test: close per-file coverage gaps opened by the message-machine refactor

Downstream packages lost the tests that exercised their agent-loop-facing
edges when the loop was rewritten. Restore 100% per-file coverage with
behavior tests through public seams: llm-retry config validation and
cancellation races, goal replay drift/staleness/teardown edges, plan-mode
disposed-flush, workspace-context empty-change commits, api-proxy
synchronous send failures, acp-snapshot spill-path extraction and refresh
write-back, ACP injection-triggered turns, cli-demo and tui inbox
lifecycle edges, and agent-loop retry/settlement/lifecycle branches.
The only source changes are narrowly-justified v8 ignore annotations on
invariant guards and one redundant-guard removal (workspace-context).

packages/context/workspace-context/src/state.ts

@@ -117,8 +117,7 @@ function isRecord(value: unknown): value is Record<string, unknown> {
   return typeof value === 'object' && value !== null && !Array.isArray(value)
 }
 
-function workspaceInstructionChanges(source: unknown): WorkspaceInstructionChange[] {
-  if (!isWorkspaceContextSource(source)) return []
+function workspaceInstructionChanges(source: { changes: unknown[] }): WorkspaceInstructionChange[] {
   const changes: WorkspaceInstructionChange[] = []
   for (const value of source.changes) {
     if (!isRecord(value)) continue

packages/context/workspace-context/tests/workspace-context.spec.ts

@@ -3260,6 +3260,29 @@ describe('workspace context pending state', () => {
     expect(versions.has(agent.session)).toBe(false)
   })
 
+  it('keeps an unrelated scope\'s version fast path when a step-close discard empties only its own scope', () => {
+    const agent = stubAgent('/')
+    const pending = new WeakMap<object, Map<string, PendingInstructionChange>>()
+    const versions: InstructionVersionCache = new WeakMap()
+    agent.session.append('step/start', { turn: 1, step: 1 })
+    commitPendingInstructionContexts(agent, [workspaceChangeContext('pkg', 'one')], pending)
+    versions.set(agent.session, new Map([
+      ['pkg', {
+        path: join('pkg', 'AGENTS.md'), version: FsVersion('v1'), digest: 'one', trimmedDigest: 'one',
+      }],
+      ['other', {
+        path: join('other', 'AGENTS.md'), version: FsVersion('v2'), digest: 'two', trimmedDigest: 'two',
+      }],
+    ]))
+
+    const ended = agent.session.append('step/end', { turn: 1, step: 1 })
+    observeInstructionSessionEvent(agent.session, ended, pending, versions)
+
+    expect(pending.has(agent.session)).toBe(false)
+    expect(versions.get(agent.session)?.has('pkg')).toBe(false)
+    expect(versions.get(agent.session)?.has('other')).toBe(true)
+  })
+
   it('rolls back only the exact current transition and releases empty session state', () => {
     const agent = stubAgent('/')
     const pending = new WeakMap<object, Map<string, PendingInstructionChange>>()
@@ -3270,6 +3293,13 @@ describe('workspace context pending state', () => {
     expect(commitPendingInstructionContexts(agent, [{
       content: [], source: { kind: 'plugin', plugin: 'workspace-context' },
     }], pending)).toEqual([])
+    // A workspace-instructions source whose change list filters to nothing
+    // must not mint per-session pending state.
+    expect(commitPendingInstructionContexts(agent, [{
+      content: [],
+      source: { kind: 'workspace-instructions', changes: [] },
+    }], pending)).toEqual([])
+    expect(pending.has(agent.session)).toBe(false)
 
     const committed = commitPendingInstructionContexts(agent, [
       workspaceChangeContext('first', 'one'),

packages/core/agent-loop/tests/config-session-id.spec.ts

@@ -441,3 +441,36 @@ describe('config-driven session id', () => {
     await ctx.fiber.dispose()
   })
 })
+
+describe('startup reporting after factory teardown', () => {
+  it('suppresses the configured-restore failure report once the loop is disposed', async () => {
+    const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-disposed-report-'))
+    dirs.push(root)
+    const ctx = await makeCoreContext()
+    await ctx.plugin(SessionPersistenceJsonl, { root })
+    ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('x')]))
+
+    // A restore lookup that hangs until after the loop is gone: the eventual
+    // failure lands with ownership inactive and must be silently dropped.
+    const gate = Promise.withResolvers<never>()
+    // The teardown path may drop the pending lookup without awaiting it.
+    gate.promise.catch(() => undefined)
+    vi.spyOn(ctx.sessionPersistence, 'list').mockReturnValue(gate.promise)
+    const failures: unknown[] = []
+    ctx.on('agent-loop/config-start-failed', (_id, error) => { failures.push(error) })
+    const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
+
+    const loop = await ctx.plugin(AgentLoop, {
+      agents: [{ id: 'main', sessionId: SessionId('config-disposed-report'), model: 'mock' }],
+    })
+    const disposal = loop.dispose()
+    gate.reject(new Error('backend failed after teardown began'))
+    await disposal
+
+    await new Promise(r => setTimeout(r, 20))
+    expect(failures).toEqual([])
+    expect(warn).not.toHaveBeenCalledWith(expect.stringContaining('config-driven restore'))
+    warn.mockRestore()
+    await ctx.fiber.dispose()
+  })
+})

packages/core/agent-loop/tests/coverage-edges.spec.ts

@@ -254,3 +254,247 @@ describe('structured tool error propagation (the runtime-validation Agent Note,
       .toEqual({ name: 'HarnessError', code: 'BOOM' })
   })
 })
+
+describe('retry() edges', () => {
+  it('throws while a turn runs with no request-error window open', async () => {
+    const adapter = new MockAdapter(['hang'])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('retry-busy'), { provider: 'mock', model: 'mock' })
+
+    send(agent, 'go')
+    // Wait until the hung request is in flight (the run owns this.abort).
+    await new Promise(r => setTimeout(r, 30))
+    expect(() => { agent.retry() }).toThrow('cannot retry while busy')
+    agent.cancel({ kind: 'user' })
+    await agent.whenIdle()
+  })
+
+  it('ignores a retry request arriving after the recovery window was aborted', async () => {
+    const { LlmError } = await import('@deepseek-ai/dsh-llm')
+    const adapter = new MockAdapter([
+      () => { throw new LlmError('busy', 'RATE_LIMIT') },
+      textResponse('never used'),
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('retry-after-cancel'), { provider: 'mock', model: 'mock' })
+    ctx.on('agent/request-error', async (subject) => {
+      // Cancellation lands first; the window survives structurally but its
+      // signal is aborted, so the request must not arm a retry turn.
+      subject.cancel({ kind: 'user' })
+      subject.retry()
+    })
+
+    send(agent, 'go')
+    await agent.whenIdle()
+
+    // One failed request, no retry turn.
+    expect(adapter.requests).toHaveLength(1)
+    const ends = agent.session.events.filter(e => e.type === 'turn/end')
+    expect(ends).toHaveLength(1)
+  })
+
+  it('completed recovery does not retry when cancellation raced the waterfall', async () => {
+    const { LlmError } = await import('@deepseek-ai/dsh-llm')
+    const adapter = new MockAdapter([
+      () => { throw new LlmError('busy', 'RATE_LIMIT') },
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('retry-raced'), { provider: 'mock', model: 'mock' })
+    ctx.on('agent/request-error', async (subject, _turn, _step, _error, _failure, signal, next) => {
+      await next()
+      // Recovery completes and requested the retry, but the turn signal
+      // aborts before the loop reads the window.
+      subject.retry()
+      subject.cancel({ kind: 'user' })
+      expect(signal.aborted).toBe(true)
+    })
+
+    send(agent, 'go')
+    await agent.whenIdle()
+
+    expect(adapter.requests).toHaveLength(1)
+    const end = agent.session.events.findLast(e => e.type === 'turn/end')
+    expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('aborted')
+  })
+})
+
+describe('stream failure edges', () => {
+  it('rethrows a mid-stream throw that carries no adapter failure facts', async () => {
+    const adapter = new MockAdapter([textResponse('will be vetoed')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('stream-no-facts'), { provider: 'mock', model: 'mock' })
+    let recoveries = 0
+    ctx.on('agent/request-error', async () => { recoveries += 1 })
+    // A pre-commit chunk veto throws INSIDE the stream-consumption try, but it
+    // is not an adapter-boundary failure, so llmFailureOf yields no facts.
+    let vetoed = false
+    ctx.on('internal/dispatch', (_mode, name, args) => {
+      if (name !== 'session/event') return
+      const event = args[1] as SessionEvent
+      if (event.type === 'assistant/chunk' && !vetoed) {
+        vetoed = true
+        throw new Error('reject the first chunk')
+      }
+    })
+
+    send(agent, 'go')
+    await agent.whenIdle()
+
+    // No facts -> not offered to recovery; the turn fails through settle().
+    expect(recoveries).toBe(0)
+    const end = agent.session.events.findLast(e => e.type === 'turn/end')
+    expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error')
+  })
+})
+
+describe('post-turn continuation edges', () => {
+  it('an agent/idle listener that enqueues a waking prompt preempts continueOrIdle', async () => {
+    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('idle-preempt'), { provider: 'mock', model: 'mock' })
+    let injected = false
+    ctx.on('agent/idle', (subject) => {
+      if (subject !== agent || injected) return
+      injected = true
+      // kick() installs the next admission synchronously, so the following
+      // continueOrIdle() sees an abort owner and yields to it.
+      send(agent, 'follow-up from idle listener')
+    })
+
+    send(agent, 'go')
+    await agent.whenIdle()
+
+    expect(adapter.requests).toHaveLength(2)
+    const starts = agent.session.events.filter(e => e.type === 'turn/start')
+    expect(starts).toHaveLength(2)
+    // The busy interval never broke between the turns: one running->idle cycle.
+  })
+
+  it('whenIdle resolves for a waiter whose awaited run fails', async () => {
+    const adapter = new MockAdapter([textResponse('unused')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('whenidle-reject'), { provider: 'mock', model: 'mock' })
+    let rejected = false
+    ctx.on('internal/dispatch', (_mode, name, args) => {
+      if (name !== 'session/event') return
+      const event = args[1] as SessionEvent
+      if (event.type === 'turn/start' && !rejected) {
+        rejected = true
+        throw new Error('veto turn start while a waiter is pending')
+      }
+    })
+
+    send(agent, 'go')
+    await expect(agent.whenIdle()).resolves.toBeUndefined()
+    expect(agent.status).toBe('idle')
+  })
+})
+
+describe('tool result meta persistence', () => {
+  it('records a presentationMeta payload on the tool/result event', async () => {
+    const { defineTool } = await import('@deepseek-ai/dsh-tools')
+    const adapter = new MockAdapter([
+      toolCallResponse('c1', 'meta-tool', {}),
+      textResponse('done'),
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('tool-meta'), { provider: 'mock', model: 'mock' })
+    ctx.tools.register(defineTool({
+      name: 'meta-tool',
+      description: 'carries presentation meta',
+      parameters: {},
+      output: {
+        schema: { type: 'string' },
+        render: (_args, value) => [{ type: 'text', text: value }],
+        presentationMeta: () => ({ presentation: 'diff-card' }),
+      },
+      async execute() {
+        return 'ran'
+      },
+    }))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    const result = agent.session.events.find(e => e.type === 'tool/result')
+    expect(result?.type === 'tool/result' && result.data.meta).toEqual({ presentation: 'diff-card' })
+  })
+})
+
+describe('turn close failure containment', () => {
+  it('a rejected turn/end append is contained: warn + agent/error, no retry', async () => {
+    const adapter = new MockAdapter([textResponse('ok')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('turnend-veto'), { provider: 'mock', model: 'mock' })
+    let vetoed = false
+    ctx.on('internal/dispatch', (_mode, name, args) => {
+      if (name !== 'session/event') return
+      const event = args[1] as SessionEvent
+      if (event.type === 'turn/end' && !vetoed) {
+        vetoed = true
+        throw new Error('reject turn end')
+      }
+    })
+    const errors: unknown[] = []
+    ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) })
+
+    send(agent, 'go')
+    await agent.whenIdle()
+
+    // The close failure is reported live; the machine still reaches idle.
+    expect(errors.map(e => e instanceof Error && e.message)).toContain('reject turn end')
+    expect(agent.status).toBe('idle')
+    expect(adapter.requests).toHaveLength(1)
+  })
+})
+
+describe('recovery without a retry request', () => {
+  it('a completed recovery that never calls retry() leaves the failed turn terminal', async () => {
+    const { LlmError } = await import('@deepseek-ai/dsh-llm')
+    const adapter = new MockAdapter([
+      () => { throw new LlmError('down', 'SERVICE_UNAVAILABLE') },
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('recovery-no-retry'), { provider: 'mock', model: 'mock' })
+    let recoveries = 0
+    ctx.on('agent/request-error', async () => { recoveries += 1 })
+
+    send(agent, 'go')
+    await agent.whenIdle()
+
+    expect(recoveries).toBe(1)
+    expect(adapter.requests).toHaveLength(1)
+    const end = agent.session.events.findLast(e => e.type === 'turn/end')
+    expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error')
+  })
+})
+
+describe('unrenderable failure settlement', () => {
+  it('drops the rendered message when the error chain cannot be rendered', async () => {
+    const { LlmError } = await import('@deepseek-ai/dsh-llm')
+    const adapter = new MockAdapter([
+      () => {
+        const error = new LlmError('will become hostile', 'SERVER')
+        // A hostile message getter makes errorChain collapse to its sentinel;
+        // settle() must then fall back to the failure facts alone.
+        Object.defineProperty(error, 'message', {
+          get() { throw new Error('hostile accessor') },
+        })
+        throw error
+      },
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('unrenderable'), { provider: 'mock', model: 'mock' })
+
+    send(agent, 'go')
+    await agent.whenIdle()
+
+    const end = agent.session.events.findLast(e => e.type === 'turn/end')
+    expect(end?.type === 'turn/end' && end.data.reason.kind).toBe('error')
+    if (end?.type === 'turn/end' && end.data.reason.kind === 'error') {
+      // The durable failure keeps the adapter facts' message, not the
+      // unrenderable chain.
+      expect(end.data.reason.failure?.message).not.toBe('<unrenderable value>')
+    }
+  })
+})

packages/core/agent-loop/tests/resume.spec.ts

@@ -555,3 +555,97 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
     await ctx.fiber.dispose()
   })
 })
+
+describe('creation and resume cancellation edges', () => {
+  it('rejects create() with a pre-aborted signal, including a non-Error reason', async () => {
+    const { ctx } = await persistentHarness(new MockAdapter([]))
+
+    const errorReason = new AbortController()
+    errorReason.abort(new Error('caller gave up'))
+    await expect(promptly(ctx.agents.create({
+      sessionId: SessionId('pre-aborted-error'),
+      agentOptions: { provider: 'mock', model: 'mock' },
+      signal: errorReason.signal,
+    }))).rejects.toThrow('caller gave up')
+
+    // A non-Error reason is wrapped into the creation-aborted error.
+    const stringReason = new AbortController()
+    stringReason.abort('operator string reason')
+    await expect(promptly(ctx.agents.create({
+      sessionId: SessionId('pre-aborted-string'),
+      agentOptions: { provider: 'mock', model: 'mock' },
+      signal: stringReason.signal,
+    }))).rejects.toThrow(/creation aborted/)
+
+    expect(ctx.agents.get(SessionId('pre-aborted-error'))).toBeUndefined()
+    expect(ctx.agents.get(SessionId('pre-aborted-string'))).toBeUndefined()
+    await ctx.fiber.dispose()
+  })
+
+  it('a non-Error abort reason arriving during setup is wrapped for the caller', async () => {
+    const { ctx } = await persistentHarness(new MockAdapter([]))
+    const controller = new AbortController()
+    const setupEntered = Promise.withResolvers<undefined>()
+    const setupGate = Promise.withResolvers<undefined>()
+
+    const creating = ctx.agents.create({
+      sessionId: SessionId('setup-string-abort'),
+      agentOptions: { provider: 'mock', model: 'mock' },
+      signal: controller.signal,
+      async setup() {
+        setupEntered.resolve(undefined)
+        await setupGate.promise
+      },
+    })
+    await setupEntered.promise
+    controller.abort('mid-setup string reason')
+    setupGate.resolve(undefined)
+
+    await expect(promptly(creating)).rejects.toThrow(/creation aborted/)
+    expect(ctx.agents.get(SessionId('setup-string-abort'))).toBeUndefined()
+    await ctx.fiber.dispose()
+  })
+
+  it('resume with a pre-aborted caller signal rejects out of the load race', async () => {
+    const sessionId = SessionId('resume-pre-aborted')
+    const root = await persistSession(sessionId)
+    const ctx = await mountPersistentHarness(root, new MockAdapter([]))
+    const controller = new AbortController()
+    controller.abort(new Error('resume abandoned'))
+
+    await expect(promptly(ctx.agents.resume({
+      resumeSessionId: sessionId,
+      agentOptions: { provider: 'mock', model: 'mock' },
+      signal: controller.signal,
+    }))).rejects.toThrow('resume abandoned')
+
+    expect(ctx.agents.get(sessionId)).toBeUndefined()
+    await ctx.fiber.dispose()
+  })
+
+  it('factory teardown during a hung resume load rejects with loop-inactive', async () => {
+    const sessionId = SessionId('resume-loop-teardown')
+    const root = await persistSession(sessionId)
+    const ctx = await mountPersistentHarness(root, new MockAdapter([]))
+    const snapshot = await ctx.sessionPersistence.load(sessionId)
+    const gate = Promise.withResolvers<typeof snapshot>()
+    const loadStarted = Promise.withResolvers<undefined>()
+    ctx.sessionPersistence.load = () => {
+      loadStarted.resolve(undefined)
+      return gate.promise
+    }
+
+    const resuming = ctx.agents.resume({
+      resumeSessionId: sessionId,
+      agentOptions: { provider: 'mock', model: 'mock' },
+    })
+    await loadStarted.promise
+    // Resolve the load only after teardown began: the post-load ownership
+    // check, not the abort race, must reject the wrapper.
+    const rejection = expect(promptly(resuming)).rejects.toThrow()
+    const disposal = ctx.fiber.dispose()
+    gate.resolve(structuredClone(snapshot))
+    await rejection
+    await disposal
+  })
+})

packages/examples/cli-demo/src/cli.ts

@@ -241,6 +241,10 @@ export async function runOneShot(ctx: Context, options: OneShotOptions): Promise
     resolveTurn()
   }
   const settleRejected = (error: Error): void => {
+    // The once-registered abort listener is the only rejecter, and a settled
+    // prompt makes targetTurn defined so onAbort skips rejection entirely;
+    // kept for symmetry with settleResolved.
+    /* v8 ignore next -- unreachable second settlement, see above */
     if (firstTurnEnded) return
     firstTurnEnded = true
     rejectTurn(error)

packages/goal/goal-session/tests/goal-session.spec.ts

@@ -1,6 +1,6 @@
 import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
-import type { Agent } from '@deepseek-ai/dsh-agent'
+import type { Agent, PromptDecision } from '@deepseek-ai/dsh-agent'
 import { agentEvents } from '@deepseek-ai/dsh-agent'
 import AgentLoop from '@deepseek-ai/dsh-agent-loop'
 import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
@@ -484,6 +484,22 @@ describe('same-session goal driving', () => {
     expect(test.adapter.requests).toHaveLength(0)
   })
 
+  it('contains a mutation failure inside the scheduler loop and fails closed', async () => {
+    const test = await harness([textResponse('the only round')])
+    // The only ctx.goals.block call in a completing one-round run is the
+    // driver's round-limit stop, so the mock fails exactly that drive pass.
+    vi.spyOn(test.ctx.goals, 'block').mockImplementationOnce(() => {
+      throw new Error('round-limit block failed')
+    })
+    test.ctx.goals.create(test.agent, { objective: 'contain a driver failure', maxGoalRounds: 1 })
+
+    const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
+
+    expect(goal).toMatchObject({ phase: 'active', roundsStarted: 1 })
+    expect(goal?.blockedReason).toBeUndefined()
+    expect(test.adapter.requests).toHaveLength(1)
+  })
+
   it('contains synchronous scheduler startup failure', async () => {
     const test = await harness([])
     vi.spyOn(test.ctx.agents, 'withoutInitiator').mockImplementationOnce(() => {
@@ -674,6 +690,152 @@ describe('same-session goal driving', () => {
     expect(test.adapter.requests).toHaveLength(1)
   })
 
+  it('leaves a queued reservation pending when the driver runs before its turn settles', async () => {
+    const test = await harness([textResponse('settled later')])
+    let woken = false
+    test.ctx.on('agent/prompt-submit', async (_agent, _content, source, _signal, next) => {
+      if (source.kind === 'goal' && !woken) {
+        woken = true
+        // A concurrent driver pass must observe the still-unsettled attempt
+        // and yield rather than double-book or clear the reservation.
+        agentEvents(test.ctx, test.agent).emit('agent/status', 'idle')
+        await new Promise<void>((resolve) => { setImmediate(resolve) })
+      }
+      return next()
+    })
+    test.ctx.goals.create(test.agent, { objective: 'wake early', maxGoalRounds: 1 })
+
+    const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
+
+    expect(goal?.blockedReason?.code).toBe('round-limit')
+    expect(goal?.roundsStarted).toBe(1)
+    expect(test.adapter.requests).toHaveLength(1)
+  })
+
+  it('disarms instead of continuing when a plugin reports a post-turn persistence failure', async () => {
+    const test = await harness([textResponse('round one')])
+    test.ctx.on('session/event', (session, event) => {
+      if (session === test.agent.session && event.type === 'turn/end') {
+        agentEvents(test.ctx, test.agent).emit('agent/error', event.data.turn, 1, new Error('post-turn flush failed'))
+      }
+    })
+    test.ctx.goals.create(test.agent, { objective: 'stop when durability is lost', maxGoalRounds: 8 })
+
+    const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
+    await test.agent.whenIdle()
+
+    expect(goal).toMatchObject({ phase: 'active', roundsStarted: 1 })
+    expect(test.adapter.requests).toHaveLength(1)
+  })
+
+  it('ignores a post-turn failure reported for a retired agent', async () => {
+    const test = await harness([textResponse('ordinary work')])
+    const handle = await test.ctx.agents.create({
+      sessionId: SessionId('goal-session-retired'),
+      agentOptions: { provider: 'mock', model: 'mock' },
+    })
+    handle.agent.followup({ content: [{ type: 'text', text: 'one ordinary turn' }], source: { kind: 'user' } })
+    await handle.agent.whenIdle()
+    const closed = handle.agent.session.events.findLast(event => event.type === 'turn/end')
+    if (closed?.type !== 'turn/end') throw new Error('expected a closed turn')
+    await handle.dispose()
+    const warn = vi.spyOn(test.ctx.logger, 'warn')
+
+    agentEvents(test.ctx, handle.agent).emit('agent/error', closed.data.turn, 1, new Error('late flush failure'))
+
+    expect(test.ctx.agents.get(handle.agent.id)).toBeUndefined()
+    expect(warn).not.toHaveBeenCalledWith(expect.stringContaining('goal-session'))
+  })
+
+  it('ignores the failed outcome of a round made stale by human work queued at turn start', async () => {
+    const test = await harness([new Error('round one broke'), textResponse('human answer')])
+    let queued = false
+    test.ctx.on('session/event', (session, event) => {
+      if (session !== test.agent.session || queued) return
+      if (event.type === 'turn/start' && event.data.trigger.kind === 'message'
+        && event.data.trigger.source.kind === 'goal') {
+        queued = true
+        test.agent.followup({ content: [{ type: 'text', text: 'human interleaved' }], source: { kind: 'user' } })
+      }
+    })
+    test.ctx.goals.create(test.agent, { objective: 'survive a stale failure', maxGoalRounds: 1 })
+
+    const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')
+
+    // The stale round's turn-error never blocks the goal; only the durable
+    // round budget does, after the interleaved human turn ran.
+    expect(goal?.blockedReason?.code).toBe('round-limit')
+    expect(test.adapter.requests).toHaveLength(2)
+    expect(requestText(test.adapter.requests[1]!)).toContain('human interleaved')
+  })
+
+  it('waits for work queued by a pause observer before considering the next round', async () => {
+    const test = await harness(['hang', textResponse('inspection answer')])
+    test.ctx.on('goal/changed', (agent, change) => {
+      if (agent === test.agent && change.operation === 'pause') {
+        agent.followup({ content: [{ type: 'text', text: 'inspect the pause' }], source: { kind: 'user' } })
+      }
+    })
+    test.ctx.goals.create(test.agent, { objective: 'pause then inspect' })
+    await waitForRequests(test.adapter, 1)
+
+    test.agent.cancel({ kind: 'user' })
+    await waitForRequests(test.adapter, 2)
+    await test.agent.whenIdle()
+
+    expect(test.ctx.goals.get(test.agent)).toMatchObject({
+      phase: 'paused',
+      roundsStarted: 1,
+      activation: 'disarmed',
+    })
+    expect(requestText(test.adapter.requests[1]!)).toContain('inspect the pause')
+  })
+
+  it('does not re-block a goal the downstream veto already saw cancelled', async () => {
+    const test = await harness([])
+    let vetoed = false
+    test.ctx.on('agent/prompt-submit', (agent, _content, source, _signal, next) => {
+      if (source.kind === 'goal' && !vetoed) {
+        vetoed = true
+        agent.cancel({ kind: 'user' })
+        return Promise.resolve<PromptDecision>({ kind: 'block', reason: 'cancelled by policy' })
+      }
+      return next()
+    })
+    test.ctx.goals.create(test.agent, { objective: 'veto after cancellation' })
+
+    const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused')
+    await test.agent.whenIdle()
+
+    // Cancellation already cleared the reservation and paused the goal, so the
+    // veto neither touches an absent attempt nor blocks the paused goal.
+    expect(goal).toMatchObject({ roundsStarted: 0, activation: 'disarmed' })
+    expect(goal?.blockedReason).toBeUndefined()
+    expect(test.adapter.requests).toHaveLength(0)
+  })
+
+  it('awaits an unadmitted reservation stuck in admission during teardown without cancelling', async () => {
+    const test = await harness([])
+    let release: (() => void) | undefined
+    test.ctx.on('agent/prompt-submit', async (_agent, _content, source, _signal, next) => {
+      if (source.kind === 'goal' && release === undefined) {
+        await new Promise<void>((resolve) => { release = resolve })
+      }
+      return next()
+    })
+    test.ctx.goals.create(test.agent, { objective: 'unload during admission' })
+    await vi.waitFor(() => { expect(release).toBeDefined() })
+
+    const disposal = Promise.resolve(test.driver.dispose())
+    await waitForGoal(test.ctx, test.agent, goal => goal?.activation === 'disarmed')
+    release?.()
+    await disposal
+
+    expect(test.ctx.goals.get(test.agent)).toMatchObject({ phase: 'active', roundsStarted: 0 })
+    expect(test.adapter.requests).toHaveLength(0)
+    expect(test.agent.session.events.some(event => event.type === 'turn/start')).toBe(false)
+  })
+
   it('ignores session events without an exact owning agent and retires disposed agent state', async () => {
     const test = await harness([])
     const orphan = test.ctx.sessions.create(SessionId('goal-session-orphan'))

packages/goal/goal/src/fold.ts

@@ -358,6 +358,9 @@ export function applyGoalEvent(state: GoalFoldState, event: SessionEvent): GoalC
     if (source === undefined) return undefined
     // A goal-sourced message without a change must be a positive-round
     // admitted continuation prompt; round zero owes a durable source change.
+    /* v8 ignore next 3 -- decodeGoalEvent returns the change or fails loud for every
+       round-zero goal source, so only positive rounds reach here; the guard keeps
+       replay fail-loud against a decoder change */
     if (source.round === 0) {
       throw new Error(`goal source at session event ${event.seq} lacks goal change data`)
     }

packages/goal/goal/tests/goal.spec.ts

@@ -821,6 +821,12 @@ describe('goal replay validation', () => {
     expect(() => foldGoal(oneChange(change, {
       source: { kind: 'goal', goalId: change.goal.id, revision: 1, round: -1 },
     }))).toThrow('source is invalid')
+    expect(() => foldGoal(oneChange(change, {
+      source: { kind: 'goal', goalId: GoalId('goal-imposter'), revision: 1, round: 0, change },
+    }))).toThrow('mismatched source attribution')
+    expect(() => foldGoal(oneChange(change, {
+      source: { kind: 'goal', goalId: change.goal.id, revision: 2, round: 0, change },
+    }))).toThrow('mismatched source attribution')
     expect(() => foldGoal(oneChange(change, { content: [{ type: 'text', text: 'wrong' }] }))).toThrow('model-visible content')
   })
 

packages/host/runtime/tests/api-proxy-cold.spec.ts

@@ -1,8 +1,9 @@
 /**
  * Cold-session and degenerate-composition paths of the host ApiProxy:
  * sessions.list merging persisted-but-unattached summaries (mtime source,
- * createdAt fallbacks, lineage projection) and the resume error split when
- * the composition has no persistence gate and no agent factory.
+ * createdAt fallbacks, lineage projection), the resume error split when
+ * the composition has no persistence gate and no agent factory, and the
+ * agent-busy mapping of a synchronous prompt rejection.
  */
 
 import { mkdtempSync, writeFileSync, utimesSync } from 'node:fs'
@@ -12,6 +13,7 @@ import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import SessionStore from '@deepseek-ai/dsh-session'
 import AgentRegistry from '@deepseek-ai/dsh-agent'
+import type { Agent } from '@deepseek-ai/dsh-agent'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 import type { SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
 import type { RpcRequest } from '@deepseek-ai/dsh-host-apiproxy/api/rpc'
@@ -95,3 +97,38 @@ describe('degenerate composition (no persistence, no factory)', () => {
     }
   })
 })
+
+describe('sessions.prompt synchronous rejection', () => {
+  it('maps a synchronous send throw (disposed/invalid input) to agent-busy with the reason attached', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(AgentRegistry)
+    await ctx.plugin(UserInteractionService)
+    const session = ctx.sessions.create(sid('session-throwing'))
+    // A live structural stub whose delivery verbs throw synchronously, the
+    // shape a disposed loop presents at this seam.
+    ctx.agents.register({
+      id: session.id,
+      session,
+      status: 'idle',
+      ctx,
+      followup: () => { throw new Error('agent "session-throwing" lifecycle disposed') },
+      steer: () => { throw new Error('agent "session-throwing" lifecycle disposed') },
+    } as unknown as Agent)
+    const api = createApiProxy(ctx, { provider: 'p', model: 'm', cwd: '/tmp' })
+
+    for (const mode of ['queue', 'steer'] as const) {
+      const response = await api.sessions.prompt(request({
+        sessionId: session.id, mode, content: [{ type: 'text' as const, text: 'x' }],
+      }))
+      expect(response.result.ok).toBe(false)
+      if (!response.result.ok) {
+        expect(response.result.error.code).toBe('agent-busy')
+        expect(response.result.error.message).toBe('prompt rejected')
+        expect(response.result.error.details).toEqual({
+          reason: 'Error: agent "session-throwing" lifecycle disposed',
+        })
+      }
+    }
+  })
+})

packages/llm/llm-retry/tests/invariant.spec.ts

@@ -63,6 +63,50 @@ describe('llm-retry invariants', () => {
     }).toThrow(message)
   })
 
+  it('rejects a retry record appended after its turn already closed', async () => {
+    const ctx = await setup()
+    const closed = closeStep(ctx, 'retry-invariant-closed-turn')
+    closed.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
+    expect(() => {
+      closed.append('llm/retry', {
+        turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+      })
+    }).toThrow(/inside an open turn/)
+  })
+
+  it('starts a fresh chain when the turn before a retry trigger did not fail structurally', async () => {
+    const ctx = await setup()
+    const session = closeStep(ctx, 'retry-invariant-completed-predecessor')
+    session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+    session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
+    session.append('step/start', { turn: 2, step: 1 })
+    session.append('step/end', { turn: 2, step: 1 })
+    expect(() => {
+      session.append('llm/retry', {
+        turn: 2, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+      })
+    }).not.toThrow()
+  })
+
+  it('walks the chain across non-boundary events and stops at an unmatched turn start', async () => {
+    const ctx = await setup()
+    // The failed predecessor's turn/start is outside this log prefix (e.g. a
+    // truncated replay): the chain walk must stop rather than loop or throw.
+    const session = ctx.sessions.create(SessionId('retry-invariant-unmatched-start'))
+    session.append('turn/end', { turn: 1, reason: { kind: 'error', step: 1, failure } })
+    // A durable non-boundary record between the turns exercises the walk over
+    // non-turn/end events.
+    session.append('todo/write', { todos: [] })
+    session.append('turn/start', { turn: 2, trigger: { kind: 'retry' } })
+    session.append('step/start', { turn: 2, step: 1 })
+    session.append('step/end', { turn: 2, step: 1 })
+    expect(() => {
+      session.append('llm/retry', {
+        turn: 2, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+      })
+    }).not.toThrow()
+  })
+
   it('requires an open turn and its latest closed step', async () => {
     const ctx = await setup()
     const absent = ctx.sessions.create(SessionId('retry-invariant-no-turn'))
@@ -158,4 +202,18 @@ describe('llm-retry invariants', () => {
     await ctx.plugin(InvariantService)
     await expect(ctx.plugin(RetryInvariant)).rejects.toThrow(/inside an open turn/)
   })
+
+  it('accepts a valid mixed pre-existing history on late registration', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('retry-invariant-late-valid'))
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    session.append('step/start', { turn: 1, step: 1 })
+    session.append('step/end', { turn: 1, step: 1 })
+    session.append('llm/retry', {
+      turn: 1, step: 1, retry: 1, maxRetries: 2, delayMs: 1, failure,
+    })
+    await ctx.plugin(InvariantService)
+    await expect(ctx.plugin(RetryInvariant)).resolves.toBeDefined()
+  })
 })

packages/llm/llm-retry/tests/retry.spec.ts

@@ -117,6 +117,23 @@ afterEach(async () => {
   context = undefined
 })
 
+describe('config validation', () => {
+  it.each([
+    [{ maxTransientRetries: 1.5 }, /maxTransientRetries must be a non-negative integer/],
+    [{ maxTransientRetries: -1 }, /maxTransientRetries must be a non-negative integer/],
+    [{ initialDelayMs: 0 }, /initialDelayMs must be a positive finite number/],
+    [{ initialDelayMs: Number.NaN }, /initialDelayMs must be a positive finite number/],
+    [{ maxDelayMs: 0 }, /maxDelayMs must be a positive finite number/],
+    [{ initialDelayMs: 600, maxDelayMs: 500 }, /initialDelayMs must be less than or equal to maxDelayMs/],
+    [{ jitterRatio: Number.NaN }, /jitterRatio must be between 0 and 1/],
+    [{ retryableCodes: [] }, /retryableCodes must not be empty/],
+    [{ retryableCodes: ['SERVER', ''] }, /retryableCodes must contain only non-empty strings/],
+    [{ retryableCodes: ['SERVER', 'SERVER'] }, /retryableCodes must not contain duplicates/],
+  ] satisfies [retry.Config, RegExp][])('rejects invalid config %j at load', (config, message) => {
+    expect(() => { retry.apply(new Context(), config) }).toThrow(message)
+  })
+})
+
 describe('bounded transient retry policy', () => {
   it('records the scheduled delay before opening a fresh request attempt', async () => {
     vi.useFakeTimers()
@@ -375,6 +392,125 @@ describe('bounded transient retry policy', () => {
     expect(vi.getTimerCount()).toBe(0)
   })
 
+  it('keeps the consumed budget when an unowned session logs an assistant message', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy one', 'SERVER'),
+      new LlmError('busy two', 'SERVER'),
+    ])
+    ;({ ctx: context } = await harness(adapter, { maxTransientRetries: 1 }))
+    const agent = context.agentLoop.create(SessionId('retry-foreign-session'), { provider: 'mock', model: 'mock' })
+    const scheduled = waitForRetry(context, agent, 1)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await scheduled
+
+    // A session no agent owns completes a response; the agent's consecutive-
+    // failure sequence must not reset from that foreign success.
+    const foreign = context.sessions.create(SessionId('retry-foreign-session-other'))
+    foreign.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    foreign.append('step/start', { turn: 1, step: 1 })
+    foreign.append('assistant/message', {
+      turn: 1,
+      step: 1,
+      content: [{ type: 'text', text: 'foreign' }],
+      provenance: { provider: 'mock', model: 'mock' },
+    }, { surfaceOp: 'append' })
+
+    const idle = waitForIdle(context, agent)
+    await vi.advanceTimersByTimeAsync(500)
+    await idle
+
+    expect(adapter.requests).toHaveLength(2)
+    expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'error', failure: { message: 'busy two', code: 'SERVER' } } },
+    })
+  })
+
+  it('drops a scheduled retry when cancellation lands between its durable record and its wait', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy', 'SERVER'),
+      textResponse('must not run'),
+    ])
+    ;({ ctx: context } = await harness(adapter))
+    const agent = context.agentLoop.create(SessionId('retry-cancel-at-record'), { provider: 'mock', model: 'mock' })
+    // The durable record commits synchronously before the cancellable wait; a
+    // user cancel observed at that exact point must skip the wait entirely.
+    const dispose = context.on('session/event', (session, event) => {
+      if (session === agent.session && event.type === 'llm/retry') {
+        dispose()
+        agent.cancel({ kind: 'user' })
+      }
+    })
+    const idle = waitForIdle(context, agent)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await idle
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.filter(event => event.type === 'llm/retry')).toHaveLength(1)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('schedules nothing when an upstream recovery listener already cancelled the turn', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy', 'SERVER'),
+      textResponse('must not run'),
+    ])
+    ;({ ctx: context } = await harness(adapter, {}, (ctx) => {
+      // Registered before the retry plugin, so it wraps the policy: it cancels
+      // the turn, then delegates into a policy that sees an aborted signal.
+      ctx.on('agent/request-error', (agent, _turn, _step, _error, _failure, _signal, next) => {
+        agent.cancel({ kind: 'user' })
+        return next()
+      })
+    }))
+    const agent = context.agentLoop.create(SessionId('retry-upstream-cancel'), { provider: 'mock', model: 'mock' })
+    const idle = waitForIdle(context, agent)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await idle
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
+  it('does nothing when its captured listener resumes after plugin disposal', async () => {
+    vi.useFakeTimers()
+    const adapter = new ScriptedAdapter([
+      new LlmError('busy', 'SERVER'),
+      textResponse('must not run'),
+    ])
+    const holder: { dispose?: () => Promise<void> } = {}
+    const mounted = await harness(adapter, {}, (ctx) => {
+      // An upstream listener captured in the same waterfall disposes the retry
+      // plugin before delegating; the stale downstream callback must bail.
+      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _signal, next) => {
+        await holder.dispose?.()
+        return next()
+      })
+    })
+    context = mounted.ctx
+    holder.dispose = () => mounted.retryFiber.dispose()
+    const agent = context.agentLoop.create(SessionId('retry-stale-listener'), { provider: 'mock', model: 'mock' })
+    const idle = waitForIdle(context, agent)
+
+    agent.followup({ content: [{ type: 'text', text: 'go' }], source: { kind: 'user' } })
+    await idle
+    await vi.advanceTimersByTimeAsync(60_000)
+
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.some(event => event.type === 'llm/retry')).toBe(false)
+    expect(vi.getTimerCount()).toBe(0)
+  })
+
   it('aborts and drains a captured backoff before plugin disposal completes', async () => {
     vi.useFakeTimers()
     const adapter = new ScriptedAdapter([

packages/support/acp-snapshot/tests/normalize.spec.ts

@@ -1,6 +1,7 @@
 import { describe, expect, it } from 'vitest'
 import {
   type NormalizeContext,
+  extractSnapshotSpillPaths,
   normalizeSessionLog,
   normalizeStdout,
   scrubRequestHeaders,
@@ -319,6 +320,24 @@ describe('normalizeSessionLog', () => {
   })
 })
 
+describe('extractSnapshotSpillPaths', () => {
+  it('maps each spill filename to its full matched path, last match wins per name', () => {
+    const log = [
+      'Full formatted result stored at: /tmp/dsh-acp-snapshot-spill/session-c22bc3f1d2af/8a7b6c5d4e3f-bash.txt. Use read with offset/limit, or grep this path to search within it.',
+      'stale copy at /tmp/dsh-acp-snap-012345678/session-aaaaaaaaaaaa/bbbbbbbbbbbb-grep.txt then',
+      'fresh copy at /tmp/dsh-acp-snap-012345678/session-cccccccccccc/dddddddddddd-grep.txt then',
+    ].join('\n')
+    expect(extractSnapshotSpillPaths(log)).toEqual(new Map([
+      ['bash.txt', '/tmp/dsh-acp-snapshot-spill/session-c22bc3f1d2af/8a7b6c5d4e3f-bash.txt'],
+      ['grep.txt', '/tmp/dsh-acp-snap-012345678/session-cccccccccccc/dddddddddddd-grep.txt'],
+    ]))
+  })
+
+  it('returns an empty map when the log carries no snapshot spill paths', () => {
+    expect(extractSnapshotSpillPaths('no spill paths here, only /tmp/other.txt\n')).toEqual(new Map())
+  })
+})
+
 describe('scrubRequestHeaders', () => {
   const headerLine = JSON.stringify({ type: 'session', version: 0, id: 's', createdAt: 1, cwd: '/w' })
   const headerEvent = (header: object) =>

packages/support/acp-snapshot/tests/suite.spec.ts

@@ -455,6 +455,29 @@ describe('refreshFixtureReplacements', () => {
       { from: '/new', to: '/old' },
     ])
   })
+
+  it('stabilizes moved snapshot spill paths by filename while skipping unchanged or unmatched names', () => {
+    const spill = (session: string, hash: string, name: string): string =>
+      `/tmp/dsh-acp-snapshot-spill/session-${session}/${hash}-${name}`
+    const record = (text: string): string =>
+      `${JSON.stringify({ type: 'session', id: 'same', cwd: '/same' })}\n`
+      + `${JSON.stringify({ type: 'tool/result', data: { content: [{ type: 'text', text: `stored at: ${text} ` }] } })}\n`
+    const freshBash = spill('aaaaaaaaaaaa', 'bbbbbbbbbbbb', 'bash.txt')
+    const oldBash = spill('cccccccccccc', 'dddddddddddd', 'bash.txt')
+    const shared = spill('eeeeeeeeeeee', 'ffffffffffff', 'grep.txt')
+    const orphan = spill('111111111111', '222222222222', 'orphan.txt')
+    const logs: HarvestedLog[] = [{
+      id: 'diagnostic',
+      createdAt: 1,
+      content: record(`${freshBash} and ${shared}`),
+    }]
+    const fixtures = [record(`${oldBash} and ${shared} and ${orphan}`)]
+    // bash.txt moved → replaced; grep.txt is identical and orphan.txt has no
+    // fresh counterpart → both skipped.
+    expect(refreshFixtureReplacements(logs, fixtures)).toEqual([
+      { from: freshBash, to: oldBash },
+    ])
+  })
 })
 
 describe('stabilizeRefreshLog', () => {

packages/ui/acp/tests/turns.spec.ts

@@ -14,6 +14,17 @@ import {
 } from './harness.ts'
 import { SessionId } from '@deepseek-ai/dsh-session'
 
+declare module '@deepseek-ai/dsh-session' {
+  interface SessionEventMap {
+    /** Test-only log-only event driven through appendOutOfBand below. */
+    'test/acp-out-of-band': { note: string }
+  }
+
+  interface OutOfBandSessionEventMap {
+    'test/acp-out-of-band': true
+  }
+}
+
 /** Boilerplate: initialize + create one session, returning its id. */
 async function newSession(h: BridgeHarness, clientCapabilities: Record<string, unknown> = {}): Promise<string> {
   await h.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities })
@@ -342,6 +353,43 @@ describe('acp bridge — turn outcomes', () => {
     expect(text).toContain('real answer')
   })
 
+  it('an out-of-band injection TURN while the prompt is queued neither captures nor settles it', async () => {
+    // Unlike the idle inject above (which appends context without a turn), a
+    // log-only out-of-band append on a closed log opens a real synthetic
+    // injection-triggered turn. Its turn/start must NOT capture inflight.turn
+    // (only message-triggered turns own the prompt) and its turn/end must not
+    // settle the prompt — the prompt settles on its OWN later message turn.
+    harness = await makeBridgeHarness({ storageDir, script: [textResponse('real answer')] })
+    const sessionId = await newSession(harness)
+    const agent = harness.ctx.agents.get(SessionId(sessionId))!
+    let appended: Promise<unknown> | undefined
+    const sessions = harness.ctx.sessions
+    harness.ctx.on('agent/inbox/enqueue', (subject) => {
+      if (subject === agent && appended === undefined) {
+        // Fires synchronously inside followup(), after the bridge installed the
+        // in-flight slot but before the prompt's own turn starts; the synthetic
+        // turn/start + turn/end land in that window.
+        appended = sessions.appendOutOfBand(
+          agent.session,
+          'test/acp-out-of-band',
+          { note: 'log-only' },
+          { kind: 'injection', source: { kind: 'plugin', plugin: 'test' } },
+        )
+      }
+    })
+    const res = await harness.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] })
+    expect(res.stopReason).toBe('end_turn')
+    await appended
+    // The synthetic injection turn precedes the prompt's own message turn.
+    const triggers = agent.session.events.flatMap(e => e.type === 'turn/start' ? [e.data.trigger.kind] : [])
+    expect(triggers).toEqual(['injection', 'message'])
+    const text = harness.updates
+      .filter(u => u.sessionUpdate === 'agent_message_chunk')
+      .map(u => (u.content.type === 'text' ? u.content.text : ''))
+      .join('')
+    expect(text).toContain('real answer')
+  })
+
   it('rejects a second prompt while one is in flight', async () => {
     harness = await makeBridgeHarness({ storageDir, script: ['hang'] })
     const sessionId = await newSession(harness)

packages/ui/tui/tests/tui.spec.ts

@@ -1431,6 +1431,32 @@ describe('pi-tui chat lifecycle and transcript', () => {
     expect(result.terminal.output).toContain('— Enter sends steering, Esc cancels')
     expect(result.terminal.output).not.toContain('queued')
 
+    // A cancellation discards queued steering: the badge clears without drains.
+    submitSteering('third')
+    submitSteering('fourth')
+    await tick()
+    expect(result.terminal.output).toContain('2 queued')
+    const discarded = result.agent.steeredIds.splice(0).map(id => ({
+      id, content: [{ type: 'text' as const, text: 'discarded' }], source: { kind: 'user' as const },
+    }))
+    // Another agent's dequeue/discard, and ones naming no pending id, leave
+    // the badge alone.
+    result.ctx.emit('agent/inbox/dequeue', other, discarded[0]!)
+    result.ctx.emit('agent/inbox/dequeue', result.agent, {
+      id: AgentMessageId('never-queued'), content: [{ type: 'text', text: 'x' }], source: { kind: 'user' },
+    })
+    result.ctx.emit('agent/inbox/discard', other, discarded)
+    result.ctx.emit('agent/inbox/discard', result.agent, [
+      { id: AgentMessageId('never-queued'), content: [{ type: 'text', text: 'x' }], source: { kind: 'user' } },
+    ])
+    await tick()
+    expect(result.terminal.output).toContain('2 queued')
+    result.terminal.output = ''
+    result.ctx.emit('agent/inbox/discard', result.agent, discarded)
+    await tick()
+    expect(result.terminal.output).toContain('— Enter sends steering, Esc cancels')
+    expect(result.terminal.output).not.toContain('queued')
+
     await dispose(result)
   })
 
@@ -1811,6 +1837,18 @@ describe('pi-tui chat lifecycle and transcript', () => {
     // /reload without a Loader in the context degrades to a warning.
     expect(result.terminal.output).toContain('/reload needs the cordis Loader')
     expect(result.exit).toHaveBeenCalledWith(0)
+
+    // The exit above left the TUI disposed (the mocked runtime.exit returns):
+    // a message submitted now is refused instead of reaching the agent. The
+    // refusal notice lands in the transcript, but the stopped UI no longer
+    // paints, so assert the refusal through the agent surface.
+    const sentBefore = result.agent.sent.length
+    const steeredBefore = result.agent.steered.length
+    result.terminal.send('after shutdown')
+    result.terminal.send('\r')
+    await tick()
+    expect(result.agent.sent).toHaveLength(sentBefore)
+    expect(result.agent.steered).toHaveLength(steeredBefore)
     await result.controller.dispose()
     await result.ctx.fiber.dispose()
 
@@ -2148,6 +2186,7 @@ describe('pi-tui chat lifecycle and transcript', () => {
     expect(result.terminal.output).not.toContain('secret full snapshot payload')
 
     const invalidCards: [JsonValue, string][] = [
+      ['plain-string-source', 'invalid-shape'],
       [{ kind: 'other' }, 'invalid-kind'],
       [{ kind: 'session-reference', references: [null] }, 'invalid-entry'],
       [{ kind: 'session-reference', references: [{}] }, 'invalid-fields'],