feat(agent-loop): turn-enclosure invariant + post-turn error model

Every session event now lives inside a turn (between turn/start and its
turn/end). The loop records queued user/message events AFTER turn/start;
an idle agent.inject() wraps its context/message in a one-shot injection
turn. This makes the turn the single durability/replay boundary so a
persistence backend can treat anything after the last turn/end as a
crash tail without dropping legitimate between-turn context.

A failure once the turn is already closed (rejecting session/flush, a
throwing agent/turn-end listener) has no in-turn position for a session
error event, so it is reported via agent/error + logger only; the turn
stays balanced. failTurn appends an error event only while the turn is
open.

The dsh-invariants plugin enforces turn-enclosure via a default case:
every non-boundary event type — including plugin-added merge-extensible
keys — must sit inside an open turn or it throws.

Documented in ADR 0017 + architecture.md.

packages/agent-loop/README.md

@@ -41,7 +41,7 @@ One invocation of `runLoop()` drives one agent for its whole lifetime:
 forever:
   wait for queued messages (idle)
   TURN (error-contained):
-    drain queued → session('user/message') → 'turn/start'
+    drain queued → 'turn/start' → session('user/message')
     STEP loop:
       drain steering
       assembly = systemPrompt.assemble()

packages/agent-loop/src/agent.ts

@@ -12,7 +12,7 @@ import type { Agent } from '@deepseek-ai/dsh-agent'
 import type { ContentBlock, MessageSource } from '@deepseek-ai/dsh-llm'
 import type { Session } from '@deepseek-ai/dsh-session'
 import { Inbox } from './inbox.ts'
-import { runLoop } from './loop.ts'
+import { isTurnOpen, lastTurnNumber, runLoop } from './loop.ts'
 
 /**
  * The concrete {@link Agent} implementation owned by the agent-loop plugin.
@@ -73,7 +73,52 @@ export class LoopAgent implements Agent {
 
   inject(content: ContentBlock[], options?: SendOptions): void {
     if (this._status === 'disposed') throw new Error(`agent "${this.id}" is disposed`)
-    this.session.append('context/message', { content, source: this.resolveSource(options) })
+    const source = this.resolveSource(options)
+    if (isTurnOpen(this.session)) {
+      // A turn is open in the LOG (decided from the log, not agent status —
+      // status can be `running` with no turn open): the context/message is
+      // turn-enclosed by that turn, so append it directly.
+      this.session.append('context/message', { content, source })
+      return
+    }
+    // No turn open: wrap the injection in a one-shot turn so every event stays
+    // turn-enclosed (the durability/replay boundary is the turn).
+    const turn = lastTurnNumber(this.session) + 1
+    // Once turn/start enters the log, a turn/end is OWED no matter what — even
+    // if a throwing `session/event` listener escapes from the turn/start append
+    // (Session.append pushes the event BEFORE notifying listeners) or the
+    // context/message append throws (non-serializable content, throwing
+    // listener). The finally re-checks the log via isTurnOpen() and closes the
+    // turn if one was actually opened, so the log never carries a permanently
+    // open injection turn that would corrupt later turns/replay. (If the
+    // turn/start append throws BEFORE pushing — non-serializable trigger, which
+    // can't happen for our fixed trigger — no turn was opened and none is owed.)
+    let turnRecorded = false
+    try {
+      this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } })
+      this.session.append('context/message', { content, source })
+    } finally {
+      // A turn was recorded iff turn/start made it into the log. Close it and
+      // mark it for the durability checkpoint below — which must run even when
+      // an append's listener threw (the turn is balanced and in memory, so it
+      // still needs a flush or a crash before the next turn/dispose loses it).
+      if (isTurnOpen(this.session)) {
+        this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
+        turnRecorded = true
+      }
+      // Checkpoint the one-shot turn for durability, exactly as the loop does at
+      // every turn/end. The loop is NOT running (we are idle), so nothing else
+      // will flush this turn. Fire-and-forget with error containment: inject()
+      // is synchronous, and a persistence backend failing must not throw into
+      // the caller (e.g. a tool-bash task-done callback). Disposal still drains
+      // independently, so a slow flush is safe. In the finally so it also runs
+      // when an append's listener threw (the turn is still balanced + durable).
+      if (turnRecorded) {
+        void Promise.resolve(this.ctx.parallel('session/flush', this.session)).catch((error: unknown) => {
+          this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${String(error)}`)
+        })
+      }
+    }
   }
 
   abort(reason?: string): void {

packages/agent-loop/src/loop.ts

@@ -91,7 +91,7 @@ export interface LoopHandle {
  * forever:
  *   wait for queued messages (idle)
  *   TURN (error-contained — a throwing plugin ends the turn, never the loop):
- *     drain queued → session('user/message'…) → 'turn/start' → emit agent/turn-start
+ *     drain queued → 'turn/start' → session('user/message'…) → emit agent/turn-start
  *     STEP loop:
  *       drain steering → session('steering/message')  ⟵ catches late steering
  *       session('step/start'); emit agent/step-start    ⟵ append before emit (ADR 0003)
@@ -118,24 +118,31 @@ export interface LoopHandle {
  */
 export async function runLoop(ctx: Context, agent: LoopAgent, handle: LoopHandle): Promise<void> {
   const { session } = agent
-  let turn = lastTurnNumber(session) // seeded/forked sessions continue numbering
 
   while (!handle.isDisposed()) {
     await agent.inbox.waitForQueued(handle.disposed)
     if (handle.isDisposed()) break
 
     handle.setStatus('running')
-    turn += 1
+    // Re-derive the turn number from the log each iteration (do NOT keep a local
+    // counter): an idle `agent.inject()` can append its own one-shot turn while
+    // the loop waits above, so the next real turn must continue from whatever
+    // turn number is actually last in the log — a stale counter would collide.
+    const turn = lastTurnNumber(session) + 1
     try {
       await runTurn(ctx, agent, handle, turn)
     } catch (error: unknown) {
-      // Backstop: a throwing emit listener (turn boundaries) or a broken
-      // finalizer must not kill the driver. Record what we can and move on.
+      // Backstop: runTurn rethrows only a PRE-turn throw (the invariant guard
+      // before turn/start) — no turn/start was appended, so no turn is open and
+      // none is owed. A session `error` here would land outside any turn (after
+      // the previous turn/end), where the persistence backend drops it as a
+      // crash tail (ADR 0017). Report via agent/error + the logger only; the
+      // driver survives and moves on.
+      const err = toError(error)
+      ctx.logger.warn(`agent "${agent.id}": turn ${turn} failed before it started: ${err.message}`)
       try {
-        const err = toError(error)
-        session.append('error', { turn, step: 0, ...errorData(err) })
         ctx.emit('agent/error', agent, turn, 0, err)
-      } catch { /* the error path itself is broken; nothing left to do */ }
+      } catch { /* contained: a throwing agent/error listener must not kill the driver */ }
     }
 
     // Steering that arrived too late to join this turn (turn-end listeners,
@@ -151,17 +158,15 @@ export async function runLoop(ctx: Context, agent: LoopAgent, handle: LoopHandle
 async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: number): Promise<void> {
   const { session } = agent
 
-  // --- Pre-turn. A throw here (the invariant guard or a user-message append)
-  // is owed NO turn/end — turn/start has not been appended — so it propagates
-  // to runLoop's backstop untouched.
+  // --- Pre-turn. A throw here (the invariant guard) is owed NO turn/end —
+  // turn/start has not been appended — so it propagates to runLoop's backstop
+  // untouched. The queued messages are drained here but appended AFTER
+  // turn/start (below), so every event in the log lives inside a turn.
   const queued = agent.inbox.drainQueued()
   const first = queued[0]
   /* v8 ignore next 3 -- invariant guard: runLoop only calls runTurn when hasQueued */
   if (!first) throw new Error('runTurn invariant violated: no queued message at turn start')
   const trigger: TurnTrigger = { kind: 'message', source: first.source }
-  for (const message of queued) {
-    session.append('user/message', { content: message.content, source: message.source })
-  }
 
   let reason: TurnEndReason = { kind: 'completed' }
   let step = 0
@@ -186,16 +191,26 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
     }
   }
 
-  // Record a step/turn failure exactly once: append the single `error` event,
-  // set the error reason, and emit agent/error (contained — trap: a throwing
-  // agent/error listener must not re-escape and strand the turn). Disposal and
-  // abort set `reason` directly without calling this (no `error` event for
-  // those — they are not failures).
+  // Record a step/turn failure exactly once: append the single `error` event
+  // (only while the turn is still open — see below), set the error reason, and
+  // emit agent/error (contained — trap: a throwing agent/error listener must not
+  // re-escape and strand the turn). Disposal and abort set `reason` directly
+  // without calling this (no `error` event for those — they are not failures).
   const failTurn = (err: CodedError): void => {
     if (errorReported) return
     errorReported = true
-    session.append('error', { turn, step, ...errorData(err) })
-    reason = { kind: 'error', ...errorData(err) }
+    // Only append the session `error` INSIDE the turn (before turn/end). If the
+    // turn has already ended — the only way here is a throwing agent/turn-end
+    // listener after closeTurn(true) already appended turn/end — appending now
+    // would land the error AFTER the last turn/end, where the persistence
+    // backend treats it as a crash tail and drops it on resume (ADR 0017). In
+    // that case report via agent/error + the logger only; the turn is balanced.
+    if (!turnEnded) {
+      session.append('error', { turn, step, ...errorData(err) })
+      reason = { kind: 'error', ...errorData(err) }
+    } else {
+      ctx.logger.warn(`agent "${agent.id}": agent/turn-end listener threw after turn ${turn} closed: ${err.message}`)
+    }
     try {
       ctx.emit('agent/error', agent, turn, step, err)
     } catch {
@@ -221,6 +236,12 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
     // matter what throws below; the catch + closeTurn guarantee it.
     session.append('turn/start', { turn, trigger })
     turnStarted = true
+    // Record the queued user messages INSIDE the turn (after turn/start), so
+    // every event in the log is turn-enclosed. turn/end is now owed, so a throw
+    // while appending these is caught below and the turn is still closed.
+    for (const message of queued) {
+      session.append('user/message', { content: message.content, source: message.source })
+    }
     ctx.emit('agent/turn-start', agent, turn)
 
     while (true) {
@@ -324,9 +345,20 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
   try {
     await ctx.parallel('session/flush', session)
   } catch (error: unknown) {
+    // The turn is already closed (turn/end appended above) and flush must run
+    // AFTER turn/end to be a checkpoint — so there is no in-turn position left
+    // for a session `error` event. Appending one here would land it after the
+    // last turn/end, where the persistence backend treats it as a crash tail
+    // and drops it on resume (ADR 0017: every event is turn-enclosed). Report
+    // the failure via agent/error + the logger only; persistence keeps the
+    // buffered events for the next flush/dispose, so nothing is lost.
     const err = toError(error)
-    session.append('error', { turn, step, ...errorData(err) })
-    ctx.emit('agent/error', agent, turn, step, err)
+    ctx.logger.warn(`agent "${agent.id}": session/flush failed at turn ${turn}: ${err.message}`)
+    try {
+      ctx.emit('agent/error', agent, turn, step, err)
+    } catch {
+      // contained: a throwing agent/error listener must not escape the loop.
+    }
   }
 }
 
@@ -445,7 +477,21 @@ async function runStep(
 }
 
 /** The last turn number in a (possibly seeded) session log, or 0. */
-function lastTurnNumber(session: Session): number {
+export function lastTurnNumber(session: Session): number {
   const lastStart = session.events.findLast(event => event.type === 'turn/start')
   return lastStart?.data.turn ?? 0
 }
+
+/**
+ * Whether a turn is currently open in the session log (a `turn/start` with no
+ * matching later `turn/end`). Decided from the LOG, not agent status: status
+ * can be `running` while no turn is open (an `agent/status` listener firing
+ * before `turn/start`, or the post-`turn/end` flush window before status
+ * returns to idle), so status is not a reliable open-turn signal. Used by
+ * `inject()` to choose between appending into an open turn vs. wrapping the
+ * injection in its own one-shot turn (ADR 0017).
+ */
+export function isTurnOpen(session: Session): boolean {
+  const last = session.events.findLast(e => e.type === 'turn/start' || e.type === 'turn/end')
+  return last?.type === 'turn/start'
+}

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

@@ -1,4 +1,4 @@
-import { describe, expect, it } from 'vitest'
+import { describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import { AgentId } from '@deepseek-ai/dsh-agent'
 import LlmService from '@deepseek-ai/dsh-llm'
@@ -82,6 +82,80 @@ describe('LoopAgent', () => {
     expect(() => { agent.inject([{ type: 'text', text: 'too late' }]) }).toThrow('disposed')
   })
 
+  it('inject() decides enclosure from the LOG (open turn), not agent status', async () => {
+    const adapter = new MockAdapter([textResponse('ok')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+
+    // 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('context/message')
+
+    // 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
+  })
+
+  it('idle inject() contains a failing flush (logs, does not throw into the caller)', 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('a1', { model: 'mock' })
+
+    // 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
+    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 () => {
+    const adapter = new MockAdapter([textResponse('ok')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { 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() 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('a1', { 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)
+  })
+
   it('steer() when idle falls through to send() and starts a turn', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)

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

@@ -57,9 +57,10 @@ describe('agent loop', () => {
     expect(order).toEqual(['agent/turn-start', 'agent/step-start', 'agent/step-end', 'agent/turn-end'])
 
     const types = agent.session.events.map(e => e.type)
-    // user message recorded before turn/start, assembled message + usage present
-    expect(types[0]).toBe('user/message')
-    expect(types[1]).toBe('turn/start')
+    // turn/start opens the turn, THEN the queued user message is recorded inside
+    // it (every event is turn-enclosed), then assembled message + usage.
+    expect(types[0]).toBe('turn/start')
+    expect(types[1]).toBe('user/message')
     expect(types).toContain('assistant/message')
     expect(types).toContain('usage')
     expect(types.at(-1)).toBe('turn/end')
@@ -199,16 +200,22 @@ describe('agent loop', () => {
     expect(agent.session.events.some(e => e.type === 'user/message')).toBe(true)
   })
 
-  it('inject() appends context visible to the next request without starting a turn', async () => {
+  it('inject() while idle wraps context in a one-shot turn, visible to the next request', async () => {
     const adapter = new MockAdapter([textResponse('ok')])
     const ctx = await harness(adapter)
     const agent = ctx.agentLoop.create('a1', { model: 'mock' })
 
     agent.inject([{ type: 'text', text: 'file changed: a.ts' }], { source: { kind: 'plugin', plugin: 'watcher' } })
-    // no turn started
+    // The idle inject records a self-contained turn (turn/start → context/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
 
     send(agent, 'go')
     await waitForIdle(ctx, agent)
@@ -217,6 +224,38 @@ describe('agent loop', () => {
     expect(flat).toContain('<context source=\\"plugin\\">')
   })
 
+  it('inject() while running appends into the open turn (no extra synthetic turn)', async () => {
+    const adapter = new MockAdapter([
+      toolCallResponse('c1', 'noticer', {}, 'calling'),
+      textResponse('done'),
+    ])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a1', { model: 'mock' })
+    // A tool that injects mid-execution: at this point the agent is running, so
+    // inject must append the context/message into the ALREADY-open turn rather
+    // than wrap it in its own one-shot turn.
+    ctx.tools.register(defineTool({
+      name: 'noticer',
+      description: 'injects a notice',
+      parameters: {},
+      async execute() {
+        agent.inject([{ type: 'text', text: 'mid-turn notice' }], { source: { kind: 'plugin', plugin: 'x' } })
+        return [{ type: 'text', text: 'ok' }]
+      },
+    }))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    // Exactly ONE turn ran (no synthetic injection turn), and the mid-turn
+    // context/message sits inside it.
+    const turnStarts = agent.session.events.filter(e => e.type === 'turn/start')
+    expect(turnStarts).toHaveLength(1)
+    const ts0 = turnStarts[0]!
+    expect(ts0.type === 'turn/start' && ts0.data.trigger.kind).toBe('message')
+    expect(agent.session.events.some(e => e.type === 'context/message')).toBe(true)
+  })
+
   it('agent/turn-continuation can force-continue (/loop pattern) and force-stop', async () => {
     // force-continue: model never calls tools, but a plugin forces 3 steps
     const adapter = new MockAdapter([

packages/agent-loop/tests/review-fixes.spec.ts

@@ -854,6 +854,39 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
     expect(adapter.requests).toHaveLength(1)
   })
 
+  it('a throwing turn-end listener on a SUCCESSFUL turn leaves no event after turn/end (loadable log)', async () => {
+    // Regression: a normal turn completes, closeTurn(true) appends turn/end and
+    // emits agent/turn-end whose listener throws. The error must NOT be appended
+    // as a session event after turn/end — that would sit past the commit
+    // boundary and be dropped as a crash tail on resume (ADR 0017). It is
+    // surfaced via agent/error instead, and the log's last event is turn/end.
+    const adapter = new MockAdapter([textResponse('done'), textResponse('next ok')])
+    const ctx = await balancedHarness(adapter)
+    const agent = ctx.agentLoop.create('a-tend', { model: 'mock' })
+
+    let threw = false
+    ctx.on('agent/turn-end', () => { if (!threw) { threw = true; throw new Error('boom turn-end') } })
+    const errors: Error[] = []
+    ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    const c = boundaryCounts(agent)
+    expect(c.turnEnd).toBe(1)
+    expect(c.errors).toBe(0) // NO session error event (it would be post-turn/end)
+    expect(agent.session.events.at(-1)?.type).toBe('turn/end') // last event is the boundary
+    expect(errors.map(e => e.message)).toEqual(['boom turn-end']) // surfaced via agent/error
+    // The whole log is loadable (nothing dropped): a fresh replay sees the turn.
+    const replay = new Session(SessionId('replay'), [...agent.session.events])
+    expect(replay.deriveMessages().map(m => m.role)).toEqual(['user', 'assistant'])
+
+    // loop survives.
+    send(agent, 'again')
+    await waitForIdle(ctx, agent)
+    expect(boundaryCounts(agent).turnEnd).toBe(2)
+  })
+
   it('a step error followed by a throwing turn-end listener logs the error exactly once (no double-report)', async () => {
     // The step fails (finish-error) → failTurn records ONE error and sets the
     // error reason. closeTurn(true) then appends turn/end and emits