From b1af35145b980d75239064b1359aa09b1a5fc75c Mon Sep 17 00:00:00 2001 From: _Kerman Date: Tue, 28 Jul 2026 15:33:00 +0800 Subject: [PATCH] fix: enforce message snapshot invariants --- ...send-and-coalesced-user-messages.i18n.yaml | 4 +- ...nified-send-and-coalesced-user-messages.md | 4 +- ...ied-send-and-coalesced-user-messages.zh.md | 4 +- ...nt-loop-observable-state-machine.i18n.yaml | 4 +- ...-24-agent-loop-observable-state-machine.md | 2 +- ...-agent-loop-observable-state-machine.zh.md | 4 +- docs/cordis-catalog/events.md | 30 ++-- docs/cordis-catalog/services.md | 9 +- docs/event-producer-consumer.md | 22 +-- .../cordis/tool-cordis/src/api-catalog.ts | 8 +- packages/core/agent-loop/README.i18n.yaml | 4 +- packages/core/agent-loop/README.md | 2 +- packages/core/agent-loop/README.zh.md | 2 +- packages/core/agent-loop/src/agent.ts | 4 +- packages/core/agent/README.i18n.yaml | 4 +- packages/core/agent/README.md | 2 +- packages/core/agent/README.zh.md | 2 +- packages/core/agent/src/types.ts | 11 +- packages/core/agent/tests/invariant.spec.ts | 4 +- packages/core/scope/tests/invariant.spec.ts | 2 +- packages/core/session/src/index.ts | 89 ++++++++++- packages/core/session/tests/session.spec.ts | 151 ++++++++++++++++++ packages/host/apiproxy/src/api-proxy.ts | 12 +- .../apiproxy/tests/api-proxy-commands.spec.ts | 12 +- .../session-persistence/README.i18n.yaml | 6 +- .../session-persistence/README.md | 4 +- .../session-persistence/README.zh.md | 4 +- .../session-persistence/src/coordinator.ts | 27 +++- .../session-persistence/src/index.ts | 5 +- .../tests/coordinator-contract.ts | 59 +++++++ .../session-query/session-query/src/index.ts | 11 +- .../session-query/src/snapshot.ts | 27 ---- .../session-query/src/tracing.ts | 5 +- packages/ui/tui/tests/tui.spec.ts | 6 +- 34 files changed, 417 insertions(+), 129 deletions(-) delete mode 100644 packages/session-query/session-query/src/snapshot.ts diff --git a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.i18n.yaml b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.i18n.yaml index f0eab75a80..46b84213b9 100644 --- a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.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 .agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md -2026-07-22-unified-send-and-coalesced-user-messages.md: e82a112408d83d7beab75c90d1a5b6474385fa17 -2026-07-22-unified-send-and-coalesced-user-messages.zh.md: bf49e5fc7f28da966ca73537f9989814298dfdc2 +2026-07-22-unified-send-and-coalesced-user-messages.md: ed171735cf483938c70291963a6e68dc02d7bde2 +2026-07-22-unified-send-and-coalesced-user-messages.zh.md: 8b2a3ebabb493954e653255e876255b9c0810c19 diff --git a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md index e82a112408..ed171735cf 100644 --- a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md +++ b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.md @@ -20,9 +20,9 @@ Separately, `context/message` and `user/message` had converged: the surface proj **Goal replay disambiguates by round, not type.** A goal state change is a round-zero goal-sourced `user/message` whose source carries the complete change; a positive round is an admitted continuation prompt. `decodeGoalEvent` takes a `user/message` and fails loud when goal-state content and its typed source disagree. -**`send` returns the message id.** `send` and its aliases return the complete message's existing opaque `MessageId`; creation and freezing are owned by the [identified immutable message decision](2026-07-28-identified-immutable-message-values.md), not by routing. +**`send` does not return identity.** Callers already own the complete message and its opaque `MessageId`; creation and freezing are owned by the [identified immutable message decision](2026-07-28-identified-immutable-message-values.md), not by routing. -**Three inbox events replace agent/queued.** `agent/inbox/enqueue` (an item entered a FIFO), `agent/inbox/dequeue` (the driver claimed one), and `agent/inbox/discard` (`cancel()` dropped pending items) carry the accepted `UserMessage`. Enqueue separately carries the resolved `queued | steering` placement captured by the producer at acceptance time, so observers and reconnect mirrors never reconstruct routing from later status or session history. Injection never touches a FIFO and emits none of these. Every FIFO entry publishes an enqueue, including steering submitted by an `agent/turn-stopping` listener, so the ledger stays balanced with its later dequeue or discard. The `dsh-agent` invariant companion asserts FIFO conservation: a per-agent outstanding count that dequeue and discard can never drive negative. +**Three inbox events replace agent/queued.** `agent/inbox/enqueue` (an item entered a FIFO), `agent/inbox/dequeue` (the driver claimed one), and `agent/inbox/discard` (`cancel()` dropped pending items) carry the accepted `UserMessage`. Enqueue and dequeue also carry the resolved `queued | steering` placement captured at acceptance, so observers and reconnect mirrors retire repeated message identities from the correct FIFO without reconstructing routing from later status or session history. Injection never touches a FIFO and emits none of these. Every FIFO entry publishes an enqueue, including steering submitted by an `agent/turn-stopping` listener, so the ledger stays balanced with its later dequeue or discard. The `dsh-agent` invariant companion asserts FIFO conservation: a per-agent outstanding count that dequeue and discard can never drive negative. **Admission accepts next-step input without becoming a turn.** The loop opens a private next-step acceptance window before `agent/prompt-submit`, keeps it open through the turn, and closes it before `turn/end`. Steering and injection received during admission therefore remain together in the outbox and join an allowed turn. If admission blocks or fails, a context-only caller batch takes idle injection's immediate append, while steering and context staged beside it remain available to retry; neither path writes the rejected prompt. When a later prompt is admitted, retained outbox input enters its turn before that prompt, while input accepted during the current admission remains after the prompt. Closing the window before `turn/end` preserves the rule that reentrant late steering becomes an independent queued turn. `Agent.acceptsNextStep` exposes whether a `next-step` send would currently join this window; `status` remains the broader activity signal rather than a routing predicate. diff --git a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.zh.md b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.zh.md index bf49e5fc7f..8b2a3ebabb 100644 --- a/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.zh.md +++ b/.agents/notes/implemented/architecture/2026-07-22-unified-send-and-coalesced-user-messages.zh.md @@ -20,9 +20,9 @@ agent 的对外驱动接口逐渐长出三个近乎平行的动词——`send` **goal 回放靠轮次而非类型来区分。** 一次 goal 状态变更是一条第 0 轮、来源为 goal 的 `user/message`，其 source 携带完整变更；正数轮次则是一条已准入的继续执行提示词。`decodeGoalEvent` 接收一条 `user/message`，并在 goal 状态内容与其类型化 source 不一致时立即报错。 -**`send` 返回消息 id。** `send` 及其别名返回完整消息已有的不透明 `MessageId`；消息的创建与冻结由[带标识的不可变消息值决策](2026-07-28-identified-immutable-message-values.md)负责，而不是由路由负责。 +**`send` 不返回标识。** 调用方已经持有完整消息及其不透明的 `MessageId`；消息的创建与冻结由[带标识的不可变消息值决策](2026-07-28-identified-immutable-message-values.md)负责，而不是由路由负责。 -**三个 inbox 事件取代 agent/queued。** `agent/inbox/enqueue`（一个队列项进入某个 FIFO）、`agent/inbox/dequeue`（驱动器认领了一个）和 `agent/inbox/discard`（`cancel()` 丢弃了待处理项）都会携带已接受的 `UserMessage`。enqueue 还会单独携带生产方在接受消息时捕获的已解析 `queued | steering` 放置方式，因此观察方和重连镜像永远不必根据后续状态或会话历史重建路由。注入从不触及 FIFO，也不发出这些事件中的任何一个。每一次 FIFO 入队都会发布一个 enqueue 事件，包括 `agent/turn-stopping` 监听器提交的 steering，因此账目会与其后的 dequeue 或 discard 保持平衡。`dsh-agent` 的不变量配套断言 FIFO 守恒：一个按 agent 计的未结算计数，dequeue 和 discard 永远无法把它压到负数。 +**三个 inbox 事件取代 agent/queued。** `agent/inbox/enqueue`（一个队列项进入某个 FIFO）、`agent/inbox/dequeue`（驱动器认领了一个）和 `agent/inbox/discard`（`cancel()` 丢弃了待处理项）都会携带已接受的 `UserMessage`。enqueue 和 dequeue 还会携带生产方在接受消息时捕获的已解析 `queued | steering` 放置方式，因此观察方和重连镜像可以从正确的 FIFO 中结算重复出现的消息标识，无需根据后续状态或会话历史重建路由。注入从不触及 FIFO，也不发出这些事件中的任何一个。每一次 FIFO 入队都会发布一个 enqueue 事件，包括 `agent/turn-stopping` 监听器提交的 steering，因此账目会与其后的 dequeue 或 discard 保持平衡。`dsh-agent` 的不变量配套断言 FIFO 守恒：一个按 agent 计的未结算计数，dequeue 和 discard 永远无法把它压到负数。 **准入接受 next-step 输入，但不会因此成为一个轮次。** 循环会在 `agent/prompt-submit` 前打开一个私有的 next-step 接受窗口，使其贯穿整个轮次，并在 `turn/end` 前关闭。因此，在准入期间收到的 steering 和注入会一起留在 outbox 中并加入获准轮次。如果准入被阻止或失败，仅含调用方上下文的批次会采用空闲注入的立即追加行为，而 steering 及与其一同暂存的上下文仍可重试；两种路径都不会写入被拒绝的提示词。后续提示词获准时，保留在 outbox 中的输入会先于该提示词进入其轮次，而当前准入期间接受的输入则留在提示词之后。在 `turn/end` 前关闭窗口，可以保留这样的规则：可重入的晚到 steering 会成为一个独立的排队轮次。`Agent.acceptsNextStep` 会公开一次 `next-step` 发送当前是否会加入该窗口；`status` 仍是更宽泛的活动信号，而非路由判据。 diff --git a/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.i18n.yaml b/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.i18n.yaml index 29043aa46a..b524e96a02 100644 --- a/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.i18n.yaml +++ b/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.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 .agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.md -2026-07-24-agent-loop-observable-state-machine.md: e0b16f8754241c632d3590d78b8ad64b448826e9 -2026-07-24-agent-loop-observable-state-machine.zh.md: 8e2b247729b3723d6d1fc86ef7e5c6c7c115298b +2026-07-24-agent-loop-observable-state-machine.md: a25657c6a41e2c0989db620046f44ea3254be151 +2026-07-24-agent-loop-observable-state-machine.zh.md: 058d89d3cb9e3d30963f95fda1510ef3c5bf281e diff --git a/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.md b/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.md index e0b16f8754..a25657c6a4 100644 --- a/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.md +++ b/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.md @@ -18,7 +18,7 @@ The public contract exposes four orthogonal state dimensions: - Registration lifetime is the `agent/created` to `agent/disposed` interval. Disposal is the terminal registry edge, not an `AgentStatus`. - Whole-agent activity is `AgentStatus = 'idle' | 'running'`. Consecutive turns may share one `running` interval. -- A FIFO-backed message progresses from `agent/inbox/enqueue` to exactly one `agent/inbox/dequeue` or `agent/inbox/discard`, correlated by its `MessageId`. The inbox events describe acceptance, claim, and removal rather than turn completion. +- A FIFO-backed message progresses from `agent/inbox/enqueue` to exactly one `agent/inbox/dequeue` or `agent/inbox/discard`. Enqueue and dequeue correlate an occurrence by `MessageId` plus its queued-or-steering placement; same-placement repeats retire in FIFO order. The inbox events describe acceptance, claim, and removal rather than turn completion. - A claimed turn passes through prompt admission and zero or more request steps. An automatic retry closes the failed turn and immediately opens another; `agent/settled` reports only the terminal turn in that chain and remains distinct from the whole-agent transition to `status === 'idle'`. The loop keeps five machine extension events. `agent/prompt-submit` admits, rewrites, or blocks a claimed prompt. `agent/step` is the single awaited between-steps checkpoint and runs before every request is derived. `agent/request` is the waterfall for the frozen call configuration; the configuration comes only from `await next()`, not from a duplicate positional argument. `agent/request-error` serializes ownership of awaited model-request recovery. `agent/turn-stopping` runs when the turn otherwise has no work left; a listener that needs another step records real steering with `agent.steer()`, and the loop decides from that data after all listeners settle. diff --git a/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.zh.md b/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.zh.md index 8e2b247729..058d89d3cb 100644 --- a/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.zh.md +++ b/.agents/notes/implemented/simplification/2026-07-24-agent-loop-observable-state-machine.zh.md @@ -18,10 +18,10 @@ agent 生命周期、agent 整体活动状态、收件箱条目的进度以及 - 注册生命周期是从 `agent/created` 到 `agent/disposed` 的区间。dispose（资源释放）是注册表的终止边界，而不是一种 `AgentStatus`。 - agent 整体活动状态为 `AgentStatus = 'idle' | 'running'`。连续多个轮次可以共用同一个 `running` 区间。 -- 由 FIFO 支撑的消息从 `agent/inbox/enqueue` 开始，最终必然进入 `agent/inbox/dequeue` 或 `agent/inbox/discard` 二者之一，并通过其 `MessageId` 关联。收件箱事件描述接受、领取和移除，而不是轮次完成。 +- 由 FIFO 支撑的消息从 `agent/inbox/enqueue` 开始，最终必然进入 `agent/inbox/dequeue` 或 `agent/inbox/discard` 二者之一。enqueue 与 dequeue 通过 `MessageId` 加 queued 或 steering（中途引导）放置方式关联一次消息出现；放置方式相同的重复项按 FIFO 顺序结算。收件箱事件描述接受、领取和移除，而不是轮次完成。 - 已领取的轮次经过提示词准入和零个或多个请求步骤。自动重试会关闭失败轮次并立即开启另一个轮次；`agent/settled` 只报告该重试链的终态轮次，且仍不同于 agent 整体转换到 `status === 'idle'`。 -循环保留五个状态机扩展事件。`agent/prompt-submit` 对已领取的提示词执行准入、改写或阻断。`agent/step` 是步骤之间唯一需要等待的检查点，在每次派生请求前运行。`agent/request` 是冻结调用配置所用的 waterfall；配置只能来自 `await next()`，不再通过重复的位置参数提供。`agent/request-error` 串行确定需要等待的模型请求恢复由谁负责。当轮次原本已经没有剩余工作时，`agent/turn-stopping` 运行；需要再执行一个步骤的监听器使用 `agent.steer()` 记录真实的 steering（中途引导），循环在所有监听器完成后根据这份数据作出决定。 +循环保留五个状态机扩展事件。`agent/prompt-submit` 对已领取的提示词执行准入、改写或阻断。`agent/step` 是步骤之间唯一需要等待的检查点，在每次派生请求前运行。`agent/request` 是冻结调用配置所用的 waterfall；配置只能来自 `await next()`，不再通过重复的位置参数提供。`agent/request-error` 串行确定需要等待的模型请求恢复由谁负责。当轮次原本已经没有剩余工作时，`agent/turn-stopping` 运行；需要再执行一个步骤的监听器使用 `agent.steer()` 记录真实的 steering，循环在所有监听器完成后根据这份数据作出决定。 是否继续和终止执行由数据表达，不再由返回的控制枚举表达。工具调用和已接受的 steering 要求再执行一个步骤。携带 `concludesTurn` 的工具结果会在其所属步骤终止工具循环。循环不再暴露通用的 `ContinuationDecision` 或终止停止返回通道。 diff --git a/docs/cordis-catalog/events.md b/docs/cordis-catalog/events.md index 66b7509bda..faea7d7728 100644 --- a/docs/cordis-catalog/events.md +++ b/docs/cordis-catalog/events.md @@ -32,7 +32,7 @@ Effective broad cancellation was requested, before queued/outbox work is cleared Types: [Agent](../core-data-structures/core.md) · [AgentCancelCause](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:277`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:284`](../../packages/core/agent/src/types.ts) ### `agent/created` — emit @@ -96,7 +96,7 @@ A step or turn errored. The machine reports a failure here (plus the logger) eve Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:391`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:398`](../../packages/core/agent/src/types.ts) ### `agent/inbox/dequeue` — emit @@ -108,16 +108,18 @@ The driver claimed one item out of the inbox: a queued item at a turn boundary, * boundary, or steering drained between steps. Fires after the item leaves * its FIFO and before it becomes a durable message. * @param agent - the agent whose inbox item was claimed. - * @param message - the claimed message (matching the `id` from its `agent/inbox/enqueue`). + * @param message - the claimed message. + * @param placement - the FIFO that claimed this occurrence; together with + * `message.id`, it matches the earliest outstanding enqueue in that FIFO. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * @mode emit */ -'agent/inbox/dequeue'(this: Scoped, agent: Agent, message: UserMessage): void +'agent/inbox/dequeue'( this: Scoped, agent: Agent, message: UserMessage, placement: InboxPlacement, ): void ``` -Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) · [UserMessage](../core-data-structures/session.md) +Types: [Agent](../core-data-structures/core.md) · [InboxPlacement](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) · [UserMessage](../core-data-structures/session.md) -Source: [`packages/core/agent/src/types.ts:255`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:257`](../../packages/core/agent/src/types.ts) ### `agent/inbox/discard` — emit @@ -140,7 +142,7 @@ Pending inbox items were dropped without delivering them, so every enqueue occur Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) · [UserMessage](../core-data-structures/session.md) -Source: [`packages/core/agent/src/types.ts:267`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:274`](../../packages/core/agent/src/types.ts) ### `agent/inbox/enqueue` — emit @@ -185,7 +187,7 @@ Allow, rewrite, or block one claimed prompt before it becomes a user message or Types: [Agent](../core-data-structures/core.md) · [PromptDecision](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) · [UserMessage](../core-data-structures/session.md) -Source: [`packages/core/agent/src/types.ts:304`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:311`](../../packages/core/agent/src/types.ts) ### `agent/request` — waterfall @@ -209,7 +211,7 @@ Replace the frozen call configuration. `await next()` yields the config the mach Types: [Agent](../core-data-structures/core.md) · [LlmCallConfig](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:330`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:337`](../../packages/core/agent/src/types.ts) ### `agent/request-error` — waterfall @@ -239,7 +241,7 @@ Handle a model-request failure after its failed step has closed but before the f Types: [Agent](../core-data-structures/core.md) · [LlmFailure](../core-data-structures/llm-streaming.md) · [RequestError](../core-data-structures/core.md) · [RequestErrorAction](../core-data-structures/core.md) · [ResolvedRetryPolicy](../core-data-structures/llm-streaming.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:349`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:356`](../../packages/core/agent/src/types.ts) ### `agent/session-start` — emit @@ -261,7 +263,7 @@ The session lifecycle began, once before the first turn. Use `agent.inject()` to Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) · [SessionStartSource](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:290`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:297`](../../packages/core/agent/src/types.ts) ### `agent/settled` — emit @@ -286,7 +288,7 @@ One drain chain reached its terminal turn: that turn's `turn/end` is already com Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) · [SettleReason](../core-data-structures/core.md) -Source: [`packages/core/agent/src/types.ts:378`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:385`](../../packages/core/agent/src/types.ts) ### `agent/status` — emit @@ -330,7 +332,7 @@ Awaited serial checkpoint before EVERY request of a turn is built (the first as Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:317`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:324`](../../packages/core/agent/src/types.ts) ### `agent/turn-stopping` — serial @@ -356,7 +358,7 @@ The turn is about to close: the model owes no response (no live tool calls, no f Types: [Agent](../core-data-structures/core.md) · [Scoped](../core-data-structures/scope.md) -Source: [`packages/core/agent/src/types.ts:364`](../../packages/core/agent/src/types.ts) +Source: [`packages/core/agent/src/types.ts:371`](../../packages/core/agent/src/types.ts) ## `agent-loop/*` diff --git a/docs/cordis-catalog/services.md b/docs/cordis-catalog/services.md index c398f72b1e..9c11f1a1e0 100644 --- a/docs/cordis-catalog/services.md +++ b/docs/cordis-catalog/services.md @@ -1029,6 +1029,8 @@ abstract append(id: SessionId, events: readonly SessionEvent[]): Promise * open live turn rejects. * A coordinator-backed cold load reserves the identity across storage awaits, * so concurrent publication of a same-id live Session rejects. + * Returned events are detached, and every identified message is deeply + * frozen; malformed identified messages reject before any stored event is returned. * @param id - the persisted session to reload. * @returns the header and a log ending on a balanced `turn/end`. */ @@ -1038,7 +1040,8 @@ abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEven * Inspect a header and its valid contiguous stored prefix without repairing * a torn tail, closing an interrupted turn, or publishing coordinator state. * This read is serialized with writes for the same id and returns detached - * values, so observers cannot mutate backend-owned state. + * values with deeply frozen identified messages, so observers cannot mutate message + * identity/content or backend-owned state. Malformed identified messages reject. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. * @returns the header and valid stored event prefix exactly as observed. @@ -1194,7 +1197,7 @@ async readEvent(request: SessionEventReadRequest, signal?: AbortSignal): Promise Types: [SessionEventReadRequest](../core-data-structures/session-query.md) · [SessionEventRecord](../core-data-structures/session-query.md) · [SessionEventResultFilter](../core-data-structures/session-query.md) · [SessionEventSearchDocument](../core-data-structures/session-query.md) · [SessionEventSearchPage](../core-data-structures/session-query.md) · [SessionEventSearchRequest](../core-data-structures/session-query.md) · [SessionEventTraceObservation](../core-data-structures/session-query.md) · [SessionEventTraceRequest](../core-data-structures/session-query.md) · [SessionEventWindow](../core-data-structures/session-query.md) · [SessionId](../core-data-structures/core.md) · [SessionLineageTrace](../core-data-structures/session-query.md) · [SessionLogSnapshot](../core-data-structures/session-query.md) · [SessionRecord](../core-data-structures/session-query.md) · [SessionResultFilter](../core-data-structures/session-query.md) · [SessionSearchExecContext](../core-data-structures/session-query.md) · [SessionSearchHit](../core-data-structures/session-query.md) · [SessionSearchPage](../core-data-structures/session-query.md) · [SessionSearchRequest](../core-data-structures/session-query.md) · [SessionSurfaceSnapshot](../core-data-structures/session-query.md) · [SessionTitleObservation](../core-data-structures/session-query.md) · [SessionTitleObservationResult](../core-data-structures/session-query.md) · [SessionTitleSnapshot](../core-data-structures/session-title.md) -Source: [`packages/session-query/session-query/src/index.ts:82`](../../packages/session-query/session-query/src/index.ts) +Source: [`packages/session-query/session-query/src/index.ts:81`](../../packages/session-query/session-query/src/index.ts) ## `ctx.sessionReferences` — `SessionReferenceService` @@ -1373,7 +1376,7 @@ fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId): Types: [CreateSessionOptions](../core-data-structures/persistence.md) · [OutOfBandSessionEventType](../core-data-structures/session.md) · [Session](../core-data-structures/session.md) · [SessionEvent](../core-data-structures/core.md) · [SessionEventMap](../core-data-structures/session.md) · [SessionId](../core-data-structures/core.md) · [TurnTrigger](../core-data-structures/session.md) -Source: [`packages/core/session/src/index.ts:618`](../../packages/core/session/src/index.ts) +Source: [`packages/core/session/src/index.ts:695`](../../packages/core/session/src/index.ts) ## `ctx.sessionTitle` — `SessionTitleService` diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index 7e638abfc2..b93ed26ec5 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -8,21 +8,21 @@ This matrix shows which packages dispatch each harness-owned event and which pac | Event | Mode | Declared in | Dispatchers | Listeners | | --- | --- | --- | --- | --- | | `agent-loop/config-start-failed` | `emit` | [`packages/core/agent-loop/src/index.ts:140`](../packages/core/agent-loop/src/index.ts) | [`agent-loop`](../packages/core/agent-loop) (`events.dispatch`) | [`tui`](../packages/ui/tui) | -| `agent/cancel-requested` | `emit` | [`packages/core/agent/src/types.ts:277`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`goal-session`](../packages/goal/goal-session) | +| `agent/cancel-requested` | `emit` | [`packages/core/agent/src/types.ts:284`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`goal-session`](../packages/goal/goal-session) | | `agent/created` | `emit` | [`packages/core/agent/src/types.ts:216`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) | | `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:225`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`agent-loop`](../packages/core/agent-loop), [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) | -| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:391`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `apiproxy`, [`goal-session`](../packages/goal/goal-session), [`session-telemetry`](../packages/telemetry/session-telemetry), [`tui`](../packages/ui/tui) | -| `agent/inbox/dequeue` | `emit` | [`packages/core/agent/src/types.ts:255`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`tui`](../packages/ui/tui) | -| `agent/inbox/discard` | `emit` | [`packages/core/agent/src/types.ts:267`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`tui`](../packages/ui/tui) | +| `agent/error` | `emit` | [`packages/core/agent/src/types.ts:398`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | `apiproxy`, [`goal-session`](../packages/goal/goal-session), [`session-telemetry`](../packages/telemetry/session-telemetry), [`tui`](../packages/ui/tui) | +| `agent/inbox/dequeue` | `emit` | [`packages/core/agent/src/types.ts:257`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`tui`](../packages/ui/tui) | +| `agent/inbox/discard` | `emit` | [`packages/core/agent/src/types.ts:274`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`tui`](../packages/ui/tui) | | `agent/inbox/enqueue` | `emit` | [`packages/core/agent/src/types.ts:245`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`goal-session`](../packages/goal/goal-session) | -| `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:304`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`goal-session`](../packages/goal/goal-session), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`repeat-tool-guard`](../packages/guard/repeat-tool-guard), [`tui`](../packages/ui/tui) | -| `agent/request` | `waterfall` | [`packages/core/agent/src/types.ts:330`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`agent`](../packages/core/agent) | -| `agent/request-error` | `waterfall` | [`packages/core/agent/src/types.ts:349`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compact-basic`](../packages/compact/compact-basic), [`llm-retry`](../packages/llm/llm-retry) | -| `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:290`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`goal`](../packages/goal/goal), [`goal-session`](../packages/goal/goal-session), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`workspace-context`](../packages/context/workspace-context) | -| `agent/settled` | `emit` | [`packages/core/agent/src/types.ts:378`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`compact-basic`](../packages/compact/compact-basic) | +| `agent/prompt-submit` | `waterfall` | [`packages/core/agent/src/types.ts:311`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`goal-session`](../packages/goal/goal-session), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`repeat-tool-guard`](../packages/guard/repeat-tool-guard), [`tui`](../packages/ui/tui) | +| `agent/request` | `waterfall` | [`packages/core/agent/src/types.ts:337`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`agent`](../packages/core/agent) | +| `agent/request-error` | `waterfall` | [`packages/core/agent/src/types.ts:356`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`waterfall`) | [`compact-basic`](../packages/compact/compact-basic), [`llm-retry`](../packages/llm/llm-retry) | +| `agent/session-start` | `emit` | [`packages/core/agent/src/types.ts:297`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`goal`](../packages/goal/goal), [`goal-session`](../packages/goal/goal-session), [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex), [`workspace-context`](../packages/context/workspace-context) | +| `agent/settled` | `emit` | [`packages/core/agent/src/types.ts:385`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`compact-basic`](../packages/compact/compact-basic) | | `agent/status` | `emit` | [`packages/core/agent/src/types.ts:234`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emitAgentEvent`) | [`agent`](../packages/core/agent), `apiproxy`, [`goal-session`](../packages/goal/goal-session), [`tui`](../packages/ui/tui) | -| `agent/step` | `serial` | [`packages/core/agent/src/types.ts:317`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`compact-basic`](../packages/compact/compact-basic), [`plan-mode`](../packages/plan/plan-mode), [`session-checkpoint-policy`](../packages/session-persistence/session-checkpoint-policy), [`time-context`](../packages/context/time-context), [`tool-skill`](../packages/skill/tool-skill), [`user-approval`](../packages/ui/user-approval), [`workspace-context`](../packages/context/workspace-context) | -| `agent/turn-stopping` | `serial` | [`packages/core/agent/src/types.ts:364`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | +| `agent/step` | `serial` | [`packages/core/agent/src/types.ts:324`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`compact-basic`](../packages/compact/compact-basic), [`plan-mode`](../packages/plan/plan-mode), [`session-checkpoint-policy`](../packages/session-persistence/session-checkpoint-policy), [`time-context`](../packages/context/time-context), [`tool-skill`](../packages/skill/tool-skill), [`user-approval`](../packages/ui/user-approval), [`workspace-context`](../packages/context/workspace-context) | +| `agent/turn-stopping` | `serial` | [`packages/core/agent/src/types.ts:371`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`hooks-codex`](../packages/hooks/hooks-codex) | | `approval/request` | `waterfall` | [`packages/ui/user-approval/src/index.ts:30`](../packages/ui/user-approval/src/index.ts) | [`user-approval`](../packages/ui/user-approval) (`waterfall`) | [`acp`](../packages/acp/acp) | | `commands/change` | `emit` | [`packages/ui/commands/src/index.ts:103`](../packages/ui/commands/src/index.ts) | [`commands`](../packages/ui/commands) (`events.dispatch`) | `apiproxy`, [`tui`](../packages/ui/tui) | | `domain/changed` | `emit` | [`packages/storage/storage-domain/src/events.ts:46`](../packages/storage/storage-domain/src/events.ts) | [`storage-domain`](../packages/storage/storage-domain) (`emit`) | `apiproxy`, [`storage-domain`](../packages/storage/storage-domain), [`workspace`](../packages/workspace/workspace) | diff --git a/packages/cordis/tool-cordis/src/api-catalog.ts b/packages/cordis/tool-cordis/src/api-catalog.ts index 9d261107fd..dd7e95403b 100644 --- a/packages/cordis/tool-cordis/src/api-catalog.ts +++ b/packages/cordis/tool-cordis/src/api-catalog.ts @@ -518,11 +518,11 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [ }, { signature: 'abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', - jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return with its stored header as a durable snapshot, while an\n * open live turn rejects.\n * A coordinator-backed cold load reserves the identity across storage awaits,\n * so concurrent publication of a same-id live Session rejects.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', + jsDoc: '/**\n * Load a header and balanced contiguous log. A complete interrupted final\n * turn is preserved and durably closed with missing tool errors plus any open\n * step and turn boundaries; only a torn final record is discarded. Unknown\n * versions and corruption in the committed prefix reject. Implementations\n * MUST NOT crash-repair an identity still bound to a live Session: a balanced\n * live log may return with its stored header as a durable snapshot, while an\n * open live turn rejects.\n * A coordinator-backed cold load reserves the identity across storage awaits,\n * so concurrent publication of a same-id live Session rejects.\n * Returned events are detached, and every identified message is deeply\n * frozen; malformed identified messages reject before any stored event is returned.\n * @param id - the persisted session to reload.\n * @returns the header and a log ending on a balanced `turn/end`.\n */', }, { signature: 'abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>', - jsDoc: '/**\n * Inspect a header and its valid contiguous stored prefix without repairing\n * a torn tail, closing an interrupted turn, or publishing coordinator state.\n * This read is serialized with writes for the same id and returns detached\n * values, so observers cannot mutate backend-owned state.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the header and valid stored event prefix exactly as observed.\n */', + jsDoc: '/**\n * Inspect a header and its valid contiguous stored prefix without repairing\n * a torn tail, closing an interrupted turn, or publishing coordinator state.\n * This read is serialized with writes for the same id and returns detached\n * values with deeply frozen identified messages, so observers cannot mutate message\n * identity/content or backend-owned state. Malformed identified messages reject.\n * @param id - the persisted session to inspect.\n * @param signal - optional cancellation for queued and backend read work.\n * @returns the header and valid stored event prefix exactly as observed.\n */', }, { signature: 'abstract list(signal?: AbortSignal): Promise', @@ -1034,8 +1034,8 @@ export const EVENT_API: readonly EventApiEntry[] = [ { name: 'agent/inbox/dequeue', mode: 'emit', - signature: '\'agent/inbox/dequeue\'(this: Scoped, agent: Agent, message: UserMessage): void', - jsDoc: '/**\n * The driver claimed one item out of the inbox: a queued item at a turn\n * boundary, or steering drained between steps. Fires after the item leaves\n * its FIFO and before it becomes a durable message.\n * @param agent - the agent whose inbox item was claimed.\n * @param message - the claimed message (matching the `id` from its `agent/inbox/enqueue`).\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode emit\n */', + signature: '\'agent/inbox/dequeue\'( this: Scoped, agent: Agent, message: UserMessage, placement: InboxPlacement, ): void', + jsDoc: '/**\n * The driver claimed one item out of the inbox: a queued item at a turn\n * boundary, or steering drained between steps. Fires after the item leaves\n * its FIFO and before it becomes a durable message.\n * @param agent - the agent whose inbox item was claimed.\n * @param message - the claimed message.\n * @param placement - the FIFO that claimed this occurrence; together with\n * `message.id`, it matches the earliest outstanding enqueue in that FIFO.\n * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent.\n * @mode emit\n */', summary: 'The driver claimed one item out of the inbox: a queued item at a turn boundary, or steering drained between steps.', }, { diff --git a/packages/core/agent-loop/README.i18n.yaml b/packages/core/agent-loop/README.i18n.yaml index 8652771f93..f0e492c2aa 100644 --- a/packages/core/agent-loop/README.i18n.yaml +++ b/packages/core/agent-loop/README.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 packages/core/agent-loop/README.md -README.md: c12140f27aed400b0f7b4246700473e877d37632 -README.zh.md: 6394cd86f5f3241be07ef711c76079624bce1bfe +README.md: 99fdc73e2dcb09ed7d47f021598dea8720cada5d +README.zh.md: fe9db9a625ea289e7e9364a182061dee1a0e773c diff --git a/packages/core/agent-loop/README.md b/packages/core/agent-loop/README.md index c12140f27a..99fdc73e2d 100644 --- a/packages/core/agent-loop/README.md +++ b/packages/core/agent-loop/README.md @@ -54,7 +54,7 @@ Configured agents start automatically. A model call requires both `provider` and The concrete `ReactLoopAgent`, its queued input, outbox, and run controls are package-internal. The package root exports only the plugin/service/config contract, and the package exports map exposes no `./src/*` escape hatch; lifecycle owners create agents through `ctx.agents` rather than naming, constructing, or starting driver internals. One prepared session can be claimed by only one concrete driver, and everything observable happens through session events and the `agent/*` event taxonomy. -The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model. Every inbox enqueue publishes `agent/inbox/enqueue` with the resolved queued-or-steering placement; taking it publishes `agent/inbox/dequeue`, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`. +The unified `send()` primitive routes content and source by (`target` × `wakeup`); `followup`/`steer`/`inject` are its fixed-preset aliases. A `next-turn` item joins the queued FIFO, waking the driver unless `wakeup: false`; admission happens before any turn opens. The loop opens a private next-step acceptance window before `agent/prompt-submit` and closes it before `turn/end`. During that window, `steer()` and `inject()` stage in one outbox; an allowed admission opens the turn, records the prompt and returned `additionalContexts`, then drains the staged input before the first request. A blocked or failed admission writes no prompt or hook-produced context. A caller-staged context-only batch then takes idle injection's immediate append, while steering and context staged beside it remain pending for retry or a later admitted prompt. Outside the window, steering becomes a waking queued prompt and injection immediately appends `user/message` without opening a turn or running the model. Every inbox enqueue publishes `agent/inbox/enqueue` with the resolved queued-or-steering placement; taking it publishes `agent/inbox/dequeue` with the same placement, and `cancel()` without `keepInbox` publishes `agent/inbox/discard`. ### Loop lifecycle (`agent.ts`) diff --git a/packages/core/agent-loop/README.zh.md b/packages/core/agent-loop/README.zh.md index 6394cd86f5..fe9db9a625 100644 --- a/packages/core/agent-loop/README.zh.md +++ b/packages/core/agent-loop/README.zh.md @@ -54,7 +54,7 @@ interface Config { 实体 `ReactLoopAgent`、其排队输入、outbox 与运行控制均为包内部实现。包根只导出插件／服务／配置契约，包导出映射不提供 `./src/*` 逃逸路径；生命周期拥有方通过 `ctx.agents` 创建 agent，而不是点名、构造或启动驱动器内部组件。一个准备完成的会话只能由一个实体驱动器认领；所有可观测行为都通过会话事件和 `agent/*` 事件分类体系发生。 -统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。每次 inbox 入队都会发布 `agent/inbox/enqueue`，并携带解析出的 queued 或 steering 路由归类；取走它会发布 `agent/inbox/dequeue`；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。 +统一的 `send()` 原语按（`target` × `wakeup`）路由内容与来源；`followup`/`steer`/`inject` 是它的固定预设别名。`next-turn` 项加入排队 FIFO，除非 `wakeup: false`，否则会唤醒驱动器；接纳发生在任何轮次开启之前。循环在 `agent/prompt-submit` 之前打开一个私有的 next-step 接收窗口，并在 `turn/end` 之前关闭它。在该窗口内，`steer()` 与 `inject()` 会暂存到同一个 outbox；接纳获准后会开启轮次，记录提示词及其返回的 `additionalContexts`，再于首次请求前排空暂存输入。接纳被阻止或失败时，不会写入提示词或钩子生成的上下文。之后，仅含调用方暂存上下文的批次会采用空闲注入的立即追加行为，而 steering（中途引导）及与其一同暂存的上下文则继续待处理，以供重试或之后获准的提示词使用。窗口之外，steering 会成为唤醒驱动器的排队提示词，而注入会立即追加 `user/message`，不开启轮次也不运行模型。每次 inbox 入队都会发布 `agent/inbox/enqueue`，并携带解析出的 queued 或 steering 路由归类；取走它会发布 `agent/inbox/dequeue`，并携带相同的路由归类；`cancel()` 在不带 `keepInbox` 时会发布 `agent/inbox/discard`。 ### 循环生命周期（`agent.ts`） diff --git a/packages/core/agent-loop/src/agent.ts b/packages/core/agent-loop/src/agent.ts index f616e77579..21759b7480 100644 --- a/packages/core/agent-loop/src/agent.ts +++ b/packages/core/agent-loop/src/agent.ts @@ -294,7 +294,7 @@ export class ReactLoopAgent implements Agent { // Published only after the abort owner and pending done are installed: a // dequeue listener that cancels or disposes must find live cancellation // and quiescence ownership, not the previous activity's settled state. - emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message) + emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', message, 'queued') } /** @@ -639,7 +639,7 @@ export class ReactLoopAgent implements Agent { for (const item of this.outbox.splice(0, limit)) { if (item.steering) { steered = true - emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.message) + emitAgentEvent(this.loopCtx, this, 'agent/inbox/dequeue', item.message, 'steering') this.session.append( 'steering/message', { turn, message: item.message }, diff --git a/packages/core/agent/README.i18n.yaml b/packages/core/agent/README.i18n.yaml index 062555cad5..f55931e999 100644 --- a/packages/core/agent/README.i18n.yaml +++ b/packages/core/agent/README.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 packages/core/agent/README.md -README.md: 44b2f81f7630834f8992f30e707956d86beac20c -README.zh.md: bec0524ebc575d382c1a70871b4cb252830499b5 +README.md: 9fe207d940ddc80153131c3d76c992804bb83773 +README.zh.md: c7975fe5e0368aa46253c47c15f04e12a24a3b7b diff --git a/packages/core/agent/README.md b/packages/core/agent/README.md index 44b2f81f76..9fe207d940 100644 --- a/packages/core/agent/README.md +++ b/packages/core/agent/README.md @@ -58,7 +58,7 @@ Turn and step boundaries and the model token stream are durable `session/event` The handle every plugin programs against: -- `agent.send(message, options)` — the one delivery primitive over the (`target` × `wakeup`) matrix. `message` is an already identified `UserMessage`; callers normally create it with `createUserMessage()` before routing begins. `SendOptions` owns only the `target` and `wakeup` policy. The agent detaches and freezes the complete value without minting or replacing its identity. The message's `agent/inbox/enqueue`/`dequeue`/`discard` events carry it so callers can correlate a queued item with its lifecycle; enqueue also carries the resolved `queued | steering` placement so listeners never reconstruct acceptance-time routing from later state. `target: 'next-turn'` queues one independent FIFO item that, if admitted, becomes the sole ordinary prompt in its turn. `target: 'next-step'` with `wakeup: true` submits steering, while `target: 'next-step'` with `wakeup: false` injects durable context without running the model. The [one-send-one-turn Agent Note](../../../.agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.md) owns the turn rationale. +- `agent.send(message, options)` — the one delivery primitive over the (`target` × `wakeup`) matrix. `message` is an already identified `UserMessage`; callers normally create it with `createUserMessage()` before routing begins. `SendOptions` owns only the `target` and `wakeup` policy. The agent detaches and freezes the complete value without minting or replacing its identity. The message's `agent/inbox/enqueue`/`dequeue`/`discard` events carry it so callers can correlate a queued item with its lifecycle; enqueue and dequeue also carry the resolved `queued | steering` placement so repeated message identities retire from the correct FIFO. `target: 'next-turn'` queues one independent FIFO item that, if admitted, becomes the sole ordinary prompt in its turn. `target: 'next-step'` with `wakeup: true` submits steering, while `target: 'next-step'` with `wakeup: false` injects durable context without running the model. The [one-send-one-turn Agent Note](../../../.agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.md) owns the turn rationale. - `agent.followup(input)` — the `next-turn`/wakeup preset of `send()`: queue an ordinary follow-up turn and wake the driver. - `agent.steer(input)` — the `next-step`/wakeup preset: during prompt admission or an open turn, stage steering for the next safe boundary without dispatching `agent/prompt-submit`; outside that acceptance window, delegate to a woken follow-up. Admission failure leaves staged steering for retry or a later admitted prompt, while cancellation or disposal may discard it. - `agent.inject(input)` — the `next-step`/no-wakeup preset: append model-facing context without running the model; the next request sees a verbatim user-role message whose provenance is carried by the required `input.source`. During prompt admission or an open turn, injection waits in the outbox for the next safe boundary. Outside that acceptance window, it appends immediately without opening a turn; a context-only admission batch takes this fallback if admission closes without a turn, while context staged beside steering remains pending with it. Persistence reacts to `session/event` independently. Injection emits no `agent/inbox/*` event. diff --git a/packages/core/agent/README.zh.md b/packages/core/agent/README.zh.md index bec0524ebc..c7975fe5e0 100644 --- a/packages/core/agent/README.zh.md +++ b/packages/core/agent/README.zh.md @@ -58,7 +58,7 @@ Agent *创建* 由实现 `AgentFactory` 的插件（`dsh-agent-loop`）提供， 每个插件面向的 handle： -- `agent.send(message, options)`：覆盖（`target` × `wakeup`）矩阵的唯一投递原语。`message` 是已有标识的 `UserMessage`；调用方通常会在开始路由前使用 `createUserMessage()` 创建它。`SendOptions` 只持有 `target` 与 `wakeup` 策略。agent 会将完整值与输入分离并冻结，但不会生成或替换其标识。该消息的 `agent/inbox/enqueue`/`dequeue`/`discard` 事件会携带其 id，调用方可据此把排队项与其生命周期关联；入队事件还会携带解析出的 `queued | steering` 路由归类，使监听器无需从后续状态重建接收时的路由。`target: 'next-turn'` 排队一条独立 FIFO 项，获准后成为其轮次中唯一的普通提示词。`target: 'next-step'` 且 `wakeup: true` 提交 steering（中途引导），而 `target: 'next-step'` 且 `wakeup: false` 注入持久上下文，不运行模型。轮次原理由 [one-send-one-turn Agent Note](../../../.agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.md)拥有。 +- `agent.send(message, options)`：覆盖（`target` × `wakeup`）矩阵的唯一投递原语。`message` 是已有标识的 `UserMessage`；调用方通常会在开始路由前使用 `createUserMessage()` 创建它。`SendOptions` 只持有 `target` 与 `wakeup` 策略。agent 会将完整值与输入分离并冻结，但不会生成或替换其标识。该消息的 `agent/inbox/enqueue`/`dequeue`/`discard` 事件会携带其 id，调用方可据此把排队项与其生命周期关联；入队与出队事件还会携带解析出的 `queued | steering` 路由归类，使重复出现的消息标识能在正确的 FIFO 中完成结算。`target: 'next-turn'` 排队一条独立 FIFO 项，获准后成为其轮次中唯一的普通提示词。`target: 'next-step'` 且 `wakeup: true` 提交 steering（中途引导），而 `target: 'next-step'` 且 `wakeup: false` 注入持久上下文，不运行模型。轮次原理由 [one-send-one-turn Agent Note](../../../.agents/notes/implemented/simplification/2026-07-17-one-send-one-turn.md)拥有。 - `agent.followup(input)`：`send()` 的 `next-turn`／wakeup 预设：排队一个普通后续轮次并唤醒驱动器。 - `agent.steer(input)`：`next-step`／wakeup 预设：提示词接纳期间或轮次打开时，为下一个安全边界暂存 steering，且不分发 `agent/prompt-submit`；该接收窗口之外则委托给会唤醒的后续轮次。接纳失败会保留暂存的 steering，以供重试或之后获准的提示词使用，而取消或 dispose 可能丢弃它。 - `agent.inject(input)`：`next-step`／不唤醒预设：追加面向模型的上下文而不运行模型；下一次请求会看到一条逐字的 user role 消息，其来源由必填的 `input.source` 携带。提示词接纳期间或轮次打开时，注入会在 outbox 中等待下一个安全边界。该接收窗口之外，它会立即追加而不开启轮次；如果接纳结束却未开启轮次，仅含上下文的接纳批次会采用这一回退，而与 steering 一同暂存的上下文则会随其继续待处理。持久化独立地响应 `session/event`。注入不发出 `agent/inbox/*` 事件。 diff --git a/packages/core/agent/src/types.ts b/packages/core/agent/src/types.ts index 46ecc7a17f..168be444db 100644 --- a/packages/core/agent/src/types.ts +++ b/packages/core/agent/src/types.ts @@ -248,11 +248,18 @@ declare module 'cordis' { * boundary, or steering drained between steps. Fires after the item leaves * its FIFO and before it becomes a durable message. * @param agent - the agent whose inbox item was claimed. - * @param message - the claimed message (matching the `id` from its `agent/inbox/enqueue`). + * @param message - the claimed message. + * @param placement - the FIFO that claimed this occurrence; together with + * `message.id`, it matches the earliest outstanding enqueue in that FIFO. * Scope-filtered dispatch (`@deepseek-ai/dsh-scope`): agent-scoped listeners receive only that agent. * @mode emit */ - 'agent/inbox/dequeue'(this: Scoped, agent: Agent, message: UserMessage): void + 'agent/inbox/dequeue'( + this: Scoped, + agent: Agent, + message: UserMessage, + placement: InboxPlacement, + ): void /** * Pending inbox items were dropped without delivering them, so every * enqueue occurrence receives exactly one terminal `agent/inbox/dequeue` OR diff --git a/packages/core/agent/tests/invariant.spec.ts b/packages/core/agent/tests/invariant.spec.ts index 96f6ddfd22..a55bdba34f 100644 --- a/packages/core/agent/tests/invariant.spec.ts +++ b/packages/core/agent/tests/invariant.spec.ts @@ -60,7 +60,7 @@ describe('agent inbox invariants', () => { expect(() => { ctx.emit(at, 'agent/inbox/enqueue', agent, info(), 'queued') ctx.emit(at, 'agent/inbox/enqueue', agent, info(), 'steering') - ctx.emit(at, 'agent/inbox/dequeue', agent, info()) + ctx.emit(at, 'agent/inbox/dequeue', agent, info(), 'queued') ctx.emit(at, 'agent/inbox/discard', agent, [info()]) }).not.toThrow() }) @@ -68,7 +68,7 @@ describe('agent inbox invariants', () => { it('rejects a dequeue with no outstanding item', async () => { const ctx = await setup() const agent = mockAgent('i2') - expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/inbox/dequeue', agent, info()) }) + expect(() => { ctx.emit(scopeTarget(agent, agent), 'agent/inbox/dequeue', agent, info(), 'queued') }) .toThrow(/without a matching prior enqueue/) }) diff --git a/packages/core/scope/tests/invariant.spec.ts b/packages/core/scope/tests/invariant.spec.ts index bc0d9f819d..54f2e1e17b 100644 --- a/packages/core/scope/tests/invariant.spec.ts +++ b/packages/core/scope/tests/invariant.spec.ts @@ -49,7 +49,7 @@ describe('scoped-dispatch invariants', () => { 'agent/disposed': [agent], 'agent/status': [agent, 'idle'], 'agent/inbox/enqueue': [agent, message, 'queued'], - 'agent/inbox/dequeue': [agent, message], + 'agent/inbox/dequeue': [agent, message, 'queued'], 'agent/inbox/discard': [agent, []], 'agent/cancel-requested': [agent, { kind: 'user' }], 'agent/session-start': [agent, 'startup'], diff --git a/packages/core/session/src/index.ts b/packages/core/session/src/index.ts index 0bbaf86728..9183b30b99 100644 --- a/packages/core/session/src/index.ts +++ b/packages/core/session/src/index.ts @@ -146,6 +146,33 @@ function snapshotSessionHeader(id: SessionId, source?: SessionHeader): SessionHe return deepFreeze(record as unknown as SessionHeader) } +/** + * Detach one event while preserving deep immutability for its identified message. + * @param event - event imported across a query or persistence boundary. + * @returns a detached event snapshot with a validated, deeply frozen message. + */ +export function snapshotSessionEvent(event: T): T { + const snapshot = structuredClone(event) + assertMessageEventShape( + snapshot, + `session event at seq ${snapshot.seq}`, + ) + switch (snapshot.type) { + case 'user/message': + deepFreeze(snapshot.data) + break + case 'assistant/message': + case 'tool/result': + case 'steering/message': + deepFreeze(snapshot.data.message) + break + default: + // SessionEventMap is merge-extensible; plugin-owned events carry no core message. + break + } + return snapshot +} + /** Validate the fixed event envelope after one-pass JSON materialization. */ function assertSessionEventEnvelope(value: Record, index: number): asserts value is SessionEvent { const event = value @@ -166,13 +193,14 @@ function assertSessionEventEnvelope(value: Record, index: numbe assertCurrentTurnEndShape(event, index) } -/** Reject obsolete request headers and pre-unification message shapes at the seed/load boundary. */ +/** Reject obsolete request headers and malformed messages at the seed/load boundary. */ function assertCurrentLlmShape(event: Record, index: number): void { const data = event['data'] - if (typeof data !== 'object' || data === null) return - const record = data as Record + const record = typeof data === 'object' && data !== null + ? data as Record + : undefined if (event['type'] === 'request/header') { - const header = record['header'] + const header = record?.['header'] const config = typeof header === 'object' && header !== null ? (header as Record)['config'] : undefined if (!hasProviderModel(config)) throw new Error(`seed request/header at index ${index} lacks provider/model`) const reasoningEffort = (config as Record)['reasoningEffort'] @@ -184,11 +212,60 @@ function assertCurrentLlmShape(event: Record, index: number): v const type = event['type'] if (type !== 'user/message' && type !== 'assistant/message' && type !== 'tool/result' && type !== 'steering/message') return - const message = type === 'user/message' ? record : record['message'] + assertMessageEventShape(event, `seed ${type} at index ${index}`) +} + +/** Validate only the event-specific invariants needed to safely replay a message. */ +function assertMessageEventShape(event: Record, subject: string): void { + const type = event['type'] + if (type !== 'user/message' && type !== 'assistant/message' + && type !== 'tool/result' && type !== 'steering/message') return + const data = event['data'] + const record = typeof data === 'object' && data !== null + ? data as Record + : undefined + const message = type === 'user/message' ? record : record?.['message'] if (typeof message !== 'object' || message === null || typeof (message as Record)['id'] !== 'string' || (message as Record)['id'] === '') { - throw new Error(`seed ${type} at index ${index} lacks an identified message`) + throw new Error(`${subject} lacks an identified message`) + } + const messageRecord = message as Record + const expectedRole = type === 'assistant/message' ? 'assistant' : 'user' + if (messageRecord['role'] !== expectedRole) { + throw new Error(`${subject} message must have role "${expectedRole}"`) + } + const source = messageRecord['source'] + if (typeof source !== 'object' || source === null + || typeof (source as Record)['kind'] !== 'string' + || (source as Record)['kind'] === '') { + throw new Error(`${subject} message has invalid source`) + } + if (!Array.isArray(messageRecord['content'])) { + throw new Error(`${subject} message has invalid content`) + } + const sourceRecord = source as Record + if (type === 'assistant/message') { + if (sourceRecord['kind'] !== 'model' || !hasProviderModel(sourceRecord)) { + throw new Error(`${subject} message must have model source`) + } + return + } + if (type !== 'tool/result') return + if (sourceRecord['kind'] !== 'tool' + || typeof sourceRecord['callId'] !== 'string' + || sourceRecord['callId'] === '') { + throw new Error(`${subject} message must have tool source`) + } + const content = messageRecord['content'] as unknown[] + const block = content[0] + if (content.length !== 1 || typeof block !== 'object' || block === null + || (block as Record)['type'] !== 'tool-result' + || !Array.isArray((block as Record)['content'])) { + throw new Error(`${subject} message must contain one tool-result block`) + } + if ((block as Record)['toolCallId'] !== sourceRecord['callId']) { + throw new Error(`${subject} message has mismatched tool call ids`) } } diff --git a/packages/core/session/tests/session.spec.ts b/packages/core/session/tests/session.spec.ts index 27571770a3..7152dd5d41 100644 --- a/packages/core/session/tests/session.spec.ts +++ b/packages/core/session/tests/session.spec.ts @@ -7,6 +7,7 @@ import SessionStore, { Session, SessionEvent, SessionId, + snapshotSessionEvent, } from '@deepseek-ai/dsh-session' import type { CreateSessionOptions, SessionEventType, SessionHeader, SessionSurface, TodoItem } from '@deepseek-ai/dsh-session' @@ -220,6 +221,156 @@ describe('Session', () => { .toEqual([unrelatedPrimitiveData]) }) + it('rejects event-specific malformed message shapes on seed/load', () => { + const user = { + id: 'user', + role: 'user', + content: [{ type: 'text', text: 'content' }], + source: { kind: 'user' }, + } + const assistant = { + id: 'assistant', + role: 'assistant', + content: [{ type: 'text', text: 'content' }], + source: { kind: 'model', provider: 'mock', model: 'mock' }, + } + const tool = { + id: 'tool', + role: 'user', + content: [{ + type: 'tool-result', + toolCallId: 'call', + content: [{ type: 'text', text: 'result' }], + }], + source: { kind: 'tool', callId: 'call' }, + } + const invalid = [ + { + name: 'message record', + event: { + type: 'user/message', seq: 0, time: 1, surfaceOp: 'append', + data: null, + }, + message: 'lacks an identified message', + }, + { + name: 'user role', + event: { + type: 'user/message', seq: 0, time: 1, surfaceOp: 'append', + data: { ...user, role: 'assistant' }, + }, + message: 'message must have role "user"', + }, + { + name: 'source', + event: { + type: 'user/message', seq: 0, time: 1, surfaceOp: 'append', + data: { ...user, source: null }, + }, + message: 'message has invalid source', + }, + { + name: 'assistant source', + event: { + type: 'assistant/message', seq: 0, time: 1, surfaceOp: 'append', + data: { + turn: 1, + step: 1, + message: { ...assistant, source: { kind: 'user' } }, + }, + }, + message: 'message must have model source', + }, + { + name: 'content block', + event: { + type: 'steering/message', seq: 0, time: 1, surfaceOp: 'append', + data: { + turn: 1, + message: { ...user, content: 'not-an-array' }, + }, + }, + message: 'message has invalid content', + }, + { + name: 'tool source', + event: { + type: 'tool/result', seq: 0, time: 1, surfaceOp: 'append', + data: { + turn: 1, + step: 1, + message: { ...tool, source: { kind: 'user' } }, + }, + }, + message: 'message must have tool source', + }, + { + name: 'tool tuple', + event: { + type: 'tool/result', seq: 0, time: 1, surfaceOp: 'append', + data: { + turn: 1, + step: 1, + message: { ...tool, content: [{ type: 'text', text: 'not a result' }] }, + }, + }, + message: 'message must contain one tool-result block', + }, + { + name: 'tool correlation', + event: { + type: 'tool/result', seq: 0, time: 1, surfaceOp: 'append', + data: { + turn: 1, + step: 1, + message: { + ...tool, + source: { kind: 'tool', callId: 'other-call' }, + }, + }, + }, + message: 'message has mismatched tool call ids', + }, + ] as const + + for (const { name, event, message } of invalid) { + expect( + () => new Session(SessionId(`invalid-${name}`), [event as unknown as SessionEvent]), + name, + ).toThrow(message) + } + }) + + it('snapshots message events without validating plugin-owned block details', () => { + const boundary = snapshotSessionEvent({ + type: 'turn/start', + seq: 0, + time: 1, + data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }, + }) + expect(boundary).toEqual({ + type: 'turn/start', + seq: 0, + time: 1, + data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }, + }) + + const extended = snapshotSessionEvent({ + type: 'user/message', + seq: 0, + time: 1, + surfaceOp: 'append', + data: { + id: 'extended-message', + role: 'user', + content: [{ type: 'plugin-block', value: 1 }], + source: { kind: 'plugin-source', value: 1 }, + }, + } as unknown as SessionEvent) + expect(extended.type === 'user/message' && extended.data.content) + .toEqual([{ type: 'plugin-block', value: 1 }]) + }) + it('round-trips a non-empty reasoning effort and rejects invalid durable values', () => { const valid = { type: 'request/header', diff --git a/packages/host/apiproxy/src/api-proxy.ts b/packages/host/apiproxy/src/api-proxy.ts index aa1478c938..b13af1d8d2 100644 --- a/packages/host/apiproxy/src/api-proxy.ts +++ b/packages/host/apiproxy/src/api-proxy.ts @@ -9,7 +9,7 @@ import { join } from 'node:path' import type { Context } from 'cordis' import { installAgentLlmTarget } from '@deepseek-ai/dsh-agent' import type { - Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, + Agent, AgentLlmTarget, AgentLlmTargetRef, AgentStatus, InboxPlacement, } from '@deepseek-ai/dsh-agent' import { createUserMessage, ReasoningEffortId } from '@deepseek-ai/dsh-llm' import { errorChain } from '@deepseek-ai/dsh-llm' @@ -428,10 +428,12 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro */ const queuedMirror = new Map() ctx.effect(() => { - const retire = (agent: Agent, id: MessageId): void => { + const retire = (agent: Agent, id: MessageId, placement?: InboxPlacement): void => { const entries = queuedMirror.get(agent.id) if (entries === undefined) return - const index = entries.findIndex(entry => entry.message.id === id) + const index = entries.findIndex(entry => + entry.message.id === id + && (placement === undefined || entry.steering === (placement === 'steering'))) if (index !== -1) entries.splice(index, 1) if (entries.length === 0) queuedMirror.delete(agent.id) } @@ -451,8 +453,8 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro steering, }) }), - ctx.on('agent/inbox/dequeue', (agent: Agent, message: UserMessage) => { - retire(agent, message.id) + ctx.on('agent/inbox/dequeue', (agent: Agent, message: UserMessage, placement) => { + retire(agent, message.id, placement) }), ctx.on('agent/inbox/discard', (agent: Agent, messages: UserMessage[]) => { for (const message of messages) retire(agent, message.id) diff --git a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts index 890d7aade2..95d440550b 100644 --- a/packages/host/apiproxy/tests/api-proxy-commands.spec.ts +++ b/packages/host/apiproxy/tests/api-proxy-commands.spec.ts @@ -284,8 +284,8 @@ describe('session/queued frames', () => { const steering = inboxMessage('m-4', 'x', 'r-1') ctx.emit('agent/inbox/enqueue', agent, queued, 'queued') ctx.emit('agent/inbox/enqueue', agent, steering, 'steering') - ctx.emit('agent/inbox/dequeue', agent, queued) - ctx.emit('agent/inbox/dequeue', agent, steering) + ctx.emit('agent/inbox/dequeue', agent, queued, 'queued') + ctx.emit('agent/inbox/dequeue', agent, steering, 'steering') const abort = new AbortController() const frames = await collect( @@ -293,15 +293,15 @@ describe('session/queued frames', () => { expect(frames.filter(f => f.type === 'session/queued')).toHaveLength(0) }) - it('retires repeated sends of one message identity by occurrence', async () => { + it('retires the matching placement when one message identity is queued and steering', async () => { const ctx = await harness() const api = createApiProxy(ctx, DEFAULTS) const agent = stubAgent(ctx) const repeated = inboxMessage('m-repeat', 'same prompt') ctx.emit('agent/inbox/enqueue', agent, repeated, 'queued') - ctx.emit('agent/inbox/enqueue', agent, repeated, 'queued') - ctx.emit('agent/inbox/dequeue', agent, inboxMessage('unknown', 'not queued')) - ctx.emit('agent/inbox/dequeue', agent, repeated) + ctx.emit('agent/inbox/enqueue', agent, repeated, 'steering') + ctx.emit('agent/inbox/dequeue', agent, inboxMessage('unknown', 'not queued'), 'queued') + ctx.emit('agent/inbox/dequeue', agent, repeated, 'steering') const abort = new AbortController() const frames = await collect( diff --git a/packages/session-persistence/session-persistence/README.i18n.yaml b/packages/session-persistence/session-persistence/README.i18n.yaml index 37714fddbd..0e2cecaada 100644 --- a/packages/session-persistence/session-persistence/README.i18n.yaml +++ b/packages/session-persistence/session-persistence/README.i18n.yaml @@ -1,6 +1,6 @@ # Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each # 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 -README.md: c99905e8aca0bdaf810de34841ea277b105a9d0f -README.zh.md: 106c28c5f9cd4330cf69b1648b669a04392e4e8a +# pnpm run verify-translation-pairing --write packages/session-persistence/session-persistence/README.md +README.md: 08d8adac8040747a6dac01dbc41525073f17060c +README.zh.md: 7676f27a1aa934eb3472e1b32b9ecd55d460fb63 diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index c99905e8ac..08d8adac80 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -13,8 +13,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l | `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. | | `create(meta): Promise` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). | | `append(id, events): Promise` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. | -| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and unknown `version` reject. | -| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. | +| `load(id): Promise<{ meta; events }>` | Return a stored header plus a balanced contiguous log whose events are detached and validated and whose identified messages are deeply frozen. A live load first flushes its snapshot and rejects while its turn is open; a cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption, malformed messages, and unknown `version` reject. | +| `inspect(id, signal?): Promise<{ meta; events }>` | Return a detached valid stored prefix with validated, deeply frozen identified messages, without truncating a torn tail, synthesizing recovery closers, or publishing coordinator state. Serialized with same-id writes; the optional signal promptly rejects a queued caller, prevents that queued backend read from starting, and cancels active backend read work. Intended for read models and other observers that must never recover a log. | | `list(signal?): Promise` | Lightweight listing from metadata, no full-log parse. The optional signal cancels backend listing work. A zero-event lazily-materialized session is absent from `list`. | | `listSnapshots(signal?): Promise` | Lightweight metadata plus an opaque branded per-log revision, without loading event logs. A revision stays equal while that log and its backing store are unchanged, changes after append or mutating load repair, and cannot collide solely because two stores use the same local counter. The optional signal requests cancellation of backend discovery work; first-party backends settle any started listing work before rejecting so an awaited call is quiescent. | diff --git a/packages/session-persistence/session-persistence/README.zh.md b/packages/session-persistence/session-persistence/README.zh.md index 106c28c5f9..7676f27a1a 100644 --- a/packages/session-persistence/session-persistence/README.zh.md +++ b/packages/session-persistence/session-persistence/README.zh.md @@ -13,8 +13,8 @@ | `locate(meta): SessionLocation \| undefined` | 在不执行 I/O 或实体化的情况下解析绝对的每会话产物目标。没有独立本地产物的后端返回 `undefined`。 | | `create(meta): Promise` | 注册新会话元数据。可以将物理写入延迟到第一次 `append`（延迟实体化）。 | | `append(id, events): Promise` | 持久保存一个批次。仅追加；任何修复后，第一个事件 `seq` == 已存储 next-seq；非 JSON 可序列化数据会被拒绝，并命名违规类型。 | -| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续日志。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏和未知 `version` 会被拒绝。 | -| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 | +| `load(id): Promise<{ meta; events }>` | 返回已存储 header 和平衡、连续的日志，其中事件已脱离并验证，带标识的消息已深度冻结。实时 load 先 flush 其快照，并在轮次开放时拒绝；冷 load 保留中断的最终轮次，并用合成 `tool/result`/`step/end?`/`turn/end {interrupted}` 事件关闭它。只丢弃撕裂尾部碎片；已提交损坏、格式错误的消息和未知 `version` 会被拒绝。 | +| `inspect(id, signal?): Promise<{ meta; events }>` | 返回脱离的有效已存储前缀，其中带标识的消息已经验证并深度冻结；不截断撕裂尾部、合成恢复 closer 或发布协调器状态。它与同 id 写入串行化；可选信号会迅速拒绝已排队调用方，阻止该后端读取启动，并取消活动后端读取工作。用于绝不应恢复日志的读模型和其他观察者。 | | `list(signal?): Promise` | 从元数据轻量列出，不解析完整日志。可选信号取消后端列表工作。零事件延迟实体化会话不在 `list` 中。 | | `listSnapshots(signal?): Promise` | 返回轻量元数据和不透明品牌化每日志修订，不加载事件日志。日志及其后端存储不变时，修订保持相等；append 或变更性 load 修复后会改变；不会仅因两个存储使用相同本地计数器而冲突。可选信号请求取消后端发现工作；第一方后端在拒绝前结算已启动列表工作，使已等待调用完全停稳。 | diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index c55f360f0a..878dddcb42 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -6,7 +6,12 @@ */ import { Context } from 'cordis' -import { interruptedTurnClosers, SESSION_FORMAT_VERSION, snapshotJsonValue } from '@deepseek-ai/dsh-session' +import { + interruptedTurnClosers, + SESSION_FORMAT_VERSION, + snapshotJsonValue, + snapshotSessionEvent, +} from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session' /** @@ -141,6 +146,12 @@ function assertSupportedEvents(events: readonly SessionEvent[], id: SessionId): } } +/** Materialize stored events as validated snapshots with immutable messages. */ +function snapshotStoredEvents(events: readonly SessionEvent[], id: SessionId): SessionEvent[] { + assertSupportedEvents(events, id) + return events.map(snapshotSessionEvent) +} + /** * Owns the backend-agnostic session write-path orchestration. A backend * constructs one (`new PersistenceCoordinator(ctx, this)`), implements @@ -307,10 +318,10 @@ export class PersistenceCoordinator { if (stored === undefined) throw new Error(`session "${id}" not found`) this.assertStoredId(id, stored.meta) this.assertVersion(stored.meta) - assertSupportedEvents(stored.events, id) + const events = snapshotStoredEvents(stored.events, id) return { meta: structuredClone(stored.meta), - events: structuredClone(stored.events), + events, } } @@ -320,11 +331,11 @@ export class PersistenceCoordinator { const { meta, events, tornMarker } = stored this.assertStoredId(id, meta) this.assertVersion(meta) - assertSupportedEvents(events, id) + const storedEvents = snapshotStoredEvents(events, id) // Preserve complete interrupted events and synthesize only missing closers. - const closers = interruptedTurnClosers(events) - const balanced = [...events, ...closers] + const closers = interruptedTurnClosers(storedEvents).map(snapshotSessionEvent) + const balanced = [...storedEvents, ...closers] // Repair storage before publishing coordinator state. if (tornMarker !== undefined || closers.length > 0) { @@ -332,12 +343,12 @@ export class PersistenceCoordinator { } // Keep coordinator metadata detached from the returned record. this.states.set(id, { meta: { ...meta }, cursor: balanced.length, materialized: true }) - return { meta, events: balanced } + return { meta: structuredClone(meta), events: balanced } } /** Return a durable balanced live snapshot without applying cold crash repair. */ private async loadLiveSnapshot(session: Session): Promise<{ meta: SessionHeader; events: SessionEvent[] }> { - const events = session.events.map(event => structuredClone(event)) + const events = session.events.map(snapshotSessionEvent) await this.flush(session) const state = this.states.get(session.id) /* v8 ignore next -- successful flush always publishes this live session's durable state */ diff --git a/packages/session-persistence/session-persistence/src/index.ts b/packages/session-persistence/session-persistence/src/index.ts index 9279e3c42c..261b26d364 100644 --- a/packages/session-persistence/session-persistence/src/index.ts +++ b/packages/session-persistence/session-persistence/src/index.ts @@ -92,6 +92,8 @@ export abstract class SessionPersistence extends Service { * open live turn rejects. * A coordinator-backed cold load reserves the identity across storage awaits, * so concurrent publication of a same-id live Session rejects. + * Returned events are detached, and every identified message is deeply + * frozen; malformed identified messages reject before any stored event is returned. * @param id - the persisted session to reload. * @returns the header and a log ending on a balanced `turn/end`. */ @@ -101,7 +103,8 @@ export abstract class SessionPersistence extends Service { * Inspect a header and its valid contiguous stored prefix without repairing * a torn tail, closing an interrupted turn, or publishing coordinator state. * This read is serialized with writes for the same id and returns detached - * values, so observers cannot mutate backend-owned state. + * values with deeply frozen identified messages, so observers cannot mutate message + * identity/content or backend-owned state. Malformed identified messages reject. * @param id - the persisted session to inspect. * @param signal - optional cancellation for queued and backend read work. * @returns the header and valid stored event prefix exactly as observed. diff --git a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts index 6b4b7b3f6c..d63cb857a2 100644 --- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts @@ -239,6 +239,65 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< } }) + it('load and inspect return immutable identified-message snapshots', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + const id = SessionId('immutable-read') + const session = ctx.sessions.create(id, { meta: { cwd: WORK } }) + send(session, oneTurnLog()) + await ctx.sessions.flush(session) + + for (const snapshot of [ + await ctx.sessionPersistence.load(id), + await ctx.sessionPersistence.inspect(id), + ]) { + const event = snapshot.events.find(candidate => candidate.type === 'user/message') + if (event?.type !== 'user/message') throw new Error('fixture lacks user/message') + expect(Object.isFrozen(event.data)).toBe(true) + expect(Object.isFrozen(event.data.content)).toBe(true) + expect(() => { + ;(event.data as { id: string }).id = 'rewritten' + }).toThrow(TypeError) + expect(() => { + ;(event.data.content[0] as { type: 'text'; text: string }).text = 'rewritten' + }).toThrow(TypeError) + } + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + + it('rejects malformed persisted message events before returning them', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + const id = SessionId('invalid-message-read') + await ctx.sessionPersistence.create(meta(id, WORK)) + await ctx.sessionPersistence.append(id, [{ + type: 'user/message', + seq: 0, + time: 1, + surfaceOp: 'append', + data: { + id: 'wrong-role', + role: 'assistant', + content: [{ type: 'text', text: 'wrong' }], + source: { kind: 'user' }, + }, + } as unknown as SessionEvent]) + + await expect(ctx.sessionPersistence.inspect(id)) + .rejects.toThrow('message must have role "user"') + await expect(ctx.sessionPersistence.load(id)) + .rejects.toThrow('message must have role "user"') + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + it('append snapshots the batch: mutating the caller array/events after the call is ignored', async () => { const fix = await makeFixture() const { ctx, fiber } = await freshCtx(fix) diff --git a/packages/session-query/session-query/src/index.ts b/packages/session-query/session-query/src/index.ts index 009ed9b8f6..809971b798 100644 --- a/packages/session-query/session-query/src/index.ts +++ b/packages/session-query/session-query/src/index.ts @@ -5,7 +5,7 @@ */ import { Context, Service } from 'cordis' -import { Session, type SessionId } from '@deepseek-ai/dsh-session' +import { Session, snapshotSessionEvent, type SessionId } from '@deepseek-ai/dsh-session' import { foldSessionTitle } from '@deepseek-ai/dsh-session-title' import type { SessionTitleSnapshot } from '@deepseek-ai/dsh-session-title' import type { @@ -45,7 +45,6 @@ import { materializeSessionResultFilters, } from './filters.ts' import * as tracing from './tracing.ts' -import { snapshotEvent } from './snapshot.ts' export type * from './types.ts' export { SessionSearchCursor } from './cursor.ts' @@ -147,7 +146,7 @@ export abstract class SessionQueryService extends Service { new Session(sessionId, loaded.events, loaded.header) return { session: structuredClone(loaded.header), - events: loaded.events.map(snapshotEvent), + events: loaded.events.map(snapshotSessionEvent), } } @@ -331,9 +330,11 @@ export abstract class SessionQueryService extends Service { } const startSeq = Math.max(0, seq - before) const endSeq = Math.min(loaded.events.length - 1, seq + after) - const targetSnapshot = snapshotEvent(target) + const targetSnapshot = snapshotSessionEvent(target) const events = loaded.events.slice(startSeq, endSeq + 1) - .map(event => event === target ? targetSnapshot : snapshotEvent(event)) + .map(event => event === target + ? targetSnapshot + : snapshotSessionEvent(event)) return { session: structuredClone(loaded.header), target: targetSnapshot, diff --git a/packages/session-query/session-query/src/snapshot.ts b/packages/session-query/session-query/src/snapshot.ts deleted file mode 100644 index 9d6db85ecc..0000000000 --- a/packages/session-query/session-query/src/snapshot.ts +++ /dev/null @@ -1,27 +0,0 @@ -/** Detached session-query snapshots that preserve message immutability. */ - -import { deepFreeze } from '@deepseek-ai/dsh-llm' -import type { SessionEvent } from '@deepseek-ai/dsh-session' - -/** - * Clone one event while retaining the invariant that every identified message is frozen. - * @param event - source event from one corpus observation. - * @returns a detached event whose message value, if any, is deeply frozen. - */ -export function snapshotEvent(event: T): T { - const snapshot = structuredClone(event) - switch (snapshot.type) { - case 'user/message': - deepFreeze(snapshot.data) - break - case 'assistant/message': - case 'tool/result': - case 'steering/message': - deepFreeze(snapshot.data.message) - break - default: - // SessionEventMap is merge-extensible; plugin-owned log-only events carry no core message. - break - } - return snapshot -} diff --git a/packages/session-query/session-query/src/tracing.ts b/packages/session-query/session-query/src/tracing.ts index e894301012..0fdf723d01 100644 --- a/packages/session-query/session-query/src/tracing.ts +++ b/packages/session-query/session-query/src/tracing.ts @@ -1,6 +1,6 @@ /** One-shot session-lineage and event-relationship tracing helpers. */ -import { foldSurface, isSurfaceEvent } from '@deepseek-ai/dsh-session' +import { foldSurface, isSurfaceEvent, snapshotSessionEvent } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionId, SurfaceEvent, SurfaceEventType } from '@deepseek-ai/dsh-session' import { SessionQueryError } from './config.ts' import type { @@ -10,7 +10,6 @@ import type { SessionLineageTrace, SessionRecord, } from './types.ts' -import { snapshotEvent } from './snapshot.ts' interface EventLogAnalysis { records: SessionEventRecord[] @@ -52,7 +51,7 @@ export function currentSurfaceEvents( 'SESSION_QUERY_INVALID_SURFACE', ) } - return snapshotEvent(event) + return snapshotSessionEvent(event) }) } diff --git a/packages/ui/tui/tests/tui.spec.ts b/packages/ui/tui/tests/tui.spec.ts index 5be755b9e5..ca2976475b 100644 --- a/packages/ui/tui/tests/tui.spec.ts +++ b/packages/ui/tui/tests/tui.spec.ts @@ -1474,7 +1474,7 @@ describe('pi-tui chat lifecycle and transcript', () => { role: 'user', content: [{ type: 'text', text }], source: { kind: 'user' }, - })) + }), 'steering') } result.session.append('steering/message', { turn: 1, @@ -1566,13 +1566,13 @@ describe('pi-tui chat lifecycle and transcript', () => { })) // Another agent's dequeue/discard, and ones naming no pending id, leave // the badge alone. - result.ctx.emit('agent/inbox/dequeue', other, discarded[0]!) + result.ctx.emit('agent/inbox/dequeue', other, discarded[0]!, 'steering') result.ctx.emit('agent/inbox/dequeue', result.agent, freezeMessage({ id: MessageId('never-queued'), role: 'user', content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, - })) + }), 'steering') result.ctx.emit('agent/inbox/discard', other, discarded) result.ctx.emit('agent/inbox/discard', result.agent, [ freezeMessage({