Merge remote-tracking branch 'origin/master' into worktree/pr468-retarget-latest-master

# Conflicts: # docs/architecture.i18n.yaml # docs/config-catalog.md # docs/event-producer-consumer.md # docs/module-graph.md # packages/examples/acp-demo/README.md # packages/examples/acp-demo/src/index.ts # packages/examples/acp-demo/tests/built-bin.e2e.ts # packages/examples/tui-demo/package.json # packages/examples/tui-demo/src/index.ts # packages/examples/tui-demo/tests/tui-agent.spec.ts
2026-07-22 22:38:17 +08:00
parent 7a8d5da3d3 54aaab6807
commit e962642926
76 changed files with 1342 additions and 108 deletions
--- a/packages/core/session/README.md
+++ b/packages/core/session/README.md
@@ -107,11 +107,11 @@ Appended surface entries preserve reusable prefixes. A `replace` operation inval

 #### What the model sees

-If a persisted turn ended with unanswered tool calls, each synthetic error result contains exactly `Tool call interrupted by a crash; no result was recorded.`
+If recovery finds an assistant tool request with no durable `tool/call`, its synthetic `TOOL_NOT_STARTED` result says `The tool call was interrupted before the Harness recorded it as started. Retry it if it is still needed.` If a durable `tool/call` has no result, its `TOOL_OUTCOME_UNKNOWN` result says `The tool call was interrupted after it was recorded, but no result was durably recorded. Its outcome is unknown. Decide whether to retry from the tool semantics: retry only if the operation is read-only or idempotent; if it may have side effects, first verify external state or ask the user. Do not retry blindly.`

 #### Token effect

-Zero tokens in an intact session. Each repaired call adds this retained error text on resume.
+Zero tokens in an intact session. Each repaired call adds its retained risk-specific error text on resume.

 #### KV Cache effect

--- a/packages/core/session/src/index.ts
+++ b/packages/core/session/src/index.ts
@@ -22,7 +22,7 @@ import { foldRequestHeader } from './request-header.ts'
 export * from './types.ts'
 export { isJsonValue, snapshotJsonValue } from './json.ts'
 export type { JsonValue } from './json.ts'
-export { interruptedTurnClosers } from './repair.ts'
+export { interruptedTurnClosers, TOOL_NOT_STARTED, TOOL_OUTCOME_UNKNOWN } from './repair.ts'
 export type { SessionSurface, SurfaceFoldReplacement, SurfaceFoldResult } from './surface.ts'
 export { foldSurface, isSurfaceEvent, isSurfaceEligibleType } from './surface.ts'
 export { canonicalHeader, foldRequestHeader, headerEquals } from './request-header.ts'
--- a/packages/core/session/src/invariant.ts
+++ b/packages/core/session/src/invariant.ts
@@ -10,6 +10,7 @@ import { assertNever } from '@deepseek-ai/dsh-llm'
 import type { CallId } from '@deepseek-ai/dsh-llm'
 import type { InvariantFailure, InvariantInstaller } from '@deepseek-ai/dsh-invariants'
 import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import { TOOL_NOT_STARTED } from './repair.ts'

 const PACKAGE_NAME = '@deepseek-ai/dsh-session'

@@ -133,8 +134,8 @@ function validateEvent(
        break
      }
      requireOpenStep(trace, 'tool/result', event.data.turn, event.data.step, fail)
-      const syntheticInterrupted = event.data.isError && event.data.error?.code === 'interrupted'
-      if (!trace.pendingCalls.has(event.data.callId) && !syntheticInterrupted) {
+      const syntheticNotStarted = event.data.isError && event.data.error?.code === TOOL_NOT_STARTED
+      if (!trace.pendingCalls.has(event.data.callId) && !syntheticNotStarted) {
        fail(`tool/result for ${event.data.callId} with no prior tool/call in this step`)
      }
      pendingCalls = { kind: 'delete', callId: event.data.callId }
--- a/packages/core/session/src/repair.ts
+++ b/packages/core/session/src/repair.ts
@@ -8,6 +8,12 @@
 import type { CallId } from '@deepseek-ai/dsh-llm'
 import type { SessionEvent } from './types.ts'

+/** Recovery code for an assistant tool request that never reached a recorded call start. */
+export const TOOL_NOT_STARTED = 'TOOL_NOT_STARTED'
+
+/** Recovery code for a recorded tool call whose completed outcome was not durably recorded. */
+export const TOOL_OUTCOME_UNKNOWN = 'TOOL_OUTCOME_UNKNOWN'
+
 /**
 * Return deterministic synthetic events that close an open tail turn. Unmatched
 * calls receive error results first, followed by an open `step/end` and an
@@ -82,6 +88,7 @@ export function interruptedTurnClosers(events: readonly SessionEvent[]): Session
  // Close calls before their step: providers reject dangling assistant calls,
  // and Map insertion order preserves their transcript order.
  for (const [callId, { step, callSeq }] of pendingCalls) {
+    const started = callSeq !== undefined
    closers.push({
      type: 'tool/result',
      seq: seq++,
@@ -90,12 +97,19 @@ export function interruptedTurnClosers(events: readonly SessionEvent[]): Session
        turn: openTurn,
        step,
        callId,
-        content: [{ type: 'text', text: 'Tool call interrupted by a crash; no result was recorded.' }],
+        content: [{
+          type: 'text',
+          text: started
+            ? 'The tool call was interrupted after it was recorded, but no result was durably recorded. Its outcome is unknown. Decide whether to retry from the tool semantics: retry only if the operation is read-only or idempotent; if it may have side effects, first verify external state or ask the user. Do not retry blindly.'
+            : 'The tool call was interrupted before the Harness recorded it as started. Retry it if it is still needed.',
+        }],
        isError: true,
-        error: { name: 'InterruptedError', code: 'interrupted' },
+        error: started
+          ? { name: 'ToolOutcomeUnknownError', code: TOOL_OUTCOME_UNKNOWN }
+          : { name: 'ToolNotStartedError', code: TOOL_NOT_STARTED },
      },
      surfaceOp: 'append',
-      ...callSeq !== undefined ? { sourceEventSeqs: [callSeq] } : {},
+      ...started ? { sourceEventSeqs: [callSeq] } : {},
    })
  }

--- a/packages/core/session/tests/invariant.spec.ts
+++ b/packages/core/session/tests/invariant.spec.ts
@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import { createScope, scopeTarget } from '@deepseek-ai/dsh-scope'
 import { CallId } from '@deepseek-ai/dsh-llm'
-import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
+import SessionStore, { SessionId, TOOL_NOT_STARTED } from '@deepseek-ai/dsh-session'
 import * as SessionInvariant from '@deepseek-ai/dsh-session/invariant'
 import InvariantService, { InvariantError } from '@deepseek-ai/dsh-invariants'

@@ -256,7 +256,7 @@ describe('session-log invariants', () => {
    })).toThrow(/outside any open turn/)
  })

-  it('allows interrupted repair results and unresolved calls at step end', async () => {
+  it('allows not-started repair results and unresolved calls at step end', async () => {
    const repaired = (await setup()).ctx.sessions.create()
    expect(() => {
      repaired.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
@@ -267,7 +267,7 @@ describe('session-log invariants', () => {
        callId: CallId('crashed'),
        content: [],
        isError: true,
-        error: { name: 'InterruptedError', code: 'interrupted' },
+        error: { name: 'ToolNotStartedError', code: TOOL_NOT_STARTED },
      }, { surfaceOp: 'append' })
      repaired.append('step/end', { turn: 1, step: 1 })
      repaired.append('turn/end', { turn: 1, reason: { kind: 'interrupted' } })
--- a/packages/core/session/tests/repair.spec.ts
+++ b/packages/core/session/tests/repair.spec.ts
@@ -1,6 +1,6 @@
 import { describe, expect, it } from 'vitest'
 import { CallId } from '@deepseek-ai/dsh-llm'
-import { interruptedTurnClosers } from '../src/index.ts'
+import { interruptedTurnClosers, TOOL_NOT_STARTED, TOOL_OUTCOME_UNKNOWN } from '../src/index.ts'
 import type { SessionEvent, SurfaceEvent } from '../src/index.ts'

 /**
@@ -47,9 +47,7 @@ describe('interruptedTurnClosers', () => {
    expect(closers.map(e => e.seq)).toEqual([2, 3])
  })

-  it('synthesizes an error tool/result for a tool-call the crash left unanswered', () => {
-    // A step issued one tool call (in the assistant message) but crashed before
-    // the tool/result was logged — the classic mid-tool crash.
+  it('marks an assistant tool request with no recorded call as not started', () => {
    const events: SessionEvent[] = [
      userTurnStart(2, 0),
      { type: 'step/start', seq: 1, time: 1, data: { turn: 2, step: 1 } },
@@ -64,8 +62,11 @@ describe('interruptedTurnClosers', () => {
    expect(closers.map(e => e.seq)).toEqual([3, 4, 5])
    const result = closers[0]!
    expect(result.type === 'tool/result' && result.data).toMatchObject({
-      turn: 2, step: 1, callId: CallId('call-1'), isError: true, error: { code: 'interrupted' },
+      turn: 2, step: 1, callId: CallId('call-1'), isError: true, error: { code: TOOL_NOT_STARTED },
    })
+    expect(result.type === 'tool/result' && result.data.content).toEqual([{
+      type: 'text', text: 'The tool call was interrupted before the Harness recorded it as started. Retry it if it is still needed.',
+    }])
  })

  it('does NOT synthesize a result for a tool-call that already has one', () => {
@@ -152,6 +153,14 @@ describe('interruptedTurnClosers', () => {
    const result = closers[0]!
    expect((result as SurfaceEvent).surfaceOp).toBe('append')
    expect((result as SurfaceEvent).sourceEventSeqs).toEqual([3])
+    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('retry only if the operation is read-only or idempotent')
+    expect(result.data.content[0].text).toContain('first verify external state or ask the user')
  })

  it('handles tool/call without a matching assistant/message entry gracefully', () => {
--- a/packages/examples/acp-demo/README.md
+++ b/packages/examples/acp-demo/README.md
@@ -16,13 +16,14 @@ stdout is the ACP JSON-RPC channel, so the cluster is defined as much by what it
 | `@deepseek-ai/dsh-user-interaction` | the human question/answer seam used by clients that can complete ACP elicitation requests |
 | `@deepseek-ai/dsh-session-persistence-jsonl` | durable JSONL session log (the bridge advertises `loadSession`) |
 | `@deepseek-ai/dsh-session-query` + `@deepseek-ai/dsh-session-reference` | exact current-surface reads and bounded `dsh-session:` snapshots |
+| `@deepseek-ai/dsh-session-checkpoint-policy` | semantic durability barriers before model requests and top-level tool effects, plus completed-step checkpoints |
 | `@deepseek-ai/dsh-acp` | the bridge that owns stdout for JSON-RPC and provides ACP-backed user answers when a leaf explicitly exposes a user-question tool |
 | ~~`@deepseek-ai/dsh-tool-ask-user`~~ | **omitted by default** — ACP elicitation support is still client-dependent, so leaves must opt in deliberately |
 | ~~`@deepseek-ai/dsh-user-approval`~~ | **omitted by default** — permission policy is deployment-specific; sandbox/approval leaves opt in and the ACP bridge then supplies the answerer |
 | ~~console logger~~ | **omitted** — it writes to stdout and would corrupt the protocol frames ([the stdout-purity footgun](../../ui/acp/README.md)) |
 | ~~`hmr`~~ | **omitted** — the editor owns the subprocess |

-Because the package wires no logger entry, an ACP leaf has **nothing to get wrong by default**: it only picks backends. A leaf author can still add `@cordisjs/plugin-logger-console` as a sibling entry, so the rule remains: never add a stdout logger to an ACP leaf; use a stderr exporter instead.
+The app owns this cluster through one ordered Cordis effect. Teardown drains the ACP bridge before removing the checkpoint policy or persistence backend, so a graceful disconnect persists the real closing `step/end` and `turn/end` events rather than leaving crash recovery to synthesize them. Because the package wires no logger entry, an ACP leaf has **nothing to get wrong by default**: it only picks backends. A leaf author can still add `@cordisjs/plugin-logger-console` as a sibling entry, so the rule remains: never add a stdout logger to an ACP leaf; use a stderr exporter instead.

 ## Config

@@ -44,6 +45,7 @@ Because the package wires no logger entry, an ACP leaf has **nothing to get wron
 | `llmRetry` | owner defaults | bounded transient model-request retry policy routed through `dsh-agent-spine-demo` |
 | `persistenceRoot` | `./.sessions` | the JSONL backend's root directory |
 | `persistenceCompression` | `'zstd'` | JSONL artifact encoding (`'zstd'` or raw `'none'`) |
+| `sessionReferences` | service defaults | cross-session candidate and snapshot limits routed to `dsh-session-reference` |

 The leaf supplies the swappable backends: an LLM adapter (`llm-deepseek` for the real model, `llm-replay` for keyless snapshot replay), a bash executor, and optionally a `ctx.fs` provider. Workspace context becomes a no-op without `ctx.fs`; the shipped [`examples/acp-agent/cordis.yml`](../../../examples/acp-agent/cordis.yml) selects `dsh-sandbox-policy`, `dsh-fs-sandbox`, `dsh-fs-policy`, and `dsh-tool-fs` so baseline instructions and model-facing `read`/`write`/`edit` share one provider, sandbox mode, workspace root, and observed-version policy.

--- a/packages/examples/acp-demo/package.json
+++ b/packages/examples/acp-demo/package.json
@@ -43,6 +43,7 @@
    "@deepseek-ai/dsh-agent-spine-demo": "^0.0.1",
    "@deepseek-ai/dsh-app-boot": "^0.0.1",
    "@deepseek-ai/dsh-invariants": "^0.0.1",
+    "@deepseek-ai/dsh-session-checkpoint-policy": "^0.0.1",
    "@deepseek-ai/dsh-session-persistence-jsonl": "^0.0.1",
    "@deepseek-ai/dsh-session-query": "^0.0.1",
    "@deepseek-ai/dsh-session-reference": "^0.0.1",
@@ -62,6 +63,7 @@
    "@deepseek-ai/dsh-agent-spine-demo": "workspace:^",
    "@deepseek-ai/dsh-app-boot": "workspace:^",
    "@deepseek-ai/dsh-invariants": "workspace:^",
+    "@deepseek-ai/dsh-session-checkpoint-policy": "workspace:^",
    "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
    "@deepseek-ai/dsh-session-query": "workspace:^",
    "@deepseek-ai/dsh-session-reference": "workspace:^",
--- a/packages/examples/acp-demo/src/index.ts
+++ b/packages/examples/acp-demo/src/index.ts
@@ -1,7 +1,9 @@
 /**
 * The ACP server app: the default agent spine ({@link @deepseek-ai/dsh-agent-spine-demo}),
 * human-command registry, JSONL session persistence, and the
- * {@link @deepseek-ai/dsh-acp} bridge. It writes nothing to stdout.
+ * {@link @deepseek-ai/dsh-acp} bridge. The app owns those plugins through one
+ * ordered lifecycle so ACP sessions quiesce before persistence detaches. It
+ * writes nothing to stdout.
 * It pre-creates no agents and leaves adapters, executors, and optional tools to
 * the leaf, which must likewise avoid stdout loggers. Named exports are
 * required so Loader retains this plugin's `Config` schema (see
@@ -21,6 +23,7 @@ import SessionPersistenceJsonl, {
  JsonlCompressionSchema,
  type JsonlCompression,
 } from '@deepseek-ai/dsh-session-persistence-jsonl'
+import * as sessionCheckpointPolicy from '@deepseek-ai/dsh-session-checkpoint-policy'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 import SessionQueryService from '@deepseek-ai/dsh-session-query'
 import SessionReferenceService, { type Config as SessionReferenceConfig } from '@deepseek-ai/dsh-session-reference'
@@ -106,23 +109,24 @@ export const Config: z<Config> = z.object({
 * NO agents (its `agents` list defaults to `[]`) and carries the deployment
 * `persona`; the JSONL backend persists under `persistenceRoot`; the ACP
 * bridge owns stdout for JSON-RPC and creates one agent per `session/new`
- * from the provider/model pair. No logger, no `hmr` — stdout stays pure.
+ * from the provider/model pair. The composite effect unloads in reverse order,
+ * keeping checkpoint and persistence listeners attached until ACP agents have
+ * flushed their closing events. No logger, no `hmr` — stdout stays pure.
 */
 export function apply(ctx: Context, config: Config): void {
  const goals = config.goals ?? {}
-  ctx.plugin(CommandService)
-  if (goals !== false) ctx.plugin(commandGoal)
-  ctx.plugin(agentCore, { ...agentCore.pickSpineConfig(config), goals })
-  // This front door owns the same persistence/reference cluster as the TUI;
-  // extracting these few calls would introduce a shared app-composition facade.
-  /* jscpd:ignore-start */
-  ctx.plugin(UserInteractionService)
-  ctx.plugin(SessionPersistenceJsonl, {
-    root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
-    ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
-  })
-  ctx.plugin(SessionQueryService)
-  ctx.plugin(SessionReferenceService, config.sessionReferences ?? {})
-  /* jscpd:ignore-end */
-  ctx.plugin(acp, { provider: config.provider, model: config.model })
+  ctx.effect(function* () {
+    yield ctx.plugin(CommandService).dispose
+    if (goals !== false) yield ctx.plugin(commandGoal).dispose
+    yield ctx.plugin(agentCore, { ...agentCore.pickSpineConfig(config), goals }).dispose
+    yield ctx.plugin(UserInteractionService).dispose
+    yield ctx.plugin(SessionPersistenceJsonl, {
+      root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
+      ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
+    }).dispose
+    yield ctx.plugin(sessionCheckpointPolicy).dispose
+    yield ctx.plugin(SessionQueryService).dispose
+    yield ctx.plugin(SessionReferenceService, config.sessionReferences ?? {}).dispose
+    yield ctx.plugin(acp, { provider: config.provider, model: config.model }).dispose
+  }, 'acp-demo.composition')
 }
--- a/packages/examples/acp-demo/tests/built-bin.e2e.ts
+++ b/packages/examples/acp-demo/tests/built-bin.e2e.ts
@@ -36,8 +36,8 @@ const dshPackages = [
  'core/tools', 'core/agent-loop', 'llm/llm', 'bash/bash',
  'bash/bash-local', 'bash/tool-bash', 'context/workspace-context', 'support/invariants', 'ui/app-boot',
  'session-persistence/session-persistence',
-  'session-persistence/session-persistence-jsonl', 'session-query/session-query',
-  'context/session-reference', 'ui/acp', 'examples/acp-demo', 'util/paths',
+  'session-persistence/session-checkpoint-policy', 'session-persistence/session-persistence-jsonl',
+  'session-query/session-query', 'context/session-reference', 'ui/acp', 'examples/acp-demo', 'util/paths',
 ]
 const vendorPackages = [
  'cordis', 'loader', 'include', 'timer', 'hmr', 'logger-console',
--- a/packages/examples/acp-demo/tsconfig.json
+++ b/packages/examples/acp-demo/tsconfig.json
@@ -50,6 +50,9 @@
    {
      "path": "../../ui/tool-ask-user"
    },
+    {
+      "path": "../../session-persistence/session-checkpoint-policy"
+    },
    {
      "path": "../../session-persistence/session-persistence-jsonl"
    },
--- a/packages/examples/cli-demo/package.json
+++ b/packages/examples/cli-demo/package.json
@@ -43,6 +43,7 @@
    "@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-checkpoint-policy": "^0.0.1",
    "@deepseek-ai/dsh-session-persistence-jsonl": "^0.0.1",
    "@deepseek-ai/dsh-tools": "^0.0.1",
    "@deepseek-ai/dsh-workspace-context": "^0.0.1",
@@ -58,6 +59,7 @@
    "@deepseek-ai/dsh-invariants": "workspace:^",
    "@deepseek-ai/dsh-llm": "workspace:^",
    "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-session-checkpoint-policy": "workspace:^",
    "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
    "@deepseek-ai/dsh-system-prompt": "workspace:^",
    "@deepseek-ai/dsh-tools": "workspace:^",
--- a/packages/examples/cli-demo/src/index.ts
+++ b/packages/examples/cli-demo/src/index.ts
@@ -15,6 +15,7 @@ import SessionPersistenceJsonl, {
  JsonlCompressionSchema,
  type JsonlCompression,
 } from '@deepseek-ai/dsh-session-persistence-jsonl'
+import * as sessionCheckpointPolicy from '@deepseek-ai/dsh-session-checkpoint-policy'
 import * as workspaceContext from '@deepseek-ai/dsh-workspace-context'

 const DEFAULT_PERSISTENCE_ROOT = './.sessions'
@@ -94,4 +95,5 @@ export function apply(ctx: Context, config: Config): void {
    root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
    ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
  })
+  ctx.plugin(sessionCheckpointPolicy)
 }
--- a/packages/examples/cli-demo/tests/built-bin.e2e.ts
+++ b/packages/examples/cli-demo/tests/built-bin.e2e.ts
@@ -24,7 +24,8 @@ const dshPackages = [
  'examples/agent-spine-demo', 'examples/cli-demo', 'core/agent', 'core/session',
  'core/system-prompt', 'core/tools', 'core/agent-loop', 'llm/llm', 'bash/bash',
  'bash/bash-local', 'bash/tool-bash', 'support/invariants', 'ui/app-boot',
-  'session-persistence/session-persistence', 'session-persistence/session-persistence-jsonl',
+  'session-persistence/session-persistence', 'session-persistence/session-checkpoint-policy',
+  'session-persistence/session-persistence-jsonl',
  'context/workspace-context',
  'spill/spill', 'spill/spill-local', 'spill/spill-policy', 'util/retention',
 ]
--- a/packages/examples/cli-demo/tsconfig.json
+++ b/packages/examples/cli-demo/tsconfig.json
@@ -32,6 +32,9 @@
    {
      "path": "../agent-spine-demo"
    },
+    {
+      "path": "../../session-persistence/session-checkpoint-policy"
+    },
    {
      "path": "../../session-persistence/session-persistence-jsonl"
    },
--- a/packages/examples/tui-demo/README.md
+++ b/packages/examples/tui-demo/README.md
@@ -12,6 +12,7 @@ Use [`@deepseek-ai/dsh-cli-demo`](../cli-demo/README.md) for pipes, scripts, and
 | `@deepseek-ai/dsh-commands` | Human-only discovery and dispatch consumed by the TUI and command plugins |
 | `@deepseek-ai/dsh-command-goal` | Direct `/goal` status and mutation over the spine's persisted-goal stack |
 | `@deepseek-ai/dsh-session-persistence-jsonl` | Durable session log under `persistenceRoot` |
+| `@deepseek-ai/dsh-session-checkpoint-policy` | Semantic durability barriers before model requests and top-level tool effects, plus completed-step checkpoints |
 | `@deepseek-ai/dsh-session-query` + `@deepseek-ai/dsh-session-reference` | Exact current-surface reads and bounded `@session` snapshots consumed by the TUI |
 | `@deepseek-ai/dsh-user-interaction` | Provider-neutral human question service |
 | `@deepseek-ai/dsh-tui` | Full-screen transcript, editor, tool cards, plan, and question overlays |
@@ -38,6 +39,7 @@ Swappable LLM, bash, filesystem, and other capability providers remain in the le
 | `workspaceContext` | required | Workspace-instruction config, or `false` |
 | `persistenceRoot` | `./.sessions` | JSONL persistence root |
 | `persistenceCompression` | `'zstd'` | JSONL artifact encoding (`'zstd'` or raw `'none'`) |
+| `sessionReferences` | service defaults | Cross-session candidate and snapshot limits routed to `dsh-session-reference` |
 | `welcome` | `ready.` | TUI subtitle |
 | `ui` | owner defaults | TUI presentation settings such as reasoning, color, and card height |
 | `resumeSessionId` | — | Exact persisted session to resume |
--- a/packages/examples/tui-demo/package.json
+++ b/packages/examples/tui-demo/package.json
@@ -46,6 +46,7 @@
    "@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-checkpoint-policy": "^0.0.1",
    "@deepseek-ai/dsh-session-query": "^0.0.1",
    "@deepseek-ai/dsh-session-reference": "^0.0.1",
    "@deepseek-ai/dsh-session-persistence-jsonl": "^0.0.1",
@@ -69,6 +70,7 @@
    "@deepseek-ai/dsh-invariants": "workspace:^",
    "@deepseek-ai/dsh-llm": "workspace:^",
    "@deepseek-ai/dsh-session": "workspace:^",
+    "@deepseek-ai/dsh-session-checkpoint-policy": "workspace:^",
    "@deepseek-ai/dsh-session-query": "workspace:^",
    "@deepseek-ai/dsh-session-reference": "workspace:^",
    "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
--- a/packages/examples/tui-demo/src/index.ts
+++ b/packages/examples/tui-demo/src/index.ts
@@ -21,6 +21,7 @@ import SessionPersistenceJsonl, {
  JsonlCompressionSchema,
  type JsonlCompression,
 } from '@deepseek-ai/dsh-session-persistence-jsonl'
+import * as sessionCheckpointPolicy from '@deepseek-ai/dsh-session-checkpoint-policy'
 import UserInteractionService from '@deepseek-ai/dsh-user-interaction'
 import SessionQueryService from '@deepseek-ai/dsh-session-query'
 import SessionReferenceService, { type Config as SessionReferenceConfig } from '@deepseek-ai/dsh-session-reference'
@@ -124,6 +125,7 @@ export function composeTuiApp(ctx: Context, config: Config): void {
    root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
    ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
  })
+  ctx.plugin(sessionCheckpointPolicy)
  ctx.plugin(SessionQueryService)
  ctx.plugin(SessionReferenceService, config.sessionReferences ?? {})
  ctx.plugin(UserInteractionService)
--- a/packages/examples/tui-demo/tests/tui-agent.spec.ts
+++ b/packages/examples/tui-demo/tests/tui-agent.spec.ts
@@ -50,6 +50,7 @@ describe('dsh-tui-demo app', () => {
      'CommandService',
      'command-goal',
      'SessionPersistenceJsonl',
+      'session-checkpoint-policy',
      'SessionQueryService',
      'SessionReferenceService',
      'UserInteractionService',
@@ -59,12 +60,12 @@ describe('dsh-tui-demo app', () => {
    ])
    expect(calls[0]?.config).toBeUndefined()
    expect(calls[2]?.config).toEqual({ root: '/tmp/tui-sessions', compression: 'none' })
-    expect(calls[4]?.config).toEqual({
+    expect(calls[5]?.config).toEqual({
      maxReferences: 2,
      candidateLimit: 7,
      maxReferenceBytes: 1234,
    })
-    const tuiConfig = calls[6]?.config as { sessionId: string }
+    const tuiConfig = calls[7]?.config as { sessionId: string }
    expect(tuiConfig).toMatchObject({
      welcome: 'TUI ready',
      resumeCommand: 'dsh --resume {session}',
@@ -72,7 +73,7 @@ describe('dsh-tui-demo app', () => {
      maxToolOutputLines: 3,
    })
    expect(tuiConfig.sessionId).toMatch(/^main-session-[0-9a-f-]{36}$/)
-    const spineConfig = calls[7]?.config as {
+    const spineConfig = calls[8]?.config as {
      readonly agents: Array<Record<string, unknown>>
      readonly goals: Record<string, never>
      readonly maxParallelToolCalls: number
@@ -106,10 +107,10 @@ describe('dsh-tui-demo app', () => {
    })

    expect(calls[2]?.config).toEqual({ root: './.sessions' })
-    expect(calls[4]?.config).toEqual({})
+    expect(calls[5]?.config).toEqual({})
    // No configured welcome forwards none: the TUI banner sweeps in without a subtitle.
-    expect(calls[6]?.config).toEqual({ sessionId: 'persisted-session' })
-    expect((calls[7]?.config as { agents: Array<Record<string, unknown>> }).agents[0]).toMatchObject({
+    expect(calls[7]?.config).toEqual({ sessionId: 'persisted-session' })
+    expect((calls[8]?.config as { agents: Array<Record<string, unknown>> }).agents[0]).toMatchObject({
      id: 'main',
      resumeSessionId: 'persisted-session',
    })
@@ -125,12 +126,12 @@ describe('dsh-tui-demo app', () => {
      workspaceContext: false,
    })

-    const tuiConfig = calls[5]?.config as { sessionId: string }
+    const tuiConfig = calls[6]?.config as { sessionId: string }
    expect(tuiConfig.sessionId).toMatch(/^main-session-[0-9a-f-]{36}$/)
-    expect((calls[6]?.config as { agents: Array<Record<string, unknown>> }).agents[0])
+    expect((calls[7]?.config as { agents: Array<Record<string, unknown>> }).agents[0])
      .toMatchObject({ sessionId: tuiConfig.sessionId })
    expect(calls.map(call => call.name)).not.toContain('command-goal')
-    expect(calls[6]?.config).toMatchObject({ goals: false })
+    expect(calls[7]?.config).toMatchObject({ goals: false })
  })

  it('has the namespace-plugin export shape so the Loader keeps its schema', () => {
--- a/packages/examples/tui-demo/tsconfig.json
+++ b/packages/examples/tui-demo/tsconfig.json
@@ -53,6 +53,9 @@
    {
      "path": "../../ui/tool-ask-user"
    },
+    {
+      "path": "../../session-persistence/session-checkpoint-policy"
+    },
    {
      "path": "../../session-persistence/session-persistence-jsonl"
    },
--- a/packages/session-persistence/README.md
+++ b/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`) |

--- a/packages/session-persistence/session-checkpoint-policy/README.md
+++ b/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.
--- a/packages/session-persistence/session-checkpoint-policy/package.json
+++ b/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"
+  }
+}
--- a/packages/session-persistence/session-checkpoint-policy/src/index.ts
+++ b/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))
+}
--- a/packages/session-persistence/session-checkpoint-policy/src/invariant.ts
+++ b/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 */
--- a/packages/session-persistence/session-checkpoint-policy/tests/crash-recovery.e2e.ts
+++ b/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.')
+  })
+})
--- a/packages/session-persistence/session-checkpoint-policy/tests/fixtures/crash-child.ts
+++ b/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()
--- a/packages/session-persistence/session-checkpoint-policy/tests/session-checkpoint-policy.spec.ts
+++ b/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')
+  })
+})
--- a/packages/session-persistence/session-checkpoint-policy/tsconfig.json
+++ b/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"
+    }
+  ]
+}
--- a/packages/session-persistence/session-persistence-jsonl/README.md
+++ b/packages/session-persistence/session-persistence-jsonl/README.md
@@ -32,7 +32,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.

@@ -46,7 +46,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

--- a/packages/session-persistence/session-persistence-jsonl/src/format.ts
+++ b/packages/session-persistence/session-persistence-jsonl/src/format.ts
@@ -211,8 +211,8 @@ export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionE
    }
  })

-  // 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 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--) {
    const p = parsed[i]
--- a/packages/session-persistence/session-persistence-sqlite/README.md
+++ b/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

--- a/packages/session-persistence/session-persistence-sqlite/src/schema.ts
+++ b/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 }
--- a/packages/session-persistence/session-persistence/README.md
+++ b/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

--- a/packages/session-persistence/session-persistence/tests/contract.ts
+++ b/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 {
--- a/packages/support/acp-snapshot/src/harness.ts
+++ b/packages/support/acp-snapshot/src/harness.ts
@@ -21,6 +21,7 @@ import { existsSync } from 'node:fs'
 import { createHash } from 'node:crypto'
 import { tmpdir } from 'node:os'
 import { basename, dirname, join, delimiter } from 'node:path'
+import { setTimeout as delay } from 'node:timers/promises'
 import {
  ClientSideConnection,
  PROTOCOL_VERSION,
@@ -34,6 +35,9 @@ import { launchAcpTestAgent, type AgentUnderTest, type LaunchedAcpTestAgent } fr

 export type { AgentUnderTest } from './launcher.ts'

+const DEFAULT_WAIT_TIMEOUT_MS = 10_000
+const WAIT_POLL_INTERVAL_MS = 10
+
 /**
 * One step of a scenario's deterministic input script (`input.json`). The
 * harness interprets these in order. `newSession` captures the server-issued
@@ -42,10 +46,13 @@ export type { AgentUnderTest } from './launcher.ts'
 *
 * `promptAndCancel` starts a prompt without awaiting completion, waits until
 * the client observes the selected update (`agent_message_chunk` by default),
- * then cancels and awaits completion. A named `waitForToolCallUpdate` keeps the
- * step open for a terminal tool update that may follow the prompt response.
+ * then cancels and awaits completion. An optional `waitForFile` first observes
+ * a cwd-relative readiness marker, and a named `waitForToolCallUpdate` keeps
+ * the step open for a terminal tool update that may follow the prompt response.
 * `promptAndWaitForAgentMessage` arms an exact text-chunk waiter before sending
 * the prompt, then keeps the application live until that later update arrives.
+ * `waitForTurnEnd` holds the subprocess open until the selected session's latest
+ * complete raw-JSONL turn boundary is `turn/end`; its timeout defaults to 10s.
 */
 export type InputStep =
  | { op: 'initialize'; terminalOutput?: boolean }
@@ -58,8 +65,10 @@ export type InputStep =
    op: 'promptAndCancel'
    text: string
    afterUpdate?: 'agent_message_chunk' | 'tool_call'
+    waitForFile?: { path: string; timeoutMs?: number }
    waitForToolCallUpdate?: string
  }
+  | { op: 'waitForTurnEnd'; timeoutMs?: number }
  | { op: 'cancel' }
  | { op: 'setMode'; modeId: string }
  | { op: 'setModeExpectError'; modeId: string }
@@ -295,7 +304,15 @@ export async function runScenario(input: InputScript, opts: RunOptions): Promise
    const { client } = active

    for (const step of input.steps) {
-      await runStep(client, step, cwd, match => active.waitForUpdate(match), () => sessionId, (id) => { sessionId = id })
+      await runStep(
+        client,
+        step,
+        cwd,
+        match => active.waitForUpdate(match),
+        () => sessionId,
+        (id) => { sessionId = id },
+        (id, timeoutMs) => waitForPersistedTurnEnd(sessionsRoot, id, timeoutMs),
+      )
      // A permission exchange happens while a step's request is in flight, so
      // by the time the step settles any script bug it exposed is captured —
      // fail the run HERE, as a harness error, rather than hoping the agent's
@@ -365,6 +382,7 @@ async function runStep(
  waitForUpdate: (match: (u: SessionNotification['update']) => boolean) => Promise<SessionNotification['update']>,
  getSessionId: () => string | undefined,
  setSessionId: (id: string) => void,
+  waitForTurnEnd: (sessionId: string, timeoutMs?: number) => Promise<void>,
 ): Promise<void> {
  switch (step.op) {
    case 'initialize':
@@ -429,6 +447,9 @@ async function runStep(
      const promptDone = client.prompt({ sessionId, prompt: [{ type: 'text', text: step.text }] })
      const afterUpdate = step.afterUpdate ?? 'agent_message_chunk'
      await waitForUpdate(u => u.sessionUpdate === afterUpdate)
+      if (step.waitForFile !== undefined) {
+        await waitForWorkspaceFile(cwd, step.waitForFile.path, step.waitForFile.timeoutMs)
+      }
      // Arm this before cancellation so a fast tool drain cannot outrun the waiter.
      const toolCallUpdateDone = step.waitForToolCallUpdate === undefined
        ? undefined
@@ -438,6 +459,12 @@ async function runStep(
      if (toolCallUpdateDone !== undefined) await toolCallUpdateDone
      return
    }
+    case 'waitForTurnEnd': {
+      const sessionId = getSessionId()
+      if (sessionId === undefined) throw new Error('snapshot-harness: waitForTurnEnd before newSession')
+      await waitForTurnEnd(sessionId, step.timeoutMs)
+      return
+    }
    case 'cancel': {
      const sessionId = getSessionId()
      if (sessionId === undefined) throw new Error('snapshot-harness: cancel before newSession')
@@ -485,6 +512,51 @@ async function runStep(
  }
 }

+/**
+ * Wait until the raw JSONL backend exposes one complete closing turn boundary.
+ * The ACP cancel notification settles its prompt before the agent necessarily
+ * reaches quiescence, so cancellation snapshots use this external boundary to
+ * keep subprocess disposal from changing an `aborted` turn into `disposed`.
+ */
+async function waitForPersistedTurnEnd(
+  root: string,
+  sessionId: string,
+  timeoutMs = DEFAULT_WAIT_TIMEOUT_MS,
+): Promise<void> {
+  const deadline = Date.now() + timeoutMs
+  while (true) {
+    const log = (await harvestSessionLogs(root)).find(candidate => candidate.id === sessionId)
+    if (log !== undefined && latestTurnIsClosed(log.content)) return
+    if (Date.now() >= deadline) {
+      throw new Error(`snapshot-harness: session "${sessionId}" did not persist turn/end within ${timeoutMs}ms`)
+    }
+    await delay(WAIT_POLL_INTERVAL_MS)
+  }
+}
+
+/** Wait for a cwd-relative marker proving an external action reached readiness. */
+async function waitForWorkspaceFile(
+  cwd: string,
+  path: string,
+  timeoutMs = DEFAULT_WAIT_TIMEOUT_MS,
+): Promise<void> {
+  const target = join(cwd, path)
+  const deadline = Date.now() + timeoutMs
+  while (!existsSync(target)) {
+    if (Date.now() >= deadline) {
+      throw new Error(`snapshot-harness: workspace file "${path}" did not appear within ${timeoutMs}ms`)
+    }
+    await delay(WAIT_POLL_INTERVAL_MS)
+  }
+}
+
+/** Return whether the last complete raw-JSONL turn boundary closes its turn. */
+function latestTurnIsClosed(content: string): boolean {
+  const complete = content.slice(0, content.lastIndexOf('\n') + 1)
+  return complete.lastIndexOf('\n{"type":"turn/end",')
+    > complete.lastIndexOf('\n{"type":"turn/start",')
+}
+
 /**
 * Harvest EVERY persisted `.jsonl` session log under a sessions root, parse each
 * header line, and return them ordered primary-first: the top-level session (no
--- a/packages/support/acp-snapshot/tests/fixtures/fake-acp-agent.ts
+++ b/packages/support/acp-snapshot/tests/fixtures/fake-acp-agent.ts
@@ -49,6 +49,8 @@ interface Behavior {
  cancelAtToolCall?: boolean
  /** Emit the parked tool call's terminal update after answering cancellation. */
  cancelToolCallUpdate?: boolean
+  /** Persist the scripted logs while handling cancellation, before stdin EOF. */
+  persistLogsOnCancel?: boolean
  /** Before responding to a prompt, send a `session/request_permission` request and echo its outcome as a chunk. */
  permissionProbe?: boolean
  /** Before responding to a prompt, send an `elicitation/create` request and echo its response as a chunk. */
@@ -63,7 +65,7 @@ interface Behavior {
  stderrNote?: string
  /** Let a short-lived descendant retain stdio and emit one final ACP update plus stderr line after this parent exits. */
  lateInheritedOutput?: boolean
-  /** Session logs to persist on stdin EOF. */
+  /** Session logs to persist on stdin EOF and, when selected, on cancellation. */
  logs?: ScriptedLog[]
  /** Leave a stray FILE directly under the sessions root (harvest must skip it). */
  strayRootFile?: boolean
@@ -304,6 +306,7 @@ function handleFrame(frame: Record<string, unknown>): void {
            },
          })
        }
+        if (behavior.persistLogsOnCancel === true) writeLogs()
      }
      return
    default:
@@ -313,12 +316,16 @@ function handleFrame(frame: Record<string, unknown>): void {
  }
 }

-function flushLogsAndExit(): void {
+function writeLogs(): void {
  for (const log of behavior.logs ?? []) {
    const target = join(sessionsRoot, log.file)
    mkdirSync(dirname(target), { recursive: true })
    writeFileSync(target, log.lines.map(l => JSON.stringify(instantiate(l))).join('\n') + '\n')
  }
+}
+
+function flushLogsAndExit(): void {
+  writeLogs()
  if (behavior.strayRootFile === true) writeFileSync(join(sessionsRoot, 'stray.txt'), 'not a bucket\n')
  if (behavior.strayBucketFile === true) {
    mkdirSync(join(sessionsRoot, 'bucket-noise'), { recursive: true })
--- a/packages/support/acp-snapshot/tests/harness.spec.ts
+++ b/packages/support/acp-snapshot/tests/harness.spec.ts
@@ -493,6 +493,37 @@ describe('runScenario', () => {
    expect(result.rawStdout.indexOf('thinking about it')).toBeLessThan(result.rawStdout.indexOf('cancelled'))
  })

+  it('promptAndCancel can wait for cwd-relative readiness before cancelling', { timeout: 20_000 }, async () => {
+    const { dir, fixtureFile } = await scenario({ prompt: 'hang-until-cancel' })
+    const workspaceDir = join(dir, 'workspace')
+    const { mkdir } = await import('node:fs/promises')
+    await mkdir(workspaceDir, { recursive: true })
+    await writeFile(join(workspaceDir, 'started.txt'), 'started')
+    const result = await runScenario(
+      {
+        steps: [...boot, {
+          op: 'promptAndCancel',
+          text: 'hang',
+          waitForFile: { path: 'started.txt' },
+        }],
+      },
+      { agent: AGENT, mode: 'replay', fixtureFile, workspaceDir },
+    )
+    expect(result.rawStdout).toContain('"stopReason":"cancelled"')
+
+    const missing = await scenario({ prompt: 'hang-until-cancel' })
+    await expect(runScenario(
+      {
+        steps: [...boot, {
+          op: 'promptAndCancel',
+          text: 'hang',
+          waitForFile: { path: 'never.txt', timeoutMs: 20 },
+        }],
+      },
+      { agent: AGENT, mode: 'replay', fixtureFile: missing.fixtureFile },
+    )).rejects.toThrow(/workspace file "never\.txt" did not appear within 20ms/)
+  })
+
  it('promptAndWaitForAgentMessage keeps the app live through a matching later update', { timeout: 20_000 }, async () => {
    const { fixtureFile } = await scenario({ prompt: 'respond' })
    const result = await runScenario(
@@ -530,6 +561,55 @@ describe('runScenario', () => {
    expect(result.rawStdout.indexOf('cancelled')).toBeLessThan(result.rawStdout.indexOf('"sessionUpdate":"tool_call_update"'))
  })

+  it('waitForTurnEnd holds cancellation open through the persisted closing boundary', { timeout: 20_000 }, async () => {
+    const { fixtureFile } = await scenario({
+      prompt: 'hang-until-cancel',
+      persistLogsOnCancel: true,
+      logs: [{
+        file: 'bucket/session.jsonl',
+        lines: [
+          { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
+          { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'aborted' } } },
+        ],
+      }],
+    })
+    const result = await runScenario(
+      { steps: [...boot, { op: 'promptAndCancel', text: 'hang' }, { op: 'waitForTurnEnd' }] },
+      { agent: AGENT, mode: 'replay', fixtureFile },
+    )
+    expect(result.sessionLogs[0]?.content).toContain('"type":"turn/end"')
+  })
+
+  it('waitForTurnEnd times out for a missing log and an open logged turn', { timeout: 20_000 }, async () => {
+    const missing = await scenario({})
+    await expect(runScenario(
+      { steps: [...boot, { op: 'waitForTurnEnd', timeoutMs: 20 }] },
+      { agent: AGENT, mode: 'replay', fixtureFile: missing.fixtureFile },
+    )).rejects.toThrow(/did not persist turn\/end within 20ms/)
+
+    const open = await scenario({
+      prompt: 'hang-until-cancel',
+      persistLogsOnCancel: true,
+      logs: [{
+        file: 'bucket/session.jsonl',
+        lines: [
+          { type: 'session', version: 0, id: '{{SID}}', createdAt: 1, delegationDepth: 0 },
+          { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
+        ],
+      }],
+    })
+    await expect(runScenario(
+      {
+        steps: [
+          ...boot,
+          { op: 'promptAndCancel', text: 'hang' },
+          { op: 'waitForTurnEnd', timeoutMs: 20 },
+        ],
+      },
+      { agent: AGENT, mode: 'replay', fixtureFile: open.fixtureFile },
+    )).rejects.toThrow(/did not persist turn\/end within 20ms/)
+  })
+
  it('promptExpectError swallows a model-error response as the expected outcome', { timeout: 20_000 }, async () => {
    const { fixtureFile } = await scenario({ prompt: 'error' })
    const result = await runScenario(
@@ -620,6 +700,7 @@ describe('runScenario', () => {
    [{ op: 'promptAndWaitForAgentMessage', text: 'x', waitForText: 'later' }, /promptAndWaitForAgentMessage before newSession/],
    [{ op: 'promptExpectError', text: 'x' }, /promptExpectError before newSession/],
    [{ op: 'promptAndCancel', text: 'x' }, /promptAndCancel before newSession/],
+    [{ op: 'waitForTurnEnd' }, /waitForTurnEnd before newSession/],
    [{ op: 'cancel' }, /cancel before newSession/],
    [{ op: 'setConfigOption', configId: 'sandbox-mode', value: 'read-only' }, /setConfigOption before newSession/],
    [{ op: 'setConfigOptionExpectError', configId: 'sandbox-mode', value: 'yolo' }, /setConfigOptionExpectError before newSession/],