Merge commit 'refs/codex-unblock/20260722-pr338-master' into HEAD

# Conflicts:
#	docs/event-producer-consumer.md
#	examples/acp-agent/tests/acp.snapshot.ts
#	packages/core/session/src/index.ts
#	packages/examples/acp-demo/src/index.ts
#	packages/session-persistence/session-persistence-jsonl/src/format.ts
#	vitest.config.ts

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,75 @@
+/**
+ * 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: { 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,59 @@
+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: {},
+  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,249 @@
+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: {},
+      execute: async () => { order.push('tool'); return [] },
+    })
+
+    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: {},
+      execute: async () => { order.push('tool'); return [] },
+    })
+
+    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: { 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: {},
+      execute: async () => { ran = true; return [] },
+    })
+    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: {}, execute: async () => [] })
+    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

@@ -34,7 +34,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the
 ## Durability and crash semantics
 
 - **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.
 
@@ -48,7 +48,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-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.
@@ -59,7 +61,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

@@ -125,6 +125,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 {
@@ -510,6 +520,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)