From 2b36620e55c94bd49a66d2956a63c97254bdafb6 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Wed, 17 Jun 2026 21:26:21 +0800 Subject: [PATCH] fix(persistence): tighten crash repair and dispose semantics --- packages/invariants/src/index.ts | 25 ++++++++++- packages/invariants/tests/invariants.spec.ts | 41 +++++++++++++++++ .../session-persistence-jsonl/src/index.ts | 37 +++++++++++----- .../tests/jsonl.spec.ts | 44 +++++++++++++------ packages/session-persistence-sqlite/README.md | 2 +- .../session-persistence-sqlite/src/index.ts | 29 +++++++++--- .../session-persistence/tests/contract.ts | 12 ++++- .../tests/persistence.spec.ts | 2 +- packages/session/src/index.ts | 2 +- packages/session/src/repair.ts | 7 +-- 10 files changed, 161 insertions(+), 40 deletions(-) diff --git a/packages/invariants/src/index.ts b/packages/invariants/src/index.ts index ebb4decfd8..fe46d9e9ee 100644 --- a/packages/invariants/src/index.ts +++ b/packages/invariants/src/index.ts @@ -57,6 +57,10 @@ interface SessionTrace { openTurn: number | null /** Open step within the current turn, or null between steps. */ openStep: number | null + /** The next turn number expected in this session log. */ + nextTurn: number + /** The next step number expected within the open turn. */ + nextStep: number /** * Tool-call ids issued in the OPEN step awaiting a result. Cleared at * `step/end` — a result must arrive in the same step as its call. @@ -114,7 +118,11 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { if (trace.openTurn !== null) { throw new InvariantError(`turn/start ${event.data.turn} while turn ${trace.openTurn} is still open`) } + if (event.data.turn !== trace.nextTurn) { + throw new InvariantError(`turn/start expected turn ${trace.nextTurn}, got ${event.data.turn}`) + } trace.openTurn = event.data.turn + trace.nextStep = 1 break } case 'turn/end': { @@ -125,6 +133,7 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { throw new InvariantError(`turn/end ${event.data.turn} while step ${trace.openStep} is still open`) } trace.openTurn = null + trace.nextTurn += 1 break } case 'step/start': { @@ -134,6 +143,9 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { if (trace.openStep !== null) { throw new InvariantError(`step/start ${event.data.step} while step ${trace.openStep} is still open`) } + if (event.data.step !== trace.nextStep) { + throw new InvariantError(`step/start expected step ${trace.nextStep} in turn ${event.data.turn}, got ${event.data.step}`) + } trace.openStep = event.data.step break } @@ -143,6 +155,7 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { // (a step that errored before its result) do not carry to the next step. trace.pendingCalls.clear() trace.openStep = null + trace.nextStep += 1 break } case 'assistant/chunk': { @@ -163,7 +176,8 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { // A result needs a prior matching call in the same step. (The converse // does NOT hold: a call may have no result — a throwing tools/execute // waterfall ends the step with no tool/result, which is legal.) - if (!trace.pendingCalls.delete(event.data.callId)) { + const syntheticInterrupted = event.data.isError && event.data.error?.code === 'interrupted' + if (!trace.pendingCalls.delete(event.data.callId) && !syntheticInterrupted) { throw new InvariantError(`tool/result for ${event.data.callId} with no prior tool/call in this step`) } break @@ -216,7 +230,14 @@ export function apply(ctx: Context, config: Config = {}): void { // (re-)apply seeds the baseline, so a reload never produces a false positive. const lastStatus = new WeakMap() - const freshTrace = (): SessionTrace => ({ lastSeq: -1, openTurn: null, openStep: null, pendingCalls: new Set() }) + const freshTrace = (): SessionTrace => ({ + lastSeq: -1, + openTurn: null, + openStep: null, + nextTurn: 1, + nextStep: 1, + pendingCalls: new Set(), + }) /** Build (or rebuild) a session's trace by replaying its whole log; freeze it. */ const seedSession = (session: Session): SessionTrace => { diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts index b96d172c22..eb935f7329 100644 --- a/packages/invariants/tests/invariants.spec.ts +++ b/packages/invariants/tests/invariants.spec.ts @@ -126,6 +126,28 @@ describe('session-log invariants', () => { .toThrow(/no prior tool\/call/) }) + it('allows a synthetic interrupted tool/result from crash repair without a prior tool/call event', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + expect(() => { + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('assistant/message', { turn: 1, step: 1, content: [ + { type: 'tool-call', id: CallId('crashed'), name: 'bash', arguments: '{}' }, + ] }) + session.append('tool/result', { + turn: 1, + step: 1, + callId: CallId('crashed'), + content: [{ type: 'text', text: 'interrupted' }], + isError: true, + error: { name: 'InterruptedError', code: 'interrupted' }, + }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('turn/end', { turn: 1, reason: { kind: 'interrupted' } }) + }).not.toThrow() + }) + it('allows a tool/call with no matching tool/result (thrown waterfall ends the step)', async () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() @@ -177,6 +199,25 @@ describe('session-log invariants', () => { }).not.toThrow() }) + it('rejects a skipped turn number', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + expect(() => session.append('turn/start', { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } })) + .toThrow(/expected turn 2, got 3/) + }) + + it('rejects a skipped step number within a turn', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('step/end', { turn: 1, step: 1 }) + expect(() => session.append('step/start', { turn: 1, step: 3 })) + .toThrow(/expected step 2 in turn 1, got 3/) + }) + it('rejects a turn/end while a step is still open', async () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() diff --git a/packages/session-persistence-jsonl/src/index.ts b/packages/session-persistence-jsonl/src/index.ts index 3e5320c3a9..0fca2bfb4d 100644 --- a/packages/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence-jsonl/src/index.ts @@ -8,8 +8,7 @@ * line, verbatim including `assistant/chunk` so `seq` stays contiguous) plus * a small atomic `.summary.json` sidecar for the mutable `SessionSummary`. * Lazy materialization (no file until the first `append`), atomic first - * write, and truncation-repair of a never-committed crash tail on the first - * `append` after a `load`. + * write, and load-time repair of a never-committed crash tail. * * 2. **The write path** — the `session/event` → buffer → `session/flush` drain * that generalizes the example `session-jsonl.ts`: snapshot each event when @@ -24,7 +23,7 @@ import { Context } from 'cordis' import z from 'schemastery' import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node:fs/promises' -import { resolve } from 'node:path' +import { dirname, resolve } from 'node:path' import { randomBytes } from 'node:crypto' import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session' @@ -111,6 +110,15 @@ function isENOENT(error: unknown): boolean { return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT' } +async function settledErrors(promises: Iterable>): Promise { + const settled = await Promise.allSettled([...promises]) + const errors: unknown[] = [] + for (const result of settled) { + if (result.status === 'rejected') errors.push(result.reason) + } + return errors +} + /** * The JSONL persistence backend. Load as a plugin; it registers as * `ctx.sessionPersistence` and installs the write-path listeners. @@ -397,8 +405,8 @@ export class SessionPersistenceJsonl extends SessionPersistence { // sidecar is best-effort) — but if we mutated state.meta first, a later // touchSummary() on a successful append would persist the rejected // title/firstPrompt, making a failed update durable after the fact. - const nextMeta: SessionMeta = { ...state.meta, ...summary } - await this.writeSidecar(nextMeta) + const nextMeta: SessionMeta = { ...state.meta, ...summary, updatedAt: summary.updatedAt ?? Date.now() } + if (state.materialized) await this.writeSidecar(nextMeta) state.meta = nextMeta } @@ -407,7 +415,10 @@ export class SessionPersistenceJsonl extends SessionPersistence { /** Atomically write the header line + first batch (temp-write, fsync, rename). */ private async materialize(state: SessionState, events: readonly SessionEvent[]): Promise { const dir = sessionDir(this.root, state.meta.cwd) + await mkdir(this.root, { recursive: true, mode: 0o700 }) + await this.syncDir(dirname(this.root)) await mkdir(dir, { recursive: true, mode: 0o700 }) + await this.syncDir(this.root) const finalPath = logPath(this.root, state.meta.cwd, state.meta.id) // Never rename over an existing committed log: materialize is the FIRST // write of a session the backend believes is new. A file here means a @@ -571,8 +582,9 @@ export class SessionPersistenceJsonl extends SessionPersistence { try { const raw = await readFile(sidecarPath(this.root, cwd, id), 'utf8') return JSON.parse(raw) as SessionSummary - } catch { - return undefined + } catch (error) { + if (isENOENT(error)) return undefined + throw error } } @@ -675,9 +687,14 @@ export class SessionPersistenceJsonl extends SessionPersistence { // Dispose must reach quiescence: await every session's init + final drain // BEFORE returning, so no write lands after teardown (orphan rename/ENOENT). ctx.effect(() => async () => { - await Promise.allSettled([...this.inits.values()]) - await Promise.allSettled([...this.buffers.keys()].map(s => this.flush(s))) - await Promise.allSettled([...this.chains.values()]) + const errors = [ + ...await settledErrors(this.inits.values()), + ...await settledErrors([...this.buffers.keys()].map(s => this.flush(s))), + ...await settledErrors(this.chains.values()), + ] + if (errors.length > 0) { + throw new AggregateError(errors, 'session-persistence-jsonl dispose failed') + } }, 'session-persistence-jsonl write path') // HMR: a hot reload does not replay session/created, so seed existing live diff --git a/packages/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence-jsonl/tests/jsonl.spec.ts index 563dd8c627..cc463f78f7 100644 --- a/packages/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence-jsonl/tests/jsonl.spec.ts @@ -644,25 +644,40 @@ describe('SessionPersistenceJsonl: edge cases', () => { expect(loaded.meta.title).toBeUndefined() }) - it('delete removes the sidecar of a lazy session that has no log', async () => { - // update() before the first append() writes a .summary.json sidecar but no - // .jsonl log (lazy create). delete() must still remove that sidecar. - const m = meta('lazy-del', '/a') + it('update before the first append keeps summary in memory and writes no orphan sidecar', async () => { + const m = meta('lazy-update', '/a') await ctx.sessionPersistence.create(m) await ctx.sessionPersistence.update(m.id, { title: 'secret', firstPrompt: 'sensitive' }) const sidecar = sidecarPath(root, '/a', m.id) - expect((await stat(sidecar)).isFile()).toBe(true) // sidecar exists, no log - await expect(stat(logPath(root, '/a', m.id))).rejects.toThrow() // no log - await ctx.sessionPersistence.delete(m.id) - await expect(stat(sidecar)).rejects.toThrow() // sidecar gone + await expect(stat(sidecar)).rejects.toThrow() + await expect(stat(logPath(root, '/a', m.id))).rejects.toThrow() + + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + const loaded = await ctx.sessionPersistence.load(m.id) + expect(loaded.meta.title).toBe('secret') + expect(loaded.meta.firstPrompt).toBe('sensitive') + expect((await stat(sidecar)).isFile()).toBe(true) }) - it('delete removes a cwd-bucket sidecar even after a restart loses the in-memory cwd', async () => { - // A lazy session writes a sidecar under cwd /a (no log). Restart the backend - // (fresh instance, empty state) and delete: the in-memory cwd is gone and - // there is no log to recover it from, so delete must scan every bucket for - // the sidecar rather than only the _no-cwd bucket. + it('a lazy update leaves no sidecar that can leak into a future same-id session after restart', async () => { + await ctx.sessionPersistence.create(meta('restart-lazy', '/a')) + await ctx.sessionPersistence.update(SessionId('restart-lazy'), { title: 'secret' }) + await expect(stat(sidecarPath(root, '/a', SessionId('restart-lazy')))).rejects.toThrow() + + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + const m2 = meta('restart-lazy', '/a') + await ctx2.sessionPersistence.create(m2) + await ctx2.sessionPersistence.append(m2.id, oneTurnLog()) + const loaded = await ctx2.sessionPersistence.load(m2.id) + expect(loaded.meta.title).toBeUndefined() + await ctx2.fiber.dispose() + }) + + it('delete removes a materialized cwd-bucket sidecar after a restart', async () => { await ctx.sessionPersistence.create(meta('restart-del', '/a')) + await ctx.sessionPersistence.append(SessionId('restart-del'), oneTurnLog()) await ctx.sessionPersistence.update(SessionId('restart-del'), { title: 'secret' }) const sidecar = sidecarPath(root, '/a', SessionId('restart-del')) expect((await stat(sidecar)).isFile()).toBe(true) @@ -671,7 +686,8 @@ describe('SessionPersistenceJsonl: edge cases', () => { await ctx2.plugin(SessionStore) await ctx2.plugin(SessionPersistenceJsonl, { root }) await ctx2.sessionPersistence.delete(SessionId('restart-del')) - await expect(stat(sidecar)).rejects.toThrow() // sidecar gone despite no in-memory cwd + await expect(stat(sidecar)).rejects.toThrow() + await expect(stat(logPath(root, '/a', SessionId('restart-del')))).rejects.toThrow() await ctx2.fiber.dispose() }) diff --git a/packages/session-persistence-sqlite/README.md b/packages/session-persistence-sqlite/README.md index 3a7a3c8163..6f8ce212d0 100644 --- a/packages/session-persistence-sqlite/README.md +++ b/packages/session-persistence-sqlite/README.md @@ -14,7 +14,7 @@ The repo targets Node ≥ 24 (the root `engines` field), which includes the stab - **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it materializes the `sessions` row (if still lazy) and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent. (`load()` already balanced the stored log, so `append` never has to repair a crash tail.) - **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `has()`/`list()` (which report exactly the sessions that have a row). -- **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `has()`/`list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`. +- **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (an error `tool/result` for every assistant tool call left unanswered, a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `has()`/`list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`. ## Configuration (schemastery) diff --git a/packages/session-persistence-sqlite/src/index.ts b/packages/session-persistence-sqlite/src/index.ts index bd4b4d9435..cfd2c5ca60 100644 --- a/packages/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence-sqlite/src/index.ts @@ -80,6 +80,15 @@ function assertSerializable(events: readonly SessionEvent[]): void { } } +async function settledErrors(promises: Iterable>): Promise { + const settled = await Promise.allSettled([...promises]) + const errors: unknown[] = [] + for (const result of settled) { + if (result.status === 'rejected') errors.push(result.reason) + } + return errors +} + /** * The SQLite persistence backend. Load as a plugin; it registers as * `ctx.sessionPersistence` and installs the write-path listeners. @@ -314,7 +323,7 @@ export class SessionPersistenceSqlite extends SessionPersistence { await this.ready let state = this.states.get(id) if (state === undefined) state = await this.adopt(id) - const nextMeta: SessionMeta = { ...state.meta, ...summary } + const nextMeta: SessionMeta = { ...state.meta, ...summary, updatedAt: summary.updatedAt ?? Date.now() } // update's only durable effect is the summary fields; the event log is // untouched. If the row is not materialized yet (a lazy session updated // before its first append) there is nothing to write — keep the pending @@ -410,11 +419,19 @@ export class SessionPersistenceSqlite extends SessionPersistence { // Dispose must reach quiescence: await every init + final drain, then close // the database, BEFORE returning, so no write lands after teardown. ctx.effect(() => async () => { - await Promise.allSettled([...this.inits.values()]) - await Promise.allSettled([...this.buffers.keys()].map(s => this.flush(s))) - await Promise.allSettled([...this.chains.values()]) - await this.ready - this.db.close() + try { + const errors = [ + ...await settledErrors(this.inits.values()), + ...await settledErrors([...this.buffers.keys()].map(s => this.flush(s))), + ...await settledErrors(this.chains.values()), + ] + if (errors.length > 0) { + throw new AggregateError(errors, 'session-persistence-sqlite dispose failed') + } + } finally { + await this.ready + this.db.close() + } }, 'session-persistence-sqlite write path') // HMR: a hot reload does not replay session/created, so seed existing live diff --git a/packages/session-persistence/tests/contract.ts b/packages/session-persistence/tests/contract.ts index f8535a70f9..73f6f653bd 100644 --- a/packages/session-persistence/tests/contract.ts +++ b/packages/session-persistence/tests/contract.ts @@ -8,7 +8,7 @@ * @module @deepseek-ai/dsh-session-persistence/tests/contract */ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionMeta } from '@deepseek-ai/dsh-session' import { CallId } from '@deepseek-ai/dsh-llm' @@ -250,11 +250,19 @@ export function runPersistenceContract(name: string, make: () => Promise): Promise { const entry = this.store.get(id) - if (entry) Object.assign(entry.meta, summary) + if (entry) Object.assign(entry.meta, summary, { updatedAt: summary.updatedAt ?? Date.now() }) } } diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 34eefb24a3..c64af8d0e4 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -30,7 +30,7 @@ declare module 'cordis' { /** * Awaited durability checkpoint. The agent loop awaits * `ctx.parallel('session/flush', session)` at every turn end; persistence - * plugins (JSONL, sqlite — TODO, future phase) drain their write-behind + * plugins (JSONL, SQLite) drain their write-behind * buffers here and on fiber dispose. */ 'session/flush'(session: Session): Promise | void diff --git a/packages/session/src/repair.ts b/packages/session/src/repair.ts index 6a3f60a681..d5e66516bc 100644 --- a/packages/session/src/repair.ts +++ b/packages/session/src/repair.ts @@ -27,9 +27,10 @@ * — which every provider rejects as an invalid transcript on the next request. * Synthesizing an error result per orphaned call keeps resume safe. * - * This module computes those synthetic closers from an event list; the backend - * returns them inline from `load` (so the reconstructed session is balanced and - * immediately usable) and persists them on the first post-load `append`. + * This module computes those synthetic closers from an event list; backends + * return them inline from `load` (so the reconstructed session is balanced and + * immediately usable) and persist them during that mutating load before any + * later append continues the log. * * @module @deepseek-ai/dsh-session/repair */