Merge latest master into native Windows CI

apps/web/tests/shipped-composition.e2e.ts

@@ -30,6 +30,7 @@ const EXPECTED_TOOLS = [
   'edit',
   'exit_plan_mode',
   'get_goal',
+  'interrupt_agent',
   'list_agents',
   'ralph',
   'read',

apps/web/tests/snapshots/subagent-interrupt/offline-composer.expected.md

@@ -0,0 +1,22 @@
+- banner:
+  - navigation "Session hierarchy":
+    - button "Ask a research subagent to"
+    - text: /
+    - button "event-sourcing researcher" [disabled]
+  - tablist:
+    - tab "Chat" [selected]
+    - tab "Trajectory"
+- text: Explain event sourcing in one sentence. {{clock}}
+- button "Copy":
+  - img
+- button "Context injection @deepseek-ai/dsh-system-prompt":
+  - img
+  - img
+  - text: Context injection @deepseek-ai/dsh-system-prompt
+- paragraph: partial
+- status: Deep diving...
+- textbox "Parent session offline; sending is unavailable but you can still stop the run" [disabled]
+- button "Commands" [disabled]:
+  - img
+- 'button "Access mode, current: Workspace Write" [disabled]': Workspace Write
+- button "Stop generating"

apps/web/tests/subagent-interrupt-ui.e2e.ts

@@ -0,0 +1,271 @@
+// Web e2e scenario: the composer's primary action interrupts a running
+// continuable child. The child holds its model turn open through a replay
+// hang entry; the browser proves the single primary Send/Stop toggle, the
+// parent-offline disabled-input-with-Stop composer, the subagent.interrupt
+// (never session.cancel) transport, the parked follow-up, and the FIFO resume
+// on a waking send.
+//
+// Replay-binding note: only the PRIMARY script can hang, and scripts bind by
+// first-call order, so the child issues the composition's first model call
+// (claiming the overridden primary) and the parent's one UI prompt — needed
+// so the non-blank parent renders its header catalog — binds to a derived
+// child fixture afterwards.
+import { existsSync } from 'node:fs'
+import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import type { Browser, Page } from 'playwright'
+import { chromium } from 'playwright'
+import { afterAll, beforeAll, describe, expect, it, onTestFailed } from 'vitest'
+import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
+import type {} from '@deepseek-ai/dsh-agent'
+import {
+  acknowledgeReloadConnectionLoss, assertFixtureInventory, captureStableAria, compareOrRefreshGolden,
+  launchWebScaffold, watchConsole, webSnapshotMode, type WebScaffold,
+} from './scaffold.ts'
+import { connectFreshWorkspace, newEnglishPage, saveFailureShot } from './support.ts'
+
+const BASE_FIXTURE = fileURLToPath(new URL('./snapshots/live-interactions/session.jsonl', import.meta.url))
+
+const SNAPSHOT_DIR = fileURLToPath(new URL('./snapshots/subagent-interrupt', import.meta.url))
+const OFFLINE_COMPOSER_EXPECTED = join(SNAPSHOT_DIR, 'offline-composer.expected.md')
+const MODE = webSnapshotMode()
+const LABEL = 'event-sourcing researcher'
+const INITIAL = 'Explain event sourcing in one sentence.'
+const FOLLOWUP = 'Now give the same explanation to a human reader.'
+const WAKING = 'And add one concrete example.'
+const PARKED_ANSWER = 'parked follow-up answer'
+const WAKING_ANSWER = 'waking answer'
+
+/** Poll a synchronous condition (hook-safe; expect.poll is test-body only). */
+async function waitFor(predicate: () => boolean, what: string, timeoutMs = 30_000): Promise<void> {
+  const deadline = Date.now() + timeoutMs
+  while (!predicate()) {
+    if (Date.now() >= deadline) throw new Error(`timed out waiting for ${what}`)
+    await new Promise<void>(resolve => setTimeout(resolve, 10))
+  }
+}
+
+/** One text-only scripted model completion (no tool calls: real tools are mounted). */
+function textCompletion(text: string): object {
+  return {
+    kind: 'chunks',
+    chunks: [
+      { type: 'block-start', index: 0, blockType: 'text' },
+      { type: 'text-delta', index: 0, text },
+      { type: 'block-end', index: 0, block: { type: 'text', text } },
+      { type: 'usage', usage: { inputTokens: 20, outputTokens: 8 } },
+      { type: 'finish', reason: { kind: 'stop' } },
+    ],
+  }
+}
+
+describe.skipIf(MODE === 'record')('web e2e: composer interrupt for a running continuable child', () => {
+  let scaffold: WebScaffold
+  let browser: Browser
+  let page: Page
+  let sidecarRoot: string
+  let childId: SessionId
+  let tripwire: ReturnType<typeof watchConsole>
+  const apiCalls: string[] = []
+
+  beforeAll(async () => {
+    sidecarRoot = await mkdtemp(join(tmpdir(), 'dsh-web-subagent-interrupt-ui-'))
+    const readyFile = join(sidecarRoot, 'hang-ready')
+    // The child claims this whole-script replacement: held turn 1, then the
+    // parked follow-up and waking turns.
+    await writeFile(join(sidecarRoot, 'replay.override.json'), JSON.stringify([
+      { kind: 'hang', readyFile },
+      textCompletion(PARKED_ANSWER),
+      textCompletion(WAKING_ANSWER),
+    ]))
+    await writeFile(
+      join(sidecarRoot, 'session.jsonl'),
+      '{"type":"session","version":0,"id":"primary","createdAt":0}\n',
+    )
+    // The parent's one prompted turn replays this recorded single text-only
+    // call (binding is positional, not lineage-aware).
+    const parentTurnPath = join(sidecarRoot, 'parent-turn.jsonl')
+    const base = await readFile(BASE_FIXTURE, 'utf8')
+    const [header, ...events] = base.trimEnd().split('\n')
+    if (header === undefined) throw new Error('base replay fixture has no header')
+    await writeFile(parentTurnPath, [
+      header
+        .replace('"id":"{{sessionId}}"', '"id":"recorded-parent-turn"')
+        .replace(/"createdAt":\d+/, '"createdAt":1784998084442'),
+      ...events,
+      '',
+    ].join('\n'))
+    scaffold = await launchWebScaffold({
+      replayFixture: join(sidecarRoot, 'session.jsonl'),
+      replayOverride: join(sidecarRoot, 'replay.override.json'),
+      replayChildFixtures: [parentTurnPath],
+    })
+    browser = await chromium.launch()
+    page = await newEnglishPage(browser)
+    page.on('request', (request) => {
+      const path = new URL(request.url()).pathname
+      if (path.startsWith('/api/')) apiCalls.push(path)
+    })
+    tripwire = watchConsole(page)
+    await page.goto(scaffold.baseUrl, { waitUntil: 'load' })
+    await page.waitForSelector('[class*="frame"]', { timeout: 30_000 })
+    await connectFreshWorkspace(page, scaffold.workspaceCwd)
+
+    const parent = scaffold.ctx.agents.roots()[0]
+    if (parent === undefined) throw new Error('fresh workspace did not publish its parent Agent')
+    // The child's first model call claims the primary override and holds.
+    const started = await scaffold.ctx.subagents.startContinuable({
+      provider: 'spawn',
+      label: LABEL,
+      signal: new AbortController().signal,
+      request: { prompt: [{ type: 'text', text: INITIAL }], parent },
+    })
+    childId = started.childId
+    await waitFor(() => existsSync(readyFile), 'the held child turn to open')
+
+    // One prompted parent turn makes the parent non-blank so the session
+    // header (and its subagent catalog action) renders.
+    const parentSettled = scaffold.whenTurnSettled()
+    const parentInput = page.locator('textarea:enabled').first()
+    await parentInput.fill('Ask a research subagent to explain event sourcing.')
+    await parentInput.press('Enter')
+    expect(await parentSettled).toBe(parent.id)
+
+    // Reload onto the restart baseline (the proven route to a freshly
+    // discovered catalog), with the child still live and running host-side.
+    const warningStart = tripwire.warnings.length
+    await page.reload({ waitUntil: 'load' })
+    await page.waitForSelector('[class*="frame"]', { timeout: 30_000 })
+    await page.getByRole('button', { name: /1 subagent/ }).waitFor({ timeout: 15_000 })
+    acknowledgeReloadConnectionLoss(tripwire, warningStart)
+    expect(scaffold.ctx.agents.get(childId)?.status).toBe('running')
+  }, 120_000)
+
+  afterAll(async () => {
+    const failures: unknown[] = []
+    await browser?.close().catch((error: unknown) => failures.push(error))
+    await scaffold?.close().catch((error: unknown) => failures.push(error))
+    if (sidecarRoot !== undefined) {
+      await rm(sidecarRoot, { recursive: true, force: true })
+        .catch((error: unknown) => failures.push(error))
+    }
+    if (failures.length === 1) throw failures[0]
+    if (failures.length > 1) throw new AggregateError(failures, 'subagent interrupt UI teardown failed')
+  })
+
+  it('locks input but keeps the same primary Stop when the parent is offline', async () => {
+    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-interrupt-offline'))
+    // Simulate a parent that went offline: the catalog delivers
+    // parentAvailable: false while the child Activation stays live (the
+    // interrupt RPC itself needs no live parent — PR 1's host coverage).
+    const pattern = '**/api/subagent.list'
+    await page.route(pattern, async (route) => {
+      const response = await route.fetch()
+      const body = await response.json() as {
+        result: { ok: true; value: { parentAvailable: boolean } } | { ok: false }
+      }
+      if (body.result.ok) body.result.value.parentAvailable = false
+      await route.fulfill({ response, json: body })
+    })
+    try {
+      await page.getByRole('button', { name: /1 subagent/ }).click()
+      await page.getByRole('treeitem', { name: new RegExp(LABEL) }).click()
+      const input = page.getByRole('textbox', {
+        name: 'Parent session offline; sending is unavailable but you can still stop the run',
+      })
+      await input.waitFor({ timeout: 15_000 })
+      expect(await input.isDisabled()).toBe(true)
+      // Still exactly one primary action, and it is an enabled Stop.
+      const stop = page.getByRole('button', { name: 'Stop generating' })
+      expect(await stop.count()).toBe(1)
+      expect(await stop.isEnabled()).toBe(true)
+      expect(await page.getByRole('button', { name: 'Send message' }).count()).toBe(0)
+      await compareOrRefreshGolden(
+        OFFLINE_COMPOSER_EXPECTED,
+        await captureStableAria(page, '[class*="centerCol"]', scaffold.workspaceCwd),
+        MODE,
+      )
+    } finally {
+      await page.unroute(pattern)
+    }
+  }, 60_000)
+
+  it('interrupts through subagent.interrupt, parks the follow-up, and resumes it FIFO', async () => {
+    onTestFailed(() => saveFailureShot(page, 'web-e2e-subagent-interrupt-flow'))
+    // Reselect the child with the truthful catalog: parent available again.
+    await page.getByRole('navigation', { name: 'Session hierarchy' })
+      .getByRole('button').first().click()
+    await page.getByRole('button', { name: /1 subagent/ }).click()
+    await page.getByRole('treeitem', { name: new RegExp(LABEL) }).click()
+    const input = page.getByRole('textbox', { name: 'Message the agent' })
+    await input.waitFor({ timeout: 15_000 })
+    expect(await input.isDisabled()).toBe(false)
+
+    // Queue a follow-up while the turn is open; the primary stays Stop.
+    const promptResponse = page.waitForResponse(response =>
+      new URL(response.url()).pathname === '/api/subagent.prompt')
+    await input.fill(FOLLOWUP)
+    await input.press('Enter')
+    expect(((await (await promptResponse).json()) as { result: { ok: boolean } }).result)
+      .toMatchObject({ ok: true })
+
+    const aborted = new Promise<void>((resolve, reject) => {
+      const timer = setTimeout(() => {
+        off()
+        reject(new Error('interrupt did not reach an aborted turn/end'))
+      }, 30_000)
+      const off = scaffold.ctx.on('session/event', (session: { id: SessionId }, event: SessionEvent) => {
+        if (session.id !== childId || event.type !== 'turn/end') return
+        clearTimeout(timer)
+        off()
+        if (event.data.reason.kind === 'aborted') resolve()
+        else reject(new Error(`expected an aborted turn/end, got ${event.data.reason.kind}`))
+      })
+    })
+    const stop = page.getByRole('button', { name: 'Stop generating' })
+    expect(await stop.count()).toBe(1)
+    const interruptResponse = page.waitForResponse(response =>
+      new URL(response.url()).pathname === '/api/subagent.interrupt')
+    await stop.click()
+    expect(((await (await interruptResponse).json()) as {
+      result: { ok: boolean; value?: { accepted: boolean } }
+    }).result).toMatchObject({ ok: true, value: { accepted: true } })
+    // The addressed child stops through its own RPC, never the generic one.
+    expect(apiCalls.filter(path => path === '/api/session.cancel')).toEqual([])
+    await aborted
+
+    // Parked: the Activation stays resident and idle with the retained
+    // follow-up; the primary returns to Send without a new turn starting.
+    await expect.poll(() => scaffold.ctx.agents.get(childId)?.status, { timeout: 15_000 }).toBe('idle')
+    const child = scaffold.ctx.agents.get(childId)
+    expect(child).toBeDefined()
+    expect(child!.inbox.nextTurn).toHaveLength(1)
+    expect(child!.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
+    await page.getByRole('button', { name: 'Send message' }).waitFor({ timeout: 15_000 })
+
+    // Only the waking send resumes the parked queue, FIFO, to settlement.
+    await input.fill(WAKING)
+    await input.press('Enter')
+    await expect.poll(() => page.getByText(PARKED_ANSWER, { exact: true }).count(), { timeout: 30_000 }).toBe(1)
+    await expect.poll(() => page.getByText(WAKING_ANSWER, { exact: true }).count(), { timeout: 30_000 }).toBe(1)
+    await expect.poll(() => scaffold.ctx.agents.get(childId), { timeout: 60_000 }).toBeUndefined()
+
+    const loaded = await scaffold.ctx.sessionPersistence.load(childId)
+    const userTexts = loaded.events.flatMap(event => event.type === 'user/message'
+      && event.data.source.kind === 'user'
+      ? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : [])
+      : [])
+    expect(userTexts).toEqual([INITIAL, FOLLOWUP, WAKING])
+    const turnEndKinds = loaded.events
+      .filter(event => event.type === 'turn/end')
+      .map(event => event.data.reason.kind)
+    expect(turnEndKinds).toEqual(['aborted', 'completed', 'completed'])
+    expect(tripwire.pageErrors).toEqual([])
+  }, 120_000)
+
+  it('keeps its snapshot inventory closed', async () => {
+    await assertFixtureInventory(SNAPSHOT_DIR, ['offline-composer.expected.md'])
+  })
+})

apps/web/tests/subagent-interrupt.e2e.ts

@@ -0,0 +1,176 @@
+// Web e2e scenario (browserless): the subagent.interrupt RPC against the real
+// composition. A live continuable child holds its model turn open through a
+// replay hang entry; plain HTTP queues a follow-up, interrupts the turn, and
+// proves from the real session state that the turn aborted, the follow-up
+// parked without auto-starting a new turn, and a later waking send resumed the
+// preserved FIFO order. No browser: the RPC surface is the product surface
+// under test, and PR-stacked UI coverage owns the composer interaction.
+import { existsSync } from 'node:fs'
+import { mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterAll, beforeAll, describe, expect, it } from 'vitest'
+import { SessionId as sessionId, type SessionId } from '@deepseek-ai/dsh-session'
+import type {} from '@deepseek-ai/dsh-agent'
+import { launchWebScaffold, webSnapshotMode, type WebScaffold } from './scaffold.ts'
+
+const MODE = webSnapshotMode()
+const INITIAL = 'Explain event sourcing in one sentence.'
+const FOLLOWUP = 'Now give the same explanation to a human reader.'
+const WAKING = 'And add one concrete example.'
+
+type RpcResult<T> = { ok: true; value: T } | { ok: false; error: { code: string; message: string } }
+
+/** POST one unary RPC through the real HTTP carrier and unwrap its result. */
+async function rpc<T>(baseUrl: string, method: string, payload: unknown): Promise<RpcResult<T>> {
+  const response = await fetch(`${baseUrl}/api/${method}`, {
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    body: JSON.stringify({
+      type: 'client-request',
+      rpcId: `interrupt-e2e-${method}-${crypto.randomUUID()}`,
+      method,
+      payload,
+    }),
+  })
+  if (!response.ok) throw new Error(`${method} failed over HTTP ${response.status}: ${await response.text()}`)
+  return (await response.json() as { result: RpcResult<T> }).result
+}
+
+/** Poll a synchronous condition (hook-safe; expect.poll is test-body only). */
+async function waitFor(predicate: () => boolean, what: string, timeoutMs = 30_000): Promise<void> {
+  const deadline = Date.now() + timeoutMs
+  while (!predicate()) {
+    if (Date.now() >= deadline) throw new Error(`timed out waiting for ${what}`)
+    await new Promise<void>(resolve => setTimeout(resolve, 10))
+  }
+}
+
+/** One text-only scripted model completion (no tool calls: real tools are mounted). */
+function textCompletion(text: string): object {
+  return {
+    kind: 'chunks',
+    chunks: [
+      { type: 'block-start', index: 0, blockType: 'text' },
+      { type: 'text-delta', index: 0, text },
+      { type: 'block-end', index: 0, block: { type: 'text', text } },
+      { type: 'usage', usage: { inputTokens: 20, outputTokens: 8 } },
+      { type: 'finish', reason: { kind: 'stop' } },
+    ],
+  }
+}
+
+describe.skipIf(MODE === 'record')('web e2e: subagent.interrupt over the real composition', () => {
+  let scaffold: WebScaffold
+  let sidecarRoot: string
+  let readyFile: string
+  let parentId: SessionId
+  let childId: SessionId
+
+  beforeAll(async () => {
+    sidecarRoot = await mkdtemp(join(tmpdir(), 'dsh-web-subagent-interrupt-'))
+    readyFile = join(sidecarRoot, 'hang-ready')
+    // Whole-script replacement: the child's three model calls are the hang
+    // (turn 1, interrupted), the parked follow-up's turn, and the waking turn.
+    // The parent never runs a turn, so the child claims this primary script.
+    await writeFile(join(sidecarRoot, 'replay.override.json'), JSON.stringify([
+      { kind: 'hang', readyFile },
+      textCompletion('resumed response one'),
+      textCompletion('resumed response two'),
+    ]))
+    // Header-only primary fixture: the bare-array override replaces the
+    // derived script entirely; the path only anchors replay installation.
+    await writeFile(
+      join(sidecarRoot, 'session.jsonl'),
+      '{"type":"session","version":0,"id":"primary","createdAt":0}\n',
+    )
+    scaffold = await launchWebScaffold({
+      replayFixture: join(sidecarRoot, 'session.jsonl'),
+      replayOverride: join(sidecarRoot, 'replay.override.json'),
+    })
+
+    // A live parent Agent through the real API; no workspace or browser.
+    const created = await rpc<{ sessionId: string }>(scaffold.baseUrl, 'session.create', {
+      cwd: scaffold.workspaceCwd,
+    })
+    if (!created.ok) throw new Error(`session.create failed: ${created.error.code}`)
+    parentId = sessionId(created.value.sessionId)
+    const parent = scaffold.ctx.agents.get(parentId)
+    if (parent === undefined) throw new Error('created parent session did not publish a live Agent')
+
+    const started = await scaffold.ctx.subagents.startContinuable({
+      provider: 'spawn',
+      label: 'event-sourcing researcher',
+      signal: new AbortController().signal,
+      request: { prompt: [{ type: 'text', text: INITIAL }], parent },
+    })
+    childId = started.childId
+    // The hang entry writes readyFile after its prefix chunks, immediately
+    // before waiting for cancellation: the deterministic "turn is open" gate.
+    await waitFor(() => existsSync(readyFile), 'the held child turn to open')
+  }, 120_000)
+
+  afterAll(async () => {
+    const failures: unknown[] = []
+    await scaffold?.close().catch((error: unknown) => failures.push(error))
+    await rm(sidecarRoot, { recursive: true, force: true }).catch((error: unknown) => failures.push(error))
+    if (failures.length === 1) throw failures[0]
+    if (failures.length > 1) throw new AggregateError(failures, 'subagent interrupt teardown failed')
+  })
+
+  it('parks a queued follow-up on interrupt and resumes it FIFO on a waking send', async () => {
+    // Queue the follow-up while the turn is still open, then interrupt.
+    const queued = await rpc<{ messageId: string }>(scaffold.baseUrl, 'subagent.prompt', {
+      parentSessionId: parentId,
+      childSessionId: childId,
+      mode: 'continuable',
+      content: [{ type: 'text', text: FOLLOWUP }],
+    })
+    expect(queued).toMatchObject({ ok: true })
+
+    const settled = scaffold.whenTurnSettled()
+    const interrupted = await rpc<{ accepted: true }>(scaffold.baseUrl, 'subagent.interrupt', {
+      parentSessionId: parentId,
+      childSessionId: childId,
+      mode: 'continuable',
+    })
+    expect(interrupted).toMatchObject({ ok: true, value: { accepted: true } })
+    // accepted acknowledges the admitted cancel, not quiescence: wait for the
+    // aborted turn/end (the composition's first turn/end) before asserting.
+    expect(await settled).toBe(childId)
+
+    // Parked, not resumed: the Activation stays resident with an idle driver,
+    // the follow-up is retained, and no second turn opened.
+    const child = scaffold.ctx.agents.get(childId)
+    expect(child).toBeDefined()
+    expect(child!.status).toBe('idle')
+    expect(child!.inbox.nextTurn).toHaveLength(1)
+    expect(child!.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
+    const lastEnd = child!.session.events.filter(event => event.type === 'turn/end').at(-1)
+    expect((lastEnd)?.data.reason.kind).toBe('aborted')
+
+    // Only an explicit waking send resumes the parked queue, FIFO, then the
+    // child runs both turns to completion and settles.
+    const waking = await rpc<{ messageId: string }>(scaffold.baseUrl, 'subagent.prompt', {
+      parentSessionId: parentId,
+      childSessionId: childId,
+      mode: 'continuable',
+      content: [{ type: 'text', text: WAKING }],
+    })
+    expect(waking).toMatchObject({ ok: true })
+    await expect.poll(() => scaffold.ctx.agents.get(childId), { timeout: 60_000 }).toBeUndefined()
+
+    const loaded = await scaffold.ctx.sessionPersistence.load(childId)
+    // Human-origin messages only: the real composition also injects
+    // runtime-context snapshots as non-user-source messages.
+    const userTexts = loaded.events.flatMap(event => event.type === 'user/message'
+      && event.data.source.kind === 'user'
+      ? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : [])
+      : [])
+    expect(userTexts).toEqual([INITIAL, FOLLOWUP, WAKING])
+    const turnEndKinds = loaded.events
+      .filter(event => event.type === 'turn/end')
+      .map(event => (event).data.reason.kind)
+    expect(turnEndKinds).toEqual(['aborted', 'completed', 'completed'])
+  }, 120_000)
+})

apps/web/tsconfig.json

@@ -67,6 +67,8 @@
     "tests/produced-file-mentions.e2e.ts",
     "tests/goal-bar.e2e.ts",
     "tests/subagent-conversation.e2e.ts",
+    "tests/subagent-interrupt.e2e.ts",
+    "tests/subagent-interrupt-ui.e2e.ts",
     "tests/sidebar-subagent-activity.e2e.ts",
     "tests/bash-abort-row.e2e.ts",
     "tests/skill-tool-row.e2e.ts",