Merge master into codex/jsonl-storage-identity

packages/session-persistence/README.md

@@ -5,6 +5,7 @@ The durable session-persistence seam and its storage backends. The interface pac
 | Package | Role | ctx key |
 |---|---|---|
 | `session-persistence/` | Persistence seam + shared write coordinator | `ctx.sessionPersistence` |
+| `session-checkpoint-policy/` | Semantic durability barriers for agent requests and tool execution | (wraps `ctx.llm` / `ctx.tools`, listens on agent events) |
 | `session-persistence-jsonl/` | JSONL-sidecar persistence backend | (registers `ctx.sessionPersistence`) |
 | `session-persistence-sqlite/` | SQLite persistence backend | (registers `ctx.sessionPersistence`) |
 

packages/session-persistence/session-checkpoint-policy/README.md

@@ -0,0 +1,45 @@
+# dsh-session-checkpoint-policy
+
+Semantic durability policy for persisted agents. It checkpoints the event-sourced session before a model adapter receives a request, before a top-level tool body may produce an external side effect, and after a step has recorded its complete assistant message and ordered tool results. The final `turn/end` checkpoint remains owned by `dsh-agent-loop`.
+
+## Plugin (namespace: `session-checkpoint-policy`)
+
+This zero-config function plugin consumes `ctx.sessions`, `ctx.llm`, `ctx.tools`, and the presence of `ctx.sessionPersistence`. Load it beside one persistence backend:
+
+```yaml
+- id: session-persistence
+  name: '@deepseek-ai/dsh-session-persistence-jsonl'
+
+- id: session-checkpoints
+  name: '@deepseek-ai/dsh-session-checkpoint-policy'
+```
+
+Persistence and checkpoint scheduling are intentionally separate Cordis plugins. A persistence backend makes each requested `session/flush` durable; this policy chooses the request, tool-dispatch, and completed-step checkpoints. Loading a backend without this policy is valid and retains checkpoints requested by the loop, including final `turn/end`, but crash recovery may lose the rest of an in-flight turn. First-party persisted apps and runtimes mount both plugins explicitly; a specialized deployment may deliberately omit or replace the policy.
+
+The policy wraps `llm/stream` lazily, so the downstream stream is not constructed until the live session's buffered request events are durable. It wraps `tools/execute` after pre-execute policy and guards; a top-level tool body runs only after its recorded call is durable. If cancellation lands while that flush is pending, the wrapper returns the canonical `ABORTED_BEFORE_DISPATCH` result without entering the tool body. Nested tool dispatches reuse the outer model-visible call's checkpoint. `agent/post-step` persists the complete response/result batch before continuation work.
+
+The loop records its assistant message and ordered tool results before dispatching `agent/post-step`, so the policy always captures that core batch. An event appended by another `agent/post-step` listener is captured at this checkpoint only when that listener is registered before the policy; Cordis registration order is the explicit composition rule for such extensions.
+
+Checkpoint rejection is fail-closed at the model and tool boundaries: neither the adapter nor the top-level tool body runs. A post-step rejection fails the turn before another request starts. Concurrent tool checkpoints share the session store's serialized persistence drain and cannot duplicate sequence numbers.
+
+## Model Experience
+
+### Interrupted calls
+
+#### What the model sees
+
+The plugin adds no prompt or tool schema. A hard crash after a tool checkpoint but before its result leaves a durable unmatched call; session recovery supplies the model-visible `TOOL_OUTCOME_UNKNOWN` result owned by `dsh-session`. The message permits retry for read-only or idempotent work and requires state verification or user confirmation for calls that may have side effects.
+
+#### Token effect
+
+Successful checkpoints add no tokens and do not change the request. Recovery adds one short tool-result message to balance the interrupted transcript.
+
+#### KV Cache effect
+
+The repair result is appended after the reusable prefix, so it does not invalidate earlier cache entries.
+
+## Known Limitations and Deferred Work
+
+- The policy durably records execution intent, not generic exactly-once effects. Side-effecting tools should forward `exec.callId` as an idempotency key when their provider supports one.
+- Streaming `assistant/chunk` events have no per-chunk checkpoint. They reach storage with the next semantic checkpoint, so a hard crash may lose the current partial response.
+- A persisted call without a result cannot prove whether its external effect completed. Recovery therefore records an unknown outcome instead of retrying automatically.

packages/session-persistence/session-checkpoint-policy/package.json

@@ -0,0 +1,52 @@
+{
+  "name": "@deepseek-ai/dsh-session-checkpoint-policy",
+  "description": "Semantic session durability checkpoints before model requests and tool side effects",
+  "version": "0.0.1",
+  "private": true,
+  "type": "module",
+  "main": "lib/index.js",
+  "types": "lib/types/index.d.ts",
+  "exports": {
+    ".": {
+      "types": "./lib/types/index.d.ts",
+      "default": "./lib/index.js"
+    },
+    "./invariant": {
+      "types": "./lib/types/invariant.d.ts",
+      "default": "./lib/invariant.js"
+    },
+    "./src/*": "./src/*",
+    "./package.json": "./package.json"
+  },
+  "files": [
+    "lib/index.js",
+    "lib/invariant.js",
+    "lib/types/**/*.d.ts",
+    "lib/types/**/*.d.ts.map",
+    "src"
+  ],
+  "license": "BSD-3-Clause",
+  "peerDependencies": {
+    "@deepseek-ai/dsh-agent": "^0.0.1",
+    "@deepseek-ai/dsh-invariants": "^0.0.1",
+    "@deepseek-ai/dsh-llm": "^0.0.1",
+    "@deepseek-ai/dsh-session": "^0.0.1",
+    "@deepseek-ai/dsh-session-persistence": "^0.0.1",
+    "@deepseek-ai/dsh-tools": "^0.0.1",
+    "cordis": "^4.0.0-rc.7"
+  },
+  "devDependencies": {
+    "@cordisjs/plugin-loader": "workspace:^",
+    "@deepseek-ai/dsh-agent": "workspace:^",
+    "@deepseek-ai/dsh-agent-loop": "workspace:^",
+    "@deepseek-ai/dsh-agent-loop-testkit": "workspace:^",
+    "@deepseek-ai/dsh-invariants": "workspace:^",
+    "@deepseek-ai/dsh-llm": "workspace:^",
+    "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-session-persistence": "workspace:^",
+    "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
+    "@deepseek-ai/dsh-system-prompt": "workspace:^",
+    "@deepseek-ai/dsh-tools": "workspace:^",
+    "cordis": "^4.0.0-rc.7"
+  }
+}

packages/session-persistence/session-checkpoint-policy/src/index.ts

@@ -0,0 +1,78 @@
+/**
+ * Semantic durability checkpoints for model requests, top-level tool dispatch,
+ * and completed agent steps.
+ * @module @deepseek-ai/dsh-session-checkpoint-policy
+ */
+
+import type { Context } from 'cordis'
+import type { Session } from '@deepseek-ai/dsh-session'
+import type { StreamChunk } from '@deepseek-ai/dsh-llm'
+import { TOOL_ABORTED_BEFORE_DISPATCH, type ToolExecutionResult } from '@deepseek-ai/dsh-tools'
+import type {} from '@deepseek-ai/dsh-agent'
+import type {} from '@deepseek-ai/dsh-session-persistence'
+
+/** Cordis plugin name used by Loader diagnostics. */
+export const name = 'session-checkpoint-policy'
+
+/** Services whose request, tool, session, and persistence boundaries this policy joins. */
+export const inject = ['llm', 'sessionPersistence', 'sessions', 'tools']
+
+/**
+ * Delay construction of the downstream model stream until the complete logged
+ * request prefix is durable. A checkpoint rejection prevents adapter dispatch.
+ *
+ * @param ctx - plugin context that owns the session store.
+ * @param session - live session named by the model request.
+ * @param next - downstream `llm/stream` chain.
+ * @returns a stream that checkpoints before requesting its first chunk.
+ */
+function afterCheckpoint(
+  ctx: Context,
+  session: Session,
+  next: () => AsyncIterable<StreamChunk>,
+): AsyncIterable<StreamChunk> {
+  return (async function* (): AsyncIterable<StreamChunk> {
+    await ctx.sessions.flush(session)
+    yield* next()
+  })()
+}
+
+/** Materialize the canonical result for a call cancelled before tool dispatch. */
+function abortedBeforeDispatchResult(): ToolExecutionResult {
+  return {
+    content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],
+    isError: true,
+    error: {
+      message: 'tool call aborted before dispatch',
+      info: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
+    },
+  }
+}
+
+/**
+ * Install semantic checkpoint listeners. Loop-built model calls checkpoint the
+ * logged request before adapter dispatch; top-level tool calls checkpoint their
+ * recorded call before the tool body; post-step checkpoints retain the complete
+ * response/result batch. Nested tool dispatches reuse the durable outer call.
+ *
+ * Checkpoint failures are fail-closed at the model and tool side-effect
+ * boundaries: the downstream adapter or tool body is not invoked.
+ *
+ * @param ctx - plugin context that owns the listeners.
+ */
+export function apply(ctx: Context): void {
+  ctx.on('llm/stream', (options, next): AsyncIterable<StreamChunk> => {
+    if (options.sessionId === undefined) return next()
+    const session = ctx.sessions.get(options.sessionId)
+    return session === undefined ? next() : afterCheckpoint(ctx, session, next)
+  })
+
+  ctx.on('tools/execute', async (exec, next): Promise<ToolExecutionResult> => {
+    if (exec.agent === undefined || exec.parent !== undefined) return next()
+    await ctx.sessions.flush(exec.agent.session)
+    if (exec.signal.aborted) return abortedBeforeDispatchResult()
+    return next()
+  })
+
+  ctx.on('agent/post-step', (agent): Promise<void> => ctx.sessions.flush(agent.session))
+}

packages/session-persistence/session-checkpoint-policy/src/invariant.ts

@@ -0,0 +1,30 @@
+/**
+ * Package-owned invariant companion for `@deepseek-ai/dsh-session-checkpoint-policy`.
+ * @module @deepseek-ai/dsh-session-checkpoint-policy/invariant
+ */
+
+/* jscpd:ignore-start */
+import type { Context } from 'cordis'
+import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
+
+const PACKAGE_NAME = '@deepseek-ai/dsh-session-checkpoint-policy'
+
+/** Cordis companion plugin name. */
+export const name = 'session-checkpoint-policy-invariant'
+/** Service required before the companion can reserve package ownership. */
+export const inject = ['invariants']
+
+/**
+ * No runtime invariant: checkpoint ordering is enforced at the intercepted waterfall and
+ * persistence seams; this stateless policy owns no independent mutable relation.
+ */
+const install: InvariantInstaller = () => {}
+
+/**
+ * Register this package's invariant companion.
+ * @param ctx - Cordis context carrying the invariant service.
+ * @returns the installed registration's disposer after setup succeeds.
+ */
+export const apply = (ctx: Context): Promise<() => void> =>
+  Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
+/* jscpd:ignore-end */

packages/session-persistence/session-checkpoint-policy/tests/crash-recovery.e2e.ts

@@ -0,0 +1,106 @@
+import { spawn } from 'node:child_process'
+import { access, mkdtemp, readFile, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { Context } from 'cordis'
+import { afterEach, describe, expect, it } from 'vitest'
+import SessionStore, {
+  SessionId, TOOL_OUTCOME_UNKNOWN,
+  type SessionEvent,
+} from '@deepseek-ai/dsh-session'
+import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+
+const repoRoot = fileURLToPath(new URL('../../../../', import.meta.url))
+const childScript = fileURLToPath(new URL('./fixtures/crash-child.ts', import.meta.url))
+const tsxLoader = fileURLToPath(import.meta.resolve('tsx'))
+const sessionId = SessionId('semantic-checkpoint-crash')
+const roots: string[] = []
+const CHILD_FAILPOINT_TIMEOUT_MS = 30_000
+
+async function waitForFile(path: string): Promise<void> {
+  const deadline = Date.now() + CHILD_FAILPOINT_TIMEOUT_MS
+  for (;;) {
+    try {
+      await access(path)
+      return
+    } catch (error: unknown) {
+      if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
+    }
+    if (Date.now() >= deadline) throw new Error(`crash child did not reach failpoint ${path}`)
+    await new Promise(resolve => setTimeout(resolve, 10))
+  }
+}
+
+async function crashAt(mode: 'request' | 'tool'): Promise<{ root: string; markerText: string }> {
+  const root = await mkdtemp(join(tmpdir(), `dsh-semantic-${mode}-`))
+  roots.push(root)
+  const marker = join(root, 'failpoint')
+  const child = spawn(process.execPath, ['--import', tsxLoader, childScript, mode, root, marker], {
+    cwd: repoRoot,
+    env: { ...process.env, TSX_TSCONFIG_PATH: join(repoRoot, 'tsconfig.json') },
+    stdio: ['ignore', 'ignore', 'pipe'],
+  })
+  let stderr = ''
+  child.stderr.setEncoding('utf8')
+  child.stderr.on('data', (chunk: string) => { stderr += chunk })
+  try {
+    await waitForFile(marker)
+    const markerText = await readFile(marker, 'utf8')
+    const closed = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => {
+      child.once('close', (code, signal) => { resolve({ code, signal }) })
+    })
+    child.kill('SIGKILL')
+    const exit = await closed
+    expect(exit).toEqual({ code: null, signal: 'SIGKILL' })
+    return { root, markerText }
+  } catch (error: unknown) {
+    if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL')
+    throw new Error(`crash child failed: ${stderr}`, { cause: error })
+  }
+}
+
+async function load(root: string): Promise<SessionEvent[]> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
+  try {
+    return (await ctx.sessionPersistence.load(sessionId)).events
+  } finally {
+    await ctx.fiber.dispose()
+  }
+}
+
+afterEach(async () => {
+  await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true })))
+})
+
+describe.skipIf(process.platform === 'win32')('semantic checkpoint hard-crash recovery', () => {
+  it('persists the complete request before model dispatch', async () => {
+    const crashed = await crashAt('request')
+    expect(crashed.markerText).toBe('request-dispatched')
+    const events = await load(crashed.root)
+    expect(events.map(event => event.type)).toEqual([
+      'turn/start', 'user/message', 'step/start', 'request/header', 'step/end', 'turn/end',
+    ])
+    expect(events.at(-1)).toMatchObject({
+      type: 'turn/end', data: { reason: { kind: 'interrupted' } },
+    })
+  })
+
+  it('persists tool intent before a side effect and repairs its missing result as unknown', async () => {
+    const crashed = await crashAt('tool')
+    expect(crashed.markerText).toBe('tool-side-effect')
+    const events = await load(crashed.root)
+    expect(events.some(event => event.type === 'assistant/message')).toBe(true)
+    expect(events.some(event => event.type === 'tool/call')).toBe(true)
+    const result = events.find(event => event.type === 'tool/result')
+    expect(result?.type === 'tool/result' && result.data.error).toEqual({
+      name: 'ToolOutcomeUnknownError', code: TOOL_OUTCOME_UNKNOWN,
+    })
+    if (result?.type !== 'tool/result' || result.data.content[0]?.type !== 'text') {
+      throw new Error('expected a text tool result')
+    }
+    expect(result.data.content[0].text).toContain('Do not retry blindly.')
+  })
+})

packages/session-persistence/session-checkpoint-policy/tests/fixtures/crash-child.ts

@@ -0,0 +1,60 @@
+import { writeFile } from 'node:fs/promises'
+import { Context } from 'cordis'
+import AgentLoop from '@deepseek-ai/dsh-agent-loop'
+import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
+import { CallId, type GenerateOptions, LlmAdapter, type StreamChunk } from '@deepseek-ai/dsh-llm'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import * as checkpointPolicy from '../../src/index.ts'
+
+function waitForCrash(): Promise<never> {
+  return new Promise(() => { setInterval(() => {}, 60_000) })
+}
+
+const [mode, root, marker] = process.argv.slice(2)
+if ((mode !== 'request' && mode !== 'tool') || root === undefined || marker === undefined) {
+  throw new Error('usage: crash-child.ts <request|tool> <persistence-root> <marker>')
+}
+const persistenceRoot = root
+const failpoint = marker
+
+class CrashAdapter extends LlmAdapter {
+  async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    if (mode === 'request') {
+      await writeFile(failpoint, 'request-dispatched')
+      await waitForCrash()
+      return
+    }
+    yield { type: 'block-start', index: 0, blockType: 'tool-call' }
+    yield {
+      type: 'block-end',
+      index: 0,
+      block: { type: 'tool-call', id: CallId('crash-call'), name: 'crash_tool', arguments: '{}' },
+    }
+    yield { type: 'finish', reason: { kind: 'tool-calls' } }
+  }
+}
+
+const ctx = new Context()
+await mountAgentLoopTestDependencies(ctx)
+await ctx.plugin(AgentLoop, { agents: [] })
+await ctx.plugin(SessionPersistenceJsonl, { root: persistenceRoot, compression: 'none' })
+await ctx.plugin(checkpointPolicy)
+ctx.llm.registerAdapter(['crash'], new CrashAdapter())
+ctx.tools.register({
+  name: 'crash_tool',
+  description: 'records an external effect and never returns',
+  parameters: {},
+  output: { schema: { type: 'null' }, render: () => [] },
+  async execute() {
+    await writeFile(failpoint, 'tool-side-effect')
+    return waitForCrash()
+  },
+})
+
+const handle = await ctx.agents.create({
+  sessionId: SessionId('semantic-checkpoint-crash'),
+  agentOptions: { provider: 'crash', model: 'crash' },
+})
+handle.agent.send([{ type: 'text', text: 'exercise the crash boundary' }])
+await waitForCrash()

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

@@ -0,0 +1,259 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import Loader from '@cordisjs/plugin-loader'
+import { agentEvents, type Agent } from '@deepseek-ai/dsh-agent'
+import LlmService, { CallId, type GenerateOptions, LlmAdapter, type StreamChunk } from '@deepseek-ai/dsh-llm'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
+import SessionPersistence from '@deepseek-ai/dsh-session-persistence'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry, { TOOL_ABORTED_BEFORE_DISPATCH } from '@deepseek-ai/dsh-tools'
+import * as checkpointPolicy from '../src/index.ts'
+
+const contexts: Context[] = []
+
+class TestPersistence extends SessionPersistence {
+  locate(_meta: SessionHeader): undefined { return undefined }
+  create(_meta: SessionHeader): Promise<void> { return Promise.resolve() }
+  append(_id: SessionId, _events: readonly SessionEvent[]): Promise<void> { return Promise.resolve() }
+  load(_id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    return Promise.reject(new Error('not used'))
+  }
+  list(): Promise<SessionHeader[]> { return Promise.resolve([]) }
+}
+
+class RecordingAdapter extends LlmAdapter {
+  constructor(private readonly order: string[]) { super() }
+  async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    this.order.push('adapter')
+    yield { type: 'finish', reason: { kind: 'stop' } }
+  }
+}
+
+async function setup(): Promise<Context> {
+  const ctx = new Context()
+  contexts.push(ctx)
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(LlmService)
+  await ctx.plugin(SystemPrompt)
+  await ctx.plugin(ToolRegistry)
+  await ctx.plugin(TestPersistence)
+  await ctx.plugin(checkpointPolicy)
+  return ctx
+}
+
+async function drain(stream: AsyncIterable<StreamChunk>): Promise<void> {
+  for await (const _chunk of stream) { /* drain */ }
+}
+
+afterEach(async () => {
+  await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose()))
+})
+
+describe('session-checkpoint-policy request boundary', () => {
+  it('awaits the live session checkpoint before constructing the downstream model stream', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('request-checkpoint'))
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    const gate = Promise.withResolvers<undefined>()
+    const order: string[] = []
+    ctx.on('session/flush', async () => {
+      order.push('flush:start')
+      await gate.promise
+      order.push('flush:end')
+    })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+
+    const pending = drain(ctx.llm.stream({
+      provider: 'mock', model: 'mock', messages: [], sessionId: session.id,
+    }))
+    await Promise.resolve()
+    expect(order).toEqual(['flush:start'])
+    gate.resolve(undefined)
+    await pending
+    expect(order).toEqual(['flush:start', 'flush:end', 'adapter'])
+  })
+
+  it('delegates a request without a live session without checkpointing', async () => {
+    const ctx = await setup()
+    const order: string[] = []
+    ctx.on('session/flush', () => { order.push('flush') })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+    await drain(ctx.llm.stream({ provider: 'mock', model: 'mock', messages: [] }))
+    expect(order).toEqual(['adapter'])
+  })
+
+  it('delegates an already-detached session id without checkpointing', async () => {
+    const ctx = await setup()
+    const order: string[] = []
+    ctx.on('session/flush', () => { order.push('flush') })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+    await drain(ctx.llm.stream({
+      provider: 'mock', model: 'mock', messages: [], sessionId: SessionId('detached'),
+    }))
+    expect(order).toEqual(['adapter'])
+  })
+
+  it('does not dispatch the adapter when the checkpoint rejects', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('request-failure'))
+    const order: string[] = []
+    ctx.on('session/flush', () => Promise.reject(new Error('disk unavailable')))
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+    await expect(drain(ctx.llm.stream({
+      provider: 'mock', model: 'mock', messages: [], sessionId: session.id,
+    }))).rejects.toThrow('disk unavailable')
+    expect(order).toEqual([])
+  })
+})
+
+describe('session-checkpoint-policy tool and step boundaries', () => {
+  it('awaits the checkpoint before a top-level tool body', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('tool-checkpoint'))
+    const agent = { session } as Agent
+    const gate = Promise.withResolvers<undefined>()
+    const order: string[] = []
+    ctx.on('session/flush', async () => {
+      order.push('flush:start')
+      await gate.promise
+      order.push('flush:end')
+    })
+    ctx.tools.register({
+      name: 'write', description: 'side effect', parameters: {},
+      output: { schema: { type: 'null' }, render: () => [] },
+      execute: async () => { order.push('tool'); return null },
+    })
+
+    const pending = ctx.tools.execute({
+      callId: CallId('write-1'), name: 'write', arguments: {}, agent,
+      signal: new AbortController().signal,
+    })
+    await Promise.resolve()
+    expect(order).toEqual(['flush:start'])
+    gate.resolve(undefined)
+    await expect(pending).resolves.toMatchObject({ isError: false })
+    expect(order).toEqual(['flush:start', 'flush:end', 'tool'])
+  })
+
+  it('does not dispatch when cancellation lands during the tool checkpoint', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('tool-checkpoint-cancel'))
+    const agent = { session } as Agent
+    const controller = new AbortController()
+    const gate = Promise.withResolvers<undefined>()
+    const order: string[] = []
+    ctx.on('session/flush', async () => {
+      order.push('flush:start')
+      await gate.promise
+      order.push('flush:end')
+    })
+    ctx.tools.register({
+      name: 'write', description: 'side effect', parameters: {},
+      output: { schema: { type: 'null' }, render: () => [] },
+      execute: async () => { order.push('tool'); return null },
+    })
+
+    const pending = ctx.tools.execute({
+      callId: CallId('write-cancelled'), name: 'write', arguments: {}, agent,
+      signal: controller.signal,
+    })
+    await Promise.resolve()
+    expect(order).toEqual(['flush:start'])
+    controller.abort('cancelled during checkpoint')
+    gate.resolve(undefined)
+
+    await expect(pending).resolves.toEqual({
+      content: [{ type: 'text', text: 'Error: tool call aborted before dispatch' }],
+      isError: true,
+      error: {
+        message: 'tool call aborted before dispatch',
+        info: { name: 'AbortError', code: TOOL_ABORTED_BEFORE_DISPATCH },
+      },
+    })
+    expect(order).toEqual(['flush:start', 'flush:end'])
+  })
+
+  it('turns a rejected checkpoint into an error result without running the tool body', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('tool-failure'))
+    const agent = { session } as Agent
+    let ran = false
+    ctx.on('session/flush', () => Promise.reject(new Error('disk unavailable')))
+    ctx.tools.register({
+      name: 'write', description: 'side effect', parameters: {},
+      output: { schema: { type: 'null' }, render: () => [] },
+      execute: async () => { ran = true; return null },
+    })
+    const result = await ctx.tools.execute({
+      callId: CallId('write-2'), name: 'write', arguments: {}, agent,
+      signal: new AbortController().signal,
+    })
+    expect(result.isError).toBe(true)
+    expect(result.content).toEqual([{ type: 'text', text: 'Error: disk unavailable' }])
+    expect(ran).toBe(false)
+  })
+
+  it('reuses the outer checkpoint for a nested tool dispatch', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('nested-tool'))
+    const agent = { session } as Agent
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+    ctx.tools.register({
+      name: 'nested', description: 'nested', parameters: {},
+      output: { schema: { type: 'null' }, render: () => [] },
+      execute: async () => null,
+    })
+    await ctx.tools.execute({
+      callId: CallId('nested-1'), name: 'nested', arguments: {}, agent,
+      parent: Symbol('outer') as never,
+      signal: new AbortController().signal,
+    })
+    expect(flushes).toBe(0)
+  })
+
+  it('checkpoints the complete recorded step at agent/post-step', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('post-step'))
+    const agent = { session } as Agent
+    const flushed: string[] = []
+    ctx.on('session/flush', (current) => { flushed.push(current.id) })
+    await agentEvents(ctx, agent).serial(
+      'agent/post-step', 1, 1, new AbortController().signal,
+    )
+    expect(flushed).toEqual([session.id])
+  })
+})
+
+describe('session-checkpoint-policy lifecycle', () => {
+  it('removes its wrappers when the owning fiber is disposed', async () => {
+    const ctx = new Context()
+    contexts.push(ctx)
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(LlmService)
+    await ctx.plugin(SystemPrompt)
+    await ctx.plugin(ToolRegistry)
+    await ctx.plugin(TestPersistence)
+    const session = ctx.sessions.create(SessionId('disposed-policy'))
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter([]))
+    const fiber = await ctx.plugin(checkpointPolicy)
+    await drain(ctx.llm.stream({ provider: 'mock', model: 'mock', messages: [], sessionId: session.id }))
+    expect(flushes).toBe(1)
+    await fiber.dispose()
+    await drain(ctx.llm.stream({ provider: 'mock', model: 'mock', messages: [], sessionId: session.id }))
+    expect(flushes).toBe(1)
+  })
+
+  it('keeps the Loader-safe namespace plugin shape', () => {
+    expect('default' in checkpointPolicy).toBe(false)
+    const loader = Object.create(Loader.prototype) as Loader
+    const unwrapped = loader.unwrapExports(checkpointPolicy) as Record<string, unknown>
+    expect(unwrapped).toBe(checkpointPolicy)
+    expect(unwrapped.name).toBe('session-checkpoint-policy')
+    expect(unwrapped.inject).toEqual(['llm', 'sessionPersistence', 'sessions', 'tools'])
+    expect(typeof unwrapped.apply).toBe('function')
+  })
+})

packages/session-persistence/session-checkpoint-policy/tsconfig.json

@@ -0,0 +1,36 @@
+{
+  "extends": "../../../tsconfig.base.json",
+  "compilerOptions": {
+    "rootDir": "src",
+    "outDir": "lib/types"
+  },
+  "include": [
+    "src"
+  ],
+  "references": [
+    {
+      "path": "../../../vendor/cosmokit"
+    },
+    {
+      "path": "../../../vendor/cordis"
+    },
+    {
+      "path": "../../core/agent"
+    },
+    {
+      "path": "../../llm/llm"
+    },
+    {
+      "path": "../../core/session"
+    },
+    {
+      "path": "../../session-persistence/session-persistence"
+    },
+    {
+      "path": "../../support/invariants"
+    },
+    {
+      "path": "../../core/tools"
+    }
+  ]
+}

packages/session-persistence/session-persistence-jsonl/README.md

@@ -11,7 +11,8 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
     <encoded-id>.jsonl           # only with compression: 'none'
 ```
 
-- The first logical line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession?, seedLength?, delegationDepth }`. `delegationDepth` is required on disk and is `0` for a top-level session; a missing or invalid value rejects the log. Every subsequent line is one `SessionEvent` JSON, **verbatim including `assistant/chunk`** so `seq` stays contiguous (`events[i].seq === i`).
+- The first logical line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession?, seedLength?, delegationDepth }`. `delegationDepth` is required on disk and is `0` for a top-level session; a missing or invalid value rejects the log. Every subsequent logical line is one storage record; `assistant/chunk` events are never dropped, and `seq` stays contiguous across the decoded log (`events[i].seq === i`).
+- A storage record is a `SessionEvent` JSON verbatim, or — written only under `packChunks` — a **packed chunk row** (`text-chunks` / `reasoning-chunks` / `tool-call-chunks`; bare slash-less tags like the header's `session`, so row tags cannot be confused with event types): one line holding a run of ≥3 consecutive same-block `assistant/chunk` delta events, `seq0`/`time0` plus per-member `dt` gaps reconstructing every member's `seq`/`time` exactly. The lossless codec lives in `@deepseek-ai/dsh-session` (`packChunkRuns`/`decodeStorageRecord`) and whitelists exact shapes — anything unrecognized stores verbatim. Reading is layout-blind: `load` always decodes rows, so packed, unpacked, and mixed files load identically.
 - Session ids are unvalidated branded strings, so they are injectively escaped to a single safe path segment before use (no traversal, no collision).
 
 ## Config
@@ -19,6 +20,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
 | Key | Type | Notes |
 |---|---|---|
 | `root` | `string` (required) | Root directory for all session files. **No default** — a `process.cwd()` default would scatter files as the process's cwd changes (bash calls, subprocesses). An existing root must be a readable directory; an absent root is created on first materialization. |
+| `packChunks` | `boolean` (default `false`) | Write delta-chunk runs as packed rows (~60% smaller logical logs measured on a real coding session). Off, the written logical layout is byte-identical to the pre-packing format; reading packed rows works regardless of this switch. Off by default while the snapshot goldens stay one-event-per-line — recording with packing on rewrites every fixture `session.jsonl`. |
 | `compression` | `'zstd' \| 'none'` | Defaults to `'zstd'`; `'none'` retains newline-delimited UTF-8 text. |
 
 `locate(meta)` returns `{ kind: 'jsonl', path }` using the resolved absolute root and the same cwd-bucket/id encoding as materialization. It performs no filesystem I/O: the target can be returned before the file exists, and an existing file contains only the last flushed prefix.
@@ -33,7 +35,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the
 
 - **Bound storage identity.** Lookup requires one matching encoded filename across the cwd buckets, then verifies that the header id equals the requested id and that the header's id/cwd derive the selected path. Listing applies the same path check and rejects duplicate ids. Identity failures occur before repair or append.
 - **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s the encoded header and first batch in a temporary file. POSIX publishes it without overwrite via a hard link and `fsync`s the parent directory. Windows publishes it without overwrite via `MoveFileExW(..., MOVEFILE_WRITE_THROUGH)` and creates missing directories through the same write-through pattern. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
-- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent raw batches append lines; compressed batches append one frame. Both paths `fsync`, and a caught write or sync failure rolls the file back to its prior byte length.
+- **Append-only.** Flushed events are never rewritten. Subsequent raw batches append lines; compressed batches append one frame. Both paths `fsync`, and a caught write or sync failure rolls the file back to its prior byte length.
 - **Crash recovery — preserve valid tail work.** `load` validates every complete compressed frame and scans their decompressed JSONL. If the last frame is structurally incomplete, the reader keeps its complete decoded records, truncates from that frame's start, and re-encodes those records with the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). Raw mode truncates from its first incomplete line. A checksum/decompression failure in a complete frame, or a defect at or before the last committed `turn/end`, is corruption and rejects.
 - **Contiguous-seq.** `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type.
 
@@ -47,7 +49,7 @@ The plugin buffers frozen session events and drains them on flush or disposal. A
 
 #### What the model sees
 
-JSONL storage contributes no live prompt or schema. Loading restores stored surface history and preserves prior request headers for reconstruction; the new loop composes its current envelope. Each unanswered call in an interrupted tail is balanced with the exact error text `Tool call interrupted by a crash; no result was recorded.` Raw `assistant/chunk` records do not duplicate messages.
+JSONL storage contributes no live prompt or schema. Loading restores stored surface history and preserves prior request headers for reconstruction; the new loop composes its current envelope. Recovery balances an assistant request without a durable call with `TOOL_NOT_STARTED`; a durable call without a result becomes `TOOL_OUTCOME_UNKNOWN`, which tells the model to retry only read-only or idempotent work and to verify possible side effects or ask the user. Raw `assistant/chunk` records do not duplicate messages.
 
 #### Token effect
 

packages/session-persistence/session-persistence-jsonl/src/format.ts

@@ -10,7 +10,8 @@
 
 import { createHash } from 'node:crypto'
 import { join } from 'node:path'
-import type { SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
+import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader, SessionId, StorageRecord } from '@deepseek-ai/dsh-session'
 
 /** Physical encoding selected for JSONL session artifacts. */
 export type JsonlCompression = 'zstd' | 'none'
@@ -151,17 +152,26 @@ export function logPath(
 }
 
 /**
- * Serialize one event as a JSONL line (no trailing newline).
- * @param event - the event to serialize verbatim.
- * @returns the event's single-line JSON text; the writer adds the newline.
+ * Serialize an event batch as JSONL lines (no trailing newline). With
+ * `packChunks` on, delta-chunk runs pack into `text-chunks` /
+ * `reasoning-chunks` / `tool-call-chunks` storage rows; off writes one event
+ * per line, byte-identical to the pre-packing layout. Reading is layout-blind
+ * either way ({@link scanLog} always decodes rows), so the switch only shapes
+ * NEW bytes.
+ * @param events - the batch to serialize, in log order.
+ * @param packChunks - whether to pack delta runs into storage rows.
+ * @returns the batch's JSONL text; the writer adds the final newline.
  */
-export function eventLine(event: SessionEvent): string {
-  return JSON.stringify(event)
+export function eventLines(events: readonly SessionEvent[], packChunks: boolean): string {
+  const records: readonly StorageRecord[] = packChunks ? packChunkRuns(events) : events
+  return records.map(record => JSON.stringify(record)).join('\n')
 }
 
 /**
  * Parse a JSONL log buffer into its preserved event prefix (the header is line
- * 0). Fully written events in an interrupted final turn remain part of the
+ * 0). Event lines pass through verbatim; packed chunk rows expand back into
+ * their events, so callers see one contiguous event list regardless of layout.
+ * Fully written events in an interrupted final turn remain part of the
  * prefix. The first unparsable record or seq gap after the last `turn/end`
  * marks a tolerated torn tail; the same hole in the committed region rejects.
  *
@@ -200,46 +210,60 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
   }
   const headerLine = parsedHeader
 
-  // Parse every complete record first so the last valid `turn/end` determines
-  // whether an earlier hole is committed corruption or an uncommitted tail.
-  interface Parsed { ok: boolean; event?: SessionEvent; endByte: number }
+  // Parse and decode every complete line first so the last valid `turn/end`
+  // determines whether an earlier hole is committed corruption or an
+  // uncommitted tail. One line yields one event, or a whole run for a packed
+  // chunk row; a row-tagged line that fails row validation is a hole, exactly
+  // like unparsable JSON.
+  interface Parsed { ok: boolean; events?: SessionEvent[]; endByte: number }
   const parsed: Parsed[] = eventEntries.map((entry) => {
     try {
-      return { ok: true, event: JSON.parse(entry.text) as SessionEvent, endByte: entry.endByte }
+      return { ok: true, events: decodeStorageRecord(JSON.parse(entry.text)), endByte: entry.endByte }
     } catch {
       return { ok: false, endByte: entry.endByte }
     }
   })
 
-  // The last index (into eventEntries) that is a valid `turn/end` — the last
-  // fully-committed boundary (the loop flushes only at turn/end).
+  // The last index (into eventEntries) that ends in a valid `turn/end` — the
+  // last fully-committed boundary (the loop flushes only at turn/end). A packed
+  // row never stores a turn/end, so only single-event lines can match.
   let lastTurnEnd = -1
   for (let i = parsed.length - 1; i >= 0; i--) {
     const p = parsed[i]
-    if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break }
+    if (p?.ok && p.events?.some(e => e.type === 'turn/end')) { lastTurnEnd = i; break }
   }
 
   // Preserve the contiguous prefix, including a complete interrupted turn;
   // holes through the last committed boundary throw, while later holes stop.
+  // Contiguity is a cursor over seqs (not the line index): a packed row
+  // advances the cursor by its whole run.
   const preserved: SessionEvent[] = []
-  for (let i = 0; i < parsed.length; i++) {
+  let lastPreservedLine = -1
+  scan: for (let i = 0; i < parsed.length; i++) {
     const p = parsed[i]
-    if (!p?.ok || p.event === undefined) {
+    if (!p?.ok || p.events === undefined) {
       if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`)
       break // torn tail fragment after the last turn/end — stop, tolerate
     }
-    if (p.event.seq !== i) {
-      if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`)
-      break // gap after the last turn/end — torn tail, stop
+    for (const event of p.events) {
+      if (event.seq !== preserved.length) {
+        if (i <= lastTurnEnd) {
+          throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${preserved.length}, got ${event.seq})`)
+        }
+        break scan // gap after the last turn/end — torn tail, stop
+      }
+      preserved.push(event)
     }
-    preserved.push(p.event)
+    lastPreservedLine = i
   }
 
-  // committedBytes = end of the last PRESERVED line (header if none): the next
-  // append truncates any torn bytes past this point before writing the
-  // synthetic closers + new events.
-  const lastPreserved = parsed[preserved.length - 1]
-  const committedBytes = preserved.length > 0 && lastPreserved ? lastPreserved.endByte : headerEntry.endByte
+  // committedBytes = end of the last FULLY preserved line (header if none): the
+  // next append truncates any torn bytes past this point before writing the
+  // synthetic closers + new events. A line is preserved whole or not at all —
+  // a mid-row seq gap discards the whole row, keeping the truncation offset on
+  // a line boundary.
+  const lastPreserved = parsed[lastPreservedLine]
+  const committedBytes = lastPreserved !== undefined ? lastPreserved.endByte : headerEntry.endByte
   return { meta: fromHeaderLine(headerLine), events: preserved, committedBytes }
 }
 

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

@@ -18,7 +18,7 @@ import {
 } from '@deepseek-ai/dsh-session-persistence'
 import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
-  encodeSegment, eventLine, logPath, logSuffix, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
+  encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, scanLog, sessionDir, toHeaderLine,
   type JsonlCompression,
 } from './format.ts'
 import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from './zstd.ts'
@@ -34,7 +34,7 @@ export const JsonlCompressionSchema: z<JsonlCompression> = z.union([
   z.const('none'),
 ]).default(DEFAULT_COMPRESSION)
 
-/** Plugin config: where the JSONL backend keeps its session logs (`root` is required — no default). */
+/** Plugin config: where the JSONL backend keeps its session logs, and the packed-row write switch. */
 export interface Config {
   /**
    * Root directory for all session files. Required (no default): a default of
@@ -44,6 +44,15 @@ export interface Config {
    * first materialization.
    */
   root: string
+  /**
+   * Write runs of consecutive `assistant/chunk` delta events as packed
+   * `text-chunks`/`reasoning-chunks`/`tool-call-chunks` rows (lossless,
+   * ~60% smaller logs measured on a real session). Off by default while
+   * snapshot fixtures stay in the one-event-per-line layout: recording with
+   * packing on rewrites every golden `session.jsonl`. READING packed rows is
+   * unconditional — a log's layout never depends on this switch.
+   */
+  packChunks?: boolean
   /** Physical encoding; defaults to checksummed Zstandard frames. */
   compression?: JsonlCompression
 }
@@ -70,6 +79,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
 
   static Config: z<Config> = z.object({
     root: z.string().required(),
+    packChunks: z.boolean().default(false),
     compression: JsonlCompressionSchema,
   })
 
@@ -81,6 +91,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
   override readonly name = 'session-persistence-jsonl'
 
   private root: string
+  private packChunks: boolean
   private compression: JsonlCompression
   private coordinator: PersistenceCoordinator<JsonlTornMarker>
   private rootEncodingCheck: Promise<void> | undefined
@@ -89,6 +100,9 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     super(ctx)
     // Resolve once so later process.cwd() changes cannot split one backend across roots.
     this.root = resolve(config.root)
+    // schemastery (static Config) applied the default before construction;
+    // the cast records that runtime fact for exactOptionalPropertyTypes.
+    this.packChunks = (config as Required<Config>).packChunks
     this.compression = config.compression ?? DEFAULT_COMPRESSION
     this.assertUsableRoot()
     this.coordinator = new PersistenceCoordinator<JsonlTornMarker>(this.ctx, this)
@@ -357,7 +371,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
   /** Encode the header and first batch without combining their frame boundaries. */
   private async encodeMaterialization(meta: SessionHeader, events: readonly SessionEvent[]): Promise<Buffer | string> {
     const header = JSON.stringify(toHeaderLine(meta)) + '\n'
-    const body = events.map(eventLine).join('\n') + '\n'
+    const body = eventLines(events, this.packChunks) + '\n'
     if (this.compression === 'none') return header + body
     const headerFrame = await compressZstdFrame(header)
     const eventFrame = await compressZstdFrame(body)
@@ -366,7 +380,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
 
   /** Encode one durable append batch in the configured physical representation. */
   private async encodeEventBatch(events: readonly SessionEvent[]): Promise<Buffer | string> {
-    const body = events.map(eventLine).join('\n') + '\n'
+    const body = eventLines(events, this.packChunks) + '\n'
     return this.compression === 'zstd' ? compressZstdFrame(body) : body
   }
 

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

@@ -6,7 +6,7 @@ import { isAbsolute, join, relative, resolve } from 'node:path'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
-import { encodeSegment, logPath, scanLog, sessionDir, toHeaderLine } from '../src/format.ts'
+import { encodeSegment, eventLines, logPath, scanLog, sessionDir, toHeaderLine } from '../src/format.ts'
 import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
 import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
 
@@ -588,6 +588,121 @@ describe('SessionPersistenceJsonl: scanLog unit', () => {
   })
 })
 
+describe('SessionPersistenceJsonl: packed chunk rows (packChunks: true)', () => {
+  let ctx: Context
+  beforeEach(async () => {
+    root = await freshRoot()
+    ctx = new Context()
+    await ctx.plugin(SessionStore)
+    // compression: 'none' — these tests assert the textual storage-record layout
+    // (row tags per line); packing is orthogonal to the physical encoding.
+    await ctx.plugin(SessionPersistenceJsonl, { root, packChunks: true, compression: 'none' })
+  })
+  afterEach(async () => { await ctx.fiber.dispose() })
+
+  /** A one-turn log whose step streams a five-member text-delta run. */
+  function chunkRunLog(): SessionEvent[] {
+    const deltas: SessionEvent[] = Array.from({ length: 5 }, (_, k) => ({
+      type: 'assistant/chunk',
+      seq: 2 + k,
+      time: 3 + k,
+      data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: `t${k}` } },
+    }))
+    return [
+      { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+      { type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
+      ...deltas,
+      { type: 'assistant/message', seq: 7, time: 8, data: { turn: 1, step: 1, content: [{ type: 'text', text: 't0t1t2t3t4' }], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append', sourceEventSeqs: [2, 3, 4, 5, 6] },
+      { type: 'step/end', seq: 8, time: 9, data: { turn: 1, step: 1 } },
+      { type: 'turn/end', seq: 9, time: 10, data: { turn: 1, reason: { kind: 'completed' } } },
+    ]
+  }
+
+  it('writes a delta run as one text-chunks row and loads back identical events', async () => {
+    const m = meta('packed', '/work')
+    const log = chunkRunLog()
+    await ctx.sessionPersistence.create(m)
+    await ctx.sessionPersistence.append(m.id, log)
+
+    const raw = (await readFile(rawLogPath(root, '/work', m.id), 'utf8')).split('\n').filter(Boolean)
+    const tags = raw.slice(1).map(line => (JSON.parse(line) as { type: string }).type)
+    expect(tags).toEqual(['turn/start', 'step/start', 'text-chunks', 'assistant/message', 'step/end', 'turn/end'])
+
+    const loaded = await ctx.sessionPersistence.load(m.id)
+    expect(loaded.events).toEqual(log)
+  })
+
+  it('loads a mixed file: verbatim lines from an unpacked writer, then packed appends', async () => {
+    const m = meta('mixed', '/work')
+    const log = chunkRunLog()
+    // First turn written line-per-event by an unpacked-config writer (an old
+    // file, hand-planted so this packed-config backend adopts it on load).
+    await mkdir(sessionDir(root, '/work'), { recursive: true })
+    await writeFile(rawLogPath(root, '/work', m.id), [
+      JSON.stringify({ type: 'session', version: 0, id: 'mixed', createdAt: 1000, cwd: '/work', delegationDepth: 0 }),
+      ...log.map(e => JSON.stringify(e)),
+    ].join('\n') + '\n')
+    // Adopt the stored log (cursor = stored length), then append a second turn
+    // through THIS packed-config backend.
+    expect((await ctx.sessionPersistence.load(m.id)).events).toEqual(log)
+    const secondTurn: SessionEvent[] = JSON.parse(JSON.stringify(log)) as SessionEvent[]
+    for (const [k, e] of secondTurn.entries()) {
+      ;(e as { seq: number }).seq = 10 + k
+      ;(e.data as { turn: number }).turn = 2
+    }
+    await ctx.sessionPersistence.append(m.id, secondTurn)
+
+    const loaded = await ctx.sessionPersistence.load(m.id)
+    expect(loaded.events).toEqual([...log, ...secondTurn])
+    // The packed append really packed: the file's tail carries a text-chunks row.
+    const tags = (await readFile(rawLogPath(root, '/work', m.id), 'utf8')).split('\n').filter(Boolean)
+      .map(line => (JSON.parse(line) as { type: string }).type)
+    expect(tags.filter(t => t === 'text-chunks')).toHaveLength(1)
+    expect(tags.filter(t => t === 'assistant/chunk')).toHaveLength(5)
+  })
+
+  it('scanLog: a packed row advances the seq cursor by its whole run', () => {
+    const logText = [
+      JSON.stringify({ type: 'session', version: 0, id: 'rows', createdAt: 1, delegationDepth: 0 }),
+      JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
+      JSON.stringify({ type: 'text-chunks', seq0: 1, time0: 2, data: { turn: 1, step: 1, index: 0, dt: [1, 1], texts: ['a', 'b', 'c'] } }),
+      JSON.stringify({ type: 'turn/end', seq: 4, time: 5, data: { turn: 1, reason: { kind: 'completed' } } }),
+    ].join('\n') + '\n'
+    const { events } = scanLog(Buffer.from(logText))
+    expect(events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4])
+    expect(events[2]).toEqual({ type: 'assistant/chunk', seq: 2, time: 3, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'b' } } })
+  })
+
+  it('scanLog: a malformed packed row in the committed region rejects like corrupt JSON', () => {
+    const logText = [
+      JSON.stringify({ type: 'session', version: 0, id: 'bad-row', createdAt: 1, delegationDepth: 0 }),
+      // dt arity mismatch — row validation throws, so the line is a committed hole.
+      JSON.stringify({ type: 'text-chunks', seq0: 0, time0: 1, data: { turn: 1, step: 1, index: 0, dt: [], texts: ['a', 'b'] } }),
+      JSON.stringify({ type: 'turn/end', seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }),
+    ].join('\n') + '\n'
+    expect(() => scanLog(Buffer.from(logText))).toThrow(/unparsable committed event/)
+  })
+
+  it('scanLog: a packed row with a mid-run seq gap after the last turn/end drops the whole row', () => {
+    const logText = [
+      JSON.stringify({ type: 'session', version: 0, id: 'row-gap', createdAt: 1, delegationDepth: 0 }),
+      JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }),
+      // seq0 skips 1 — the run's first member is already a gap; no turn/end follows.
+      JSON.stringify({ type: 'text-chunks', seq0: 2, time0: 2, data: { turn: 1, step: 1, index: 0, dt: [1, 1], texts: ['a', 'b', 'c'] } }),
+    ].join('\n') + '\n'
+    const scanned = scanLog(Buffer.from(logText))
+    expect(scanned.events.map(e => e.seq)).toEqual([0])
+    // committedBytes stays on the line boundary BEFORE the dropped row.
+    const headerAndTurn = logText.split('\n').slice(0, 2).join('\n') + '\n'
+    expect(scanned.committedBytes).toBe(Buffer.byteLength(headerAndTurn, 'utf8'))
+  })
+
+  it('eventLines(packChunks: false) is byte-identical to the pre-packing layout', () => {
+    const log = chunkRunLog()
+    expect(eventLines(log, false)).toBe(log.map(e => JSON.stringify(e)).join('\n'))
+  })
+})
+
 describe('SessionPersistenceJsonl: edge cases', () => {
   let ctx: Context
   beforeEach(async () => {

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

@@ -7,7 +7,7 @@ import { join } from 'node:path'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
 import type { SessionEvent } from '@deepseek-ai/dsh-session'
 import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
-import { eventLine, logPath, scanLog, sessionDir, toHeaderLine, type JsonlCompression } from '../src/format.ts'
+import { logPath, scanLog, sessionDir, toHeaderLine, type JsonlCompression } from '../src/format.ts'
 import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts'
 import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts'
 import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
@@ -217,7 +217,7 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
     const plaintext = await decodeCompleteFrames(buffer)
     expect(plaintext.toString()).toBe([
       JSON.stringify(toHeaderLine(header)),
-      ...oneTurnLog().map(eventLine),
+      ...oneTurnLog().map(e => JSON.stringify(e)),
       '',
     ].join('\n'))
     expect((await ctx.sessionPersistence.load(header.id)).events).toEqual(oneTurnLog())
@@ -288,7 +288,7 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
       { type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
       { type: 'assistant/chunk', seq: 8, time: 9, data: { turn: 2, step: 1, chunk: { type: 'text-delta', index: 0, text: deterministicNoise(300_000) } } },
     ] as SessionEvent[]
-    const plaintext = openTurn.map(eventLine).join('\n') + '\n'
+    const plaintext = openTurn.map(e => JSON.stringify(e)).join('\n') + '\n'
     const partial = await tornFrame(plaintext, (decoded) => {
       const newlines = decoded.match(/\n/g)?.length ?? 0
       return newlines >= 2 && !decoded.endsWith('\n')
@@ -334,7 +334,7 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => {
       { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
       { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
     ] as SessionEvent[]
-    const frame = await compressZstdFrame(secondTurn.map(eventLine).join('\n') + '\n')
+    const frame = await compressZstdFrame(secondTurn.map(e => JSON.stringify(e)).join('\n') + '\n')
     await appendFile(path, frame.subarray(0, -1))
 
     const loaded = await ctx.sessionPersistence.load(header.id)
@@ -456,7 +456,7 @@ describe('SessionPersistenceJsonl: encoding selection', () => {
     await mkdir(sessionDir(root, loadHeader.cwd), { recursive: true })
     await writeFile(logPath(root, loadHeader.cwd, loadHeader.id, 'none'), [
       JSON.stringify(toHeaderLine(loadHeader)),
-      ...oneTurnLog().map(eventLine),
+      ...oneTurnLog().map(e => JSON.stringify(e)),
       '',
     ].join('\n'))
     await expect(ctx.sessionPersistence.load(loadHeader.id)).rejects.toThrow(/uses \.jsonl/)
@@ -474,7 +474,7 @@ describe('SessionPersistenceJsonl: encoding selection', () => {
     await mkdir(sessionDir(root, header.cwd), { recursive: true })
     await writeFile(logPath(root, header.cwd, header.id, 'none'), [
       JSON.stringify(toHeaderLine(header)),
-      ...oneTurnLog().map(eventLine),
+      ...oneTurnLog().map(e => JSON.stringify(e)),
       '',
     ].join('\n'))
     await expect(ctx.sessionPersistence.append(header.id, oneTurnLog())).rejects.toThrow(/uses \.jsonl/)

packages/session-persistence/session-persistence-sqlite/README.md

@@ -39,7 +39,7 @@ Like the JSONL backend, the plugin also installs the `session/event` → buffer
 
 #### What the model sees
 
-SQLite storage contributes no live prompt or schema. Loading restores the same surface history as JSONL and preserves prior headers for reconstruction; the new loop composes its current envelope. Each unanswered call in interrupted rows is balanced with the exact error text `Tool call interrupted by a crash; no result was recorded.` Row metadata and raw chunks are not messages.
+SQLite storage contributes no live prompt or schema. Loading restores the same surface history as JSONL and preserves prior headers for reconstruction; the new loop composes its current envelope. Recovery balances an assistant request without a durable call with `TOOL_NOT_STARTED`; a durable call without a result becomes `TOOL_OUTCOME_UNKNOWN`, which tells the model to retry only read-only or idempotent work and to verify possible side effects or ask the user. Row metadata and raw chunks are not messages.
 
 #### Token effect
 

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

@@ -166,8 +166,8 @@ export function scanRows(rows: readonly EventRow[]): { preserved: SessionEvent[]
     }
   })
 
-  // The last index that is a valid `turn/end` — the last fully-committed
-  // boundary (the loop flushes only at turn/end).
+  // The last index that is a valid `turn/end` — holes through a closed turn
+  // are always committed corruption.
   let lastTurnEnd = -1
   for (let i = parsed.length - 1; i >= 0; i--) {
     if (parsed[i]?.ok && rows[i]?.type === 'turn/end') { lastTurnEnd = i; break }

packages/session-persistence/session-persistence/README.md

@@ -16,7 +16,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 
 ## Invariants every backend must honor
 
-- **Append-only; a crashed turn is closed, not truncated.** Committed events (at or below a flushed `turn/end`) are never rewritten. A crash can leave an unclosed final turn whose events are real and possibly large; `load` preserves them and durably appends synthetic closers (an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}`) to balance the log and keep the rehydrated history a valid provider transcript. Only a never-fully-written torn tail fragment is discarded.
+- **Append-only; a crashed turn is closed, not truncated.** Flushed events are never rewritten. A crash can leave an unclosed final turn whose events are real and possibly large; `load` preserves them and durably appends synthetic closers (a risk-classified error `tool/result` per unanswered assistant call, then `step/end?`+`turn/end {interrupted}`) to balance the log and keep the rehydrated history a valid provider transcript. Only a never-fully-written torn tail fragment is discarded.
 - **Contiguous seq.** `load` rejects a `seq` gap/parse error in the MIDDLE of the log; `append`'s first `seq` must equal the stored next-seq.
 - **JSON-serializable data.** `append` materializes each direct/replay batch through the shared one-pass lossless-JSON boundary. Live `Session` events are already deep-frozen, but the write coordinator still copies each event into a persistence-owned buffer.
 - **Durability.** `append` returns only once the batch is durable.
@@ -25,6 +25,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 
 `PersistenceCoordinator` owns per-id state, write-behind buffers and serialization, the `session/event` → `session/flush` drain, lazy materialization, crash-tail repair, session adoption, and quiescent disposal. A first-party backend composes one, implements the small `PersistenceBackend` storage hook interface, and delegates its four public service methods. JSONL and SQLite therefore share lifecycle correctness while retaining different storage primitives; see the [coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
 
+The coordinator implements durability at checkpoints but does not select their schedule. Persisted deployments explicitly compose [`dsh-session-checkpoint-policy`](../session-checkpoint-policy) when they want request-, tool-dispatch-, and completed-step recovery boundaries; omitting it leaves the loop's coarser checkpoints intact.
+
 When a live session emits `session/disposed`, the coordinator waits for its initialization, serializes a final buffer drain, then releases every map entry owned by that exact `Session` object. A failed final drain keeps the pending buffer for backend teardown to retry. Backend teardown stops event admission first, awaits all in-flight session retirements and remaining per-id operations, drains any retained buffers, and only then closes the storage handle.
 
 The side-effect-free `locate` query remains backend-owned because it describes storage topology rather than write orchestration.
@@ -58,7 +60,7 @@ Re-exported from `dsh-session`: `SessionHeader` (immutable session metadata: `ve
 
 #### What the model sees
 
-This seam adds no prompt or schema. Resume restores stored surface events as message history; stored request headers reconstruct earlier calls, while the new loop composes the current system prompt, tools, and session prefix for its next request. Crash repair inserts exactly `Tool call interrupted by a crash; no result was recorded.` as the error result for each unanswered tool call.
+This seam adds no prompt or schema. Resume restores stored surface events as message history; stored request headers reconstruct earlier calls, while the new loop composes the current system prompt, tools, and session prefix for its next request. Crash repair marks an assistant request without a durable call as `TOOL_NOT_STARTED`; a durable call without a result becomes `TOOL_OUTCOME_UNKNOWN`, whose text lets the model retry read-only or idempotent work but directs it to verify side effects or ask the user instead of retrying blindly.
 
 #### Token effect
 

packages/session-persistence/session-persistence/src/coordinator.ts

@@ -117,6 +117,11 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId):
   if (legacy !== undefined) {
     throw new Error(`session "${id}" contains unsupported legacy request/header-delta event at seq ${legacy.seq}`)
   }
+  const legacyModeType: string = 'mode/set'
+  const legacyMode = events.find(event => event.type === legacyModeType)
+  if (legacyMode !== undefined) {
+    throw new Error(`session "${id}" contains unsupported legacy mode/set event at seq ${legacyMode.seq}`)
+  }
   const fallback = events.find(event => event.type === 'request/header'
     && (event.data as { reason?: string }).reason === 'fallback')
   if (fallback !== undefined) {

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

@@ -9,8 +9,8 @@
  */
 
 import { describe, expect, it } from 'vitest'
-import { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
-import type { Session, SessionEvent, SessionHeader, SurfaceEventType, SurfaceIntent } from '@deepseek-ai/dsh-session'
+import { SESSION_FORMAT_VERSION, Session, SessionId, TOOL_NOT_STARTED, TOOL_OUTCOME_UNKNOWN } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader, SurfaceEventType, SurfaceIntent } from '@deepseek-ai/dsh-session'
 import { CallId } from '@deepseek-ai/dsh-llm'
 import type { SessionPersistence } from '../src/index.ts'
 
@@ -122,7 +122,7 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
       }
     })
 
-    it('crash recovery: an interrupted tool call gets a synthetic error result so resume is a valid transcript', async () => {
+    it('crash recovery: an unstarted assistant tool request gets a retryable synthetic result', async () => {
       const { persistence, dispose } = await make()
       try {
         const m = meta('interrupted-toolcall')
@@ -149,7 +149,7 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
         ])
         const synthetic = loaded.events.find(e => e.type === 'tool/result')
         expect(synthetic?.type === 'tool/result' && synthetic.data).toMatchObject({
-          callId: CallId('call-x'), isError: true, error: { code: 'interrupted' },
+          callId: CallId('call-x'), isError: true, error: { code: TOOL_NOT_STARTED },
         })
         // The synthetic result carries the SAME callId as the orphaned tool-call,
         // so deriveMessages() pairs them — no provider-invalid dangling call.
@@ -162,6 +162,40 @@ export function runPersistenceContract(name: string, make: () => Promise<Contrac
       }
     })
 
+    it('crash recovery: a recorded tool call with no result tells the model to assess retry risk', async () => {
+      const { persistence, dispose } = await make()
+      try {
+        const m = meta('unknown-tool-outcome')
+        await persistence.create(m)
+        await persistence.append(m.id, [
+          { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+          { type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
+          { type: 'assistant/message', seq: 2, time: 3, data: { turn: 1, step: 1, content: [
+            { type: 'tool-call', id: CallId('call-risk'), name: 'write', arguments: '{}' },
+          ], provenance: { provider: 'mock', model: 'mock' } }, surfaceOp: 'append' },
+          { type: 'tool/call', seq: 3, time: 4, data: { turn: 1, step: 1, callId: CallId('call-risk'), name: 'write', arguments: '{}' } },
+        ])
+
+        const loaded = await persistence.load(m.id)
+        const synthetic = loaded.events.find(e => e.type === 'tool/result')
+        expect(synthetic?.type === 'tool/result' && synthetic.data.error).toEqual({
+          name: 'ToolOutcomeUnknownError', code: TOOL_OUTCOME_UNKNOWN,
+        })
+        if (synthetic?.type !== 'tool/result' || synthetic.data.content[0]?.type !== 'text') {
+          throw new Error('expected a text tool result')
+        }
+        expect(synthetic.data.content[0].text).toContain('retry only if the operation is read-only or idempotent')
+        expect(synthetic.data.content[0].text).toContain('if it may have side effects, first verify external state or ask the user')
+        const resumed = new Session(m.id, loaded.events, loaded.meta)
+        const resumedResult = resumed.deriveMessages().find(message => message.content.some(block => block.type === 'tool-result'))
+        expect(resumedResult?.content[0]).toMatchObject({
+          type: 'tool-result', toolCallId: CallId('call-risk'), isError: true,
+        })
+      } finally {
+        await dispose()
+      }
+    })
+
     it('list() excludes a created-but-never-appended (zero-event) session', async () => {
       const { persistence, dispose } = await make()
       try {

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

@@ -22,6 +22,16 @@ function legacyHeaderDelta(seq = 0): SessionEvent {
   } as unknown as SessionEvent
 }
 
+/** An unsupported named-mode fixture emulating an untyped producer. */
+function legacyModeSet(seq = 0): SessionEvent {
+  return {
+    type: 'mode/set',
+    seq,
+    time: 1,
+    data: { mode: 'plan' },
+  } as unknown as SessionEvent
+}
+
 /** An obsolete full-header reason fixture from the removed delta codec. */
 function legacyFallbackHeader(seq = 0): SessionEvent {
   return {
@@ -534,6 +544,19 @@ describe('SessionPersistence service registration', () => {
     await fiber.dispose()
   })
 
+  it('rejects a stored legacy named-mode event during load', async () => {
+    const id = SessionId('legacy-mode-load')
+    const m = meta(id, '/legacy')
+    const store: MemoryStore = new Map([[id, { meta: m, events: [legacyModeSet()] }]])
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(MemoryPersistence, { store })
+
+    await expect(ctx.sessionPersistence.load(id))
+      .rejects.toThrow('unsupported legacy mode/set event at seq 0')
+    await fiber.dispose()
+  })
+
   it('retires all coordinator bookkeeping for disposed sessions', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)