fix(session): close checkpoint cancellation races

packages/core/agent-loop/README.md

@@ -56,7 +56,7 @@ The driver owns one agent for its lifetime and runs inside `ctx.agents.withIniti
 
 Every provider call that reaches a successful finish appends exactly one `assistant/message` completion anchor, including content-less calls and `max-tokens` finishes. A successful `agent/step-result` stores its transformed content; a rejected result records empty content before the original failure continues. The anchor retains exact chunk provenance (`[]` for a stream with no chunks) and usage when available, while empty content stays out of derived message history.
 
-Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery receives the exact live error, immutable provider facts, and immutable prior failures after the failed step closes. A retry rebuilds from the durable log in a new numbered step, success clears the consecutive history, and exhaustion records the structured failure once on `turn/end`. Effective cancellation resolves its reason and emits `agent/cancel-requested` before clearing pending work or aborting the current step; notification failures are contained, queued work added by an observer is included in the same broad clear, and idle cancellation emits nothing. Undispatched model tool calls receive synthetic `tool/call` and aborted result pairs. Terminal continuation stops remain authoritative through turn close and durability flush.
+Plugin failure ends the current turn, not the loop. Only final adapter dispatch/iteration failures and terminal in-band error or aborted finishes enter `agent/request-error`; middleware, result processing, tools, and `agent/post-step` remain ordinary turn failures. Recovery receives the exact live error, immutable provider facts, and immutable prior failures after the failed step closes. A retry rebuilds from the durable log in a new numbered step, success clears the consecutive history, and exhaustion records the structured failure once on `turn/end`. Effective cancellation resolves its reason and emits `agent/cancel-requested` before clearing pending work or aborting the current step; notification failures are contained, queued work added by an observer is included in the same broad clear, and idle cancellation emits nothing. If structural disposal follows an effective cancellation before the turn closes, the earlier cancellation retains its `aborted` reason; disposal alone closes the turn as `disposed`. Undispatched model tool calls receive synthetic `tool/call` and aborted result pairs. Terminal continuation stops remain authoritative through turn close and durability flush.
 
 Within a step, exclusive calls form barriers; parallel-safe calls use a bounded rolling pool and are reclassified before start. Only dispatch/body overlaps. Policy, durable results, and result context remain model-ordered. Abort stops new calls, drains started results, then drains accepted batch context before the turn closes through the normal abort path.
 

packages/core/agent-loop/src/loop.ts

@@ -88,6 +88,19 @@ function stepFinishReason(finish: FinishReason): TurnEndReason | undefined {
   }
 }
 
+/** Classify an interruption before a step controller can provide an abort reason. */
+function pendingInterruptionReason(handle: LoopHandle): TurnEndReason {
+  if (handle.isCancelled()) return { kind: 'aborted', reason: handle.cancelReason() }
+  return { kind: 'disposed' }
+}
+
+/** Preserve an effective cancel reason when structural disposal follows it. */
+function stepInterruptionReason(handle: LoopHandle, signal: AbortSignal): TurnEndReason {
+  if (handle.isCancelled()) return { kind: 'aborted', reason: handle.cancelReason() }
+  if (handle.isDisposed()) return { kind: 'disposed' }
+  return { kind: 'aborted', reason: String(signal.reason) }
+}
+
 /** Mutable agent controls supplied to the loop driver. */
 export interface LoopHandle {
   /** Native-private agent inbox handed to the driver only at internal startup. */
@@ -307,7 +320,7 @@ async function runTurn(
       // Cancellation or disposal during assembly ends the turn before any step opens.
       if (handle.isCancelled() || handle.isDisposed()) {
         handle.setAbort(undefined)
-        reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
+        reason = pendingInterruptionReason(handle)
         break
       }
 
@@ -324,7 +337,7 @@ async function runTurn(
         // Never cache an interrupted composition; the next turn recomposes it.
         if (handle.isCancelled() || handle.isDisposed()) {
           handle.setAbort(undefined)
-          reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
+          reason = pendingInterruptionReason(handle)
           break
         }
         transmission.sessionPrefix = deepFreeze(structuredClone(composed))
@@ -336,7 +349,7 @@ async function runTurn(
       // Interruption landing during the pre-step seam: do not open an empty step.
       if (handle.isCancelled() || handle.isDisposed()) {
         handle.setAbort(undefined)
-        reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
+        reason = pendingInterruptionReason(handle)
         break
       }
 
@@ -356,7 +369,7 @@ async function runTurn(
       // turn accordingly. closeStep balances the already-appended step/start.
       if (handle.isCancelled() || handle.isDisposed()) {
         handle.setAbort(undefined)
-        reason = handle.isDisposed() ? { kind: 'disposed' } : { kind: 'aborted', reason: handle.cancelReason() }
+        reason = pendingInterruptionReason(handle)
         closeStep()
         break
       }
@@ -382,9 +395,7 @@ async function runTurn(
         closeStep()
         if (handle.isDisposed() || abort.signal.aborted) {
           handle.setAbort(undefined)
-          reason = handle.isDisposed()
-            ? { kind: 'disposed' }
-            : { kind: 'aborted', reason: String(abort.signal.reason) }
+          reason = stepInterruptionReason(handle, abort.signal)
           break
         }
 
@@ -407,9 +418,7 @@ async function runTurn(
         // or a recovery-listener failure.
         // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
         if (handle.isDisposed() || abort.signal.aborted) {
-          reason = handle.isDisposed()
-            ? { kind: 'disposed' }
-            : { kind: 'aborted', reason: String(abort.signal.reason) }
+          reason = stepInterruptionReason(handle, abort.signal)
           break
         }
         switch (recoveryDecision.action) {
@@ -434,11 +443,8 @@ async function runTurn(
         handle.setAbort(undefined)
         const { error } = stepOutcome
         /* v8 ignore next -- narrow race: disposal while non-request step work throws. */
-        if (handle.isDisposed()) {
-          reason = { kind: 'disposed' }
-        } else if (abort.signal.aborted) {
-          /* v8 ignore next -- signal.reason always set: cancel()/disposal provide a default */
-          reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
+        if (handle.isDisposed() || handle.isCancelled() || abort.signal.aborted) {
+          reason = stepInterruptionReason(handle, abort.signal)
         } else {
           failTurn(error)
         }
@@ -464,11 +470,8 @@ async function runTurn(
         closeStep()
         handle.setAbort(undefined)
         /* v8 ignore next -- narrow race: disposal while a post-step listener throws. */
-        if (handle.isDisposed()) {
-          reason = { kind: 'disposed' }
-        } else if (abort.signal.aborted) {
-          /* v8 ignore next -- signal.reason always set by cancellation or disposal. */
-          reason = { kind: 'aborted', reason: String(abort.signal.reason ?? 'aborted') }
+        if (handle.isDisposed() || handle.isCancelled() || abort.signal.aborted) {
+          reason = stepInterruptionReason(handle, abort.signal)
         } else {
           failTurn(stepOutcome.error)
         }
@@ -476,9 +479,7 @@ async function runTurn(
       }
 
       if (handle.isDisposed() || abort.signal.aborted) {
-        reason = handle.isDisposed()
-          ? { kind: 'disposed' }
-          : { kind: 'aborted', reason: String(abort.signal.reason) }
+        reason = stepInterruptionReason(handle, abort.signal)
         closeStep()
         handle.setAbort(undefined)
         break

packages/core/agent-loop/tests/cancel.spec.ts

@@ -343,6 +343,35 @@ describe('Agent.cancel()', () => {
     expect(adapter.requests).toHaveLength(1)
   })
 
+  it('preserves cancellation when disposal follows during post-step work', async () => {
+    const adapter = new MockAdapter([textResponse('done')])
+    const ctx = await harness(adapter)
+    const entered = Promise.withResolvers<undefined>()
+    const release = Promise.withResolvers<undefined>()
+    const handle = await ctx.agents.create({
+      sessionId: SessionId('cancel-then-dispose'),
+      agentOptions: { provider: 'mock', model: 'mock' },
+    })
+    const agent = handle.agent
+    ctx.on('agent/post-step', async (subject) => {
+      if (subject !== agent) return
+      entered.resolve(undefined)
+      await release.promise
+    })
+
+    send(agent, 'go')
+    await entered.promise
+    agent.cancel('user cancelled')
+    const disposed = handle.dispose()
+    release.resolve(undefined)
+    await disposed
+
+    const turnEnd = agent.session.events.findLast(event => event.type === 'turn/end')
+    expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason)
+      .toEqual({ kind: 'aborted', reason: 'user cancelled' })
+    await ctx.fiber.dispose()
+  })
+
   it('cancel() with no reason defaults to "cancelled" when aborting an in-flight step', async () => {
     const adapter = new MockAdapter(['hang'])
     const ctx = await harness(adapter)

packages/examples/acp-demo/README.md

@@ -15,13 +15,14 @@ stdout is the ACP JSON-RPC channel, so the cluster is defined as much by what it
 | `@deepseek-ai/dsh-command-goal` | the discoverable direct `/goal` producer; the app enables the spine's persisted-goal stack with 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-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
 

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
@@ -99,18 +101,22 @@ 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 })
-  ctx.plugin(UserInteractionService)
-  ctx.plugin(SessionPersistenceJsonl, {
-    root: config.persistenceRoot ?? DEFAULT_PERSISTENCE_ROOT,
-    ...(config.persistenceCompression === undefined ? {} : { compression: config.persistenceCompression }),
-  })
-  ctx.plugin(sessionCheckpointPolicy)
-  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(acp, { provider: config.provider, model: config.model }).dispose
+  }, 'acp-demo.composition')
 }

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

@@ -16,7 +16,7 @@ This zero-config function plugin consumes `ctx.sessions`, `ctx.llm`, `ctx.tools`
 
 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. Nested tool dispatches reuse the outer model-visible call's checkpoint. `agent/post-step` persists the complete response/result batch before continuation work.
+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` 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.
 

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

@@ -37,6 +37,15 @@ function afterCheckpoint(
   })()
 }
 
+/** Materialize the canonical result for a call cancelled before tool dispatch. */
+function abortedToolResult(): ToolExecutionResult {
+  return {
+    content: [{ type: 'text', text: 'Error: tool call skipped because the step was aborted before execution' }],
+    isError: true,
+    error: { name: 'AbortError', code: 'ABORTED' },
+  }
+}
+
 /**
  * Install semantic checkpoint listeners. Loop-built model calls checkpoint the
  * logged request before adapter dispatch; top-level tool calls checkpoint their
@@ -58,6 +67,7 @@ export function apply(ctx: Context): void {
   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 === true) return abortedToolResult()
     return next()
   })
 

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

@@ -134,6 +134,40 @@ describe('session-checkpoint-policy tool and step boundaries', () => {
     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 skipped because the step was aborted before execution' }],
+      isError: true,
+      error: { name: 'AbortError', code: 'ABORTED' },
+    })
+    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'))