From 3e1ca8a4256d1b9489d17bd0a322037ce258a52d Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 00:37:18 +0800 Subject: [PATCH 1/3] fix(agent-loop): contain finalizer append-listener throws (review #32 round 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior fix handled a throwing session/event listener on the turn/start append, but the SAME push-before-notify hazard remained on the three FINALIZER appends. Session.append pushes the event before notifying, so a throwing listener on a finalizer event left the event logged but aborted the rest of finalization — stranding the turn open. - failTurn(): set `reason` BEFORE appending the `error` event, and contain a throwing session/event listener on it (the event is already logged either way). Otherwise reason stayed unset, agent/error was skipped, and the caller's closeTurn(false) never ran → open turn. - closeStep(): the try/catch wrapped only the agent/step-end EMIT, not the step/end APPEND. A throwing session/event listener on step/end escaped — fatal when closeStep runs from the outer catch during finalization (turn/start + step/end but no turn/end). Now both the append and the emit are contained and surface as a turn error via failTurn. - closeTurn(): contain a throwing session/event listener on the turn/end append (it would propagate to the runLoop backstop from closeTurn(false), or skip the turn-end emit from closeTurn(true)). turn/end is logged either way, so the turn stays balanced. Regressions: a throwing session/event listener on the error event, on step/end during finalization (driven by a throwing agent/step-start), and on turn/end — each leaves a balanced turn and the loop survives. --- packages/agent-loop/src/loop.ts | 49 +++++++-- .../agent-loop/tests/review-fixes.spec.ts | 104 ++++++++++++++++++ 2 files changed, 145 insertions(+), 8 deletions(-) diff --git a/packages/agent-loop/src/loop.ts b/packages/agent-loop/src/loop.ts index 3497892c1c..d2b1ed270b 100644 --- a/packages/agent-loop/src/loop.ts +++ b/packages/agent-loop/src/loop.ts @@ -181,16 +181,27 @@ 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 @@ -208,8 +219,18 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: // 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) }) + // 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}`) } @@ -229,7 +250,19 @@ 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) } diff --git a/packages/agent-loop/tests/review-fixes.spec.ts b/packages/agent-loop/tests/review-fixes.spec.ts index 438e246009..d0ab2e7e3f 100644 --- a/packages/agent-loop/tests/review-fixes.spec.ts +++ b/packages/agent-loop/tests/review-fixes.spec.ts @@ -969,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', () => { From 9838fdb626791f0bdcd43e980bbd675937a9dd7d Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 00:38:55 +0800 Subject: [PATCH 2/3] docs(adr-0016): soften 'typed error' to 'clear error' (review #34) The resume seam intentionally throws a plain Error (the JSDoc and #34 were aligned to "clear error"). Match the ADR 0016 prose, which still said "typed error". Docs-only; no behavior change. (Also carries the #32 finalizer-containment fixes via the forward merge.) --- docs/adr/0016-session-persistence.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/adr/0016-session-persistence.md b/docs/adr/0016-session-persistence.md index 9be7437841..d83e8829f5 100644 --- a/docs/adr/0016-session-persistence.md +++ b/docs/adr/0016-session-persistence.md @@ -21,7 +21,7 @@ Key choices recorded here because they are durable, contested, and surprising: - **Append-only with a single exception.** Committed events — those at or below a flushed `turn/end` — are never rewritten. The loop only flushes at `turn/end`, so a crash can leave a half-written final turn below the last checkpoint; `load` returns events only up to the **last complete `turn/end`**, and the first post-load `append` runs a one-time **truncation-repair** (`ftruncate` + `fsync`) that physically discards only that never-committed crash tail before writing. - **File backend canonical, DB backend a drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. A future `dsh-session-persistence-sqlite` is a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL). - **Metadata is out-of-log.** Format version, cwd, and lineage are storage concerns, not replayable conversation state, so they live in a `SessionMeta` (`SessionHeader & SessionSummary`) owned by `dsh-session` and attached to a `Session` via a new readonly `session.header` — never in `SessionEventMap`, never reaching `deriveMessages()`. The alternative (a merge-extensible `session/meta` event as log line 0) was rejected: an in-log event would ride along with a seeded/forked session for free, but metadata is not replayable state, so the explicit out-of-log header seam is the cleaner cost. -- **`load` returns a resumable event log, not just bytes.** `load(sessionId)` yields the `SessionMeta` plus the committed `SessionEvent[]` (through the last complete `turn/end`), shaped so a caller can reconstruct a live session with the loaded events as seed (so `lastTurnNumber`/`deriveMessages` continue) on the SAME session id. The agent-facing create/resume factory that consumes this is a separate seam (a follow-up on `ctx.agents`); the persistence layer deliberately stops at the `load` primitive and does NOT reach into the loop. The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever), so any resume path built on this rejects with a typed error when the backend is absent. +- **`load` returns a resumable event log, not just bytes.** `load(sessionId)` yields the `SessionMeta` plus the committed `SessionEvent[]` (through the last complete `turn/end`), shaped so a caller can reconstruct a live session with the loaded events as seed (so `lastTurnNumber`/`deriveMessages` continue) on the SAME session id. The agent-facing create/resume factory that consumes this is a separate seam (a follow-up on `ctx.agents`); the persistence layer deliberately stops at the `load` primitive and does NOT reach into the loop. The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever), so any resume path built on this rejects with a clear error when the backend is absent. Format versioning: the header carries a `version`; `load` rejects an unknown version (no v1 migration). Stated honestly: append-only + flush is robust to partial trailing writes (tolerated on load) but not to fsync-less power loss mid-line; a DB/WAL backend is the stronger option later. From 1b1385e4f7d6cd1d1faae72f5514c4775549a4ce Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 16 Jun 2026 00:43:40 +0800 Subject: [PATCH 3/3] fix(session-persistence-jsonl): never wedge a published log on temp-cleanup failure (review #33) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit materialize() removed the temp hard link in a finally that ran BEFORE syncDir() and before state.materialized/cursor advanced. If link() succeeded (log published) but the temp rm then threw, materialize() rejected after publishing — leaving state.materialized false, so the buffered events stayed unpersisted and every retry wedged on the "already exists" exists() backstop. Restructured to the robust shape: track link() success; on link failure remove the temp (the only reference) before propagating; on success fsync the directory, mark materialized, THEN best-effort remove the now-redundant temp link (a leftover *.tmp is harmless and never read). A temp-rm failure can no longer reject a session whose log published. --- .../session-persistence-jsonl/src/index.ts | 34 ++++++++++++++----- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/packages/session-persistence-jsonl/src/index.ts b/packages/session-persistence-jsonl/src/index.ts index 47d593d3c0..7f8b80a215 100644 --- a/packages/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence-jsonl/src/index.ts @@ -424,20 +424,38 @@ export class SessionPersistenceJsonl extends SessionPersistence { // final path already exists, so two processes materializing the same id // concurrently cannot clobber each other (both could pass the exists() check // above, but only one link() wins). rename() would silently overwrite the - // log the other process just committed. The temp link is always removed, - // whether link() succeeds or throws (EEXIST on a race, or any I/O error). + // log the other process just committed. + let linked = false try { await link(tmp, finalPath) + linked = true } finally { - await rm(tmp, { force: true }) + // If link FAILED (EEXIST on a race, or any I/O error), the temp is the + // only reference and must be removed before the original error propagates. + // If link SUCCEEDED, the temp cleanup is deferred to AFTER the publish is + // durable (below) so a temp-rm failure can never reject a session whose + // log already published — that would leave state.materialized false and + // wedge every retry on the exists() backstop above. + /* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */ + if (!linked) await rm(tmp, { force: true }) } - // fsync the directory so the new entry survives a power loss: on POSIX - // filesystems the new link is not crash-durable until the parent directory's - // metadata is synced. The seam contract is "append returns once durable", - // and materialize is the first append's write — so the directory entry must - // be durable before we return. + // link() succeeded — the log is published. fsync the directory so the new + // entry survives a power loss: on POSIX filesystems the new link is not + // crash-durable until the parent directory's metadata is synced. The seam + // contract is "append returns once durable", and materialize is the first + // append's write — so the directory entry must be durable before we return. await this.syncDir(dir) state.materialized = true + // Best-effort temp cleanup: the log is already published and durable, so a + // failure to remove the (now-redundant) temp hard link must NOT reject the + // append. A leftover `*.tmp` is harmless — it is never read, and the next + // materialize of this id is guarded by exists()/link(). Swallow only the + // rm failure; nothing else of consequence runs in the try. + try { + await rm(tmp, { force: true }) + } catch { + /* v8 ignore next -- redundant temp link; publish already durable, rm failure is an unreachable IO edge */ + } } /** fsync a directory so a just-created/renamed entry inside it is crash-durable. */