fix(persistence): tighten crash repair and dispose semantics

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<Agent, AgentStatus>()
 
-  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 => {

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()

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<unknown>>): Promise<unknown[]> {
+  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<void> {
     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

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()
   })
 

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)
 

packages/session-persistence-sqlite/src/index.ts

@@ -80,6 +80,15 @@ function assertSerializable(events: readonly SessionEvent[]): void {
   }
 }
 
+async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unknown[]> {
+  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

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<Contrac
         const log = oneTurnLog()
         await persistence.create(m)
         await persistence.append(m.id, log)
-        await persistence.update(m.id, { title: 'My session', firstPrompt: 'hi' })
+        const beforeUpdate = (await persistence.load(m.id)).meta.updatedAt
+        vi.useFakeTimers()
+        vi.setSystemTime(beforeUpdate + 1_000)
+        try {
+          await persistence.update(m.id, { title: 'My session', firstPrompt: 'hi' })
+        } finally {
+          vi.useRealTimers()
+        }
 
         const loaded = await persistence.load(m.id)
         expect(loaded.meta.title).toBe('My session')
         expect(loaded.meta.firstPrompt).toBe('hi')
+        expect(loaded.meta.updatedAt).toBe(beforeUpdate + 1_000)
         expect(loaded.events).toEqual(log) // log untouched
       } finally {
         await dispose()

packages/session-persistence/tests/persistence.spec.ts

@@ -69,7 +69,7 @@ class MemoryPersistence extends SessionPersistence {
 
   async update(id: SessionId, summary: Partial<SessionSummary>): Promise<void> {
     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() })
   }
 }
 

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> | void

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
  */