fix: commit step context before request dispatch

2026-08-15 21:04:50 +00:00 · 2026-07-30 18:18:04 +08:00
parent a31331cb0e
commit a4ae0b4126
17 changed files with 132 additions and 228 deletions
--- a/packages/context/time-context/src/index.ts
+++ b/packages/context/time-context/src/index.ts
@@ -174,6 +174,6 @@ export function apply(ctx: Context, config: Config): void {
    const previous = step === 1
      ? precedingMessageTime(agent)
      : precedingStepContextTime(agent, turn)
-    agent.inject(createUserMessage({ content: [{ type: 'text', text: renderText(now, turn, step, previous, formatter, resolvedTimeZone) }], source: { kind: 'plugin', plugin: name } }))
+    agent.session.append('user/message', createUserMessage({ content: [{ type: 'text', text: renderText(now, turn, step, previous, formatter, resolvedTimeZone) }], source: { kind: 'plugin', plugin: name } }), { surfaceOp: 'append' })
  }, { prepend: true })
 }
--- a/packages/context/time-context/src/invariant.ts
+++ b/packages/context/time-context/src/invariant.ts
@@ -18,31 +18,22 @@ export const name = 'time-context-invariant'
 /** Service required before the companion can reserve package ownership. */
 export const inject = ['invariants']

-/** Derive the pre-step position at which a time-context reading may append. */
+/** Derive the open step in which a time-context reading may append. */
 function preparationPosition(history: readonly SessionEvent[], fail: InvariantFailure): { turn: number; step: number } {
-  const currentTurnEvents: SessionEvent[] = []
-  let openTurn: number | undefined
  for (const event of history.slice().reverse()) {
-    if (event.type === 'turn/end') {
-      fail('time-context reading must be appended inside an open turn')
-    }
-    if (event.type === 'turn/start') {
-      openTurn = event.data.turn
-      break
-    }
-    currentTurnEvents.push(event)
-  }
-  if (openTurn === undefined) fail('time-context reading must be appended inside an open turn')
-
-  for (const event of currentTurnEvents) {
-    if (event.type === 'step/start') {
-      fail(`time-context reading must precede step/start, but step ${event.data.step} is already open`)
-    }
-    if (event.type === 'step/end') {
-      return { turn: openTurn, step: event.data.step + 1 }
+    switch (event.type) {
+      case 'step/start':
+        return event.data
+      case 'step/end':
+      case 'turn/start':
+      case 'turn/end':
+        fail('time-context reading must be appended inside an open step')
+        break
+      default:
+        break
    }
  }
-  return { turn: openTurn, step: 1 }
+  fail('time-context reading must be appended inside an open step')
 }

 /** Validate one plugin-attributed time reading against its session position and timestamp. */
--- a/packages/context/time-context/tests/invariant.spec.ts
+++ b/packages/context/time-context/tests/invariant.spec.ts
@@ -58,6 +58,7 @@ function preparing(turn: number, step: number): Session {
    session.append('step/start', { turn, step: priorStep })
    session.append('step/end', { turn, step: priorStep })
  }
+  session.append('step/start', { turn, step })
  return session
 }

@@ -92,8 +93,8 @@ describe('time-context invariants', () => {
      content: [{ type: 'text', text: 'prepare' }],
      source: { kind: 'user' },
    }), { surfaceOp: 'append' })
-    appendReading(session, reading())
    session.append('step/start', { turn: 1, step: 1 })
+    appendReading(session, reading())

    await ctx.plugin(InvariantService, { enabled: true })
    await expect(ctx.plugin(TimeInvariant)).resolves.toBeDefined()
@@ -108,6 +109,7 @@ describe('time-context invariants', () => {
      content: [{ type: 'text', text: 'prepare' }],
      source: { kind: 'user' },
    }), { surfaceOp: 'append' })
+    session.append('step/start', { turn: 1, step: 1 })
    appendReading(session, reading('1', '2', 'step context'))

    await ctx.plugin(InvariantService, { enabled: true })
@@ -127,17 +129,17 @@ describe('time-context invariants', () => {
    const session = preparing(1, 2)
    session.append('turn/end', { turn: 1, reason: { kind: 'aborted', reason: { kind: 'user' } } })
    expect(() => { ctx.emit('session/event', session, event(reading('1', '2', 'step context'))) })
-      .toThrow(/inside an open turn/)
+      .toThrow(/inside an open step/)
  })

-  it('rejects a reading after step/start or without any open turn', async () => {
+  it('rejects a reading outside an open step', async () => {
    const ctx = await setup()
-    const started = preparing(1, 1)
-    started.append('step/start', { turn: 1, step: 1 })
-    expect(() => { ctx.emit('session/event', started, event(reading())) }).toThrow(/must precede step\/start/)
+    const ended = preparing(1, 1)
+    ended.append('step/end', { turn: 1, step: 1 })
+    expect(() => { ctx.emit('session/event', ended, event(reading())) }).toThrow(/inside an open step/)
    expect(() => {
      ctx.emit('session/event', new Session(SessionId('time-invariant-empty')), event(reading()))
-    }).toThrow(/inside an open turn/)
+    }).toThrow(/inside an open step/)
  })

  it.each([
--- a/packages/context/time-context/tests/time-context.e2e.ts
+++ b/packages/context/time-context/tests/time-context.e2e.ts
@@ -54,7 +54,7 @@ describe('time-context through a real headless cordis.yml', () => {
    expect(contexts).toHaveLength(2)
    expect(starts).toHaveLength(2)
    for (let index = 0; index < contexts.length; index += 1) {
-      expect(contexts[index]!.seq).toBeLessThan(starts[index]!.seq)
+      expect(contexts[index]!.seq).toBeGreaterThan(starts[index]!.seq)
      expect(contexts[index]!.surfaceOp).toBe('append')
      expect(contexts[index]!.data.source).toEqual({ kind: 'plugin', plugin: 'time-context' })
    }
--- a/packages/context/time-context/tests/time-context.spec.ts
+++ b/packages/context/time-context/tests/time-context.spec.ts
@@ -45,9 +45,7 @@ function sessionAgent(session: Session, id = 'agent'): Agent {
    ctx: new Context(),
    followup: () => {},
    steer: () => {},
-    inject(input) {
-      session.append('user/message', input, { surfaceOp: 'append' })
-    },
+    inject: () => { throw new Error('time-context must append directly to the open step') },
    cancel() {},
    whenIdle: () => Promise.resolve(),
  }
@@ -358,7 +356,7 @@ describe('real agent-loop request history', () => {
  it.each([
    ['throws', 'error'],
    ['cancels', 'aborted'],
-  ] as const)('discards the pending preparation reading when a later step listener %s', async (mode, reasonKind) => {
+  ] as const)('retains the durable preparation reading when a later step listener %s', async (mode, reasonKind) => {
    const adapter = new ScriptedAdapter([textResponse('unused')])
    const ctx = await loopHarness(adapter)
    let laterSawReading = false
@@ -372,10 +370,10 @@ describe('real agent-loop request history', () => {
    agent.followup(createUserMessage({ content: [{ type: 'text', text: 'start' }], source: { kind: 'user' } }))
    await agent.whenIdle()

-    expect(laterSawReading).toBe(false)
-    expect(contextTexts(agent.session)).toHaveLength(0)
+    expect(laterSawReading).toBe(true)
+    expect(contextTexts(agent.session)).toHaveLength(1)
    expect(adapter.requests).toHaveLength(0)
-    expect(agent.session.events.some(event => event.type === 'step/start')).toBe(false)
+    expect(agent.session.events.some(event => event.type === 'step/start')).toBe(true)
    const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
    expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe(reasonKind)
    await ctx.fiber.dispose()
@@ -405,7 +403,7 @@ describe('real agent-loop request history', () => {
    expect(contexts).toHaveLength(adapter.requests.length)
    expect(starts).toHaveLength(adapter.requests.length)
    for (let index = 0; index < contexts.length; index += 1) {
-      expect(contexts[index]!.seq).toBeLessThan(starts[index]!.seq)
+      expect(contexts[index]!.seq).toBeGreaterThan(starts[index]!.seq)
    }
    expect(contexts.every(event => event.data.source.kind === 'plugin'
      && event.data.source.plugin === 'time-context'
--- a/packages/context/tmux-context/src/index.ts
+++ b/packages/context/tmux-context/src/index.ts
@@ -233,9 +233,9 @@ export function apply(ctx: Context, config: Config): void {
    if (location === undefined) return
    const state = renderState(location)
    if (previous !== undefined && previous.state === state) return
-    agent.inject(createUserMessage({
+    agent.session.append('user/message', createUserMessage({
      content: [{ type: 'text', text: renderReading(location, turn) }],
      source: { kind: 'plugin', plugin: name },
-    }))
+    }), { surfaceOp: 'append' })
  }, { prepend: true })
 }
--- a/packages/context/tmux-context/tests/tmux-context.spec.ts
+++ b/packages/context/tmux-context/tests/tmux-context.spec.ts
@@ -101,9 +101,7 @@ function sessionAgent(session: Session, id = 'agent'): Agent {
    ctx: new Context(),
    followup: () => {},
    steer: () => {},
-    inject(input) {
-      session.append('user/message', input, { surfaceOp: 'append' })
-    },
+    inject: () => { throw new Error('tmux-context must append directly to the open step') },
    cancel() {},
    whenIdle: () => Promise.resolve(),
  }
--- a/packages/context/workspace-context/src/index.ts
+++ b/packages/context/workspace-context/src/index.ts
@@ -116,20 +116,20 @@ export function apply(ctx: Context, config: Config): void {
      { includeBaselineScopes: false, signal },
    )
    if (update !== undefined) {
-      agent.inject(update.context)
+      agent.session.append('user/message', update.context, { surfaceOp: 'append' })
      applyInstructionVersionUpdates(agent.session, update.versionUpdates, instructionVersions)
    }
    const keepVisibleBaseline = !lifecycleWitnessed.has(agent.session) && hasVisibleBaseline(agent)
    if (!keepVisibleBaseline && instructions !== undefined && instructions.rendered.text.length > 0) {
      const baselineMessage = workspaceContextMessage(instructions.rendered.text)
-      agent.inject(createUserMessage({
+      agent.session.append('user/message', createUserMessage({
        content: baselineMessage.content,
        source: {
          kind: 'workspace-instructions',
          baseline: true,
          changes: [...baseline.changes.values()],
        },
-      }))
+      }), { surfaceOp: 'append' })
    }
    baselineLoaded.add(agent.session)
  })
--- a/packages/context/workspace-context/tests/workspace-context.spec.ts
+++ b/packages/context/workspace-context/tests/workspace-context.spec.ts
@@ -180,9 +180,7 @@ function stubAgent(cwd?: string, seed: SessionEvent[] = []): Agent {
    status: 'idle',
    followup: () => {},
    steer: () => {},
-    inject(input) {
-      session.append('user/message', input, { surfaceOp: 'append' })
-    },
+    inject: () => { throw new Error('workspace-context must append directly to the open step') },
    cancel() {},
    whenIdle: () => Promise.resolve(),
  }
@@ -1164,7 +1162,7 @@ describe('workspace context request injection', () => {
      const ctx = new Context()
      await mountWorkspaceContext(ctx, { dshHome: home, maxBytes: 65536 })
      ctx.on('agent/step', (agent) => {
-        agent.inject(createUserMessage({ content: [{ type: 'text', text: '<system-reminder>Available skills</system-reminder>' }], source: { kind: 'plugin', plugin: 'test-skills' } }))
+        agent.session.append('user/message', createUserMessage({ content: [{ type: 'text', text: '<system-reminder>Available skills</system-reminder>' }], source: { kind: 'plugin', plugin: 'test-skills' } }), { surfaceOp: 'append' })
      })

      const prefix = await composeBaselinePrefix(ctx, stubAgent(root))
--- a/packages/core/agent-loop/src/invariant.ts
+++ b/packages/core/agent-loop/src/invariant.ts
@@ -6,7 +6,7 @@
 import type { Context } from 'cordis'
 import { isAgentLoopRequest, type GenerateOptions } from '@deepseek-ai/dsh-llm'
 import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
-import { Session, SessionId, foldRequestHeader } from '@deepseek-ai/dsh-session'
+import { foldRequestHeader } from '@deepseek-ai/dsh-session'

 const PACKAGE_NAME = '@deepseek-ai/dsh-agent-loop'

@@ -17,8 +17,7 @@ export const inject = ['invariants']

 /** Install the request-reconstruction contribution into its child registration fiber. */
 const install: InvariantInstaller = Object.assign((ctx: Context, fail: InvariantFailure) => {
-  // Prepend prevents a short-circuiting replay listener from silencing the
-  // check; correctness itself comes from the sequence-bounded reconstruction.
+  // Prepend prevents a short-circuiting replay listener from silencing the check.
  ctx.on('llm/stream', (options: GenerateOptions, next) => {
    if (!isAgentLoopRequest(options)) return next()
    if (!Object.isFrozen(options)) fail('a loop-built request must be frozen')
@@ -30,27 +29,16 @@ const install: InvariantInstaller = Object.assign((ctx: Context, fail: Invariant
    }

    const events = session.events
-    let boundary = -1
-    for (let index = events.length - 1; index >= 0; index -= 1) {
-      if (events[index]?.type === 'step/start') {
-        boundary = index
-        break
-      }
-    }
-    if (boundary === -1) {
+    if (!events.some(event => event.type === 'step/start')) {
      return fail('a loop-built request with no step/start in its session log')
    }
    const header = foldRequestHeader(events)
    if (header === undefined) {
      return fail('a loop-built request with no request/header event in its session log')
    }
-    const rebuilt = new Session(
-      SessionId(`${String(session.id)}-invariant-rebuild`),
-      structuredClone(events.slice(0, boundary)),
-    )
-    const expected = rebuilt.deriveMessages()
+    const expected = session.deriveMessages()
    if (JSON.stringify(options.messages) !== JSON.stringify(expected)) {
-      fail(`llm request for session "${String(session.id)}" diverges from the boundary derivation (log-reconstruction desync)`)
+      fail(`llm request for session "${String(session.id)}" diverges from the dispatch-time durable derivation (log-reconstruction desync)`)
    }

    const headerMatches = options.model === header.config.model
--- a/packages/core/agent-loop/tests/invariant.spec.ts
+++ b/packages/core/agent-loop/tests/invariant.spec.ts
@@ -42,12 +42,16 @@ describe('request-reconstruction invariant', () => {
    expect(() => { dispatch(ctx, options) }).not.toThrow()
  })

-  it('uses the step boundary rather than content appended afterward', async () => {
-    const { ctx, session, boundary } = await requestSetup()
+  it('includes context appended inside the open step before dispatch', async () => {
+    const { ctx, session } = await requestSetup()
    session.append('user/message', createUserMessage({
-      content: [{ type: 'text', text: '[late]' }], source: { kind: 'plugin', plugin: 'x' },
+      content: [{ type: 'text', text: '[step context]' }], source: { kind: 'plugin', plugin: 'x' },
    }), { surfaceOp: 'append' })
-    const options = loopRequest({ model: 'm', messages: Object.freeze(boundary), sessionId: session.id })
+    const options = loopRequest({
+      model: 'm',
+      messages: Object.freeze(session.deriveMessages()),
+      sessionId: session.id,
+    })
    expect(() => { dispatch(ctx, options) }).not.toThrow()
  })

@@ -57,16 +61,16 @@ describe('request-reconstruction invariant', () => {
    expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([...boundary]), sessionId: session.id })) })
      .not.toThrow()
    expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([extra, ...boundary]), sessionId: session.id })) })
-      .toThrow(/diverges from the boundary derivation/)
+      .toThrow(/diverges from the dispatch-time durable derivation/)
    expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze([...boundary, extra]), sessionId: session.id })) })
-      .toThrow(/diverges from the boundary derivation/)
+      .toThrow(/diverges from the dispatch-time durable derivation/)
  })

  it('rejects message and header divergence', async () => {
    const { ctx, session, boundary } = await requestSetup()
    const divergent = [...boundary, { role: 'user', content: [{ type: 'text', text: 'phantom' }] }]
    expect(() => { dispatch(ctx, loopRequest({ model: 'm', messages: Object.freeze(divergent), sessionId: session.id })) })
-      .toThrow(/diverges from the boundary derivation/)
+      .toThrow(/diverges from the dispatch-time durable derivation/)
    expect(() => { dispatch(ctx, loopRequest({ model: 'other', messages: Object.freeze(boundary), sessionId: session.id })) })
      .toThrow(/diverges from the folded request header/)
  })
@@ -133,6 +137,6 @@ describe('request-reconstruction invariant', () => {
      messages: Object.freeze([{ role: 'user', content: [{ type: 'text', text: 'phantom' }] }]),
      sessionId: session.id,
    })
-    expect(() => { dispatch(ctx, divergent) }).toThrow(/diverges from the boundary derivation/)
+    expect(() => { dispatch(ctx, divergent) }).toThrow(/diverges from the dispatch-time durable derivation/)
  })
 })
--- a/packages/skill/tool-skill/src/index.ts
+++ b/packages/skill/tool-skill/src/index.ts
@@ -149,7 +149,7 @@ export function apply(ctx: Context, config: Config = {}): void {
    const catalog = history.published
      ? renderCatalogUpdate(skills, catalogDescriptionMaxLength)
      : renderCatalogMessage(skills, catalogDescriptionMaxLength)
-    agent.inject(catalog)
+    agent.session.append('user/message', catalog, { surfaceOp: 'append' })
  })
 }

--- a/packages/skill/tool-skill/tests/tool-skill.spec.ts
+++ b/packages/skill/tool-skill/tests/tool-skill.spec.ts
@@ -48,9 +48,7 @@ function agentForCwd(cwd: string): Agent {
    status: 'idle',
    followup: () => {},
    steer: () => {},
-    inject(input) {
-      session.append('user/message', input, { surfaceOp: 'append' })
-    },
+    inject: () => { throw new Error('step-boundary catalog must not use agent.inject()') },
    cancel() {},
    whenIdle: () => Promise.resolve(),
  }
@@ -66,9 +64,7 @@ function sessionAgent(session: Session, id = 'tool-skill-agent'): Agent {
    ctx: new Context(),
    followup: () => {},
    steer: () => {},
-    inject(input) {
-      session.append('user/message', input, { surfaceOp: 'append' })
-    },
+    inject: () => { throw new Error('step-boundary catalog must not use agent.inject()') },
    cancel() {},
    whenIdle: () => Promise.resolve(),
  }
@@ -200,7 +196,10 @@ describe('dsh-tool-skill', () => {
      content: 'User-only body.',
    })
    ctx.on('agent/step', (agent) => {
-      agent.inject(createUserMessage({ content: [{ type: 'text', text: 'later contribution' }], source: { kind: 'plugin', plugin: 'later-contribution' } }))
+      agent.session.append('user/message', createUserMessage({
+        content: [{ type: 'text', text: 'later contribution' }],
+        source: { kind: 'plugin', plugin: 'later-contribution' },
+      }), { surfaceOp: 'append' })
    })

    const prefix = await composePrefix(ctx, '/workspace')
--- a/packages/ui/jsonrpc/tests/plugin-apply.spec.ts
+++ b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts
@@ -185,7 +185,12 @@ describe('dsh-jsonrpc plugin apply', () => {
        params: { sessionId: 'main', contentBlocks: [{ type: 'text', text: 'fix it' }] },
      })
      const response = await harness.waitForFrame(frame => frame.id === 2, 'prompt response')
-      expect(response.result).toEqual({ accepted: true })
+      expect((response.result as { messageId?: unknown }).messageId).toBeTypeOf('string')
+      await harness.waitForFrame(
+        frame => frame.method === 'session.status'
+          && (frame.params as { status?: string } | undefined)?.status === 'idle',
+        'idle session status',
+      )

      expect(llmServer.requests).toHaveLength(1)
      const body = llmServer.requests[0] as { model: string; messages: { role: string }[] }
@@ -195,9 +200,9 @@ describe('dsh-jsonrpc plugin apply', () => {
      // Notifications use the same transport and arrive as id-less frames.
      const notifications = harness.frames().filter(frame => frame.id === undefined)
      expect(notifications.some(frame => frame.method === 'session.event')).toBe(true)
-      expect(notifications.find(frame => frame.method === 'session.finished')).toMatchObject({
+      expect(notifications.findLast(frame => frame.method === 'session.status')).toMatchObject({
        jsonrpc: '2.0',
-        params: { sessionId: 'main', status: 'ok' },
+        params: { sessionId: 'main', status: 'idle' },
      })
    } finally {
      await harness.dispose()
--- a/packages/ui/jsonrpc/tests/server.spec.ts
+++ b/packages/ui/jsonrpc/tests/server.spec.ts
@@ -8,7 +8,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import AgentRegistry, { type Agent, type AgentHandle } from '@deepseek-ai/dsh-agent'

-import SessionStore, { SessionId, type UserMessage } from '@deepseek-ai/dsh-session'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import * as agentCore from '@deepseek-ai/dsh-agent-spine-demo'
 import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
 import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
@@ -127,12 +127,13 @@ describe('HarnessSdkServer', () => {
      }) as { serverInfo: { name: string } }
      expect(init.serverInfo.name).toBe('deepseek-harness-sdk-runtime')

-      await server.handleRequest('session/prompt', {
+      const receipt = await server.handleRequest('session/prompt', {
        sessionId: 'main',
        contentBlocks: [{ type: 'text', text: 'fix it' }],
      })
+      expect((receipt as { messageId?: unknown }).messageId).toBeTypeOf('string')

-      expect(llmServer.requests).toHaveLength(1)
+      await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(1) })
      const body = llmServer.requests[0] as { model: string; messages: { role: string }[]; max_tokens?: number }
      expect(body.model).toBe('dsagent-model')
      expect(body.max_tokens).toBe(321)
@@ -140,16 +141,18 @@ describe('HarnessSdkServer', () => {
      expect(body.messages.at(-1)?.role).toBe('user')
      expect(llmServer.headers[0]?.authorization).toBe('Bearer test-key')
      expect(transport.notifications.some(n => n.method === 'session.event')).toBe(true)
-      expect(transport.notifications.at(-1)).toMatchObject({
-        method: 'session.finished',
-        params: { sessionId: 'main', status: 'ok' },
+      await vi.waitFor(() => {
+        expect(transport.notifications.findLast(n => n.method === 'session.status')).toEqual({
+          method: 'session.status',
+          params: { sessionId: 'main', status: 'idle' },
+        })
      })

      await server.handleRequest('session/prompt', {
        sessionId: 'main',
        contentBlocks: [{ type: 'text', text: 'again' }],
      })
-      expect(llmServer.requests).toHaveLength(2)
+      await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(2) })

      const orphanHandle = await ctx.agents.create({
        sessionId: SessionId('orphan-session'),
@@ -168,24 +171,17 @@ describe('HarnessSdkServer', () => {
    }
  })

-  it('rejects overlapping prompts for one session without serializing other sessions', async () => {
-    let releaseMain: (() => void) | undefined
-    const firstMainIdle = new Promise<void>((resolve) => { releaseMain = resolve })
-    const mainWhenIdle = vi.fn<() => Promise<void>>()
-      .mockReturnValueOnce(firstMainIdle)
-      .mockResolvedValue(undefined)
+  it('queues overlapping prompts for one session without blocking other sessions', async () => {
    const mainFollowup = vi.fn<Agent['followup']>()
    const mainAgent = ({
      id: SessionId('main'),
      followup: mainFollowup,
-      whenIdle: mainWhenIdle,
-    } satisfies Pick<Agent, 'id' | 'followup' | 'whenIdle'>) as unknown as Agent
+    } satisfies Pick<Agent, 'id' | 'followup'>) as unknown as Agent
    const otherFollowup = vi.fn<Agent['followup']>()
    const otherAgent = ({
      id: SessionId('other'),
      followup: otherFollowup,
-      whenIdle: vi.fn(() => Promise.resolve()),
-    } satisfies Pick<Agent, 'id' | 'followup' | 'whenIdle'>) as unknown as Agent
+    } satisfies Pick<Agent, 'id' | 'followup'>) as unknown as Agent
    const mainHandle = { agent: mainAgent, dispose: vi.fn(() => Promise.resolve()) }
    const otherHandle = { agent: otherAgent, dispose: vi.fn(() => Promise.resolve()) }
    const create = vi.fn(async (options: { sessionId: SessionId }) =>
@@ -202,20 +198,11 @@ describe('HarnessSdkServer', () => {
      contentBlocks: [{ type: 'text', text }],
    })

-    const first = prompt('main', 'first')
-    await vi.waitFor(() => { expect(mainFollowup).toHaveBeenCalledOnce() })
+    expect((await prompt('main', 'first')).messageId).toBeTypeOf('string')
+    expect((await prompt('main', 'overlap')).messageId).toBeTypeOf('string')
+    expect((await prompt('other', 'independent')).messageId).toBeTypeOf('string')

-    await expect(prompt('main', 'overlap')).rejects.toThrow('session already has an active prompt: main')
-    await expect(prompt('other', 'independent')).resolves.toEqual({ accepted: true })
-    releaseMain?.()
-    await expect(first).resolves.toEqual({ accepted: true })
-    await expect(prompt('main', 'sequential')).resolves.toEqual({ accepted: true })
-
-    mainWhenIdle.mockRejectedValueOnce(new Error('turn wait failed'))
-    await expect(prompt('main', 'failing')).rejects.toThrow('turn wait failed')
-    await expect(prompt('main', 'after failure')).resolves.toEqual({ accepted: true })
-
-    expect(mainFollowup).toHaveBeenCalledTimes(4)
+    expect(mainFollowup).toHaveBeenCalledTimes(2)
    expect(otherFollowup).toHaveBeenCalledOnce()
    await server.shutdown()
    expect(mainHandle.dispose).toHaveBeenCalledOnce()
@@ -247,7 +234,7 @@ describe('HarnessSdkServer', () => {
      contentBlocks: [{ type: 'text', text }],
    })

-    await expect(prompt('while live')).resolves.toEqual({ accepted: true })
+    expect((await prompt('while live')).messageId).toBeTypeOf('string')
    live = false
    await expect(prompt('after detach')).rejects.toThrow('session agent was disposed outside the server: zombie')
    // The detached agent was never driven by the rejected prompt.
@@ -255,59 +242,26 @@ describe('HarnessSdkServer', () => {
    await server.shutdown()
  })

-  it('reports the final whole-agent outcome after later activity settles', async () => {
+  it('forwards whole-agent status without attributing a turn outcome', async () => {
    const ctx = new Context()
    await ctx.plugin(SessionStore)
    await ctx.plugin(AgentRegistry)
    const transport = new FakeTransport()
-    const server = new HarnessSdkServer(ctx, transport) as unknown as {
-      prompt(params: { sessionId: string; contentBlocks: { type: 'text'; text: string }[] }): Promise<unknown>
-      sessions: Map<string, { handle: AgentHandle; lastTurnEnd: undefined; activePrompt: boolean }>
-      shutdown(): Promise<Record<string, never>>
-    }
+    const server = new HarnessSdkServer(ctx, transport)
    const session = ctx.sessions.create(SessionId('message-outcome'))
    const agent = ({
      id: SessionId('message-outcome'),
      session,
-      followup(input: UserMessage) {
-        session.append('turn/start', {
-          turn: 1,
-        })
-        session.append('user/message', input, { surfaceOp: 'append' })
-        session.append('turn/end', { turn: 1, reason: { kind: 'max-tokens' } })
-        session.append('turn/start', {
-          turn: 2,
-        })
-        session.append('user/message', createUserMessage({
-          content: [{ type: 'text', text: 'late metadata' }],
-          source: { kind: 'plugin', plugin: 'late-metadata' },
-        }), { surfaceOp: 'append' })
-        session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
-        return input.id
-      },
-      whenIdle: () => Promise.resolve(),
-    } satisfies Pick<Agent, 'id' | 'session' | 'followup' | 'whenIdle'>) as unknown as Agent
-    ctx.agents.register(agent)
-    server.sessions.set('message-outcome', {
-      handle: { agent, dispose: () => Promise.resolve() },
-      lastTurnEnd: undefined,
-      activePrompt: false,
-    })
+    } satisfies Pick<Agent, 'id' | 'session'>) as Agent

-    await server.prompt({
-      sessionId: 'message-outcome',
-      contentBlocks: [{ type: 'text', text: 'bounded prompt' }],
-    })
+    ctx.emit('agent/status', agent, 'running')
+    ctx.emit('agent/status', agent, 'idle')

-    expect(transport.notifications.findLast(notification => notification.method === 'session.finished'))
-      .toEqual({
-        method: 'session.finished',
-        params: {
-          sessionId: 'message-outcome',
-          status: 'ok',
-          reason: { kind: 'completed' },
-        },
-      })
+    expect(transport.notifications.filter(notification => notification.method === 'session.status'))
+      .toEqual([
+        { method: 'session.status', params: { sessionId: 'message-outcome', status: 'running' } },
+        { method: 'session.status', params: { sessionId: 'message-outcome', status: 'idle' } },
+      ])
    await server.shutdown()
    await ctx.fiber.dispose()
  })
@@ -356,7 +310,7 @@ describe('HarnessSdkServer', () => {
        contentBlocks: [{ type: 'text', text: 'hello' }],
      })

-      expect(llmServer.requests).toHaveLength(1)
+      await vi.waitFor(() => { expect(llmServer.requests).toHaveLength(1) })
      await server.shutdown()
    } finally {
      await ctx.fiber.dispose()
@@ -881,43 +835,6 @@ describe('HarnessSdkServer', () => {
    },
  )

-  it('classifies defensive finish states', async () => {
-    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-finish-states-'))
-    const ctx = await makeHarness(storageDir)
-    try {
-      const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
-        finishedStatus(reason: unknown): 'ok' | 'error'
-        shutdown(): Promise<Record<string, never>>
-      }
-
-      expect(server.finishedStatus(undefined)).toBe('error')
-      expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('error')
-      expect(server.finishedStatus({ kind: 'error' })).toBe('error')
-      await server.shutdown()
-    } finally {
-      await ctx.fiber.dispose()
-      await rm(storageDir, { recursive: true, force: true })
-    }
-  })
-
-  it('can report max-token turn termination as an accepted evaluation result', async () => {
-    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-max-tokens-success-'))
-    const ctx = await makeHarness(storageDir)
-    try {
-      const server = new HarnessSdkServer(ctx, new FakeTransport(), { maxTokensAsSuccess: true }) as unknown as {
-        finishedStatus(reason: unknown): 'ok' | 'error'
-        shutdown(): Promise<Record<string, never>>
-      }
-
-      expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('ok')
-      expect(server.finishedStatus({ kind: 'error' })).toBe('error')
-      await server.shutdown()
-    } finally {
-      await ctx.fiber.dispose()
-      await rm(storageDir, { recursive: true, force: true })
-    }
-  })
-
  it('reports no adapter when the LLM service is absent', async () => {
    const ctx = new Context()
    try {
@@ -1045,6 +962,6 @@ describe('HarnessSdkServer', () => {
    const server = new HarnessSdkServer(ctx, new FakeTransport())

    await expect(server.shutdown()).rejects.toBe(listenerFailure)
-    expect(on).toHaveBeenCalledTimes(3)
+    expect(on).toHaveBeenCalledTimes(4)
  })
 })
--- a/packages/ui/user-approval/src/index.ts
+++ b/packages/ui/user-approval/src/index.ts
@@ -277,10 +277,10 @@ export class ApprovalService extends Service {
      const cause = overrideSource === 'delegation'
        ? 'inherited from the delegating session'
        : overrideIndex > headerIndex ? 'changed by the user' : 'changed by the operator/config'
-      agent.inject(createUserMessage({
+      session.append('user/message', createUserMessage({
        content: [{ type: 'text', text: `The approval policy changed from "${told}" to "${current}" (${cause}).` }],
        source: { kind: 'plugin', plugin: 'user-approval' },
-      }))
+      }), { surfaceOp: 'append' })
    })
  }

--- a/packages/ui/user-approval/tests/approval.spec.ts
+++ b/packages/ui/user-approval/tests/approval.spec.ts
@@ -358,23 +358,27 @@ describe('approval policy (the approval/policy fold)', () => {
   * An agent stand-in over a REAL Session — gate, section, and narrator fold
   * real events; the opened turn satisfies request()'s enclosure precondition.
   */
-  function sessionAgent(id: string): { agent: Agent; session: Session; injected: string[] } {
+  function sessionAgent(id: string): { agent: Agent; session: Session } {
    const session = new Session(SessionId(id))
    session.append('turn/start', { turn: 1 })
-    const injected: string[] = []
    const agent = {
      id,
      session,
-      inject: (input: { content: Array<{ type: string; text: string }> }) => {
-        injected.push(input.content[0]?.text ?? '')
-      },
+      inject: () => { throw new Error('step-boundary narration must not use agent.inject()') },
    } as unknown as Agent
-    return { agent, session, injected }
+    return { agent, session }
  }

  const preStep = (ctx: Context, agent: Agent): Promise<void> =>
    agentEvents(ctx, agent).serial('agent/step', 1, 1, new AbortController().signal)

+  const narrations = (session: Session): string[] => session.events.flatMap(event =>
+    event.type === 'user/message'
+      && event.data.source.kind === 'plugin'
+      && event.data.source.plugin === 'user-approval'
+      ? [event.data.content.flatMap(block => block.type === 'text' ? [block.text] : []).join('')]
+      : [])
+
  /** Append a `request/header` snapshot whose system text is exactly `system`. */
  function appendHeader(session: Session, system: string): void {
    session.append('request/header', { header: { config: { provider: 'mock', model: 'mock' }, system }, reason: 'initial' })
@@ -482,20 +486,20 @@ describe('approval policy (the approval/policy fold)', () => {
  it('narrates nothing cold, once per coalesced switch (user wording), and idempotently', async () => {
    const ctx = new Context()
    await ctx.plugin(ApprovalService)
-    const { agent, session, injected } = sessionAgent('sess-narr-1')
+    const { agent, session } = sessionAgent('sess-narr-1')
    await preStep(ctx, agent)
-    expect(injected).toEqual([])
+    expect(narrations(session)).toEqual([])
    setApprovalPolicy(session, 'never')
    setApprovalPolicy(session, 'ask')
    setApprovalPolicy(session, 'never')
    await preStep(ctx, agent)
-    expect(injected).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).'])
+    expect(narrations(session)).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).'])
    await preStep(ctx, agent)
-    expect(injected).toHaveLength(1)
+    expect(narrations(session)).toHaveLength(1)
    setApprovalPolicy(session, 'ask')
    setApprovalPolicy(session, 'never')
    await preStep(ctx, agent)
-    expect(injected).toHaveLength(1)
+    expect(narrations(session)).toHaveLength(1)
  })

  it('reads what the model was told back from the folded header text after a restart', async () => {
@@ -503,69 +507,69 @@ describe('approval policy (the approval/policy fold)', () => {
    // an ask default: the narrator attributes the change to the operator.
    const ctx = new Context()
    await ctx.plugin(ApprovalService)
-    const { agent, session, injected } = sessionAgent('sess-narr-2')
+    const { agent, session } = sessionAgent('sess-narr-2')
    appendHeader(session, `persona\n\n${NEVER_SENTENCE}\n${NEVER_MARKER}`)
    await preStep(ctx, agent)
-    expect(injected).toEqual(['The approval policy changed from "never" to "ask" (changed by the operator/config).'])
+    expect(narrations(session)).toEqual(['The approval policy changed from "never" to "ask" (changed by the operator/config).'])
  })

  it('attributes a constructor-seeded policy event to delegation', async () => {
    const ctx = new Context()
    await ctx.plugin(ApprovalService)
-    const { agent, session, injected } = sessionAgent('sess-narr-inherited')
+    const { agent, session } = sessionAgent('sess-narr-inherited')
    appendHeader(session, ASK_MARKER)
    session.append('approval/policy', { policy: 'never', source: 'delegation' })

    await preStep(ctx, agent)

-    expect(injected).toEqual(['The approval policy changed from "ask" to "never" (inherited from the delegating session).'])
+    expect(narrations(session)).toEqual(['The approval policy changed from "ask" to "never" (inherited from the delegating session).'])
  })

  it('narrates a config default drift from the logged ask marker', async () => {
    const ctx = new Context()
    await ctx.plugin(ApprovalService, { policy: 'never' })
-    const { agent, session, injected } = sessionAgent('sess-narr-3')
+    const { agent, session } = sessionAgent('sess-narr-3')
    appendHeader(session, `persona only\n${ASK_MARKER}`)
    await preStep(ctx, agent)
-    expect(injected).toEqual(['The approval policy changed from "ask" to "never" (changed by the operator/config).'])
+    expect(narrations(session)).toEqual(['The approval policy changed from "ask" to "never" (changed by the operator/config).'])
  })

  it('a pinned override survives a default change silently', async () => {
    const ctx = new Context()
    await ctx.plugin(ApprovalService, { policy: 'never' })
-    const { agent, session, injected } = sessionAgent('sess-narr-4')
+    const { agent, session } = sessionAgent('sess-narr-4')
    appendHeader(session, `persona only\n${ASK_MARKER}`)
    setApprovalPolicy(session, 'ask')
    appendHeader(session, `persona only\n${ASK_MARKER}`)
    await preStep(ctx, agent)
-    expect(injected).toEqual([])
+    expect(narrations(session)).toEqual([])
  })

  it('does not infer never from deployment prose that quotes the never sentence', async () => {
    const ctx = new Context()
    await ctx.plugin(ApprovalService)
-    const { agent, session, injected } = sessionAgent('sess-narr-spoof-prose')
+    const { agent, session } = sessionAgent('sess-narr-spoof-prose')
    appendHeader(session, `persona quotes this warning: ${NEVER_SENTENCE}\n${ASK_MARKER}`)
    await preStep(ctx, agent)
-    expect(injected).toEqual([])
+    expect(narrations(session)).toEqual([])
  })

  it('treats a legacy header with no source-owned marker as untold', async () => {
    const ctx = new Context()
    await ctx.plugin(ApprovalService, { policy: 'never' })
-    const { agent, session, injected } = sessionAgent('sess-narr-unmarked-header')
+    const { agent, session } = sessionAgent('sess-narr-unmarked-header')
    appendHeader(session, 'legacy persona-only header')
    await preStep(ctx, agent)
-    expect(injected).toEqual([])
+    expect(narrations(session)).toEqual([])
  })

  it('uses the service marker after an earlier persona marker', async () => {
    const ctx = new Context()
    await ctx.plugin(ApprovalService)
-    const { agent, session, injected } = sessionAgent('sess-narr-spoof-marker')
+    const { agent, session } = sessionAgent('sess-narr-spoof-marker')
    appendHeader(session, `persona quotes ${NEVER_MARKER}\n${ASK_MARKER}`)
    await preStep(ctx, agent)
-    expect(injected).toEqual([])
+    expect(narrations(session)).toEqual([])
  })

  it('disposes the service prompt section and pre-step narrator together (HMR safety)', async () => {
@@ -581,7 +585,7 @@ describe('approval policy (the approval/policy fold)', () => {
    appendHeader(live.session, `persona\n${ASK_MARKER}`)
    setApprovalPolicy(live.session, 'never')
    await preStep(ctx, live.agent)
-    expect(live.injected).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).'])
+    expect(narrations(live.session)).toEqual(['The approval policy changed from "ask" to "never" (changed by the user).'])

    appendHeader(afterDispose.session, `persona\n${ASK_MARKER}`)
    setApprovalPolicy(afterDispose.session, 'never')
@@ -589,6 +593,6 @@ describe('approval policy (the approval/policy fold)', () => {

    expect(await sectionFor()).toBeUndefined()
    await preStep(ctx, afterDispose.agent)
-    expect(afterDispose.injected).toEqual([])
+    expect(narrations(afterDispose.session)).toEqual([])
  })
 })