diff --git a/.env.example b/.env.example index f5a2872154..3a594eadca 100644 --- a/.env.example +++ b/.env.example @@ -168,6 +168,12 @@ POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_TIMEOUT_SECONDS=30 # POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_MODEL_SETTINGS={"dimensions":2560} # Fixed-dimension models reject the dimensions request field. Keep EMBEDDING_DIMENSION and disable sending: # POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_SEND_DIMENSIONS=false + +# Optional Atomic Memory extraction over a same-host gateway's cached turn prefix (off unless the URL is set). +# Requires GENERATION_MODEL, which normal extraction uses as the fallback. Headers are secrets. +# POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_URL=http://127.0.0.1:8787 +# POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_HEADERS={"Authorization":"Bearer replace-me"} +# POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_CACHED_INPUT_RATIO=0.1 # A dedicated reranker LLM is optional; without it, reranking reuses the generation model. # POWERCONTEXT_SERVER_INFERENCE_RERANK_MODEL=openai-chat:local-reranker # POWERCONTEXT_SERVER_INFERENCE_RERANK_BASE_URL=http://127.0.0.1:8082/v1 diff --git a/docs/en/development/capture-coverage.md b/docs/en/development/capture-coverage.md index 970ca757ad..6e4240bb8a 100644 --- a/docs/en/development/capture-coverage.md +++ b/docs/en/development/capture-coverage.md @@ -38,7 +38,7 @@ Do not add content, paths, message ID lists, prompts, model configuration or cre | Field | Contract | | --- | --- | | `schema` | Exactly `powercontext.capture-coverage.v1`. | -| `projection` | One of the three profiles below. | +| `projection` | One of the four profiles below. | | `unit` | The exact unit paired with that profile. | | `status` | `complete`, `partial` or `unknown`. | | `reasons` | Unique supported strings in ascending lexicographic order. | @@ -65,7 +65,8 @@ interpreted as unknown. Interpretation never rewrites the stored payload. | --- | --- | --- | | `hermes-turn-text.v1` | `user_assistant_text_pair` | The current normalized user/assistant text pair with the existing role labels and separators. | | `hermes-precompress-text.v1` | `new_user_assistant_window` | The new user/assistant entries delivered by the host after existing filtering and overlap removal. | -| `gateway-turn-text.v1` | `gateway_turn_projection` | The existing supported Chat/Responses normalized turn from the last user message through the terminal assistant response. | +| `gateway-turn-text.v1` | `gateway_turn_projection` | The earlier Chat/Responses normalized turn from the last user message through the terminal assistant response, with byte limits per field and per record. Used only to interpret existing Sources. | +| `gateway-turn-text.v2` | `gateway_turn_projection` | The complete turn the Gateway now writes, over the same range as v1, with nothing truncated or omitted. | ### Hermes completed turn @@ -90,7 +91,28 @@ window before sending a partial Source or advancing the captured snapshot. Scope binding, overlap identities and checkpoint transitions retain their existing semantics. Once capture is durable, later extraction failure does not invalidate the checkpoint. -### Gateway normalized turn +### Gateway complete turn (v2) + +New Gateway Sources use `gateway-turn-text.v2`. It keeps everything the model read in the turn: the user text, other +messages after it, visible reasoning summaries, every argument field of each tool call, tool outputs and errors, +results returned by built-in tools, unrecognized items, and the final assistant text. No field is truncated and no +step is omitted. Injected prepare content, provider reasoning ciphertext, IDs, statuses and other transport metadata +are outside the profile, and `data:` binary content becomes a placeholder with its media type and size. Excluding +them does not itself make this projection partial. + +So that Core's normal extraction can process each Source on its own (default input budget 24,000 estimated tokens), +a turn over 12,000 tokens by Core's character estimate (of the escaped text: one token per 4 ASCII characters or per +other character) is split, in order, into several Sources. Each part is a +complete JSON document with `part` and `parts`. A single step that does not fit has its longest text field split into +consecutive steps of the same type, and the later pieces carry `"continued": true`. Splitting loses nothing, so every +part declares `complete`. The metadata `turn` of each part gives the turn's `source_id` and the part number. Terminal +response detection is the same as in v1. + +### Gateway normalized turn (v1) + +The following describes the v1 projection, for interpreting existing Sources that declare it; the Gateway no longer +writes v1. + The projection retains its existing visible text, tool calls and tool outputs. Injected prepare content, opaque reasoning, transport IDs and unsupported media remain outside the profile. diff --git a/docs/en/docs/operate/configuration.md b/docs/en/docs/operate/configuration.md index 68e099c75d..03072bc657 100644 --- a/docs/en/docs/operate/configuration.md +++ b/docs/en/docs/operate/configuration.md @@ -214,6 +214,9 @@ still occupies an HTTP slot until it finishes; reserve other slots for concurren | `POWERCONTEXT_SERVER_INFERENCE_GENERATION_STREAMING` | `disabled` | When `auto`, only generation models whose provider prefix is `anthropic`, `openai`, `openai-chat`, or `openai-responses` use a streamed request aggregated locally into one complete response. Other prefixes stay on non-streaming requests. `disabled` keeps every model on non-streaming requests | | `POWERCONTEXT_SERVER_INFERENCE_GENERATION_STREAM_IDLE_TIMEOUT_SECONDS` | `60` | Maximum idle seconds while aggregating a streamed request: while the stream opens and waits for its first model event, and between model events (text, thinking, or tool-call parts). Provider heartbeat frames do not reset this timer; a longer gap fails as a timeout | | `POWERCONTEXT_SERVER_INFERENCE_GENERATION_MODEL_CONTEXT_WINDOW_TOKENS` | `125000` | Total generation-model context window used to budget Topic processing | +| `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_URL` | unset | Base URL of a gateway on the same host; extracts Atomic Memory for a single gateway turn over its cached prefix and falls back to normal extraction. Requires `GENERATION_MODEL`. See [Memory extraction over the cached prefix](deploy-server.md#memory-extraction-over-the-cached-prefix) | +| `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_CACHED_INPUT_RATIO` | `0.1` | Price of a cached input token relative to an uncached one for the agent's target, greater than 0 and at most 1; the prefix is reused only when it is estimated to cost less than normal extraction | +| `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_HEADERS` | `{}` | JSON object of static headers for the gateway, such as its `Authorization` key; values are secrets | | `POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_MODEL` | unset | Pydantic AI embedding model; requires profile ID and dimension | | `POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_BASE_URL` | provider default | Custom OpenAI-compatible embeddings base URL | | `POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_HEADERS` | `{}` | JSON object of static embedding client headers; values are secrets | diff --git a/docs/en/docs/operate/deploy-server.md b/docs/en/docs/operate/deploy-server.md index fb49deedaf..543dc42d52 100644 --- a/docs/en/docs/operate/deploy-server.md +++ b/docs/en/docs/operate/deploy-server.md @@ -317,6 +317,51 @@ These are per-body and per-class concurrency limits. They do not measure retaine RSS ceiling. Process-wide retained-byte accounting and output-capacity discovery belong to the remaining [G5-B work in #2464](https://github.com/oceanbase/powercontext/issues/2464). +### Memory extraction over the cached prefix + +When the gateway runs on the same host as the Server, Atomic Memory extraction for a captured turn can reuse the +provider's prompt cache for the agent's conversation instead of sending the transcript again. It is off by default and +needs both sides: + +- **Gateway:** `PC_GATEWAY_PREFIX_EXTRACTION=on` together with capture (`PC_GATEWAY_SEDIMENT=on`). For a Responses + conversation, the gateway keeps the exact request it sent and the response output of each turn's final response in + memory, for at most 30 minutes and at most 64 turns. It records an opaque handle in the captured Source's metadata, + with the input and output token counts of that final response. + The request and response never enter the Source, evidence, or logs. The captured Source records the turn in full + (projection `gateway-turn-text.v2`): no field is truncated and no step is omitted, and every argument field of a tool + call, tool outputs and errors, other messages, and unrecognized items are included. Only the provider's encrypted + reasoning, IDs, and other transport metadata are left out, and binary data becomes a placeholder with its size. + Injected memory blocks are not in the Source. Each Source must fit Core's default extraction input on its own, so + a turn over about 12,000 tokens by Core's character estimate is captured as consecutive Sources. Their metadata + `turn` names the turn and the part, and the handle is on the last part. +- **Server:** `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_URL` set to the gateway's listen address, and + `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_HEADERS` set to `{"Authorization": "Bearer "}` + when the gateway requires a key. A generation model is still required, because normal extraction is the fallback. + +Cached input is not free: the provider still bills it at a reduced rate. When a processing window holds exactly one +such turn, the Server first estimates both requests. The prefixed request costs Core's instruction plus the final +response's output tokens at the full rate, plus the cached input tokens multiplied by +`POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_CACHED_INPUT_RATIO` (default `0.1`). Normal extraction costs its +instructions plus the Source evidence. The window must hold exactly one whole turn: one Source, or every part of one +turn in order. The Server uses the prefix only when that estimate is lower. This usually +holds for turns with substantial tool output, and stops holding as a long conversation's prefix grows. It then sends its extraction instruction, the output schema, +and the Source's provenance to `POST /gateway/v1/extract`. The gateway appends them as one user message after the +kept request and response, sets `tool_choice` to `none`, and sends it to the agent's own target and account. Fields +that would change the cached prefix, such as tools, `prompt_cache_key`, reasoning settings, and a response format, are +not changed. The request passes the Server's normal `before_send` and `before_release` checks. The prefix holds the +memory blocks the gateway injected; the extraction instruction states that content between `BEGIN_POWERCONTEXT_*` and +`END_POWERCONTEXT_*` markers is not evidence for the turn. An enforcing Server withholds a +request that replays the provider's encrypted reasoning, so with Codex and similar clients extraction then falls back +to normal extraction. It is not injected and is not captured. The Server validates the reply with the same schema +and rules as normal extraction. + +Any other case falls back to normal extraction: a window that is not exactly one whole turn or has no handle, a prefix that is not +estimated to cost less or whose size is unknown, an expired or unknown +handle, a target mismatch, a withheld or blocked check, a provider or gateway error, or a reply that fails validation. +Each attempt is logged as `atomic_memory.prefixed_extraction`; see +[Logs, metrics, and tracing](observability.md#operational-fields-and-correlation). Model usage, including cached input +tokens, is reported like any other generation. + ## Run with Docker Build the image from the repository root: diff --git a/docs/en/docs/operate/observability.md b/docs/en/docs/operate/observability.md index 2f9bd0b159..9cf8cd47d9 100644 --- a/docs/en/docs/operate/observability.md +++ b/docs/en/docs/operate/observability.md @@ -147,6 +147,13 @@ When `auto` is set and the model provider prefix is outside `anthropic`, `openai `openai-responses`, an `inference.generation.streaming_not_applied` WARNING with `unit` `inference` reports that the model stayed on a non-streaming request. The event carries no prompt or response content and does not change the request result. +With prefixed extraction configured (`POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_URL`), an INFO +`atomic_memory.prefixed_extraction` event with `operation` `atomic_memory.extract` and `unit` `inference` reports each +attempt for a window that holds one gateway turn with a prefix handle. `outcome` is `success` when the reply over the +cached prefix was used, or `fallback` when normal extraction ran instead; `fallback_reason` is then one of +`not_cheaper`, `prefix_size_unknown`, `prefix_unavailable`, `target_mismatch`, `disabled`, `invalid_instruction`, `check_rejected`, `gateway_unavailable`, +`provider_error`, `gateway_error`, or `invalid_output`. Windows without such a turn emit no event. The event carries no handle, prompt, +or response content. For child cancellation, the supervisor emits `artifact_processing.failed` at `INFO` with outcome/category `cancelled` and diagnostic code `worker_cancelled`. Application spans classify mapped diagnostic codes even when sensitive exception values are excluded from logs and traces. diff --git a/docs/zh/development/capture-coverage.md b/docs/zh/development/capture-coverage.md index b1b6531f90..4e39897215 100644 --- a/docs/zh/development/capture-coverage.md +++ b/docs/zh/development/capture-coverage.md @@ -36,7 +36,7 @@ prefix extraction 扩展。不得加入正文、路径、message ID 列表、pro | 字段 | 契约 | | --- | --- | | `schema` | 固定为 `powercontext.capture-coverage.v1`。 | -| `projection` | 下表定义的三个 profile 之一。 | +| `projection` | 下表定义的四个 profile 之一。 | | `unit` | 与 profile 精确配对的投影单元。 | | `status` | `complete`、`partial` 或 `unknown`。 | | `reasons` | 受支持字符串，去重并按字典序升序排列。 | @@ -61,7 +61,8 @@ reason 重复或乱序、status 与 reasons 矛盾，都解释为 unknown。解 | --- | --- | --- | | `hermes-turn-text.v1` | `user_assistant_text_pair` | 当前规范化后的 user/assistant 文本对，包含既有角色标签和分隔符。 | | `hermes-precompress-text.v1` | `new_user_assistant_window` | 宿主提供、经现有过滤及重叠去重后，本次待捕获的新 user/assistant 条目窗口。 | -| `gateway-turn-text.v1` | `gateway_turn_projection` | 当前受支持的 Chat/Responses 规范投影，从最后一条 user 到 terminal assistant。 | +| `gateway-turn-text.v1` | `gateway_turn_projection` | 早期的 Chat/Responses 规范投影，从最后一条 user 到 terminal assistant，字段与整条记录有字节上限。只用于解读已有 Source。 | +| `gateway-turn-text.v2` | `gateway_turn_projection` | Gateway 当前写入的完整回合投影，范围同 v1，不截断、不省略。 | ### Hermes completed turn @@ -85,7 +86,22 @@ fingerprint 和 overlap 处理。投影单元是本次请求捕获的完整新 Scope 固定、重叠身份和 checkpoint 状态转换保持现有语义。capture 已持久化之后，后续 extraction 失败不否定 checkpoint。 -### Gateway normalized turn +### Gateway complete turn（v2） + +Gateway 新写入的 Source 使用 `gateway-turn-text.v2`。它保留这一回合中模型读到的全部内容：用户文本、其后的其他消息、 +推理的可见摘要、工具调用的全部参数字段、工具输出与错误、内置工具返回的结果和无法识别的条目，以及最终助手文本。 +字段不截断，步骤不省略。注入的 prepare、推理密文、ID、状态和其他传输元数据不在 profile 内；`data:` 二进制内容记为带 +媒体类型和字节数的占位文本。排除这些内容本身不使投影变为 partial。 + +为使每段 Source 都能被 Core 的普通抽取单独处理（默认输入预算 24,000 个估算 token），一个回合按 Core 的字符估算 +（转义后的文本，ASCII 每 4 个字符、其他每个字符各一个 token）超过 12,000 个 token 时，按顺序拆成几段 Source，每段都是完整的 JSON 文档，带 `part` 与 +`parts`；单个步骤放不下时，其最长的文本字段拆成连续的同类步骤，后续部分带 `"continued": true`。拆分不丢失内容， +每段的声明都是 `complete`。各段元数据中的 `turn` 给出所属回合的 `source_id` 与段号。终止响应的判定与 v1 相同。 + +### Gateway normalized turn（v1） + +以下描述 v1 投影，供解读 v1 声明的已有 Source；Gateway 不再写入 v1。 + 继续保留既有可见文本、工具调用和工具输出投影。注入的 prepare、opaque reasoning、传输 ID 及不支持 的媒体仍在 profile 之外；排除它们本身不代表该投影 partial。 diff --git a/docs/zh/docs/operate/configuration.md b/docs/zh/docs/operate/configuration.md index 1f01844222..f318a3a4c2 100644 --- a/docs/zh/docs/operate/configuration.md +++ b/docs/zh/docs/operate/configuration.md @@ -201,6 +201,9 @@ MCP 使用 stateless Streamable HTTP，每次工具执行都采用当前请求 | `POWERCONTEXT_SERVER_INFERENCE_GENERATION_STREAMING` | `disabled` | `auto` 时，仅模型名前缀为 `anthropic`、`openai`、`openai-chat`、`openai-responses` 的 generation 模型改用流式请求并在本地聚合为完整响应。其他前缀仍走非流式。`disabled` 时全部保持非流式请求 | | `POWERCONTEXT_SERVER_INFERENCE_GENERATION_STREAM_IDLE_TIMEOUT_SECONDS` | `60` | 流式聚合时的最大空闲秒数：覆盖等待流建立并收到首个模型事件，以及相邻模型事件（正文、思考、工具调用）之间的间隔。供应商心跳不重置这个计时，超过即按超时处理 | | `POWERCONTEXT_SERVER_INFERENCE_GENERATION_MODEL_CONTEXT_WINDOW_TOKENS` | `125000` | Topic 处理预算使用的 generation model 总上下文窗口 | +| `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_URL` | 未设置 | 同一主机上网关的 base URL；为单个网关回合复用其缓存前缀抽取 Atomic Memory，失败时回退到普通抽取。需要 `GENERATION_MODEL`。参见[复用缓存前缀抽取记忆](deploy-server.md#复用缓存前缀抽取记忆) | +| `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_CACHED_INPUT_RATIO` | `0.1` | Agent 目标上命中缓存的输入 token 相对未命中缓存的价格比例，大于 0 且不超过 1；只有估算复用前缀比普通抽取更便宜时才复用 | +| `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_HEADERS` | `{}` | 发往网关的静态 header JSON 对象，例如网关的 `Authorization` 密钥；值为 secret | | `POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_MODEL` | 未设置 | Pydantic AI embedding model；必须同时设置 profile ID 和 dimension | | `POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_BASE_URL` | provider 默认值 | 自定义 OpenAI-compatible embeddings base URL | | `POWERCONTEXT_SERVER_INFERENCE_EMBEDDING_HEADERS` | `{}` | embedding client 静态 header JSON object；value 按 secret 处理 | diff --git a/docs/zh/docs/operate/deploy-server.md b/docs/zh/docs/operate/deploy-server.md index 0ad4cf2bed..6460ee0bcf 100644 --- a/docs/zh/docs/operate/deploy-server.md +++ b/docs/zh/docs/operate/deploy-server.md @@ -287,6 +287,38 @@ G5-A 的“可能已计费进入调用证据”仍待接入 [#2506](https://gith 这些是逐正文和分类并发限制，不是驻留字节测量，也不构成进程 RSS 上限。 全局驻留字节记账与输出容量发现属于 [#2464 的 G5-B 后续工作](https://github.com/oceanbase/powercontext/issues/2464)。 +### 复用缓存前缀抽取记忆 + +网关与 Server 部署在同一主机时，为一个已沉淀回合抽取 Atomic Memory 可以复用服务商为 Agent 会话写下的提示缓存， +而不必再次发送对话记录。该功能默认关闭，需要两端同时开启： + +- **网关：** 设置 `PC_GATEWAY_PREFIX_EXTRACTION=on`，并开启沉淀（`PC_GATEWAY_SEDIMENT=on`）。对 Responses 会话，网关 + 在内存中保留每个回合终止响应对应的、实际发出的请求及其响应输出，最多 30 分钟、最多 64 个回合，并在沉淀的 Source + 元数据中记录一个不透明的句柄，以及该终止响应的输入和输出 token 数。请求和响应本身从不进入 Source、证据或日志。 + 沉淀的 Source 完整记录这一回合（投影 `gateway-turn-text.v2`）：字段不截断、步骤不省略，工具调用的全部参数字段、 + 工具输出与错误、其他消息和无法识别的条目都在其中；只有推理密文、ID 和其他传输元数据不进入记录，二进制数据记为带字节数 + 的占位文本。注入的记忆块不在 Source 里。每段 Source 都要能被 Core 的普通抽取单独处理，因此一个回合按 Core 的字符 + 估算超过约 12,000 个 token 时，按顺序拆成几段 Source，元数据的 `turn` 标明所属回合与段号，句柄在最后一段上。 +- **Server：** 将 `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_URL` 设为网关的监听地址；网关要求密钥时， + 将 `POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_HEADERS` 设为 + `{"Authorization": "Bearer "}`。仍需配置生成模型，因为普通抽取是回退路径。 + +命中缓存的输入并非免费，服务商仍按较低的价格计费。处理窗口恰好只含一个这样的回合时，Server 先估算两种请求的成本： +前缀请求为 Core 的指令加上终止响应的输出 token（按全价），再加上命中缓存的输入 token 乘以 +`POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_CACHED_INPUT_RATIO`（默认 `0.1`）；普通抽取为其指令加上 Source +证据。处理窗口须恰好含一个完整回合：一段 Source，或同一回合按顺序排列的全部分段。只有前者更低时才复用前缀。工具输出较多的回合通常满足这一条件；长会话的前缀不断增长后，就不再满足。随后 Server 把抽取指令、输出 Schema 和该 Source 的来源信息发给 +`POST /gateway/v1/extract`。网关把它们作为一条用户消息追加在保留的请求和响应之后，把 `tool_choice` 设为 `none`， +发往 Agent 自己的目标和账号。工具、`prompt_cache_key`、推理设置、响应格式等会改变缓存前缀的字段保持不变。该请求 +照常经过 Server 的 `before_send` 和 `before_release` 检查。前缀里含有网关注入的记忆块，抽取指令说明 +`BEGIN_POWERCONTEXT_*` 与 `END_POWERCONTEXT_*` 标记之间的内容不是这一回合的证据。enforcing 模式下，Server 会暂扣重放服务商加密推理的请求， +因此对 Codex 等客户端，抽取会回退到普通抽取。它不注入、也不沉淀。Server +用与普通抽取相同的 Schema 和规则校验回复。 + +其他情况都回退到普通抽取：窗口不是恰好一个完整回合或没有句柄、估算复用前缀并不更便宜或前缀规模未知、句柄过期或未知、目标不一致、检查被暂扣或阻断、服务商或网关 +出错，以及回复未通过校验。每次尝试都记录为 `atomic_memory.prefixed_extraction` 日志事件，参见 +[日志、指标与追踪](observability.md#operational-fields-与关联)。模型用量（包括命中缓存的输入 token）与其他 +生成一样上报。 + ## 使用 Docker 运行 在仓库根目录构建镜像： diff --git a/docs/zh/docs/operate/observability.md b/docs/zh/docs/operate/observability.md index ae973506dd..212acc0568 100644 --- a/docs/zh/docs/operate/observability.md +++ b/docs/zh/docs/operate/observability.md @@ -140,6 +140,12 @@ Supervisor 的 worker 和 wait spans 在跨进程后保留 child 的有限错误 `auto` 已设置且模型名前缀不属于 `anthropic`、`openai`、`openai-chat`、`openai-responses` 时，`unit` 为 `inference` 的 `inference.generation.streaming_not_applied` WARNING 表示该模型仍走非流式请求。该事件不携带 prompt 或响应内容，也不改变本次请求的结果。 +配置了前缀抽取（`POWERCONTEXT_SERVER_INFERENCE_PREFIX_EXTRACTION_GATEWAY_URL`）时，对含一个带前缀句柄的网关回合的 +窗口，每次尝试输出一条 INFO 级别的 `atomic_memory.prefixed_extraction` 事件，`operation` 为 `atomic_memory.extract`， +`unit` 为 `inference`。使用了基于缓存前缀的回复时 `outcome` 为 `success`；改用普通抽取时为 `fallback`，此时 +`fallback_reason` 取 `not_cheaper`、`prefix_size_unknown`、`prefix_unavailable`、`target_mismatch`、`disabled`、`invalid_instruction`、`check_rejected`、 +`gateway_unavailable`、`provider_error`、`gateway_error` 或 `invalid_output` 之一。不含这种回合的窗口不输出该事件。该事件不携带句柄、 +prompt 或响应内容。 Child 取消时，supervisor 以 `INFO` 级别输出 `artifact_processing.failed`，outcome/category 为 `cancelled`， 诊断代码为 `worker_cancelled`。Application spans 根据已映射的诊断代码分类， 即使敏感 exception values 已从 logs 和 traces 中排除，也保留正确分类。 diff --git a/gateway/Cargo.lock b/gateway/Cargo.lock index 134cb05c60..97325525c3 100644 --- a/gateway/Cargo.lock +++ b/gateway/Cargo.lock @@ -658,6 +658,7 @@ name = "pc-gateway-core" version = "0.1.0" dependencies = [ "bytes", + "getrandom 0.4.3", "pc-gateway-evidence", "pc-gateway-protocol", "pc-gateway-transport", diff --git a/gateway/README.md b/gateway/README.md index 02b3ac50cf..2c96e984f5 100644 --- a/gateway/README.md +++ b/gateway/README.md @@ -19,7 +19,7 @@ An optional Rust executor for PowerContext, defined by the gateway design in #21 - **Auto-recall injection**: for each new user message the gateway calls `prepare` with `include_reuse`, and injects the returned prepared context **verbatim** into the current user message. The exact UTF-8 byte range is declared in `prepared_contexts` (`ModelJsonTextLocation`); Core verifies the digest of the declared range against the signed reuse binding — a single byte changed by the gateway is rejected. Empty or failed prepares inject nothing and never block the main path. Injection is tail-only (current message), so the client-owned history prefix stays byte-stable (KV prefix-cache friendly, verified with DeepSeek `cached_tokens`). A request that belongs to a conversation gets the RFC 2199 lifecycle. The gateway takes the conversation from `X-PowerContext-Conversation-ID` when the host sends one; otherwise from the client's native session header: Claude Code `x-claude-code-session-id` combined with `x-claude-code-agent-id` (subagents share the session ID), OpenCode `x-opencode-session-id` (a child session has its own ID; the shared affinity headers are not used), or, on `/gateway/v1/responses` only, Codex `thread-id` (a subagent has its own thread ID; the `session-id` and `prompt_cache_key` it shares with its parent are not used). Claude Code speaks the Anthropic Messages API, which the gateway does not serve, so its headers take effect only behind a proxy that translates its requests to the Chat or Responses route. Within a conversation, a request whose messages up to and including the last user message are unchanged (a tool continuation or retry) replays the exact prepared content and binding without another retrieval, and a new user message prepares once. Replay state is partitioned by Scope and conversation ID, kept in memory, bounded, and expires with the reuse binding (at most 5 minutes). Without a conversation, every request prepares on its own and never inherits an earlier injection. An empty result is replayed as well; a failed `prepare` is retried by the next request and never blocks it. Every send and release still goes through `check`. -- **Sedimentation**: each user turn is captured once, synchronously, via `capture_content_source` when its terminal response is released: one that needs no client-executed tool. On the Responses route, built-in tools that complete inside the response (web search, file search, code interpreter, image generation, MCP calls) count as finished, while client-executed calls (`*_call` items, MCP approval requests) and unrecognized items keep the turn open (millisecond-scale durable write; extraction happens later in PC background workers). Tool-calling responses in between are not captured. The captured `gateway_turn` record is normalized from the pre-injection request and the terminal response: the user text, then each tool call (name and arguments) and tool output, including built-in tool calls and their returned results, then the final assistant text. Provider reasoning ciphertext, IDs, usage, other transport metadata and `data:` URLs (screenshots, images) are left out; each field is capped at 8 KiB and the record at 64 KiB, dropping the oldest steps first (`omitted_steps`). `source_id` identifies the turn: a digest of the Scope, the trusted conversation ID and the pre-injection messages up to the last user message, so a retried, repeated or concurrent terminal request writes the same Source (Core keeps the first content and answers a different one with 409, recorded as `sediment_duplicate`). Without a trusted conversation, identical messages in different conversations are never merged, and each release keeps its own `source_id` from the release binding. Blocked and degraded turns are never captured, and a failed capture is recorded without blocking the release. The full loop — conversation → capture → background extraction into Atomic Memory → `prepare` recall in a *new* session — has been verified end-to-end. +- **Sedimentation**: each user turn is captured once, synchronously, via `capture_content_source` when its terminal response is released: one that needs no client-executed tool. On the Responses route, built-in tools that complete inside the response (web search, file search, code interpreter, image generation, MCP calls) count as finished, while client-executed calls (`*_call` items, MCP approval requests) and unrecognized items keep the turn open (millisecond-scale durable write; extraction happens later in PC background workers). Tool-calling responses in between are not captured. The captured `gateway_turn` record (projection `gateway-turn-text.v2`) is normalized from the pre-injection request and the terminal response, and keeps the whole turn: the user text, then other messages, visible reasoning summaries, each tool call (name and every argument field), tool outputs and errors, including built-in tool calls and their returned results, and unrecognized items, then the final assistant text. Provider reasoning ciphertext, IDs, statuses, usage and other transport metadata are left out, and `data:` URLs (screenshots, images) become placeholders with their size. No field is truncated and no step is omitted; a turn over 12,000 tokens by Core's character estimate is captured as consecutive Sources, so that each fits Core's default extraction input `-of` whose metadata `turn` names the turn and the part. `source_id` identifies the turn: a digest of the Scope, the trusted conversation ID and the pre-injection messages up to the last user message, so a retried, repeated or concurrent terminal request writes the same Source (Core keeps the first content and answers a different one with 409, recorded as `sediment_duplicate`). Without a trusted conversation, identical messages in different conversations are never merged, and each release keeps its own `source_id` from the release binding. Blocked and degraded turns are never captured, and a failed capture is recorded without blocking the release. The full loop — conversation → capture → background extraction into Atomic Memory → `prepare` recall in a *new* session — has been verified end-to-end. ### Deployment & ops @@ -100,7 +100,9 @@ PC_GATEWAY_UPSTREAM_API_KEY_CHAT=... \ # base prompt; read and validated at startup), PC_GATEWAY_SEDIMENT=on (capture after release), # PC_GATEWAY_DEGRADE=passthrough (no-memory passthrough when Core is unreachable; never applies # to deployments with enforcing protection), PC_GATEWAY_MAX_CONCURRENT / PC_GATEWAY_MAX_BODY_BYTES, -# PC_GATEWAY_CORE_TOKEN (bearer for Core when authentication is enabled). +# PC_GATEWAY_CORE_TOKEN (bearer for Core when authentication is enabled), +# PC_GATEWAY_PREFIX_EXTRACTION=on (with PC_GATEWAY_SEDIMENT=on: keep each Responses turn's prefix in memory for +# 30 minutes and serve POST /gateway/v1/extract, so Core extracts memory over the provider's prompt cache). ``` Credentials live only in server-side environment variables; they are never accepted from request fields and never enter check payloads. diff --git a/gateway/crates/pc-gateway-app/src/ingress.rs b/gateway/crates/pc-gateway-app/src/ingress.rs index ad713dd58a..4f17ddd1ea 100644 --- a/gateway/crates/pc-gateway-app/src/ingress.rs +++ b/gateway/crates/pc-gateway-app/src/ingress.rs @@ -137,6 +137,158 @@ enum Format { Responses, } +/// 网关密钥认证（替代服务商 key；上游凭据由受控配置持有）。 +fn unauthorized(state: &AppState, headers: &HeaderMap) -> Option { + let expected = state.api_key.as_ref()?; + (!crate::authentication::authenticate(expected, headers)).then(|| { + ( + StatusCode::UNAUTHORIZED, + [(header::WWW_AUTHENTICATE, "Bearer")], + Json(serde_json::json!({ "error": "invalid_gateway_key" })), + ) + .into_response() + }) +} + +/// Core 复用会话前缀抽取记忆的请求（#2413）。 +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ExtractRequest { + /// 沉淀的 Source 元数据中的 `prefix_handle`。 + handle: String, + /// Core 的抽取指令，作为一条用户消息追加在前缀之后。 + instruction: String, + #[serde(default)] + max_output_tokens: Option, +} + +const MAX_EXTRACT_INSTRUCTION_BYTES: usize = 64 * 1024; + +/// 在保留的前缀后追加 Core 的抽取指令并发送。请求照常经过 before_send/before_release 检查；不注入、不沉淀、 +/// 不保留前缀。前缀不可用时返回 404，由 Core 回退到普通抽取。 +pub async fn extract( + State((state, _)): State, + headers: HeaderMap, + Json(request): Json, +) -> Response { + let error = |status: StatusCode, code: &str| { + (status, Json(serde_json::json!({ "error": code }))).into_response() + }; + if !state.prefix_extraction { + return error(StatusCode::NOT_FOUND, "prefix_extraction_disabled"); + } + if let Some(rejected) = unauthorized(&state, &headers) { + return rejected; + } + if request.instruction.trim().is_empty() + || request.instruction.len() > MAX_EXTRACT_INSTRUCTION_BYTES + { + return error(StatusCode::BAD_REQUEST, "invalid_instruction"); + } + let Some(prefix) = state.executor.prefix(&request.handle, &state.default_scope) else { + return error(StatusCode::NOT_FOUND, "prefix_unavailable"); + }; + if prefix.target_profile != state.upstream_profile { + return error(StatusCode::CONFLICT, "target_mismatch"); + } + let canonical = prefixed_request(&prefix, &request) + .and_then(|body| serde_json::from_value::(body).ok()) + .and_then(|parsed| to_canonical_responses(parsed).ok()); + let Some(canonical) = canonical else { + return error(StatusCode::INTERNAL_SERVER_ERROR, "prefix_invalid"); + }; + let call = ManagedCall { + scope_id: state.default_scope.clone(), + target_profile: state.upstream_profile.clone(), + request: canonical, + target_url: state.upstream_responses.clone(), + headers: bearer_headers(&state.upstream_responses_api_key), + degrade_policy: pc_gateway_core::DegradePolicy::Pause, + inject_context: false, + include_code: false, + capture_prefix: false, + conversation_id: Some(prefix.conversation_id.clone()), + sediment: false, + prepared_contexts: Vec::new(), + }; + let renderer = pc_gateway_protocol::GatewayRenderer::Responses(Default::default()); + match state + .executor + .run( + RequestControl::new( + tokio::time::Instant::now() + std::time::Duration::from_secs(180), + None, + ), + call, + &renderer, + state.evidence.as_ref(), + ) + .await + { + ManagedOutcome::Released { status, body, .. } if (200..300).contains(&status) => { + match serde_json::from_slice::(&body) { + Ok(response) => Json(serde_json::json!({ + "text": responses_output_text(&response), + "usage": response.get("usage").cloned().unwrap_or(serde_json::Value::Null), + "model": response.get("model").cloned().unwrap_or(serde_json::Value::Null), + })) + .into_response(), + Err(_) => error(StatusCode::BAD_GATEWAY, "invalid_upstream_response"), + } + } + outcome => respond(outcome, OutputProtocol::OpenAiResponses), + } +} + +/// 前缀请求：原始请求的 input 依次接上终止响应的输出（去掉 `id`/`status`，与客户端重放的形式一致）和 Core 的 +/// 指令。其余字段（工具、`prompt_cache_key`、推理设置等）不变，只把 `tool_choice` 设为 `none`、改为非流式， +/// 并在 Core 给出时设置输出上限：这些字段不影响前缀缓存。 +fn prefixed_request( + prefix: &pc_gateway_core::PrefixEntry, + request: &ExtractRequest, +) -> Option { + let mut body: serde_json::Value = serde_json::from_slice(&prefix.request_body).ok()?; + let input = body.get_mut("input")?.as_array_mut()?; + for item in &prefix.response_output { + let mut item = item.clone(); + if let Some(fields) = item.as_object_mut() { + fields.remove("id"); + fields.remove("status"); + } + input.push(item); + } + input.push(serde_json::json!({ + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": request.instruction}], + })); + body["tool_choice"] = "none".into(); + body["stream"] = false.into(); + if let Some(limit) = request.max_output_tokens { + body["max_output_tokens"] = limit.into(); + } + Some(body) +} + +/// Responses 响应中助手消息的文本。 +fn responses_output_text(response: &serde_json::Value) -> String { + response + .get("output") + .and_then(|output| output.as_array()) + .into_iter() + .flatten() + .filter(|item| item.get("type").and_then(|t| t.as_str()) == Some("message")) + .flat_map(|item| { + item.get("content") + .and_then(|c| c.as_array()) + .into_iter() + .flatten() + }) + .filter_map(|part| part.get("text").and_then(|t| t.as_str())) + .collect::>() + .join("") +} + async fn ingress_common( state: Arc, headers: HeaderMap, @@ -184,16 +336,8 @@ async fn ingress_model( format: Format, control: RequestControl, ) -> Response { - // 网关密钥认证（替代服务商 key；上游凭据由受控配置持有）。 - if let Some(expected) = &state.api_key { - if !crate::authentication::authenticate(expected, &headers) { - return ( - StatusCode::UNAUTHORIZED, - [(header::WWW_AUTHENTICATE, "Bearer")], - Json(serde_json::json!({ "error": "invalid_gateway_key" })), - ) - .into_response(); - } + if let Some(rejected) = unauthorized(&state, &headers) { + return rejected; } if state.max_body_bytes == 0 { @@ -322,6 +466,7 @@ async fn ingress_model( conversation.is_some(), ), include_code: state.inject_code, + capture_prefix: state.prefix_extraction && matches!(format, Format::Responses), conversation_id: conversation, sediment: state.sediment, prepared_contexts: Vec::new(), @@ -1023,6 +1168,7 @@ mod tests { inject_code: false, sediment: false, steering: None, + prefix_extraction: false, default_scope: "test".into(), max_body_bytes: 1024 * 1024, degradation_allowed: false, @@ -1308,6 +1454,61 @@ mod tests { serde_json::from_slice(&payload.bytes).unwrap() } + #[test] + fn an_extraction_request_extends_the_sent_prefix_unchanged() { + let sent = render_responses(&codex_request(false)); + let output = vec![ + serde_json::json!({"id": "rs_1", "type": "reasoning", "summary": [], "encrypted_content": "gAAAA-x"}), + serde_json::json!({"id": "msg_1", "type": "message", "role": "assistant", "status": "completed", + "content": [{"type": "output_text", "text": "done", "annotations": []}]}), + ]; + let prefix = pc_gateway_core::PrefixEntry { + scope_id: "s".into(), + target_profile: "codex".into(), + conversation_id: "codex-thread:a".into(), + request_body: serde_json::to_vec(&sent).unwrap().into(), + response_output: output, + }; + let request = ExtractRequest { + handle: "h".into(), + instruction: "Extract memories.".into(), + max_output_tokens: Some(2000), + }; + let extract = render_responses(&prefixed_request(&prefix, &request).unwrap()); + + let before = sent["input"].as_array().unwrap(); + let after = extract["input"].as_array().unwrap(); + assert_eq!( + after[..before.len()], + before[..], + "the sent input is kept as is" + ); + // 响应输出按客户端重放的形式接上：去掉 id 与 status。 + assert_eq!( + after[before.len()..], + [ + serde_json::json!({"type": "reasoning", "summary": [], "encrypted_content": "gAAAA-x"}), + serde_json::json!({"type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "done", "annotations": []}]}), + // 与客户端追加的用户消息渲染方式相同。 + serde_json::json!({"role": "user", + "content": [{"type": "input_text", "text": "Extract memories."}]}), + ] + ); + assert_eq!(extract["tool_choice"], "none"); + assert_eq!(extract["max_output_tokens"], 2000); + assert_ne!(extract["stream"], true); + // 影响缓存的其余字段（模型、指令、工具、prompt_cache_key、推理设置）不变。 + let rest = |body: &serde_json::Value| { + let mut fields = body.as_object().unwrap().clone(); + for key in ["input", "tool_choice", "stream", "max_output_tokens"] { + fields.remove(key); + } + fields + }; + assert_eq!(rest(&extract), rest(&sent)); + } + #[test] fn steering_is_rendered_after_the_base_prompt_and_nothing_else_changes() { let client = codex_request(false); diff --git a/gateway/crates/pc-gateway-app/src/main.rs b/gateway/crates/pc-gateway-app/src/main.rs index 1a73031199..8f044df8f0 100644 --- a/gateway/crates/pc-gateway-app/src/main.rs +++ b/gateway/crates/pc-gateway-app/src/main.rs @@ -67,6 +67,8 @@ pub struct AppState { pub sediment: bool, /// 运维配置的固定指令（`PC_GATEWAY_STEERING_FILE`），只插入 Responses 请求；未配置时为空。 pub steering: Option, + /// 为 Core 保留会话前缀以复用缓存抽取记忆：`PC_GATEWAY_PREFIX_EXTRACTION=on`（默认关，#2413）。 + pub prefix_extraction: bool, /// 默认 Scope（启动时读一次，避免热路径读环境变量）。 pub default_scope: String, /// Confirmed model input intersection in bytes; zero pauses model routes. @@ -207,6 +209,10 @@ async fn main() { ), sediment: matches!(std::env::var("PC_GATEWAY_SEDIMENT").as_deref(), Ok("on")), steering, + prefix_extraction: matches!( + std::env::var("PC_GATEWAY_PREFIX_EXTRACTION").as_deref(), + Ok("on") + ), default_scope: std::env::var("PC_GATEWAY_DEFAULT_SCOPE") .unwrap_or_else(|_| "gateway:default".into()), max_body_bytes: model_body_bytes, @@ -241,6 +247,7 @@ fn router(state: Arc, proxy_state: proxy::ProxyState) -> Router { post(ingress::chat_completions), ) .route("/gateway/v1/responses", post(ingress::responses)) + .route("/gateway/v1/extract", post(ingress::extract)) .fallback(proxy::fallback) .with_state((state, proxy_state)) } diff --git a/gateway/crates/pc-gateway-core/Cargo.toml b/gateway/crates/pc-gateway-core/Cargo.toml index f92693d763..7e372f87c6 100644 --- a/gateway/crates/pc-gateway-core/Cargo.toml +++ b/gateway/crates/pc-gateway-core/Cargo.toml @@ -31,3 +31,4 @@ thiserror = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } reqwest = { workspace = true } +getrandom = "0.4.3" diff --git a/gateway/crates/pc-gateway-core/src/capture_coverage.rs b/gateway/crates/pc-gateway-core/src/capture_coverage.rs index 7e0bfd0e56..83e2175414 100644 --- a/gateway/crates/pc-gateway-core/src/capture_coverage.rs +++ b/gateway/crates/pc-gateway-core/src/capture_coverage.rs @@ -16,39 +16,16 @@ use serde_json::{json, Value}; -#[derive(Default)] -pub(crate) struct CaptureLosses { - pub field_byte_limit: bool, - pub transcript_byte_limit: bool, - pub unmeasured_projection: bool, -} - -impl CaptureLosses { - pub fn declaration(&self) -> Value { - let mut reasons = Vec::new(); - if self.field_byte_limit { - reasons.push("field_byte_limit"); - } - if self.transcript_byte_limit { - reasons.push("transcript_byte_limit"); - } - if self.unmeasured_projection { - reasons.push("unmeasured_projection"); - } - declaration("gateway-turn-text.v1", &reasons) - } -} - fn unit(projection: &str) -> Option<&'static str> { match projection { "hermes-turn-text.v1" => Some("user_assistant_text_pair"), "hermes-precompress-text.v1" => Some("new_user_assistant_window"), - "gateway-turn-text.v1" => Some("gateway_turn_projection"), + "gateway-turn-text.v1" | "gateway-turn-text.v2" => Some("gateway_turn_projection"), _ => None, } } -fn declaration(projection: &str, reasons: &[&str]) -> Value { +pub(crate) fn declaration(projection: &str, reasons: &[&str]) -> Value { let mut reasons = reasons.to_vec(); reasons.sort_unstable(); reasons.dedup(); diff --git a/gateway/crates/pc-gateway-core/src/executor.rs b/gateway/crates/pc-gateway-core/src/executor.rs index 4b2066f4da..3d4aebb543 100644 --- a/gateway/crates/pc-gateway-core/src/executor.rs +++ b/gateway/crates/pc-gateway-core/src/executor.rs @@ -30,6 +30,7 @@ use crate::evidence::CallRecorder; use crate::lifecycle::Execution; +use crate::prefix_store::{PrefixEntry, PrefixStore}; use crate::replay_cache::{message_anchor, PreparedBlock, ReplayCache, MAX_REPLAY_MS}; use crate::transcript::turn_transcript; @@ -71,6 +72,8 @@ pub struct ManagedCall { pub include_code: bool, /// 宿主提供的会话 ID（`X-PowerContext-Conversation-ID`）。只有带它的请求才在会话内重放准备结果。 pub conversation_id: Option, + /// 沉淀时保留本回合的上游请求与响应输出，供 Core 复用前缀缓存抽取记忆（#2413，默认关）。 + pub capture_prefix: bool, pub sediment: bool, pub prepared_contexts: Vec, } @@ -143,6 +146,7 @@ pub struct Executor { replays: ReplayCache, max_input_bytes: usize, require_privacy_coverage: bool, + prefixes: PrefixStore, } /// 单次发送尝试的准入凭证。 @@ -166,6 +170,7 @@ where replays: ReplayCache::default(), max_input_bytes: pc_gateway_types::CONTEXT_CHECK_BODY_LIMIT, require_privacy_coverage: true, + prefixes: PrefixStore::default(), } } @@ -285,6 +290,7 @@ where degrade_policy: DegradePolicy::Pause, inject_context: false, include_code: false, + capture_prefix: false, conversation_id: None, sediment: false, prepared_contexts: Vec::new(), @@ -497,23 +503,51 @@ where .map(simple_digest) .unwrap_or_else(|| simple_digest(&release_body)) }); - let capture = CaptureContentSourceRequest { - scope_id: call.scope_id.clone(), - source_id: format!("gw-turn-{digest}"), - content: transcript.text, - metadata: Some(serde_json::json!({ + let source_id = format!("gw-turn-{digest}"); + // 保留前缀供 Core 复用缓存抽取：句柄写在这一回合最后一段 Source 的元数据里。注入的准备块在 + // 前缀里、不在 Source 里，Core 的抽取指令说明标记内的内容不是证据。 + let prefix = self.keep_prefix( + &call, + &payload.bytes, + &release_body, + renderer.message_list_key(), + ); + let count = transcript.parts.len(); + for (index, content) in transcript.parts.into_iter().enumerate() { + let mut metadata = serde_json::json!({ "kind": "gateway_turn", "target_profile": call.target_profile, "powercontext_capture_coverage": transcript.coverage, - })), - }; - match self.gate.capture(capture).await { - Ok(_) => {} - // 同一回合已以不同内容沉淀（重试或重复的终止请求）：先写入的记录保留。 - Err(GateError::GateRejected(409)) => { - evidence.record_not_sent(0, "sediment_duplicate") + }); + let part_id = if count > 1 { + metadata["turn"] = serde_json::json!({ + "source_id": source_id, + "part": index + 1, + "parts": count, + }); + format!("{source_id}-{}of{count}", index + 1) + } else { + source_id.clone() + }; + if index + 1 == count { + for (key, value) in prefix.iter().flatten() { + metadata[*key] = value.clone(); + } + } + let capture = CaptureContentSourceRequest { + scope_id: call.scope_id.clone(), + source_id: part_id, + content, + metadata: Some(metadata), + }; + match self.gate.capture(capture).await { + Ok(_) => {} + // 同一回合已以不同内容沉淀（重试或重复的终止请求）：先写入的记录保留。 + Err(GateError::GateRejected(409)) => { + evidence.record_not_sent(0, "sediment_duplicate") + } + Err(_) => evidence.record_not_sent(0, "sediment_failed:gate_error"), } - Err(_) => evidence.record_not_sent(0, "sediment_failed:gate_error"), } // P1-4 残留：capture 完成后复核——沉淀耗时可能跨越截止时间。 if now_ms() >= release_deadline { @@ -547,6 +581,47 @@ where } } + /// Core 凭句柄取回的前缀（同一 Scope、未过期）。 + pub fn prefix(&self, handle: &str, scope_id: &str) -> Option { + self.prefixes.get(handle, scope_id, now_ms()) + } + + /// 保留一个 Responses 回合的终止请求与响应输出，返回要写入 Source 元数据的句柄和前缀规模。只用于可信会话。 + /// 规模取自终止响应的用量：输入部分在抽取时命中缓存，输出部分作为新输入计费，Core 据此判断复用前缀是否比 + /// 普通抽取便宜。 + fn keep_prefix( + &self, + call: &ManagedCall, + sent: &bytes::Bytes, + release_body: &str, + list_key: &str, + ) -> Option> { + if !call.capture_prefix || list_key != "input" { + return None; + } + let conversation_id = call.conversation_id.clone()?; + let response: serde_json::Value = serde_json::from_str(release_body).ok()?; + let response_output = response.get("output")?.as_array()?.clone(); + let usage = response.get("usage"); + let tokens = |field: &str| usage.and_then(|u| u.get(field)).and_then(|v| v.as_u64()); + let handle = self.prefixes.add( + PrefixEntry { + scope_id: call.scope_id.clone(), + target_profile: call.target_profile.clone(), + conversation_id, + request_body: sent.clone(), + response_output, + }, + now_ms(), + )?; + let mut fields = vec![("prefix_handle", handle.into())]; + if let (Some(input), Some(output)) = (tokens("input_tokens"), tokens("output_tokens")) { + fields.push(("prefix_input_tokens", input.into())); + fields.push(("prefix_output_tokens", output.into())); + } + Some(fields) + } + /// 有可信会话 ID 时：同一消息锚点（工具续写、重试）精确重放已准备的内容与绑定，不再检索； /// 新锚点（新的用户消息）准备一次。没有会话 ID 时每个请求各自准备，不继承任何旧注入。 /// 准备失败不写入，下一次请求重试，且不阻塞本次请求；空结果同样重放。 diff --git a/gateway/crates/pc-gateway-core/src/lib.rs b/gateway/crates/pc-gateway-core/src/lib.rs index 0f8af9b2c9..749d1d1991 100644 --- a/gateway/crates/pc-gateway-core/src/lib.rs +++ b/gateway/crates/pc-gateway-core/src/lib.rs @@ -30,6 +30,7 @@ mod executor; mod gate_client; mod http_gate_client; mod lifecycle; +mod prefix_store; mod replay_cache; mod transcript; mod uds_gate_client; @@ -38,6 +39,7 @@ pub use executor::*; pub use gate_client::*; pub use http_gate_client::*; pub use lifecycle::*; +pub use prefix_store::{PrefixEntry, PREFIX_TTL_MS}; pub use uds_gate_client::*; mod evidence; diff --git a/gateway/crates/pc-gateway-core/src/prefix_store.rs b/gateway/crates/pc-gateway-core/src/prefix_store.rs new file mode 100644 index 0000000000..520915f38b --- /dev/null +++ b/gateway/crates/pc-gateway-core/src/prefix_store.rs @@ -0,0 +1,115 @@ +// Copyright (c) 2026 OceanBase. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! 记忆抽取复用的会话前缀（#2413）。 +//! +//! 每个用户回合的终止响应放行后，网关在内存中保留这次发给上游的原始请求和响应输出，用一个随机句柄引用。 +//! Core 抽取记忆时凭句柄请网关在这段前缀后追加抽取指令再发送，这样请求几乎全部命中服务商为 Agent 写下的 +//! 前缀缓存。前缀只是临时的缓存线索：只在网关内存里，有上限，按时间过期；从不进入 Source 或证据。 + +use std::collections::VecDeque; +use std::sync::Mutex; + +/// 保留的前缀数上限：超出后淘汰最早的。 +const MAX_PREFIXES: usize = 64; +/// 前缀保留时长：Core 的后台抽取通常在回合结束后几分钟内运行，服务商缓存也只保留这么久。 +pub const PREFIX_TTL_MS: u64 = 30 * 60 * 1000; + +/// 一个用户回合的前缀。 +#[derive(Clone, Debug)] +pub struct PrefixEntry { + pub scope_id: String, + pub target_profile: String, + pub conversation_id: String, + /// 发给上游的原始请求体（终止响应对应的那次请求）。 + pub request_body: bytes::Bytes, + /// 终止响应的 output 条目。 + pub response_output: Vec, +} + +#[derive(Default)] +pub struct PrefixStore { + entries: Mutex>, +} + +impl PrefixStore { + /// 保存一个前缀，返回 128 位随机句柄；取不到随机数时不保存。 + pub fn add(&self, entry: PrefixEntry, now_ms: u64) -> Option { + let mut random = [0u8; 16]; + getrandom::fill(&mut random).ok()?; + let handle: String = random.iter().map(|b| format!("{b:02x}")).collect(); + let mut entries = lock(&self.entries); + entries.retain(|(_, _, expires)| *expires > now_ms); + if entries.len() >= MAX_PREFIXES { + entries.pop_front(); + } + entries.push_back((handle.clone(), entry, now_ms + PREFIX_TTL_MS)); + Some(handle) + } + + /// 句柄对应的前缀；句柄未知、已过期或属于其他 Scope 时为 `None`。 + pub fn get(&self, handle: &str, scope_id: &str, now_ms: u64) -> Option { + lock(&self.entries) + .iter() + .find(|(h, entry, expires)| { + h == handle && entry.scope_id == scope_id && *expires > now_ms + }) + .map(|(_, entry, _)| entry.clone()) + } +} + +fn lock(m: &Mutex) -> std::sync::MutexGuard<'_, T> { + m.lock().unwrap_or_else(|e| e.into_inner()) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn entry(scope: &str) -> PrefixEntry { + PrefixEntry { + scope_id: scope.into(), + target_profile: "codex".into(), + conversation_id: "codex-thread:a".into(), + request_body: bytes::Bytes::from_static(b"{}"), + response_output: Vec::new(), + } + } + + #[test] + fn handles_are_random_scoped_and_expire() { + let store = PrefixStore::default(); + let first = store.add(entry("s"), 0).unwrap(); + let second = store.add(entry("s"), 0).unwrap(); + assert_ne!(first, second); + assert_eq!(first.len(), 32); + assert!(store.get(&first, "s", 1).is_some()); + assert!( + store.get(&first, "other", 1).is_none(), + "another Scope cannot use it" + ); + assert!(store.get("unknown", "s", 1).is_none()); + assert!(store.get(&first, "s", PREFIX_TTL_MS).is_none(), "expired"); + } + + #[test] + fn the_store_is_bounded() { + let store = PrefixStore::default(); + let oldest = store.add(entry("s"), 0).unwrap(); + for _ in 0..MAX_PREFIXES { + store.add(entry("s"), 0); + } + assert!(store.get(&oldest, "s", 1).is_none()); + } +} diff --git a/gateway/crates/pc-gateway-core/src/transcript.rs b/gateway/crates/pc-gateway-core/src/transcript.rs index f049a3d592..0c21c76ac0 100644 --- a/gateway/crates/pc-gateway-core/src/transcript.rs +++ b/gateway/crates/pc-gateway-core/src/transcript.rs @@ -12,28 +12,66 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! 一个用户回合的规范化记录，供网关在回合结束时沉淀一次。 +//! 一个用户回合的规范化记录，供网关在回合结束时沉淀。 //! -//! 回合从最后一条用户消息开始，到不再调用工具的响应（终止响应）结束。记录只保留模型可见的对话： -//! 用户文本、工具调用（名称与参数）、工具输出和助手文本；推理密文、各类 ID、用量和其他传输元数据都不进入记录。 -//! 用户文本取自注入前的请求，不含准备结果。 +//! 回合从最后一条用户消息开始，到不再调用工具的响应（终止响应）结束。记录完整保留这一回合里模型读到的内容： +//! 用户文本、其后的其他消息、工具调用的全部参数字段、工具输出与错误、推理的可见摘要、无法识别的条目，以及助手 +//! 文本；字段不截断，步骤不省略。只有推理密文、各类 ID、状态、用量等传输元数据不进入记录，二进制数据（截图、 +//! 图片、文件）记为带字节数的占位文本。用户文本取自注入前的请求，不含准备结果。 +//! +//! 每个 Source 都要能被 Core 的普通抽取单独处理：估算规模超过 [`MAX_SOURCE_TOKENS`] 时，记录按顺序拆成几段， +//! 每段都是独立的 Source。 -use crate::capture_coverage::CaptureLosses; +use crate::capture_coverage::declaration; use pc_gateway_protocol::{CanonicalMessage, CanonicalRequest, Role}; use serde::Serialize; -use serde_json::Value; - -/// 单个字段（用户文本、工具参数、工具输出、助手文本）的字节上限，超出部分截断。 -pub(crate) const MAX_FIELD_BYTES: usize = 8 * 1024; -/// 整条记录的字节上限：超出时从最早的步骤开始省略。 -pub(crate) const MAX_TRANSCRIPT_BYTES: usize = 64 * 1024; +use serde_json::{Map, Value}; + +/// 投影名称，见捕获覆盖声明契约。 +pub(crate) const PROJECTION: &str = "gateway-turn-text.v2"; +/// 一段 Source 的估算 token 上限。Core 的 Atomic Memory 抽取默认只接受 24,000 个估算 token 的输入，其中还有 +/// 指令和输出格式；Source 正文在输入里是转义后的 JSON 字符串，这里按转义后的文本估算，留出余量。 +pub(crate) const MAX_SOURCE_TOKENS: usize = 12_000; +/// Core 的字符估算：ASCII 字符 4 个单位，其他字符 16 个单位，每 16 个单位一个 token（与 +/// `character_token_estimator` 一致：ASCII 4 字符一个 token，非 ASCII 1 字符一个 token）。 +const UNITS_PER_TOKEN: usize = 16; + +/// 文本作为 JSON 字符串转义后在 Core 估算中的单位数。 +fn units(text: &str) -> usize { + serde_json::to_string(text) + .unwrap_or_default() + .chars() + .map(|c| if c.is_ascii() { 4 } else { 16 }) + .sum() +} #[derive(Debug, PartialEq, Serialize)] #[serde(tag = "type", rename_all = "snake_case")] pub(crate) enum Step { - ToolCall { name: String, arguments: String }, - ToolOutput { output: String }, - Assistant { text: String }, + ToolCall { + name: String, + arguments: String, + }, + ToolOutput { + output: String, + }, + Assistant { + text: String, + }, + /// 用户消息之后、非助手角色的消息（如 developer、system）。 + Message { + role: String, + text: String, + }, + /// 推理的可见摘要；推理密文不进入记录。 + ReasoningSummary { + text: String, + }, + /// 无法识别类型的条目，去掉 ID、状态和二进制数据后原样记录。 + Item { + item_type: String, + content: String, + }, } #[derive(Debug, Serialize)] @@ -41,24 +79,18 @@ struct Transcript { kind: &'static str, user: String, steps: Vec, - #[serde(skip_serializing_if = "is_zero")] - omitted_steps: usize, assistant: String, } -fn is_zero(n: &usize) -> bool { - *n == 0 -} - /// 渲染后消息列表所在的顶层键决定线格式：`input` 为 Responses，其余为 Chat。 fn is_responses(list_key: &str) -> bool { list_key == "input" } -/// 正文与同一次投影计算观察到的覆盖事实。 +/// 回合的记录：放得下时是一段，否则按顺序拆成几段；以及投影的覆盖声明。 #[derive(Debug, PartialEq)] pub(crate) struct TurnProjection { - pub text: String, + pub parts: Vec, pub coverage: Value, } @@ -69,10 +101,9 @@ pub(crate) fn turn_transcript( response: &str, list_key: &str, ) -> Option { - let mut losses = CaptureLosses::default(); let response: Value = serde_json::from_str(response).ok()?; let (final_steps, assistant) = if is_responses(list_key) { - responses_final(&response, &mut losses)? + responses_final(&response)? } else { (Vec::new(), chat_final_text(&response)?) }; @@ -85,41 +116,121 @@ pub(crate) fn turn_transcript( .iter() .flat_map(|m| { if is_responses(list_key) { - if m.content.len() != 1 { - losses.unmeasured_projection = true; - } - m.content - .first() - .map(|item| item_steps(item, &mut losses)) - .unwrap_or_default() + m.content.iter().flat_map(item_steps).collect() } else { - chat_steps(m, &mut losses) + chat_steps(m) } }) .collect(); steps.extend(final_steps); - let mut transcript = Transcript { + let transcript = Transcript { kind: "gateway_turn", - user: bounded(user, &mut losses), - steps: Vec::new(), - omitted_steps: 0, - assistant: bounded(assistant, &mut losses), + user, + steps, + assistant, }; - loop { - transcript.steps = std::mem::take(&mut steps); - let text = serde_json::to_string(&transcript).ok()?; - if text.len() <= MAX_TRANSCRIPT_BYTES || transcript.steps.is_empty() { - return Some(TurnProjection { - text, - coverage: losses.declaration(), - }); + Some(TurnProjection { + parts: split(&transcript)?, + coverage: declaration(PROJECTION, &[]), + }) +} + +/// 整条记录放得下时原样返回一段。否则拆成按顺序编号的几段，每段的估算规模都不超过 [`MAX_SOURCE_TOKENS`]：每段是 +/// `{"kind": "gateway_turn", "part": i, "parts": n, "steps": [...]}`，用户文本作为第一段的第一个步骤（`user`）， +/// 助手文本作为最后一段的最后一个步骤；单个步骤放不下时，它最长的文本字段按字符拆成连续的几个同类步骤， +/// 后续部分带 `"continued": true`。 +fn split(transcript: &Transcript) -> Option> { + let budget = MAX_SOURCE_TOKENS * UNITS_PER_TOKEN; + let whole = serde_json::to_string(transcript).ok()?; + if units(&whole) <= budget { + return Some(vec![whole]); + } + let mut entries = vec![serde_json::json!({"type": "user", "text": transcript.user})]; + for step in &transcript.steps { + entries.push(serde_json::to_value(step).ok()?); + } + entries.push(serde_json::json!({"type": "assistant", "text": transcript.assistant})); + // 段的外层（kind、part、parts、括号与分隔符）留出余量。 + let budget = budget - 200 * 16; + let pieces: Vec = entries + .into_iter() + .flat_map(|entry| fit(entry, budget)) + .collect(); + let mut groups: Vec> = vec![Vec::new()]; + let mut used = 0; + for piece in pieces { + let size = units(&piece.to_string()) + 4; + if used + size > budget && !groups.last().is_some_and(Vec::is_empty) { + groups.push(Vec::new()); + used = 0; } - steps = std::mem::take(&mut transcript.steps); - let drop = (steps.len() / 4).max(1); - steps.drain(..drop); - transcript.omitted_steps += drop; - losses.transcript_byte_limit = true; + used += size; + groups.last_mut()?.push(piece); } + let count = groups.len(); + groups + .into_iter() + .enumerate() + .map(|(index, steps)| { + serde_json::to_string(&serde_json::json!({ + "kind": "gateway_turn", + "part": index + 1, + "parts": count, + "steps": steps, + })) + .ok() + }) + .collect() +} + +/// 把一个放不下的步骤按其最长的字符串字段拆成连续的几个步骤，每个的估算单位数都不超过 `budget`。 +fn fit(entry: Value, budget: usize) -> Vec { + if units(&entry.to_string()) < budget { + return vec![entry]; + } + let Some(fields) = entry.as_object() else { + return vec![entry]; + }; + let Some((key, text)) = fields + .iter() + .filter_map(|(key, value)| value.as_str().map(|text| (key.clone(), text.to_owned()))) + .max_by_key(|(_, text)| text.len()) + else { + return vec![entry]; + }; + let mut shell = fields.clone(); + shell.insert(key.clone(), Value::from("")); + shell.insert("continued".into(), Value::from(true)); + let room = budget + .saturating_sub(units(&Value::Object(shell).to_string())) + .max(1); + // 每个字符在步骤 JSON 里转义一次、在 Source 字符串里再转义一次后的单位数。 + let cost = |c: char| units(&Value::from(c.to_string()).to_string()) - units("\"\""); + let mut chunks = vec![String::new()]; + let mut used = 0; + for c in text.chars() { + let size = cost(c); + if used + size > room && !chunks.last().is_some_and(String::is_empty) { + chunks.push(String::new()); + used = 0; + } + used += size; + if let Some(chunk) = chunks.last_mut() { + chunk.push(c); + } + } + chunks + .into_iter() + .enumerate() + .map(|(index, chunk)| { + let mut piece = fields.clone(); + piece.insert(key.clone(), Value::from(chunk)); + if index > 0 { + piece.insert("continued".into(), Value::from(true)); + } + Value::Object(piece) + }) + .collect() } /// Chat 终止响应：第一个 choice 的消息既没有 `tool_calls`，也没有旧式的 `function_call`。 @@ -140,7 +251,9 @@ fn chat_final_text(response: &Value) -> Option { #[derive(Debug, PartialEq)] enum ItemKind { Message, - /// 不进入记录、也不影响回合是否结束的条目：推理与 MCP 工具清单。 + /// 推理条目：密文不进入记录，可见摘要记为步骤；不影响回合是否结束。 + Reasoning, + /// MCP 工具清单：不进入记录，也不影响回合是否结束。 Ignored, /// 服务商在同一响应内执行完毕的内置工具，结果随条目返回，客户端不需要再发请求。 HostedCall, @@ -166,7 +279,8 @@ fn classify(item: &Value) -> ItemKind { .unwrap_or("message") { "message" => ItemKind::Message, - "reasoning" | "mcp_list_tools" => ItemKind::Ignored, + "reasoning" => ItemKind::Reasoning, + "mcp_list_tools" => ItemKind::Ignored, kind if HOSTED_CALLS.contains(&kind) => ItemKind::HostedCall, kind if kind.ends_with("_call_output") || kind == "mcp_approval_response" => { ItemKind::CallOutput @@ -176,17 +290,18 @@ fn classify(item: &Value) -> ItemKind { } } -/// Responses 终止响应：output 只含助手消息、可忽略的条目和已在响应内完成的内置工具。 +/// Responses 终止响应：output 只含助手消息、推理、可忽略的条目和已在响应内完成的内置工具。 /// 需要客户端执行的调用或无法识别的条目都表示回合尚未结束。 -fn responses_final(response: &Value, losses: &mut CaptureLosses) -> Option<(Vec, String)> { +fn responses_final(response: &Value) -> Option<(Vec, String)> { let output = response.get("output")?.as_array()?; let mut steps = Vec::new(); let mut texts = Vec::new(); for item in output { match classify(item) { ItemKind::Message => texts.extend(assistant_text(item)), - ItemKind::Ignored => {} - ItemKind::HostedCall => steps.extend(item_steps(item, losses)), + ItemKind::Reasoning | ItemKind::Ignored | ItemKind::HostedCall => { + steps.extend(item_steps(item)) + } ItemKind::ClientCall | ItemKind::CallOutput | ItemKind::Unknown => return None, } } @@ -198,58 +313,97 @@ fn assistant_text(item: &Value) -> Option { .then(|| content_text(item.get("content"))) } -fn chat_steps(message: &CanonicalMessage, losses: &mut CaptureLosses) -> Vec { +fn chat_steps(message: &CanonicalMessage) -> Vec { match message.role { Role::Assistant => { let mut steps = Vec::new(); let text = parts_text(&message.content); if !text.is_empty() { - steps.push(Step::Assistant { - text: bounded(text, losses), - }); + steps.push(Step::Assistant { text }); } for call in message.tool_calls.iter().flatten() { let function = call.get("function"); - if call - .get("type") - .is_some_and(|kind| kind.as_str() != Some("function")) - || function - .and_then(|value| value.get("name")) - .and_then(Value::as_str) - .is_none() - || function - .and_then(|value| value.get("arguments")) - .and_then(Value::as_str) - .is_none() - { - losses.unmeasured_projection = true; - } - steps.push(Step::ToolCall { - name: string_field(function, "name"), - arguments: bounded(string_field(function, "arguments"), losses), - }); + let named = function + .and_then(|value| value.get("name")) + .and_then(Value::as_str); + let arguments = function.and_then(|value| value.get("arguments")); + // 标准函数调用记名称与参数字符串；其他形态的调用整体记录。 + let (name, arguments) = match (named, arguments.and_then(Value::as_str)) { + (Some(name), Some(arguments)) => (name.to_owned(), arguments.to_owned()), + _ => ( + named + .or_else(|| call.get("type").and_then(Value::as_str)) + .unwrap_or_default() + .to_owned(), + without_binary(&drop_keys(call, &["id"])).to_string(), + ), + }; + steps.push(Step::ToolCall { name, arguments }); } steps } Role::Tool => vec![Step::ToolOutput { - output: bounded(parts_text(&message.content), losses), + output: parts_text(&message.content), }], - _ => Vec::new(), + Role::User => Vec::new(), + _ => { + let text = parts_text(&message.content); + (!text.is_empty()) + .then(|| Step::Message { + role: role_name(&message.role), + text, + }) + .into_iter() + .collect() + } } } -/// 一个 Responses 条目在记录中的步骤。工具调用取名称（没有名称时用条目类型）与参数；内置工具随条目返回的结果 -/// 作为工具输出。条目 ID、`call_id`、状态等字段不进入记录。 -fn item_steps(item: &Value, losses: &mut CaptureLosses) -> Vec { +fn role_name(role: &Role) -> String { + serde_json::to_value(role) + .ok() + .and_then(|value| value.as_str().map(str::to_owned)) + .unwrap_or_default() +} + +/// 调用条目里不属于参数的字段：标识、状态、名称和随条目返回的结果。 +const CALL_METADATA: &[&str] = &["id", "call_id", "status", "type", "name"]; +/// 内置工具随条目返回的结果字段。 +const CALL_RESULTS: &[&str] = &["output", "outputs", "results", "error"]; + +/// 一个 Responses 条目在记录中的步骤。工具调用取名称（没有名称时用条目类型）和全部参数字段：只有一个参数字段时 +/// 记它的文本，否则记这些字段组成的 JSON 对象；内置工具随条目返回的结果与错误作为工具输出。条目 ID、`call_id`、 +/// 状态等字段不进入记录。 +fn item_steps(item: &Value) -> Vec { match classify(item) { - ItemKind::Message => assistant_text(item) - .filter(|text| !text.is_empty()) - .map(|text| { - vec![Step::Assistant { - text: bounded(text, losses), + ItemKind::Message => { + let role = item + .get("role") + .and_then(Value::as_str) + .unwrap_or("assistant"); + let text = content_text(item.get("content")); + if text.is_empty() { + Vec::new() + } else if role == "assistant" { + vec![Step::Assistant { text }] + } else { + vec![Step::Message { + role: role.to_owned(), + text, }] - }) - .unwrap_or_default(), + } + } + ItemKind::Reasoning => { + let text = item + .get("summary") + .and_then(Value::as_array) + .map(|parts| parts_text(parts)) + .unwrap_or_default(); + (!text.is_empty()) + .then_some(Step::ReasoningSummary { text }) + .into_iter() + .collect() + } ItemKind::ClientCall | ItemKind::HostedCall => { let name = item .get("name") @@ -257,79 +411,78 @@ fn item_steps(item: &Value, losses: &mut CaptureLosses) -> Vec { .or_else(|| item.get("type").and_then(Value::as_str)) .unwrap_or_default() .to_owned(); - let arguments = first_field( - item, - &["arguments", "input", "action", "code", "query", "queries"], - ); - // 后缀分类只决定工具循环是否结束，不证明 v1 能测量该调用的载荷。 - let known_call = item - .get("type") - .and_then(Value::as_str) - .is_some_and(|kind| { - HOSTED_CALLS.contains(&kind) - || matches!( - kind, - "function_call" - | "custom_tool_call" - | "computer_call" - | "local_shell_call" - | "shell_call" - | "mcp_approval_request" - ) - }); - if !known_call || arguments.is_none() { - losses.unmeasured_projection = true; - } - let arguments = arguments.map(value_text).unwrap_or_default(); + let excluded: Vec<&str> = CALL_METADATA.iter().chain(CALL_RESULTS).copied().collect(); let mut steps = vec![Step::ToolCall { name, - arguments: bounded(arguments, losses), + arguments: fields_text(&drop_keys(item, &excluded)), }]; - if let Some(output) = first_field(item, &["output", "outputs", "results"]) { + let results = keep_keys(item, CALL_RESULTS); + if results.as_object().is_some_and(|fields| !fields.is_empty()) { steps.push(Step::ToolOutput { - output: bounded(value_text(output), losses), + output: fields_text(&results), }); } steps } - ItemKind::CallOutput => { - let output = item.get("output"); - let known_output = matches!( - item.get("type").and_then(Value::as_str), - Some( - "function_call_output" - | "custom_tool_call_output" - | "computer_call_output" - | "local_shell_call_output" - | "shell_call_output" - ) - ); - // MCP 审批的 approve/reason 和未知输出载荷不在现有 output 投影中。 - if !known_output || output.is_none_or(Value::is_null) { - losses.unmeasured_projection = true; - } - vec![Step::ToolOutput { - output: bounded(output.map(value_text).unwrap_or_default(), losses), - }] - } + ItemKind::CallOutput => vec![Step::ToolOutput { + output: fields_text(&drop_keys(item, &["id", "call_id", "status", "type"])), + }], ItemKind::Ignored => Vec::new(), - ItemKind::Unknown => { - losses.unmeasured_projection = true; - Vec::new() - } + ItemKind::Unknown => vec![Step::Item { + item_type: item + .get("type") + .and_then(Value::as_str) + .unwrap_or_default() + .to_owned(), + content: without_binary(&drop_keys(item, &["id", "status", "type"])).to_string(), + }], } } -fn first_field<'a>(item: &'a Value, keys: &[&str]) -> Option<&'a Value> { - keys.iter() - .filter_map(|key| item.get(*key)) - .find(|value| !value.is_null()) +/// 一组字段的文本：只有一个非空字段时记它的文本，否则记整个对象（二进制数据换成占位文本）。 +fn fields_text(fields: &Value) -> String { + let present: Vec<&Value> = fields + .as_object() + .map(|map| map.values().filter(|value| !value.is_null()).collect()) + .unwrap_or_default(); + match present.as_slice() { + [] => String::new(), + [value] => value_text(value), + _ => without_binary(fields).to_string(), + } +} + +fn drop_keys(item: &Value, keys: &[&str]) -> Value { + let fields: Map = item + .as_object() + .map(|map| { + map.iter() + .filter(|(key, _)| !keys.contains(&key.as_str())) + .map(|(key, value)| (key.clone(), value.clone())) + .collect() + }) + .unwrap_or_default(); + Value::Object(fields) +} + +fn keep_keys(item: &Value, keys: &[&str]) -> Value { + let fields: Map = item + .as_object() + .map(|map| { + map.iter() + .filter(|(key, value)| keys.contains(&key.as_str()) && !value.is_null()) + .map(|(key, value)| (key.clone(), value.clone())) + .collect() + }) + .unwrap_or_default(); + Value::Object(fields) } /// 字段的可读文本：字符串原样，内容块数组取其文本；其余结构序列化为 JSON，其中的 `data:` URL /// （截图、图片等二进制内容）替换为占位文本。 fn value_text(value: &Value) -> String { match value { + Value::String(text) if text.starts_with("data:") => binary_placeholder(text), Value::String(text) => text.clone(), Value::Array(parts) if !parts_text(parts).is_empty() => parts_text(parts), other => without_binary(other).to_string(), @@ -338,7 +491,7 @@ fn value_text(value: &Value) -> String { fn without_binary(value: &Value) -> Value { match value { - Value::String(text) if text.starts_with("data:") => Value::from("[binary data omitted]"), + Value::String(text) if text.starts_with("data:") => Value::from(binary_placeholder(text)), Value::Array(items) => Value::Array(items.iter().map(without_binary).collect()), Value::Object(map) => Value::Object( map.iter() @@ -349,12 +502,13 @@ fn without_binary(value: &Value) -> Value { } } -fn string_field(value: Option<&Value>, key: &str) -> String { - value - .and_then(|v| v.get(key)) - .and_then(Value::as_str) - .unwrap_or_default() - .to_owned() +fn binary_placeholder(data_url: &str) -> String { + let media = data_url + .strip_prefix("data:") + .and_then(|rest| rest.split([';', ',']).next()) + .filter(|media| !media.is_empty()) + .unwrap_or("binary"); + format!("[{media} data omitted: {} bytes]", data_url.len()) } /// 字符串或内容块数组中的文本。 @@ -366,54 +520,43 @@ fn content_text(content: Option<&Value>) -> String { } } +/// 内容块的文本。图片、文件等非文本块记为占位文本：有 URL 或文件名时带上，内联数据只记字节数。 fn parts_text(parts: &[Value]) -> String { - let texts: Vec<&str> = parts + let texts: Vec = parts .iter() .filter_map(|part| match part { - Value::String(text) => Some(text.as_str()), - _ if matches!( - part.get("type").and_then(Value::as_str), - Some("text" | "input_text" | "output_text") - ) => - { - part.get("text").and_then(Value::as_str) - } - _ => None, + Value::String(text) => Some(text.clone()), + _ => match part.get("type").and_then(Value::as_str) { + Some("text" | "input_text" | "output_text" | "summary_text") => { + part.get("text").and_then(Value::as_str).map(str::to_owned) + } + Some(kind) => Some(match first_text(part, &["image_url", "file_url", "url"]) { + Some(url) if url.starts_with("data:") => binary_placeholder(&url), + Some(url) => format!("[{kind}: {url}]"), + None => match first_text(part, &["filename", "file_id"]) { + Some(name) => format!("[{kind}: {name}]"), + None => format!("[{kind}]"), + }, + }), + None => None, + }, }) .collect(); texts.join("\n") } -/// 截断到 [`MAX_FIELD_BYTES`]（按字符边界），并注明省略的字节数。 -fn bounded(text: String, losses: &mut CaptureLosses) -> String { - if text.len() <= MAX_FIELD_BYTES { - return text; - } - losses.field_byte_limit = true; - let mut end = MAX_FIELD_BYTES; - while !text.is_char_boundary(end) { - end -= 1; - } - format!("{}\n[truncated {} bytes]", &text[..end], text.len() - end) +fn first_text(part: &Value, keys: &[&str]) -> Option { + keys.iter().find_map(|key| match part.get(*key) { + Some(Value::String(text)) => Some(text.clone()), + Some(Value::Object(inner)) => inner.get("url").and_then(Value::as_str).map(str::to_owned), + _ => None, + }) } #[cfg(test)] mod tests { use super::*; - #[test] - fn long_fields_are_truncated_on_a_character_boundary() { - let text = "é".repeat(MAX_FIELD_BYTES); - let mut losses = CaptureLosses::default(); - let out = bounded(text.clone(), &mut losses); - assert!(out.len() < text.len()); - assert!(out.ends_with(&format!( - "[truncated {} bytes]", - text.len() - MAX_FIELD_BYTES - ))); - assert_eq!(bounded("short".into(), &mut losses), "short"); - } - fn chat_turn(outputs: &[String]) -> CanonicalRequest { let message = |role, content: Vec, tool_calls| CanonicalMessage { role, @@ -446,256 +589,182 @@ mod tests { .unwrap() } - const FINAL: &str = r#"{"choices":[{"message":{"role":"assistant","content":"done"}}]}"#; - - #[test] - fn coverage_tracks_real_losses_including_fields_in_omitted_steps() { - for (outputs, reasons) in [ - (vec!["[truncated 999 bytes]".to_owned()], vec![]), - (vec!["文".repeat(MAX_FIELD_BYTES)], vec!["field_byte_limit"]), - ( - vec!["x".repeat(MAX_FIELD_BYTES); 40], - vec!["transcript_byte_limit"], - ), - ( - vec!["x".repeat(MAX_FIELD_BYTES + 1); 40], - vec!["field_byte_limit", "transcript_byte_limit"], - ), - ] { - let projection = turn_transcript(&chat_turn(&outputs), FINAL, "messages").unwrap(); - assert_eq!(projection.coverage["reasons"], serde_json::json!(reasons)); - assert_eq!( - projection.coverage["status"], - if reasons.is_empty() { - "complete" - } else { - "partial" - } - ); - let body: Value = serde_json::from_str(&projection.text).unwrap(); - assert_eq!(body["user"], "task"); - assert_eq!(body["assistant"], "done"); - assert_eq!(projection.coverage["unit"], "gateway_turn_projection"); - } - - let mut outputs = vec!["x".repeat(MAX_FIELD_BYTES + 1)]; - outputs.extend(vec!["y".repeat(MAX_FIELD_BYTES); 40]); - let projection = turn_transcript(&chat_turn(&outputs), FINAL, "messages").unwrap(); - assert!(!projection.text.contains("[truncated")); - assert_eq!( - projection.coverage["reasons"], - serde_json::json!(["field_byte_limit", "transcript_byte_limit"]) - ); - } - - #[test] - fn user_and_terminal_assistant_field_limits_are_measured() { - for (user, assistant) in [ - ("文".repeat(MAX_FIELD_BYTES), "done".into()), - ("task".into(), "x".repeat(MAX_FIELD_BYTES + 1)), - ] { - let mut request = chat_turn(&[]); - request.messages[0].content = vec![Value::from(user)]; - let response = serde_json::json!({"choices":[{"message":{"role":"assistant","content":assistant}}]}).to_string(); - let projection = turn_transcript(&request, &response, "messages").unwrap(); - assert_eq!( - projection.coverage["reasons"], - serde_json::json!(["field_byte_limit"]) - ); + fn item(value: Value) -> CanonicalMessage { + CanonicalMessage { + role: Role::Tool, + content: vec![value], + content_null: false, + tool_calls: None, + extra_fields: Default::default(), + tool_call_id: None, } } - #[test] - fn unprojectable_chat_calls_are_unmeasured_without_changing_the_body() { - for call in [ - serde_json::json!({"type":"custom", "custom":{"name":"exec", "input":"list files"}}), - serde_json::json!({"type":"future", "function":{"name":"exec", "arguments":"{}"}}), - serde_json::json!({"type":"function", "function":{"name":"exec", "arguments":{}}}), - serde_json::json!({"type":"function", "function":{"arguments":"{}"}}), - ] { - let mut request = chat_turn(&["done".into()]); - request.messages[1].tool_calls = Some(vec![call.clone()]); - let projection = turn_transcript(&request, FINAL, "messages").unwrap(); - assert_eq!(projection.coverage["status"], "unknown"); - assert_eq!( - projection.coverage["reasons"], - serde_json::json!(["unmeasured_projection"]) - ); - let body: Value = serde_json::from_str(&projection.text).unwrap(); - assert_eq!( - body["steps"][0]["name"], - string_field(call.get("function"), "name") - ); - assert_eq!( - body["steps"][0]["arguments"], - string_field(call.get("function"), "arguments") - ); + const FINAL: &str = r#"{"choices":[{"message":{"role":"assistant","content":"done"}}]}"#; + const RESPONSES_FINAL: &str = r#"{"output":[{"type":"message","role":"assistant","content":[{"type":"output_text","text":"done"}]}]}"#; - request.messages[0].content = vec![Value::from("x".repeat(MAX_FIELD_BYTES + 1))]; - let partial = turn_transcript(&request, FINAL, "messages").unwrap(); - assert_eq!(partial.coverage["status"], "partial"); - assert_eq!( - partial.coverage["reasons"], - serde_json::json!(["field_byte_limit", "unmeasured_projection"]) - ); - } + fn single(projection: &TurnProjection) -> Value { + assert_eq!(projection.parts.len(), 1); + serde_json::from_str(&projection.parts[0]).unwrap() } #[test] - fn accepted_unknown_history_is_unmeasured_but_unknown_final_stays_uncaptured() { - let mut request = chat_turn(&[]); - let mut item = request.messages[0].clone(); - item.role = Role::Tool; - item.content = vec![serde_json::json!({"type":"future_history_item"})]; - request.messages.push(item); - let projection = turn_transcript( - &request, - r#"{"output":[{"role":"assistant","type":"message","content":"done"}]}"#, - "input", - ) - .unwrap(); - assert_eq!(projection.coverage["status"], "unknown"); - assert_eq!( - projection.coverage["reasons"], - serde_json::json!(["unmeasured_projection"]) - ); - assert_eq!( - projection.text, - r#"{"kind":"gateway_turn","user":"task","steps":[],"assistant":"done"}"# - ); - request.messages[0].content = vec![Value::from("x".repeat(MAX_FIELD_BYTES + 1))]; - let partial = turn_transcript( - &request, - r#"{"output":[{"role":"assistant","type":"message","content":"done"}]}"#, - "input", - ) - .unwrap(); - assert_eq!(partial.coverage["status"], "partial"); + fn a_turn_is_recorded_in_full_and_declares_the_complete_projection() { + let long = "文".repeat(10_000); + let projection = + turn_transcript(&chat_turn(std::slice::from_ref(&long)), FINAL, "messages").unwrap(); + let transcript = single(&projection); + assert_eq!(transcript["user"], "task"); assert_eq!( - partial.coverage["reasons"], - serde_json::json!(["field_byte_limit", "unmeasured_projection"]) + transcript["steps"][1]["output"], long, + "no field is truncated" ); + assert_eq!(transcript["assistant"], "done"); assert_eq!( - turn_transcript( - &request, - r#"{"output":[{"type":"future_history_item"}]}"#, - "input" - ), - None + projection.coverage, + serde_json::json!({ + "schema": "powercontext.capture-coverage.v1", "projection": "gateway-turn-text.v2", + "unit": "gateway_turn_projection", "status": "complete", "reasons": [] + }) ); } #[test] - fn responses_tool_fields_share_the_same_loss_facts() { - let response = serde_json::json!({"output":[ - {"type":"reasoning", "encrypted_content":"opaque"}, - {"type":"web_search_call", "query":"x".repeat(MAX_FIELD_BYTES + 1), "results":"文".repeat(MAX_FIELD_BYTES)}, - {"type":"message", "role":"assistant", "content":[{"type":"output_text", "text":"done"}]} - ]}).to_string(); - let projection = turn_transcript(&chat_turn(&[]), &response, "input").unwrap(); - assert_eq!(projection.coverage["status"], "partial"); + fn an_oversized_turn_is_split_in_order_without_losing_anything() { + let outputs: Vec = (0..12) + .map(|i| format!("{i:02}{}", "x".repeat(30_000))) + .collect(); + let mut request = chat_turn(&outputs); + // 一个单独就放不下的工具输出，拆成连续的几个步骤。 + request.messages.push(request.messages[1].clone()); + request.messages.push(request.messages[2].clone()); + let huge = "y\n".repeat(MAX_SOURCE_TOKENS * 3); + request.messages.last_mut().unwrap().content = vec![Value::from(huge.as_str())]; + let projection = turn_transcript(&request, FINAL, "messages").unwrap(); + let count = projection.parts.len(); + assert!(count > 2); + let mut entries = Vec::new(); + for (index, part) in projection.parts.iter().enumerate() { + assert!( + units(part) <= MAX_SOURCE_TOKENS * UNITS_PER_TOKEN, + "each part fits Core's default extraction budget" + ); + let part: Value = serde_json::from_str(part).unwrap(); + assert_eq!(part["kind"], "gateway_turn"); + assert_eq!(part["part"], index + 1); + assert_eq!(part["parts"], count); + entries.extend(part["steps"].as_array().unwrap().clone()); + } assert_eq!( - projection.coverage["reasons"], - serde_json::json!(["field_byte_limit"]) + entries.first().unwrap(), + &serde_json::json!({"type": "user", "text": "task"}) ); - assert!(!projection.text.contains("opaque")); - } - - #[test] - fn responses_history_with_multiple_items_keeps_its_body_but_cannot_claim_complete() { - let mut request = chat_turn(&[]); - let mut message = request.messages[0].clone(); - message.role = Role::Assistant; - message.content = vec![ - serde_json::json!({"type":"message", "role":"assistant", "content":"first"}), - serde_json::json!({"type":"message", "role":"assistant", "content":"second"}), - ]; - request.messages.push(message); - let projection = turn_transcript( - &request, - r#"{"output":[{"type":"message","role":"assistant","content":"done"}]}"#, - "input", - ) - .unwrap(); - assert_eq!(projection.coverage["status"], "unknown"); assert_eq!( - projection.coverage["reasons"], - serde_json::json!(["unmeasured_projection"]) + entries.last().unwrap(), + &serde_json::json!({"type": "assistant", "text": "done"}) ); + let outputs: Vec<&str> = entries + .iter() + .filter(|entry| entry["type"] == "tool_output") + .map(|entry| entry["output"].as_str().unwrap()) + .collect(); + for (i, output) in outputs.iter().take(12).enumerate() { + assert!(output.starts_with(&format!("{i:02}"))); + } assert_eq!( - projection.text, - r#"{"kind":"gateway_turn","user":"task","steps":[{"type":"assistant","text":"first"}],"assistant":"done"}"# + outputs[12..].concat(), + huge, + "the oversized output is kept whole, in order" ); + assert!(entries + .iter() + .filter(|entry| entry["type"] == "tool_output") + .skip(13) + .all(|entry| entry["continued"] == true)); } #[test] - fn a_long_tool_output_is_truncated() { - let request = chat_turn(&["x".repeat(MAX_FIELD_BYTES * 2)]); - let text = turn_transcript(&request, FINAL, "messages").unwrap().text; - let transcript: Value = serde_json::from_str(&text).unwrap(); - let output = transcript["steps"][1]["output"].as_str().unwrap(); - assert!(output.ends_with(&format!("[truncated {MAX_FIELD_BYTES} bytes]"))); - assert_eq!(transcript["assistant"], "done"); - } - - #[test] - fn an_oversized_turn_omits_its_oldest_steps() { - let outputs: Vec = (0..40) - .map(|i| format!("{i:02}{}", "x".repeat(MAX_FIELD_BYTES - 2))) - .collect(); - let request = chat_turn(&outputs); - let text = turn_transcript(&request, FINAL, "messages").unwrap().text; - assert!(text.len() <= MAX_TRANSCRIPT_BYTES); - let transcript: Value = serde_json::from_str(&text).unwrap(); - let omitted = transcript["omitted_steps"].as_u64().unwrap(); - assert!(omitted > 0); - let steps = transcript["steps"].as_array().unwrap(); - // 保留的是最近的步骤，最后一步是最后一次工具输出。 - assert!(steps.last().unwrap()["output"] - .as_str() - .unwrap() - .starts_with("39")); - assert_eq!(steps.len() as u64 + omitted, 80); - assert_eq!(transcript["user"], "task"); - assert_eq!(transcript["assistant"], "done"); + fn responses_items_keep_every_field_the_model_read() { + let mut request = chat_turn(&[]); + request.messages.extend([ + item(serde_json::json!({"type": "message", "role": "developer", + "content": [{"type": "input_text", "text": "Run tests before finishing."}]})), + item(serde_json::json!({"type": "reasoning", "encrypted_content": "gAAAA", + "summary": [{"type": "summary_text", "text": "Checking the loader."}]})), + item(serde_json::json!({"type": "apply_patch_call", "id": "ap_1", "call_id": "call_1", "status": "completed", + "operation": {"type": "update_file", "path": "a.py", "diff": "@@\n-old\n+new"}})), + item(serde_json::json!({"type": "apply_patch_call_output", "call_id": "call_1", "output": "patched"})), + item(serde_json::json!({"type": "computer_call_output", "call_id": "call_2", + "output": {"type": "computer_screenshot", "image_url": "data:image/png;base64,AAAA"}})), + item(serde_json::json!({"type": "function_call_output", "call_id": "call_3", + "output": [{"type": "input_image", "image_url": "https://example.com/a.png"}]})), + item(serde_json::json!({"type": "mcp_approval_response", "approval_request_id": "r1", + "approve": false, "reason": "too risky"})), + item(serde_json::json!({"type": "frobnicate", "id": "f_1", "data": "visible payload"})), + ]); + let response = serde_json::json!({"output": [ + {"type": "mcp_call", "id": "mcp_1", "name": "search_docs", "server_label": "docs", + "arguments": "{\"q\":\"loader\"}", "error": "server timed out"}, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "done"}]} + ]}) + .to_string(); + let projection = turn_transcript(&request, &response, "input").unwrap(); + let transcript = single(&projection); + assert_eq!( + transcript["steps"], + serde_json::json!([ + {"type": "message", "role": "developer", "text": "Run tests before finishing."}, + {"type": "reasoning_summary", "text": "Checking the loader."}, + {"type": "tool_call", "name": "apply_patch_call", + "arguments": "{\"diff\":\"@@\\n-old\\n+new\",\"path\":\"a.py\",\"type\":\"update_file\"}"}, + {"type": "tool_output", "output": "patched"}, + {"type": "tool_output", + "output": "{\"image_url\":\"[image/png data omitted: 26 bytes]\",\"type\":\"computer_screenshot\"}"}, + {"type": "tool_output", "output": "[input_image: https://example.com/a.png]"}, + {"type": "tool_output", "output": "{\"approval_request_id\":\"r1\",\"approve\":false,\"reason\":\"too risky\"}"}, + {"type": "item", "item_type": "frobnicate", "content": "{\"data\":\"visible payload\"}"}, + {"type": "tool_call", "name": "search_docs", + "arguments": "{\"arguments\":\"{\\\"q\\\":\\\"loader\\\"}\",\"server_label\":\"docs\"}"}, + {"type": "tool_output", "output": "server timed out"} + ]) + ); + assert_eq!(projection.coverage["status"], "complete"); + for absent in ["call_1", "ap_1", "mcp_1", "f_1", "base64", "gAAAA"] { + assert!(!projection.parts[0].contains(absent), "{absent}"); + } } #[test] - fn built_in_calls_and_outputs_in_the_history_are_kept_without_binary_data() { - let item = |value: Value| CanonicalMessage { - role: Role::Tool, - content: vec![value], - content_null: false, - tool_calls: None, - extra_fields: Default::default(), - tool_call_id: None, - }; + fn a_single_argument_field_is_recorded_as_its_text() { let mut request = chat_turn(&[]); request.messages.extend([ - item(serde_json::json!({"type": "computer_call", "id": "cu_1", "call_id": "call_1", - "status": "completed", "action": {"type": "click", "x": 1, "y": 2}})), - item(serde_json::json!({"type": "computer_call_output", "call_id": "call_1", - "output": {"type": "computer_screenshot", "image_url": "data:image/png;base64,AAAA"}})), - item(serde_json::json!({"type": "mcp_call", "id": "mcp_1", "name": "search_docs", - "server_label": "docs", "arguments": "{\"q\":\"loader\"}", "output": "3 results"})), - item(serde_json::json!({"type": "reasoning", "encrypted_content": "gAAAA"})), - item(serde_json::json!({"type": "frobnicate", "data": "opaque"})), + item(serde_json::json!({"type": "function_call", "call_id": "c", "name": "run", "arguments": "{\"cmd\":\"ls\"}"})), + item(serde_json::json!({"type": "custom_tool_call", "call_id": "d", "name": "exec", "input": "ls -la"})), + item(serde_json::json!({"type": "computer_call", "call_id": "e", "action": {"type": "screenshot"}})), ]); - let response = r#"{"output":[{"type":"message","role":"assistant","content":[{"type":"output_text","text":"done"}]}]}"#; - let text = turn_transcript(&request, response, "input").unwrap().text; - let transcript: Value = serde_json::from_str(&text).unwrap(); + let transcript = single(&turn_transcript(&request, RESPONSES_FINAL, "input").unwrap()); assert_eq!( transcript["steps"], serde_json::json!([ - {"type": "tool_call", "name": "computer_call", "arguments": "{\"type\":\"click\",\"x\":1,\"y\":2}"}, - {"type": "tool_output", "output": "{\"image_url\":\"[binary data omitted]\",\"type\":\"computer_screenshot\"}"}, - {"type": "tool_call", "name": "search_docs", "arguments": "{\"q\":\"loader\"}"}, - {"type": "tool_output", "output": "3 results"} + {"type": "tool_call", "name": "run", "arguments": "{\"cmd\":\"ls\"}"}, + {"type": "tool_call", "name": "exec", "arguments": "ls -la"}, + {"type": "tool_call", "name": "computer_call", "arguments": "{\"type\":\"screenshot\"}"} ]) ); - for absent in ["call_1", "cu_1", "mcp_1", "base64", "gAAAA", "opaque"] { - assert!(!text.contains(absent), "{absent} in {text}"); - } + } + + #[test] + fn chat_calls_of_other_shapes_are_recorded_whole() { + let mut request = chat_turn(&["done".into()]); + request.messages[1].tool_calls = Some(vec![ + serde_json::json!({"type": "custom", "id": "x1", "custom": {"name": "exec", "input": "list files"}}), + ]); + let transcript = single(&turn_transcript(&request, FINAL, "messages").unwrap()); + assert_eq!( + transcript["steps"][0], + serde_json::json!({"type": "tool_call", "name": "custom", + "arguments": "{\"custom\":{\"input\":\"list files\",\"name\":\"exec\"},\"type\":\"custom\"}"}) + ); } #[test] diff --git a/gateway/crates/pc-gateway-core/tests/executor_semantics.rs b/gateway/crates/pc-gateway-core/tests/executor_semantics.rs index 85e9d615d2..adf27655fd 100644 --- a/gateway/crates/pc-gateway-core/tests/executor_semantics.rs +++ b/gateway/crates/pc-gateway-core/tests/executor_semantics.rs @@ -274,6 +274,7 @@ fn call(degrade: DegradePolicy) -> ManagedCall { degrade_policy: degrade, inject_context: false, include_code: false, + capture_prefix: false, conversation_id: None, sediment: false, prepared_contexts: Vec::new(), @@ -1318,7 +1319,7 @@ async fn sediment_captures_released_turn_synchronously() { assert_eq!( metadata[0]["powercontext_capture_coverage"], serde_json::json!({ - "schema": "powercontext.capture-coverage.v1", "projection": "gateway-turn-text.v1", + "schema": "powercontext.capture-coverage.v1", "projection": "gateway-turn-text.v2", "unit": "gateway_turn_projection", "status": "complete", "reasons": [] }) ); @@ -2742,318 +2743,64 @@ async fn responses_tool_loop_captures_once_at_the_terminal_response() { } #[tokio::test] -async fn responses_call_coverage_requires_a_measurable_payload() { - for (item, name, arguments, measurable) in [ - ( - serde_json::json!({"type": "apply_patch_call", "call_id": "call_0", - "operation": {"type": "update_file", "path": "example.py", "diff": "@@\n-old\n+new"}}), - "apply_patch_call", - "", - false, - ), - ( - serde_json::json!({"type": "future_tool_call", "call_id": "call_0", "payload": {"text": "hidden"}}), - "future_tool_call", - "", - false, - ), - ( - serde_json::json!({"type": "future_tool_call", "call_id": "call_0", "arguments": "visible", "payload": "hidden"}), - "future_tool_call", - "visible", - false, - ), - ( - serde_json::json!({"type": "function_call", "call_id": "call_0", "name": "run"}), - "run", - "", - false, - ), - ( - serde_json::json!({"type": "function_call", "call_id": "call_0", "name": "run", "arguments": null}), - "run", - "", - false, - ), - ( - serde_json::json!({"type": "function_call", "call_id": "call_0", "name": "run", "arguments": ""}), - "run", - "", - true, - ), - ( - serde_json::json!({"type": "custom_tool_call", "call_id": "call_0", "name": "run", "input": "ls"}), - "run", - "ls", - true, - ), - ( - serde_json::json!({"type": "computer_call", "call_id": "call_0", "action": {"type": "screenshot"}}), - "computer_call", - "{\"type\":\"screenshot\"}", - true, - ), - ] { - for clipped in [false, true] { - let (executor, _renderer, transport, gate) = harness( - allow_runs(2), - vec![ - responses_output(serde_json::json!([item.clone()])), - responses_output(serde_json::json!([ - {"type": "message", "role": "assistant", "content": "done"} - ])), - ], - 2, - ); - let output = if clipped { - "x".repeat(8193) - } else { - "ok".into() - }; - let output_item = serde_json::json!({ - "type": format!("{}_output", item["type"].as_str().unwrap()), - "call_id": "call_0", "output": output - }); - for step in 0..2 { - let mut c = tool_loop_call("fix the bug", step, true); - c.inject_context = false; - c.sediment = true; - if step == 1 { - c.request.messages[1].content = vec![item.clone()]; - c.request.messages[2].content = vec![output_item.clone()]; - } - let outcome = executor - .run( - request_control(), - c, - &pc_gateway_protocol::OpenAiResponsesRenderer, - &NoopEvidenceSink, - ) - .await; - assert!(matches!(outcome, ManagedOutcome::Released { .. })); - assert_eq!(gate.captured.lock().unwrap().len(), step); - } - let sent: serde_json::Value = - serde_json::from_slice(&transport.sends.lock().unwrap()[1]).unwrap(); - assert_eq!(sent["input"][1], item); - assert_eq!(sent["input"][2], output_item); - let captured = gate.captured.lock().unwrap(); - let body: serde_json::Value = serde_json::from_str(&captured[0].1).unwrap(); - let projected_output = if clipped { - format!("{}\n[truncated 1 bytes]", "x".repeat(8192)) - } else { - output - }; - assert_eq!( - body, - serde_json::json!({ - "kind": "gateway_turn", "user": "fix the bug", - "steps": [ - {"type": "tool_call", "name": name, "arguments": arguments}, - {"type": "tool_output", "output": projected_output} - ], - "assistant": "done" - }) - ); - let mut reasons = Vec::new(); - if clipped { - reasons.push("field_byte_limit"); - } - if !measurable { - reasons.push("unmeasured_projection"); - } - assert_eq!( - gate.captured_metadata.lock().unwrap()[0]["powercontext_capture_coverage"], - serde_json::json!({ - "schema": "powercontext.capture-coverage.v1", "projection": "gateway-turn-text.v1", - "unit": "gateway_turn_projection", - "status": if clipped { "partial" } else if measurable { "complete" } else { "unknown" }, - "reasons": reasons - }), - "item: {item}" - ); - } - } -} - -#[tokio::test] -async fn responses_hosted_call_without_projected_payload_is_unmeasured() { +async fn a_captured_turn_keeps_every_call_field_and_output_uncut() { + let call_item = serde_json::json!({"type": "apply_patch_call", "call_id": "call_0", + "operation": {"type": "update_file", "path": "example.py", "diff": "@@\n-old\n+new"}}); + let long = "x".repeat(40_000); + let output_item = + serde_json::json!({"type": "apply_patch_call_output", "call_id": "call_0", "output": long}); let (executor, _renderer, _transport, gate) = harness( - allow_runs(1), - vec![responses_output(serde_json::json!([ - {"type": "code_interpreter_call", "output": "tool result"}, - {"type": "message", "role": "assistant", "content": "done"} - ]))], + allow_runs(2), + vec![ + responses_output(serde_json::json!([call_item.clone()])), + responses_output(serde_json::json!([ + {"type": "message", "role": "assistant", "content": "done"} + ])), + ], 2, ); - let mut c = tool_loop_call("fix the bug", 0, true); - c.inject_context = false; - c.sediment = true; - let outcome = executor - .run( - request_control(), - c, - &pc_gateway_protocol::OpenAiResponsesRenderer, - &NoopEvidenceSink, - ) - .await; - assert!(matches!(outcome, ManagedOutcome::Released { .. })); + for step in 0..2 { + let mut c = tool_loop_call("fix the bug", step, true); + c.inject_context = false; + c.sediment = true; + if step == 1 { + c.request.messages[1].content = vec![call_item.clone()]; + c.request.messages[2].content = vec![output_item.clone()]; + } + let outcome = executor + .run( + request_control(), + c, + &pc_gateway_protocol::OpenAiResponsesRenderer, + &NoopEvidenceSink, + ) + .await; + assert!(matches!(outcome, ManagedOutcome::Released { .. })); + } let captured = gate.captured.lock().unwrap(); + assert_eq!(captured.len(), 1); let body: serde_json::Value = serde_json::from_str(&captured[0].1).unwrap(); assert_eq!( body, serde_json::json!({ "kind": "gateway_turn", "user": "fix the bug", "steps": [ - {"type": "tool_call", "name": "code_interpreter_call", "arguments": ""}, - {"type": "tool_output", "output": "tool result"} + {"type": "tool_call", "name": "apply_patch_call", + "arguments": "{\"diff\":\"@@\\n-old\\n+new\",\"path\":\"example.py\",\"type\":\"update_file\"}"}, + {"type": "tool_output", "output": long} ], "assistant": "done" }) ); - let metadata = gate.captured_metadata.lock().unwrap(); - assert_eq!( - metadata[0]["powercontext_capture_coverage"]["status"], - "unknown" - ); assert_eq!( - metadata[0]["powercontext_capture_coverage"]["reasons"], - serde_json::json!(["unmeasured_projection"]) + gate.captured_metadata.lock().unwrap()[0]["powercontext_capture_coverage"], + serde_json::json!({ + "schema": "powercontext.capture-coverage.v1", "projection": "gateway-turn-text.v2", + "unit": "gateway_turn_projection", "status": "complete", "reasons": [] + }) ); } -#[tokio::test] -async fn responses_output_coverage_requires_a_measurable_payload() { - for (output, projected_output, measurable) in [ - ( - serde_json::json!({"type": "mcp_approval_response", "approval_request_id": "approval-1", - "approve": false, "reason": "Denied because production data"}), - "", - false, - ), - ( - serde_json::json!({"type": "mcp_approval_response", "approval_request_id": "approval-1", - "approve": true, "output": "visible", "reason": "Unprojected decision"}), - "visible", - false, - ), - ( - serde_json::json!({"type": "future_tool_call_output", "call_id": "call_0", "payload": "hidden"}), - "", - false, - ), - ( - serde_json::json!({"type": "future_tool_call_output", "call_id": "call_0", "output": "visible", "payload": "hidden"}), - "visible", - false, - ), - ( - serde_json::json!({"type": "function_call_output", "call_id": "call_0"}), - "", - false, - ), - ( - serde_json::json!({"type": "function_call_output", "call_id": "call_0", "output": null}), - "null", - false, - ), - ( - serde_json::json!({"type": "function_call_output", "call_id": "call_0", "output": ""}), - "", - true, - ), - ( - serde_json::json!({"type": "custom_tool_call_output", "call_id": "call_0", "output": "ok"}), - "ok", - true, - ), - ] { - for clipped in [false, true] { - let assistant = if clipped { - "x".repeat(8193) - } else { - "done".into() - }; - let (executor, _renderer, transport, gate) = harness( - allow_runs(2), - vec![ - responses_output(serde_json::json!([output.clone()])), - responses_output(serde_json::json!([ - {"type": "message", "role": "assistant", "content": assistant} - ])), - ], - 2, - ); - for step in 0..2 { - let mut c = tool_loop_call("fix the bug", step, true); - c.inject_context = false; - c.sediment = true; - if step == 1 { - if output["type"] == "mcp_approval_response" { - c.request.messages[1].content = vec![serde_json::json!({ - "type": "mcp_approval_request", "id": "approval-1", - "name": "run", "server_label": "docs", "arguments": "{}" - })]; - } else if output["type"] == "custom_tool_call_output" { - c.request.messages[1].content = vec![serde_json::json!({ - "type": "custom_tool_call", "call_id": "call_0", "name": "run", "input": "{}" - })]; - } - c.request.messages[2].content = vec![output.clone()]; - } - let outcome = executor - .run( - request_control(), - c, - &pc_gateway_protocol::OpenAiResponsesRenderer, - &NoopEvidenceSink, - ) - .await; - assert!(matches!(outcome, ManagedOutcome::Released { .. })); - assert_eq!(gate.captured.lock().unwrap().len(), step); - } - let sent: serde_json::Value = - serde_json::from_slice(&transport.sends.lock().unwrap()[1]).unwrap(); - assert_eq!(sent["input"][2], output); - let captured = gate.captured.lock().unwrap(); - let body: serde_json::Value = serde_json::from_str(&captured[0].1).unwrap(); - let projected_assistant = if clipped { - format!("{}\n[truncated 1 bytes]", "x".repeat(8192)) - } else { - assistant - }; - assert_eq!( - body, - serde_json::json!({ - "kind": "gateway_turn", "user": "fix the bug", - "steps": [ - {"type": "tool_call", "name": "run", "arguments": "{}"}, - {"type": "tool_output", "output": projected_output} - ], - "assistant": projected_assistant - }) - ); - let mut reasons = Vec::new(); - if clipped { - reasons.push("field_byte_limit"); - } - if !measurable { - reasons.push("unmeasured_projection"); - } - assert_eq!( - gate.captured_metadata.lock().unwrap()[0]["powercontext_capture_coverage"], - serde_json::json!({ - "schema": "powercontext.capture-coverage.v1", "projection": "gateway-turn-text.v1", - "unit": "gateway_turn_projection", - "status": if clipped { "partial" } else if measurable { "complete" } else { "unknown" }, - "reasons": reasons - }), - "output: {output}" - ); - } - } -} - #[tokio::test] async fn a_failed_capture_still_releases_and_is_recorded() { let (executor, renderer, _transport, gate) = @@ -4196,3 +3943,132 @@ async fn real_body_timeout_exports_timeout_attempt_and_unknown_usage_without_rel } } } + +// ---------- #2413：为复用缓存前缀的记忆抽取保留回合前缀 ---------- + +#[tokio::test] +async fn a_terminal_responses_turn_keeps_its_prefix_for_extraction() { + let output = serde_json::json!([ + reasoning_item(), + {"id": "msg_1", "type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "done", "annotations": []}]} + ]); + let (executor, _renderer, transport, gate) = harness( + allow_runs(2), + vec![ + responses_output(output.clone()), + responses_output(output.clone()), + ], + 2, + ); + for capture_prefix in [true, false] { + let mut c = tool_loop_call("fix the bug", 0, true); + c.inject_context = false; + c.sediment = true; + c.capture_prefix = capture_prefix; + let outcome = executor + .run( + request_control(), + c, + &pc_gateway_protocol::OpenAiResponsesRenderer, + &NoopEvidenceSink, + ) + .await; + assert!(matches!(outcome, ManagedOutcome::Released { .. })); + } + let metadata = gate.captured_metadata.lock().unwrap().clone(); + assert!( + metadata[1].get("prefix_handle").is_none(), + "without opt-in nothing is kept" + ); + let handle = metadata[0]["prefix_handle"].as_str().unwrap().to_owned(); + let prefix = executor.prefix(&handle, "project:test").unwrap(); + assert_eq!(prefix.conversation_id, "conversation-a"); + assert_eq!(prefix.target_profile, "openai:test"); + assert_eq!( + prefix.request_body, + transport.sends.lock().unwrap()[0], + "the prefix is the exact body sent upstream" + ); + assert_eq!(serde_json::Value::from(prefix.response_output), output); + assert!(executor.prefix(&handle, "project:other").is_none()); + // 前缀规模取自终止响应的用量，供 Core 判断复用前缀是否更便宜。 + assert_eq!(metadata[0]["prefix_input_tokens"], 10); + assert_eq!(metadata[0]["prefix_output_tokens"], 2); +} + +#[tokio::test] +async fn an_injected_turn_keeps_its_prefix_and_a_source_without_the_injection() { + let output = serde_json::json!([ + {"id": "msg_1", "type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "done", "annotations": []}]} + ]); + let (executor, _renderer, transport, gate) = + harness(allow_runs(1), vec![responses_output(output)], 2); + *gate.prepare_result.lock().unwrap() = Some(Ok(reuse_result("memory-snapshot-1"))); + let mut c = tool_loop_call("fix the bug", 0, true); + c.sediment = true; + c.capture_prefix = true; + c.inject_context = true; + let outcome = executor + .run( + request_control(), + c, + &pc_gateway_protocol::OpenAiResponsesRenderer, + &NoopEvidenceSink, + ) + .await; + assert!(matches!(outcome, ManagedOutcome::Released { .. })); + // 前缀是实际发出的请求，含注入块；Source 取自注入前的请求，不含注入块。 + let handle = gate.captured_metadata.lock().unwrap()[0]["prefix_handle"] + .as_str() + .unwrap() + .to_owned(); + let prefix = executor.prefix(&handle, "project:test").unwrap(); + assert_eq!(prefix.request_body, transport.sends.lock().unwrap()[0]); + assert!(String::from_utf8_lossy(&prefix.request_body).contains("memory-snapshot-1")); + assert!(!gate.captured.lock().unwrap()[0] + .1 + .contains("memory-snapshot-1")); +} + +#[tokio::test] +async fn an_oversized_turn_is_captured_as_ordered_parts_with_the_handle_on_the_last() { + let output = serde_json::json!([ + {"id": "msg_1", "type": "message", "role": "assistant", + "content": [{"type": "output_text", "text": "done", "annotations": []}]} + ]); + let (executor, _renderer, _transport, gate) = + harness(allow_runs(1), vec![responses_output(output)], 2); + let mut c = tool_loop_call(&"y".repeat(450_000), 0, true); + c.inject_context = false; + c.sediment = true; + c.capture_prefix = true; + let outcome = executor + .run( + request_control(), + c, + &pc_gateway_protocol::OpenAiResponsesRenderer, + &NoopEvidenceSink, + ) + .await; + assert!(matches!(outcome, ManagedOutcome::Released { .. })); + let captured = gate.captured.lock().unwrap().clone(); + let metadata = gate.captured_metadata.lock().unwrap().clone(); + let count = captured.len(); + assert!(count >= 3); + let turn = captured[0].0.rsplit_once('-').unwrap().0.to_owned(); + for (index, ((source_id, content), metadata)) in captured.iter().zip(&metadata).enumerate() { + assert_eq!(source_id, &format!("{turn}-{}of{count}", index + 1)); + assert!(content.chars().count() <= 200_000); + assert_eq!( + metadata["turn"], + serde_json::json!({"source_id": turn, "part": index + 1, "parts": count}) + ); + assert_eq!( + metadata.get("prefix_handle").is_some(), + index + 1 == count, + "only the last part carries the handle" + ); + } +} diff --git a/integrations/hermes/plugins/powercontext/helpers.py b/integrations/hermes/plugins/powercontext/helpers.py index 8f7e4444a1..d929cc1e0d 100644 --- a/integrations/hermes/plugins/powercontext/helpers.py +++ b/integrations/hermes/plugins/powercontext/helpers.py @@ -40,6 +40,7 @@ "hermes-turn-text.v1": "user_assistant_text_pair", "hermes-precompress-text.v1": "new_user_assistant_window", "gateway-turn-text.v1": "gateway_turn_projection", + "gateway-turn-text.v2": "gateway_turn_projection", } _CAPTURE_REASONS = frozenset({"character_limit", "field_byte_limit", "transcript_byte_limit", "unmeasured_projection"}) SCOPE_SAFE_RE = re.compile(r"[^\w:./@+-]+", re.UNICODE) diff --git a/src/powercontext/builtin/artifacts/atomic_memory/extraction.py b/src/powercontext/builtin/artifacts/atomic_memory/extraction.py index 61f91ba10c..b23a12f159 100644 --- a/src/powercontext/builtin/artifacts/atomic_memory/extraction.py +++ b/src/powercontext/builtin/artifacts/atomic_memory/extraction.py @@ -16,14 +16,20 @@ from __future__ import annotations +import logging from collections.abc import Callable from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal from pydantic import BaseModel, ConfigDict, Field, JsonValue, TypeAdapter, ValidationError +from powercontext._logging import log_safely from powercontext.artifacts import ArtifactRef from powercontext.builtin.artifacts.atomic_memory.models import AtomicMemoryContent +from powercontext.builtin.artifacts.atomic_memory.prefixed_extraction import ( + PrefixedExtractionUnavailable, + prefix_handle, +) from powercontext.builtin.artifacts.atomic_memory.summary import DESCRIPTION_RULES from powercontext.builtin.artifacts.atomic_memory.temporal import preserves_time_literals, source_time_literals from powercontext.builtin.artifacts.memory.prompts import MemoryExtractionProfile @@ -45,6 +51,8 @@ SourceSequenceLocator, ) +logger = logging.getLogger(__name__) + _JSON = TypeAdapter(JsonValue) _TEXT_EVIDENCE_KEY = TEXT_EVIDENCE_PROJECTION_KEY.model_dump(mode="json") @@ -158,22 +166,69 @@ class AtomicMemoryGenerationPipeline: estimator: TokenEstimator extraction_instructions: str = atomic_memory_extraction_instructions(MemoryExtractionProfile.CODING) input_renderer: Callable[[AtomicMemoryExtractionInput | AtomicMemoryReconciliationInput], str] | None = None + # Opt-in extraction over the agent's cached conversation prefix; any failure falls back to ``extractor``. + prefixed: StructuredGenerator[AtomicMemoryExtractionInput, AtomicMemoryExtractionOutput] | None = None async def extract(self, value: AtomicMemoryExtractionInput) -> tuple[AtomicMemoryCandidate, ...]: + if self.prefixed is not None and prefix_handle(value) is not None: + candidates = await self._extract_prefixed(self.prefixed, value) + if candidates is not None: + return candidates result = await self.extractor.generate(value) + return validated_candidates(value, result.output) + + async def _extract_prefixed( + self, + prefixed: StructuredGenerator[AtomicMemoryExtractionInput, AtomicMemoryExtractionOutput], + value: AtomicMemoryExtractionInput, + ) -> tuple[AtomicMemoryCandidate, ...] | None: try: - output = AtomicMemoryExtractionOutput.model_validate(result.output) - except ValidationError as error: - raise attach_diagnostics( - InvalidInferenceOutputError("atomic-memory-extract", "output does not match the extraction schema"), - ProcessingDiagnostics( - stage=ProcessingStage.OUTPUT_PARSING, - reason=ProcessingFailureReason.SCHEMA_INVALID, - failure_location="atomic-memory-extract", - ), - ) from error - validate_extraction_output(value, output) - return output.candidates + result = await prefixed.generate(value) + candidates = validated_candidates(value, result.output) + except PrefixedExtractionUnavailable as error: + reason = error.reason + except InvalidInferenceOutputError: + reason = "invalid_output" + else: + _log_prefixed_extraction("success") + return candidates + _log_prefixed_extraction("fallback", reason) + return None + + +def _log_prefixed_extraction(outcome: str, reason: str | None = None) -> None: + log_safely( + logger, + logging.INFO, + "Prefixed Atomic Memory extraction completed" + if reason is None + else "Prefixed Atomic Memory extraction fell back", + extra={ + "event": "atomic_memory.prefixed_extraction", + "operation": "atomic_memory.extract", + "unit": "inference", + "outcome": outcome, + **({} if reason is None else {"fallback_reason": reason}), + }, + ) + + +def validated_candidates(value: AtomicMemoryExtractionInput, raw: object) -> tuple[AtomicMemoryCandidate, ...]: + """Candidates that satisfy the extraction schema, the content rules and cite only supplied evidence.""" + + try: + output = AtomicMemoryExtractionOutput.model_validate(raw) + except ValidationError as error: + raise attach_diagnostics( + InvalidInferenceOutputError("atomic-memory-extract", "output does not match the extraction schema"), + ProcessingDiagnostics( + stage=ProcessingStage.OUTPUT_PARSING, + reason=ProcessingFailureReason.SCHEMA_INVALID, + failure_location="atomic-memory-extract", + ), + ) from error + validate_extraction_output(value, output) + return output.candidates def validate_extraction_output(value: AtomicMemoryExtractionInput, output: AtomicMemoryExtractionOutput) -> None: diff --git a/src/powercontext/builtin/artifacts/atomic_memory/prefixed_extraction.py b/src/powercontext/builtin/artifacts/atomic_memory/prefixed_extraction.py new file mode 100644 index 0000000000..329d3d7ab2 --- /dev/null +++ b/src/powercontext/builtin/artifacts/atomic_memory/prefixed_extraction.py @@ -0,0 +1,248 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Atomic Memory extraction over the agent's cached conversation prefix. + +A gateway deployed with Core keeps the request and response of each terminal agent turn for a short time and records +an opaque handle and the prefix's token counts in the turn's Source metadata. A turn too long for one Source is +captured as consecutive parts, and the handle is on the last. For a window holding exactly one turn, Core asks the +gateway to append its extraction instruction to that prefix, so the provider reads the conversation from its prompt +cache instead of Core sending the transcript again. Cached input is still billed at a reduced rate, so the path is used +only when the cached prefix plus Core's instruction is estimated to cost less than normal extraction of the same +Sources. The gateway runs Core's normal context checks on that request. Anything this path cannot complete raises +:class:`PrefixedExtractionUnavailable`, and the caller extracts the normal way. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, TypeGuard + +import httpx +from pydantic import ValidationError + +from powercontext.builtin.artifacts.prompt.service import current_prompt +from powercontext.builtin.inference import InferenceError, InvalidInferenceOutputError, TokenEstimator +from powercontext.builtin.inference.models import GenerationResult, InferenceUsage + +if TYPE_CHECKING: + from powercontext.builtin.artifacts.atomic_memory.extraction import ( + AtomicMemoryEvidence, + AtomicMemoryExtractionInput, + AtomicMemoryExtractionOutput, + ) + +PREFIX_HANDLE_KEY = "prefix_handle" +# Token counts of the terminal response that ends the prefix: its input is read from the cache, its output is new +# input to the extraction request. +PREFIX_INPUT_TOKENS_KEY = "prefix_input_tokens" +PREFIX_OUTPUT_TOKENS_KEY = "prefix_output_tokens" +_GATEWAY_METADATA_KEYS = {PREFIX_HANDLE_KEY, PREFIX_INPUT_TOKENS_KEY, PREFIX_OUTPUT_TOKENS_KEY} +# Part of a turn captured as several Sources: ``{"source_id": , "part": i, "parts": n}``. +TURN_KEY = "turn" +EXTRACT_PATH = "/gateway/v1/extract" + +# Gateway refusals that name their cause; other refusals are reported by kind. +_GATEWAY_REASONS = { + "prefix_extraction_disabled": "disabled", + "prefix_unavailable": "prefix_unavailable", + "target_mismatch": "target_mismatch", + "invalid_instruction": "invalid_instruction", +} + + +class PrefixedExtractionUnavailable(InferenceError): + """The prefixed path cannot extract this window; ``reason`` is a bounded fallback code.""" + + def __init__(self, reason: str) -> None: + self.reason = reason + super().__init__(f"prefixed extraction unavailable: {reason}") + + +def _metadata(evidence: AtomicMemoryEvidence) -> dict[str, Any]: + source = evidence.source_metadata + metadata = source.get("metadata") if isinstance(source, dict) else None + return metadata if isinstance(metadata, dict) else {} + + +def _one_turn(evidence: tuple[AtomicMemoryEvidence, ...]) -> bool: + """Whether the evidence is one whole gateway turn: a single Source, or all parts of one turn in order.""" + + if len(evidence) == 1: + return TURN_KEY not in _metadata(evidence[0]) + turns = [turn for item in evidence if isinstance(turn := _metadata(item).get(TURN_KEY), dict)] + if len(turns) != len(evidence): + return False + first = turns[0] + return all( + turn.get("source_id") == first.get("source_id") + and turn.get("parts") == len(evidence) + and turn.get("part") == index + for index, turn in enumerate(turns, start=1) + ) + + +def prefix_handle(value: AtomicMemoryExtractionInput) -> str | None: + """The gateway prefix handle of a window that holds exactly one gateway turn, which its last Source carries.""" + + if not value.evidence or not _one_turn(value.evidence): + return None + handle = _metadata(value.evidence[-1]).get(PREFIX_HANDLE_KEY) + return handle if isinstance(handle, str) and handle else None + + +def _prefix_tokens(evidence: AtomicMemoryEvidence) -> tuple[int, int] | None: + metadata = _metadata(evidence) + cached, uncached = metadata.get(PREFIX_INPUT_TOKENS_KEY), metadata.get(PREFIX_OUTPUT_TOKENS_KEY) + if _count(cached) and _count(uncached): + return cached, uncached + return None + + +def _count(value: object) -> TypeGuard[int]: + return isinstance(value, int) and not isinstance(value, bool) and value >= 0 + + +@dataclass(frozen=True) +class GatewayPrefixedExtractor: + """Generate extraction output through the gateway's prefixed extraction route.""" + + client: httpx.AsyncClient + instructions: str + estimator: TokenEstimator + # Price of a cached input token relative to an uncached one. + cached_input_ratio: float = 0.1 + max_output_tokens: int | None = None + + async def generate(self, value: AtomicMemoryExtractionInput, /) -> GenerationResult[AtomicMemoryExtractionOutput]: + from powercontext.builtin.artifacts.atomic_memory.extraction import AtomicMemoryExtractionOutput + + handle = prefix_handle(value) + if handle is None: + raise PrefixedExtractionUnavailable("no_prefix") + selection = current_prompt("atomic_memory.extract") + instructions = self.instructions if selection is None else selection.compiled_instructions + instruction = prefixed_instruction(instructions, value.evidence, AtomicMemoryExtractionOutput) + tokens = _prefix_tokens(value.evidence[-1]) + if tokens is None: + raise PrefixedExtractionUnavailable("prefix_size_unknown") + cached, uncached = tokens + prefixed_cost = self.estimator.estimate(instruction) + uncached + self.cached_input_ratio * cached + # Normal extraction sends the instructions and the Source evidence uncached, estimated as its budget is. + if prefixed_cost >= self.estimator.estimate(f"{instructions}\n{value.model_dump_json()}"): + raise PrefixedExtractionUnavailable("not_cheaper") + body: dict[str, Any] = {"handle": handle, "instruction": instruction} + if self.max_output_tokens is not None: + body["max_output_tokens"] = self.max_output_tokens + try: + response = await self.client.post(EXTRACT_PATH, json=body) + except httpx.HTTPError as error: + raise PrefixedExtractionUnavailable("gateway_unavailable") from error + payload = _json_object(response) + if response.status_code != httpx.codes.OK: + raise PrefixedExtractionUnavailable(_refusal_reason(payload)) + usage = _usage(payload.get("usage"), payload.get("model")) + try: + output = AtomicMemoryExtractionOutput.model_validate_json(_unfenced(payload["text"])) + except (KeyError, TypeError, ValidationError) as error: + failure = InvalidInferenceOutputError( + "atomic-memory-extract", "prefixed output does not match the extraction schema" + ) + failure.usage = usage + raise failure from error + return GenerationResult(output=output, usage=usage) + + +def prefixed_instruction( + instructions: str, evidence: tuple[AtomicMemoryEvidence, ...], output_type: type[AtomicMemoryExtractionOutput] +) -> str: + """The user message appended to the prefix. + + The output format is stated in the message because a response-format field would change the cached request. + The Source records keep the provenance and timing the normal extraction input carries, without the transcript. + """ + + records = [] + for item in evidence: + metadata = item.source_metadata + if isinstance(metadata, dict) and isinstance(metadata.get("metadata"), dict): + inner = {key: value for key, value in metadata["metadata"].items() if key not in _GATEWAY_METADATA_KEYS} + metadata = {**metadata, "metadata": inner} + records.append({ + "evidence_id": item.evidence_id, + "journal_position": item.journal_position, + "source_metadata": metadata, + }) + return "\n\n".join(( + instructions, + "The conversation above is the Source evidence for this extraction. Only its most recent user turn is " + "evidence: the last user message before this one and everything after it. Earlier turns are context only; " + "do not extract a fact that only they support. Content between a BEGIN_POWERCONTEXT_ marker and its " + "END_POWERCONTEXT_ marker was supplied by PowerContext from existing memory or context: it is not " + "evidence for this turn, and a fact that only it supports must not be extracted. This message is not " + "evidence. Cite the evidence_id of the Source records below; a turn captured in several parts is listed in " + "order, and a fact may cite every part that supports it.", + "Source records:\n" + json.dumps(records, ensure_ascii=False, sort_keys=True), + "Reply with only a JSON object that matches this JSON Schema, without code fences or other text:\n" + + json.dumps(output_type.model_json_schema(), sort_keys=True), + )) + + +def _json_object(response: httpx.Response) -> dict[str, Any]: + try: + payload = response.json() + except ValueError: + return {} + return payload if isinstance(payload, dict) else {} + + +def _refusal_reason(payload: dict[str, Any]) -> str: + error = payload.get("error") + if isinstance(error, str) and error in _GATEWAY_REASONS: + return _GATEWAY_REASONS[error] + # The gateway passes a released provider error through unchanged, such as an OpenAI rate-limit object. + if isinstance(error, dict): + return "provider_error" + if payload.get("decision") in {"block", "withhold"}: + return "check_rejected" + return "gateway_error" + + +def _unfenced(text: object) -> str: + if not isinstance(text, str): + raise TypeError("text") + stripped = text.strip() + if stripped.startswith("```") and stripped.endswith("```"): + stripped = stripped.removeprefix("```json").removeprefix("```").removesuffix("```") + return stripped + + +def _usage(value: object, model: object) -> InferenceUsage: + """Portable usage from a Responses ``usage`` object.""" + + usage = value if isinstance(value, dict) else {} + + def count(field: object, name: str) -> int | None: + item = field.get(name) if isinstance(field, dict) else None + return item if isinstance(item, int) and not isinstance(item, bool) else None + + return InferenceUsage( + requests=1, + input_tokens=count(usage, "input_tokens"), + output_tokens=count(usage, "output_tokens"), + cache_read_tokens=count(usage.get("input_tokens_details"), "cached_tokens"), + reasoning_tokens=count(usage.get("output_tokens_details"), "reasoning_tokens"), + target_model=model if isinstance(model, str) and model else None, + ) diff --git a/src/powercontext/builtin/runtime/composition.py b/src/powercontext/builtin/runtime/composition.py index 9a44026378..1e20a99f93 100644 --- a/src/powercontext/builtin/runtime/composition.py +++ b/src/powercontext/builtin/runtime/composition.py @@ -1547,6 +1547,32 @@ async def _dream_generator( ) +def _prefixed_extractor( + settings: InferenceConfig, runtime: RuntimeConfig, resources: AsyncExitStack, estimator: TokenEstimator +) -> UsageReportingStructuredGenerator[AtomicMemoryExtractionInput, AtomicMemoryExtractionOutput] | None: + if settings.prefix_extraction_gateway_url is None: + return None + import httpx + + from powercontext.builtin.artifacts.atomic_memory.prefixed_extraction import GatewayPrefixedExtractor + + client = httpx.AsyncClient( + base_url=str(settings.prefix_extraction_gateway_url), + headers={name: value.get_secret_value() for name, value in settings.prefix_extraction_gateway_headers.items()}, + timeout=settings.generation_timeout_seconds, + follow_redirects=False, + ) + resources.push_async_callback(client.aclose) + return UsageReportingStructuredGenerator( + GatewayPrefixedExtractor( + client=client, + instructions=atomic_memory_extraction_instructions(runtime.memory_extraction_profile), + estimator=estimator, + cached_input_ratio=settings.prefix_extraction_cached_input_ratio, + ) + ) + + async def _generation_pipelines( settings: InferenceConfig, runtime: RuntimeConfig, @@ -1787,6 +1813,7 @@ async def _generation_pipelines( estimator=memory_estimator, extraction_instructions=atomic_memory_extraction_instructions(runtime.memory_extraction_profile), input_renderer=memory_input_renderer, + prefixed=_prefixed_extractor(settings, runtime, resources, memory_estimator), ) generated_incubation = LLMExperienceCandidatePipeline(UsageReportingStructuredGenerator(experience_generator)) generated_experience = LLMExperienceGenerator(UsageReportingStructuredGenerator(explicit_experience_generator)) diff --git a/src/powercontext/builtin/runtime/config.py b/src/powercontext/builtin/runtime/config.py index 5e2dbead92..3c8a9fc173 100644 --- a/src/powercontext/builtin/runtime/config.py +++ b/src/powercontext/builtin/runtime/config.py @@ -327,6 +327,11 @@ class InferenceConfig(BaseModel): generation_streaming: Literal["auto", "disabled"] = "disabled" generation_stream_idle_timeout_seconds: float = Field(default=60.0, gt=0) generation_model_context_window_tokens: int = Field(default=125_000, ge=1) + # Opt-in: extract Atomic Memory over the agent's cached conversation prefix through a gateway on the same host. + prefix_extraction_gateway_url: AnyHttpUrl | None = None + prefix_extraction_gateway_headers: dict[str, SecretStr] = Field(default_factory=dict, repr=False) + # Price of a cached input token relative to an uncached one for the agent's target. + prefix_extraction_cached_input_ratio: float = Field(default=0.1, gt=0, le=1) embedding_model: str | None = None embedding_base_url: AnyHttpUrl | None = None embedding_headers: dict[str, SecretStr] = Field(default_factory=dict, repr=False) @@ -378,7 +383,13 @@ def validate_normalization(cls, value: object) -> object: raise ValueError("embedding normalization must be 'none' or 'unit'") # noqa: TRY003 return normalized - @field_validator("generation_headers", "embedding_headers", "rerank_headers", "decision_headers") + @field_validator( + "generation_headers", + "embedding_headers", + "rerank_headers", + "decision_headers", + "prefix_extraction_gateway_headers", + ) @classmethod def validate_headers(cls, value: dict[str, SecretStr]) -> dict[str, SecretStr]: normalized_names: set[str] = set() @@ -416,6 +427,17 @@ def validate_embedding_profile(self) -> Self: ) return self + @model_validator(mode="after") + def validate_prefix_extraction(self) -> Self: + if self.generation_model is None and ( + self.prefix_extraction_gateway_url is not None or self.prefix_extraction_gateway_headers + ): + # Normal extraction is the fallback, so it must be configured. + raise ValueError("prefix extraction requires generation_model") # noqa: TRY003 + if self.prefix_extraction_gateway_headers and self.prefix_extraction_gateway_url is None: + raise ValueError("prefix_extraction_gateway_headers requires prefix_extraction_gateway_url") # noqa: TRY003 + return self + @model_validator(mode="after") def validate_workload_overrides(self) -> Self: if self.generation_model is None and self.generation_model_settings: diff --git a/tests/builtin/artifacts/atomic_memory/test_prefixed_extraction.py b/tests/builtin/artifacts/atomic_memory/test_prefixed_extraction.py new file mode 100644 index 0000000000..72fe92345a --- /dev/null +++ b/tests/builtin/artifacts/atomic_memory/test_prefixed_extraction.py @@ -0,0 +1,283 @@ +# Copyright (c) 2026 OceanBase. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Atomic Memory extraction over a gateway turn's cached prefix, with fallback to normal extraction.""" + +from __future__ import annotations + +import json +import logging +from typing import cast + +import httpx +import pytest +from fastapi.testclient import TestClient + +from powercontext.builtin.artifacts.atomic_memory.extraction import ( + AtomicMemoryExtractionInput, + AtomicMemoryExtractionOutput, + AtomicMemoryGenerationPipeline, +) +from powercontext.builtin.artifacts.atomic_memory.prefixed_extraction import GatewayPrefixedExtractor +from powercontext.builtin.artifacts.atomic_memory.reconciliation import ( + AtomicMemoryReconciliationContent, + AtomicMemoryReconciliationInput, + AtomicMemoryReconciliationOutput, +) +from powercontext.builtin.inference import GenerationResult, character_token_estimator +from powercontext.builtin.persistence.sqlite import SQLiteConfig +from powercontext.builtin.runtime import RuntimeConfig +from powercontext.server.factory import create_server_app +from powercontext.server.settings import McpConfig, ServerSettings + +# A turn whose tool output makes resending the transcript cost more than reading the cached prefix. +TRANSCRIPT = json.dumps({ + "kind": "gateway_turn", + "user": "Our team deploys only on Tuesdays. Check the release script.", + "steps": [ + {"type": "tool_call", "name": "exec", "arguments": '{"cmd": "cat release.sh"}'}, + {"type": "tool_output", "output": "echo release\n" * 2000}, + ], + "assistant": "Noted.", +}) +CANDIDATE = { + "kind": "constraint", + "text": "The team deploys only on Tuesdays.", + "description": "Deployment day", + "evidence_ids": ["source:1"], +} + + +class _Extractor: + def __init__(self) -> None: + self.calls = 0 + + async def generate(self, request: AtomicMemoryExtractionInput, /) -> GenerationResult[AtomicMemoryExtractionOutput]: + self.calls += 1 + return GenerationResult(output=AtomicMemoryExtractionOutput(candidates=())) + + +class _CreatingReconciler: + async def generate( + self, request: AtomicMemoryReconciliationInput, / + ) -> GenerationResult[AtomicMemoryReconciliationOutput]: + return GenerationResult( + output=AtomicMemoryReconciliationOutput( + action="create", + compared_ids=tuple(item.item_id for item in request.related), + content=AtomicMemoryReconciliationContent( + kind=request.proposal.kind, text=request.proposal.text, description="Deployment day" + ), + evidence_ids=request.proposal.evidence_ids, + reason="Keep the extracted candidate.", + ) + ) + + +class _Gateway: + """The gateway's extraction route: records requests and returns a scripted reply.""" + + def __init__(self, status: int, payload: object) -> None: + self.status = status + self.payload = payload + self.requests: list[dict[str, object]] = [] + # Atomic Memory text in the Scope after the flush. + self.memories: list[str] = [] + + def __call__(self, request: httpx.Request) -> httpx.Response: + assert request.url.path == "/gateway/v1/extract" + assert request.headers["authorization"] == "Bearer gateway-key" + self.requests.append(json.loads(request.content)) + return httpx.Response(self.status, json=self.payload) + + +def _flush(tmp_path, gateway: _Gateway, sources: list[dict[str, object]]) -> _Extractor: + extractor = _Extractor() + client = httpx.AsyncClient( + base_url="http://gateway.local", + headers={"Authorization": "Bearer gateway-key"}, + transport=httpx.MockTransport(gateway), + ) + app = create_server_app( + settings=ServerSettings( + database=SQLiteConfig(url=f"sqlite+aiosqlite:///{tmp_path / 'prefixed.db'}"), + runtime=RuntimeConfig(artifact_processing_families=()), + mcp=McpConfig(enabled=False), + ), + candidate_pipeline=AtomicMemoryGenerationPipeline( + extractor=extractor, + reconciler=_CreatingReconciler(), + estimator=character_token_estimator(), + prefixed=GatewayPrefixedExtractor( + client=client, instructions="Extract memories.", estimator=character_token_estimator() + ), + ), + ) + with TestClient(app) as test_client: + scope_id = test_client.get("/v1/scopes/default").json()["scope_id"] + for source in sources: + captured = test_client.post("/v1/sources/content", json={"scope_id": scope_id, **source}) + assert captured.status_code == 202, captured.text + flushed = test_client.post("/v1/memory/flush", json={"scope_id": scope_id}) + assert flushed.status_code == 200, flushed.text + listed = test_client.post("/v1/memory/entries/list", json={"scope_id": scope_id}) + assert listed.status_code == 200, listed.text + gateway.memories = [entry["text"] for entry in listed.json()["entries"]] + return extractor + + +def _turn( + source_id: str, handle: str | None = "a" * 32, prefix_tokens: tuple[int, int] | None = (30_000, 50) +) -> dict[str, object]: + metadata: dict[str, object] = {"kind": "gateway_turn", "target_profile": "codex"} + if handle is not None: + metadata["prefix_handle"] = handle + if prefix_tokens is not None: + metadata["prefix_input_tokens"], metadata["prefix_output_tokens"] = prefix_tokens + return {"source_id": source_id, "content": TRANSCRIPT, "metadata": metadata} + + +def test_a_gateway_turn_is_extracted_over_its_prefix_without_resending_the_transcript(tmp_path, caplog) -> None: + reply = {"candidates": [CANDIDATE]} + gateway = _Gateway(200, {"text": json.dumps(reply), "usage": {"input_tokens": 9000, "output_tokens": 40}}) + with caplog.at_level(logging.INFO): + extractor = _flush(tmp_path, gateway, [_turn("turn-1")]) + + assert extractor.calls == 0 + assert gateway.memories == [CANDIDATE["text"]] + (request,) = gateway.requests + assert request["handle"] == "a" * 32 + instruction = request["instruction"] + assert isinstance(instruction, str) + assert instruction.startswith("Extract memories.") + assert '"evidence_id": "source:1"' in instruction + # The conversation comes from the cached prefix: neither the transcript nor gateway metadata is sent again. + assert "deploys only on Tuesdays" not in instruction + assert "a" * 32 not in instruction + assert "prefix_input_tokens" not in instruction + # Memory injected into the conversation is marked; the instruction excludes it as evidence. + assert "BEGIN_POWERCONTEXT_" in instruction + assert any( + getattr(record, "event", None) == "atomic_memory.prefixed_extraction" + and getattr(record, "outcome", None) == "success" + for record in caplog.records + ) + + +@pytest.mark.parametrize( + ("status", "payload", "reason"), + [ + (404, {"error": "prefix_unavailable"}, "prefix_unavailable"), + (409, {"error": "target_mismatch"}, "target_mismatch"), + (409, {"decision": "withhold", "reason_codes": ["encrypted-origin-unverified"]}, "check_rejected"), + (502, {"decision": "not_sent", "attempts": 2}, "gateway_error"), + ( + 429, + {"error": {"message": "Rate limit reached", "type": "rate_limit_error", "code": "rate_limit_exceeded"}}, + "provider_error", + ), + (500, ["not", "an", "object"], "gateway_error"), + (200, {"text": "These are the memories: none."}, "invalid_output"), + (200, {"text": json.dumps({"candidates": [{**CANDIDATE, "evidence_ids": ["source:9"]}]})}, "invalid_output"), + ], +) +def test_a_failed_prefixed_extraction_falls_back_with_its_reason(tmp_path, caplog, status, payload, reason) -> None: + gateway = _Gateway(status, payload) + with caplog.at_level(logging.INFO): + extractor = _flush(tmp_path, gateway, [_turn("turn-1")]) + + assert len(gateway.requests) == 1 + assert extractor.calls == 1 + assert gateway.memories == [] + assert [ + getattr(record, "fallback_reason", None) + for record in caplog.records + if getattr(record, "event", None) == "atomic_memory.prefixed_extraction" + ] == [reason] + + +@pytest.mark.parametrize( + ("prefix_tokens", "reason"), + [((10_000_000, 50), "not_cheaper"), ((30_000, 40_000), "not_cheaper"), (None, "prefix_size_unknown")], + ids=["large-cached-prefix", "large-uncached-output", "unknown-size"], +) +def test_a_prefix_that_would_cost_more_is_not_used(tmp_path, caplog, prefix_tokens, reason) -> None: + gateway = _Gateway(500, {}) + with caplog.at_level(logging.INFO): + extractor = _flush(tmp_path, gateway, [_turn("turn-1", prefix_tokens=prefix_tokens)]) + + assert gateway.requests == [] + assert extractor.calls == 1 + assert [ + getattr(record, "fallback_reason", None) + for record in caplog.records + if getattr(record, "event", None) == "atomic_memory.prefixed_extraction" + ] == [reason] + + +@pytest.mark.parametrize( + "sources", + [ + [_turn("turn-1", handle=None)], + [_turn("turn-1"), _turn("turn-2", handle="b" * 32)], + ], + ids=["no-handle", "several-sources"], +) +def test_windows_without_a_single_prefixed_turn_use_normal_extraction(tmp_path, sources) -> None: + gateway = _Gateway(500, {}) + extractor = _flush(tmp_path, gateway, sources) + + assert gateway.requests == [] + assert extractor.calls == 1 + + +def _part(index: int, count: int, *, handle: bool) -> dict[str, object]: + source = _turn(f"turn-1-{index}of{count}", handle="a" * 32 if handle else None) + # Each part is a slice of one long turn; together they fit the extraction budget. + source["content"] = json.dumps({ + "kind": "gateway_turn", + "part": index, + "parts": count, + "steps": [{"type": "tool_output", "output": "echo release\n" * 700}], + }) + metadata = cast("dict[str, object]", source["metadata"]) + metadata["turn"] = {"source_id": "turn-1", "part": index, "parts": count} + if not handle: + del metadata["prefix_input_tokens"], metadata["prefix_output_tokens"] + return source + + +def test_a_turn_captured_in_parts_is_extracted_over_the_prefix_of_its_last_part(tmp_path) -> None: + candidate = {**CANDIDATE, "evidence_ids": ["source:1", "source:3"]} + gateway = _Gateway(200, {"text": json.dumps({"candidates": [candidate]}), "usage": {}}) + extractor = _flush( + tmp_path, gateway, [_part(1, 3, handle=False), _part(2, 3, handle=False), _part(3, 3, handle=True)] + ) + + assert extractor.calls == 0 + assert gateway.memories == [CANDIDATE["text"]] + (request,) = gateway.requests + assert request["handle"] == "a" * 32 + instruction = request["instruction"] + assert isinstance(instruction, str) + for evidence_id in ("source:1", "source:2", "source:3"): + assert f'"evidence_id": "{evidence_id}"' in instruction + + +def test_a_window_with_only_some_parts_of_a_turn_uses_normal_extraction(tmp_path) -> None: + gateway = _Gateway(500, {}) + extractor = _flush(tmp_path, gateway, [_part(2, 3, handle=False), _part(3, 3, handle=True)]) + + assert gateway.requests == [] + assert extractor.calls == 1 diff --git a/tests/integration_conformance/capture_coverage_cases.json b/tests/integration_conformance/capture_coverage_cases.json index 3f46bf7acc..b8e1290581 100644 --- a/tests/integration_conformance/capture_coverage_cases.json +++ b/tests/integration_conformance/capture_coverage_cases.json @@ -153,6 +153,18 @@ "unmeasured_projection" ] } + }, + { + "id": "gateway-turn-text.v2-0", + "projection": "gateway-turn-text.v2", + "reasons": [], + "expected": { + "schema": "powercontext.capture-coverage.v1", + "projection": "gateway-turn-text.v2", + "unit": "gateway_turn_projection", + "status": "complete", + "reasons": [] + } } ], "interpretations": [ @@ -332,6 +344,20 @@ }, "expected": "partial" }, + { + "id": "valid-gateway-turn-text.v2-0", + "metadata": { + "powercontext_capture_coverage": { + "schema": "powercontext.capture-coverage.v1", + "projection": "gateway-turn-text.v2", + "unit": "gateway_turn_projection", + "status": "complete", + "reasons": [] + }, + "prefix_handle": "synthetic-opaque" + }, + "expected": "complete" + }, { "id": "missing", "metadata": {}, diff --git a/tests/test_server_generation.py b/tests/test_server_generation.py index 4df2decd57..d76bf7d60b 100644 --- a/tests/test_server_generation.py +++ b/tests/test_server_generation.py @@ -49,3 +49,26 @@ def test_decision_overrides_require_decision_model() -> None: def test_decision_base_url_requires_decision_model() -> None: with pytest.raises(ValidationError, match="decision_base_url requires decision_model"): InferenceConfig.model_validate({"decision_base_url": "https://decision.example/v1"}) + + +def test_prefix_extraction_requires_normal_extraction_as_its_fallback() -> None: + with pytest.raises(ValidationError, match="prefix extraction requires generation_model"): + InferenceConfig.model_validate({"prefix_extraction_gateway_url": "http://127.0.0.1:8787"}) + with pytest.raises(ValidationError, match="requires prefix_extraction_gateway_url"): + InferenceConfig.model_validate({ + "generation_model": "provider:model", + "prefix_extraction_gateway_headers": {"Authorization": "Bearer gateway-key"}, + }) + configured = InferenceConfig.model_validate({ + "generation_model": "provider:model", + "prefix_extraction_gateway_url": "http://127.0.0.1:8787", + "prefix_extraction_gateway_headers": {"Authorization": "Bearer gateway-key"}, + }) + assert "gateway-key" not in repr(configured) + + +def test_prefix_extraction_cached_input_ratio_is_a_price_fraction() -> None: + for ratio in (0, 1.5): + with pytest.raises(ValidationError, match="prefix_extraction_cached_input_ratio"): + InferenceConfig.model_validate({"prefix_extraction_cached_input_ratio": ratio}) + assert InferenceConfig().prefix_extraction_cached_input_ratio == 0.1