Merge branch 'split/turn-enclosure' into split/session-persistence

packages/agent-loop/src/agent.ts

@@ -93,29 +93,51 @@ export class LoopAgent implements Agent {
     // 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).
+      // Close the turn if turn/start made it into the log. Contain a throwing
+      // turn/end listener: Session.append pushes before notifying, so a throw
+      // here still leaves turn/end in the log (the turn is balanced) — swallow
+      // it so it neither replaces the original exception nor skips the flush
+      // decision below. (It surfaces through the flush path is not needed; the
+      // turn-balance contract is what matters and it holds.)
       if (isTurnOpen(this.session)) {
-        this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
-        turnRecorded = true
+        try {
+          this.session.append('turn/end', { turn, reason: { kind: 'completed' } })
+        } catch {
+          // turn/end is already in the log (pushed before the listener threw),
+          // so the turn is balanced; the throw is the listener's bug.
+        }
       }
+      // Decide the durability checkpoint from the LOG, not a flag: a turn was
+      // recorded iff this turn's turn/start is logged (it may have been closed
+      // by a throwing-listener turn/end above, which still counts). A
+      // `turnRecorded` boolean set after append('turn/end') would be skipped by
+      // a throwing turn/end listener, losing the flush for a balanced in-memory
+      // turn (crash before the next turn/dispose would drop the idle injection).
+      const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
       // 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).
+      // independently, so a slow flush is safe. A flush failure is reported via
+      // agent/error (step 0 — the idle-injection convention, there is no real
+      // step) AND the logger, mirroring the loop's post-turn/end flush path so
+      // plugins monitoring agent/error see idle-injection persistence failures
+      // too. A throwing agent/error listener is contained.
       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)}`)
+          const err = error instanceof Error ? error : new Error(String(error))
+          this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`)
+          try {
+            this.ctx.emit('agent/error', this, turn, 0, err)
+          } catch {
+            // contained: the failure is already logged; a throwing agent/error
+            // listener must not escape this fire-and-forget catch.
+          }
         })
       }
     }

packages/agent-loop/src/loop.ts

@@ -170,7 +170,6 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
 
   let reason: TurnEndReason = { kind: 'completed' }
   let step = 0
-  let turnStarted = false
   let turnEnded = false
   let stepOpen = false
   let errorReported = false
@@ -236,9 +235,10 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
 
   try {
     // --- Turn boundary. Once turn/start is appended, a turn/end is owed no
-    // matter what throws below; the catch + closeTurn guarantee it.
+    // matter what throws below; the catch + closeTurn guarantee it (the catch
+    // decides "owed" from the log via isTurnOpen, so even a throwing turn/start
+    // listener — append pushes before notifying — still gets its turn/end).
     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.
@@ -320,12 +320,20 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
     // Normal / inline-error loop exit: close the turn and notify.
     closeTurn(true)
   } catch (error: unknown) {
-    // A pre-turn throw (turn/start append) is owed no turn/end — rethrow to
-    // the backstop. Otherwise a boundary emit (turn-start, step-start, the
-    // normal-path turn-end) or other unhandled throw escaped: close any open
-    // step, choose the reason (disposal wins only if no error was reported),
-    // record the error, and close the turn WITHOUT re-emitting agent/turn-end.
-    if (!turnStarted) throw error
+    // Decide whether this turn was ever opened from the LOG, not a flag.
+    // Session.append pushes the event BEFORE notifying session/event listeners,
+    // so a throwing listener on the `turn/start` append leaves turn/start in the
+    // log even though execution never reached the lines after that append.
+    // Gating on a "turn started" boolean would skip turn/end and leave a
+    // permanently OPEN turn that poisons the next turn/replay (ADR 0017). We
+    // check the log for THIS turn's turn/start: present means a turn/end is owed
+    // (or was already appended — closeTurn/failTurn are idempotent, so running
+    // them again is a safe no-op that still preserves the disposed/error reason
+    // chosen below). Absent means the turn/start append threw BEFORE its push (a
+    // non-serializable trigger — impossible for our fixed trigger); nothing was
+    // opened, so rethrow to the runLoop backstop.
+    const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn)
+    if (!turnStartLogged) throw error
     closeStep()
     // Choose the close reason. Disposal wins only if no error was already
     // reported: a turn disposed mid-step sets reason=disposed in the step-error