From 380a94febba74b1973c5541a34ce376c226eb383 Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Tue, 21 Jul 2026 17:58:34 +0800 Subject: [PATCH] fix(session): close checkpoint cancellation races --- ...-21-semantic-session-checkpoints.i18n.yaml | 4 +- ...2026-07-21-semantic-session-checkpoints.md | 6 ++- ...6-07-21-semantic-session-checkpoints.zh.md | 6 ++- docs/architecture.md | 4 +- docs/config-catalog.md | 2 +- .../snapshots/cancel-tool-calls/session.jsonl | 4 +- .../cancel-tool-calls/stdout.expected.jsonl | 2 +- packages/core/agent-loop/README.md | 2 +- packages/core/agent-loop/src/loop.ts | 47 ++++++++++--------- packages/core/agent-loop/tests/cancel.spec.ts | 29 ++++++++++++ packages/examples/acp-demo/README.md | 3 +- packages/examples/acp-demo/src/index.ts | 30 +++++++----- .../session-checkpoint-policy/README.md | 2 +- .../session-checkpoint-policy/src/index.ts | 10 ++++ .../tests/session-checkpoint-policy.spec.ts | 34 ++++++++++++++ 15 files changed, 136 insertions(+), 49 deletions(-) diff --git a/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.i18n.yaml b/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.i18n.yaml index 82465ffbb6..5907d9bc7f 100644 --- a/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.i18n.yaml +++ b/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write -2026-07-21-semantic-session-checkpoints.md: b672375a177964589ef4a3bf3661d66a2086d0d0 -2026-07-21-semantic-session-checkpoints.zh.md: eedb4df471b17d42fad9b8f1c35f6139b9abe470 +2026-07-21-semantic-session-checkpoints.md: 3dc2699a1a48274ede8e5f9abde0302d5bebeda4 +2026-07-21-semantic-session-checkpoints.zh.md: 0d5adccf4f582286894d0f1404f1d4523ec80f74 diff --git a/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md b/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md index b672375a17..3dc2699a1a 100644 --- a/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md +++ b/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.md @@ -14,7 +14,9 @@ Persistence buffered every synchronous `session/event` until the loop's final tu Persistence and checkpoint scheduling remain separate Cordis plugins. A backend makes requested `session/flush` boundaries durable but does not choose them; loading it without this policy is valid and retains the loop's coarser checkpoints. First-party persisted apps and runtimes explicitly mount both, while a specialized deployment may intentionally omit or replace the policy. Registration order governs whether events appended by other `agent/post-step` listeners join this checkpoint; the loop-owned assistant message and ordered results always precede the event. -Checkpoint failure is fail-closed at effect boundaries. A rejected request checkpoint prevents adapter dispatch; a rejected tool checkpoint becomes an error result without invoking the tool body; a rejected post-step checkpoint stops continuation before another model request. Persistence serialization continues to belong to the coordinator, so concurrent tool checkpoints cannot duplicate event sequences. +Checkpoint failure and cancellation are fail-closed at effect boundaries. A rejected request checkpoint prevents adapter dispatch; a rejected tool checkpoint becomes an error result without invoking the tool body. If cancellation lands while the tool checkpoint is pending, the policy rechecks the signal and returns the canonical `ABORTED` result before dispatch. A rejected post-step checkpoint stops continuation before another model request. Persistence serialization continues to belong to the coordinator, so concurrent tool checkpoints cannot duplicate event sequences. + +The ACP app owns its bridge, checkpoint policy, and persistence backend in one ordered Cordis effect. Cordis unloads sibling plugin effects concurrently, so independent mounts would let persistence detach while bridge teardown was still closing an interrupted turn. The composite lifecycle unloads the bridge first, waits for its agents to quiesce and flush the real `step/end` and `turn/end`, then removes checkpoint scheduling and persistence. If connection disposal follows an effective `session/cancel` while post-step persistence is pending, the loop preserves the earlier `aborted` reason instead of replacing it with `disposed`. Crash repair distinguishes durable evidence. An assistant tool request without a `tool/call` becomes `TOOL_NOT_STARTED` and may be retried if still needed. A durable `tool/call` without a result becomes `TOOL_OUTCOME_UNKNOWN`; its model-visible result permits retry only for read-only or idempotent operations and directs the model to verify external state or ask the user before deciding about side-effecting work. A provider that supports idempotency keys can receive the stable `callId`, but the Harness does not claim generic exactly-once effects. @@ -24,4 +26,4 @@ Flushing every event or streaming chunk minimizes loss but turns local append an ## Consequences -Hard-crash recovery retains the complete model request, durable tool intent, and complete settled step at the nearest semantic boundary while allowing partial streaming chunks since the previous boundary to remain lossy. Default CLI, TUI, ACP, Python SDK runtime, headless persistence tests, and JSON-RPC compositions mount the policy with their persistence backend. Unit tests cover ordering, fail-closed behavior, nested dispatch, disposal, and Loader shape; a real child process killed with `SIGKILL` proves request and tool-intent recovery through JSONL, and the shared persistence contract proves both recovery classifications across backends. A keyless ACP snapshot loads a seeded unknown-outcome session through the shipped ACP example and proves that the retry-risk guidance reaches both resumed history and the next model turn. +Hard-crash recovery retains the complete model request, durable tool intent, and complete settled step at the nearest semantic boundary while allowing partial streaming chunks since the previous boundary to remain lossy. Default CLI, TUI, ACP, Python SDK runtime, headless persistence tests, and JSON-RPC compositions mount the policy with their persistence backend. Unit tests cover ordering, cancellation during a checkpoint, fail-closed behavior, nested dispatch, disposal, and Loader shape; a real child process killed with `SIGKILL` proves request and tool-intent recovery through JSONL, and the shared persistence contract proves both recovery classifications across backends. Keyless ACP snapshots prove both that retry-risk guidance reaches resumed history and the next model turn and that graceful cancellation persists the loop's real closing boundaries. diff --git a/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.zh.md b/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.zh.md index eedb4df471..0d5adccf4f 100644 --- a/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.zh.md +++ b/.agents/notes/implemented/bug-fix/2026-07-21-semantic-session-checkpoints.zh.md @@ -14,7 +14,9 @@ Status: implemented 持久化与检查点调度仍是相互独立的 Cordis 插件。后端使请求的 `session/flush` 边界持久化，但不选择边界；只加载后端而不加载本策略仍是有效组合，并保留循环提供的较粗检查点。第一方持久化应用与运行时会显式加载两者，专用部署则可以有意省略或替换本策略。注册顺序决定其他 `agent/post-step` 监听器追加的事件是否会纳入本检查点；循环自身记录的助手消息与有序结果始终先于该事件。 -检查点失败在副作用边界上采取失败关闭策略。请求检查点被拒绝时，系统不会分发给适配器；工具检查点被拒绝时，系统会返回错误结果，不调用工具主体；步骤后检查点被拒绝时，系统会在发起下一个模型请求前停止继续执行。持久化写入的串行化仍由协调器负责，因此并发的工具检查点不会产生重复的事件序号。 +检查点失败与取消在副作用边界上采取失败关闭策略。请求检查点被拒绝时，系统不会分发给适配器；工具检查点被拒绝时，系统会返回错误结果，不调用工具主体。如果在工具检查点等待期间收到取消，策略会重新检查信号，并在分发前返回标准的 `ABORTED` 结果。步骤后检查点被拒绝时，系统会在发起下一个模型请求前停止继续执行。持久化写入的串行化仍由协调器负责，因此并发的工具检查点不会产生重复的事件序号。 + +ACP（Agent Client Protocol）应用在一个有序 Cordis effect 中统一持有其桥接层、检查点策略与持久化后端。Cordis 会并发卸载同级插件的 effect；如果分别加载，桥接层仍在为被中断的轮次收尾时，持久化后端就可能已经卸载。组合生命周期会先卸载桥接层，等待其各 agent 达到静止，并刷新真实的 `step/end` 与 `turn/end`，再移除检查点调度与持久化。如果在步骤后持久化仍在等待时，连接释放发生在已生效的 `session/cancel` 之后，循环会保留先前的 `aborted` 原因，而不会将其替换为 `disposed`。 崩溃修复会区分持久化证据。如果模型发出了工具请求，却没有 `tool/call`，系统会将其标记为 `TOOL_NOT_STARTED`；如果仍有需要，可以重试。如果持久化的 `tool/call` 没有结果，系统会将其标记为 `TOOL_OUTCOME_UNKNOWN`；对应的模型可见结果只允许重试只读或幂等操作，并指示模型在决定是否重试有副作用的工作前，先验证外部状态或询问用户。支持幂等键的模型提供方可以获取稳定的 `callId`，但 Harness 不承诺通用的副作用恰好执行一次保证。 @@ -24,4 +26,4 @@ Status: implemented ## 后果 -发生硬崩溃时，崩溃恢复会在最近的语义边界保留完整的模型请求、持久化的工具意图与完整且已结束的步骤，但允许上一个边界之后的部分流式分片仍可能丢失。默认的 CLI（命令行界面）、TUI、ACP（Agent Client Protocol）、Python SDK 运行时、headless 持久化测试与 JSON-RPC 组合都会在持久化后端旁加载该策略。单元测试覆盖顺序、失败关闭行为、嵌套分发、dispose（资源释放）与 Loader 形状；一个被 `SIGKILL` 终止的真实子进程通过 JSONL 证明系统可以恢复请求与工具意图，共享持久化契约则证明各后端都支持这两种恢复分类。一项无密钥 ACP 快照通过已交付的 ACP 示例加载预置的结果未知会话，并证明重试风险指引会同时出现在恢复后的历史记录与下一个模型轮次中。 +发生硬崩溃时，崩溃恢复会在最近的语义边界保留完整的模型请求、持久化的工具意图与完整且已结束的步骤，但允许上一个边界之后的部分流式分片仍可能丢失。默认的 CLI（命令行界面）、TUI、ACP、Python SDK 运行时、headless 持久化测试与 JSON-RPC 组合都会在持久化后端旁加载该策略。单元测试覆盖顺序、检查点期间的取消、失败关闭行为、嵌套分发、dispose（资源释放）与 Loader 形状；一个被 `SIGKILL` 终止的真实子进程通过 JSONL 证明系统可以恢复请求与工具意图，共享持久化契约则证明各后端都支持这两种恢复分类。无密钥 ACP 快照既证明重试风险指引会进入恢复后的历史记录与下一个模型轮次，也证明取消流程正常收尾时，系统会持久化由循环实际生成的闭合边界。 diff --git a/docs/architecture.md b/docs/architecture.md index 5878280e28..6c1acc8805 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -118,9 +118,9 @@ Pruning precedes summaries; overflow retries require durable progress. Bounded t The turn is the containment boundary. Adapter failures close the step, entering `agent/request-error` with the exact `Error`, `LlmFailure`, and retry history. Retry opens a numbered step; success clears history; exhaustion stores the failure on `turn/end`. Failed chunks commit no message or tool. -Other failures use `agent/error`. Cancellation beats recovery; undispatched calls get synthetic `ABORTED` results. Effective `cancel()` emits `agent/cancel-requested` before queue clearing or abort; observers cannot veto it, and idle calls emit nothing. Disposal awaits quiescence. +Other failures use `agent/error`. Cancellation beats recovery and later disposal; undispatched calls get synthetic `ABORTED` results. Effective `cancel()` emits `agent/cancel-requested` before clearing or aborting work; observers cannot veto it, and idle calls emit nothing. Disposal alone yields `disposed` and always awaits quiescence. -Every session event is turn-enclosed. Reloading preserves an interrupted tail and closes it with a synthetic `interrupted` turn end. Failures after durable turn close report only through `agent/error` because no safe in-turn position remains. Each turn has one `TurnEndReason`; [TurnEndReasonMap](core-data-structures/session.md#why-a-turn-ended-turnendreasonmap) owns the variants. +Session events are turn-enclosed. Reload closes an interrupted tail with a synthetic `interrupted` turn end. Post-close failures report only through `agent/error`; no safe in-turn position remains. Each turn has one `TurnEndReason`; [TurnEndReasonMap](core-data-structures/session.md#why-a-turn-ended-turnendreasonmap) owns the variants. ### Agent Handles diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 5ca397db73..2a168b8994 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -77,7 +77,7 @@ export interface Config { Depends on: [`agentCore`](../packages/examples/agent-spine-demo/src/index.ts) · [`JsonlCompression`](../packages/session-persistence/session-persistence-jsonl/src/index.ts) · [`ToolsConfig`](#deepseek-aidsh-tools) -Source: [`packages/examples/acp-demo/src/index.ts:39`](../packages/examples/acp-demo/src/index.ts) +Source: [`packages/examples/acp-demo/src/index.ts:41`](../packages/examples/acp-demo/src/index.ts) ## `@deepseek-ai/dsh-agent-loop` diff --git a/examples/acp-agent/tests/snapshots/cancel-tool-calls/session.jsonl b/examples/acp-agent/tests/snapshots/cancel-tool-calls/session.jsonl index 682c13b761..7d39416dd6 100644 --- a/examples/acp-agent/tests/snapshots/cancel-tool-calls/session.jsonl +++ b/examples/acp-agent/tests/snapshots/cancel-tool-calls/session.jsonl @@ -13,6 +13,8 @@ {"type":"assistant/chunk","seq":11,"time":1784437195077,"data":{"turn":1,"step":1,"chunk":{"type":"finish","reason":{"kind":"tool-calls"}}}} {"type":"assistant/message","seq":12,"time":1784437195078,"data":{"turn":1,"step":1,"content":[{"type":"tool-call","id":"call_wait","name":"bash","arguments":"{\"command\":\"node -e \\\"setInterval(() => {}, 1000)\\\"\",\"description\":\"Wait until cancellation\"}"},{"type":"tool-call","id":"call_skipped","name":"bash","arguments":"{\"command\":\"printf skipped > skipped.txt\",\"description\":\"Write skipped marker\"}"}],"provenance":{"provider":"deepseek","model":"deepseek-v4-flash"},"usage":{"inputTokens":10,"outputTokens":10}},"sourceEventSeqs":[4,5,6,7,8,9,10,11],"surfaceOp":"append"} {"type":"tool/call","seq":13,"time":1784437195078,"data":{"turn":1,"step":1,"callId":"call_wait","name":"bash","arguments":"{\"command\":\"node -e \\\"setInterval(() => {}, 1000)\\\"\",\"description\":\"Wait until cancellation\"}"}} -{"type":"tool/result","seq":14,"time":1784437195089,"data":{"turn":1,"step":1,"callId":"call_wait","content":[{"type":"text","text":"Error: command aborted"}],"isError":true},"sourceEventSeqs":[13],"surfaceOp":"append"} +{"type":"tool/result","seq":14,"time":1784437195089,"data":{"turn":1,"step":1,"callId":"call_wait","content":[{"type":"text","text":"Error: tool call skipped because the step was aborted before execution"}],"isError":true,"error":{"name":"AbortError","code":"ABORTED"}},"sourceEventSeqs":[13],"surfaceOp":"append"} {"type":"tool/call","seq":15,"time":1784437195089,"data":{"turn":1,"step":1,"callId":"call_skipped","name":"bash","arguments":"{\"command\":\"printf skipped > skipped.txt\",\"description\":\"Write skipped marker\"}"}} {"type":"tool/result","seq":16,"time":1784437195089,"data":{"turn":1,"step":1,"callId":"call_skipped","content":[{"type":"text","text":"Error: tool call skipped because the step was aborted before execution"}],"isError":true,"error":{"name":"AbortError","code":"ABORTED"}},"sourceEventSeqs":[15],"surfaceOp":"append"} +{"type":"step/end","seq":17,"time":1784437195090,"data":{"turn":1,"step":1}} +{"type":"turn/end","seq":18,"time":1784437195090,"data":{"turn":1,"reason":{"kind":"aborted","reason":"session/cancel"}}} diff --git a/examples/acp-agent/tests/snapshots/cancel-tool-calls/stdout.expected.jsonl b/examples/acp-agent/tests/snapshots/cancel-tool-calls/stdout.expected.jsonl index 9cce0b2875..2b905c13dd 100644 --- a/examples/acp-agent/tests/snapshots/cancel-tool-calls/stdout.expected.jsonl +++ b/examples/acp-agent/tests/snapshots/cancel-tool-calls/stdout.expected.jsonl @@ -3,6 +3,6 @@ {"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"available_commands_update","availableCommands":[{"name":"goal","description":"set or view the goal for a long-running task","input":{"hint":"[|clear|edit |pause|resume]"}}]}}} {"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"tool_call","toolCallId":"call_wait","title":"node -e \"setInterval(() => {}, 1000)\"","kind":"execute","status":"in_progress","rawInput":"node -e \"setInterval(() => {}, 1000)\"","content":[{"type":"content","content":{"type":"text","text":"Wait until cancellation"}}]}}} {"jsonrpc":"2.0","id":3,"result":{"stopReason":"cancelled"}} -{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"tool_call_update","toolCallId":"call_wait","status":"failed","content":[{"type":"content","content":{"type":"text","text":"```console\nError: command aborted\n```"}}]}}} +{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"tool_call_update","toolCallId":"call_wait","status":"failed","content":[{"type":"content","content":{"type":"text","text":"```console\nError: tool call skipped because the step was aborted before execution\n```"}}]}}} {"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"tool_call","toolCallId":"call_skipped","title":"printf skipped > skipped.txt","kind":"execute","status":"in_progress","rawInput":"printf skipped > skipped.txt","content":[{"type":"content","content":{"type":"text","text":"Write skipped marker"}}]}}} {"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"tool_call_update","toolCallId":"call_skipped","status":"failed","content":[{"type":"content","content":{"type":"text","text":"```console\nError: tool call skipped because the step was aborted before execution\n```"}}]}}} diff --git a/packages/core/agent-loop/README.md b/packages/core/agent-loop/README.md index f4193d44ad..3f4e599da6 100644 --- a/packages/core/agent-loop/README.md +++ b/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. diff --git a/packages/core/agent-loop/src/loop.ts b/packages/core/agent-loop/src/loop.ts index 97a32e2f38..08f72ff72d 100644 --- a/packages/core/agent-loop/src/loop.ts +++ b/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 diff --git a/packages/core/agent-loop/tests/cancel.spec.ts b/packages/core/agent-loop/tests/cancel.spec.ts index f6df171365..eeebf2719f 100644 --- a/packages/core/agent-loop/tests/cancel.spec.ts +++ b/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() + const release = Promise.withResolvers() + 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) diff --git a/packages/examples/acp-demo/README.md b/packages/examples/acp-demo/README.md index 38c2b7852d..ea692fdee7 100644 --- a/packages/examples/acp-demo/README.md +++ b/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 diff --git a/packages/examples/acp-demo/src/index.ts b/packages/examples/acp-demo/src/index.ts index a99c55bec0..923c0acea8 100644 --- 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 @@ -99,18 +101,22 @@ export const Config: z = 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') } diff --git a/packages/session-persistence/session-checkpoint-policy/README.md b/packages/session-persistence/session-checkpoint-policy/README.md index 1f526ce580..dc830d5183 100644 --- a/packages/session-persistence/session-checkpoint-policy/README.md +++ b/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. diff --git a/packages/session-persistence/session-checkpoint-policy/src/index.ts b/packages/session-persistence/session-checkpoint-policy/src/index.ts index 720988186d..2c19b089bf 100644 --- a/packages/session-persistence/session-checkpoint-policy/src/index.ts +++ b/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 => { 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() }) diff --git 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 index faee24fa21..0a507d70a8 100644 --- 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 @@ -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() + 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'))