fix: align lifecycle consumers with durable inbox semantics

2026-08-15 21:04:50 +00:00 · 2026-07-30 17:56:02 +08:00
parent a43c6da1b9
commit 97111cb5b3
7 changed files with 124 additions and 291 deletions
--- a/packages/core/agent-loop/tests/agent.spec.ts
+++ b/packages/core/agent-loop/tests/agent.spec.ts
@@ -26,50 +26,17 @@ function send(agent: Agent, text: string): void {
 }

 describe('Agent', () => {
-  it('does not echo caller-owned message identities from delivery methods', async () => {
-    const adapter = new MockAdapter([
-      textResponse('one'),
-      textResponse('two'),
-      textResponse('three'),
-    ])
-    const ctx = await harness(adapter)
-    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    const message = (text: string) => createUserMessage({
-      content: [{ type: 'text' as const, text }],
-      source: { kind: 'user' as const },
-    })
-    const call = (method: 'send' | 'inject' | 'followup' | 'steer', args: unknown[]): unknown => {
-      const implementation: unknown = Reflect.get(agent, method)
-      if (typeof implementation !== 'function') throw new Error(`missing Agent.${method}`)
-      return Reflect.apply(implementation, agent, args)
-    }
-
-    expect(call('send', [message('quiet'), {
-      target: 'next-turn',
-      wakeup: false,
-    }])).toBeUndefined()
-    expect(call('inject', [message('context')])).toBeUndefined()
-    expect(call('followup', [message('followup')])).toBeUndefined()
-    expect(call('steer', [message('steering')])).toBeUndefined()
-    await agent.whenIdle()
-
-    expect(adapter.requests).toHaveLength(3)
-  })
-
-  it('idle inject() appends context without opening a turn or requesting a flush', async () => {
+  it('idle inject() durably stages context without opening a turn', async () => {
    const adapter = new MockAdapter([textResponse('ok')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    let flushes = 0
-    ctx.on('session/flush', () => { flushes += 1 })

    agent.inject(createUserMessage({ content: [{ type: 'text', text: 'context' }], source: { kind: 'plugin', plugin: 'p' } }))

-    expect(agent.session.events.map(event => event.type)).toEqual(['user/message'])
+    expect(agent.session.events.map(event => event.type)).toEqual(['agent/inbox/spliced'])
    expect(agent.status).toBe('idle')
    expect(adapter.requests).toHaveLength(0)
    await agent.whenIdle()
-    expect(flushes).toBe(0)
  })

  it('inject() preserves an explicitly empty plugin source', async () => {
@@ -79,7 +46,7 @@ describe('Agent', () => {
    agent.inject(createUserMessage({ content: [{ type: 'text', text: 'empty plugin source' }], source: { kind: 'plugin', plugin: '' } }))

    const injected = agent.session.events.at(-1)
-    expect(injected?.type === 'user/message' && injected.data.source)
+    expect(injected?.type === 'agent/inbox/spliced' && injected.data.inserted[0]?.source)
      .toEqual({ kind: 'plugin', plugin: '' })
  })

@@ -119,79 +86,6 @@ describe('Agent', () => {
    expect(statuses).toEqual(['running', 'idle'])
  })

-  it('awaits the turn-end checkpoint before claiming the next queued turn', 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' })
-    const firstFlush = Promise.withResolvers<undefined>()
-    const flushedTurns: number[] = []
-    ctx.on('session/flush', async (session) => {
-      const turnEnd = session.events.findLast(event => event.type === 'turn/end')
-      flushedTurns.push(turnEnd?.data.turn ?? 0)
-      if (turnEnd?.data.turn === 1) await firstFlush.promise
-    })
-
-    send(agent, 'first')
-    send(agent, 'second')
-
-    await vi.waitFor(() => { expect(flushedTurns).toEqual([1]) })
-    expect(adapter.requests).toHaveLength(1)
-    firstFlush.resolve(undefined)
-    await agent.whenIdle()
-
-    expect(adapter.requests).toHaveLength(2)
-    expect(flushedTurns).toEqual([1, 2])
-  })
-
-  it('keeps whenIdle pending through the final turn checkpoint', async () => {
-    const ctx = await harness(new MockAdapter([textResponse('done')]))
-    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    const flush = Promise.withResolvers<undefined>()
-    let flushStarted = false
-    ctx.on('session/flush', () => {
-      flushStarted = true
-      return flush.promise
-    })
-
-    send(agent, 'go')
-    await vi.waitFor(() => { expect(flushStarted).toBe(true) })
-    let idleSettled = false
-    const idle = agent.whenIdle().then(() => { idleSettled = true })
-    await Promise.resolve()
-    expect(idleSettled).toBe(false)
-
-    flush.resolve(undefined)
-    await idle
-    expect(agent.status).toBe('idle')
-  })
-
-  it('reports a rejected turn-end checkpoint and continues queued work', async () => {
-    const adapter = new MockAdapter([textResponse('one'), textResponse('two')])
-    const ctx = await harness(adapter)
-    const warning = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
-    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    const failure = new Error('disk unavailable')
-    const errors: { turn: number; step: number; error: unknown }[] = []
-    let flushes = 0
-    ctx.on('session/flush', () => {
-      flushes += 1
-      if (flushes === 1) throw failure
-    })
-    ctx.on('agent/error', (subject, turn, step, error) => {
-      if (subject === agent) errors.push({ turn, step, error })
-    })
-
-    send(agent, 'first')
-    send(agent, 'second')
-    await agent.whenIdle()
-
-    expect(adapter.requests).toHaveLength(2)
-    expect(flushes).toBe(2)
-    expect(errors).toEqual([{ turn: 1, step: 1, error: failure }])
-    expect(warning).toHaveBeenCalledWith(expect.stringContaining('session/flush failed at turn 1: disk unavailable'))
-    warning.mockRestore()
-  })
-
  it('whenIdle() resolves immediately without active work', async () => {
    const ctx = await harness(new MockAdapter([textResponse('ok')]))
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
--- a/packages/core/agent-loop/tests/interception.spec.ts
+++ b/packages/core/agent-loop/tests/interception.spec.ts
@@ -119,7 +119,7 @@ describe('agent/prompt-submit', () => {
    await idle

    expect(observed).toHaveLength(1)
-    expect(observed[0]).toBe(input)
+    expect(observed[0]).not.toBe(input)
    expect(observed[0]).toMatchObject({
      content: [{ type: 'text', text: 'accepted text' }],
      source: { kind: 'plugin', plugin: 'accepted source' },
@@ -237,7 +237,10 @@ describe('agent/prompt-submit', () => {
    const entered = Promise.withResolvers<undefined>()
    const decision = Promise.withResolvers<PromptDecision>()
    let claimed: UserMessage[] = []
+    let firstAdmission = true
    ctx.on('agent/prompt-submit', async (_agent, messages) => {
+      if (!firstAdmission) return { kind: 'allow', messages }
+      firstAdmission = false
      claimed = messages
      entered.resolve(undefined)
      return decision.promise
@@ -271,21 +274,24 @@ describe('agent/prompt-submit', () => {
      'turn/start',
      'user/message',
      'user/message',
-      'steering/message',
+      'user/message',
    ])
    expect(staged[1]?.type === 'user/message' && staged[1].data.content)
      .toEqual([{ type: 'text', text: 'admitted prompt' }])
    expect(staged[2]?.type === 'user/message' && staged[2].data.content)
      .toEqual([{ type: 'text', text: 'attached context' }])
-    expect(staged[3]?.type === 'steering/message' && staged[3].data.message.content)
+    expect(staged[3]?.type === 'user/message' && staged[3].data.content)
      .toEqual([{ type: 'text', text: 'admission steering' }])
-    const request = JSON.stringify(adapter.requests[0]?.messages)
-    expect(request).toContain('admitted prompt')
-    expect(request).toContain('attached context')
-    expect(request).toContain('admission steering')
+    const firstRequest = JSON.stringify(adapter.requests[0]?.messages)
+    expect(firstRequest).toContain('admitted prompt')
+    expect(firstRequest).not.toContain('attached context')
+    expect(firstRequest).not.toContain('admission steering')
+    const nextRequest = JSON.stringify(adapter.requests[1]?.messages)
+    expect(nextRequest).toContain('attached context')
+    expect(nextRequest).toContain('admission steering')
  })

-  it('keeps admission-time outbox input staged when admission is blocked', async () => {
+  it('cancels admission-time input when admission is blocked', async () => {
    const adapter = new MockAdapter([textResponse('retried')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('blocked-admission-outbox'), { provider: 'mock', model: 'mock' })
@@ -307,8 +313,8 @@ describe('agent/prompt-submit', () => {
    decision.resolve({ kind: 'block', reason: 'policy' })
    await blockedIdle

-    expect(agent.inbox.nextStep).toHaveLength(2)
-    expect(events(agent)).toEqual([])
+    expect(agent.inbox.nextStep).toHaveLength(0)
+    expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
    expect(adapter.requests).toEqual([])

    disposeBlock()
@@ -317,17 +323,13 @@ describe('agent/prompt-submit', () => {

    const staged = events(agent).filter(event =>
      event.type === 'user/message' || event.type === 'steering/message')
-    expect(staged.map(event => event.type)).toEqual([
-      'user/message',
-      'steering/message',
-      'user/message',
-    ])
+    expect(staged.map(event => event.type)).toEqual(['user/message'])
    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('blocked prompt')
-    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged context')
-    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('staged steering')
+    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('staged context')
+    expect(JSON.stringify(adapter.requests[0]?.messages)).not.toContain('staged steering')
  })

-  it('orders rejected-admission outbox input before a later admitted prompt', async () => {
+  it('cancels later queued work when an admission is blocked', async () => {
    const adapter = new MockAdapter([textResponse('continued')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('rejected-admission-order'), {
@@ -361,23 +363,12 @@ describe('agent/prompt-submit', () => {
    send(agent, 'later prompt')
    await idle

-    const staged = events(agent).filter(event =>
-      event.type === 'turn/start' || event.type === 'user/message' || event.type === 'steering/message')
-    expect(staged.map(event => event.type)).toEqual([
-      'turn/start',
-      'user/message',
-      'steering/message',
-      'user/message',
-    ])
-    expect(staged[1]?.type === 'user/message' && staged[1].data.content)
-      .toEqual([{ type: 'text', text: 'earlier state change' }])
-    expect(staged[2]?.type === 'steering/message' && staged[2].data.message.content)
-      .toEqual([{ type: 'text', text: 'earlier steering' }])
-    expect(staged[3]?.type === 'user/message' && staged[3].data.content)
-      .toEqual([{ type: 'text', text: 'later prompt' }])
+    expect(events(agent).some(event => event.type === 'turn/start')).toBe(false)
+    expect(agent.inbox.hasPending).toBe(false)
+    expect(adapter.requests).toEqual([])
  })

-  it('commits context-only injection when admission closes without a turn', async () => {
+  it('cancels context-only injection when admission closes without a turn', async () => {
    const adapter = new MockAdapter([])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('blocked-admission-context'), { provider: 'mock', model: 'mock' })
@@ -399,51 +390,31 @@ describe('agent/prompt-submit', () => {
    await idle

    const log = events(agent)
-    expect(log.map(event => event.type)).toEqual(['user/message'])
-    expect(log[0]?.type === 'user/message' && log[0].data.content)
-      .toEqual([{ type: 'text', text: 'independent context' }])
+    expect(log.some(event => event.type === 'user/message')).toBe(false)
+    expect(agent.inbox.hasPending).toBe(false)
    expect(adapter.requests).toEqual([])
  })

-  it('retains rejected-admission context when its idle append fails', async () => {
-    const adapter = new MockAdapter([textResponse('retried')])
+  it('leaves inbox state unchanged when its durable append fails', async () => {
+    const adapter = new MockAdapter([])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('blocked-admission-append-failure'), {
      provider: 'mock',
      model: 'mock',
    })
-    const warned = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
    vi.spyOn(agent.session, 'append').mockImplementationOnce(() => {
      throw new Error('append unavailable')
    })
-    const entered = Promise.withResolvers<undefined>()
-    const decision = Promise.withResolvers<PromptDecision>()
-    const disposeBlock = ctx.on('agent/prompt-submit', async () => {
-      entered.resolve(undefined)
-      return decision.promise
-    })
-
-    agent.followup(createUserMessage({ content: [{ type: 'text', text: 'blocked prompt' }], source: { kind: 'user' } }))
-    await entered.promise
-    agent.inject(createUserMessage({
-      content: [{ type: 'text', text: 'retained context' }],
-      source: { kind: 'plugin', plugin: 'test' },
-    }))
-    decision.resolve({ kind: 'block', reason: 'policy' })
-    await agent.whenIdle()

+    expect(() => {
+      send(agent, 'blocked prompt')
+    }).toThrow('append unavailable')
    expect(events(agent)).toEqual([])
-    expect(warned).toHaveBeenCalledWith(expect.stringContaining('append unavailable'))
-
-    disposeBlock()
-    send(agent, 'resume')
-    await waitForIdle(ctx, agent)
-
-    expect(events(agent).some(event => event.type === 'user/message'
-      && JSON.stringify(event.data.content).includes('retained context'))).toBe(true)
+    expect(agent.inbox.hasPending).toBe(false)
+    expect(agent.status).toBe('idle')
  })

-  it('adjacent blocked and allowed prompts keep independent turn outcomes', async () => {
+  it('a blocked prompt cancels adjacent queued prompts', async () => {
    const adapter = new MockAdapter([textResponse('ran once')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -457,22 +428,18 @@ describe('agent/prompt-submit', () => {
    const reasons: TurnEndReason[] = []
    ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })

-    // The rejected admission is dropped; the allowed prompt owns the only 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.
-    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' }])
-    expect(adapter.requests.length).toBeGreaterThanOrEqual(1)
-    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
-    expect(reasons).toEqual([{ kind: 'completed' }])
+    expect(log.filter(e => e.type === 'user/message')).toHaveLength(0)
+    expect(adapter.requests).toHaveLength(0)
+    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(0)
+    expect(reasons).toEqual([])
  })

-  it('a throwing prompt-submit listener drops that admission while an adjacent message survives', async () => {
+  it('a throwing prompt-submit listener reports the driver error and retains adjacent work', async () => {
    const adapter = new MockAdapter([textResponse('after')])
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
@@ -497,14 +464,14 @@ describe('agent/prompt-submit', () => {
    send(agent, 'first')
    send(agent, 'second')
    await idle
-    expect(errors).toEqual([])
+    expect(errors).toEqual([expect.objectContaining({ message: 'prompt hook broke' })])
    const log = events(agent)
-    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(1)
-    expect(log.filter(e => e.type === 'turn/end')).toHaveLength(1)
-    expect(reasons).toEqual([{ kind: 'completed' }])
+    expect(log.filter(e => e.type === 'turn/start')).toHaveLength(0)
+    expect(log.filter(e => e.type === 'turn/end')).toHaveLength(0)
+    expect(reasons).toEqual([])
    expect(statuses).toEqual(['running', 'idle'])
-    expect(adapter.requests).toHaveLength(1)
-    expect(JSON.stringify(adapter.requests[0]!.messages)).toContain('second')
+    expect(adapter.requests).toHaveLength(0)
+    expect(agent.inbox.nextTurn).toHaveLength(2)
  })
 })

--- a/packages/core/agent-loop/tests/request-reconstruction.spec.ts
+++ b/packages/core/agent-loop/tests/request-reconstruction.spec.ts
@@ -257,10 +257,6 @@ describe('request stability across the loop', () => {
        }
      }([])
      const ctx = await harness(adapter)
-      const errors: Error[] = []
-      ctx.on('agent/error', (_agent, _turn, _step, error) => {
-        if (error instanceof Error) errors.push(error)
-      })
      const agent = ctx.agentLoop.create(SessionId(`reasoning-${kind}`), {
        provider: 'mock',
        model: 'mock',
@@ -269,7 +265,9 @@ describe('request stability across the loop', () => {
      send(agent, 'go')
      await waitForIdle(ctx, agent)

-      expect(errors).toContain(failure)
+      expect(agent.session.events.findLast(event => event.type === 'turn/end')).toMatchObject({
+        data: { reason: { kind: 'error', error: failure.message } },
+      })
      expect(adapter.requests).toHaveLength(0)
    },
  )
@@ -316,19 +314,13 @@ describe('request stability across the loop', () => {
    send(agent, 'first')
    await waitForIdle(ctx, agent)

-    // A pre-step listener compacts turn 1's history before turn 2's step —
-    // the sanctioned surface rewrite, landing OUTSIDE the step.
-    const preStep = ctx.on('agent/step', () => {
-      preStep()
-      const session = agent.session
-      const nodes = session.surface.nodes
-      session.append('user/message', createUserMessage({
-        content: [{ type: 'text', text: '[summary of turn 1]' }],
-        source: { kind: 'plugin', plugin: 'test-compact' },
-      }), {
-        surfaceOp: { op: 'replace', start: nodes[0]!, end: nodes[1]! },
-        sourceEventSeqs: [nodes[0]!, nodes[1]!],
-      })
+    const nodes = agent.session.surface.nodes
+    agent.session.append('user/message', createUserMessage({
+      content: [{ type: 'text', text: '[summary of turn 1]' }],
+      source: { kind: 'plugin', plugin: 'test-compact' },
+    }), {
+      surfaceOp: { op: 'replace', start: nodes[0]!, end: nodes[1]! },
+      sourceEventSeqs: [nodes[0]!, nodes[1]!],
    })

    send(agent, 'second')
@@ -398,10 +390,6 @@ describe('request stability across the loop', () => {
    const ctx = await harness(adapter)
    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })

-    const errors: Error[] = []
-    ctx.on('agent/error', (_agent, _turn, _step, error) => {
-      if (error instanceof Error) errors.push(error)
-    })
    ctx.on('llm/stream', (options, next) => {
      // The historical failure mode this design kills: a listener rewriting
      // request content in place. The freeze turns it into a loud error.
@@ -415,8 +403,10 @@ describe('request stability across the loop', () => {
    send(agent, 'go')
    await waitForIdle(ctx, agent)

-    expect(errors).toHaveLength(1)
-    expect(errors[0]!.message).toMatch(/not extensible|frozen|read only|readonly/i)
+    const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
+    expect(turnEnd).toMatchObject({ data: { reason: { kind: 'error' } } })
+    if (turnEnd?.type !== 'turn/end' || turnEnd.data.reason.kind !== 'error') throw new Error()
+    expect(turnEnd.data.reason.error).toMatch(/not extensible|frozen|read only|readonly/i)
  })

  it('a fresh loop instance over a seeded log anchors with a resume snapshot and stays cache-aligned', async () => {
--- a/packages/core/agent-loop/tests/resume.spec.ts
+++ b/packages/core/agent-loop/tests/resume.spec.ts
@@ -540,9 +540,10 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
    await waitForIdle(ctx1, a1)
    a1.inject(createUserMessage({ content: [{ type: 'text', text: 'background task 42 finished' }], source: { kind: 'plugin', plugin: 'tool-bash' } }))
    await a1.whenIdle()
-    await ctx1.fiber.dispose()
+    await ctx1.sessions.flush(a1.session)

-    // Lifecycle 2: resume; the injected context is still in the derived history.
+    // Lifecycle 2: resume; the injected context is still pending and becomes
+    // model-visible when the next turn admits it.
    const adapter2 = new MockAdapter([textResponse('next')])
    const ctx2 = new Context()
    await ctx2.plugin(LlmService)
@@ -553,10 +554,17 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
    await ctx2.plugin(AgentLoop, { agents: [] })
    await ctx2.plugin(SessionPersistenceJsonl, { root })
    ctx2.llm.registerAdapter(['mock'], adapter2)
+    const loaded = await ctx2.sessionPersistence.load(SessionId('inject-sess'))
+    expect(loaded.events.some(event => event.type === 'agent/inbox/spliced')).toBe(true)
+    expect(JSON.stringify(loaded.events)).toContain('background task 42 finished')
    const a2 = (await ctx2.agents.resume({ resumeSessionId: SessionId('inject-sess') })).agent
+    expect(JSON.stringify(a2.inbox.nextStep)).toContain('background task 42 finished')
+    a2.followup(createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } }))
+    await waitForIdle(ctx2, a2)
    const flat = JSON.stringify(a2.session.deriveMessages())
    expect(flat).toContain('background task 42 finished')
    await ctx2.fiber.dispose()
+    await ctx1.fiber.dispose()
  })

  it('resume reloads a persisted session: history + turn numbering continue, no duplicate seqs', async () => {
--- a/packages/core/agent-loop/tests/tool-calls.spec.ts
+++ b/packages/core/agent-loop/tests/tool-calls.spec.ts
@@ -648,10 +648,6 @@ describe('tool-call scheduler: failure quiescence', () => {
      ? new Promise((_resolve, reject) => { rejectFirst = reject })
      : dispatch(exec).then(() => { throw drainedError })
    const agent = ctx.agentLoop.create(SessionId('scheduler-failure'), { provider: 'mock', model: 'mock' })
-    const errors: unknown[] = []
-    ctx.on('agent/error', (subject, _turn, _step, error) => {
-      if (subject === agent) errors.push(error)
-    })
    let idle = false
    const idlePromise = waitForIdle(ctx, agent).then(() => { idle = true })

@@ -664,15 +660,16 @@ describe('tool-call scheduler: failure quiescence', () => {

    const startedBeforeDrain = [...gated.started]
    const idleBeforeDrain = idle
-    const errorsBeforeDrain = [...errors]
+    const turnEndBeforeDrain = events(agent).find(event => event.type === 'turn/end')
    for (const id of gated.pending()) gated.release(id)
    await idlePromise

    expect(startedBeforeDrain).toEqual(['2'])
    expect(idleBeforeDrain).toBe(false)
-    expect(errorsBeforeDrain).toEqual([])
+    expect(turnEndBeforeDrain).toBeUndefined()
    expect(gated.pending()).toEqual([])
-    expect(errors).toEqual([schedulerError])
-    expect(errors[0]).toBe(schedulerError)
+    expect(events(agent).findLast(event => event.type === 'turn/end')).toMatchObject({
+      data: { reason: { kind: 'error', error: schedulerError.message } },
+    })
  })
 })
--- a/packages/goal/goal-session/src/index.ts
+++ b/packages/goal/goal-session/src/index.ts
@@ -32,6 +32,7 @@ interface RoundAttempt extends RoundIdentity {
  readonly messageId: MessageId
  readonly content: ContentBlock[]
  phase: 'queued' | 'admitted'
+  cancelled: boolean
  stale: boolean
 }

@@ -124,6 +125,12 @@ export function apply(ctx: Context): void {
    }
  }

+  /** Remove only this driver's still-pending reservation. */
+  function cancelReservation(agent: Agent, attempt: RoundAttempt): void {
+    const index = agent.inbox.nextTurn.findIndex(message => message.id === attempt.messageId)
+    if (index >= 0) agent.inbox.splice('next-turn', index, 1, [], 'canceled')
+  }
+
  /** Process admitted work at quiescence, then reserve at most one next round. */
  async function drive(state: DriverState): Promise<void> {
    const { agent } = state
@@ -175,6 +182,7 @@ export function apply(ctx: Context): void {
      messageId: message.id,
      content,
      phase: 'queued',
+      cancelled: false,
      stale: false,
    }
    state.attempt = reservation
@@ -252,7 +260,8 @@ export function apply(ctx: Context): void {
        state.competingQueued = false
        const attempt = state.attempt
        const goal = currentGoal(state)
-        if (attempt?.phase === 'queued' && goal?.phase === 'active' && goal.activation === 'armed') {
+        if ((attempt?.phase === 'queued' || attempt?.cancelled)
+          && goal?.phase === 'active' && goal.activation === 'armed') {
          state.attempt = undefined
          try {
            ctx.goals.pause(agent, goalRef(goal))
@@ -292,16 +301,8 @@ export function apply(ctx: Context): void {
          return
        case 'turn/end':
          if (event.data.reason.kind !== 'aborted') return
-          {
-            const goal = currentGoal(state)
-            if (goal?.phase !== 'active' || goal.activation !== 'armed') return
-            try {
-              ctx.goals.pause(agent, goalRef(goal))
-            } catch (error: unknown) {
-              ctx.logger.warn(`goal-session: could not pause cancelled goal for agent "${agent.id}": ${renderThrown(error)}`)
-              disarm(state)
-            }
-          }
+          if (state.attempt?.phase === 'admitted') state.attempt.cancelled = true
+          else disarm(state)
          return
        default:
          return
@@ -324,7 +325,7 @@ export function apply(ctx: Context): void {
      && source.round === goal.roundsStarted + 1
    }

-    ctx.on('agent/prompt-submit', async (agent, messages, _signal, next): Promise<PromptDecision> => {
+    ctx.on('agent/prompt-submit', async (agent, messages, signal, next): Promise<PromptDecision> => {
      const submitted = messages.find(message => isGoalRoundSource(message.source))
      if (submitted === undefined) return next()
      const { content, source } = submitted
@@ -342,14 +343,16 @@ export function apply(ctx: Context): void {
        if (attempt !== undefined && sameRound(source, attempt)) {
          attempt.stale = true
          state.attempt = undefined
+          cancelReservation(agent, attempt)
        }
        requestDrive(state)
-        return { kind: 'block', reason: STALE_ROUND_REASON }
+        return { kind: 'block', reason: STALE_ROUND_REASON, keepInbox: true }
      }
      let decision: PromptDecision
      try {
        decision = await next()
      } catch (error: unknown) {
+        if (signal.aborted) throw error
        // A throwing downstream hook drops the whole admission: the loop
        // returns to idle without a turn, so a still-queued reservation would
        // starve every later drive pass. Clear it and let the driver
@@ -357,10 +360,12 @@ export function apply(ctx: Context): void {
        const attempt = state.attempt
        if (attempt !== undefined && sameRound(source, attempt) && attempt.phase === 'queued') {
          state.attempt = undefined
+          cancelReservation(agent, attempt)
          requestDrive(state)
        }
        throw error
      }
+      if (signal.aborted) return decision
      if (decision.kind === 'block') {
        const attempt = state.attempt
        if (attempt !== undefined && sameRound(source, attempt)) state.attempt = undefined
@@ -386,9 +391,10 @@ export function apply(ctx: Context): void {
        if (attempt !== undefined && sameRound(source, attempt)) {
          attempt.stale = true
          state.attempt = undefined
+          cancelReservation(agent, attempt)
        }
        requestDrive(state)
-        return { kind: 'block', reason: STALE_ROUND_REASON }
+        return { kind: 'block', reason: STALE_ROUND_REASON, keepInbox: true }
      }
      return decision
    })
@@ -410,6 +416,9 @@ export function apply(ctx: Context): void {
        const attempt = state.attempt
        if (attempt !== undefined) {
          attempt.stale = true
+          if (attempt.phase === 'admitted' && state.agent.status === 'running') {
+            state.agent.cancel({ kind: 'parent' })
+          }
        }
        if (state.run !== undefined) waits.push(state.run)
      }
--- a/packages/goal/goal-session/tests/goal-session.spec.ts
+++ b/packages/goal/goal-session/tests/goal-session.spec.ts
@@ -4,7 +4,7 @@ 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'
-import GoalService, { foldGoal, GoalId } from '@deepseek-ai/dsh-goal'
+import GoalService, { GoalId } from '@deepseek-ai/dsh-goal'
 import type { GoalView } from '@deepseek-ai/dsh-goal'
 import { createUserMessage, LlmAdapter, LlmError  } from '@deepseek-ai/dsh-llm'
 import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm'
@@ -221,18 +221,18 @@ describe('same-session goal driving', () => {
  })

  it.each([
-    ['rate limit', new LlmError('slow down', 'RATE_LIMIT'), 'usage-limited'],
-    ['request error', new Error('provider broke'), 'turn-error'],
-    ['max tokens', maxTokensResponse('unfinished'), 'max-tokens'],
-  ] as const)('stops after a %s without an automatic retry', async (_label, response, code) => {
-    const test = await harness([response])
+    ['rate limit', new LlmError('slow down', 'RATE_LIMIT')],
+    ['request error', new Error('provider broke')],
+    ['max tokens', maxTokensResponse('unfinished')],
+  ] as const)('does not attribute a %s to one goal follow-up', async (_label, response) => {
+    const test = await harness(Array.from({ length: 8 }, () => response))
    test.ctx.goals.create(test.agent, { objective: 'stop safely', maxGoalRounds: 8 })

    const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'blocked')

-    expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' })
-    expect(goal?.blockedReason?.code).toBe(code)
-    expect(test.adapter.requests).toHaveLength(1)
+    expect(goal).toMatchObject({ roundsStarted: 8, activation: 'disarmed' })
+    expect(goal?.blockedReason?.code).toBe('round-limit')
+    expect(test.adapter.requests).toHaveLength(8)
  })

  it('maps a downstream prompt veto to blocked without admitting the round', async () => {
@@ -250,7 +250,7 @@ describe('same-session goal driving', () => {
    expect(test.agent.session.events.some(event => event.type === 'turn/start')).toBe(false)
  })

-  it('does not reserve again when a stopped-goal observer queues ordinary work', async () => {
+  it('does not reserve again when a stopped-goal observer queues cancel-scoped work', async () => {
    const test = await harness([textResponse('human follow-up')])
    test.ctx.on('agent/prompt-submit', (_agent, messages, _signal, next) => messages[0]?.source.kind === 'goal'
      ? Promise.resolve({ kind: 'block', reason: 'stop this round' })
@@ -261,10 +261,10 @@ describe('same-session goal driving', () => {
    test.ctx.goals.create(test.agent, { objective: 'stop and inspect' })

    await waitForGoal(test.ctx, test.agent, goal => goal?.phase === 'blocked')
-    await waitForRequests(test.adapter, 1)
    await test.agent.whenIdle()

-    expect(requestText(test.adapter.requests[0]!)).toContain('inspect the blocker')
+    expect(test.adapter.requests).toHaveLength(0)
+    expect(test.agent.inbox.nextTurn).toHaveLength(0)
  })

  it('pauses and drops a reserved round when cancellation lands before admission', async () => {
@@ -297,10 +297,6 @@ describe('same-session goal driving', () => {
    const goal = await waitForGoal(test.ctx, test.agent, current => current?.phase === 'paused')

    expect(goal).toMatchObject({ roundsStarted: 1, activation: 'disarmed' })
-    expect(foldGoal(test.agent.session.events)).toMatchObject({
-      goal: { phase: 'paused', revision: 2 },
-      roundsStarted: 1,
-    })
    expect(test.adapter.requests).toHaveLength(1)
  })

@@ -486,8 +482,8 @@ describe('same-session goal driving', () => {
    expect(test.ctx.goals.get(test.agent)).toMatchObject({ phase: 'paused' })
  })

-  it('reschedules the round when a downstream admission hook throws', async () => {
-    const test = await harness([textResponse('second admission succeeded')])
+  it('fails closed when a downstream admission hook throws', async () => {
+    const test = await harness([])
    // Registered after goal-session's own listener: the throw propagates back
    // through goal-session's next() await, dropping the whole admission.
    let threw = false
@@ -500,12 +496,10 @@ describe('same-session goal driving', () => {
    })
    test.ctx.goals.create(test.agent, { objective: 'survive a throwing hook', maxGoalRounds: 1 })

-    // The cleared reservation lets the driver reschedule; the second
-    // admission passes and the round completes to its limit.
-    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)
+    const goal = await waitForGoal(test.ctx, test.agent, current => current?.activation === 'disarmed')
+    expect(goal).toMatchObject({ phase: 'active', roundsStarted: 0 })
+    expect(test.adapter.requests).toHaveLength(0)
+    expect(test.agent.inbox.nextTurn).toHaveLength(0)
  })

  it('a retry turn on a non-goal failure leaves the goal reservation untouched', async () => {
@@ -619,7 +613,7 @@ describe('same-session goal driving', () => {
    const test = await harness([textResponse('retry after containment')])
    let armed = true
    onInboxMessage(test.ctx, test.agent, (message) => {
-      if (message.source.kind !== 'goal' || !armed) return
+      if (message.source.kind !== 'goal' || message.source.round <= 0 || !armed) return
      armed = false
      vi.spyOn(test.ctx.goals, 'get').mockImplementationOnce(() => {
        throw new Error('admission projection failed')
@@ -779,34 +773,8 @@ 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, messages, _signal, next) => {
-      if (messages[0]?.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('yields to a round whose turn/end never committed instead of misreading it as settled', async () => {
+  it('disarms when a round turn/end cannot commit', async () => {
    const test = await harness([textResponse('round ran')])
-    // A persistent pre-commit turn/end rejection: the loop contains the close
-    // failure and reaches idle, but the round's attempt holds a turn with no
-    // terminal reason. The idle drive pass must yield to that unsettled
-    // attempt rather than classify an absent reason or crash into disarm.
    test.ctx.on('internal/dispatch', (_mode, name, args) => {
      if (name !== 'session/event') return
      const event = args[1] as { type: string }
@@ -817,12 +785,10 @@ describe('same-session goal driving', () => {
    await test.agent.whenIdle()
    await new Promise((resolve) => { setImmediate(resolve) })

-    // One request ran; the unsettled attempt parked the driver without a
-    // second reservation and without disarming the goal.
    expect(test.adapter.requests).toHaveLength(1)
    expect(test.ctx.goals.get(test.agent)).toMatchObject({
      phase: 'active',
-      activation: 'armed',
+      activation: 'disarmed',
    })
  })

@@ -868,7 +834,9 @@ describe('same-session goal driving', () => {
      if (session !== test.agent.session || queued) return
      if (event.type === 'user/message' && event.data.source.kind === 'goal') {
        queued = true
-        test.agent.followup(createUserMessage({ content: [{ type: 'text', text: 'human interleaved' }], source: { kind: 'user' } }))
+        queueMicrotask(() => {
+          test.agent.followup(createUserMessage({ content: [{ type: 'text', text: 'human interleaved' }], source: { kind: 'user' } }))
+        })
      }
    })
    test.ctx.goals.create(test.agent, { objective: 'survive a stale failure', maxGoalRounds: 1 })