Merge remote-tracking branch 'origin/master' into codex/enforce-tool-cancellation

# Conflicts:
#	docs/cookbook/adding-a-tool.i18n.yaml
#	examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/session.jsonl
#	examples/acp-agent/tests/snapshots/cordis-inspect-jsdoc/stdout.expected.jsonl
#	packages/bash/tool-bash/src/index.ts
#	packages/core/agent-loop/README.md
#	packages/core/tools/README.md
#	packages/core/tools/tests/scoped.spec.ts
#	packages/fs/tool-fs-search/tests/tools.spec.ts
#	website/zh-CN/api/harness/events.md
#	website/zh-CN/api/harness/tools.md

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

@@ -79,7 +79,8 @@ describe('Agent.cancel()', () => {
 
     // send() queues synchronously (status still idle, loop microtask not yet
     // resumed). Cancel in that pre-step window: the queued turn must not run.
-    send(agent, 'drop me')
+    send(agent, 'drop me first')
+    send(agent, 'drop me second')
     agent.cancel('pre-step')
 
     // Give the loop a chance to wake and process the cancel.
@@ -91,6 +92,35 @@ describe('Agent.cancel()', () => {
     expect(agent.status).toBe('idle')
   })
 
+  it('disposal from the running notification drops queued work before turn start', async () => {
+    const adapter = new MockAdapter([textResponse('should not run')])
+    const ctx = await harness(adapter)
+    const handle = await ctx.agents.create({
+      sessionId: SessionId('dispose-running-session'),
+      agentOptions: { provider: 'mock', model: 'mock' },
+    })
+    const agent = handle.agent
+
+    const running = Promise.withResolvers<undefined>()
+    let disposalDone: Promise<void> | undefined
+    ctx.on('agent/status', (subject, status) => {
+      if (subject !== agent || status !== 'running') return
+      disposalDone = handle.dispose()
+      running.resolve(undefined)
+    })
+
+    send(agent, 'drop before claim')
+    await running.promise
+    if (disposalDone === undefined) throw new Error('running listener did not start disposal')
+    await disposalDone
+    await driverDone(agent)
+
+    expect(agent.status).toBe('disposed')
+    expect(agent.session.events.some(event => event.type === 'turn/start')).toBe(false)
+    expect(userTexts(agent)).toEqual([])
+    expect(adapter.requests).toHaveLength(0)
+  })
+
   it('a whenIdle() waiter registered BEFORE a pre-step cancel resolves (F1 hang guard)', async () => {
     const adapter = new MockAdapter([textResponse('x')])
     const ctx = await harness(adapter)
@@ -110,7 +140,162 @@ describe('Agent.cancel()', () => {
     expect(agent.status).toBe('idle')
   })
 
-  it('cancel() mid-step aborts the in-flight model call; the turn ends aborted', async () => {
+  it('cancel() between consecutive turns restores idle and leaves idle steer usable', async () => {
+    const adapter = new MockAdapter([textResponse('first reply'), textResponse('steer reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('between-turn-cancel'), { provider: 'mock', model: 'mock' })
+
+    let rejectFirstFlush = true
+    ctx.on('session/flush', (session) => {
+      if (session !== agent.session || !rejectFirstFlush) return
+      rejectFirstFlush = false
+      throw new Error('first flush failed')
+    })
+
+    const cancelled = Promise.withResolvers<undefined>()
+    ctx.on('agent/error', (subject, _turn, _step, error) => {
+      if (subject !== agent || error.message !== 'first flush failed') return
+      // The first hop runs before runLoop resumes from runTurn; the second lands
+      // before its resolved waitForQueued continuation checks cancellation.
+      queueMicrotask(() => {
+        queueMicrotask(() => {
+          agent.cancel('between turns')
+          cancelled.resolve(undefined)
+        })
+      })
+    })
+
+    const statuses: string[] = []
+    ctx.on('agent/status', (subject, status) => {
+      if (subject === agent) statuses.push(status)
+    })
+
+    send(agent, 'first')
+    send(agent, 'queued tail')
+    await cancelled.promise
+
+    expect(agent.status).toBe('idle')
+    expect(statuses).toEqual(['running', 'idle'])
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
+    expect(userTexts(agent)).toEqual(['first'])
+
+    let idleResolved = false
+    void agent.whenIdle().then(() => { idleResolved = true })
+    await Promise.resolve()
+    expect(idleResolved).toBe(true)
+
+    const idle = waitForIdle(ctx, agent)
+    agent.steer([{ type: 'text', text: 'idle steer' }])
+    await idle
+
+    expect(statuses).toEqual(['running', 'idle', 'running', 'idle'])
+    expect(adapter.requests).toHaveLength(2)
+    expect(userTexts(agent)).toEqual(['first', 'idle steer'])
+  })
+
+  it('an idle-listener replacement keeps whenIdle pending until the replacement turn finishes', async () => {
+    const adapter = new MockAdapter([textResponse('first reply'), textResponse('replacement reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('between-turn-idle-listener'), { provider: 'mock', model: 'mock' })
+
+    let rejectFirstFlush = true
+    ctx.on('session/flush', (session) => {
+      if (session !== agent.session || !rejectFirstFlush) return
+      rejectFirstFlush = false
+      throw new Error('first flush failed')
+    })
+
+    ctx.on('agent/error', (subject, _turn, _step, error) => {
+      if (subject !== agent || error.message !== 'first flush failed') return
+      queueMicrotask(() => {
+        queueMicrotask(() => { agent.cancel('between turns') })
+      })
+    })
+
+    const replacementRegistered = Promise.withResolvers<undefined>()
+    let replacementObservation: Promise<{ status: string; requests: number; turns: number }> | undefined
+    ctx.on('agent/status', (subject, status) => {
+      if (subject !== agent || status !== 'idle' || replacementObservation !== undefined) return
+      send(agent, 'replacement')
+      replacementObservation = agent.whenIdle().then(() => ({
+        status: agent.status,
+        requests: adapter.requests.length,
+        turns: agent.session.events.filter(event => event.type === 'turn/start').length,
+      }))
+      replacementRegistered.resolve(undefined)
+    })
+
+    send(agent, 'first')
+    send(agent, 'cancelled tail')
+    await replacementRegistered.promise
+    if (replacementObservation === undefined) throw new Error('idle listener did not register replacement work')
+
+    await expect(replacementObservation).resolves.toEqual({ status: 'idle', requests: 2, turns: 2 })
+    expect(userTexts(agent)).toEqual(['first', 'replacement'])
+  })
+
+  it('idle-listener cancellation settles its waiter without cancelling later work', async () => {
+    const adapter = new MockAdapter([textResponse('first reply'), textResponse('later reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('idle-listener-cancel'), { provider: 'mock', model: 'mock' })
+
+    const replacementRegistered = Promise.withResolvers<undefined>()
+    let replacementObservation: Promise<{ status: string; requests: number; turns: number }> | undefined
+    ctx.on('agent/status', (subject, status) => {
+      if (subject !== agent || status !== 'idle' || replacementObservation !== undefined) return
+      send(agent, 'cancelled replacement')
+      replacementObservation = agent.whenIdle().then(() => ({
+        status: agent.status,
+        requests: adapter.requests.length,
+        turns: agent.session.events.filter(event => event.type === 'turn/start').length,
+      }))
+      agent.cancel('idle listener')
+      replacementRegistered.resolve(undefined)
+    })
+
+    send(agent, 'first')
+    await replacementRegistered.promise
+    if (replacementObservation === undefined) throw new Error('idle listener did not register replacement work')
+
+    await expect(Promise.race([
+      replacementObservation,
+      new Promise((_resolve, reject) => setTimeout(() => { reject(new Error('whenIdle hung after idle-listener cancel')) }, 1000)),
+    ])).resolves.toEqual({ status: 'idle', requests: 1, turns: 1 })
+
+    const idle = waitForIdle(ctx, agent)
+    send(agent, 'later')
+    await idle
+    expect(adapter.requests).toHaveLength(2)
+    expect(userTexts(agent)).toEqual(['first', 'later'])
+  })
+
+  it('replacement work queued after idle-listener cancellation still runs', async () => {
+    const adapter = new MockAdapter([textResponse('first reply'), textResponse('replacement reply')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('idle-listener-post-cancel-send'), { provider: 'mock', model: 'mock' })
+
+    const replacementRegistered = Promise.withResolvers<undefined>()
+    let replacementIdle: Promise<void> | undefined
+    ctx.on('agent/status', (subject, status) => {
+      if (subject !== agent || status !== 'idle' || replacementIdle !== undefined) return
+      send(agent, 'cancelled replacement')
+      agent.cancel('idle listener')
+      send(agent, 'surviving replacement')
+      replacementIdle = agent.whenIdle()
+      replacementRegistered.resolve(undefined)
+    })
+
+    send(agent, 'first')
+    await replacementRegistered.promise
+    if (replacementIdle === undefined) throw new Error('idle listener did not register replacement work')
+    await replacementIdle
+
+    expect(adapter.requests).toHaveLength(2)
+    expect(userTexts(agent)).toEqual(['first', 'surviving replacement'])
+  })
+
+  it('cancel() mid-step aborts the active turn and drops every queued tail item', async () => {
     const adapter = new MockAdapter(['hang'])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -121,10 +306,14 @@ describe('Agent.cancel()', () => {
     send(agent, 'go')
     await new Promise(r => setTimeout(r, 30))
     expect(agent.status).toBe('running')
+    send(agent, 'queued tail')
     agent.cancel('mid-step')
     await waitForIdle(ctx, agent)
 
     expect(reasons).toEqual([{ kind: 'aborted', reason: 'mid-step' }])
+    expect(userTexts(agent)).toEqual(['go'])
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
+    expect(adapter.requests).toHaveLength(1)
   })
 
   it('cancel() with no reason defaults to "cancelled" when aborting an in-flight step', async () => {

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

@@ -47,9 +47,9 @@ describe('config-driven session id', () => {
   it('accepts one exact fresh id and rejects it alongside a resume id', async () => {
     const exact = await makeCoreContext()
     await exact.plugin(AgentLoop, {
-      agents: [{ id: 'main', sessionId: SessionId('stdio-exact'), model: 'mock' }],
+      agents: [{ id: 'main', sessionId: SessionId('config-exact'), model: 'mock' }],
     })
-    expect(exact.agents.get(SessionId('stdio-exact'))?.session.id).toBe('stdio-exact')
+    expect(exact.agents.get(SessionId('config-exact'))?.session.id).toBe('config-exact')
     await exact.fiber.dispose()
 
     const conflicting = await makeCoreContext()
@@ -89,13 +89,13 @@ describe('config-driven session id', () => {
     const ctx = await makeCoreContext()
     await ctx.plugin(SessionPersistenceJsonl, { root })
     ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first'), textResponse('second')]))
-    const config = { agents: [{ id: 'main', sessionId: SessionId('stdio-exact-reload'), model: 'mock' }] }
+    const config = { agents: [{ id: 'main', sessionId: SessionId('config-exact-reload'), model: 'mock' }] }
 
     const firstLoop = await ctx.plugin(AgentLoop, config)
     let first: Agent | undefined
     for (let i = 0; i < 50 && first === undefined; i++) {
       await new Promise(resolve => setTimeout(resolve, 5))
-      first = ctx.agents.get(SessionId('stdio-exact-reload'))
+      first = ctx.agents.get(SessionId('config-exact-reload'))
     }
     expect(first).toBeDefined()
     first!.send([{ type: 'text', text: 'remember me' }], { source: { kind: 'user' } })
@@ -106,14 +106,14 @@ describe('config-driven session id', () => {
     let second: Agent | undefined
     for (let i = 0; i < 50 && second === undefined; i++) {
       await new Promise(resolve => setTimeout(resolve, 5))
-      second = ctx.agents.get(SessionId('stdio-exact-reload'))
+      second = ctx.agents.get(SessionId('config-exact-reload'))
     }
     expect(second).toBeDefined()
     expect(JSON.stringify(second!.session.deriveMessages())).toContain('remember me')
     second!.send([{ type: 'text', text: 'continue' }], { source: { kind: 'user' } })
     await waitForIdle(ctx, second!)
     await ctx.sessions.flush(second!.session)
-    const loaded = await ctx.sessionPersistence.load(SessionId('stdio-exact-reload'))
+    const loaded = await ctx.sessionPersistence.load(SessionId('config-exact-reload'))
     expect(loaded.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
 
     await secondLoop.dispose()
@@ -125,7 +125,7 @@ describe('config-driven session id', () => {
     dirs.push(root)
     const ctx = await makeCoreContext()
     await ctx.plugin(SessionPersistenceJsonl, { root })
-    const sessionId = SessionId('stdio-exact-overlap')
+    const sessionId = SessionId('config-exact-overlap')
     const config = { agents: [{ id: 'main', sessionId, model: 'mock' }] }
     const firstLoop = await ctx.plugin(AgentLoop, config)
     await expect.poll(() => ctx.agents.get(sessionId)).toBeDefined()
@@ -169,7 +169,7 @@ describe('config-driven session id', () => {
     dirs.push(root)
     const ctx = await makeCoreContext()
     await ctx.plugin(SessionPersistenceJsonl, { root })
-    const sessionId = SessionId('stdio-exact-cancel')
+    const sessionId = SessionId('config-exact-cancel')
     const config = { agents: [{ id: 'main', sessionId, model: 'mock' }] }
     const firstLoop = await ctx.plugin(AgentLoop, config)
     await expect.poll(() => ctx.agents.get(sessionId)).toBeDefined()
@@ -213,20 +213,20 @@ describe('config-driven session id', () => {
     const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
 
     await ctx.plugin(AgentLoop, {
-      agents: [{ id: 'main', sessionId: SessionId('stdio-exact-failure'), model: 'mock' }],
+      agents: [{ id: 'main', sessionId: SessionId('config-exact-failure'), model: 'mock' }],
     })
 
     await expect.poll(() => warn).toHaveBeenCalledWith(expect.stringContaining(
-      'config-driven restore of "stdio-exact-failure" failed: Error: persistence index failed',
+      'config-driven restore of "config-exact-failure" failed: persistence index failed',
     ))
-    expect(failures).toEqual([{ sessionId: SessionId('stdio-exact-failure'), error: failure }])
+    expect(failures).toEqual([{ sessionId: SessionId('config-exact-failure'), error: failure }])
     expect(warn).toHaveBeenCalledWith(
-      'agent "main": config-start-failed listener threw: Error: failure observer failed',
+      'agent "main": config-start-failed listener threw: failure observer failed',
     )
     await expect.poll(() => warn).toHaveBeenCalledWith(
-      'agent "main": config-start-failed listener rejected: Error: async failure observer failed',
+      'agent "main": config-start-failed listener rejected: async failure observer failed',
     )
-    expect(ctx.agents.get(SessionId('stdio-exact-failure'))).toBeUndefined()
+    expect(ctx.agents.get(SessionId('config-exact-failure'))).toBeUndefined()
     warn.mockRestore()
     await ctx.fiber.dispose()
   })
@@ -251,18 +251,18 @@ describe('config-driven session id', () => {
     const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
 
     await ctx.plugin(AgentLoop, {
-      agents: [{ id: 'main', sessionId: SessionId('stdio-exact-unrenderable'), model: 'mock' }],
+      agents: [{ id: 'main', sessionId: SessionId('config-exact-unrenderable'), model: 'mock' }],
     })
 
     await expect.poll(() => failures).toEqual([unrenderable])
     expect(warn).toHaveBeenCalledWith(
-      'agent "main": config-driven restore of "stdio-exact-unrenderable" failed: <unrenderable thrown value>',
+      'agent "main": config-driven restore of "config-exact-unrenderable" failed: <unrenderable value>',
     )
     expect(warn).toHaveBeenCalledWith(
-      'agent "main": config-start-failed listener threw: <unrenderable thrown value>',
+      'agent "main": config-start-failed listener threw: <unrenderable value>',
     )
     await expect.poll(() => warn).toHaveBeenCalledWith(
-      'agent "main": config-start-failed listener rejected: <unrenderable thrown value>',
+      'agent "main": config-start-failed listener rejected: <unrenderable value>',
     )
     await ctx.fiber.dispose()
   })
@@ -281,7 +281,7 @@ describe('config-driven session id', () => {
       ctx.on('agent-loop/config-start-failed', (_sessionId, error) => { failures.push(error) })
 
       const loop = await ctx.plugin(AgentLoop, {
-        agents: [{ id: 'main', sessionId: SessionId('stdio-exact-dispose'), model: 'mock' }],
+        agents: [{ id: 'main', sessionId: SessionId('config-exact-dispose'), model: 'mock' }],
       })
       let disposed = false
       const disposal = loop.dispose().then(() => { disposed = true })
@@ -291,7 +291,7 @@ describe('config-driven session id', () => {
       if (outcome === 'resolve') listing.resolve([])
       else listing.reject(new Error('startup cancelled by teardown'))
       await disposal
-      expect(ctx.agents.get(SessionId('stdio-exact-dispose'))).toBeUndefined()
+      expect(ctx.agents.get(SessionId('config-exact-dispose'))).toBeUndefined()
       expect(failures).toEqual([])
       expect(warn).not.toHaveBeenCalled()
       warn.mockRestore()

packages/core/agent-loop/tests/contract-regressions.spec.ts

@@ -1,6 +1,6 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
-import LlmService, { CallId, ContentBlock, MessageSource, StreamChunk } from '@deepseek-ai/dsh-llm'
+import LlmService, { CallId, ContentBlock, MessageSource, ProviderRequestId, StreamChunk } from '@deepseek-ai/dsh-llm'
 import SessionStore, { Session, SessionEvent, SessionId, TurnEndReason } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineTool, TOOL_ABORTED, TOOL_ABORTED_BEFORE_DISPATCH, type PostToolDecision } from '@deepseek-ai/dsh-tools'
@@ -640,15 +640,20 @@ describe('plugin exceptions are contained', () => {
     expect(agent.status).toBe('idle')
   })
 
-  it('a rejecting session/flush listener is reported but does not kill the agent', async () => {
+  it('a rejecting first-turn flush settles before the queued tail starts', async () => {
     const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
 
-    let rejectedOnce = false
-    ctx.on('session/flush', async () => {
-      if (!rejectedOnce) {
-        rejectedOnce = true
+    const firstFlush = Promise.withResolvers<undefined>()
+    const releaseFirstFlush = Promise.withResolvers<undefined>()
+    let flushes = 0
+    ctx.on('session/flush', async (session) => {
+      if (session !== agent.session) return
+      flushes += 1
+      if (flushes === 1) {
+        firstFlush.resolve(undefined)
+        await releaseFirstFlush.promise
         throw new Error('disk full')
       }
     })
@@ -656,18 +661,25 @@ describe('plugin exceptions are contained', () => {
     const errors: Error[] = []
     ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
 
+    const idle = waitForIdle(ctx, agent)
     send(agent, 'first')
-    await waitForIdle(ctx, agent)
-    expect(errors.map(e => e.message)).toEqual(['disk full'])
-
     send(agent, 'second')
-    await waitForIdle(ctx, agent)
+
+    await firstFlush.promise
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
+
+    releaseFirstFlush.resolve(undefined)
+    await idle
+
+    expect(errors.map(e => e.message)).toEqual(['disk full'])
     expect(adapter.requests).toHaveLength(2)
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
   })
 })
 
 describe('disposed status is part of the agent/status contract', () => {
-  it('disposing the fiber emits agent/status(disposed) and ends the turn with reason disposed', async () => {
+  it('disposing the fiber ends the active turn and never starts its queued tail', async () => {
     const adapter = new MockAdapter(['hang'])
     const ctx = await harness(adapter)
 
@@ -683,11 +695,19 @@ describe('disposed status is part of the agent/status contract', () => {
 
     send(agent, 'go')
     await new Promise(r => setTimeout(r, 30))
+    send(agent, 'queued tail')
     await fiber.dispose()
     await driverDone(agent)
 
     expect(statuses).toEqual(['running', 'disposed'])
     expect(reasons).toEqual([{ kind: 'disposed' }])
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
+    const messages = agent.session.events
+      .filter(event => event.type === 'user/message')
+      .flatMap(event => event.data.content)
+      .flatMap(block => block.type === 'text' ? [block.text] : [])
+    expect(messages).toEqual(['go'])
+    expect(adapter.requests).toHaveLength(1)
   })
 
   it('a throwing agent/status listener cannot break disposal or leak the registry entry', async () => {
@@ -933,8 +953,15 @@ describe('discriminated SessionEvent narrows without casts', () => {
 describe('a finish-error stream chunk ends the turn as error, not completed', () => {
   it('translates finish {kind:error} into a turn error with a logged error event', async () => {
     // A finish-error chunk must not produce a completed assistant turn.
+    const failure = {
+      message: 'provider 401',
+      code: 'AUTH',
+      status: 401,
+      providerRetryAfterMs: 2_000,
+      requestId: ProviderRequestId('finish-request-1'),
+    }
     const errorStream: StreamChunk[] = [
-      { type: 'finish', reason: { kind: 'error', message: 'provider 401', code: 'AUTH' } },
+      { type: 'finish', reason: { kind: 'error', failure } },
     ]
     const adapter = new MockAdapter([errorStream])
     const ctx = await harness(adapter)
@@ -946,20 +973,20 @@ describe('a finish-error stream chunk ends the turn as error, not completed', ()
     send(agent, 'go')
     await waitForIdle(ctx, agent)
 
-    expect(reasons).toEqual([{ kind: 'error', step: 1, message: 'provider 401', code: 'AUTH' }])
+    expect(reasons).toEqual([{ kind: 'error', step: 1, failure }])
 
     const events = [...agent.session.events]
     // The durable failure lives on turn/end.reason (with the failing step), not
     // a standalone error event.
     const turnEnd = events.find(event => event.type === 'turn/end')
-    expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'error', step: 1, message: 'provider 401', code: 'AUTH' })
+    expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'error', step: 1, failure })
     // A failed step must not synthesize an assistant message.
     expect(events.some(event => event.type === 'assistant/message')).toBe(false)
   })
 
   it('translates finish {kind:aborted} into a turn error coded ABORTED', async () => {
     const abortedStream: StreamChunk[] = [
-      { type: 'finish', reason: { kind: 'aborted' } },
+      { type: 'finish', reason: { kind: 'aborted', failure: { message: 'model stream aborted', code: 'ABORTED' } } },
     ]
     const adapter = new MockAdapter([abortedStream])
     const ctx = await harness(adapter)
@@ -971,13 +998,13 @@ describe('a finish-error stream chunk ends the turn as error, not completed', ()
     send(agent, 'go')
     await waitForIdle(ctx, agent)
 
-    expect(reasons).toEqual([{ kind: 'error', step: 1, message: 'model stream aborted', code: 'ABORTED' }])
+    expect(reasons).toEqual([{ kind: 'error', step: 1, failure: { message: 'model stream aborted', code: 'ABORTED' } }])
     expect([...agent.session.events].some(event => event.type === 'assistant/message')).toBe(false)
   })
 
   it('handles a finish error without a code (code key omitted)', async () => {
     const errorStream: StreamChunk[] = [
-      { type: 'finish', reason: { kind: 'error', message: 'codeless failure' } },
+      { type: 'finish', reason: { kind: 'error', failure: { message: 'codeless failure', code: 'UNKNOWN' } } },
     ]
     const adapter = new MockAdapter([errorStream])
     const ctx = await harness(adapter)
@@ -989,7 +1016,7 @@ describe('a finish-error stream chunk ends the turn as error, not completed', ()
     send(agent, 'go')
     await waitForIdle(ctx, agent)
 
-    expect(reasons).toEqual([{ kind: 'error', step: 1, message: 'codeless failure' }])
+    expect(reasons).toEqual([{ kind: 'error', step: 1, failure: { message: 'codeless failure', code: 'UNKNOWN' } }])
   })
 })
 
@@ -1109,7 +1136,7 @@ describe('turn and step boundary recovery', () => {
   })
 
   it('a one-shot turn/end validation failure preserves the earlier turn error on retry', async () => {
-    const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider failed' } }]
+    const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', failure: { message: 'provider failed', code: 'UNKNOWN' } } }]
     const adapter = new MockAdapter([errorStream])
     const ctx = await balancedHarness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a-turnend-veto'), { provider: 'mock', model: 'mock' })
@@ -1139,7 +1166,7 @@ describe('turn and step boundary recovery', () => {
     const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
     expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toMatchObject({
       kind: 'error',
-      message: 'provider failed',
+      failure: { message: 'provider failed', code: 'UNKNOWN' },
     })
   })
 
@@ -1175,7 +1202,7 @@ describe('turn and step boundary recovery', () => {
 
   it('a throwing agent/error listener during a step-error path still balances the turn, loop survives', async () => {
     // Listener failure cannot interrupt error finalization or the next turn.
-    const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider 500' } }]
+    const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', failure: { message: 'provider 500', code: 'SERVER' } } }]
     const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')])
     const ctx = await balancedHarness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a-errorlistener'), { provider: 'mock', model: 'mock' })
@@ -1191,7 +1218,11 @@ describe('turn and step boundary recovery', () => {
     expect(c.turnStart).toBe(1)
     expect(c.turnEnd).toBe(1)
     expect(c.stepStart).toBe(c.stepEnd)
-    expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason).toMatchObject({ kind: 'error', step: 1, message: 'provider 500' })
+    expect(c.lastTurnEnd?.type === 'turn/end' && c.lastTurnEnd.data.reason).toMatchObject({
+      kind: 'error',
+      step: 1,
+      failure: { message: 'provider 500', code: 'SERVER' },
+    })
 
     // loop survives: a second turn runs to completion (invariants oracle would
     // throw on its turn/start if turn 1 had been left open).
@@ -1336,7 +1367,7 @@ describe('turn and step boundary recovery', () => {
 
   it('a throwing step/end observer cannot interrupt error finalization', async () => {
     // Observer failure after step/end commit cannot interrupt turn finalization.
-    const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider 500' } }]
+    const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', failure: { message: 'provider 500', code: 'SERVER' } } }]
     const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a-stependthrow'), { provider: 'mock', model: 'mock' })

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

@@ -146,12 +146,22 @@ describe('toError normalization', () => {
     const errors: Error[] = []
     ctx.on('agent/error', (_agent, _turn, _step, error) => void errors.push(error))
 
-    send(agent, 'go')
+    send(agent, 'fails before turn start')
+    send(agent, 'survives as the next item')
     await waitForIdle(ctx, agent)
     expect(errors).toHaveLength(1)
     expect(errors[0]).toMatchObject({ message: 'naked string error', code: 'UNKNOWN' })
-    expect(adapter.requests).toEqual([])
-    expect(agent.session.events.some(event => event.type === 'turn/start' || event.type === 'turn/end')).toBe(false)
+    expect(adapter.requests).toHaveLength(1)
+    const starts = agent.session.events.filter(event => event.type === 'turn/start')
+    const ends = agent.session.events.filter(event => event.type === 'turn/end')
+    const messages = agent.session.events.filter(event => event.type === 'user/message')
+    expect(starts).toHaveLength(1)
+    expect(starts[0]?.type === 'turn/start' && starts[0].data.turn).toBe(1)
+    expect(ends).toHaveLength(1)
+    expect(messages).toHaveLength(1)
+    expect(messages[0]?.type === 'user/message' && messages[0].data.content).toEqual([
+      { type: 'text', text: 'survives as the next item' },
+    ])
   })
 
   it('normalizes non-Error throws from agent/request waterfall via inline toError in runStep catch', async () => {
@@ -177,7 +187,9 @@ describe('toError normalization', () => {
     // String() of { code: 500 } is '[object Object]'
     expect(errors[0]!.message).toBe('[object Object]')
     const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
-    expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error' && turnEnd.data.reason.code).toBe('UNKNOWN')
+    expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error'
+      && ('failure' in turnEnd.data.reason ? turnEnd.data.reason.failure.code : turnEnd.data.reason.code))
+      .toBe('UNKNOWN')
   })
 })
 
@@ -208,7 +220,8 @@ describe('coded error data emission', () => {
     const turnEnd = agent.session.events.find(e => e.type === 'turn/end')
     expect(turnEnd).toBeDefined()
     if (turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind === 'error') {
-      expect(turnEnd.data.reason.code).toBe('RATE_LIMIT')
+      expect('failure' in turnEnd.data.reason ? turnEnd.data.reason.failure.code : turnEnd.data.reason.code)
+        .toBe('RATE_LIMIT')
     }
   })
 })

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

@@ -8,17 +8,17 @@ function resolverPair() {
 }
 
 describe('Inbox', () => {
-  it('enqueues and drains queued messages in FIFO order', () => {
+  it('dequeues one queued message at a time in FIFO order', () => {
     const inbox = new Inbox()
     inbox.enqueue({ content: [{ type: 'text', text: 'first' }], source: { kind: 'user' } })
     inbox.enqueue({ content: [{ type: 'text', text: 'second' }], source: { kind: 'user' } })
     expect(inbox.hasQueued).toBe(true)
 
-    const drained = inbox.drainQueued()
-    expect(drained).toHaveLength(2)
-    expect(drained[0]!.content[0]).toMatchObject({ text: 'first' })
-    expect(drained[1]!.content[0]).toMatchObject({ text: 'second' })
+    expect(inbox.dequeueQueued()?.content[0]).toMatchObject({ text: 'first' })
+    expect(inbox.hasQueued).toBe(true)
+    expect(inbox.dequeueQueued()?.content[0]).toMatchObject({ text: 'second' })
     expect(inbox.hasQueued).toBe(false)
+    expect(inbox.dequeueQueued()).toBeUndefined()
   })
 
   it('pushes and drains steering messages separately from queued', () => {

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

@@ -99,7 +99,6 @@ describe('agent/prompt-submit', () => {
         additionalContexts: [{
           content: [{ type: 'text', text: '<system-reminder>extra ctx</system-reminder>' }],
           source: { kind: 'plugin', plugin: 'test' },
-          envelope: 'raw',
           meta,
         }],
       }))
@@ -113,7 +112,6 @@ describe('agent/prompt-submit', () => {
     expect(userMsg).toBeDefined()
     expect(ctxMsg?.type === 'context/message' && ctxMsg.data.content).toEqual([{ type: 'text', text: '<system-reminder>extra ctx</system-reminder>' }])
     expect(ctxMsg?.type === 'context/message' && ctxMsg.data.source).toEqual({ kind: 'plugin', plugin: 'test' })
-    expect(ctxMsg?.type === 'context/message' && ctxMsg.data.envelope).toBe('raw')
     expect(ctxMsg?.type === 'context/message' && ctxMsg.data.meta).toEqual(meta)
     const sent = JSON.stringify(adapter.requests[0]!.messages)
     expect(sent).toContain('extra ctx')
@@ -179,9 +177,7 @@ describe('agent/prompt-submit', () => {
     expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason).toEqual({ kind: 'rejected', reason: 'blocked by policy' })
   })
 
-  it('a mixed batch records a prompt/blocked for the vetoed prompt while the allowed one runs', async () => {
-    // Blocking one prompt in a mixed batch must persist its reason even though
-    // the allowed prompt keeps the turn from ending rejected.
+  it('adjacent blocked and allowed prompts keep independent turn outcomes', async () => {
     const adapter = new MockAdapter([textResponse('ran once')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -194,13 +190,13 @@ describe('agent/prompt-submit', () => {
     const reasons: TurnEndReason[] = []
     ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
 
-    // both sends land before the loop drains → one batched turn
+    // Both sends land before the driver wakes, but each remains its own turn.
     send(agent, 'secret')
     send(agent, 'safe')
     await waitForIdle(ctx, agent)
 
     const log = events(agent)
-    // the allowed prompt became a user/message and drove exactly one model call
+    // The allowed prompt became a user/message and drove exactly one model call.
     const userMsgs = log.filter(e => e.type === 'user/message')
     expect(userMsgs).toHaveLength(1)
     expect(userMsgs[0]?.type === 'user/message' && userMsgs[0].data.content).toEqual([{ type: 'text', text: 'safe' }])
@@ -212,12 +208,14 @@ describe('agent/prompt-submit', () => {
       content: [{ type: 'text', text: 'secret' }],
       reason: 'policy: no secrets',
     })
-    // the turn did NOT reject — a sibling was allowed — so the boundary reason
-    // alone would not have preserved the block
-    expect(reasons.some(r => r.kind === 'rejected')).toBe(false)
+    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(2)
+    expect(reasons).toEqual([
+      { kind: 'rejected', reason: 'policy: no secrets' },
+      { kind: 'completed' },
+    ])
   })
 
-  it('a throwing prompt-submit listener ends the turn balanced (error), loop survives', async () => {
+  it('a throwing prompt-submit listener ends its turn balanced while an adjacent message survives', async () => {
     const adapter = new MockAdapter([textResponse('after')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -228,20 +226,31 @@ describe('agent/prompt-submit', () => {
       return { kind: 'allow' as const }
     })
     const errors: Error[] = []
+    const reasons: TurnEndReason[] = []
+    const statuses: string[] = []
     ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
+    ctx.on('agent/status', (subject, status) => { if (subject === agent) statuses.push(status) })
+    ctx.on('session/event', (session, event) => {
+      if (session === agent.session && event.type === 'turn/end') reasons.push(event.data.reason)
+    })
 
+    const idle = waitForIdle(ctx, agent)
     send(agent, 'first')
-    await waitForIdle(ctx, agent)
-    expect(errors.map(e => e.message)).toEqual(['prompt hook broke'])
-    // turn balanced
-    const log = events(agent)
-    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
-    expect(log.filter(e => e.type === 'turn/end')).toHaveLength(1)
-
-    // loop survives: a second prompt runs normally
     send(agent, 'second')
-    await waitForIdle(ctx, agent)
-    expect(adapter.requests.length).toBeGreaterThanOrEqual(1)
+    await idle
+    expect(errors.map(e => e.message)).toEqual(['prompt hook broke'])
+    // The failed prompt forms one balanced error turn; the adjacent prompt forms
+    // the following normal turn without an intermediate idle transition.
+    const log = events(agent)
+    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(2)
+    expect(log.filter(e => e.type === 'turn/end')).toHaveLength(2)
+    expect(reasons).toEqual([
+      { kind: 'error', step: 0, message: 'prompt hook broke' },
+      { kind: 'completed' },
+    ])
+    expect(statuses).toEqual(['running', 'idle'])
+    expect(adapter.requests).toHaveLength(1)
+    expect(JSON.stringify(adapter.requests[0]!.messages)).toContain('second')
   })
 })
 
@@ -556,7 +565,6 @@ describe('tool additionalContexts buffering across a step', () => {
         additionalContexts: [{
           content: [{ type: 'text', text: `ctx-${exec.callId}` }],
           source: { kind: 'plugin', plugin: 'p' },
-          envelope: 'raw',
           meta: { callId: exec.callId },
         }],
       }))
@@ -580,7 +588,6 @@ describe('tool additionalContexts buffering across a step', () => {
       .map(b => (b.type === 'text' ? b.text : ''))
     expect(ctxTexts).toEqual(['ctx-c1', 'ctx-c2'])
     const contextEvents = events(agent).filter(e => e.type === 'context/message')
-    expect(contextEvents.map(e => e.type === 'context/message' && e.data.envelope)).toEqual(['raw', 'raw'])
     expect(contextEvents.map(e => e.type === 'context/message' && e.data.meta)).toEqual([{ callId: 'c1' }, { callId: 'c2' }])
   })
 
@@ -591,7 +598,7 @@ describe('tool additionalContexts buffering across a step', () => {
       name: 'composite', description: 'composite', parameters: {},
       async execute(_args, exec) {
         exec.deferContext({ content: [{ type: 'text', text: 'nested-a' }], source: { kind: 'plugin', plugin: 'a' }, meta: { order: 1 } })
-        exec.deferContext({ content: [{ type: 'text', text: 'nested-b' }], source: { kind: 'plugin', plugin: 'b' }, envelope: 'raw', meta: { order: 2 } })
+        exec.deferContext({ content: [{ type: 'text', text: 'nested-b' }], source: { kind: 'plugin', plugin: 'b' }, meta: { order: 2 } })
         return [{ type: 'text', text: 'outer result' }]
       },
     }))

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

@@ -354,14 +354,24 @@ describe('agent loop', () => {
     expect(flat).toContain('change of plans')
   })
 
-  it('steering while idle behaves like send (starts a turn)', async () => {
-    const adapter = new MockAdapter([textResponse('ok')])
+  it('same-tick idle steering inherits one-send-one-turn FIFO behavior', async () => {
+    const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
 
-    agent.steer([{ type: 'text', text: 'hello' }])
-    await waitForIdle(ctx, agent)
-    expect(agent.session.events.some(e => e.type === 'user/message')).toBe(true)
+    const idle = waitForIdle(ctx, agent)
+    agent.steer([{ type: 'text', text: 'first idle steer' }])
+    agent.steer([{ type: 'text', text: 'second idle steer' }])
+    await idle
+
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
+    expect(agent.session.events
+      .filter(event => event.type === 'user/message')
+      .map(event => event.data.content)).toEqual([
+      [{ type: 'text', text: 'first idle steer' }],
+      [{ type: 'text', text: 'second idle steer' }],
+    ])
+    expect(adapter.requests).toHaveLength(2)
   })
 
   it('inject() while idle wraps context in a one-shot turn, visible to the next request', async () => {
@@ -385,10 +395,10 @@ describe('agent loop', () => {
     await waitForIdle(ctx, agent)
     const flat = JSON.stringify(adapter.requests[0]!.messages)
     expect(flat).toContain('file changed: a.ts')
-    expect(flat).toContain('<context source=\\"plugin\\">')
+    expect(flat).not.toContain('<context source=')
   })
 
-  it('inject() can persist raw structured context without the generic context envelope', async () => {
+  it('inject() persists structured context content verbatim with durable hidden meta', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('raw-context'), { provider: 'mock', model: 'mock' })
@@ -401,14 +411,13 @@ describe('agent loop', () => {
 
     agent.inject([{ type: 'text', text }], {
       source: { kind: 'plugin', plugin: 'workspace-context' },
-      envelope: 'raw',
       meta,
     })
     send(agent, 'go')
     await waitForIdle(ctx, agent)
 
     const contextEvent = agent.session.events.find(event => event.type === 'context/message')
-    expect(contextEvent?.type === 'context/message' && contextEvent.data).toMatchObject({ envelope: 'raw', meta })
+    expect(contextEvent?.type === 'context/message' && contextEvent.data).toMatchObject({ meta })
     const requestText = JSON.stringify(adapter.requests[0]!.messages)
     expect(requestText).toContain('Additional instructions from: pkg/AGENTS.md')
     expect(requestText).not.toContain('<context source=')
@@ -432,7 +441,6 @@ describe('agent loop', () => {
         const first = { type: 'text' as const, text: 'mid-turn notice' }
         agent.inject([first], {
           source: { kind: 'plugin', plugin: 'x' },
-          envelope: 'raw',
           meta,
         })
         first.text = 'mutated after inject'
@@ -458,7 +466,6 @@ describe('agent loop', () => {
     expect(contexts).toHaveLength(2)
     expect(result.seq).toBeLessThan(contexts[0]!.seq)
     expect(contexts[0]?.type === 'context/message' && contexts[0].data).toMatchObject({
-      envelope: 'raw',
       meta,
     })
     expect(contexts.flatMap(event => event.type === 'context/message' ? event.data.content : []))
@@ -925,7 +932,149 @@ describe('agent loop', () => {
     expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('completed')
   })
 
-  it('chains queued messages into consecutive turns', async () => {
+  it('keeps same-tick sends in separate turns and checkpoints before the next starts', async () => {
+    const adapter = new MockAdapter([textResponse('first answer'), textResponse('second answer')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+
+    const firstFlush = Promise.withResolvers<undefined>()
+    const releaseFirstFlush = Promise.withResolvers<undefined>()
+    let flushes = 0
+    ctx.on('session/flush', async (session) => {
+      if (session !== agent.session) return
+      flushes += 1
+      if (flushes === 1) {
+        firstFlush.resolve(undefined)
+        await releaseFirstFlush.promise
+      }
+    })
+
+    const turns: number[] = []
+    ctx.on('session/event', (session, event) => {
+      if (session === agent.session && event.type === 'turn/start') turns.push(event.data.turn)
+    })
+
+    const idle = waitForIdle(ctx, agent)
+    send(agent, 'first message')
+    send(agent, 'second message')
+
+    await firstFlush.promise
+    expect(turns).toEqual([1])
+    expect(adapter.requests).toHaveLength(1)
+
+    releaseFirstFlush.resolve(undefined)
+    await idle
+
+    expect(turns).toEqual([1, 2])
+    expect(flushes).toBe(2)
+    expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('first answer')
+    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('second message')
+  })
+
+  it('holds a turn-end listener send behind the closing turn checkpoint', async () => {
+    const adapter = new MockAdapter([textResponse('first answer'), textResponse('second answer')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+
+    const firstFlush = Promise.withResolvers<undefined>()
+    const releaseFirstFlush = Promise.withResolvers<undefined>()
+    let flushes = 0
+    ctx.on('session/flush', async (session) => {
+      if (session !== agent.session) return
+      flushes += 1
+      if (flushes === 1) {
+        firstFlush.resolve(undefined)
+        await releaseFirstFlush.promise
+      }
+    })
+
+    const turns: number[] = []
+    const statuses: string[] = []
+    ctx.on('agent/status', (subject, status) => {
+      if (subject === agent) statuses.push(status)
+    })
+    ctx.on('session/event', (session, event) => {
+      if (session !== agent.session) return
+      if (event.type === 'turn/start') turns.push(event.data.turn)
+      if (event.type === 'turn/end' && event.data.turn === 1) send(agent, 'turn-end listener message')
+    })
+
+    const idle = waitForIdle(ctx, agent)
+    send(agent, 'first message')
+    await firstFlush.promise
+
+    expect(turns).toEqual([1])
+    expect(adapter.requests).toHaveLength(1)
+
+    releaseFirstFlush.resolve(undefined)
+    await idle
+
+    expect(turns).toEqual([1, 2])
+    expect(statuses).toEqual(['running', 'idle'])
+    expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('first answer')
+    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('turn-end listener message')
+  })
+
+  it('keeps a reentrant agent/queued send as the next independent turn', async () => {
+    const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+
+    let nested = false
+    ctx.on('agent/queued', (subject) => {
+      if (subject !== agent || nested) return
+      nested = true
+      send(agent, 'queued listener message')
+    })
+
+    const idle = waitForIdle(ctx, agent)
+    send(agent, 'outer message')
+    await idle
+
+    const turns = agent.session.events.filter(event => event.type === 'turn/start')
+    const messages = agent.session.events
+      .filter(event => event.type === 'user/message')
+      .map(event => event.data.content)
+    expect(turns).toHaveLength(2)
+    expect(messages).toEqual([
+      [{ type: 'text', text: 'outer message' }],
+      [{ type: 'text', text: 'queued listener message' }],
+    ])
+  })
+
+  it('preserves independent turn sources across an adjacent microtask send', async () => {
+    const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+
+    const idle = waitForIdle(ctx, agent)
+    agent.send([{ type: 'text', text: 'user message' }])
+    await Promise.resolve()
+    agent.send(
+      [{ type: 'text', text: 'plugin message' }],
+      { source: { kind: 'plugin', plugin: 'test' } },
+    )
+    await idle
+
+    const triggers = agent.session.events
+      .filter(event => event.type === 'turn/start')
+      .map(event => event.data.trigger)
+    const sources = agent.session.events
+      .filter(event => event.type === 'user/message')
+      .map(event => event.data.source)
+    expect(triggers).toEqual([
+      { kind: 'message', source: { kind: 'user' } },
+      { kind: 'message', source: { kind: 'plugin', plugin: 'test' } },
+    ])
+    expect(sources).toEqual([
+      { kind: 'user' },
+      { kind: 'plugin', plugin: 'test' },
+    ])
+  })
+
+  it('keeps a session-listener send after dequeue in the following turn', async () => {
     const adapter = new MockAdapter([textResponse('first'), textResponse('second')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -948,6 +1097,37 @@ describe('agent loop', () => {
 
     expect(turns).toEqual([1, 2])
     expect(adapter.requests).toHaveLength(2)
+    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('first')
+    expect(JSON.stringify(adapter.requests[1]!.messages)).toContain('second message')
+  })
+
+  it('keeps a model-adapter callback send in the following turn', async () => {
+    const agentRef: { current?: Agent } = {}
+    const adapter = new MockAdapter([
+      () => {
+        const agent = agentRef.current
+        if (agent === undefined) throw new Error('model callback ran before agent setup')
+        send(agent, 'model callback message')
+        return textResponse('first')
+      },
+      textResponse('second'),
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+    agentRef.current = agent
+
+    const idle = waitForIdle(ctx, agent)
+    send(agent, 'outer message')
+    await idle
+
+    const messages = agent.session.events
+      .filter(event => event.type === 'user/message')
+      .map(event => event.data.content)
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
+    expect(messages).toEqual([
+      [{ type: 'text', text: 'outer message' }],
+      [{ type: 'text', text: 'model callback message' }],
+    ])
   })
 
   it('awaits session/flush at turn end (persistence checkpoint)', async () => {

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

@@ -81,6 +81,21 @@ function turnNumbers(agent: Agent): number[] {
     .map(e => (e.data as { turn: number }).turn)
 }
 
+function turnEndNumbers(agent: Agent): number[] {
+  return agent.session.events
+    .filter(e => e.type === 'turn/end')
+    .map(e => (e.data as { turn: number }).turn)
+}
+
+function userMessageCountsByTurn(agent: Agent): number[] {
+  const counts: number[] = []
+  for (const event of agent.session.events) {
+    if (event.type === 'turn/start') counts.push(0)
+    if (event.type === 'user/message') counts[counts.length - 1]! += 1
+  }
+  return counts
+}
+
 /** Assert a status trace is a legal run: idle/running alternating, ending idle. */
 function assertLegalStatusTrace(trace: string[]): void {
   for (let i = 1; i < trace.length; i++) {
@@ -90,7 +105,7 @@ function assertLegalStatusTrace(trace: string[]): void {
 }
 
 describe('agent loop scheduling properties', () => {
-  it('a synchronous burst loses no message and uses strictly increasing turns', async () => {
+  it('a synchronous burst gives every message its own strictly increasing turn', async () => {
     await fc.assert(fc.asyncProperty(
       fc.array(fc.string({ minLength: 1 }), { minLength: 1, maxLength: 6 }),
       async (texts) => {
@@ -105,8 +120,11 @@ describe('agent loop scheduling properties', () => {
 
           // No message lost: every send appears as a user/message, in order.
           expect(userMessageTexts(agent)).toEqual(texts)
-          // A synchronous burst batches into exactly one turn.
-          expect(turnNumbers(agent)).toEqual([1])
+          // This failure-free fixture maps every item to an independent turn.
+          expect(turnNumbers(agent)).toEqual(texts.map((_, i) => i + 1))
+          expect(turnEndNumbers(agent)).toEqual(texts.map((_, i) => i + 1))
+          expect(userMessageCountsByTurn(agent)).toEqual(texts.map(() => 1))
+          expect(trace).toEqual(['running', 'idle'])
           assertLegalStatusTrace(trace)
         } finally {
           await ctx.fiber.dispose()
@@ -137,9 +155,9 @@ describe('agent loop scheduling properties', () => {
     ), { numRuns: 20, timeout: 2000 })
   })
 
-  it('mixed schedule (send, optionally settle) loses no message and orders turns', async () => {
-    // Each step is a (text, settle?) pair: settle=true awaits idle before the
-    // next send (own turn); settle=false sends in the same tick (batches).
+  it('mixed settled and same-tick sends preserve one turn per message', async () => {
+    // Each step optionally waits for idle before the next send; that scheduling
+    // choice must not change the ordinary message-to-turn mapping.
     const stepArb = fc.record({ text: fc.string({ minLength: 1 }), settle: fc.boolean() })
     await fc.assert(fc.asyncProperty(
       fc.array(stepArb, { minLength: 1, maxLength: 6 }),
@@ -158,14 +176,13 @@ describe('agent loop scheduling properties', () => {
           }
           await lastIdle
 
-          // No message lost or reordered, regardless of batching.
+          // No message is lost or reordered, regardless of driver timing.
           expect(userMessageTexts(agent)).toEqual(steps.map(s => s.text))
-          // Turn numbers are a strictly increasing 1..N prefix (N = turn count).
+          // Every item forms one FIFO-ordered turn containing only that message.
           const turns = turnNumbers(agent)
-          expect(turns).toEqual(turns.map((_, i) => i + 1))
-          // Every message landed in some turn; turns never exceed messages.
-          expect(turns.length).toBeLessThanOrEqual(steps.length)
-          expect(turns.length).toBeGreaterThanOrEqual(1)
+          expect(turns).toEqual(steps.map((_, i) => i + 1))
+          expect(turnEndNumbers(agent)).toEqual(turns)
+          expect(userMessageCountsByTurn(agent)).toEqual(steps.map(() => 1))
         } finally {
           await ctx.fiber.dispose()
         }

packages/core/agent-loop/tests/request-recovery.spec.ts

@@ -3,10 +3,12 @@ import { Context } from 'cordis'
 import LlmService, {
   CallId,
   CONTEXT_WINDOW_EXCEEDED_CODE,
+  HarnessError,
   LlmAdapter,
   LlmError,
+  ProviderRequestId,
 } from '@deepseek-ai/dsh-llm'
-import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
+import type { GenerateOptions, LlmFailure, StreamChunk } from '@deepseek-ai/dsh-llm'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools'
@@ -258,16 +260,17 @@ describe('agent post-step and request-error lifecycle', () => {
 
   it.each([
     ['thrown', contextError()],
-    ['in-band', [{ type: 'finish', reason: { kind: 'error', message: 'too large', code: CONTEXT_WINDOW_EXCEEDED_CODE } }] satisfies StreamChunk[]],
+    ['in-band', [{ type: 'finish', reason: { kind: 'error', failure: { message: 'too large', code: CONTEXT_WINDOW_EXCEEDED_CODE, status: 400 } } }] satisfies StreamChunk[]],
   ] as const)('recovers a %s request failure in a new reconstructable step', async (_style, failure) => {
     const adapter = new FailureScriptAdapter([failure, textResponse('recovered')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId(`recover-${_style}`), { provider: 'mock', model: 'mock' })
     const attempts: number[] = []
-    ctx.on('agent/request-error', async (subject, turn, step, error, attempt) => {
+    ctx.on('agent/request-error', async (subject, turn, step, error, facts, history) => {
       expect(subject).toBe(agent)
       expect({ turn, step, code: error.code }).toEqual({ turn: 1, step: 1, code: CONTEXT_WINDOW_EXCEEDED_CODE })
-      attempts.push(attempt)
+      expect(facts.code).toBe(CONTEXT_WINDOW_EXCEEDED_CODE)
+      attempts.push(history.length)
       subject.session.append('context/message', {
         content: [{ type: 'text', text: 'RECOVERY SURFACE MUTATION' }],
         source: { kind: 'plugin', plugin: 'test-recovery' },
@@ -295,7 +298,7 @@ describe('agent post-step and request-error lifecycle', () => {
     const agent = ctx.agentLoop.create(SessionId(`stream-plugin-${_name.replaceAll(' ', '-')}`), { provider: 'mock', model: 'mock' })
     let recoveries = 0
     install(ctx)
-    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => {
       recoveries += 1
       return next()
     })
@@ -326,7 +329,7 @@ describe('agent post-step and request-error lifecycle', () => {
     })
     const agent = ctx.agentLoop.create(SessionId('nested-stream-not-recoverable'), { provider: 'mock', model: 'mock' })
     let recoveries = 0
-    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => {
       recoveries += 1
       return next()
     })
@@ -359,7 +362,7 @@ describe('agent post-step and request-error lifecycle', () => {
       }
       const agent = ctx.agentLoop.create(SessionId(`${boundary}-not-recoverable`), { provider: 'mock', model: 'mock' })
       let recoveries = 0
-      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => {
         recoveries += 1
         return next()
       })
@@ -387,7 +390,7 @@ describe('agent post-step and request-error lifecycle', () => {
       }
       const agent = ctx.agentLoop.create(SessionId(`${failure}-not-recoverable`), { provider: 'mock', model: 'mock' })
       let recoveries = 0
-      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, _signal, next) => {
+      ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, _signal, next) => {
         recoveries += 1
         return next()
       })
@@ -406,7 +409,7 @@ describe('agent post-step and request-error lifecycle', () => {
     const ctx = await harness(makeAdapter(original))
     const agent = ctx.agentLoop.create(SessionId(`identity-${_name.replaceAll(' ', '-')}`), { provider: 'mock', model: 'mock' })
     let seen: Error | undefined
-    ctx.on('agent/request-error', async (_agent, _turn, _step, error, _attempt, _signal, next) => {
+    ctx.on('agent/request-error', async (_agent, _turn, _step, error, _failure, _history, _signal, next) => {
       seen = error
       return next()
     })
@@ -417,12 +420,90 @@ describe('agent post-step and request-error lifecycle', () => {
     expect(seen).toBe(original)
   })
 
+  it('keeps an adapter error with a hostile message accessor on the recovery path', async () => {
+    const original = Object.defineProperty(new HarnessError('provider failed', 'SERVER'), 'message', {
+      get() { throw new Error('SDK message accessor trap') },
+    })
+    const ctx = await harness(new SynchronousDispatchFailureAdapter(original))
+    const agent = ctx.agentLoop.create(SessionId('hostile-message-recovery'), { provider: 'mock', model: 'mock' })
+    let seenError: Error | undefined
+    let seenFailure: LlmFailure | undefined
+    ctx.on('agent/request-error', async (_agent, _turn, _step, error, failure, _history, _signal, next) => {
+      seenError = error
+      seenFailure = failure
+      return next()
+    })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(seenError).toBe(original)
+    expect(seenFailure).toEqual({ message: 'LLM adapter failed', code: 'SERVER' })
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: { reason: { kind: 'error', failure: { message: 'LLM adapter failed', code: 'SERVER' } } },
+    })
+  })
+
+  it('passes structured facts beside the original Error and records its cause chain on exhaustion', async () => {
+    const original = new LlmError('provider busy', 'RATE_LIMIT', {
+      cause: new Error('upstream connection reset'),
+      status: 429,
+      providerRetryAfterMs: 2_000,
+      requestId: ProviderRequestId('req-9'),
+    })
+    Object.freeze(original)
+    const ctx = await harness(new SynchronousDispatchFailureAdapter(original))
+    const agent = ctx.agentLoop.create(SessionId('structured-request-failure'), { provider: 'mock', model: 'mock' })
+    let seenError: Error | undefined
+    let seenFailure: LlmFailure | undefined
+    let seenHistory: readonly LlmFailure[] | undefined
+    ctx.on('agent/request-error', async (
+      _agent, _turn, _step, error, failure, history, _signal, next,
+    ) => {
+      seenError = error
+      seenFailure = failure
+      seenHistory = history
+      return next()
+    })
+
+    send(agent)
+    await waitForIdle(ctx, agent)
+
+    expect(seenError).toBe(original)
+    expect(seenFailure).toEqual({
+      message: 'provider busy',
+      code: 'RATE_LIMIT',
+      status: 429,
+      providerRetryAfterMs: 2_000,
+      requestId: ProviderRequestId('req-9'),
+    })
+    expect(seenHistory).toEqual([])
+    expect(Object.isFrozen(seenHistory)).toBe(true)
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'turn/end',
+      data: {
+        reason: {
+          kind: 'error',
+          step: 1,
+          failure: {
+            message: 'provider busy: upstream connection reset',
+            code: 'RATE_LIMIT',
+            status: 429,
+            providerRetryAfterMs: 2_000,
+            requestId: ProviderRequestId('req-9'),
+          },
+        },
+      },
+    })
+  })
+
   it('classifies iterator construction and explicit NO_ADAPTER as model-request failures', async () => {
     for (const scenario of ['iterator', 'no-adapter'] as const) {
       const ctx = scenario === 'iterator' ? await harness(new IteratorConstructionFailureAdapter()) : await harness()
       const agent = ctx.agentLoop.create(SessionId(`request-boundary-${scenario}`), { provider: 'mock', model: 'mock' })
       let seen = ''
-      ctx.on('agent/request-error', async (_agent, _turn, _step, error, _attempt, _signal, next) => {
+      ctx.on('agent/request-error', async (_agent, _turn, _step, error, _failure, _history, _signal, next) => {
         seen = error.code ?? ''
         return next()
       })
@@ -436,14 +517,17 @@ describe('agent post-step and request-error lifecycle', () => {
     const capped = new FailureScriptAdapter([contextError('first overflow'), contextError('second overflow')])
     const cappedCtx = await harness(capped)
     const cappedAgent = cappedCtx.agentLoop.create(SessionId('retry-cap'), { provider: 'mock', model: 'mock' })
-    const cappedAttempts: number[] = []
-    cappedCtx.on('agent/request-error', async (_agent, _turn, _step, _error, attempt, _signal, next) => {
-      cappedAttempts.push(attempt)
-      return attempt < 1 ? { action: 'retry' } : next()
+    const cappedHistories: string[][] = []
+    cappedCtx.on('agent/request-error', async (
+      _agent, _turn, _step, _error, _failure, history, _signal, next,
+    ) => {
+      const codes = history.map(entry => entry.code)
+      cappedHistories.push(codes)
+      return codes.length < 1 ? { action: 'retry' } : next()
     })
     send(cappedAgent)
     await waitForIdle(cappedCtx, cappedAgent)
-    expect(cappedAttempts).toEqual([0, 1])
+    expect(cappedHistories).toEqual([[], [CONTEXT_WINDOW_EXCEEDED_CODE]])
 
     const reset = new FailureScriptAdapter([
       contextError('first overflow'),
@@ -458,14 +542,16 @@ describe('agent post-step and request-error lifecycle', () => {
       async execute() { return [{ type: 'text', text: 'worked' }] },
     }))
     const resetAgent = resetCtx.agentLoop.create(SessionId('retry-reset'), { provider: 'mock', model: 'mock' })
-    const resetAttempts: { step: number; attempt: number }[] = []
-    resetCtx.on('agent/request-error', async (_agent, _turn, step, _error, attempt, _signal, next) => {
-      resetAttempts.push({ step, attempt })
-      return resetAttempts.length === 1 ? { action: 'retry' } : next()
+    const resetHistories: { step: number; codes: string[] }[] = []
+    resetCtx.on('agent/request-error', async (
+      _agent, _turn, step, _error, _failure, history, _signal, next,
+    ) => {
+      resetHistories.push({ step, codes: history.map(entry => entry.code) })
+      return resetHistories.length === 1 ? { action: 'retry' } : next()
     })
     send(resetAgent)
     await waitForIdle(resetCtx, resetAgent)
-    expect(resetAttempts).toEqual([{ step: 1, attempt: 0 }, { step: 3, attempt: 0 }])
+    expect(resetHistories).toEqual([{ step: 1, codes: [] }, { step: 3, codes: [] }])
   })
 
   it('preserves the original provider error when recovery throws', async () => {
@@ -479,7 +565,7 @@ describe('agent post-step and request-error lifecycle', () => {
 
     expect(agent.session.events.at(-1)).toMatchObject({
       type: 'turn/end',
-      data: { reason: { kind: 'error', message: 'original overflow', code: CONTEXT_WINDOW_EXCEEDED_CODE } },
+      data: { reason: { kind: 'error', failure: { message: 'original overflow', code: CONTEXT_WINDOW_EXCEEDED_CODE } } },
     })
   })
 
@@ -489,7 +575,7 @@ describe('agent post-step and request-error lifecycle', () => {
     const agent = ctx.agentLoop.create(SessionId(`${action}-recovery`), { provider: 'mock', model: 'mock' })
     let entered!: () => void
     const recoveryEntered = new Promise<void>((resolve) => { entered = resolve })
-    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _attempt, signal) => {
+    ctx.on('agent/request-error', async (_agent, _turn, _step, _error, _failure, _history, signal) => {
       entered()
       await new Promise<void>((resolve) => {
         signal.addEventListener('abort', () => { resolve() }, { once: true })

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

@@ -411,7 +411,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
     await ctx.fiber.dispose()
   })
 
-  it('resume of a forked session preserves the parentSession lineage and seed boundary in the header', async () => {
+  it('resume of a forked session preserves the lineage, seed boundary, and delegation depth in the header', async () => {
     // Lifecycle 1: persist a FORKED session (carries parentSession + seedLength
     // in its header) by creating it with a complete-turn seed — the write path
     // materializes the fork (header + seed) on disk.
@@ -423,7 +423,7 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
     const { ctx: ctx1, root } = await persistentHarness(adapter1)
     const forked = ctx1.sessions.create(SessionId('forked-sess'), {
       seed,
-      meta: { cwd: '/w', parentSession: SessionId('parent-sess'), seedLength: seed.length },
+      meta: { cwd: '/w', parentSession: SessionId('parent-sess'), seedLength: seed.length, delegationDepth: 1 },
     })
     await ctx1.parallel('session/flush', forked)
     await ctx1.fiber.dispose()
@@ -447,6 +447,9 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
     expect(a2.session.header.parentSession).toBe('parent-sess')
     expect(a2.session.header.cwd).toBe('/w')
     expect(a2.session.header.seedLength).toBe(seed.length)
+    // The recursion budget survives resume — a dropped depth would let a
+    // resumed child delegate as if it were top-level.
+    expect(a2.session.header.delegationDepth).toBe(1)
     await ctx2.fiber.dispose()
   })