diff --git a/docs/adr/0017-turn-enclosure-invariant.md b/docs/adr/0017-turn-enclosure-invariant.md new file mode 100644 index 0000000000..f4b88420ea --- /dev/null +++ b/docs/adr/0017-turn-enclosure-invariant.md @@ -0,0 +1,38 @@ +# ADR 0017: Every session event is enclosed in a turn + +Status: accepted (2026-06-15) + +## Context + +A durable session-persistence backend (added in a companion change) uses the **turn** as its crash-recovery boundary: `load` returns events only up to the last complete `turn/end`, and the first post-load `append` truncates whatever follows as a never-committed crash tail. This is safe only if nothing *legitimately* durable can sit after the last `turn/end`. + +That assumption did not hold. Two paths recorded events outside any turn: + +1. **Queued user messages.** The loop drained queued messages and appended `user/message` *before* `turn/start` — so a turn's own prompt sat in the gap between the previous `turn/end` and the next `turn/start`. +2. **Idle context injection.** `agent.inject()` appends a `context/message` directly. Its real production caller is `dsh-tool-bash`, which injects a background-task completion notice from `ctx.bash.onTaskDone` — a callback that fires whenever a background bash task finishes, frequently while the agent is **idle** (between turns). + +In case 2, if the injected `context/message` is the last event before a flush/dispose (no later turn appends a `turn/end`), `scanLog` treats it as crash debris and **drops it on resume** — the injected context is durably on disk but silently lost on reload. Case 1 was benign in isolation (a `user/message` is always followed by the turn it triggered) but made the "what may appear outside a turn" rule fuzzy. + +Two ways to fix it: relax the *reader* (let `scanLog` commit events that sit outside an open turn), or constrain the *producer* (make every event turn-enclosed so the reader's simple "last `turn/end`" rule is both correct and complete). We chose the producer-side invariant: a single, checkable rule beats a more permissive boundary scan that has to reason about partial turns *and* loose between-turn events. + +## Decision + +**Every session event lives inside a turn** — between a `turn/start` and its matching `turn/end`. Concretely: + +- The loop appends queued `user/message` events **after** `turn/start` (inside the turn), not before it. `turn/end` is therefore owed the moment those messages are recorded, and the existing finalizer guarantees it. +- An `agent.inject()` made while the agent is **running** appends its `context/message` into the already-open turn (unchanged). +- An `agent.inject()` made while **idle** wraps its `context/message` in a one-shot turn: `turn/start{trigger:{kind:'injection'}}` → `context/message` → `turn/end{completed}`. A new `injection` variant joins the merge-extensible `TurnTriggerMap`. +- The loop derives the next turn number from the log each iteration (`lastTurnNumber(session) + 1`) instead of keeping a private counter, so an idle injection's one-shot turn cannot collide with the next real turn's number. +- The `dsh-invariants` plugin **enforces** the invariant in dev: a `user/message` / `context/message` / `steering/message` appended while no turn is open throws an `InvariantError`. + +The serializability invariant is enforced at the same source boundary (`Session.append` throws on non-JSON-serializable data), so "what may enter the log" is now governed in one place rather than discovered downstream by whichever backend happens to be watching. + +## Consequences + +The turn is now the *single* durability/replay boundary, so a persistence backend's "last `turn/end` = commit point" rule is complete, not merely sufficient: a backend can discard everything after the last `turn/end` with zero risk of losing between-turn context, because there is no between-turn context. `scanLog` stays simple (no partial-turn boundary walk), and an idle background-task notice survives persist + resume. + +Costs: `agent.inject()` while idle now writes three log lines instead of one, and the derived history gains a turn that carries only injected context (no assistant output) — `deriveMessages()` already derives purely by event type, so this renders identically. The `injection` trigger is a new on-disk vocabulary value; like every `SessionEventMap`/`TurnTriggerMap` addition it is part of the frozen format. Event ordering within a turn changed (`turn/start` now precedes `user/message`), which is observable to anything that asserted the old order — the loop's own tests were the only such consumers. + +The rule is intentionally producer-enforced and dev-checked rather than reader-tolerated: a future backend (SQLite/WAL) inherits the same clean boundary for free, and a plugin that records an event outside a turn fails loudly in dev instead of silently losing data on the next reload. + +The invariant also constrains where the loop may record an `error` event. A failure detected while a turn is open is appended INSIDE the turn (before `turn/end`); but a failure that surfaces once the turn is already closed — a rejecting `session/flush` (which runs as the post-`turn/end` durability checkpoint) or a throwing `agent/turn-end` listener (after `closeTurn` already appended `turn/end`) — has no in-turn position left. Appending an `error` there would land it past the last `turn/end`, exactly the crash-tail position a backend discards. So those post-turn failures are reported via the `agent/error` event and the logger only, never as a `SessionEvent`; the turn stays balanced and persistence keeps its buffered events for the next checkpoint. If durable operational diagnostics are ever needed, they belong on a separate telemetry channel, not the replayable session log. diff --git a/docs/adr/README.md b/docs/adr/README.md index b634e6eb15..6b2454a110 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -28,3 +28,4 @@ Do NOT write an ADR for: a mechanical or local choice (a variable name, a one-fi | [0014](0014-doc-sync-enforcement.md) | Doc-sync enforcement (doc code blocks + event taxonomy) | accepted | | [0015](0015-structured-error-taxonomy.md) | Structured error taxonomy (HarnessError base) | accepted | | [0016](0016-pnpm-over-yarn.md) | pnpm as the package manager instead of Yarn 4 | accepted | +| [0017](0017-turn-enclosure-invariant.md) | Every session event is enclosed in a turn | accepted | diff --git a/docs/architecture.md b/docs/architecture.md index 6cab7201b5..17408cb764 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -90,7 +90,7 @@ A `Session` is an append-only log of typed `SessionEvent`s — the single source - `tool/result` → user message carrying a `tool-result` block - `context/message`, `steering/message` → user-role messages wrapped in a tagged envelope (``) at their chronological position — the "system-reminder" pattern; models distinguish them from real user prompts by the envelope. **TODO(review)**: the real adapters now exist (the original precondition); the envelope still wants a deliberate review against live model behavior (`TODO(review)` in dsh-session). -Replay/fork = `ctx.sessions.create(id, seedEvents)`. Trace/telemetry = listen to `session/event`. +Replay/fork = `ctx.sessions.create(id, { seed: seedEvents })`. Trace/telemetry = listen to `session/event`. **Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end (see `examples/echo-agent/src/session-jsonl.ts` for the pattern). **TODO**: real persistence backends (JSONL per session dir, sqlite) are a future phase. @@ -114,7 +114,7 @@ Tool schemas are deliberately **part of the assembly**: "what the model is told - `send(content)` — queued message; starts a turn when idle, else next turn - `steer(content)` — mid-turn injection, drained **between steps**; behaves like `send` when idle -- `inject(content)` — in-session context (`context/message` event) without triggering a turn; the next request sees it (Claude Code attachment / system-reminder analog) +- `inject(content)` — in-session context (`context/message` event); the next request sees it (Claude Code attachment / system-reminder analog). An inject made while the agent is *running* joins the open turn; an inject while *idle* is wrapped in a one-shot turn (`turn/start{trigger:injection}` → `context/message` → `turn/end`) so every event stays turn-enclosed (see ADR 0017). - `abort(reason)` — aborts the in-flight step via `AbortSignal` - `session`, `status`, `options` @@ -131,7 +131,7 @@ forever: wait for queued messages (idle) emit agent/status(running) 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 (late steering from previous step's listeners) session('step/start'); emit agent/step-start @@ -160,7 +160,11 @@ forever: emit agent/status(idle) unless more queued ``` -Error containment: a throwing `agent/turn-continuation` listener or a rejecting `session/flush` ends the **turn** with an `error` event — never the driver loop. An adapter that ends its stream with a `finish {kind:'error'}` or `{kind:'aborted'}` chunk (the in-band error path, for adapters that can't throw mid-stream) is likewise translated into a step error, so the turn ends `error`/`aborted` instead of logging a normal `completed` assistant message. `abort()` is honored mid-stream **and** between tool calls; disposal mid-turn ends the turn with reason `disposed` and emits `agent/status('disposed')`. +Error containment: a throwing `agent/turn-continuation` listener or a broken step ends the **turn** with an `error` event (appended INSIDE the turn, before `turn/end`) — never the driver loop. An adapter that ends its stream with a `finish {kind:'error'}` or `{kind:'aborted'}` chunk (the in-band error path, for adapters that can't throw mid-stream) is likewise translated into a step error, so the turn ends `error`/`aborted` instead of logging a normal `completed` assistant message. `abort()` is honored mid-stream **and** between tool calls; disposal mid-turn ends the turn with reason `disposed` and emits `agent/status('disposed')`. + +A failure that happens once the turn is already closed has no in-turn position for a session `error` event (appending one after `turn/end` would put it past a persistence backend's commit boundary, where it is dropped as a crash tail — ADR 0017). So a rejecting `session/flush` (the post-`turn/end` durability checkpoint) and a throwing `agent/turn-end` listener are reported via `agent/error` + the logger only, NOT as a session event; the turn stays balanced and the backend keeps its buffered events for the next flush. + +**Turn-enclosure invariant**: every session event lives inside a turn (between a `turn/start` and its `turn/end`). The loop appends queued `user/message` events *after* `turn/start`, and an idle `agent.inject()` wraps its `context/message` in a one-shot `injection` turn. This makes the turn the single durability/replay boundary: a persistence backend can treat anything after the last `turn/end` as an interrupted-crash tail without risking the loss of legitimately-recorded between-turn context. The `dsh-invariants` plugin enforces it in dev (a message event outside an open turn throws). See ADR 0017. ### Event taxonomy @@ -221,7 +225,7 @@ Every MVP feature (including the TODO-marked ones), with the mechanism that impl | Memory | section provider + tool | | Scheduled tasks (cron) | plugin registers model-callable scheduling tools; timer fires → `send(…, {source: {kind: 'cron', …}})` when idle / `inject()` notification when busy | | UI (GUI; CLI emits JSONL) | listen `agent/stream-chunk` + `session/event`; input → `send()` | -| Telemetry / replayable trace | `session/event` → JSONL; replay = `sessions.create(id, seed)` | +| Telemetry / replayable trace | `session/event` → JSONL; replay = `sessions.create(id, { seed })` | | DeepSeek V4 (and other) models | `LlmAdapter` subclass via `registerAdapter`. **Implemented twice**: `dsh-llm-deepseek` (hand-rolled) and `dsh-llm-pi-ai` (pi-ai-backed) | | Plugin hot-reload | every registration is a `ctx.effect` → vendored HMR just works | diff --git a/packages/agent-loop/README.md b/packages/agent-loop/README.md index 49176701c5..2a99fe2b6c 100644 --- a/packages/agent-loop/README.md +++ b/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() diff --git a/packages/agent-loop/src/agent.ts b/packages/agent-loop/src/agent.ts index bc8c72650f..64576186c3 100644 --- a/packages/agent-loop/src/agent.ts +++ b/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,74 @@ 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.) + try { + this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } }) + this.session.append('context/message', { content, source }) + } finally { + // 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)) { + 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. 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) => { + 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. + } + }) + } + } } abort(reason?: string): void { diff --git a/packages/agent-loop/src/loop.ts b/packages/agent-loop/src/loop.ts index 230775de94..d2b1ed270b 100644 --- a/packages/agent-loop/src/loop.ts +++ b/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 { 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,21 +158,18 @@ export async function runLoop(ctx: Context, agent: LoopAgent, handle: LoopHandle async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: number): Promise { 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 - let turnStarted = false let turnEnded = false let stepOpen = false let errorReported = false @@ -177,28 +181,59 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: const closeStep = (): void => { if (!stepOpen) return stepOpen = false - session.append('step/end', { turn, step }) + // Session.append pushes step/end BEFORE notifying session/event listeners, + // so a throwing listener leaves step/end in the log (balance holds) but + // would otherwise abort finalization. Contain it and surface it as a turn + // error below — the same outcome as a throwing agent/step-end listener. + let failure: unknown + try { + session.append('step/end', { turn, step }) + } catch (error: unknown) { + failure = error + } try { ctx.emit('agent/step-end', agent, turn, step) } catch (error: unknown) { - // step/end is already recorded so balance holds; surface the throwing - // listener as a turn error via failTurn (idempotent). This prevents a - // throwing step-end listener from producing a silent "completed" turn - // when the step itself succeeded (the normal-path closeStep call). - failTurn(toError(error)) + failure ??= error } + // A throwing step/end session-event listener OR a throwing agent/step-end + // listener surfaces as a turn error via failTurn (idempotent). This prevents + // a throwing listener from producing a silent "completed" turn when the step + // itself succeeded, AND keeps finalization going when closeStep runs from + // the outer catch. + if (failure !== undefined) failTurn(toError(failure)) } - // 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) { + // Set `reason` BEFORE the append: Session.append pushes the error event + // before notifying session/event listeners, so a throwing listener would + // otherwise leave `reason` unset (and closeTurn would record the wrong + // reason / the outer catch would skip closeTurn). The append is contained + // — the error event is already in the log either way; a throwing listener + // must not abort finalization. + reason = { kind: 'error', ...errorData(err) } + try { + session.append('error', { turn, step, ...errorData(err) }) + } catch (appendError: unknown) { + ctx.logger.warn(`agent "${agent.id}": session/event listener threw on the error event at turn ${turn}: ${toError(appendError).message}`) + } + } 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 { @@ -215,15 +250,34 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: const closeTurn = (emit: boolean): void => { if (turnEnded) return turnEnded = true - session.append('turn/end', { turn, reason }) + // Session.append pushes turn/end BEFORE notifying session/event listeners, + // so a throwing listener leaves turn/end in the log (the turn is balanced) + // but would otherwise escape — from the outer catch's closeTurn(false) it + // would propagate to the runLoop backstop, and from the normal-path + // closeTurn(true) it would skip the agent/turn-end emit. Contain it: the + // boundary is durable either way, and finalization must not abort on a bad + // listener. (On the normal path the outer catch also re-runs closeTurn, + // which is an idempotent no-op once turnEnded is set.) + try { + session.append('turn/end', { turn, reason }) + } catch (error: unknown) { + ctx.logger.warn(`agent "${agent.id}": session/event listener threw on turn/end at turn ${turn}: ${toError(error).message}`) + } if (emit) ctx.emit('agent/turn-end', agent, turn, reason) } 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. + for (const message of queued) { + session.append('user/message', { content: message.content, source: message.source }) + } ctx.emit('agent/turn-start', agent, turn) while (true) { @@ -299,12 +353,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 @@ -327,9 +389,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. + } } } @@ -448,7 +521,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' +} diff --git a/packages/agent-loop/tests/agent.spec.ts b/packages/agent-loop/tests/agent.spec.ts index 26933188fc..6ec8d7d8f5 100644 --- a/packages/agent-loop/tests/agent.spec.ts +++ b/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,124 @@ 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() 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('a1', { model: 'mock' }) + let flushes = 0 + ctx.on('session/flush', () => { flushes += 1 }) + // A session/event listener that throws on the synthetic turn/end. Append + // pushes before notifying, so turn/end is in the log (turn balanced) but the + // throw must NOT skip the durability checkpoint — the flush decision is made + // from the log, not a flag set after the (throwing) append. + 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', 'context/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('a1', { 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('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) diff --git a/packages/agent-loop/tests/coverage-edges.spec.ts b/packages/agent-loop/tests/coverage-edges.spec.ts index 260c521de1..90b87d02aa 100644 --- a/packages/agent-loop/tests/coverage-edges.spec.ts +++ b/packages/agent-loop/tests/coverage-edges.spec.ts @@ -35,9 +35,11 @@ function send(agent: LoopAgent, text: string) { agent.send([{ type: 'text', text }]) } -describe('loop backstop catch', () => { - it('a throwing turn-start listener is caught by the backstop and loop survives', async () => { - // The first turn will abort before the model call (turn-start throw). +describe('turn boundary listener throws (handled in-turn, loop survives)', () => { + it('a throwing agent/turn-start listener surfaces via agent/error and the loop survives', async () => { + // The agent/turn-start emit happens AFTER turn/start is appended to the log, + // so a throwing listener is handled inside runTurn (the turn is balanced and + // closed via failTurn → agent/error), NOT rethrown to the runLoop backstop. // The second turn should proceed normally and consume the first script entry. const adapter = new MockAdapter([textResponse('turn 2')]) const ctx = await harness(adapter) @@ -57,6 +59,9 @@ describe('loop backstop catch', () => { send(agent, 'first') await waitForIdle(ctx, agent) expect(errors.map(e => e.message)).toEqual(['broken turn-start listener']) + // The turn is balanced: its turn/start was logged, so a turn/end was owed + // and appended (decided from the log, not a flag). + expect(agent.session.events.at(-1)?.type).toBe('turn/end') // loop survives: second turn works fine and makes the model call send(agent, 'second') @@ -65,7 +70,7 @@ describe('loop backstop catch', () => { expect(adapter.requests[0]!.messages.some(m => m.content.some(b => 'text' in b && b.text === 'second'))).toBe(true) }) - it('a throwing turn-end listener is caught by the backstop and loop survives', async () => { + it('a throwing agent/turn-end listener surfaces via agent/error and the loop survives', async () => { const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create('a1', { model: 'mock' }) @@ -84,7 +89,8 @@ describe('loop backstop catch', () => { send(agent, 'first') await waitForIdle(ctx, agent) // The turn-end throw happens after the model call is complete, so turn 1's - // request is consumed. The error is surfaced by the backstop. + // request is consumed. turn/end is already in the log (append pushes before + // notifying), so the turn is balanced; the error is surfaced via agent/error. expect(errors.map(e => e.message)).toEqual(['broken turn-end listener']) // loop survives: second turn works fine @@ -92,6 +98,35 @@ describe('loop backstop catch', () => { await waitForIdle(ctx, agent) expect(adapter.requests).toHaveLength(2) }) + + it('a pre-push turn/start failure (non-serializable source) is rethrown to the runLoop backstop', async () => { + // A non-serializable message source makes the turn/start append throw BEFORE + // the event is pushed (Session.append validates before push), so turn/start + // never enters the log. runTurn sees no logged turn/start and rethrows; the + // runLoop backstop reports via agent/error (step 0) + the logger and the + // driver survives. This is the ONLY path that reaches the backstop. + const adapter = new MockAdapter([textResponse('turn 2')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a1', { 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 })) + + // A non-serializable source (BigInt) on the queued message. + agent.send([{ type: 'text', text: 'first' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never }) + await waitForIdle(ctx, agent) + + expect(errors).toHaveLength(1) + expect(errors[0]!.step).toBe(0) + expect(errors[0]!.message).toMatch(/non-JSON-serializable/) + // No turn boundary was written (the turn/start append threw before push). + expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false) + + // loop survives: a well-formed second turn runs normally. + send(agent, 'second') + await waitForIdle(ctx, agent) + expect(adapter.requests).toHaveLength(1) + }) }) describe('tool JSON parse', () => { @@ -157,7 +192,7 @@ describe('tool JSON parse', () => { }) describe('toError normalization', () => { - it('normalizes non-Error throws from turn-start listeners via toError in the backstop', async () => { + it('normalizes non-Error throws from turn-start listeners via toError', async () => { const adapter = new MockAdapter([textResponse('ok')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create('a1', { model: 'mock' }) @@ -166,7 +201,7 @@ describe('toError normalization', () => { ctx.on('agent/turn-start', () => { if (!threwOnce) { threwOnce = true - throw 'naked string error' // non-Error throw, goes through backstop's toError + throw 'naked string error' // non-Error throw, normalized via toError } }) diff --git a/packages/agent-loop/tests/loop.spec.ts b/packages/agent-loop/tests/loop.spec.ts index 1b47fd8706..94a7181771 100644 --- a/packages/agent-loop/tests/loop.spec.ts +++ b/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('') }) + 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([ diff --git a/packages/agent-loop/tests/review-fixes.spec.ts b/packages/agent-loop/tests/review-fixes.spec.ts index 3cdf68e752..d0ab2e7e3f 100644 --- a/packages/agent-loop/tests/review-fixes.spec.ts +++ b/packages/agent-loop/tests/review-fixes.spec.ts @@ -822,13 +822,15 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar expect(errorEmits).toHaveLength(0) }) - it('a throw at the turn/start append (before turnStarted) is rethrown to the runLoop backstop', async () => { - // A session/event listener that throws specifically on the turn/start - // event makes session.append('turn/start') throw while turnStarted is - // still false. runTurn must NOT try to close a turn it never opened — it - // rethrows, and the runLoop backstop records the error and survives. - // (Uses the plain harness — NOT the invariants oracle — because the - // throwing listener is itself a session/event subscriber.) + it('a throwing session/event listener on the turn/start append still balances the turn', async () => { + // Session.append pushes the event BEFORE notifying session/event listeners, + // so a listener throwing on turn/start leaves turn/start IN THE LOG. The + // loop must therefore still owe (and append) a turn/end — deciding "owed" + // from the log via isTurnOpen, not a "turn started" flag that the throw + // skipped. Otherwise the turn stays permanently open and poisons the next + // turn/replay (ADR 0017). (Uses the plain harness — NOT the invariants + // oracle — because the throwing listener is itself a session/event + // subscriber.) const adapter = new MockAdapter([textResponse('turn 2')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create('a-preturn', { model: 'mock' }) @@ -843,10 +845,18 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar send(agent, 'go') await waitForIdle(ctx, agent) - // The backstop logged exactly one error for the failed pre-turn append. + // The error was surfaced exactly once via agent/error. expect(errors.map(e => e.message)).toEqual(['boom turn/start append']) - // No turn/end was appended (none is owed — the turn never opened). - expect([...agent.session.events].some(e => e.type === 'turn/end')).toBe(false) + // The turn is BALANCED: turn/start is in the log (it was pushed before the + // listener threw), so a turn/end was owed and appended — no open turn. The + // last turn-boundary event being turn/end is exactly the loop's isTurnOpen + // check (no open turn remains). + const types = [...agent.session.events].map(e => e.type) + expect(types.filter(t => t === 'turn/start')).toHaveLength(1) + expect(types.filter(t => t === 'turn/end')).toHaveLength(1) + const lastBoundary = [...agent.session.events].reverse().find(e => e.type === 'turn/start' || e.type === 'turn/end') + expect(lastBoundary?.type).toBe('turn/end') + expect(agent.session.events.at(-1)?.type).toBe('turn/end') // loop survives: a second turn runs normally. send(agent, 'second') @@ -854,6 +864,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 throwing agent/step-end listener during a successful step ends the turn as error, not completed', async () => { // closeStep() must surface a throwing step-end listener via failTurn so the // turn ends with reason error, not a silent "completed" with the throw @@ -926,6 +969,110 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar await waitForIdle(ctx, agent) expect(boundaryCounts(agent).turnEnd).toBe(2) }) + + it('a throwing session/event listener on the error event still closes the turn (finalizer containment)', async () => { + // failTurn appends the `error` event; Session.append pushes it BEFORE + // notifying session/event listeners, so a throwing listener leaves `error` + // in the log but must NOT abort finalization — `reason` is set before the + // append and the throw is contained, so closeTurn(false) still runs and + // turn/end is appended (the turn is balanced, not left open). + // Plain harness (no invariants oracle): the throwing listener is itself a + // session/event subscriber. A finish-error drives the boundary-error path. + const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider down' } }] + const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a-errthrow', { model: 'mock' }) + + let threw = false + ctx.on('session/event', (_s, event) => { + if (!threw && event.type === 'error') { threw = true; throw new Error('boom error-event listener') } + }) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + const e = [...agent.session.events] + // The error event is in the log (pushed before the listener threw)… + expect(e.some(x => x.type === 'error')).toBe(true) + // …and the turn was still closed with the error reason (finalization did not + // abort): the last event is turn/end carrying the error reason. + const last = e.at(-1) + expect(last?.type).toBe('turn/end') + expect(last?.type === 'turn/end' && last.data.reason).toMatchObject({ kind: 'error', message: 'provider down' }) + + // loop survives: a second turn runs normally. + send(agent, 'again') + await waitForIdle(ctx, agent) + expect(adapter.requests).toHaveLength(2) + }) + + it('a throwing session/event listener on step/end during finalization still appends turn/end', async () => { + // A throwing agent/step-start listener drives the outer catch, which calls + // closeStep() during finalization. closeStep appends step/end; a + // session/event listener throwing on THAT must not abort the catch before + // closeTurn(false) — step/end is already logged (balance holds) and the + // throw is contained + surfaced via failTurn, so turn/end is still appended. + const adapter = new MockAdapter([textResponse('never reached')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a-stependthrow', { model: 'mock' }) + + // Open a step, then make the agent/step-start emit throw (boundary throw → + // outer catch → closeStep during finalization). + ctx.on('agent/step-start', () => { throw new Error('boom step-start') }) + let threw = false + ctx.on('session/event', (_s, event) => { + if (!threw && event.type === 'step/end') { threw = true; throw new Error('boom step/end listener') } + }) + const errors: Error[] = [] + ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error)) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + const e = [...agent.session.events] + // Both step/end and turn/end are present — finalization ran to completion. + expect(e.some(x => x.type === 'step/end')).toBe(true) + expect(e.some(x => x.type === 'turn/end')).toBe(true) + expect(e.at(-1)?.type).toBe('turn/end') + expect(errors.length).toBeGreaterThanOrEqual(1) // surfaced via agent/error + + // loop survives. + send(agent, 'again') + await waitForIdle(ctx, agent) + expect(e.filter(x => x.type === 'turn/start').length).toBeGreaterThanOrEqual(1) + }) + + it('a throwing session/event listener on turn/end is contained (turn still balanced, loop survives)', async () => { + // closeTurn appends turn/end; Session.append pushes it BEFORE notifying + // session/event listeners, so a throwing listener leaves turn/end in the log + // (the turn is balanced) but must not escape — from the normal-path + // closeTurn(true) it would otherwise propagate; the append is contained so + // the turn/end emit + loop continue. (A throwing agent/turn-end LISTENER is + // a separate, already-tested path; here the session/event append notify is + // what throws.) + const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a-turnendappend', { model: 'mock' }) + + let threw = false + ctx.on('session/event', (_s, event) => { + if (!threw && event.type === 'turn/end') { threw = true; throw new Error('boom turn/end listener') } + }) + + send(agent, 'go') + await waitForIdle(ctx, agent) + // turn 1 is balanced despite the throwing turn/end listener. + const e1 = [...agent.session.events] + expect(e1.filter(x => x.type === 'turn/start')).toHaveLength(1) + expect(e1.filter(x => x.type === 'turn/end')).toHaveLength(1) + expect(e1.at(-1)?.type).toBe('turn/end') + + // loop survives: a second turn runs to completion. + send(agent, 'again') + await waitForIdle(ctx, agent) + expect(adapter.requests).toHaveLength(2) + expect([...agent.session.events].filter(x => x.type === 'turn/end')).toHaveLength(2) + }) }) describe('P1-7: tool/result is logged under the originating call.id, not result.callId', () => { diff --git a/packages/agent/README.md b/packages/agent/README.md index 5f6fb85893..57b1b87fc4 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -45,7 +45,7 @@ The handle every plugin programs against: - `agent.send(content, options?)` — queue a message; starts a turn when idle - `agent.steer(content, options?)` — steer a running turn (inject between steps); behaves like `send` when idle -- `agent.inject(content, options?)` — inject in-session context without triggering a turn (context/message event); next request sees it +- `agent.inject(content, options?)` — inject in-session context (context/message event); next request sees it. While running it joins the open turn; while idle it is wrapped in a one-shot `injection` turn so every event stays turn-enclosed (ADR 0017) - `agent.abort(reason?)` — abort the in-flight step - `agent.session`, `agent.status`, `agent.options`, `agent.id` diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 0441938b6d..cee1e02666 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -60,9 +60,17 @@ export interface Agent { /** * Inject in-session context (file-change notices, skill content, cron - * notifications, …): appends a `context/message` session event without - * triggering a turn — the next model request sees it at its chronological - * position, rendered as tagged synthetic context rather than a user prompt. + * notifications, …): appends a `context/message` session event the next model + * request sees at its chronological position, rendered as tagged synthetic + * context rather than a user prompt. Does not run the model. + * + * Turn-enclosure (ADR 0017): an inject while a turn is open joins that turn; + * an inject while idle wraps its `context/message` in a one-shot `injection` + * turn (`turn/start` → `context/message` → `turn/end`) and checkpoints it for + * durability, so every event stays inside a turn and a persistence backend + * never loses a between-turn notice. The idle checkpoint is fire-and-forget + * (inject is synchronous): a failing flush is reported via `agent/error` + * (step `0`) and the logger, never thrown into the caller. * * TODO(review): exact envelope/rendering rules live in dsh-session and need * review once a real adapter exists. diff --git a/packages/invariants/src/index.ts b/packages/invariants/src/index.ts index fa72788ae4..ebb4decfd8 100644 --- a/packages/invariants/src/index.ts +++ b/packages/invariants/src/index.ts @@ -105,11 +105,10 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { } trace.lastSeq = event.seq - // Intentionally non-exhaustive: only events that carry ordering structure - // are checked; the rest are trace/replay data with no nesting contract. - // SessionEventMap is merge-extensible, so no assertNever — unknown event - // types fall through untouched. - // eslint-disable-next-line @typescript-eslint/switch-exhaustiveness-check + // Boundary/step-scoped events have explicit cases; every OTHER event type — + // including plugin-added (merge-extensible) SessionEventMap keys — is caught + // by the `default` and must be turn-enclosed (ADR 0017). No assertNever: an + // unknown variant is valid, not a compile error. switch (event.type) { case 'turn/start': { if (trace.openTurn !== null) { @@ -169,6 +168,22 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { } break } + // Turn-enclosure (ADR 0017): EVERY session event not handled by a boundary + // case above must sit inside an open turn. The durable session log uses the + // turn as its commit/replay boundary (the JSONL backend treats anything + // after the last turn/end as a crash tail), so a bare event between turns is + // silently dropped on reload. The loop records queued user messages after + // turn/start, an idle agent.inject() wraps its context/message in a one-shot + // turn, and usage/error are only appended inside an open turn. A `default` + // (not an enumerated list) is deliberate: SessionEventMap is + // merge-extensible, so a PLUGIN-added event type appended while idle must + // also fail here rather than fall through and be dropped on resume. + default: { + if (trace.openTurn === null) { + throw new InvariantError(`${event.type} appended outside any open turn (every event must be turn-enclosed)`) + } + break + } } } diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts index a2e205457f..b96d172c22 100644 --- a/packages/invariants/tests/invariants.spec.ts +++ b/packages/invariants/tests/invariants.spec.ts @@ -85,6 +85,38 @@ describe('session-log invariants', () => { .toThrow(/open is turn 1\/step null/) }) + it('rejects a message event appended outside any open turn (turn-enclosure)', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + // No turn open: every message-bearing event must be turn-enclosed (ADR 0017). + expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })) + .toThrow(/outside any open turn/) + expect(() => session.append('context/message', { content: [{ type: 'text', text: 'ctx' }], source: { kind: 'user' } })) + .toThrow(/outside any open turn/) + }) + + it('rejects usage/error and plugin-added events appended outside any open turn', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + // usage and error are turn-scoped: outside a turn they would land past the + // commit boundary and be dropped on resume (ADR 0017). + expect(() => session.append('usage', { turn: 1, step: 1, usage: { inputTokens: 1, outputTokens: 1 } })) + .toThrow(/outside any open turn/) + expect(() => session.append('error', { turn: 1, step: 1, message: 'boom' })) + .toThrow(/outside any open turn/) + // A PLUGIN-added (merge-extensible) event type is caught by the default too. + expect(() => session.append('compaction/marker' as never, { foo: 'bar' } as never)) + .toThrow(/outside any open turn/) + }) + + it('accepts message events once a turn is open', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + expect(() => session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } })) + .not.toThrow() + }) + it('rejects a tool/result with no prior tool/call', async () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() @@ -211,6 +243,7 @@ describe('dev-freeze', () => { it('freezes appended event data so mutating a logged event throws', async () => { const { ctx } = await setup() // freeze defaults true const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) expect(Object.isFrozen(event)).toBe(true) expect(Object.isFrozen(event.data)).toBe(true) @@ -221,6 +254,7 @@ describe('dev-freeze', () => { it('does not freeze when freeze:false', async () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) const event = session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }) expect(Object.isFrozen(event)).toBe(false) }) @@ -228,7 +262,8 @@ describe('dev-freeze', () => { it('freezes seeded events on session/created', async () => { const { ctx } = await setup() const seed = [ - { type: 'user/message' as const, seq: 0, time: 0, data: { content: [{ type: 'text' as const, text: 'seeded' }], source: { kind: 'user' as const } } }, + { type: 'turn/start' as const, seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message' as const, source: { kind: 'user' as const } } } }, + { type: 'user/message' as const, seq: 1, time: 0, data: { content: [{ type: 'text' as const, text: 'seeded' }], source: { kind: 'user' as const } } }, ] const session = ctx.sessions.create(undefined, { seed }) expect(Object.isFrozen(session.events[0])).toBe(true) @@ -237,6 +272,7 @@ describe('dev-freeze', () => { it('freezes mutable descendants of a shallow-frozen event datum', async () => { const { ctx } = await setup() const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) // A caller hands in a SHALLOW-frozen block whose nested array is still // mutable. deepFreeze must descend into the already-frozen object and // freeze the descendant, not short-circuit on the frozen container — @@ -260,7 +296,7 @@ describe('dev-freeze', () => { // non-serializable (incl. cyclic) data at the source, so drive the freeze // handler directly via hand-built session/events — exactly the shape the // invariants listener receives. Open a turn first (seq 0) so the cyclic - // user/message (seq 1) satisfies seq-contiguity. + // user/message (seq 1) satisfies the turn-enclosure invariant. ctx.emit('session/event', session, { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } } as never) const cyclic: Record = { type: 'text', text: 'x' } cyclic['self'] = cyclic