refactor(agent-loop): separate injected context from turns

packages/core/agent-loop/tests/MIGRATION.md

@@ -25,8 +25,8 @@ New emit: `agent/idle (agent, turn, reason: IdleReason)` fires once per closed t
 
 - `send()` — unchanged (queued FIFO, one turn each).
 - `steer()` while running — enters the outbox; taken whole at the next step
-  boundary. Steering left when the turn closes becomes a queued prompt.
-  There is NO terminal-stop discard of steering anymore.
+  boundary. A turn failure leaves untaken steering staged without waking the
+  agent; `retry()` or a later prompt takes it.
 - `inject()` while the machine is busy — enters the outbox (a `context/message`
   appears at the NEXT step boundary, not immediately). While idle — writes a
   one-shot turn (`turn/start(injection)` + `context/message` + `turn/end`) and
@@ -43,10 +43,10 @@ New emit: `agent/idle (agent, turn, reason: IdleReason)` fires once per closed t
 - `kick()` runs SYNCHRONOUSLY from `send()` when idle: status flips to
   `running` inside the `send()` call. There is no parked driver loop, no
   waitForQueued, no microtask collection window.
-- One `run()` = one turn. The idle tail (`idle()`) runs after turn/end +
-  flush: it sets `busy=false`, emits `agent/idle`, requeues leftover steering,
-  then either kicks the next turn or settles `whenIdle` waiters and flips
-  status to `idle`. Status stays `running` continuously across queued turns.
+- One `run()` = one turn. After `turn/end`, it emits `agent/idle`, then either
+  starts the next waking queued prompt or flips status to `idle`. Residual
+  outbox input does not wake the agent. Status stays `running` continuously
+  across queued turns.
 - `step/end` is appended INSIDE the step (after tools + the in-step outbox
   drain), before `agent/continue` runs. The old `post-step → step/end`
   window no longer exists.

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

@@ -83,130 +83,71 @@ describe('Agent', () => {
     await ctx.fiber.dispose()
   })
 
-  it('inject() decides enclosure from the LOG (open turn), not agent status', async () => {
+  it('idle inject() appends context and flushes 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' })
+    const release = Promise.withResolvers<void>()
+    let flushes = 0
+    ctx.on('session/flush', async (session) => {
+      if (session !== agent.session) return
+      flushes += 1
+      await release.promise
+    })
 
-    // Simulate an OPEN turn in the log while the agent is idle (status is not a
-    // reliable open-turn signal). inject must append into that open turn, NOT
-    // wrap a new one.
-    agent.session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
-    agent.inject([{ type: 'text', text: 'mid' }], { source: { kind: 'plugin', plugin: 'p' } })
-    expect(agent.session.events.filter(e => e.type === 'turn/start')).toHaveLength(1)
-    expect(agent.session.events.at(-1)!.type).toBe('user/message')
+    agent.inject([{ type: 'text', text: 'context' }], { source: { kind: 'plugin', plugin: 'p' } })
+    expect(agent.session.events.map(event => event.type)).toEqual(['user/message'])
+    expect(agent.status).toBe('idle')
+    expect(adapter.requests).toHaveLength(0)
 
-    // Close the turn; now inject must wrap its own one-shot injection turn.
-    agent.session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
-    agent.inject([{ type: 'text', text: 'after' }], { source: { kind: 'plugin', plugin: 'p' } })
-    const starts = agent.session.events.filter(e => e.type === 'turn/start')
-    expect(starts).toHaveLength(2)
-    const last = starts[1]!
-    expect(last.type === 'turn/start' && last.data.trigger.kind).toBe('injection')
-    expect(agent.session.events.at(-1)!.type).toBe('turn/end') // turn-enclosed
+    let idle = false
+    const settled = agent.whenIdle().then(() => { idle = true })
+    await Promise.resolve()
+    expect(flushes).toBe(1)
+    expect(idle).toBe(false)
+    release.resolve()
+    await settled
   })
 
   it('inject() defaults its source to an empty plugin, never user', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    agent.session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
     agent.inject([{ type: 'text', text: 'no explicit source' }])
     const injected = agent.session.events.at(-1)!
     expect(injected.type === 'user/message' && injected.data.source).toEqual({ kind: 'plugin', plugin: '' })
+    await agent.whenIdle()
   })
 
-  it('idle inject() contains a failing flush (logs, does not throw into the caller)', async () => {
+  it('idle inject() contains a failing flush without inventing an agent turn error', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
-    // A persistence-like listener whose flush rejects.
     ctx.on('session/flush', () => { throw new Error('disk gone') })
     const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
     const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
+    const errors: Error[] = []
+    ctx.on('agent/error', (_agent, _turn, _step, error) => { errors.push(error) })
 
-    // inject() is synchronous and fires a fire-and-forget flush; a rejecting
-    // flush must be contained (logged), never thrown into the caller.
     expect(() => { agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) }).not.toThrow()
-    await new Promise(r => setTimeout(r, 20)) // let the contained flush settle
+    await agent.whenIdle()
+    expect(errors).toEqual([])
+    expect(agent.session.events.map(event => event.type)).toEqual(['user/message'])
     expect(warn).toHaveBeenCalledWith(expect.stringContaining('flush after idle injection failed'))
     warn.mockRestore()
   })
 
-  it('idle inject() closes its one-shot turn AND still checkpoints even if the append throws', async () => {
+  it('idle inject() does not flush input rejected before append', 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 })
 
-    // Non-serializable injected content makes Session.append throw AFTER
-    // turn/start was recorded. The turn/end must still be appended (finally),
-    // AND the durability checkpoint must still fire — the balanced turn is in
-    // memory and a crash before the next turn/dispose would otherwise lose it.
     expect(() => {
       agent.inject([{ type: 'text', text: 'x', bad: 1n } as never], { source: { kind: 'plugin', plugin: 'p' } })
     }).toThrow(/non-JSON-serializable/)
-    const types = agent.session.events.map(e => e.type)
-    expect(types).toEqual(['turn/start', 'turn/end']) // balanced, no open turn
-    await new Promise(r => setTimeout(r, 10)) // let the fire-and-forget flush run
-    expect(flushes).toBe(1) // checkpoint fired despite the throw
-  })
-
-  it('idle inject() still checkpoints when a listener throws on the synthetic turn/end', 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 })
-    // Session contains a throwing post-commit turn/end observer. The accepted
-    // boundary still triggers the idle injection's durability checkpoint.
-    let threw = false
-    ctx.on('session/event', (_s, event) => {
-      if (!threw && event.type === 'turn/end') { threw = true; throw new Error('boom turn/end') }
-    })
-
-    expect(() => { agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) }).not.toThrow()
-    const types = agent.session.events.map(e => e.type)
-    expect(types).toEqual(['turn/start', 'user/message', 'turn/end']) // balanced
-    await new Promise(r => setTimeout(r, 10))
-    expect(flushes).toBe(1) // checkpoint fired despite the throwing turn/end listener
-  })
-
-  it('idle inject() reports a failing flush via agent/error (step 0) AND the logger', async () => {
-    const adapter = new MockAdapter([textResponse('ok')])
-    const ctx = await harness(adapter)
-    // A non-Error rejection exercises the String() normalization branch.
-    ctx.on('session/flush', () => { throw 'disk gone' })
-    const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined)
-    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-    const errors: { turn: number; step: number; message: string }[] = []
-    ctx.on('agent/error', (_a, turn, step, error) => void errors.push({ turn, step, message: error.message }))
-
-    agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } })
-    await new Promise(r => setTimeout(r, 20)) // let the contained flush settle
-
-    // Reported via agent/error (step 0 — the idle-injection convention) so
-    // plugins monitoring agent/error see idle-injection persistence failures,
-    // mirroring the loop's post-turn/end flush path. A non-Error throw is
-    // normalized to an Error.
-    expect(errors).toEqual([{ turn: 1, step: 0, message: 'disk gone' }])
-    expect(warn).toHaveBeenCalledWith(expect.stringContaining('flush after idle injection failed'))
-    warn.mockRestore()
-  })
-
-  it('idle inject() with a non-serializable source opens no turn (nothing to close)', async () => {
-    const adapter = new MockAdapter([textResponse('ok')])
-    const ctx = await harness(adapter)
-    const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' })
-
-    // A non-serializable source makes the turn/start append throw BEFORE the
-    // event is pushed (Session.append validates before push), so NO turn opens.
-    // The finally's isTurnOpen() guard sees no open turn and appends nothing —
-    // the log stays empty, not left with a dangling turn/start.
-    expect(() => {
-      agent.inject([{ type: 'text', text: 'x' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never })
-    }).toThrow(/non-JSON-serializable/)
     expect(agent.session.events).toHaveLength(0)
+    expect(flushes).toBe(0)
   })
 
   it('steer() when idle falls through to send() and starts a turn', async () => {

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

@@ -114,53 +114,6 @@ describe('agent/prompt-submit', () => {
     expect(sent).toContain('extra ctx')
   })
 
-  it('bakes prompt-prefix contexts and a request delimiter into one durable user message', async () => {
-    const adapter = new MockAdapter([textResponse('ok')])
-    const ctx = await harness(adapter)
-    const agent = ctx.agentLoop.create(SessionId('prefixed'), { provider: 'mock', model: 'mock' })
-
-    ctx.on('agent/prompt-submit', async (_agent, _content, _source, _signal, next): Promise<PromptDecision> => {
-      const downstream = await next()
-      return downstream.kind === 'block'
-        ? downstream
-        : { ...downstream, content: [{ type: 'text', text: 'rewritten request' }] }
-    })
-    agent.send([{ type: 'text', text: 'original request' }], {
-      contexts: [{
-        content: [{ type: 'text', text: 'untrusted prefix' }],
-        source: { kind: 'plugin', plugin: 'prefix' },
-        placement: 'prompt-prefix',
-      }],
-    })
-    await waitForIdle(ctx, agent)
-
-    const log = events(agent)
-    const user = log.find(event => event.type === 'user/message')
-    expect(user?.type === 'user/message' && user.data).toEqual({
-      content: [
-        { type: 'text', text: 'untrusted prefix' },
-        { type: 'text', text: '\n\n## My request:\n' },
-        { type: 'text', text: 'rewritten request' },
-      ],
-      source: { kind: 'user' },
-      envelope: {
-        displayContent: [{ type: 'text', text: 'rewritten request' }],
-        prefixContexts: [{
-          source: { kind: 'plugin', plugin: 'prefix' },
-        }],
-      },
-    })
-    expect(log.some(event => event.type === 'user/message' && event.data.source.kind === 'plugin')).toBe(false)
-    expect(adapter.requests[0]?.messages.at(-1)).toEqual({
-      role: 'user',
-      content: [
-        { type: 'text', text: 'untrusted prefix' },
-        { type: 'text', text: '\n\n## My request:\n' },
-        { type: 'text', text: 'rewritten request' },
-      ],
-    })
-  })
-
   it('runs pre-step after prompt rewrites and injected context become durable', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
@@ -198,9 +151,7 @@ describe('agent/prompt-submit', () => {
     const reasons: TurnEndReason[] = []
     ctx.on('session/event', (_s, event: SessionEvent) => { if (event.type === 'turn/end') reasons.push(event.data.reason) })
 
-    agent.send([{ type: 'text', text: 'do something' }], {
-      contexts: [{ content: [{ type: 'text', text: 'must be dropped' }], source: { kind: 'plugin', plugin: 'test' } }],
-    })
+    agent.send([{ type: 'text', text: 'do something' }])
     await waitForIdle(ctx, agent)
 
     // the model was never called

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

@@ -374,25 +374,52 @@ describe('agent loop', () => {
     expect(adapter.requests).toHaveLength(2)
   })
 
-  it('inject() while idle wraps context in a one-shot turn, visible to the next request', async () => {
+  it('keeps steering staged after a failed step until retry', async () => {
+    const adapter = new MockAdapter([textResponse('recovered')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create(SessionId('failed-steering'), { provider: 'mock', model: 'mock' })
+    let fail = true
+    ctx.on('agent/step', (subject) => {
+      if (subject !== agent || !fail) return
+      fail = false
+      subject.steer([{ type: 'text', text: 'pending steering' }])
+      throw new Error('step failed')
+    })
+
+    send(agent, 'prompt')
+    await waitForIdle(ctx, agent)
+
+    expect(adapter.requests).toHaveLength(0)
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
+    expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(false)
+
+    const idle = waitForIdle(ctx, agent)
+    agent.retry()
+    await idle
+
+    expect(adapter.requests).toHaveLength(1)
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(2)
+    expect(agent.session.events.some(event => event.type === 'steering/message')).toBe(true)
+    expect(JSON.stringify(adapter.requests[0]?.messages)).toContain('pending steering')
+  })
+
+  it('inject() while idle appends 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' })
 
     agent.inject([{ type: 'text', text: 'file changed: a.ts' }], { source: { kind: 'plugin', plugin: 'watcher' } })
-    // The idle inject records a self-contained turn (turn/start → user/message
-    // → turn/end) so the event stays turn-enclosed, but does NOT run the model.
-    await new Promise(r => setTimeout(r, 20))
     expect(agent.status).toBe('idle')
     expect(adapter.requests).toHaveLength(0)
-    const injectedTurn = agent.session.events.filter(e => e.type === 'turn/start')
-    expect(injectedTurn).toHaveLength(1)
-    const it0 = injectedTurn[0]!
-    expect(it0.type === 'turn/start' && it0.data.trigger.kind).toBe('injection')
-    expect(agent.session.events.at(-1)!.type).toBe('turn/end') // turn-enclosed
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(0)
+    expect(agent.session.events.at(-1)).toMatchObject({
+      type: 'user/message',
+      data: { source: { kind: 'plugin', plugin: 'watcher' } },
+    })
 
     send(agent, 'go')
     await waitForIdle(ctx, agent)
+    expect(agent.session.events.filter(event => event.type === 'turn/start')).toHaveLength(1)
     const flat = JSON.stringify(adapter.requests[0]!.messages)
     expect(flat).toContain('file changed: a.ts')
     expect(flat).not.toContain('<context source=')

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

@@ -454,16 +454,15 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
   })
 
   it('an idle inject() is flushed durably on its own (survives without explicit flush/dispose)', async () => {
-    // Idle injection creates and flushes a one-shot turn. No explicit flush or
-    // clean disposal follows, so disk presence proves its own checkpoint ran.
+    // No clean disposal follows, so disk presence proves the idle injection's
+    // own checkpoint ran without a synthetic turn.
     const adapter1 = new MockAdapter([textResponse('answer')])
     const { ctx: ctx1, root } = await persistentHarness(adapter1)
     const a1 = (await ctx1.agents.create({ sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent
     a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
     await waitForIdle(ctx1, a1)
     a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } })
-    // Let inject()'s fire-and-forget flush settle (NO explicit flush/dispose).
-    await new Promise(r => setTimeout(r, 30))
+    await a1.whenIdle()
 
     // A SEPARATE backend reads the on-disk log — proving the inject persisted
     // itself, not a later dispose drain.
@@ -476,16 +475,14 @@ describe('the session-persistence Agent Note: AgentLoop factory create/resume',
     await ctx1.fiber.dispose()
   })
 
-  it('an idle inject() survives persist + resume (turn-enclosed, not dropped as crash tail)', async () => {
-    // Turn enclosure keeps idle context out of crash-tail repair, so it must
-    // survive persistence and resume.
+  it('an idle inject() survives persist + resume without a synthetic turn', async () => {
     const adapter1 = new MockAdapter([textResponse('answer')])
     const { ctx: ctx1, root } = await persistentHarness(adapter1)
     const a1 = (await ctx1.agents.create({ sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent
     a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
     await waitForIdle(ctx1, a1)
     a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } })
-    await ctx1.sessions.flush(a1.session)
+    await a1.whenIdle()
     await ctx1.fiber.dispose()
 
     // Lifecycle 2: resume; the injected context is still in the derived history.