fix(agent): align consumers with inbox lifecycle

apps/web/tests/queue-actions.e2e.ts

@@ -32,6 +32,7 @@ const REMOVE = 'Queue item to remove'
 const EDIT = 'Queue item to edit'
 const EDITED = 'Edited queue item'
 const TAIL = 'Queue item preserved after stop'
+const WAKE = 'Wake the preserved queue'
 
 /** Durable turn-end classifications observed by the scenario. */
 function turnEndReasons(events: readonly SessionEvent[]): string[] {
@@ -63,13 +64,13 @@ describe('web e2e: queue row actions', () => {
   it.skipIf(MODE === 'record')('edits and removes exact occurrences and preserves Queue across stop', async () => {
     overrideDir = await mkdtemp(join(tmpdir(), 'dsh-web-queue-actions-'))
     const readyFile = join(overrideDir, '.hang-ready')
-    const nextReadyFile = join(overrideDir, '.next-hang-ready')
     const overridePath = join(overrideDir, 'replay.override.json')
     const recorded = deriveReplayScript(parseSessionLog(await readFile(FIXTURE, 'utf8')))
     expect(recorded).toHaveLength(1)
     const replay: ReplayEntry[] = [
       { kind: 'hang', readyFile },
-      { kind: 'hang', readyFile: nextReadyFile },
+      recorded[0]!,
+      recorded[0]!,
       recorded[0]!,
     ]
     await writeFile(overridePath, JSON.stringify(replay))
@@ -86,7 +87,7 @@ describe('web e2e: queue row actions', () => {
     onTestFailed(() => saveFailureShot(page, 'web-e2e-queue-actions'))
 
     const input = page.locator('textarea').first()
-    const settled = scaffold.whenTurnSettled()
+    const firstSettled = scaffold.whenTurnSettled()
     await input.fill(ACTIVE_PROMPT)
     await input.press('Enter')
     await expect.poll(() => existsSync(readyFile), { timeout: 15_000 }).toBe(true)
@@ -157,19 +158,22 @@ describe('web e2e: queue row actions', () => {
     ).toBe(2)
 
     await page.getByRole('button', { name: 'Stop generating' }).click()
-    await expect.poll(() => existsSync(nextReadyFile), { timeout: 15_000 }).toBe(true)
-    await page.getByText(TAIL, { exact: true }).waitFor()
+    await firstSettled
     await expect.poll(() => page.getByRole('button', { name: 'Remove queued message' }).count())
-      .toBe(1)
+      .toBe(2)
 
     const preservedSnapshot = await captureStableAria(page, '[class*="centerCol"]', scaffold.workspaceCwd)
     await compareOrRefreshGolden(PRESERVED_EXPECTED, preservedSnapshot, MODE)
 
-    await page.getByRole('button', { name: 'Stop generating' }).click()
+    const settled = scaffold.whenTurnSettled()
+    await input.fill(WAKE)
+    await input.press('Enter')
     await settled
-    expect(turnEndReasons(sessionEvents)).toEqual(['aborted', 'aborted', 'completed'])
-    expect(sessionEvents.filter(event => event.type === 'user/message' && event.data.source.kind === 'user'))
-      .toHaveLength(3)
+    await expect.poll(() => turnEndReasons(sessionEvents), { timeout: 15_000 })
+      .toEqual(['aborted', 'completed', 'completed', 'completed'])
+    expect(sessionEvents.flatMap(event => event.type === 'user/message' && event.data.source.kind === 'user'
+      ? event.data.content.flatMap(block => block.type === 'text' ? [block.text] : [])
+      : [])).toEqual([ACTIVE_PROMPT, EDITED, TAIL, WAKE])
     await expect.poll(() => page.locator('[data-queue-dock]').count()).toBe(0)
   }, 120_000)
 

apps/web/tests/seeded-history.e2e.ts

@@ -216,7 +216,7 @@ describe('web e2e: seeded history renders through cold resume', () => {
 
     const agent = scaffold.ctx.agents.get(SessionId(SEED_ID))
     if (agent === undefined) throw new Error('seeded session did not attach an agent')
-    agent.inject(createUserMessage({
+    agent.session.append('user/message', createUserMessage({
       content: [{
         type: 'text',
         text: '<system-reminder>\n'
@@ -235,7 +235,7 @@ describe('web e2e: seeded history renders through cold resume', () => {
           digest: 'context-injection-browser-snapshot',
         }],
       },
-    }))
+    }), { surfaceOp: 'append' })
     await page.getByRole('button', { name: 'Context injection' }).waitFor({ timeout: 10_000 })
   }, 60_000)
 
@@ -357,13 +357,13 @@ describe('web e2e: seeded history renders through cold resume', () => {
     await compareOrRefreshGolden(COMMAND_ROW_EXPECTED, snapshot, MODE)
   }, 60_000)
 
-  it.skipIf(MODE === 'record')('fits short injected context without a scrollport', async () => {
+  it.skipIf(MODE === 'record')('fits short logged context without a scrollport', async () => {
     const agent = scaffold.ctx.agents.get(SessionId(SEED_ID))
     if (agent === undefined) throw new Error('seeded session did not attach an agent')
-    agent.inject(createUserMessage({
+    agent.session.append('user/message', createUserMessage({
       content: [{ type: 'text', text: 'Short injected context.' }],
       source: { kind: 'plugin', plugin: 'fixture' },
-    }))
+    }), { surfaceOp: 'append' })
 
     const disclosures = page.getByRole('button', { name: 'Context injection' })
     await expect.poll(() => disclosures.count(), { timeout: 10_000 }).toBe(2)

apps/web/tests/smoke-real.e2e.ts

@@ -196,16 +196,19 @@ describe('dsh web keyless CLI smoke', () => {
       messages?: { role?: string; content?: string }[]
       tools?: { function?: { name?: string } }[]
     }
-    let resolveProviderRequest!: (request: NativeProviderRequest) => void
-    const providerRequest = new Promise<NativeProviderRequest>((resolve) => {
-      resolveProviderRequest = resolve
+    let resolveProviderRequests!: (requests: NativeProviderRequest[]) => void
+    const requests: NativeProviderRequest[] = []
+    const providerRequests = new Promise<NativeProviderRequest[]>((resolve) => {
+      resolveProviderRequests = resolve
     })
     const provider = createServer((request, response) => {
       let body = ''
       request.setEncoding('utf8')
       request.on('data', (chunk: string) => { body += chunk })
       request.on('end', () => {
-        resolveProviderRequest(JSON.parse(body) as NativeProviderRequest)
+        const parsed = JSON.parse(body) as NativeProviderRequest
+        if ((parsed.tools?.length ?? 0) > 0) requests.push(parsed)
+        if (requests.length === 2) resolveProviderRequests(requests)
         response.writeHead(200, { 'content-type': 'text/event-stream' })
         response.end([
           'data: {"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}',
@@ -244,14 +247,21 @@ describe('dsh web keyless CLI smoke', () => {
         mode: 'queue',
         content: [{ type: 'text', text: 'go' }],
       })
-      const captured = await Promise.race([
-        providerRequest,
+      const capturedRequests = await Promise.race([
+        providerRequests,
         new Promise<never>((_resolve, reject) => {
           setTimeout(() => { reject(new Error('provider request not received in 10s')) }, 10_000).unref()
         }),
       ])
-      expect(captured.messages?.some(message =>
+      const initial = capturedRequests[0]
+      const captured = capturedRequests[1]
+      if (initial === undefined || captured === undefined) {
+        throw new Error('provider did not receive both workspace projection requests')
+      }
+      expect(initial.messages?.some(message =>
         message.role === 'user' && message.content?.includes('<available_skills>'))).toBe(false)
+      expect(initial.messages?.some(message =>
+        message.role === 'user' && message.content?.includes('web-workspace-context-probe'))).toBe(false)
       const workspaceMessage = captured.messages?.find(message =>
         message.role === 'user' && message.content?.includes('web-workspace-context-probe'))
       const systemMessage = captured.messages?.find(message => message.role === 'system')

apps/web/tests/snapshots/queue-actions/layout.expected.md

@@ -16,10 +16,6 @@
   - img
   - img
   - text: Context injection
-- button "Context injection":
-  - img
-  - img
-  - text: Context injection
 - paragraph: partial
 - status: Deep diving...
 - region "To-dos":

apps/web/tests/snapshots/queue-actions/preserved.expected.md

@@ -19,21 +19,24 @@
   - img
 - button "Branch into a new conversation":
   - img
-- text: {{clock}} Edited queue item {{clock}}
-- button "Copy":
-  - img
-- button "Branch into a new conversation":
-  - img
-- paragraph: partial
-- status: Deep diving...
+- text: {{clock}}
+- button "2 queued messages" [expanded]
 - list:
+  - listitem:
+    - text: Edited queue item
+    - button "Edit queued message":
+      - img
+    - button "Remove queued message":
+      - img
+    - button "Steer queued message" [disabled]:
+      - img
   - listitem:
     - text: Queue item preserved after stop
     - button "Edit queued message":
       - img
     - button "Remove queued message":
       - img
-    - button "Steer queued message":
+    - button "Steer queued message" [disabled]:
       - img
 - textbox "Message the agent"
 - button "Commands":
@@ -42,5 +45,5 @@
 - button "Select model, current DeepSeek-V4-Flash":
   - text: DeepSeek-V4-Flash
   - img
-- button "Stop generating"
+- button "Send message" [disabled]
 - text: 1 turns · 1 steps Input 0 tok · Output 0 tok

apps/web/tests/steering.e2e.ts

@@ -21,7 +21,7 @@ const FIXTURE = join(SNAPSHOT_DIR, 'session.jsonl')
 // Two goldens pin the transient Host projection and its durable handoff: the
 // mid-turn state renders accepted steering from session/queue while the
 // question blocks admission, then the settled state renders the same message
-// from steering/message beside the reply that obeys it.
+// from user/message beside the reply that obeys it.
 const MID_EXPECTED = join(SNAPSHOT_DIR, 'mid-steer.expected.md')
 const SETTLED_EXPECTED = join(SNAPSHOT_DIR, 'settled.expected.md')
 const MODE = webSnapshotMode()
@@ -45,6 +45,12 @@ function assistantText(events: SessionEvent[]): string {
     .join('')
 }
 
+/** Claimed user messages whose payload contains the exact scenario text. */
+function claimedMessages(events: readonly SessionEvent[], text: string): SessionEvent<'user/message'>[] {
+  return events.filter((event): event is SessionEvent<'user/message'> =>
+    event.type === 'user/message' && JSON.stringify(event.data.content).includes(text))
+}
+
 describe('web e2e: mid-turn steering lands durably and visibly', () => {
   let scaffold: WebScaffold
   let browser: Browser
@@ -74,7 +80,7 @@ describe('web e2e: mid-turn steering lands durably and visibly', () => {
   it('strictly steers one queued row; the interjection is logged, rendered, and obeyed', async () => {
     onTestFailed(() => saveFailureShot(page, 'web-e2e-steering'))
     if (MODE !== 'record') {
-      // The steer must NOT be a user/message — it lands as steering/message.
+      // The recorded prompt inventory excludes the later same-turn steer.
       expect(fixtureUserPrompts(await readFile(FIXTURE, 'utf8'))).toEqual([PROMPT])
     }
     const input = page.locator('textarea').first()
@@ -112,7 +118,7 @@ describe('web e2e: mid-turn steering lands durably and visibly', () => {
     }
 
     // Answer the composer; the tool result closes the step, the loop drains
-    // the steer as steering/message, and the steered continuation runs the
+    // the steer as user/message, and the steered continuation runs the
     // final model call.
     await composer.getByRole('radio', { name: 'Yes' }).click()
     await composer.getByRole('radio', { name: 'Yes' }).press('Enter')
@@ -124,15 +130,14 @@ describe('web e2e: mid-turn steering lands durably and visibly', () => {
       // Fixture honesty: a recording where the live model ignored the steer
       // would replay as a vacuous scenario — reject it and re-record instead.
       const recorded = parseSessionLog(await readFile(FIXTURE, 'utf8'))
-      expect(recorded.filter(e => e.type === 'steering/message')).toHaveLength(1)
+      expect(claimedMessages(recorded, STEER)).toHaveLength(1)
       expect(assistantText(recorded)).toContain('BANANA')
       return
     }
 
-    // Durable: exactly one steering/message, inside turn 1, carrying the text.
-    const steerEvents = sessionEvents.filter(e => e.type === 'steering/message')
+    // Durable: exactly one claimed user/message carrying the steering text.
+    const steerEvents = claimedMessages(sessionEvents, STEER)
     expect(steerEvents).toHaveLength(1)
-    expect((steerEvents[0] as SessionEvent & { data: { turn: number } }).data.turn).toBe(1)
     expect(JSON.stringify(steerEvents[0])).toContain('BANANA')
     const turnEnds = sessionEvents.filter(e => e.type === 'turn/end')
     expect(turnEnds).toHaveLength(1)
@@ -203,9 +208,8 @@ describe('web e2e: composer shortcut steers directly', () => {
     await composer.getByRole('radio', { name: 'Yes' }).press('Enter')
     await settled
 
-    const steerEvents = sessionEvents.filter(event => event.type === 'steering/message')
+    const steerEvents = claimedMessages(sessionEvents, STEER)
     expect(steerEvents).toHaveLength(1)
-    expect((steerEvents[0] as SessionEvent & { data: { turn: number } }).data.turn).toBe(1)
     await expect.poll(() => page.getByText(STEER, { exact: true }).count(), { timeout: 15_000 }).toBe(1)
     expect(await pendingSteering.count()).toBe(0)
     await expect.poll(() => page.getByText('BANANA', { exact: false }).count(), { timeout: 10_000 })
@@ -259,7 +263,7 @@ describe('web e2e: composer shortcut follows the swapped busy behavior', () => {
     const queuedRow = page.locator('[data-queue-dock]').getByRole('listitem').filter({ hasText: queuedText })
     await queuedRow.getByText(queuedText, { exact: true }).waitFor({ timeout: 10_000 })
     expect(await page.locator('[data-pending-steering]').filter({ hasText: queuedText }).count()).toBe(0)
-    expect(sessionEvents.filter(event => event.type === 'steering/message')).toHaveLength(0)
+    expect(claimedMessages(sessionEvents, queuedText)).toHaveLength(0)
 
     // Remove the asserted Queue row, then finish the recorded question turn
     // so replay teardown still proves that every fixture call was consumed.