From 7e95a00c8a5eed37fc8d16487b6a1a9b772b075c Mon Sep 17 00:00:00 2001 From: Yichen Jiang Date: Sat, 15 Aug 2026 16:07:30 +0800 Subject: [PATCH] fix(llm): align replay state with assembled content and degrade unusable state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A max-tokens response that included a tool call persisted assembler-transformed content next to replay metadata projected from the untransformed native message, so the next request died in history reconstruction with INVALID_REPLAY_STATE and the session stayed permanently stuck. Write side: the finish chunk's replayState becomes a typed ReplayEnvelope — opaque response-level metadata plus optional per-block entries aligned with the emitted block sequence. BlockAssembler computes one keep/drop decision for blocks and entries together, so stored metadata always describes stored content and retained blocks keep their signatures. pi-ai splits its state into a version-2 response half and per-block signature entries. Read side: durable content is authoritative. toPiAssistant degrades any unusable state — foreign kind, other versions (including the flat v1 form already on disk), malformed metadata, or content/block mismatches — to the existing provider-neutral conversion with an onReplayDegrade diagnostic instead of failing the request, which un-bricks sessions poisoned before this change. Covered by assembler and replay unit tests, an agent-loop continuation regression, keyless real-composition continuation tests (native pruned-envelope replay and legacy flat-state degrade), and the authored keyless snapshot scenario max-tokens-continue through the assembled ACP app. --- ...-14-provider-routed-llm-adapters.i18n.yaml | 4 +- ...2026-07-14-provider-routed-llm-adapters.md | 4 +- ...6-07-14-provider-routed-llm-adapters.zh.md | 4 +- ...max-token-replay-state-alignment.i18n.yaml | 6 + ...-08-15-max-token-replay-state-alignment.md | 33 +++ ...-15-max-token-replay-state-alignment.zh.md | 33 +++ docs/subsystems/llm-streaming.i18n.yaml | 4 +- docs/subsystems/llm-streaming.md | 39 ++- docs/subsystems/llm-streaming.zh.md | 39 ++- examples/acp-agent/tests/acp.snapshot.ts | 7 + .../snapshots/max-tokens-continue/input.json | 8 + .../max-tokens-continue/session.jsonl | 33 +++ .../max-tokens-continue/stdout.expected.jsonl | 6 + .../tests/contract-regressions.spec.ts | 2 +- packages/core/agent-loop/tests/loop.spec.ts | 35 ++- .../extensions/tool-cordis/src/api-catalog.ts | 6 +- packages/llm/llm-pi-ai/README.i18n.yaml | 4 +- packages/llm/llm-pi-ai/README.md | 4 +- packages/llm/llm-pi-ai/README.zh.md | 4 +- packages/llm/llm-pi-ai/src/adapter.ts | 12 +- packages/llm/llm-pi-ai/src/context.ts | 37 ++- packages/llm/llm-pi-ai/src/index.ts | 6 + packages/llm/llm-pi-ai/src/replay.ts | 98 +++++--- packages/llm/llm-pi-ai/tests/convert.spec.ts | 234 +++++++++++------- .../tests/loader-composition.spec.ts | 130 +++++++++- .../llm/llm-pi-ai/tests/provider-apis.e2e.ts | 18 +- packages/llm/llm/README.i18n.yaml | 4 +- packages/llm/llm/README.md | 2 +- packages/llm/llm/README.zh.md | 2 +- packages/llm/llm/src/assembler.ts | 41 ++- packages/llm/llm/src/types.ts | 25 +- packages/llm/llm/tests/assembler.spec.ts | 79 ++++++ scripts/type-equiv.manifest.json | 5 + 33 files changed, 782 insertions(+), 186 deletions(-) create mode 100644 .agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.i18n.yaml create mode 100644 .agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.md create mode 100644 .agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.zh.md create mode 100644 examples/acp-agent/tests/snapshots/max-tokens-continue/input.json create mode 100644 examples/acp-agent/tests/snapshots/max-tokens-continue/session.jsonl create mode 100644 examples/acp-agent/tests/snapshots/max-tokens-continue/stdout.expected.jsonl diff --git a/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.i18n.yaml b/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.i18n.yaml index a79104468c..3f4683f480 100644 --- a/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.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-14-provider-routed-llm-adapters.md -2026-07-14-provider-routed-llm-adapters.md: e1eaf52f21481a7c65e85effb7607b16f9b0ffdd -2026-07-14-provider-routed-llm-adapters.zh.md: 4e620408dabffc8368293635613afb5778b6e822 +2026-07-14-provider-routed-llm-adapters.md: 78c8d6788006c503b532ff2bbddd30342415f0a4 +2026-07-14-provider-routed-llm-adapters.zh.md: 5e73cab5f1f2b1296c9a486d1c833e95bb5674a0 diff --git a/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.md b/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.md index e1eaf52f21..78c8d67880 100644 --- a/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.md +++ b/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.md @@ -40,9 +40,9 @@ pi-ai's common stream options do not expose stop sequences. `dsh-llm-pi-ai` reje Assistant messages carry the request's `provider` and `model`, plus an optional JSON-serializable adapter replay state. A successful `assistant/message` session event records those fields and `deriveMessages()` returns them with the assistant message. User, system, context, and tool-result messages carry no assistant route fields. The provider/model fields are authoritative loop data; an adapter owns only its opaque replay-state payload. -A terminal successful `finish` chunk may carry replay state, and `BlockAssembler` retains it alongside usage and finish reason. The loop attaches that state to the assembled assistant message's model source without exposing a response-rewrite hook. Error and aborted responses do not produce a normal assistant message and therefore do not enter future model history. +A terminal successful `finish` chunk may carry replay state as a `ReplayEnvelope`: opaque response-level metadata plus optional per-block entries aligned with the emitted block sequence. `BlockAssembler` makes one keep/drop decision for content and metadata — when max-token assembly drops a tool call, the envelope loses the entry at the same position — so the state the loop attaches to the assembled assistant message's model source always describes the stored blocks, per the [max-token replay-state alignment decision](../bug-fix/2026-08-15-max-token-replay-state-alignment.md). The loop exposes no response-rewrite hook. Error and aborted responses do not produce a normal assistant message and therefore do not enter future model history. -The pi-ai replay state is a versioned, minimal projection of its successful `AssistantMessage`: source API/provider/model, response id/model, stop reason, and index-aligned text, thinking, and tool-call signatures. It does not duplicate text or tool arguments already carried by Harness content blocks, and it omits diagnostics, timestamps, usage, and errors. On a later request, `LlmRuntime` gives replay state to the target adapter only when the historical provider and target provider are currently owned by the same adapter instance. That adapter combines the logged Harness content with replay state when it can restore the historical response, and owns any required cross-model or cross-provider conversion. An adapter receiving replay state with an unknown version or mismatched block shape fails explicitly; a different adapter receives only provider-neutral content plus provider/model fields. +The pi-ai replay state fills that envelope with a versioned, minimal projection of its successful `AssistantMessage`: a response half (source API/provider/model, response id/model, stop reason) and per-block text, thinking, and tool-call signatures. It does not duplicate text or tool arguments already carried by Harness content blocks, and it omits diagnostics, timestamps, usage, and errors. On a later request, `LlmRuntime` gives replay state to the target adapter only when the historical provider and target provider are currently owned by the same adapter instance. That adapter combines the logged Harness content with replay state when it can restore the historical response, and owns any required cross-model or cross-provider conversion. Durable content stays authoritative: an adapter receiving replay state it cannot use — an unknown kind or version, malformed metadata, or a block shape that no longer matches the content — degrades that message to provider-neutral conversion with a diagnostic; a different adapter receives only provider-neutral content plus provider/model fields. This state is model-visible replay input and therefore follows the existing [reconstructable-request rule](2026-07-05-reconstructable-requests.md): it is present in both the terminal `finish` chunk and the assembled `assistant/message` model source that drives derivation. Resume and fork preserve it verbatim. Compaction that shadows the assistant message also removes its replay state from the active surface; the summary is ordinary provider-neutral content. diff --git a/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.zh.md b/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.zh.md index 4e620408da..5e73cab5f1 100644 --- a/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.zh.md +++ b/.agents/notes/implemented/architecture/2026-07-14-provider-routed-llm-adapters.zh.md @@ -40,9 +40,9 @@ pi-ai 的通用流选项不支持停止序列。若 Harness `stop` 选项已定 助手消息携带请求的 `provider` 和 `model`，以及可选的 JSON 可序列化适配器回放状态。成功的 `assistant/message` 会话事件记录这些字段，`deriveMessages()` 返回助手消息时也会包含它们。用户、系统、上下文与工具结果消息不携带助手路由字段。提供方/模型字段是 agent loop 的权威数据；适配器仅拥有其不透明回放状态 payload。 -成功的终止 `finish` 分片可以携带回放状态，`BlockAssembler` 会将其与 token 用量和结束原因一起保留。agent loop 会把该状态附加到已组装助手消息的模型来源中，但不公开响应改写钩子。错误或中止响应不会生成正常助手消息，因此不会进入后续模型历史。 +成功的终止 `finish` 分片可以以 `ReplayEnvelope` 形式携带回放状态：不透明的响应级元数据，加上与发射块序列对齐的可选逐块条目。`BlockAssembler` 对内容与元数据只做一次保留/丢弃决定——max-token 组装丢弃工具调用时，数据同一位置的条目一并丢弃——因此 agent loop 附加到已组装助手消息模型来源中的状态始终描述存储的块，见 [max-token 回放状态对齐决定](../bug-fix/2026-08-15-max-token-replay-state-alignment.md)。agent loop 不公开响应改写钩子。错误或中止响应不会生成正常助手消息，因此不会进入后续模型历史。 -pi-ai 回放状态是其成功 `AssistantMessage` 的带版本最小投影，包含源 API/提供方/模型、响应 ID/模型、停止原因，以及按索引对齐的文本签名、thinking 签名和工具调用签名。它不会重复 Harness 内容块中已有的文本或工具参数，也不包含诊断信息、时间戳、用量或错误。后续请求中，只有历史提供方和目标提供方当前归同一个适配器实例所有时，`LlmRuntime` 才会把回放状态交给目标适配器。适配器在能够恢复历史响应时，将 Harness 记录的内容与回放状态组合，并负责所需的跨模型或跨提供方转换。适配器收到未知版本或块形状不匹配的回放状态时会显式失败；其他适配器只能收到提供方无关的内容以及提供方/模型字段。 +pi-ai 回放状态用其成功 `AssistantMessage` 的带版本最小投影填充该结构：一个响应半区（源 API/提供方/模型、响应 ID/模型、停止原因），以及逐块的文本签名、thinking 签名和工具调用签名。它不会重复 Harness 内容块中已有的文本或工具参数，也不包含诊断信息、时间戳、用量或错误。后续请求中，只有历史提供方和目标提供方当前归同一个适配器实例所有时，`LlmRuntime` 才会把回放状态交给目标适配器。适配器在能够恢复历史响应时，将 Harness 记录的内容与回放状态组合，并负责所需的跨模型或跨提供方转换。持久化内容保持权威：适配器收到无法使用的回放状态——未知 kind 或版本、格式错误的元数据、或与内容不再匹配的块结构——会把该消息降级为提供方无关转换并带出诊断；其他适配器只能收到提供方无关的内容以及提供方/模型字段。 该状态属于模型可见的回放输入，因此遵循现有的[请求可重建规则](2026-07-05-reconstructable-requests.md)：它同时存在于终止 `finish` 分片和驱动派生的已组装 `assistant/message` 模型来源中。恢复和 fork 会原样保留该状态。压缩（compaction）遮蔽助手消息时，也会从活动 surface 中移除其回放状态；摘要属于普通的提供方无关内容。 diff --git a/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.i18n.yaml b/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.i18n.yaml new file mode 100644 index 0000000000..af691f1175 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.i18n.yaml @@ -0,0 +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 .agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.md +2026-08-15-max-token-replay-state-alignment.md: 256a64403a08377cf35ba645175698678eaa7f8b +2026-08-15-max-token-replay-state-alignment.zh.md: a24f3e194dca100d2ea0faf659b1868c605c26c1 diff --git a/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.md b/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.md new file mode 100644 index 0000000000..256a64403a --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.md @@ -0,0 +1,33 @@ +# Agent Note: Replay state aligns with assembled content by construction + +Status: implemented + +English | [中文](2026-08-15-max-token-replay-state-alignment.zh.md) + +## Problem + +pi-ai recorded one opaque replay blob per response, projected from the provider's native message, while `BlockAssembler.blocks()` separately dropped tool calls from a `max-tokens` response because a truncated call is unsafe to execute. The durable assistant message therefore stored transformed content next to metadata describing the untransformed native block list. The next request failed during history reconstruction with `INVALID_REPLAY_STATE: block count does not match assistant content`, and because the mismatch was already on disk, every later request on that session failed the same way — the session was permanently stuck. The root cause is structural: two representations of one response were snapshotted at different pipeline points, with their index alignment enforced only by a read-time hard error. + +## Decision + +Two changes, one per side of the durable boundary. + +**Write side — one keep/drop decision.** The finish chunk's `replayState` becomes a typed `ReplayEnvelope`: an opaque `response` half plus optional opaque per-block entries aligned with the emitted block sequence. `BlockAssembler` computes its keep/drop decision once and applies it to blocks and envelope entries together, so any transformation assembly performs — today's max-token tool-call drop or a future one — prunes the matching metadata by construction. Retained blocks keep their entries, so a truncated response keeps signatures for the reasoning and text it kept. An envelope whose entries do not match the emitted block count is discarded whole (a misemitting adapter must not publish misattributed metadata). pi-ai splits its former flat state into a version-2 response half and per-block signature entries. + +**Read side — durable content is authoritative.** `toPiAssistant` treats replay state as fidelity metadata, not as a load-bearing input: any state the reading build cannot use — another adapter's kind, another version (including the flat version-1 form already on disk), malformed metadata, or a block shape that no longer matches the content — degrades that one message to the existing foreign provider-neutral conversion and reports the `INVALID_REPLAY_STATE` diagnostic through the plugin's `onReplayDegrade` hook (a logger warning). The request proceeds. This is what lets sessions poisoned before this change continue instead of erroring forever, and it bounds every future divergence source to a fidelity loss on one message. + +## Verification + +Assembler unit tests prove pruning, misalignment discard, and pass-through for untransformed and per-block-free envelopes. pi-ai unit tests prove the version-2 envelope round-trip and that every formerly-throwing invalid-state case now degrades to foreign conversion with the diagnostic. An agent-loop regression drives a truncated text-plus-tool-call response through persistence and shows the follow-up request carrying the pruned envelope. Keyless real-composition tests boot `dsh-llm-pi-ai` through the Loader and prove a native continuation without `tool_calls` after truncation, and a successful continuation over a legacy flat-state message whose block count no longer matches. The authored keyless snapshot scenario `max-tokens-continue` pins the assembled application's durable log — truncated turn, pruned envelope on the stored message, continued turn — through the real ACP subprocess path. + +## Alternatives considered + +**Suppress the whole replay state when assembly drops a tool call.** Works for today's one transformation, but re-derives the drop condition beside `blocks()` (the two drift silently), discards valid signatures for the retained blocks, and leaves read-time divergence — legacy sessions on disk foremost — a hard error. + +**Keep the state and relax pi-ai's block-count validation to attach what fits.** Rejected: index-aligned signatures attached to a different block list would present false native history to the provider. Degrading attaches nothing. + +**Teach each adapter to rewrite its state after assembly.** Rejected as an adapter obligation with an opaque blob; the envelope moves exactly the needed structure — and nothing else — into shared vocabulary, and the assembler's single decision does the rewrite mechanically. + +## Consequences + +Continuing after a max-token response that included a tool call works, retains the kept blocks' native signatures, and replays as a native pi-ai message. Sessions recorded before this change replay their affected assistant messages as provider-neutral content (with a diagnostic) instead of failing the turn; on-disk `replayState` values changed shape under the pre-release no-compatibility stance, with the old flat form handled by the same degrade path. This supersedes the read-time hard-error rule in the [provider-routed adapter decision](../architecture/2026-07-14-provider-routed-llm-adapters.md) for unusable state; validation itself is unchanged and still precedes any native reconstruction. diff --git a/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.zh.md b/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.zh.md new file mode 100644 index 0000000000..a24f3e194d --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-08-15-max-token-replay-state-alignment.zh.md @@ -0,0 +1,33 @@ +# Agent Note: 回放状态与组装内容按构造对齐 + +Status: implemented + +[English](2026-08-15-max-token-replay-state-alignment.md) | 中文 + +## 问题 + +pi-ai 为每个响应记录一个从提供方原生消息投影而来的不透明回放数据，而 `BlockAssembler.blocks()` 会另行从 `max-tokens` 响应中丢弃工具调用，因为被截断的调用不能安全执行。持久化的 assistant 消息因此把变换后的内容与描述未变换原生块清单的元数据存在一起。下一个请求在历史重建阶段以 `INVALID_REPLAY_STATE: block count does not match assistant content` 失败；由于不一致已经落盘，该会话之后的每个请求都以同样方式失败——会话被永久卡死。根因是结构性的：同一响应的两种表示在流水线的不同位置各自拍摄快照，其索引对齐只靠读取时的硬错误来维持。 + +## 决定 + +两处改动，各覆盖持久化边界的一侧。 + +**写侧——一次保留/丢弃决定。** finish 分片的 `replayState` 变为有类型的 `ReplayEnvelope`：一个不透明的 `response` 半区，加上与发射块序列对齐的可选不透明逐块条目。`BlockAssembler` 只计算一次保留/丢弃决定，并把它同时应用于块和逐块条目，因此组装执行的任何变换——今天的 max-token 工具调用丢弃或未来的其他变换——都按构造裁剪掉对应元数据。保留的块保留其条目，所以被截断的响应仍为其保留的推理（reasoning）与文本保有签名。条目数与发射块数不一致的数据整体丢弃（发射不当的适配器不得发布归属错误的元数据）。pi-ai 把原先的平铺状态拆为版本 2 的 response 半区和逐块签名条目。 + +**读侧——持久化内容是权威记录。** `toPiAssistant` 把回放状态当作保真度元数据，而非承重输入：读取方无法使用的任何状态——其他适配器的 kind、其他版本（包括已落盘的平铺版本 1 形式）、格式错误的元数据、或与内容不再匹配的块结构——都把这一条消息降级为既有的外来提供方无关转换，并通过插件的 `onReplayDegrade` 钩子（logger 警告）上报 `INVALID_REPLAY_STATE` 诊断。请求继续执行。正是这一点让本次改动之前已被毒化的会话得以继续而不是永远报错，也把未来一切分叉源约束为单条消息的保真度损失。 + +## 验证 + +组装器单元测试证明裁剪、错位丢弃、以及未变换与无逐块条目数据的透传。pi-ai 单元测试证明版本 2 数据的往返，以及先前每个抛错的无效状态用例现在都降级为外来转换并带出诊断。agent loop 回归用例驱动一个被截断的文本加工具调用响应穿过持久化，并证明后续请求携带裁剪后的数据。无密钥真实组合测试通过 loader 启动 `dsh-llm-pi-ai`，证明截断后不带 `tool_calls` 的原生续聊，以及在块数不再匹配的旧平铺状态消息之上成功续聊。手工编写的无密钥快照场景 `max-tokens-continue` 通过真实 ACP 子进程路径钉住组装应用的持久化日志——截断轮次、存储消息上裁剪后的数据、以及继续的轮次。 + +## 已考虑的替代方案 + +**组装丢弃工具调用时抑制整个回放状态。** 对今天唯一的变换有效，但在 `blocks()` 旁边重新推导丢弃条件（两处会无声漂移），丢掉保留块的有效签名，并让读取时的分叉——首当其冲是已落盘的旧会话——仍然是硬错误。 + +**保留状态并放宽 pi-ai 的块数校验、能贴多少贴多少。** 否决：索引对齐的签名贴到不同的块清单上，会向提供方呈现虚假的原生历史。降级则什么都不贴。 + +**让每个适配器在组装后改写自己的状态。** 否决：这把义务压给持有不透明数据的适配器；信封只把恰好需要的结构——不多一分——纳入共享词汇，组装器的单一决定即可机械完成改写。 + +## 影响 + +包含工具调用的 max-token 响应之后的续聊可以工作，保留块保有原生签名，并作为原生 pi-ai 消息回放。本次改动之前记录的会话，其受影响的 assistant 消息作为提供方无关内容回放（带诊断）而不是让轮次失败；`replayState` 落盘形状在预发布无兼容承诺立场下发生变化，旧平铺形式由同一降级路径处理。对不可用状态而言，这取代了[提供方路由适配器决定](../architecture/2026-07-14-provider-routed-llm-adapters.md)中读取时硬错误的规则；校验本身不变，仍先于任何原生重建执行。 diff --git a/docs/subsystems/llm-streaming.i18n.yaml b/docs/subsystems/llm-streaming.i18n.yaml index 5708a7b6d5..8f287e2a0f 100644 --- a/docs/subsystems/llm-streaming.i18n.yaml +++ b/docs/subsystems/llm-streaming.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 docs/subsystems/llm-streaming.md -llm-streaming.md: 0d3a0d53c875c9d943146ba44b775d81fc9cae01 -llm-streaming.zh.md: fbaa47d14d57e7377be4db6ecaa04f11997572a6 +llm-streaming.md: 7c0e0865f8dcc0e7722bb2205d0129d9e0ca3086 +llm-streaming.zh.md: 5c31909ee79137c6c5eef101235b43a2419b1339 diff --git a/docs/subsystems/llm-streaming.md b/docs/subsystems/llm-streaming.md index 0d3a0d53c8..7c0e0865f8 100644 --- a/docs/subsystems/llm-streaming.md +++ b/docs/subsystems/llm-streaming.md @@ -157,6 +157,29 @@ type ContextFormed = A streaming response interleaves several typed blocks (text, reasoning, multiple tool calls). `index` ties each delta to its block; `block-end` carries the fully-assembled `ContentBlock` so consumers don't have to re-assemble deltas themselves. It is a **closed** discriminated union — a `switch` over `type` ends with `assertNever`, so adding a variant breaks compilation at every consumer that must handle it. +```ts type-equiv +/** + * Adapter-private lossless-JSON state for replaying a successful response, + * carried by a terminal `finish` chunk and stored on the assembled assistant + * message's model source. Both halves stay opaque to the harness; only the + * split is shared vocabulary, so assembly can keep stored metadata aligned + * with stored content without reading either half. + */ +interface ReplayEnvelope { + /** Response-level adapter-private metadata (ids, native stop reason). */ + response: unknown + /** + * Per-block adapter-private metadata, one entry per emitted block in + * first-seen stream order. When assembly drops a block it drops the entry at + * the same position; entries whose length does not match the emitted block + * count discard the whole envelope. An adapter whose metadata is independent + * of block structure omits this field and the envelope passes through + * assembly unchanged. + */ + blocks?: readonly unknown[] +} +``` + ```ts type-equiv /** * Raw streaming protocol emitted by adapters. @@ -176,8 +199,8 @@ type StreamChunk = | { type: 'finish' reason: FinishReason - /** Adapter-private lossless-JSON state for replaying a successful response. */ - replayState?: unknown + /** Replay metadata for a successful response; see {@link ReplayEnvelope}. */ + replayState?: ReplayEnvelope } ``` @@ -213,7 +236,7 @@ Every adapter MUST obey these, and every consumer may rely on them: - **Context overflow has one canonical code.** Both DeepSeek adapters classify explicit provider detail through `isContextWindowExceededError()` and surface `CONTEXT_WINDOW_EXCEEDED`, whether the failure arrives as a thrown HTTP `LlmError` or an in-band finish error. Consumers route on the code, never provider text. - **An empty completion is a retryable error, not a silent success.** Both adapters map a terminal `stop` finish that carried no content blocks to `finish {kind:'error'}` with the canonical `EMPTY_RESPONSE` code, and `dsh-llm-retry` retries it by default; see [empty model responses are retryable](../../.agents/notes/implemented/bug-fix/2026-07-24-empty-model-response-is-retryable.md). - **Every provider HTTP request carries the app-attribution header.** Adapters send `attributionHeaders()` (below) - the `User-Agent` baseline - and prove it with a wire-level test. -- **Replay state is adapter-owned.** A successful `finish` may carry lossless-JSON state needed to reconstruct a native provider response. The loop stores it with the assembled assistant message. On a later request, `LlmRuntime` passes the state only when the historical provider and target provider are currently registered to the exact same adapter instance. That adapter validates the state and owns any cross-model or cross-provider conversion; other adapters receive the provider-neutral content plus provider/model fields without the private state. +- **Replay state is adapter-owned; its split is shared.** A successful `finish` may carry a `ReplayEnvelope`: opaque response-level metadata plus optional per-block entries aligned with the emitted block sequence. The alignment is the harness's vocabulary — when assembly drops a block it drops the entry at the same position, so stored metadata always describes stored content. The loop stores the pruned envelope with the assembled assistant message. On a later request, `LlmRuntime` passes the state only when the historical provider and target provider are currently registered to the exact same adapter instance. That adapter validates the state and owns any cross-model or cross-provider conversion; other adapters receive the provider-neutral content plus provider/model fields without the private state. Durable content stays authoritative: a stored state the reading adapter cannot use degrades that one message to provider-neutral conversion with a diagnostic instead of failing the request. ## `ResolvedRetryPolicy` @@ -267,6 +290,8 @@ interface TokenUsage { `BlockAssembler` ([`packages/llm/llm/src/assembler.ts`](../../packages/llm/llm/src/assembler.ts)) is the single shared implementation that folds a `StreamChunk` stream back into `ContentBlock`s, usage, finish reason, and replay state. The loop logs the raw chunks while feeding the same chunks through an assembler, then stores the assembled assistant content with the provider and model that produced it. A consumer that needs the assembled result without re-implementing the fold uses this. +One keep/drop decision covers content and metadata together: a `max-tokens` finish drops every tool call because a truncated call is unsafe to execute, and the same decision prunes the replay envelope's per-block entry at each dropped position. `blocks()` and `replayState` therefore cannot disagree, whatever assembly removes. + ```ts public-api /** * Incrementally assembles raw {@link StreamChunk}s into complete @@ -296,8 +321,12 @@ declare class BlockAssembler { get usage(): TokenUsage | undefined; /** Finish reason from the `finish` chunk; `{kind: 'stop'}` when the stream ended without one. */ get finish(): FinishReason; - /** Adapter-private replay state from the terminal finish chunk, if any. */ - get replayState(): unknown; + /** + * Replay metadata from the terminal finish chunk, if any, with per-block + * entries pruned in step with {@link blocks}. Undefined when the envelope's + * entries do not align with the emitted blocks. + */ + get replayState(): ReplayEnvelope | undefined; /** * The assembled assistant message. * @param source - producer attribution for the assembled message. diff --git a/docs/subsystems/llm-streaming.zh.md b/docs/subsystems/llm-streaming.zh.md index fbaa47d14d..5c31909ee7 100644 --- a/docs/subsystems/llm-streaming.zh.md +++ b/docs/subsystems/llm-streaming.zh.md @@ -157,6 +157,29 @@ type ContextFormed = 一个流式响应交错包含多种类型的块（文本、推理（reasoning）、多个工具调用）。`index` 将每个 delta 关联到其所属块；`block-end` 携带完整组装好的 `ContentBlock`，消费方无需自行重新组装 delta。这是一个**封闭的**可辨识联合类型：对 `type` 的 `switch` 以 `assertNever` 结尾，因此新增变体会在每个必须处理它的消费方处触发编译错误。 +```ts type-equiv +/** + * Adapter-private lossless-JSON state for replaying a successful response, + * carried by a terminal `finish` chunk and stored on the assembled assistant + * message's model source. Both halves stay opaque to the harness; only the + * split is shared vocabulary, so assembly can keep stored metadata aligned + * with stored content without reading either half. + */ +interface ReplayEnvelope { + /** Response-level adapter-private metadata (ids, native stop reason). */ + response: unknown + /** + * Per-block adapter-private metadata, one entry per emitted block in + * first-seen stream order. When assembly drops a block it drops the entry at + * the same position; entries whose length does not match the emitted block + * count discard the whole envelope. An adapter whose metadata is independent + * of block structure omits this field and the envelope passes through + * assembly unchanged. + */ + blocks?: readonly unknown[] +} +``` + ```ts type-equiv /** * Raw streaming protocol emitted by adapters. @@ -176,8 +199,8 @@ type StreamChunk = | { type: 'finish' reason: FinishReason - /** Adapter-private lossless-JSON state for replaying a successful response. */ - replayState?: unknown + /** Replay metadata for a successful response; see {@link ReplayEnvelope}. */ + replayState?: ReplayEnvelope } ``` @@ -215,7 +238,7 @@ interface LlmFailure { - **上下文溢出只有一个规范 code。** 两个 DeepSeek 适配器都通过 `isContextWindowExceededError()` 对提供方的显式细节分类并暴露 `CONTEXT_WINDOW_EXCEEDED`，无论失败以抛出的 HTTP `LlmError` 还是带内 finish error 到达。消费方按 code 路由，绝不依赖提供方文本。 - **空 completion 是可重试错误，而不是静默的成功结果。** 两个适配器都把没有携带任何内容块的终止性 `stop` 结束映射为携带规范 `EMPTY_RESPONSE` code 的 `finish {kind:'error'}`，`dsh-llm-retry` 默认会重试它；详见[空模型响应可重试](../../.agents/notes/implemented/bug-fix/2026-07-24-empty-model-response-is-retryable.md)。 - **每个提供方 HTTP 请求都携带应用归属头。** 适配器发送 `attributionHeaders()`（见下文）作为 `User-Agent` 基线，并通过协议级测试加以证明。 -- **回放状态归适配器所有。** 成功的 `finish` 可以携带重建提供方原生响应所需的无损 JSON 状态。循环会将其与组装后的 assistant 消息一起存储。后续请求中，仅当历史提供方与目标提供方当前注册到完全相同的适配器实例时，`LlmRuntime` 才会传递该状态。该适配器负责校验状态并拥有所有跨模型或跨提供方转换；其他适配器只会收到提供方无关的内容以及提供方／模型字段，不会收到私有状态。 +- **回放状态归适配器所有；其切分是共享词汇。** 成功的 `finish` 可以携带一个 `ReplayEnvelope`：不透明的响应级元数据，加上与发射块序列对齐的可选逐块条目。对齐关系是 harness 的词汇——组装丢弃某个块时，同一位置的条目一并丢弃，因此存储的元数据始终描述存储的内容。循环把裁剪后的数据与组装后的 assistant 消息一起存储。后续请求中，仅当历史提供方与目标提供方当前注册到完全相同的适配器实例时，`LlmRuntime` 才会传递该状态。该适配器负责校验状态并拥有所有跨模型或跨提供方转换；其他适配器只会收到提供方无关的内容以及提供方／模型字段，不会收到私有状态。持久化内容保持权威：读取适配器无法使用的已存状态只会把这一条消息降级为提供方无关转换并带出诊断，而不是让请求失败。 ## `ResolvedRetryPolicy` @@ -273,6 +296,8 @@ interface TokenUsage { `BlockAssembler`（[`packages/llm/llm/src/assembler.ts`](../../packages/llm/llm/src/assembler.ts)）是唯一的共享实现，负责把 `StreamChunk` 流折叠回 `ContentBlock`、usage、结束原因与回放状态。循环在记录原始分片的同时，把同一批分片送入 assembler，再将组装后的 assistant 内容连同生成它的提供方和模型一起存储。需要组装结果、又不想重新实现 fold 的消费方使用它。 +内容与元数据共用同一次保留/丢弃决定：`max-tokens` 结束会丢弃每个工具调用，因为被截断的调用不能安全执行，而同一决定会在每个被丢弃的位置裁剪回放数据的逐块条目。无论组装移除什么，`blocks()` 与 `replayState` 都不可能不一致。 + ```ts public-api /** * Incrementally assembles raw {@link StreamChunk}s into complete @@ -302,8 +327,12 @@ declare class BlockAssembler { get usage(): TokenUsage | undefined; /** Finish reason from the `finish` chunk; `{kind: 'stop'}` when the stream ended without one. */ get finish(): FinishReason; - /** Adapter-private replay state from the terminal finish chunk, if any. */ - get replayState(): unknown; + /** + * Replay metadata from the terminal finish chunk, if any, with per-block + * entries pruned in step with {@link blocks}. Undefined when the envelope's + * entries do not align with the emitted blocks. + */ + get replayState(): ReplayEnvelope | undefined; /** * The assembled assistant message. * @param source - producer attribution for the assembled message. diff --git a/examples/acp-agent/tests/acp.snapshot.ts b/examples/acp-agent/tests/acp.snapshot.ts index db4a2b5d2f..87ef397ee2 100644 --- a/examples/acp-agent/tests/acp.snapshot.ts +++ b/examples/acp-agent/tests/acp.snapshot.ts @@ -348,6 +348,13 @@ const SCENARIOS: Scenario[] = [ // reply, and a clean completed retry turn. Its overlay only pins a deterministic // 1 ms zero-jitter delay, so it shares the default header class. { name: 'empty-response-retry', hasModelTurn: true, recorded: false, configPath: RETRY_CONFIG }, + // Keyless, authored (like error-finish): a live model cannot be coaxed into + // a deterministic mid-tool-call output-limit truncation. Turn 1's script ends + // at `max-tokens` with an unfinished tool call and adapter replay metadata for + // both blocks; the durable assistant/message pins assembly dropping the tool + // call AND pruning its per-block replay entry in the same decision, and turn 2 + // proves the session continues past the truncated step. + { name: 'max-tokens-continue', hasModelTurn: true, recorded: false }, // Keyless, authored (like error-finish/cancel): deterministically forcing a // LIVE model to repeat one call three times is not a stable recording, so // the fixture scripts five identical todo_write calls and pins BOTH reminder diff --git a/examples/acp-agent/tests/snapshots/max-tokens-continue/input.json b/examples/acp-agent/tests/snapshots/max-tokens-continue/input.json new file mode 100644 index 0000000000..ebd0c642bb --- /dev/null +++ b/examples/acp-agent/tests/snapshots/max-tokens-continue/input.json @@ -0,0 +1,8 @@ +{ + "steps": [ + { "op": "initialize" }, + { "op": "newSession" }, + { "op": "prompt", "text": "This turn is cut off at the output limit while calling a tool." }, + { "op": "prompt", "text": "Continue: summarize what happened without retrying the tool." } + ] +} diff --git a/examples/acp-agent/tests/snapshots/max-tokens-continue/session.jsonl b/examples/acp-agent/tests/snapshots/max-tokens-continue/session.jsonl new file mode 100644 index 0000000000..6e2f7a6d1c --- /dev/null +++ b/examples/acp-agent/tests/snapshots/max-tokens-continue/session.jsonl @@ -0,0 +1,33 @@ +{"type":"session","version":0,"id":"7f1c9a04-5b52-4a7e-9a63-1d2ab7c90d11","createdAt":1786348800000,"cwd":"{{cwd}}","delegationDepth":0} +{"type":"agent/inbox/spliced","seq":0,"time":1786348800001,"data":{"target":"next-turn","start":0,"inserted":[{"content":[{"type":"text","text":"This turn is cut off at the output limit while calling a tool."}],"source":{"kind":"user"},"role":"user","id":"3a6a5c9e-0f9c-4c8f-9f57-6f2f7f3d5a01"}]}} +{"type":"turn/start","seq":1,"time":1786348800002,"data":{"turn":1}} +{"type":"agent/inbox/spliced","seq":2,"time":1786348800002,"data":{"target":"next-turn","start":0,"removedCount":1,"inserted":[]}} +{"type":"step/start","seq":3,"time":1786348800003,"data":{"turn":1,"step":1}} +{"type":"user/message","seq":4,"time":1786348800004,"data":{"content":[{"type":"text","text":"This turn is cut off at the output limit while calling a tool."}],"source":{"kind":"user"},"role":"user","id":"3a6a5c9e-0f9c-4c8f-9f57-6f2f7f3d5a01"},"surfaceOp":"append"} +{"type":"user/message","seq":5,"time":1786348800005,"data":{"content":[{"type":"text","text":"Current runtime context. This snapshot supersedes earlier runtime-context snapshots.\n\nCurrent DSH file policy: danger-full-access. The DSH file sandbox does not restrict file modifications by available operations.\n\nApproval prompts are disabled in this session: actions that require approval are rejected automatically — do not request sandbox escalation (do not set `sandbox_permissions`)."}],"source":{"kind":"plugin","plugin":"@deepseek-ai/dsh-system-prompt","form":"snapshot","sections":[{"name":"sandbox:policy","text":"Current DSH file policy: danger-full-access. The DSH file sandbox does not restrict file modifications by available operations."},{"name":"approval:policy","text":"Approval prompts are disabled in this session: actions that require approval are rejected automatically — do not request sandbox escalation (do not set `sandbox_permissions`)."}]},"role":"user","id":"5b7f2d1c-9c44-4c58-8a3e-2f6f8b9d4c02"},"surfaceOp":"append"} +{"type":"session/title","seq":6,"time":1786348800005,"data":{"title":"This turn is cut off","messageSeqs":[4],"source":{"kind":"fallback"}}} +{"type":"request/header","seq":7,"time":1786348800006,"data":{"header":{"config":{"provider":"deepseek-official","model":"deepseek-v4-flash"},"system":"{{system}}","tools":"{{tools}}"},"reason":"initial"}} +{"type":"request/context","seq":8,"time":1786348800006,"data":{"provider":"deepseek-official","model":"deepseek-v4-flash"}} +{"type":"assistant/chunk","seq":9,"time":1786348800010,"data":{"turn":1,"step":1,"chunk":{"type":"block-start","index":0,"blockType":"text"}}} +{"type":"assistant/chunk","seq":10,"time":1786348800011,"data":{"turn":1,"step":1,"chunk":{"type":"text-delta","index":0,"text":"Starting the write now."}}} +{"type":"assistant/chunk","seq":11,"time":1786348800012,"data":{"turn":1,"step":1,"chunk":{"type":"block-end","index":0,"block":{"type":"text","text":"Starting the write now."}}}} +{"type":"assistant/chunk","seq":12,"time":1786348800013,"data":{"turn":1,"step":1,"chunk":{"type":"block-start","index":1,"blockType":"tool-call"}}} +{"type":"assistant/chunk","seq":13,"time":1786348800014,"data":{"turn":1,"step":1,"chunk":{"type":"tool-call-delta","index":1,"id":"call-cut","name":"bash","argumentsDelta":"{\"command\":\"echo demo > "}}} +{"type":"assistant/chunk","seq":14,"time":1786348800015,"data":{"turn":1,"step":1,"chunk":{"type":"usage","usage":{"inputTokens":2864,"outputTokens":12}}}} +{"type":"assistant/chunk","seq":15,"time":1786348800016,"data":{"turn":1,"step":1,"chunk":{"type":"finish","reason":{"kind":"max-tokens"},"replayState":{"response":{"kind":"pi-ai","version":2,"api":"openai-completions","provider":"deepseek-official","model":"deepseek-v4-flash","stopReason":"length"},"blocks":[{"type":"text"},{"type":"tool-call"}]}}}} +{"type":"assistant/message","seq":16,"time":1786348800016,"data":{"turn":1,"step":1,"message":{"role":"assistant","content":[{"type":"text","text":"Starting the write now."}],"source":{"kind":"model","provider":"deepseek-official","model":"deepseek-v4-flash","replayState":{"response":{"kind":"pi-ai","version":2,"api":"openai-completions","provider":"deepseek-official","model":"deepseek-v4-flash","stopReason":"length"},"blocks":[{"type":"text"}]}},"id":"9d5f7c2a-1e63-4d6b-8f14-7a2c5e9b3d03"},"usage":{"inputTokens":2864,"outputTokens":12}},"sourceEventSeqs":[9,10,11,12,13,14,15],"surfaceOp":"append"} +{"type":"step/end","seq":17,"time":1786348800016,"data":{"turn":1,"step":1}} +{"type":"turn/end","seq":18,"time":1786348800016,"data":{"turn":1,"reason":{"kind":"max-tokens"}}} +{"type":"agent/inbox/spliced","seq":19,"time":1786348800020,"data":{"target":"next-turn","start":0,"inserted":[{"content":[{"type":"text","text":"Continue: summarize what happened without retrying the tool."}],"source":{"kind":"user"},"role":"user","id":"1c8e6b4f-3d27-4a91-b5c8-9e4f7a2d6c04"}]}} +{"type":"turn/start","seq":20,"time":1786348800021,"data":{"turn":2}} +{"type":"agent/inbox/spliced","seq":21,"time":1786348800021,"data":{"target":"next-turn","start":0,"removedCount":1,"inserted":[]}} +{"type":"step/start","seq":22,"time":1786348800022,"data":{"turn":2,"step":1}} +{"type":"user/message","seq":23,"time":1786348800023,"data":{"content":[{"type":"text","text":"Continue: summarize what happened without retrying the tool."}],"source":{"kind":"user"},"role":"user","id":"1c8e6b4f-3d27-4a91-b5c8-9e4f7a2d6c04"},"surfaceOp":"append"} +{"type":"assistant/chunk","seq":24,"time":1786348800030,"data":{"turn":2,"step":1,"chunk":{"type":"block-start","index":0,"blockType":"text"}}} +{"type":"assistant/chunk","seq":25,"time":1786348800031,"data":{"turn":2,"step":1,"chunk":{"type":"text-delta","index":0,"text":"The previous reply hit the output limit while a tool call was still streaming, so that call was discarded and no tool ran."}}} +{"type":"assistant/chunk","seq":26,"time":1786348800032,"data":{"turn":2,"step":1,"chunk":{"type":"block-end","index":0,"block":{"type":"text","text":"The previous reply hit the output limit while a tool call was still streaming, so that call was discarded and no tool ran."}}}} +{"type":"assistant/chunk","seq":27,"time":1786348800033,"data":{"turn":2,"step":1,"chunk":{"type":"usage","usage":{"inputTokens":64,"outputTokens":28}}}} +{"type":"assistant/chunk","seq":28,"time":1786348800034,"data":{"turn":2,"step":1,"chunk":{"type":"finish","reason":{"kind":"stop"}}}} +{"type":"assistant/message","seq":29,"time":1786348800034,"data":{"turn":2,"step":1,"message":{"role":"assistant","content":[{"type":"text","text":"The previous reply hit the output limit while a tool call was still streaming, so that call was discarded and no tool ran."}],"source":{"kind":"model","provider":"deepseek-official","model":"deepseek-v4-flash"},"id":"7e3d9f6b-5a18-4c72-9b4e-1f8c6d2a7e05"},"usage":{"inputTokens":64,"outputTokens":28}},"sourceEventSeqs":[24,25,26,27,28],"surfaceOp":"append"} +{"type":"step/end","seq":30,"time":1786348800034,"data":{"turn":2,"step":1}} +{"type":"turn/end","seq":31,"time":1786348800034,"data":{"turn":2,"reason":{"kind":"completed"}}} diff --git a/examples/acp-agent/tests/snapshots/max-tokens-continue/stdout.expected.jsonl b/examples/acp-agent/tests/snapshots/max-tokens-continue/stdout.expected.jsonl new file mode 100644 index 0000000000..bf555a8d1c --- /dev/null +++ b/examples/acp-agent/tests/snapshots/max-tokens-continue/stdout.expected.jsonl @@ -0,0 +1,6 @@ +{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1,"agentInfo":{"name":"deepseek-harness-acp","version":"0.0.1"},"agentCapabilities":{"promptCapabilities":{"image":false,"audio":false,"embeddedContext":false}},"authMethods":[]}} +{"jsonrpc":"2.0","id":2,"result":{"sessionId":"{{sessionId}}"}} +{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"Starting the write now."}}}} +{"jsonrpc":"2.0","id":3,"result":{"stopReason":"end_turn"}} +{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":"{{sessionId}}","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"The previous reply hit the output limit while a tool call was still streaming, so that call was discarded and no tool ran."}}}} +{"jsonrpc":"2.0","id":4,"result":{"stopReason":"end_turn"}} diff --git a/packages/core/agent-loop/tests/contract-regressions.spec.ts b/packages/core/agent-loop/tests/contract-regressions.spec.ts index acdc66865e..c085dc7de0 100644 --- a/packages/core/agent-loop/tests/contract-regressions.spec.ts +++ b/packages/core/agent-loop/tests/contract-regressions.spec.ts @@ -62,7 +62,7 @@ function inboxText(message: UserMessage): string { describe('assistant replay provider and model fields', () => { it('records adapter replay state with the assembled assistant content', async () => { const response = textResponse('unchanged') - const replayState = { private: 'state' } + const replayState = { response: { private: 'state' }, blocks: ['block-meta'] } response[response.length - 1] = { type: 'finish', reason: { kind: 'stop' }, replayState } const adapter = new MockAdapter([response]) const ctx = await harness(adapter) diff --git a/packages/core/agent-loop/tests/loop.spec.ts b/packages/core/agent-loop/tests/loop.spec.ts index ca4a5309c9..2105b86f42 100644 --- a/packages/core/agent-loop/tests/loop.spec.ts +++ b/packages/core/agent-loop/tests/loop.spec.ts @@ -1192,15 +1192,29 @@ describe('agent loop', () => { { type: 'block-end', index: 0, block: { type: 'text', text: 'partial text' } }, { type: 'block-start', index: 1, blockType: 'tool-call' }, { type: 'tool-call-delta', index: 1, id: callId, name: 'echo', argumentsDelta: '{"text"' }, - { type: 'finish', reason: { kind: 'max-tokens' } }, - ]]) + { + type: 'finish', + reason: { kind: 'max-tokens' }, + replayState: { response: { responseId: 'resp-1' }, blocks: ['text-meta', 'tool-meta'] }, + }, + ], textResponse('continued')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create(SessionId('a1'), { provider: 'mock', model: 'mock' }) send(agent, 'go') await waitForIdle(ctx, agent) + send(agent, 'continue') + await waitForIdle(ctx, agent) expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false) + // The follow-up request replays the truncated message with its replay + // metadata pruned in step with the dropped tool call. + expect(adapter.requests[1]?.messages[1]?.source).toEqual({ + kind: 'model', + provider: 'mock', + model: 'mock', + replayState: { response: { responseId: 'resp-1' }, blocks: ['text-meta'] }, + }) expect(agent.session.deriveMessages()).toEqual([ { id: expect.any(String) as unknown, @@ -1212,6 +1226,23 @@ describe('agent loop', () => { id: expect.any(String) as unknown, role: 'assistant', content: [{ type: 'text', text: 'partial text' }], + source: { + kind: 'model', + provider: 'mock', + model: 'mock', + replayState: { response: { responseId: 'resp-1' }, blocks: ['text-meta'] }, + }, + }, + { + id: expect.any(String) as unknown, + role: 'user', + content: [{ type: 'text', text: 'continue' }], + source: { kind: 'user' }, + }, + { + id: expect.any(String) as unknown, + role: 'assistant', + content: [{ type: 'text', text: 'continued' }], source: { kind: 'model', provider: 'mock', model: 'mock' }, }, ]) diff --git a/packages/extensions/tool-cordis/src/api-catalog.ts b/packages/extensions/tool-cordis/src/api-catalog.ts index 5a812806da..7fb624d21f 100644 --- a/packages/extensions/tool-cordis/src/api-catalog.ts +++ b/packages/extensions/tool-cordis/src/api-catalog.ts @@ -3577,6 +3577,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [ name: 'RedactedSecret', declaration: 'export interface RedactedSecret {\n path: string[];\n set: boolean;\n}', }, + { + name: 'ReplayEnvelope', + declaration: 'export interface ReplayEnvelope {\n response: unknown;\n blocks?: readonly unknown[];\n}', + }, { name: 'RequestContext', declaration: 'export interface RequestContext {\n provider: string;\n model: string;\n contextWindow?: number;\n}', @@ -4091,7 +4095,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [ }, { name: 'StreamChunk', - declaration: 'export type StreamChunk = {\n type: \'block-start\';\n index: number;\n blockType: ContentBlockType;\n} | {\n type: \'text-delta\';\n index: number;\n text: string;\n} | {\n type: \'reasoning-delta\';\n index: number;\n text: string;\n} | {\n type: \'tool-call-delta\';\n index: number;\n id: CallId;\n name?: string;\n argumentsDelta: string;\n} | {\n type: \'block-end\';\n index: number;\n block: ContentBlock;\n} | {\n type: \'usage\';\n usage: TokenUsage;\n} | {\n type: \'finish\';\n reason: FinishReason;\n replayState?: unknown;\n};', + declaration: 'export type StreamChunk = {\n type: \'block-start\';\n index: number;\n blockType: ContentBlockType;\n} | {\n type: \'text-delta\';\n index: number;\n text: string;\n} | {\n type: \'reasoning-delta\';\n index: number;\n text: string;\n} | {\n type: \'tool-call-delta\';\n index: number;\n id: CallId;\n name?: string;\n argumentsDelta: string;\n} | {\n type: \'block-end\';\n index: number;\n block: ContentBlock;\n} | {\n type: \'usage\';\n usage: TokenUsage;\n} | {\n type: \'finish\';\n reason: FinishReason;\n replayState?: ReplayEnvelope;\n};', }, { name: 'SubagentCapabilities', diff --git a/packages/llm/llm-pi-ai/README.i18n.yaml b/packages/llm/llm-pi-ai/README.i18n.yaml index 31e6ac3b8a..553b7d4557 100644 --- a/packages/llm/llm-pi-ai/README.i18n.yaml +++ b/packages/llm/llm-pi-ai/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/llm/llm-pi-ai/README.md -README.md: 6120f8d982c6d475cd508e6cf9e41cabfc9ba159 -README.zh.md: 4b47976c6c6c67968b5b93edbdfd5dfa9530eb1d +README.md: d775e72616822ce0deee063ac0f3fc453af1a126 +README.zh.md: 621d67d1c181c6d4c78ea0078f521acccce92653 diff --git a/packages/llm/llm-pi-ai/README.md b/packages/llm/llm-pi-ai/README.md index 6120f8d982..d775e72616 100644 --- a/packages/llm/llm-pi-ai/README.md +++ b/packages/llm/llm-pi-ai/README.md @@ -137,9 +137,9 @@ Credentials never enter that collection. The harness resolves a route's key thro The selected model descriptor supplies the protocol implementation. This includes native API differences such as OpenAI models whose descriptor uses the Responses API rather than Chat Completions; the harness adapter does not hardcode endpoint selection by model name. -Successful assistant responses store a versioned, lossless-JSON replay state beside the provider and model that produced them. At request time, `LlmRuntime` passes replay state only when the historical provider route and target provider route are currently owned by this same `PiAiAdapter` instance. The adapter validates the state and restores pi-ai response ids and provider signatures even when the target provider or model changes; pi-ai then decides which metadata its target API can reuse. History without replay state is translated as foreign provider-neutral content and never impersonates a native pi-ai response. +Successful assistant responses store a versioned, lossless-JSON replay state beside the provider and model that produced them, as a `ReplayEnvelope`: a response-level half (kind, version, API, route, response ids, native stop reason) plus one per-block entry per streamed block carrying that block's signatures. The per-block alignment is what `BlockAssembler` prunes when assembly drops a block (a `max-tokens` tool call), so the stored entries always describe the stored content — the retained blocks keep their signatures. At request time, `LlmRuntime` passes replay state only when the historical provider route and target provider route are currently owned by this same `PiAiAdapter` instance. The adapter validates the state and restores pi-ai response ids and provider signatures even when the target provider or model changes; pi-ai then decides which metadata its target API can reuse. History without replay state is translated as foreign provider-neutral content and never impersonates a native pi-ai response. -If a listener rewrites assembled assistant content, the loop drops replay state before logging the message because its provider metadata no longer describes the content. Invalid versions, malformed metadata, provider/model mismatches between the message and replay state, and content/block mismatches fail explicitly with `LlmError('INVALID_REPLAY_STATE')`. +Durable content is the authoritative record; replay state only restores native fidelity. A stored state this build cannot use — another adapter's kind, another version (including the flat pre-envelope form older logs carry), malformed metadata, provider/model mismatches between the message and replay state, or content/block mismatches — degrades that one assistant message to the same foreign provider-neutral conversion instead of failing the request, and the plugin logs the `INVALID_REPLAY_STATE` diagnostic through its `onReplayDegrade` hook. ## Vocabulary differences diff --git a/packages/llm/llm-pi-ai/README.zh.md b/packages/llm/llm-pi-ai/README.zh.md index 4b47976c6c..621d67d1c1 100644 --- a/packages/llm/llm-pi-ai/README.zh.md +++ b/packages/llm/llm-pi-ai/README.zh.md @@ -138,9 +138,9 @@ profile 的 `models` 列表是*替换*该路由已安装 catalog，而不是扩 所选模型 descriptor 提供协议实现。这包括原生 API 差异，例如 descriptor 使用 Responses API 而非 Chat Completions 的 OpenAI 模型；harness 适配器不会按模型名称硬编码端点选择。 -成功的 assistant 响应会将经版本化的无损 JSON 回放状态与生成该响应的提供方和模型一同存储。请求时，`LlmRuntime` 只有在历史提供方路由与目标提供方路由当前由同一个 `PiAiAdapter` 实例拥有时，才会传递回放状态。即使目标提供方或模型改变，适配器也会验证状态并恢复 pi-ai 响应 id 与提供方 signature；随后由 pi-ai 判定目标 API 可以复用哪些元数据。没有回放状态的历史会被转换为外来的、与提供方无关的内容，绝不伪装为原生 pi-ai 响应。 +成功的 assistant 响应会将经版本化的无损 JSON 回放状态与生成该响应的提供方和模型一同存储，其形式是 `ReplayEnvelope`：一个响应级半区（kind、版本、API、路由、响应 id、原生停止原因），加上每个流式块一条、携带该块 signature 的逐块条目。逐块对齐正是 `BlockAssembler` 在组装丢弃某个块（`max-tokens` 下的工具调用）时裁剪的对象，因此存储的条目始终描述存储的内容——保留的块保有其 signature。请求时，`LlmRuntime` 只有在历史提供方路由与目标提供方路由当前由同一个 `PiAiAdapter` 实例拥有时，才会传递回放状态。即使目标提供方或模型改变，适配器也会验证状态并恢复 pi-ai 响应 id 与提供方 signature；随后由 pi-ai 判定目标 API 可以复用哪些元数据。没有回放状态的历史会被转换为外来的、与提供方无关的内容，绝不伪装为原生 pi-ai 响应。 -如果 listener 改写已组装 assistant 内容，loop 会在记录消息前丢弃回放状态，因为其提供方元数据不再描述该内容。无效版本、格式错误元数据、消息与回放状态之间的提供方／模型不匹配，以及内容／块不匹配都会显式以 `LlmError('INVALID_REPLAY_STATE')` 失败。 +持久化内容是权威记录；回放状态只负责恢复原生保真度。当前构建无法使用的已存状态——其他适配器的 kind、其他版本（包括旧日志携带的平铺前信封形式）、格式错误的元数据、消息与回放状态之间的提供方／模型不匹配，或内容／块不匹配——会把这一条 assistant 消息降级为同样的外来提供方无关转换而不是让请求失败，插件通过其 `onReplayDegrade` 钩子记录 `INVALID_REPLAY_STATE` 诊断。 ## 词汇差异 diff --git a/packages/llm/llm-pi-ai/src/adapter.ts b/packages/llm/llm-pi-ai/src/adapter.ts index 66964c5339..ab1c784351 100644 --- a/packages/llm/llm-pi-ai/src/adapter.ts +++ b/packages/llm/llm-pi-ai/src/adapter.ts @@ -76,6 +76,11 @@ export interface PiAiAdapterOptions { resolveApiKey: (provider: string, profile: ResolvedPiAiProviderProfile) => Promise /** Resolve the optional durable attachment service at request time. */ resolveAttachments?: () => AttachmentStore | undefined + /** + * Observe one assistant history message degrading to provider-neutral + * conversion because its stored replay state is unusable by this build. + */ + onReplayDegrade?: (detail: { provider: string; model: string; reason: string }) => void } /** Copy profile stream knobs into pi-ai's common option vocabulary. */ @@ -307,9 +312,12 @@ export class PiAiAdapter extends LlmAdapter { if (containsImage && attachments === undefined) { throw new LlmError('pi-ai image input requires the durable attachment service', 'UNSUPPORTED_CONTENT') } + const onReplayDegrade = (reason: string): void => { + this.config.onReplayDegrade?.({ provider: options.provider, model: options.model, reason }) + } const context = attachments === undefined - ? toPiContext(options) - : await toPiContext(options, attachments) + ? toPiContext(options, undefined, onReplayDegrade) + : await toPiContext(options, attachments, onReplayDegrade) const events = snapshot.models.streamSimple(model, context, { ...profileOptions(profile, reasoning, apiKey), ...options.temperature === undefined ? {} : { temperature: options.temperature }, diff --git a/packages/llm/llm-pi-ai/src/context.ts b/packages/llm/llm-pi-ai/src/context.ts index 678820510e..dcbaabc815 100644 --- a/packages/llm/llm-pi-ai/src/context.ts +++ b/packages/llm/llm-pi-ai/src/context.ts @@ -84,7 +84,7 @@ function piContext(options: GenerateOptions, messages: PiMessage[]): PiContext { } } -function textOnlyContext(options: GenerateOptions): PiContext { +function textOnlyContext(options: GenerateOptions, onReplayDegrade?: (reason: string) => void): PiContext { const toolNames = new Map() const messages: PiMessage[] = [] for (const message of options.messages) { @@ -96,7 +96,7 @@ function textOnlyContext(options: GenerateOptions): PiContext { continue } if (message.role === 'assistant') { - const assistant = toPiAssistant(message) + const assistant = toPiAssistant(message, onReplayDegrade) for (const block of assistant.content) if (block.type === 'toolCall') toolNames.set(CallId(block.id), block.name) messages.push(assistant) continue @@ -125,22 +125,43 @@ function textOnlyContext(options: GenerateOptions): PiContext { * Convert text-only harness history to a synchronous pi-ai Context. Tool * result names are recovered from preceding assistant tool calls. * @param options - the harness request; `options.system` maps to pi-ai's single `systemPrompt` slot. + * @param attachments - absent; selects the synchronous conversion. + * @param onReplayDegrade - forwarded to {@link toPiAssistant} for each assistant message. * @returns the pi-ai context; `tools` is omitted when the request declares none. */ -export function toPiContext(options: GenerateOptions): PiContext +export function toPiContext( + options: GenerateOptions, + attachments?: undefined, + onReplayDegrade?: (reason: string) => void, +): PiContext /** * Convert harness history to a pi-ai Context while resolving durable images. * Tool result names are recovered from preceding assistant tool calls. * @param options - the harness request; `options.system` maps to pi-ai's single `systemPrompt` slot. * @param attachments - durable byte resolver for image references. + * @param onReplayDegrade - forwarded to {@link toPiAssistant} for each assistant message. * @returns the asynchronously resolved pi-ai context. */ -export function toPiContext(options: GenerateOptions, attachments: AttachmentStore): Promise -export function toPiContext(options: GenerateOptions, attachments?: AttachmentStore): PiContext | Promise { - return attachments === undefined ? textOnlyContext(options) : toPiContextWithImages(options, attachments) +export function toPiContext( + options: GenerateOptions, + attachments: AttachmentStore, + onReplayDegrade?: (reason: string) => void, +): Promise +export function toPiContext( + options: GenerateOptions, + attachments?: AttachmentStore, + onReplayDegrade?: (reason: string) => void, +): PiContext | Promise { + return attachments === undefined + ? textOnlyContext(options, onReplayDegrade) + : toPiContextWithImages(options, attachments, onReplayDegrade) } -async function toPiContextWithImages(options: GenerateOptions, attachments: AttachmentStore): Promise { +async function toPiContextWithImages( + options: GenerateOptions, + attachments: AttachmentStore, + onReplayDegrade?: (reason: string) => void, +): Promise { const toolNames = new Map() const messages: PiMessage[] = [] @@ -156,7 +177,7 @@ async function toPiContextWithImages(options: GenerateOptions, attachments: Atta continue } if (message.role === 'assistant') { - const assistant = toPiAssistant(message) + const assistant = toPiAssistant(message, onReplayDegrade) for (const block of assistant.content) { if (block.type === 'toolCall') toolNames.set(CallId(block.id), block.name) } diff --git a/packages/llm/llm-pi-ai/src/index.ts b/packages/llm/llm-pi-ai/src/index.ts index 2e550771fc..1bbeec79db 100644 --- a/packages/llm/llm-pi-ai/src/index.ts +++ b/packages/llm/llm-pi-ai/src/index.ts @@ -201,6 +201,12 @@ export function apply(ctx: Context, config: Config): void { profiles, resolveApiKey, resolveAttachments: () => ctx.get('attachments'), + onReplayDegrade: ({ provider, model, reason }) => { + ctx.logger.warn( + `llm-pi-ai: unusable replay state on assistant history for route "${provider}/${model}";` + + ` sending that message as provider-neutral content (${reason})`, + ) + }, }) // The full installed catalog is configurable from the moment the plugin // mounts — dormant or not — so configuration surfaces can offer every diff --git a/packages/llm/llm-pi-ai/src/replay.ts b/packages/llm/llm-pi-ai/src/replay.ts index 10a39c655b..aa9d542e33 100644 --- a/packages/llm/llm-pi-ai/src/replay.ts +++ b/packages/llm/llm-pi-ai/src/replay.ts @@ -9,24 +9,30 @@ */ import { LlmError } from '@deepseek-ai/dsh-llm' -import type { Message, ModelMessageSource } from '@deepseek-ai/dsh-llm' +import type { Message, ModelMessageSource, ReplayEnvelope } from '@deepseek-ai/dsh-llm' import type { Api, AssistantMessage, Usage as PiUsage } from '@earendil-works/pi-ai' -type PiAiReplayBlock = +/** Per-block half of the pi-ai replay envelope, one entry per content block. */ +export type PiAiReplayBlock = | { type: 'text'; textSignature?: string } | { type: 'reasoning'; thinkingSignature?: string; redacted?: boolean } | { type: 'tool-call'; thoughtSignature?: string } -/** Versioned adapter-private projection required to replay a pi-ai response. */ -export interface PiAiReplayState { +/** Versioned response-level half of the pi-ai replay envelope. */ +export interface PiAiReplayResponse { kind: 'pi-ai' - version: 1 + version: 2 api: Api provider: string model: string responseModel?: string responseId?: string stopReason: AssistantMessage['stopReason'] +} + +/** The validated halves of one pi-ai replay envelope. */ +interface PiAiReplayState { + response: PiAiReplayResponse blocks: PiAiReplayBlock[] } @@ -57,19 +63,25 @@ function emptyPiUsage(): PiUsage { /** * Project a successful pi-ai response into the minimal durable replay state. + * The per-block half is index-aligned with the streamed blocks (pi-ai content + * order), so `BlockAssembler` prunes an entry with its block whenever assembly + * removes one. * @param message - completed native pi-ai assistant response. * @returns the versioned lossless-JSON replay projection. */ -export function toPiReplayState(message: AssistantMessage): PiAiReplayState { - return { +export function toPiReplayState(message: AssistantMessage): ReplayEnvelope { + const response: PiAiReplayResponse = { kind: 'pi-ai', - version: 1, + version: 2, api: message.api, provider: message.provider, model: message.model, ...message.responseModel === undefined ? {} : { responseModel: message.responseModel }, ...message.responseId === undefined ? {} : { responseId: message.responseId }, stopReason: message.stopReason, + } + return { + response, blocks: message.content.map((block): PiAiReplayBlock => { switch (block.type) { case 'text': return { @@ -94,22 +106,26 @@ function invalidReplay(message: string): never { throw new LlmError(`invalid pi-ai replay state: ${message}`, 'INVALID_REPLAY_STATE') } -/** Validate the adapter-private state before it reaches pi-ai. */ +/** Validate the durable adapter-private envelope before it reaches pi-ai. */ function readReplayState(value: unknown): PiAiReplayState { - if (typeof value !== 'object' || value === null || Array.isArray(value)) return invalidReplay('expected an object') - const state = value as Record - if (state['kind'] !== 'pi-ai') return invalidReplay('unknown state kind') - if (state['version'] !== 1) return invalidReplay(`unsupported version ${String(state['version'])}`) + if (typeof value !== 'object' || value === null || Array.isArray(value)) return invalidReplay('expected a replay envelope') + const envelope = value as Record + const rawResponse = envelope['response'] + if (typeof rawResponse !== 'object' || rawResponse === null || Array.isArray(rawResponse)) return invalidReplay('expected a response object') + const response = rawResponse as Record + if (response['kind'] !== 'pi-ai') return invalidReplay('unknown state kind') + if (response['version'] !== 2) return invalidReplay(`unsupported version ${String(response['version'])}`) for (const key of ['api', 'provider', 'model'] as const) { - if (typeof state[key] !== 'string' || state[key].length === 0) return invalidReplay(`${key} must be a non-empty string`) + if (typeof response[key] !== 'string' || response[key].length === 0) return invalidReplay(`${key} must be a non-empty string`) } - if (!['stop', 'length', 'toolUse', 'error', 'aborted'].includes(String(state['stopReason']))) { + if (!['stop', 'length', 'toolUse', 'error', 'aborted'].includes(String(response['stopReason']))) { return invalidReplay('unknown stopReason') } - if (state['responseModel'] !== undefined && typeof state['responseModel'] !== 'string') return invalidReplay('responseModel must be a string') - if (state['responseId'] !== undefined && typeof state['responseId'] !== 'string') return invalidReplay('responseId must be a string') - if (!Array.isArray(state['blocks'])) return invalidReplay('blocks must be an array') - for (const [index, value] of state['blocks'].entries()) { + if (response['responseModel'] !== undefined && typeof response['responseModel'] !== 'string') return invalidReplay('responseModel must be a string') + if (response['responseId'] !== undefined && typeof response['responseId'] !== 'string') return invalidReplay('responseId must be a string') + const blocks = envelope['blocks'] + if (!Array.isArray(blocks)) return invalidReplay('blocks must be an array') + for (const [index, value] of blocks.entries()) { if (typeof value !== 'object' || value === null || Array.isArray(value)) return invalidReplay(`block ${index} must be an object`) const block = value as Record if (!['text', 'reasoning', 'tool-call'].includes(String(block['type']))) return invalidReplay(`block ${index} has an unknown type`) @@ -118,7 +134,10 @@ function readReplayState(value: unknown): PiAiReplayState { } if (block['redacted'] !== undefined && typeof block['redacted'] !== 'boolean') return invalidReplay(`block ${index} redacted must be boolean`) } - return state as unknown as PiAiReplayState + return { + response: response as unknown as PiAiReplayResponse, + blocks: blocks as PiAiReplayBlock[], + } } /** Convert provider-neutral blocks without trusting them as same-model replay. */ @@ -159,8 +178,8 @@ function foreignAssistant(message: Message): AssistantMessage { /** Recombine durable Harness content with validated pi-ai replay metadata. */ function replayedAssistant(message: Message, source: ModelMessageSource, rawState: unknown): AssistantMessage { const state = readReplayState(rawState) - if (state.provider !== source.provider) return invalidReplay('provider does not match assistant source') - if (state.model !== source.model) return invalidReplay('model does not match assistant source') + if (state.response.provider !== source.provider) return invalidReplay('provider does not match assistant source') + if (state.response.model !== source.model) return invalidReplay('model does not match assistant source') if (state.blocks.length !== message.content.length) return invalidReplay('block count does not match assistant content') const content: AssistantMessage['content'] = message.content.map((block, index) => { const replay = state.blocks[index] @@ -191,25 +210,40 @@ function replayedAssistant(message: Message, source: ModelMessageSource, rawStat return { role: 'assistant', content, - api: state.api, - provider: state.provider, - model: state.model, - ...state.responseModel === undefined ? {} : { responseModel: state.responseModel }, - ...state.responseId === undefined ? {} : { responseId: state.responseId }, + api: state.response.api, + provider: state.response.provider, + model: state.response.model, + ...state.response.responseModel === undefined ? {} : { responseModel: state.response.responseModel }, + ...state.response.responseId === undefined ? {} : { responseId: state.response.responseId }, usage: emptyPiUsage(), - stopReason: state.stopReason, + stopReason: state.response.stopReason, timestamp: 0, } } /** * Convert one durable Harness assistant message into pi-ai history. + * + * Durable content is the authoritative record; replay metadata only restores + * native fidelity (ids, signatures). A replay state this build cannot use — + * another adapter's kind, another version, a malformed value, or metadata that + * no longer matches the content — therefore degrades the one message to + * provider-neutral history instead of failing the request. * @param message - assistant content with required source and optional adapter-owned replay metadata. + * @param onDegrade - called with the diagnostic reason when an unusable replay + * state falls back to provider-neutral conversion. * @returns a native pi-ai assistant message reconstructed from durable content. */ -export function toPiAssistant(message: Message): AssistantMessage { +export function toPiAssistant(message: Message, onDegrade?: (reason: string) => void): AssistantMessage { const source = message.source - return source.kind !== 'model' || source.replayState === undefined - ? foreignAssistant(message) - : replayedAssistant(message, source, source.replayState) + if (source.kind !== 'model' || source.replayState === undefined) return foreignAssistant(message) + try { + return replayedAssistant(message, source, source.replayState) + } catch (error: unknown) { + /* v8 ignore next -- replayedAssistant throws only INVALID_REPLAY_STATE LlmErrors today; the + guard keeps a future non-replay failure loud instead of silently degrading it */ + if (!(error instanceof LlmError) || error.code !== 'INVALID_REPLAY_STATE') throw error + onDegrade?.(error.message) + return foreignAssistant(message) + } } diff --git a/packages/llm/llm-pi-ai/tests/convert.spec.ts b/packages/llm/llm-pi-ai/tests/convert.spec.ts index 5af58e6630..1a42e4b085 100644 --- a/packages/llm/llm-pi-ai/tests/convert.spec.ts +++ b/packages/llm/llm-pi-ai/tests/convert.spec.ts @@ -1,7 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import { AttachmentId } from '@deepseek-ai/dsh-attachment' import type { AttachmentStore } from '@deepseek-ai/dsh-attachment' -import { createUserMessage, CallId, CONTEXT_WINDOW_EXCEEDED_CODE, EMPTY_RESPONSE_CODE, LlmError, createMessage } from '@deepseek-ai/dsh-llm' +import { createUserMessage, CallId, CONTEXT_WINDOW_EXCEEDED_CODE, EMPTY_RESPONSE_CODE, createMessage } from '@deepseek-ai/dsh-llm' import type { ContentBlock, StreamChunk } from '@deepseek-ai/dsh-llm' import type { AssistantMessage, AssistantMessageEvent, Usage } from '@earendil-works/pi-ai' import { toPiContext } from '../src/context.ts' @@ -415,35 +415,68 @@ describe('toPiContext', () => { expect(context.messages[0]).not.toHaveProperty('responseId') }) - it('rejects unsupported replay-state versions with a stable error code', () => { - try { - toPiContext({ - provider: 'deepseek', - model: 'm', - messages: [createMessage({ - role: 'assistant', - content: [{ type: 'text', text: 'done' }], - source: { - kind: 'model', - ...{ - provider: 'deepseek', - model: 'old', - replayState: { kind: 'pi-ai', version: 2 }, - }, + it('degrades unsupported replay-state versions to provider-neutral history', () => { + const onDegrade = vi.fn() + const context = toPiContext({ + provider: 'deepseek', + model: 'm', + messages: [createMessage({ + role: 'assistant', + content: [{ type: 'text', text: 'done' }], + source: { + kind: 'model', + ...{ + provider: 'deepseek', + model: 'old', + replayState: { response: { kind: 'pi-ai', version: 3 }, blocks: [] }, }, - })], - }) - expect.fail('expected invalid replay state') - } catch (error: unknown) { - expect(error).toBeInstanceOf(LlmError) - expect((error as LlmError).code).toBe('INVALID_REPLAY_STATE') - expect((error as Error).message).toContain('unsupported version 2') - } + }, + })], + }, undefined, onDegrade) + expect(context.messages[0]).toMatchObject({ + role: 'assistant', + api: 'dsh-foreign', + provider: 'deepseek', + model: 'old', + content: [{ type: 'text', text: 'done' }], + }) + expect(onDegrade).toHaveBeenCalledWith(expect.stringContaining('unsupported version 3')) }) - it('rejects replay metadata whose blocks do not match the durable content', () => { + it('degrades the flat pre-envelope replay state a legacy session log carries', () => { + const onDegrade = vi.fn() + const context = toPiContext({ + provider: 'deepseek', + model: 'm', + messages: [createMessage({ + role: 'assistant', + content: [{ type: 'text', text: 'done' }], + source: { + kind: 'model', + ...{ + provider: 'deepseek', + model: 'deepseek-v4-flash', + replayState: { + kind: 'pi-ai', + version: 1, + api: 'openai-completions', + provider: 'deepseek', + model: 'deepseek-v4-flash', + stopReason: 'stop', + blocks: [{ type: 'text' }], + }, + }, + }, + })], + }, undefined, onDegrade) + expect(context.messages[0]).toMatchObject({ role: 'assistant', api: 'dsh-foreign' }) + expect(onDegrade).toHaveBeenCalledWith(expect.stringContaining('expected a response object')) + }) + + it('degrades replay metadata whose blocks do not match the durable content', () => { + const onDegrade = vi.fn() const state = toPiReplayState(assistant({ content: [{ type: 'text', text: 'done' }] })) - expect(() => toPiContext({ + const context = toPiContext({ provider: 'deepseek', model: 'm', messages: [createMessage({ @@ -454,12 +487,19 @@ describe('toPiContext', () => { ...{ provider: 'deepseek', model: 'deepseek-v4-flash', replayState: state }, }, })], - })).toThrow(/block 0 does not match assistant content/) + }, undefined, onDegrade) + expect(context.messages[0]).toMatchObject({ + role: 'assistant', + api: 'dsh-foreign', + content: [{ type: 'thinking', thinking: 'done' }], + }) + expect(onDegrade).toHaveBeenCalledWith(expect.stringContaining('block 0 does not match assistant content')) }) - it('rejects replay metadata whose block count differs from durable content', () => { + it('degrades replay metadata whose block count differs from durable content', () => { + const onDegrade = vi.fn() const state = toPiReplayState(assistant()) - expect(() => toPiContext({ + const context = toPiContext({ provider: 'deepseek', model: 'm', messages: [createMessage({ @@ -470,66 +510,34 @@ describe('toPiContext', () => { ...{ provider: 'deepseek', model: 'deepseek-v4-flash', replayState: state }, }, })], - })).toThrow(/block count does not match assistant content/) + }, undefined, onDegrade) + expect(context.messages[0]).toMatchObject({ + role: 'assistant', + api: 'dsh-foreign', + provider: 'deepseek', + model: 'deepseek-v4-flash', + content: [{ type: 'text', text: 'done' }], + stopReason: 'stop', + }) + expect(onDegrade).toHaveBeenCalledWith(expect.stringContaining('block count does not match assistant content')) }) - const validReplay = { + const validResponse = { kind: 'pi-ai', - version: 1, + version: 2, api: 'openai-completions', provider: 'deepseek', model: 'deepseek-v4-flash', stopReason: 'stop', - blocks: [{ type: 'text' }], } + const validReplay = { response: validResponse, blocks: [{ type: 'text' }] } - it.each([ - ['provider', { ...validReplay, provider: 'openai' }], - ['model', { ...validReplay, model: 'deepseek-v4-pro' }], - ])('rejects replay metadata whose %s differs from assistant source', (field, replayState) => { - try { - toPiContext({ - provider: 'deepseek', - model: 'next-model', - messages: [createMessage({ - role: 'assistant', - content: [{ type: 'text', text: 'done' }], - source: { - kind: 'model', - ...{ provider: 'deepseek', model: 'deepseek-v4-flash', replayState }, - }, - })], - }) - expect.fail('expected invalid replay state') - } catch (error: unknown) { - expect(error).toBeInstanceOf(LlmError) - expect((error as LlmError).code).toBe('INVALID_REPLAY_STATE') - expect((error as Error).message).toContain(`${field} does not match assistant source`) - } - }) - - it.each([ - ['number state', 1, 'expected an object'], - ['null state', null, 'expected an object'], - ['array state', [], 'expected an object'], - ['unknown kind', { ...validReplay, kind: 'other' }, 'unknown state kind'], - ['non-string api', { ...validReplay, api: 1 }, 'api must be a non-empty string'], - ['empty provider', { ...validReplay, provider: '' }, 'provider must be a non-empty string'], - ['missing model', { ...validReplay, model: undefined }, 'model must be a non-empty string'], - ['unknown stop reason', { ...validReplay, stopReason: 'pause' }, 'unknown stopReason'], - ['non-string response model', { ...validReplay, responseModel: 1 }, 'responseModel must be a string'], - ['non-string response id', { ...validReplay, responseId: 1 }, 'responseId must be a string'], - ['non-array blocks', { ...validReplay, blocks: 'text' }, 'blocks must be an array'], - ['number block', { ...validReplay, blocks: [1] }, 'block 0 must be an object'], - ['null block', { ...validReplay, blocks: [null] }, 'block 0 must be an object'], - ['array block', { ...validReplay, blocks: [[]] }, 'block 0 must be an object'], - ['unknown block type', { ...validReplay, blocks: [{ type: 'audio' }] }, 'block 0 has an unknown type'], - ['non-string signature', { ...validReplay, blocks: [{ type: 'text', textSignature: 1 }] }, 'textSignature must be a string'], - ['non-boolean redaction', { ...validReplay, blocks: [{ type: 'reasoning', redacted: 'yes' }] }, 'redacted must be boolean'], - ])('rejects malformed replay state: %s', (_name, replayState, message) => { - expect(() => toPiContext({ + /** Convert with the given state and assert the message degraded to foreign with the given reason. */ + function expectDegraded(replayState: unknown, message: string): void { + const onDegrade = vi.fn() + const context = toPiContext({ provider: 'deepseek', - model: 'm', + model: 'next-model', messages: [createMessage({ role: 'assistant', content: [{ type: 'text', text: 'done' }], @@ -538,7 +546,45 @@ describe('toPiContext', () => { ...{ provider: 'deepseek', model: 'deepseek-v4-flash', replayState }, }, })], - })).toThrow(message) + }, undefined, onDegrade) + expect(context.messages[0]).toMatchObject({ + role: 'assistant', + api: 'dsh-foreign', + content: [{ type: 'text', text: 'done' }], + }) + expect(onDegrade).toHaveBeenCalledWith(expect.stringContaining(message)) + } + + it.each([ + ['provider', { ...validReplay, response: { ...validResponse, provider: 'openai' } }], + ['model', { ...validReplay, response: { ...validResponse, model: 'deepseek-v4-pro' } }], + ])('degrades replay metadata whose %s differs from assistant source', (field, replayState) => { + expectDegraded(replayState, `${field} does not match assistant source`) + }) + + it.each([ + ['number state', 1, 'expected a replay envelope'], + ['null state', null, 'expected a replay envelope'], + ['array state', [], 'expected a replay envelope'], + ['missing response', { blocks: [] }, 'expected a response object'], + ['array response', { ...validReplay, response: [] }, 'expected a response object'], + ['unknown kind', { ...validReplay, response: { ...validResponse, kind: 'other' } }, 'unknown state kind'], + ['non-string api', { ...validReplay, response: { ...validResponse, api: 1 } }, 'api must be a non-empty string'], + ['empty provider', { ...validReplay, response: { ...validResponse, provider: '' } }, 'provider must be a non-empty string'], + ['missing model', { ...validReplay, response: { ...validResponse, model: undefined } }, 'model must be a non-empty string'], + ['unknown stop reason', { ...validReplay, response: { ...validResponse, stopReason: 'pause' } }, 'unknown stopReason'], + ['non-string response model', { ...validReplay, response: { ...validResponse, responseModel: 1 } }, 'responseModel must be a string'], + ['non-string response id', { ...validReplay, response: { ...validResponse, responseId: 1 } }, 'responseId must be a string'], + ['missing blocks', { response: validResponse }, 'blocks must be an array'], + ['non-array blocks', { ...validReplay, blocks: 'text' }, 'blocks must be an array'], + ['number block', { ...validReplay, blocks: [1] }, 'block 0 must be an object'], + ['null block', { ...validReplay, blocks: [null] }, 'block 0 must be an object'], + ['array block', { ...validReplay, blocks: [[]] }, 'block 0 must be an object'], + ['unknown block type', { ...validReplay, blocks: [{ type: 'audio' }] }, 'block 0 has an unknown type'], + ['non-string signature', { ...validReplay, blocks: [{ type: 'text', textSignature: 1 }] }, 'textSignature must be a string'], + ['non-boolean redaction', { ...validReplay, blocks: [{ type: 'reasoning', redacted: 'yes' }] }, 'redacted must be boolean'], + ])('degrades malformed replay state: %s', (_name, replayState, message) => { + expectDegraded(replayState, message) }) }) @@ -565,12 +611,14 @@ describe('toStreamChunks', () => { type: 'finish', reason: { kind: 'stop' }, replayState: { - kind: 'pi-ai', - version: 1, - api: 'openai-completions', - provider: 'deepseek', - model: 'deepseek-v4-flash', - stopReason: 'stop', + response: { + kind: 'pi-ai', + version: 2, + api: 'openai-completions', + provider: 'deepseek', + model: 'deepseek-v4-flash', + stopReason: 'stop', + }, blocks: [{ type: 'text' }], }, }, @@ -614,12 +662,14 @@ describe('toStreamChunks', () => { type: 'finish', reason: { kind: 'tool-calls' }, replayState: { - kind: 'pi-ai', - version: 1, - api: 'openai-completions', - provider: 'deepseek', - model: 'deepseek-v4-flash', - stopReason: 'toolUse', + response: { + kind: 'pi-ai', + version: 2, + api: 'openai-completions', + provider: 'deepseek', + model: 'deepseek-v4-flash', + stopReason: 'toolUse', + }, blocks: [{ type: 'tool-call' }], }, }, diff --git a/packages/llm/llm-pi-ai/tests/loader-composition.spec.ts b/packages/llm/llm-pi-ai/tests/loader-composition.spec.ts index 0ed1eee440..a4e89424f6 100644 --- a/packages/llm/llm-pi-ai/tests/loader-composition.spec.ts +++ b/packages/llm/llm-pi-ai/tests/loader-composition.spec.ts @@ -16,13 +16,22 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { Context } from '@deepseek-ai/cordis' import Loader from '@deepseek-ai/cordis-plugin-loader' import Include from '@deepseek-ai/cordis-plugin-include' -import LlmRuntime from '@deepseek-ai/dsh-llm' +import LlmRuntime, { createMessage, createUserMessage } from '@deepseek-ai/dsh-llm' import LocalCredentialProvider from '@deepseek-ai/dsh-credentials-local' import FileSettingsProvider from '@deepseek-ai/dsh-settings-file' import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai' import { assemble } from './assemble.ts' import { closeMockServers, mockServer, textEvents } from './mock-server.ts' +/** One text block, then a tool call truncated by the output-token ceiling. */ +const truncatedToolCallEvents = [ + '{"choices":[{"delta":{"role":"assistant","content":""},"index":0,"finish_reason":null}]}', + '{"choices":[{"delta":{"content":"partial"},"index":0,"finish_reason":null}]}', + '{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"echo","arguments":"{\\"text\\":"}}]},"index":0,"finish_reason":null}]}', + '{"choices":[{"delta":{},"index":0,"finish_reason":"length"}],"usage":{"prompt_tokens":3,"completion_tokens":4}}', + '[DONE]', +] + let root: string | undefined let context: Context | undefined @@ -113,4 +122,123 @@ describe('llm-pi-ai real dormant composition', () => { expect(result.message.content).toEqual([{ type: 'text', text: 'hello' }]) expect(server.headers[0]?.authorization).toBe('Bearer key-from-store') }) + + it('continues natively after max-token assembly drops a tool call, with pruned replay metadata', async () => { + vi.stubEnv('PI_COMPOSITION_KEY', '') + const server = await mockServer([ + { events: truncatedToolCallEvents }, + { events: textEvents }, + ]) + const { ctx, settingsPath } = await loadComposition() + await writeFile(settingsPath, [ + 'llm-pi-ai:', + ' providers:', + ' deepseek:', + ' apiKeyEnv: PI_COMPOSITION_KEY', + ` baseURL: ${server.url}`, + '', + ].join('\n')) + await vi.waitFor(() => { + expect(ctx.llm.listProviders().map(provider => provider.id)).toEqual(['deepseek']) + }, { timeout: 5000 }) + + const truncated = await assemble(ctx, { + provider: 'deepseek', + model: 'deepseek-v4-flash', + messages: [], + }) + expect(truncated.finish).toEqual({ kind: 'max-tokens' }) + expect(truncated.message.content).toEqual([{ type: 'text', text: 'partial' }]) + expect(truncated.message.source).toEqual({ + kind: 'model', + provider: 'deepseek', + model: 'deepseek-v4-flash', + replayState: { + response: { + kind: 'pi-ai', + version: 2, + api: 'openai-completions', + provider: 'deepseek', + model: 'deepseek-v4-flash', + stopReason: 'length', + }, + blocks: [{ type: 'text' }], + }, + }) + + const continued = await assemble(ctx, { + provider: 'deepseek', + model: 'deepseek-v4-flash', + messages: [ + truncated.message, + createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } }), + ], + }) + expect(continued.message.content).toEqual([{ type: 'text', text: 'hello' }]) + expect(server.requests).toHaveLength(2) + expect(server.requests[1]).toMatchObject({ + messages: [ + { role: 'assistant', content: 'partial' }, + { role: 'user', content: 'continue' }, + ], + }) + const followup = server.requests[1] as { messages?: unknown[] } + expect(followup.messages?.[0]).not.toHaveProperty('tool_calls') + }) + + it('continues a legacy session whose stored replay state no longer matches its content', async () => { + vi.stubEnv('PI_COMPOSITION_KEY', '') + const server = await mockServer([{ events: textEvents }]) + const { ctx, settingsPath } = await loadComposition() + await writeFile(settingsPath, [ + 'llm-pi-ai:', + ' providers:', + ' deepseek:', + ' apiKeyEnv: PI_COMPOSITION_KEY', + ` baseURL: ${server.url}`, + '', + ].join('\n')) + await vi.waitFor(() => { + expect(ctx.llm.listProviders().map(provider => provider.id)).toEqual(['deepseek']) + }, { timeout: 5000 }) + + // A pre-envelope session log entry: max-token assembly dropped the tool + // call from content while the flat v1 state still describes both blocks. + const poisoned = createMessage({ + role: 'assistant', + content: [{ type: 'text', text: 'partial' }], + source: { + kind: 'model', + ...{ + provider: 'deepseek', + model: 'deepseek-v4-flash', + replayState: { + kind: 'pi-ai', + version: 1, + api: 'openai-completions', + provider: 'deepseek', + model: 'deepseek-v4-flash', + stopReason: 'length', + blocks: [{ type: 'text' }, { type: 'tool-call' }], + }, + }, + }, + }) + const continued = await assemble(ctx, { + provider: 'deepseek', + model: 'deepseek-v4-flash', + messages: [ + poisoned, + createUserMessage({ content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } }), + ], + }) + expect(continued.finish).toEqual({ kind: 'stop' }) + expect(continued.message.content).toEqual([{ type: 'text', text: 'hello' }]) + expect(server.requests[0]).toMatchObject({ + messages: [ + { role: 'assistant', content: 'partial' }, + { role: 'user', content: 'continue' }, + ], + }) + }) }) diff --git a/packages/llm/llm-pi-ai/tests/provider-apis.e2e.ts b/packages/llm/llm-pi-ai/tests/provider-apis.e2e.ts index b1731d7071..a2a583abd8 100644 --- a/packages/llm/llm-pi-ai/tests/provider-apis.e2e.ts +++ b/packages/llm/llm-pi-ai/tests/provider-apis.e2e.ts @@ -11,7 +11,7 @@ import type { import LlmRuntime, { createUserMessage, CallId } from '@deepseek-ai/dsh-llm' import type { Message, ToolSchema } from '@deepseek-ai/dsh-llm' import * as LlmPiAi from '@deepseek-ai/dsh-llm-pi-ai' -import type { PiAiReplayState } from '../src/replay.ts' +import type { PiAiReplayResponse } from '../src/replay.ts' import { assemble, type AssembledResult } from './assemble.ts' interface ProviderCase { @@ -118,18 +118,20 @@ function expectFinish(result: AssembledResult, expected: 'stop' | 'tool-calls'): expect(result.finish.kind).toBe(expected) } -function expectNativeReplay(result: AssembledResult, profile: ProviderCase): PiAiReplayState { +function expectNativeReplay(result: AssembledResult, profile: ProviderCase): PiAiReplayResponse { const replayState = result.message.source.kind === 'model' ? result.message.source.replayState : undefined expect(replayState).toMatchObject({ - kind: 'pi-ai', - version: 1, - api: profile.api, - provider: profile.provider, - model: profile.model, + response: { + kind: 'pi-ai', + version: 2, + api: profile.api, + provider: profile.provider, + model: profile.model, + }, }) - return replayState as PiAiReplayState + return (replayState as { response: PiAiReplayResponse }).response } const lookupTool: ToolSchema = { diff --git a/packages/llm/llm/README.i18n.yaml b/packages/llm/llm/README.i18n.yaml index 2d0d68e7ac..fce8fa059b 100644 --- a/packages/llm/llm/README.i18n.yaml +++ b/packages/llm/llm/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/llm/llm/README.md -README.md: 2cae9a05a58295b06d25382729a3304dbdc9a6fa -README.zh.md: 110cc128f1c19bef1741ae59d8106139c4a72649 +README.md: fb6bd84240b41dd730d45b3eb34c35827dc4c991 +README.zh.md: 5c22767a7c654972cbf614505382fa4755d7d318 diff --git a/packages/llm/llm/README.md b/packages/llm/llm/README.md index 2cae9a05a5..fb6bd84240 100644 --- a/packages/llm/llm/README.md +++ b/packages/llm/llm/README.md @@ -53,7 +53,7 @@ Exact-model metadata is a separate correctness query, not a catalog decoration o Message content is an array of typed blocks: `text`, `reasoning`, `tool-call`, `tool-result`. The union is derived from the merge-extensible `ContentBlockMap`, so plugins can add block types via declaration merging. Assistant messages use a model source carrying the provider and model that produced them plus optional adapter-private replay state. Before dispatch, `LlmRuntime` retains that state only when the historical provider route and target provider route are currently owned by the exact same adapter instance; the adapter then decides whether it can restore or convert the state across models/providers. The core block set is limited to blocks every shipping path honors — multimodal content (images, audio, …) has no core block type; a feature that needs one adds it via the map together with the adapter/UI/compaction support that honors it. -Streaming is a raw chunk protocol (`block-start`, `text-delta`, `reasoning-delta`, `tool-call-delta`, `block-end`, `usage`, `finish`). Every adapter outcome reaches consumers as one terminal `finish`; operational failure uses its `error` or `aborted` reason rather than throwing across the stream API. `BlockAssembler` is the single shared implementation that assembles chunks into blocks/messages. +Streaming is a raw chunk protocol (`block-start`, `text-delta`, `reasoning-delta`, `tool-call-delta`, `block-end`, `usage`, `finish`). Every adapter outcome reaches consumers as one terminal `finish`; operational failure uses its `error` or `aborted` reason rather than throwing across the stream API. `BlockAssembler` is the single shared implementation that assembles chunks into blocks/messages. A successful `finish` may carry a `ReplayEnvelope` — opaque response-level replay metadata plus optional per-block entries aligned with the emitted block sequence. Assembly makes one keep/drop decision for content and metadata together: a `max-tokens` finish drops tool calls that may have been truncated, and the envelope loses the entry at each dropped position, so stored metadata always describes stored content. ### Call configuration (`call-config.ts`) diff --git a/packages/llm/llm/README.zh.md b/packages/llm/llm/README.zh.md index 110cc128f1..5c22767a7c 100644 --- a/packages/llm/llm/README.zh.md +++ b/packages/llm/llm/README.zh.md @@ -53,7 +53,7 @@ 消息内容是类型化内容块数组：`text`、`reasoning`、`tool-call`、`tool-result`。联合从可合并扩展的 `ContentBlockMap` 派生，因此插件可以通过 declaration merging 添加块类型。assistant 消息使用模型来源，其中携带生成该消息的提供方和模型，以及可选的适配器私有回放状态。dispatch 前，`LlmRuntime` 只在历史提供方路由与目标提供方路由当前由完全相同的适配器实例拥有时才保留该状态；随后由适配器判定能否在模型／提供方间恢复或转换该状态。核心块集只包含每条已发布路径都支持的块。多模态内容（图像、音频等）没有核心块类型；需要它的功能会通过 map 添加，并一并添加相应的适配器／UI／压缩（compaction）支持。 -流式输出是原始分片协议（`block-start`、`text-delta`、`reasoning-delta`、`tool-call-delta`、`block-end`、`usage`、`finish`）。每个适配器结果都以一个终止 `finish` 到达消费方；运行故障使用 `error` 或 `aborted` 作为结束原因，而不会跨流 API 抛出。`BlockAssembler` 是将分片组装为块／消息的唯一共享实现。 +流式输出是原始分片协议（`block-start`、`text-delta`、`reasoning-delta`、`tool-call-delta`、`block-end`、`usage`、`finish`）。每个适配器结果都以一个终止 `finish` 到达消费方；运行故障使用 `error` 或 `aborted` 作为结束原因，而不会跨流 API 抛出。`BlockAssembler` 是将分片组装为块／消息的唯一共享实现。成功的 `finish` 可以携带 `ReplayEnvelope`——不透明的响应级回放元数据，加上与发射块序列对齐的可选逐块条目。组装对内容与元数据只做一次保留/丢弃决定：`max-tokens` 结束会丢弃可能被截断的工具调用，数据在每个被丢弃的位置同步失去对应条目，因此存储的元数据始终描述存储的内容。 ### 调用配置（`call-config.ts`） diff --git a/packages/llm/llm/src/assembler.ts b/packages/llm/llm/src/assembler.ts index a0e1332417..5eb3668915 100644 --- a/packages/llm/llm/src/assembler.ts +++ b/packages/llm/llm/src/assembler.ts @@ -10,7 +10,7 @@ import { CallId } from './brand.ts' import { assertNever } from './never.ts' import { createMessage } from './message.ts' import type { Message, MessageSource } from './message.ts' -import type { ContentBlock, FinishReason, StreamChunk, TokenUsage } from './types.ts' +import type { ContentBlock, FinishReason, ReplayEnvelope, StreamChunk, TokenUsage } from './types.ts' interface PartialBlock { blockType: string @@ -38,7 +38,7 @@ export class BlockAssembler { private order: number[] = [] private _usage: TokenUsage | undefined private _finish: FinishReason | undefined - private _replayState: unknown = undefined + private _replayState: ReplayEnvelope | undefined /** * Feed one chunk into the assembly state. @@ -125,6 +125,28 @@ export class BlockAssembler { return partial } + /** + * The one shared keep/drop decision over all seen blocks: max-token + * truncation drops tool calls that cannot be executed safely. Emitted blocks + * and replay metadata both derive from this result, so they cannot disagree. + */ + private assembled(): { blocks: ContentBlock[]; replay: ReplayEnvelope | undefined } { + const all = this.order.map(index => this.assemble(this.mustGet(index), index)) + const kept = this.finish.kind === 'max-tokens' + ? all.map(block => block.type !== 'tool-call') + : undefined + const blocks = kept === undefined ? all : all.filter((_, position) => kept[position]) + const envelope = this._replayState + if (envelope?.blocks === undefined) return { blocks, replay: envelope } + if (envelope.blocks.length !== all.length) return { blocks, replay: undefined } + return { + blocks, + replay: kept === undefined || blocks.length === all.length + ? envelope + : { response: envelope.response, blocks: envelope.blocks.filter((_, position) => kept[position]) }, + } + } + /** * Assemble all blocks seen so far, in stream order. * @returns one block per seen index, except that max-token truncation drops @@ -132,10 +154,7 @@ export class BlockAssembler { * its accumulated deltas (an unknown block type never closed by `block-end` throws). */ blocks(): ContentBlock[] { - const blocks = this.order.map(index => this.assemble(this.mustGet(index), index)) - return this.finish.kind === 'max-tokens' - ? blocks.filter(block => block.type !== 'tool-call') - : blocks + return this.assembled().blocks } /** Usage from the `usage` chunk; undefined until one arrives. */ @@ -148,9 +167,13 @@ export class BlockAssembler { return this._finish ?? { kind: 'stop' } } - /** Adapter-private replay state from the terminal finish chunk, if any. */ - get replayState(): unknown { - return this._replayState + /** + * Replay metadata from the terminal finish chunk, if any, with per-block + * entries pruned in step with {@link blocks}. Undefined when the envelope's + * entries do not align with the emitted blocks. + */ + get replayState(): ReplayEnvelope | undefined { + return this.assembled().replay } /** diff --git a/packages/llm/llm/src/types.ts b/packages/llm/llm/src/types.ts index 326db1cb14..8c5be187dd 100644 --- a/packages/llm/llm/src/types.ts +++ b/packages/llm/llm/src/types.ts @@ -280,6 +280,27 @@ export interface LlmResolvedModelInfo extends LlmModelInfo { reasoning?: LlmModelReasoningInfo } +/** + * Adapter-private lossless-JSON state for replaying a successful response, + * carried by a terminal `finish` chunk and stored on the assembled assistant + * message's model source. Both halves stay opaque to the harness; only the + * split is shared vocabulary, so assembly can keep stored metadata aligned + * with stored content without reading either half. + */ +export interface ReplayEnvelope { + /** Response-level adapter-private metadata (ids, native stop reason). */ + response: unknown + /** + * Per-block adapter-private metadata, one entry per emitted block in + * first-seen stream order. When assembly drops a block it drops the entry at + * the same position; entries whose length does not match the emitted block + * count discard the whole envelope. An adapter whose metadata is independent + * of block structure omits this field and the envelope passes through + * assembly unchanged. + */ + blocks?: readonly unknown[] +} + /** * Raw streaming protocol emitted by adapters. * Block indexes correlate interleaved deltas, and `block-end` carries the @@ -298,8 +319,8 @@ export type StreamChunk = | { type: 'finish' reason: FinishReason - /** Adapter-private lossless-JSON state for replaying a successful response. */ - replayState?: unknown + /** Replay metadata for a successful response; see {@link ReplayEnvelope}. */ + replayState?: ReplayEnvelope } /** diff --git a/packages/llm/llm/tests/assembler.spec.ts b/packages/llm/llm/tests/assembler.spec.ts index bf2276a218..9f73ee3f96 100644 --- a/packages/llm/llm/tests/assembler.spec.ts +++ b/packages/llm/llm/tests/assembler.spec.ts @@ -118,6 +118,85 @@ describe('BlockAssembler', () => { }) }) +describe('BlockAssembler replay metadata', () => { + const response = { responseId: 'resp-1' } + + it('prunes per-block replay entries with the tool calls a max-tokens finish drops', () => { + const assembler = new BlockAssembler() + assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'lead' } }) + assembler.push({ + type: 'block-end', + index: 1, + block: { type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{"text":' }, + }) + assembler.push({ type: 'block-end', index: 2, block: { type: 'reasoning', text: 'tail' } }) + assembler.push({ + type: 'finish', + reason: { kind: 'max-tokens' }, + replayState: { response, blocks: ['meta-0', 'meta-1', 'meta-2'] }, + }) + + expect(assembler.blocks()).toEqual([ + { type: 'text', text: 'lead' }, + { type: 'reasoning', text: 'tail' }, + ]) + expect(assembler.replayState).toEqual({ response, blocks: ['meta-0', 'meta-2'] }) + }) + + it('omits replay metadata whose per-block entries misalign with the emitted blocks', () => { + const assembler = new BlockAssembler() + assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'one' } }) + assembler.push({ type: 'block-end', index: 1, block: { type: 'text', text: 'two' } }) + assembler.push({ + type: 'finish', + reason: { kind: 'stop' }, + replayState: { response, blocks: ['meta-0'] }, + }) + + expect(assembler.blocks()).toHaveLength(2) + expect(assembler.replayState).toBeUndefined() + }) + + it('passes replay metadata through unchanged when assembly drops nothing', () => { + const replayState = { response, blocks: ['meta-0', 'meta-1'] } + const assembler = new BlockAssembler() + assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'partial' } }) + assembler.push({ + type: 'block-end', + index: 1, + block: { type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{}' }, + }) + assembler.push({ type: 'finish', reason: { kind: 'tool-calls' }, replayState }) + + expect(assembler.replayState).toBe(replayState) + }) + + it('keeps a max-tokens replay state with no per-block entries across a tool-call drop', () => { + const replayState = { response } + const assembler = new BlockAssembler() + assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'partial' } }) + assembler.push({ + type: 'block-end', + index: 1, + block: { type: 'tool-call', id: CallId('c1'), name: 'echo', arguments: '{"text":' }, + }) + assembler.push({ type: 'finish', reason: { kind: 'max-tokens' }, replayState }) + + expect(assembler.blocks()).toEqual([{ type: 'text', text: 'partial' }]) + expect(assembler.replayState).toBe(replayState) + }) + + it('keeps a text-only max-tokens response and its replay metadata intact', () => { + const replayState = { response, blocks: ['meta-0'] } + const assembler = new BlockAssembler() + assembler.push({ type: 'block-end', index: 0, block: { type: 'text', text: 'partial' } }) + assembler.push({ type: 'finish', reason: { kind: 'max-tokens' }, replayState }) + + expect(assembler.blocks()).toEqual([{ type: 'text', text: 'partial' }]) + expect(assembler.replayState).toBe(replayState) + }) +}) + describe('assertNever', () => { it('throws with diagnostics when a value escapes a closed union at runtime', async () => { const { assertNever } = await import('@deepseek-ai/dsh-llm') diff --git a/scripts/type-equiv.manifest.json b/scripts/type-equiv.manifest.json index 5e6184b88d..95a573541c 100644 --- a/scripts/type-equiv.manifest.json +++ b/scripts/type-equiv.manifest.json @@ -351,6 +351,11 @@ "symbol": "ToolProviderResult", "source": "packages/core/system-prompt/src/index.ts" }, + { + "doc": "docs/subsystems/llm-streaming.md", + "symbol": "ReplayEnvelope", + "source": "packages/llm/llm/src/types.ts" + }, { "doc": "docs/subsystems/llm-streaming.md", "symbol": "StreamChunk",