Merge remote-tracking branch 'origin/master' into fix/subagent-depth-budget

# Conflicts: # docs/persistence-catalog.md
2026-07-20 18:11:59 +08:00
parent ff6359a0cd 09bcaaa66d
commit 51ba53766e
34 changed files with 795 additions and 240 deletions
--- a/packages/core/agent-loop/tests/cancel.spec.ts
+++ b/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 () => {
--- a/packages/core/agent-loop/tests/contract-regressions.spec.ts
+++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts
@@ -632,15 +632,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')
      }
    })
@@ -648,18 +653,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)

@@ -675,11 +687,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 () => {
--- a/packages/core/agent-loop/tests/coverage-edges.spec.ts
+++ b/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 () => {
--- a/packages/core/agent-loop/tests/inbox.spec.ts
+++ b/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', () => {
--- a/packages/core/agent-loop/tests/interception.spec.ts
+++ b/packages/core/agent-loop/tests/interception.spec.ts
@@ -177,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' })
@@ -192,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' }])
@@ -210,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' })
@@ -226,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')
  })
 })

--- a/packages/core/agent-loop/tests/loop.spec.ts
+++ b/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 () => {
@@ -922,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' })
@@ -945,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 () => {
--- a/packages/core/agent-loop/tests/properties.spec.ts
+++ b/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()
        }