From a6d8416c7633b3e4eedcee3e127b7976b3f139bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E6=9C=8D=E5=8A=A1=E9=83=A8-=E5=8F=B6=E6=9E=97?= =?UTF-8?q?=E7=AB=8B?= <1361666059@qq.com> Date: Wed, 21 Oct 2026 19:51:36 -0700 Subject: [PATCH] =?UTF-8?q?refactor(pi-compaction):=20=E6=94=B9=E4=B8=BA?= =?UTF-8?q?=E5=8E=8B=E5=8A=9B=E8=A7=A6=E5=8F=91=E7=9A=84=E8=A7=82=E5=AF=9F?= =?UTF-8?q?=E9=81=AE=E8=94=BD=E4=B8=8E=E7=A9=BA=E9=97=B2=E9=98=88=E5=80=BC?= =?UTF-8?q?=E6=95=B4=E6=AE=B5=E6=91=98=E8=A6=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 原实现对所有工具结果（含 read）在下一次请求即替换为约 4K 的首尾存根， 不看上下文压力，导致模型看不到大文件中段；后台逐条语义摘要又反复改写 历史、打碎 prompt 缓存。参照 Anthropic 上下文编辑、Manus 与 JetBrains 观察遮蔽研究重构为： - L2：上下文达到阈值时在 turn_end 批量提交 Pi 原生 context_edit，把最近 keepRecentTurns 轮之外的旧 toolResult 换成确定性占位符；trigger/target 回差与 minClearTokens 保证少触发、一次清够；原始条目保留，可经 context_tree_query 取回 - L3：空闲 agent_settled 且超过绝对阈值时调用一次 ctx.compact - 阈值按绝对 token 计并受窗口比例约束，默认针对 1M 窗口（DeepSeek V4.1 Flash）：遮蔽 200K→100K、整段摘要 400K - 删除逐条存根、语义 worker、soft/hard/target 比例与自行取消 Pi 压缩的门控； Pi 原生阈值压缩保留为最终安全网 Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- AGENTS.md | 2 +- README.md | 8 +- config/pi-compaction.json | 26 +- extensions/context-view.ts | 2 +- pi-compaction/AGENTS.md | 20 +- pi-compaction/README.md | 64 +++-- pi-compaction/UPSTREAM.md | 4 + pi-compaction/index.ts | 250 +++++++++++----- pi-compaction/src/config.ts | 93 +++--- pi-compaction/src/masking.ts | 239 ++++++++++++++++ pi-compaction/src/segment-index.ts | 277 ++---------------- pi-compaction/src/semantic-worker.ts | 189 ------------ pi-compaction/src/service.ts | 10 +- pi-compaction/src/token-accounting.ts | 16 +- pi-compaction/src/types.ts | 78 ++--- pi-compaction/test/config.test.ts | 88 ++++-- pi-compaction/test/extension-runtime.test.ts | 286 +++++++++++++++---- pi-compaction/test/masking.test.ts | 188 ++++++++++++ pi-compaction/test/performance.test.ts | 5 +- pi-compaction/test/segment-index.test.ts | 109 ++++--- pi-compaction/test/token-accounting.test.ts | 24 +- 21 files changed, 1174 insertions(+), 804 deletions(-) create mode 100644 pi-compaction/src/masking.ts delete mode 100644 pi-compaction/src/semantic-worker.ts create mode 100644 pi-compaction/test/masking.test.ts diff --git a/AGENTS.md b/AGENTS.md index ac1b57d..f0de8c8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,7 +18,7 @@ | `pi-hermes-memory/` | Hermes 风格持久记忆：policy-only 注入、六记忆工具、`skill_manage` | | `pi-minimal-footer/` | 紧凑 footer：上下文仪表、订阅用量条、扩展状态 | | `pi-notify/` | Kitty OSC 99 完成与权限确认通知，精确聚焦来源窗口 | -| `pi-compaction/` | 增量 Session segment、确定性压缩、后台语义摘要、effective token、native compaction | +| `pi-compaction/` | 压力触发的观察遮蔽（Pi context_edit）、阈值整段摘要、effective token、native compaction | | `extensions/fff-override.ts` | FFF `override` 模式接管 `find`/`grep`/`multi_grep` 与 `@` 补全 | | `extensions/codegraph.ts` | 原生 `codegraph_explore`，自有 stdio JSON-RPC 直连 `codegraph serve`（`codegraph-client.ts`） | | `extensions/context-mode.ts` | 经 `prompt-guard.ts` 加载 context-mode adapter 与 skills（`context-mode-filter.ts` 过滤冲突锚点） | diff --git a/README.md b/README.md index 5873815..58967c5 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,7 @@ - `pi-hermes-memory/`：从官方 `pi-hermes-memory v0.9.10` 源码导入的持久记忆扩展，提供全局/项目/用户/失败四类记忆、SQLite FTS5 会话与记忆检索、`SKILL.md` 过程记忆、correction 检测与自动 consolidation。 - 本仓库维护的 `pi-minimal-footer`：基于 `@ogulcancelik/pi-minimal-footer@0.1.10`，用紧凑的上下文仪表和订阅用量条替换默认 footer。 - 本仓库维护的 `pi-notify`：基于 `@smoose/pi-notify@0.1.1`，使用 Kitty OSC 99 在 Pi 完全 settled 或权限系统即将请求人工确认时通知，并支持点击精确聚焦来源 Kitty 窗口。 -- 本仓库维护的 `pi-compaction`：以增量投影统一确定性工具结果压缩、后台语义摘要、effective token 计量、`context_tree_query`、portable checkpoint 与 Codex native compaction。 +- 本仓库维护的 `pi-compaction`：压力触发时以 Pi `context_edit` 批量遮蔽旧工具结果、空闲时按阈值触发整段摘要、effective token 计量、`context_tree_query`、portable checkpoint 与 Codex native compaction。 - 本仓库维护的 `pi-ask-user`：提供模型主动调用的 `ask_user_question`；TUI 中保留单题、多题、选择、文本、自定义回答与提交前复核，Paseo 等兼容 RPC 客户端通过标准 Pi extension UI 逐题交互。 - 本仓库自维护的 `pi-worktree`：纯斜杠命令（`/wt` / `/worktree`）git worktree 管理，`/wt new` 一键拆分、`/wt merge` 一键合并回来源分支（默认 ff-only，可选 `--squash`），通过 `SessionManager.forkFrom` + `switchSession` 切换会话 cwd；登记状态在 `/pi-worktree/state.json`，不注册模型工具、不 push。 - 本仓库维护的 `pi-tool-search`：从完整工具定义生成并缓存经过校验的工作流分组，默认以最多 5 个动态组、28 个动态工具的 LRU 策略加载原始完整 schema，并支持原子批量激活。 @@ -377,11 +377,11 @@ compaction 和排队 follow-up 完成前不会提前通知；同时监听 `pi-pe ### Pi Compaction -组合包只加载仓库内维护的 `pi-compaction/`。它以增量 Segment Index 管理已经进入 Session 的历史：已闭合工具结果先同步替换为有界的确定性证据 stub，大于阈值的结果再由单 Session 有界后台 worker 生成语义摘要；Provider 请求不会等待后台摘要。原始工具证据不重复写入 custom entry，可通过 `context_tree_query` 使用 source key 或 tool call ID 有界恢复。 +组合包只加载仓库内维护的 `pi-compaction/`。它管理已经进入 Session 的历史投影：上下文压力达到阈值时，在 `turn_end` 批量提交 Pi 原生 `context_edit`，把最近 `keepRecentTurns` 轮之外的大型工具结果遮蔽为指向 `context_tree_query` 的单行占位符；遮蔽成批发生、受 trigger/target 回差与 `minClearTokens` 约束。原始工具结果保留在 Session 中（UI、导出不变），可通过 `context_tree_query` 按 tool call ID 有界恢复。 -普通 Provider、阈值判断、Codex native compaction、`/context`、`/compaction-status` 和 minimal footer 都读取同一个 effective projection snapshot。Pi Core 的 raw threshold 只负责唤醒：只有 effective token 下界达到默认 90% 才执行 threshold compaction；manual 与 overflow 始终允许继续并在 native 失败时回退 Pi 默认 compaction。Codex native API 直接消费 effective ResponseItems，同时保存有界 portable summary 供模型或 Provider 切换。 +普通 Provider、阈值判断、Codex native compaction、`/context`、`/compaction-status` 和 minimal footer 都读取同一个 effective projection snapshot。整段摘要只在空闲 `agent_settled` 且达到绝对 token 阈值（默认 400K，受窗口比例约束，272K 窗口时约为 90%）时经 `ctx.compact` 触发一次；Pi 自身的阈值压缩作为最终安全网始终有效，`session_before_compact` 不再按扩展自身阈值取消。Codex native API 直接消费 effective ResponseItems，同时保存有界 portable summary 供模型或 Provider 切换。 -唯一配置位于已安装组合包的 `config/pi-compaction.json`，默认 soft/hard/target 比例为 72%/90%/55%，扩展只在内存中归一化且不写回配置。`/compaction-status` 显示 raw/effective/reclaimed token、置信度、segment 和后台任务。完整配置、Session version 1 格式和双上游来源见 [`pi-compaction/README.md`](pi-compaction/README.md) 与 [`pi-compaction/UPSTREAM.md`](pi-compaction/UPSTREAM.md)。 +唯一配置位于已安装组合包的 `config/pi-compaction.json`：`masking` 段为遮蔽触发/目标阈值、`keepRecentTurns`、`minEntryTokens` 与 `excludeTools`，`summary` 段为整段摘要阈值与 thinking 档位，扩展只在内存中归一化且不写回配置。`/compaction-status` 显示 raw/effective/reclaimed token、置信度、segment 和已遮蔽结果数。完整配置、Session version 1 格式和双上游来源见 [`pi-compaction/README.md`](pi-compaction/README.md) 与 [`pi-compaction/UPSTREAM.md`](pi-compaction/UPSTREAM.md)。 原生远程压缩同时支持 `openai-codex` + `openai-codex-responses` 和 `sub-api` + `openai-responses`，使用触发时当前会话模型。SubAPI 保留现有模型配置，从 `baseUrl` 调用公共 `/responses` 并使用 `compaction_trigger` 流式协议；优先保留非空显式 `Authorization`，否则使用 API key，也支持仅凭非空 `cf-aig-authorization` 的代理认证。缺失或空白凭据在请求前拒绝。SubAPI 不要求客户端解析 ChatGPT account ID，也不调用独立 `/responses/compact`；Codex 的 JWT 要求保持不变。网关部署须支持 Sub2API 原生压缩 v2；请求失败沿用 Pi 回退，切换模型或 Provider 时使用 portable summary。 diff --git a/config/pi-compaction.json b/config/pi-compaction.json index ad0c00c..6b86ce9 100644 --- a/config/pi-compaction.json +++ b/config/pi-compaction.json @@ -1,20 +1,20 @@ { "enabled": true, - "softRatio": 0.72, - "hardRatio": 0.9, - "targetRatio": 0.55, - "deterministic": { + "masking": { "enabled": true, - "maxStubTokens": 1024 + "triggerTokens": 200000, + "triggerRatio": 0.6, + "targetTokens": 100000, + "targetRatio": 0.35, + "minClearTokens": 50000, + "keepRecentTurns": 10, + "minEntryTokens": 256, + "excludeTools": ["ask_user_question"] }, - "semantic": { - "enabled": true, - "model": "default", - "thinking": "low", - "concurrency": 1, - "minInputChars": 4000, - "maxInputChars": 24000, - "maxRetries": 2 + "summary": { + "triggerTokens": 400000, + "triggerRatio": 0.9, + "thinking": "low" }, "native": { "enabled": true diff --git a/extensions/context-view.ts b/extensions/context-view.ts index 4276a20..ee0b28e 100644 --- a/extensions/context-view.ts +++ b/extensions/context-view.ts @@ -17,7 +17,7 @@ export default function contextView(pi: ExtensionAPI): void { `Raw: ${formatNumber(metrics.rawTokens)} / ${formatNumber(metrics.contextWindow)}`, `Reclaimed: ${formatNumber(metrics.reclaimedTokens)}`, `Segments: ${formatNumber(metrics.segmentCount)}`, - `Semantic jobs: ${formatNumber(metrics.semanticJobs)}`, + `Masked tool results: ${formatNumber(metrics.maskedResults)}`, `Generation: ${formatNumber(metrics.generation)}`, ].join("\n") : "No pi-compaction projection is available yet. Send one message to build the first snapshot."; diff --git a/pi-compaction/AGENTS.md b/pi-compaction/AGENTS.md index 7361fac..666beee 100644 --- a/pi-compaction/AGENTS.md +++ b/pi-compaction/AGENTS.md @@ -1,19 +1,23 @@ # my-pi 维护约定（pi-compaction） -- 扩展入口是 `pi-compaction/index.ts`，增量索引、投影、后台摘要、服务与 Codex native 实现在 `src/`，测试位于 `test/`，双上游来源记录是 `UPSTREAM.md`。 +- 扩展入口是 `pi-compaction/index.ts`，遮蔽规划、投影索引、遮蔽台账、服务与 Codex native 实现在 `src/`，测试位于 `test/`，双上游来源记录是 `UPSTREAM.md`。 - 根包必须通过 `file:./pi-compaction` 提供依赖并直接加载源码；不得重新声明或加载 `pi-condense`、npm `@ogulcancelik/pi-codex-compaction` 或 npm `pi-context-view`。 -- `pi-compaction` 只支持自己的 version 1 记录和 `config/pi-compaction.json` 配置，不读取或迁移旧 `context-prune-*`、`contextPrune` 或 `openai-codex-native-compaction` 状态；切换后应使用新 Session。配置不写回 `settings.json`，默认 soft/hard/target 比例为 72%/90%/55%。 -- Context Authority 是 compaction 唯一生产投影者。普通 Provider、Codex native、`/context`、`/compaction-status` 和 footer 必须消费同一个不可变 projection snapshot；`pi-minimal-footer` 优先显示同一 effective usage，snapshot 尚未建立时才回退宿主原始 usage；background worker 不得在 Provider 请求中被 await。 +- `pi-compaction` 只支持自己的 version 1 记录和 `config/pi-compaction.json` 配置，不读取或迁移旧 `context-prune-*`、`contextPrune`、`openai-codex-native-compaction` 或旧 `segment-summary` 状态；切换后应使用新 Session。配置不写回 `settings.json`。 +- Context Authority 是 compaction 唯一生产投影者。普通 Provider、Codex native、`/context`、`/compaction-status` 和 footer 必须消费同一个不可变 projection snapshot；`pi-minimal-footer` 优先显示同一 effective usage，snapshot 尚未建立时才回退宿主原始 usage。 - `pi-compaction` 负责已经进入 Session 的历史投影，与 Context Mode 的输入隔离职责互补。首次 Session hydration 允许一次完整扫描；后续普通 `context` hot path 必须复用不可变 snapshot 或只处理 delta，不得调用 summarizer、写 Session 或重新构建完整 branch。 -- 已闭合工具结果先同步生成保留工具名、参数、状态、首尾证据和 source key 的有界确定性 stub；大于 `semantic.minInputChars` 的结果再进入单 Session 有界后台语义 worker。后台结果只在安全事件边界提交，陈旧 hash 必须丢弃，Provider 请求不得等待后台摘要。 -- Segment custom entry 只能保存摘要、hash 和 source reference，不得复制完整原始工具输出。`context_tree_query` 恢复结果必须有界，native checkpoint 的 portable summary 也必须有界。 -- effective token 默认使用 Provider usage-backed baseline 加 projected delta；无可靠 baseline 时使用带上下界的估算。Pi Core raw threshold 只负责唤醒，只有 effective lower bound 达到 `hardRatio` 才执行 threshold compaction；manual 与 overflow 不得被阻断。 +- L2 遮蔽只用 Pi 原生 `context_edit`（`{ type: "context_edit", targetId, replacement }`）处理保护窗口外的旧 toolResult：成批提交、受 trigger/target 回差与 `minClearTokens` 约束，不逐条即时替换；不为遮蔽写 `segment-summary` 等 custom entry，原始条目、UI 与导出保持不变。 +- 不得改写最近 `keepRecentTurns` 轮；不得遮蔽 user/assistant 消息、`excludeTools` 中的工具、`SKILL.md` 的 `read` 结果、已遮蔽或已省略的条目、以及小于 `minEntryTokens` 的结果。 +- 遮蔽占位符必须确定性、单行、以 `MASK_PREFIX`（`"[pi-compaction] Earlier "`）开头，并在文本中指向 `context_tree_query` 恢复；`context_tree_query` 与 native checkpoint 的 portable summary 都必须有界。 +- L3 整段摘要只能从空闲的 `agent_settled` 通过 `ctx.compact` 触发（`ctx.compact` 会中止当前运行，非空闲时调用会破坏运行）；同一 leaf 不重复触发。阈值以绝对 token 计并受窗口比例约束（`effectiveThreshold = min(tokens, floor(ratio × contextWindow))`）。 +- `context` 钩子不得改写消息——非 checkpoint 路径只更新 snapshot 并返回 undefined；唯一例外是 native checkpoint 路径：checkpoint 之后的条目在投影时必须应用各 target 最新的 `context_edit`（null 省略、字符串规范成 text block），因为 `sessionEntryToContextMessages` 不处理 edits。 +- `session_before_compact` 只负责生成 portable/native 摘要，不得按自身阈值取消 Pi 的压缩（Pi 的 `contextTokens > window - reserveTokens` 是最终安全网）。 +- effective token 默认使用 Provider usage-backed baseline 加 projected delta；无可靠 baseline 时使用带上下界的估算。遮蔽台账在 `session_start` / `session_tree` / `session_compact` 从当前 branch 的 `context_edit` 条目重建，指标暴露 `maskedResults` 与 `reclaimedTokens`。 - Codex native compaction 必须直接消费当前 projection 生成的 ResponseItems，不得从 raw branch 重建输入；native checkpoint 同时保存有界 portable summary 供模型或 Provider 切换。 -- 原生远程压缩限定为 `openai-codex` + `openai-codex-responses` 与 `sub-api` + `openai-responses`，默认在 turn boundary 达到 effective 下界 90% 时触发。`src/native-provider.ts` 统一资格与传输选择：Codex 保留原 endpoint/JWT/account ID；SubAPI 从显式 `baseUrl` 解析公共 `/responses`，使用配置的 API key 和请求头，不解析 ChatGPT account ID、不硬编码网关域名，也不调用独立 `/responses/compact`。两者复用 projection、`compaction_trigger`、SSE 结果、native checkpoint 恢复和 portable 模型切换摘要；SubAPI 部署须支持原生 compaction v2。普通聊天支持不代表压缩支持，失败回退 Pi，不自动改写用户模型配置。 +- 原生远程压缩限定为 `openai-codex` + `openai-codex-responses` 与 `sub-api` + `openai-responses`，默认由 L3 阈值或 Pi 自身阈值触发。`src/native-provider.ts` 统一资格与传输选择：Codex 保留原 endpoint/JWT/account ID；SubAPI 从显式 `baseUrl` 解析公共 `/responses`，使用配置的 API key 和请求头，不解析 ChatGPT account ID、不硬编码网关域名，也不调用独立 `/responses/compact`。两者复用 projection、`compaction_trigger`、SSE 结果、native checkpoint 恢复和 portable 模型切换摘要；SubAPI 部署须支持原生 compaction v2。普通聊天支持不代表压缩支持，失败回退 Pi，不自动改写用户模型配置。 - SubAPI 原生压缩认证必须保留非空显式 `Authorization`（大小写不敏感）的优先级；没有该头时才使用非空 API key 生成 Bearer，均缺失时可接受非空 `cf-aig-authorization` 代理认证头。不得要求 header-only 配置额外提供假 API key，也不得生成 `Bearer undefined`；空白或缺失凭据在网络请求前拒绝并回退 Pi。Codex 仍必须提供有效 JWT，不能用请求头绕过。 - 同步 Codex 协议代码必须从官方 `ogulcancelik/pi-extensions` 明确 commit 移植并保留对应 MIT 许可证；不得从 npm tarball、缓存、`node_modules` 或构建产物导入。 ## 验证 - 在本目录内至少运行 `npm test`、`npm run check` 和 `npm pack --dry-run --json`；涉及依赖、加载入口、context writer、footer 或 observer 时还要验证根 `npm install` 幂等、锁文件仅包含预期本地 link 变化、根扩展联合加载、性能回归，以及实际 packed 根 tarball 的隔离安装/加载。 -- 修改 projection、token accounting、threshold、checkpoint、模型切换、后台并发或恢复索引后，至少运行 `npm test`、`npm run check`、`npm pack --dry-run --json`、根扩展联合加载、packed 根 tarball 隔离加载和性能回归。性能门禁为 cache-hit P95 <2ms、delta projection P95 <10ms，普通 hot path 不得扫描完整 branch。 +- 修改 projection、token accounting、遮蔽规划/台账、threshold、checkpoint、模型切换或恢复索引后，至少运行 `npm test`、`npm run check`、`npm pack --dry-run --json`、根扩展联合加载、packed 根 tarball 隔离加载和性能回归。性能门禁为 cache-hit P95 <2ms、delta projection P95 <10ms，普通 hot path 不得扫描完整 branch。 diff --git a/pi-compaction/README.md b/pi-compaction/README.md index d83a09f..4999f14 100644 --- a/pi-compaction/README.md +++ b/pi-compaction/README.md @@ -1,17 +1,20 @@ # pi-compaction -High-performance unified context projection and compaction for Pi. +Pressure-aware context masking and threshold compaction for Pi. ## Responsibilities -- incrementally index completed tool calls; -- replace large tool results with deterministic bounded evidence stubs; -- pass small tool results through unchanged whenever a stub would not be shorter; -- refine large stubs with non-blocking semantic summaries; -- expose one effective context projection to normal provider requests and Codex/SubAPI native compaction; -- cancel Pi's raw threshold compaction until the effective lower-bound reaches the configured threshold; -- retain original tool evidence through `context_tree_query`; -- publish read-only effective metrics for the footer and context observer. +Pi context management is layered; this extension owns the middle two layers: + +- **L1 — entry truncation** (future `pi-overflow`): truncate new tool results as they are written. Not implemented here. +- **L2 — observation masking** (this extension): under real context pressure, replace old tool results outside a protected recent window with deterministic one-line placeholders via Pi `context_edit` entries. Originals stay in the session untouched and are recoverable with `context_tree_query`. +- **L3 — whole-context summary** (this extension): when the session idles (`agent_settled`) above the summary threshold, trigger `ctx.compact` once per leaf. +- **Pi native threshold compaction** is the final safety net and is never blocked; this extension no longer cancels `session_before_compact` based on its own thresholds. + +Additionally it: + +- publishes one effective-context snapshot for normal provider requests and Codex/SubAPI native compaction; +- maintains a masked-results ledger so `/context`, `/compaction-status`, the footer, and AutoReview see the same placeholders the model sees. Context Mode remains responsible for preventing large external data from entering the session. RTK remains responsible for ordinary tool-output cleanup. @@ -22,21 +25,21 @@ Configuration lives in the installed bundle's `config/pi-compaction.json` (the b ```json { "enabled": true, - "softRatio": 0.72, - "hardRatio": 0.9, - "targetRatio": 0.55, - "deterministic": { + "masking": { "enabled": true, - "maxStubTokens": 1024 + "triggerTokens": 200000, + "triggerRatio": 0.6, + "targetTokens": 100000, + "targetRatio": 0.35, + "minClearTokens": 50000, + "keepRecentTurns": 10, + "minEntryTokens": 256, + "excludeTools": ["ask_user_question"] }, - "semantic": { - "enabled": true, - "model": "default", - "thinking": "low", - "concurrency": 1, - "minInputChars": 4000, - "maxInputChars": 24000, - "maxRetries": 2 + "summary": { + "triggerTokens": 400000, + "triggerRatio": 0.9, + "thinking": "low" }, "native": { "enabled": true @@ -44,7 +47,9 @@ Configuration lives in the installed bundle's `config/pi-compaction.json` (the b } ``` -Invalid or missing fields are normalized in memory. +All thresholds are absolute token counts constrained by a context-window ratio: `effectiveThreshold = min(tokens, floor(ratio × contextWindow))`. Masking fires when `currentTokens ≥ triggerEff` and clears the oldest eligible results until `currentTokens - cleared ≤ targetEff`; if the clearable volume stays under `min(minClearTokens, triggerEff - targetEff)` nothing is masked, so masking is rare, batched, and hysteretic. `masking.target` is normalized below `masking.trigger` and `summary.trigger` above `masking.trigger` when a config would violate that ordering. Invalid or missing fields fall back to defaults in memory. + +Masking never touches the most recent `keepRecentTurns` assistant turns, user/assistant messages, `excludeTools`, `SKILL.md` reads, entries already masked, or results smaller than `minEntryTokens`. ## Native providers @@ -67,17 +72,16 @@ The gateway must have a compatible deployed version, model and available upstrea ## Session format -The only custom entry type is `pi-compaction`, version 1. Segment summary entries contain hashes, summaries, and source references; they never duplicate complete raw tool output. Native checkpoints use Pi compaction-entry details with the same kind/version. +The extension writes no `segment-summary` or other custom entries for masking: L2 uses Pi's native `context_edit` entries (`{ type: "context_edit", targetId, replacement }`), which alter only future model context. The raw session entries — and therefore UI, exports, and the evidence ledger — are unchanged; every masked result remains recoverable through `context_tree_query`. Masking placeholders are deterministic single lines starting with `[pi-compaction] Earlier `. -Projection replaces a tool result only when the resulting summary plus its `Source:` recovery line is strictly shorter than the original text content. Small tool results are therefore passed through byte-for-byte and keep their original content parts, including any non-text parts. Because a stub is only larger than the text it would replace when the original is short, this comparison is the sole gate; no separate size option exists. Every indexed result remains recoverable through `context_tree_query` whether or not the message was replaced. - -This implementation intentionally does not read legacy `context-prune-*` or `openai-codex-native-compaction` entries. Start a new session after switching to this extension. +Compaction entries still carry `pi-compaction` version 1 details; native checkpoints use the same kind/version. This implementation intentionally does not read legacy `context-prune-*`, `openai-codex-native-compaction`, or older `segment-summary` entries. Start a new session after switching to this extension. ## Interfaces -- `context_tree_query`: recover original bounded tool evidence by source key or tool call ID. -- `/compaction-status`: show raw/effective tokens, reclaimed tokens, confidence, segments, and pending semantic jobs. +- `context_tree_query`: recover the original output of a tool result that pi-compaction cleared from context, by tool call ID (1–20 ids per call, bounded output). Use it when a `[pi-compaction] Earlier` placeholder lacks evidence needed for the current task. +- `/compaction-status`: show raw/effective tokens, reclaimed tokens, confidence, segments, masked results, and generation. +- Service `my-pi.pi-compaction.service.v1`: `getSnapshot`, `getMetrics`, and `query(sessionId, ids)` — `query` returns a placeholder record only for tool call IDs that are currently masked, so consumers such as the permission system's AutoReview see the same evidence the model does. ## Performance model -The first session hydration scans the branch once. Appended turns use delta projection and immutable cached snapshots. The `context` hook performs no model call and no session write. Semantic summaries run in a bounded background worker and become visible only at later safe event boundaries. +The first session hydration scans the branch once. Appended turns use delta projection and immutable cached snapshots; the `context` hook performs no model call, no session write, and returns the message list unchanged except on the native-checkpoint path, where post-checkpoint `context_edit` entries are applied during projection. Masking drafts are produced in `turn_end` in one batch and only when both the trigger and the `minClear` hysteresis pass, so the hot path stays allocation-free for unprompted turns. diff --git a/pi-compaction/UPSTREAM.md b/pi-compaction/UPSTREAM.md index 2a88ebd..2323aae 100644 --- a/pi-compaction/UPSTREAM.md +++ b/pi-compaction/UPSTREAM.md @@ -29,3 +29,7 @@ - 协议证据：`backend/internal/service/openai_compact_body_signal.go` 指出原生远端压缩 v2 为 Responses 加 `stream: true` 与 `compaction_trigger`，并处理 Codex beta 特性。 - 本地 `src/native-provider.ts` 适配器是针对该协议独立实现的。没有导入任何 Sub2API 源码；复用现有 Codex 序列化器、流解析器和 checkpoint 格式。 - 该参考描述的是上游能力，不构成对任何特定部署、模型或账号的保证。本适配器不使用单独的 Responses compact 端点。 + +## 本地改动记录 + +- 上下文管理重构为 L2/L3：L2 在上下文压力下经 `turn_end` 提交 Pi 原生 `context_edit`，把保护窗口外的旧 toolResult 批量遮蔽为指向 `context_tree_query` 的确定性占位符（原始条目保留、不写 custom entry）；L3 只在空闲 `agent_settled` 且达到绝对 token 阈值（受窗口比例约束）时调用一次 `ctx.compact`。删除逐条确定性存根、后台语义 worker、`softRatio`/`hardRatio`/`targetRatio` 配置与 `session_before_compact` 的自行阈值取消；Pi 原生阈值压缩保留为最终安全网。 diff --git a/pi-compaction/index.ts b/pi-compaction/index.ts index a9edd5e..9e87bd2 100644 --- a/pi-compaction/index.ts +++ b/pi-compaction/index.ts @@ -14,6 +14,12 @@ import { type SessionEntry, } from "@earendil-works/pi-coding-agent"; import { DEFAULT_CONFIG, loadConfig } from "./src/config.js"; +import { + applyContextEdits, + planMasking, + rebuildMaskedLedger, + type MaskedLedgerEntry, +} from "./src/masking.js"; import { buildNativeCompactionTransport, isNativeCompactionModel } from "./src/native-provider.js"; import { buildCompactionRequestBody, @@ -32,27 +38,28 @@ import { type NativeCompactionDetails, type ResponseItem, } from "./src/native-compaction.js"; -import { SegmentIndex } from "./src/segment-index.js"; -import { SemanticWorker } from "./src/semantic-worker.js"; -import { shouldCompact } from "./src/token-accounting.js"; +import { contentText, SegmentIndex } from "./src/segment-index.js"; import { CompactionServiceRegistry, publishCompactionService, unpublishCompactionService, } from "./src/service.js"; +import { effectiveThreshold } from "./src/token-accounting.js"; import { COMPACTION_CUSTOM_TYPE, COMPACTION_VERSION, QUERY_TOOL_NAME, + type MaskedResultRecord, type PiCompactionConfig, type ProjectionSnapshot, - type ToolHistoryRecord, } from "./src/types.js"; interface SessionRuntime { index: SegmentIndex; config: PiCompactionConfig; codexCache?: { fingerprint: string; modelKey: string; toolsHash: string; items: ResponseItem[] }; + masked: Map; + lastAutoCompactLeafId: string | null; } interface CachedPayload { @@ -64,10 +71,6 @@ function errorMessage(error: unknown): string { return error instanceof Error ? error.message : String(error); } -function hash(value: string): string { - return createHash("sha256").update(value).digest("hex").slice(0, 20); -} - interface EnvironmentValue { hash: string; tokens: number; @@ -81,6 +84,10 @@ interface EnvironmentCacheEntry extends EnvironmentValue { tools: unknown[]; } +function hash(value: string): string { + return createHash("sha256").update(value).digest("hex").slice(0, 20); +} + function environment( pi: ExtensionAPI, ctx: ExtensionContext, @@ -182,34 +189,46 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { const runtimes = new Map(); const payloads = new Map(); const environments = new Map(); - const worker = new SemanticWorker(); const service = new CompactionServiceRegistry(); publishCompactionService(service); + const maskedRecords = (sessionId: string, ids: string[]): MaskedResultRecord[] => { + const masked = runtimes.get(sessionId)?.masked; + if (!masked) return []; + return ids.flatMap((id) => { + const entry = masked.get(id); + return entry + ? [{ + toolCallId: id, + resultTimestamp: entry.resultTimestamp, + deterministicSummary: entry.placeholder, + }] + : []; + }); + }; + const runtimeFor = (ctx: ExtensionContext): SessionRuntime => { const sessionId = ctx.sessionManager.getSessionId(); let runtime = runtimes.get(sessionId); if (!runtime) { - const index = new SegmentIndex(sessionId); - index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], DEFAULT_CONFIG); - runtime = { index, config: structuredClone(DEFAULT_CONFIG) }; + runtime = { + index: new SegmentIndex(sessionId), + config: structuredClone(DEFAULT_CONFIG), + masked: rebuildLedger(ctx), + lastAutoCompactLeafId: null, + }; runtimes.set(sessionId, runtime); - service.setSession(sessionId, (ids) => index.query(ids)); + service.setSession(sessionId, (ids) => maskedRecords(sessionId, ids)); } return runtime; }; - const persist = (record: unknown): void => { - pi.appendEntry(COMPACTION_CUSTOM_TYPE, record); - }; - - const drainSemantic = (ctx: ExtensionContext, runtime: SessionRuntime): void => { - const sessionId = ctx.sessionManager.getSessionId(); - for (const result of worker.drain(sessionId)) { - const record = runtime.index.applySemantic(result.sourceKey, result.sourceHash, result.summary); - if (record) persist(record); + const ledgerStats = (runtime: SessionRuntime): { maskedResults: number; reclaimedTokens: number } => { + let reclaimedTokens = 0; + for (const entry of runtime.masked.values()) { + reclaimedTokens += Math.max(0, entry.originalTokens - entry.placeholderTokens); } - runtime.index.setSemanticJobs(worker.count(sessionId)); + return { maskedResults: runtime.masked.size, reclaimedTokens }; }; const projectMessages = ( @@ -218,16 +237,32 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { messages: readonly any[], ): ProjectionSnapshot => { const env = environment(pi, ctx, environments); - runtime.index.setSemanticJobs(worker.count(ctx.sessionManager.getSessionId())); + const stats = ledgerStats(runtime); const snapshot = runtime.index.project(messages, { environmentHash: env.hash, environmentTokens: env.tokens, contextWindow: env.contextWindow, + maskedResults: stats.maskedResults, + reclaimedTokens: stats.reclaimedTokens, }); service.setSnapshot(snapshot); return snapshot; }; + const rebuildLedger = (ctx: ExtensionContext): Map => + rebuildMaskedLedger( + ctx.sessionManager.getBranch(), + ctx.sessionManager.getLeafId(), + (message) => estimateTokens(message as any), + ); + + const hydrateRuntime = (ctx: ExtensionContext, runtime: SessionRuntime): void => { + runtime.index.hydrate(); + runtime.masked = rebuildLedger(ctx); + runtime.lastAutoCompactLeafId = null; + runtime.codexCache = undefined; + }; + const branchMessages = ( ctx: ExtensionContext, runtime: SessionRuntime, @@ -237,9 +272,10 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { const checkpoint = findNativeCheckpoint(branch); let messages: any[]; if (checkpoint.status === "valid") { - messages = branch - .slice(checkpoint.checkpoint.entryIndex + 1) - .flatMap((entry) => sessionEntryToContextMessages(entry)); + // sessionEntryToContextMessages sees only raw entries; applyContextEdits + // additionally replays the latest context_edit per target (omissions and + // placeholder replacements included). + messages = applyContextEdits(branch.slice(checkpoint.checkpoint.entryIndex + 1)); if (checkpoint.checkpoint.details.modelKey !== (ctx.model ? modelKey(ctx.model) : "")) { messages = [asPortableMessage(checkpoint.checkpoint.details.portableSummary), ...messages]; } @@ -278,15 +314,22 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { publishCompactionService(service); const sessionId = ctx.sessionManager.getSessionId(); const config = await loadConfig(); - const index = new SegmentIndex(sessionId); - index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], config); - runtimes.set(sessionId, { index, config }); - service.setSession(sessionId, (ids) => index.query(ids)); + const runtime = runtimes.get(sessionId) ?? { + index: new SegmentIndex(sessionId), + config, + masked: new Map(), + lastAutoCompactLeafId: null, + }; + runtime.config = config; + runtime.index.hydrate(); + runtime.masked = rebuildLedger(ctx); + runtime.lastAutoCompactLeafId = null; + runtimes.set(sessionId, runtime); + service.setSession(sessionId, (ids) => maskedRecords(sessionId, ids)); }); pi.on("session_shutdown", (_event, ctx) => { const sessionId = ctx.sessionManager.getSessionId(); - worker.cancelSession(sessionId); runtimes.delete(sessionId); payloads.delete(sessionId); environments.delete(sessionId); @@ -296,15 +339,13 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { pi.on("session_tree", (_event, ctx) => { const runtime = runtimeFor(ctx); - runtime.index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], runtime.config); - runtime.codexCache = undefined; + hydrateRuntime(ctx, runtime); service.clearSnapshot(ctx.sessionManager.getSessionId()); }); pi.on("session_compact", (_event, ctx) => { const runtime = runtimeFor(ctx); - runtime.index.hydrate(ctx.sessionManager.getBranch() as SessionEntry[], runtime.config); - runtime.codexCache = undefined; + hydrateRuntime(ctx, runtime); service.clearSnapshot(ctx.sessionManager.getSessionId()); payloads.delete(ctx.sessionManager.getSessionId()); }); @@ -319,21 +360,9 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { pi.on("turn_end", (event, ctx) => { const runtime = runtimeFor(ctx); - drainSemantic(ctx, runtime); - const previous = runtime.index.getSnapshot(); - const captured = runtime.index.captureTurn(event.message, event.toolResults, runtime.config); - for (const record of captured) persist(runtime.index.toPersisted(record)); - const pressure = previous && previous.contextWindow > 0 - ? previous.upperBound / previous.contextWindow >= runtime.config.softRatio - : false; - const semanticCandidates = pressure - ? runtime.index.allRecords() - : captured.filter((record) => record.rawText.length >= runtime.config.semantic.maxInputChars); - for (const record of semanticCandidates) { - worker.enqueue(ctx.sessionManager.getSessionId(), record, ctx, runtime.config); - } - runtime.index.setSemanticJobs(worker.count(ctx.sessionManager.getSessionId())); + if (!runtime.config.enabled) return undefined; + const previous = runtime.index.getSnapshot(); const usage = ctx.getContextUsage(); if (usage?.tokens && previous) { const env = environment(pi, ctx, environments); @@ -343,6 +372,50 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { env.hash, ); } + + const currentTokens = usage?.tokens + ?? (event.context?.contextMessages ?? []).reduce((sum, message) => sum + estimateTokens(message as any), 0) + + environment(pi, ctx, environments).tokens; + const plan = planMasking({ + contextEntries: event.context?.contextEntries ?? [], + currentTokens, + contextWindow: ctx.model?.contextWindow ?? 0, + masking: runtime.config.masking, + estimate: (message) => estimateTokens(message as any), + }); + if (!plan.drafts.length) return undefined; + for (const result of plan.masked) { + runtime.masked.set(result.toolCallId, { + placeholder: result.placeholder, + originalTokens: result.originalTokens, + placeholderTokens: result.placeholderTokens, + resultTimestamp: result.resultTimestamp, + }); + } + return { entries: [...(event.entries ?? []), ...plan.drafts] }; + }); + + pi.on("agent_settled", (_event, ctx) => { + const runtime = runtimeFor(ctx); + if (!runtime.config.enabled || ctx.isIdle?.() === false) return; + const snapshot = runtime.index.getSnapshot(); + const tokens = ctx.getContextUsage()?.tokens ?? snapshot?.effectiveTokens; + if (tokens == null) return; + const contextWindow = ctx.model?.contextWindow ?? snapshot?.contextWindow ?? 0; + const threshold = effectiveThreshold( + runtime.config.summary.triggerTokens, + runtime.config.summary.triggerRatio, + contextWindow, + ); + if (tokens < threshold) return; + const leafId = ctx.sessionManager.getLeafId(); + if (leafId !== null && runtime.lastAutoCompactLeafId === leafId) return; + runtime.lastAutoCompactLeafId = leafId; + ctx.compact({ + onError: (error) => { + if (ctx.hasUI) ctx.ui.notify(`pi-compaction auto summary failed: ${errorMessage(error)}`, "warning"); + }, + }); }); pi.on("context", (event, ctx) => { @@ -358,7 +431,10 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { const projected = branchMessages(ctx, runtime, branch); return { messages: projected.snapshot.messages as any }; } - return { messages: projectMessages(ctx, runtime, event.messages).messages as any }; + // Non-checkpoint path only refreshes the shared snapshot; the message list + // is returned to Pi unchanged — masking is append-only via context_edit. + projectMessages(ctx, runtime, event.messages); + return undefined; }); pi.on("before_provider_headers", (event, ctx) => { @@ -387,7 +463,6 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { pi.on("session_before_compact", async (event, ctx) => { const runtime = runtimeFor(ctx); if (!runtime.config.enabled) return undefined; - drainSemantic(ctx, runtime); const { snapshot, checkpoint } = branchMessages( ctx, runtime, @@ -395,8 +470,6 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { event.reason === "overflow" && event.willRetry, ); - if (!shouldCompact(event.reason, snapshot, runtime.config)) return { cancel: true }; - const model = ctx.model; if (!model) return undefined; if (!runtime.config.native.enabled || !isNativeCompactionModel(model)) { @@ -412,7 +485,7 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { event.signal, event.customInstructions, undefined, - runtime.config.semantic.thinking, + runtime.config.summary.thinking, ); return { compaction: { @@ -463,7 +536,7 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { replacementHistory: buildReplacementHistory( input, remote.compactionItem, - Math.min(64_000, Math.max(4_096, Math.floor(snapshot.contextWindow * Math.max(0.05, runtime.config.targetRatio - 0.3)))), + Math.min(64_000, Math.max(4_096, Math.floor(snapshot.contextWindow * 0.25))), ), portableSummary: portableSummary(snapshot.messages), sourceFingerprint: snapshot.fingerprint, @@ -504,7 +577,7 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { event.signal, event.preparation.customInstructions, undefined, - runtime.config.semantic.thinking, + runtime.config.summary.thinking, undefined, undefined, undefined, @@ -531,38 +604,64 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { pi.registerTool({ name: QUERY_TOOL_NAME, label: "Query Compacted Tool History", - description: "Recover original tool outputs by source key or tool call id after pi-compaction replaced them with bounded summaries.", - promptSnippet: "Recover original compacted tool evidence", + description: "Recover the original output of a tool result that pi-compaction cleared from context, by tool call id.", + promptSnippet: "Recover original tool output cleared from context", promptGuidelines: [ - "Use context_tree_query only when a pi-compaction summary lacks evidence needed for the current task.", + "Use context_tree_query when you see a \"[pi-compaction] Earlier\" placeholder and still need the original tool output.", ], parameters: Type.Object({ toolCallIds: Type.Array(Type.String(), { minItems: 1, maxItems: 20 }), }), async execute(_id, params, _signal, _update, ctx) { - const records = service.query(ctx.sessionManager.getSessionId(), params.toolCallIds); - const blocks = records.map((record: ToolHistoryRecord) => { - const perResultBytes = Math.max(1_024, Math.floor(DEFAULT_MAX_BYTES / Math.max(1, records.length))); - const args = truncateHead(JSON.stringify(record.args), { maxLines: 20, maxBytes: 2_048 }).content; - const truncated = truncateHead(record.rawText, { - maxLines: Math.max(20, Math.floor(DEFAULT_MAX_LINES / Math.max(1, records.length))), + const branch = ctx.sessionManager.getBranch() as SessionEntry[]; + const calls = new Map(); + const results = new Map(); + for (const entry of branch) { + const message = (entry as any)?.message; + if ((entry as any)?.type !== "message" || !message) continue; + if (message.role === "assistant" && Array.isArray(message.content)) { + for (const block of message.content) { + if (block?.type === "toolCall" && typeof block.id === "string") { + calls.set(block.id, { + toolName: String(block.name ?? "unknown"), + args: block.arguments ?? block.input ?? block.args ?? {}, + }); + } + } + } else if (message.role === "toolResult" && typeof message.toolCallId === "string") { + results.set(message.toolCallId, message); + } + } + + const found: string[] = []; + const blocks: string[] = []; + for (const id of params.toolCallIds) { + const result = results.get(id); + if (!result) continue; + found.push(id); + const call = calls.get(id) ?? { toolName: String(result.toolName ?? "unknown"), args: {} }; + const perResultBytes = Math.max(1_024, Math.floor(DEFAULT_MAX_BYTES / Math.max(1, params.toolCallIds.length))); + const args = truncateHead(JSON.stringify(call.args ?? {}), { maxLines: 20, maxBytes: 2_048 }).content; + const rawText = contentText(result.content); + const truncated = truncateHead(rawText, { + maxLines: Math.max(20, Math.floor(DEFAULT_MAX_LINES / Math.max(1, params.toolCallIds.length))), maxBytes: perResultBytes, }); - return [ - `## ${record.sourceKey}`, - `Tool: ${record.toolName}`, + blocks.push([ + `## ${id}`, + `Tool: ${call.toolName}`, `Args: ${args}`, - `Status: ${record.isError ? "ERROR" : "OK"}`, + `Status: ${result.isError ? "ERROR" : "OK"}`, "", truncated.content, ...(truncated.truncated ? [`[truncated: ${truncated.outputLines}/${truncated.totalLines} lines]`] : []), - ].join("\n"); - }); - const missing = params.toolCallIds.filter((id: string) => !records.some((record) => record.sourceKey === id || record.toolCallId === id)); + ].join("\n")); + } + const missing = params.toolCallIds.filter((id: string) => !results.has(id)); for (const id of missing) blocks.push(`## ${id}\n(not found)`); return { content: [{ type: "text", text: blocks.join("\n\n---\n\n") }], - details: { sourceKeys: records.map((record) => record.sourceKey), missing }, + details: { toolCallIds: found, missing }, }; }, }); @@ -576,12 +675,11 @@ export default function piCompactionExtension(pi: ExtensionAPI): void { `Effective: ${metrics.effectiveTokens}/${metrics.contextWindow} (${metrics.percent.toFixed(1)}%, ${metrics.confidence})`, `Raw: ${metrics.rawTokens}/${metrics.contextWindow}`, `Reclaimed: ${metrics.reclaimedTokens}`, - `Segments: ${metrics.segmentCount}; semantic jobs: ${metrics.semanticJobs}; generation: ${metrics.generation}`, + `Segments: ${metrics.segmentCount}; masked tool results: ${metrics.maskedResults}; generation: ${metrics.generation}`, ].join("\n") : "pi-compaction has no projection snapshot yet."; if (ctx.hasUI) ctx.ui.notify(text, "info"); else process.stderr.write(`${text}\n`); }, }); - } diff --git a/pi-compaction/src/config.ts b/pi-compaction/src/config.ts index 36a378e..b0b80d5 100644 --- a/pi-compaction/src/config.ts +++ b/pi-compaction/src/config.ts @@ -12,27 +12,29 @@ export const CONFIG_PATH = fileURLToPath( export const DEFAULT_CONFIG: PiCompactionConfig = { enabled: true, - softRatio: 0.72, - hardRatio: 0.9, - targetRatio: 0.55, - deterministic: { + masking: { enabled: true, - maxStubTokens: 1_024, + triggerTokens: 200_000, + triggerRatio: 0.6, + targetTokens: 100_000, + targetRatio: 0.35, + minClearTokens: 50_000, + keepRecentTurns: 10, + minEntryTokens: 256, + excludeTools: ["ask_user_question"], }, - semantic: { - enabled: true, - model: "default", + summary: { + triggerTokens: 400_000, + triggerRatio: 0.9, thinking: "low", - concurrency: 1, - minInputChars: 4_000, - maxInputChars: 24_000, - maxRetries: 2, }, native: { enabled: true, }, }; +const THINKING_LEVELS = ["off", "minimal", "low", "medium", "high", "xhigh"]; + function ratio(value: unknown, fallback: number): number { return typeof value === "number" && Number.isFinite(value) && value > 0 && value < 1 ? value : fallback; } @@ -41,46 +43,67 @@ function integer(value: unknown, fallback: number, min = 0, max = Number.MAX_SAF return typeof value === "number" && Number.isInteger(value) && value >= min && value <= max ? value : fallback; } +function positiveInteger(value: unknown, fallback: number): number { + return integer(value, fallback, 1); +} + function object(value: unknown): Record { return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; } +function stringArray(value: unknown, fallback: string[]): string[] { + if (!Array.isArray(value) || !value.every((entry) => typeof entry === "string")) return fallback; + return value.map((entry) => entry.trim()).filter((entry) => entry.length > 0); +} + export function normalizeConfig(value: unknown): PiCompactionConfig { const input = object(value); - const deterministic = object(input.deterministic); - const semantic = object(input.semantic); + const masking = object(input.masking); + const summary = object(input.summary); const native = object(input.native); - const softRatio = ratio(input.softRatio, DEFAULT_CONFIG.softRatio); - const hardRatio = Math.max(softRatio + 0.01, ratio(input.hardRatio, DEFAULT_CONFIG.hardRatio)); - const targetRatio = Math.min(softRatio - 0.01, ratio(input.targetRatio, DEFAULT_CONFIG.targetRatio)); - const thinking = ["off", "minimal", "low", "medium", "high", "xhigh"].includes(String(semantic.thinking)) - ? semantic.thinking as PiCompactionConfig["semantic"]["thinking"] - : DEFAULT_CONFIG.semantic.thinking; - return { + const normalized: PiCompactionConfig = { enabled: typeof input.enabled === "boolean" ? input.enabled : DEFAULT_CONFIG.enabled, - softRatio, - hardRatio: Math.min(0.99, hardRatio), - targetRatio: Math.max(0.1, targetRatio), - deterministic: { - enabled: typeof deterministic.enabled === "boolean" ? deterministic.enabled : DEFAULT_CONFIG.deterministic.enabled, - maxStubTokens: integer(deterministic.maxStubTokens, DEFAULT_CONFIG.deterministic.maxStubTokens, 32, 4_096), + masking: { + enabled: typeof masking.enabled === "boolean" ? masking.enabled : DEFAULT_CONFIG.masking.enabled, + triggerTokens: positiveInteger(masking.triggerTokens, DEFAULT_CONFIG.masking.triggerTokens), + triggerRatio: ratio(masking.triggerRatio, DEFAULT_CONFIG.masking.triggerRatio), + targetTokens: positiveInteger(masking.targetTokens, DEFAULT_CONFIG.masking.targetTokens), + targetRatio: ratio(masking.targetRatio, DEFAULT_CONFIG.masking.targetRatio), + minClearTokens: positiveInteger(masking.minClearTokens, DEFAULT_CONFIG.masking.minClearTokens), + keepRecentTurns: integer(masking.keepRecentTurns, DEFAULT_CONFIG.masking.keepRecentTurns, 1, 100), + minEntryTokens: integer(masking.minEntryTokens, DEFAULT_CONFIG.masking.minEntryTokens, 0, 100_000), + excludeTools: stringArray(masking.excludeTools, DEFAULT_CONFIG.masking.excludeTools), }, - semantic: { - enabled: typeof semantic.enabled === "boolean" ? semantic.enabled : DEFAULT_CONFIG.semantic.enabled, - model: typeof semantic.model === "string" && semantic.model.trim() ? semantic.model.trim() : DEFAULT_CONFIG.semantic.model, - thinking, - concurrency: integer(semantic.concurrency, DEFAULT_CONFIG.semantic.concurrency, 1, 4), - minInputChars: integer(semantic.minInputChars, DEFAULT_CONFIG.semantic.minInputChars, 0, 1_000_000), - maxInputChars: integer(semantic.maxInputChars, DEFAULT_CONFIG.semantic.maxInputChars, 1_000, 1_000_000), - maxRetries: integer(semantic.maxRetries, DEFAULT_CONFIG.semantic.maxRetries, 0, 5), + summary: { + triggerTokens: positiveInteger(summary.triggerTokens, DEFAULT_CONFIG.summary.triggerTokens), + triggerRatio: ratio(summary.triggerRatio, DEFAULT_CONFIG.summary.triggerRatio), + thinking: THINKING_LEVELS.includes(String(summary.thinking)) + ? summary.thinking as PiCompactionConfig["summary"]["thinking"] + : DEFAULT_CONFIG.summary.thinking, }, native: { enabled: typeof native.enabled === "boolean" ? native.enabled : DEFAULT_CONFIG.native.enabled, }, }; + + // Trigger/target must keep their hysteresis ordering; fix up rather than + // letting a degenerate pair mask-or-compact every turn. + if (normalized.masking.targetTokens >= normalized.masking.triggerTokens) { + normalized.masking.targetTokens = Math.floor(normalized.masking.triggerTokens / 2); + } + if (normalized.masking.targetRatio >= normalized.masking.triggerRatio) { + normalized.masking.targetRatio = normalized.masking.triggerRatio / 2; + } + if (normalized.summary.triggerTokens <= normalized.masking.triggerTokens) { + normalized.summary.triggerTokens = normalized.masking.triggerTokens * 2; + } + if (normalized.summary.triggerRatio <= normalized.masking.triggerRatio) { + normalized.summary.triggerRatio = Math.min(0.99, normalized.masking.triggerRatio + 0.01); + } + return normalized; } export async function loadConfig( diff --git a/pi-compaction/src/masking.ts b/pi-compaction/src/masking.ts new file mode 100644 index 0000000..fd7b30e --- /dev/null +++ b/pi-compaction/src/masking.ts @@ -0,0 +1,239 @@ +import { + buildContextEntries, + sessionEntryToContextMessages, + type ContextEditEntryDraft, +} from "@earendil-works/pi-coding-agent"; +import { contentText, timestampOf } from "./segment-index.js"; +import { effectiveThreshold } from "./token-accounting.js"; +import type { MaskingConfig } from "./types.js"; + +// Every placeholder written through a context_edit starts with this prefix so +// re-masking, ledger rebuilds, and the model-facing recovery hint all key off +// one marker. +export const MASK_PREFIX = "[pi-compaction] Earlier "; + +const MAX_PLACEHOLDER_ARGS_CHARS = 300; +const EDITABLE_ROLES = new Set(["user", "assistant", "toolResult", "custom"]); + +export interface MaskingProjectedEntry { + sourceEntry: { id?: string; type?: string; message?: any }; + messages: any[]; +} + +export interface MaskedResult { + toolCallId: string; + entryId: string; + placeholder: string; + originalTokens: number; + placeholderTokens: number; + resultTimestamp: number; +} + +export interface MaskingPlan { + drafts: ContextEditEntryDraft[]; + clearedTokens: number; + masked: MaskedResult[]; +} + +export interface MaskedLedgerEntry { + placeholder: string; + originalTokens: number; + placeholderTokens: number; + resultTimestamp: number; +} + +const EMPTY_PLAN: MaskingPlan = { drafts: [], clearedTokens: 0, masked: [] }; + +function json(value: unknown): string { + try { + return JSON.stringify(value) ?? "undefined"; + } catch { + return "[unserializable]"; + } +} + +function maskPlaceholder(toolName: string, args: unknown, toolCallId: string, originalTokens: number): string { + let argsText = json(args); + if (argsText.length > MAX_PLACEHOLDER_ARGS_CHARS) { + argsText = `${argsText.slice(0, MAX_PLACEHOLDER_ARGS_CHARS)}…`; + } + return `${MASK_PREFIX}${toolName} result cleared to save context (~${originalTokens} tokens). ` + + `Call args: ${argsText}. Re-run the call if still needed, or recover the original with ` + + `context_tree_query {"toolCallIds":["${toolCallId}"]}.`; +} + +function isSkillRead(toolName: string, args: unknown): boolean { + if (toolName !== "read" || args === null || typeof args !== "object") return false; + const record = args as Record; + const raw = record.path ?? record.filePath ?? record.file; + if (typeof raw !== "string") return false; + return (raw.split(/[\\/]/).pop() ?? raw) === "SKILL.md"; +} + +function toolResultCalls(entries: readonly MaskingProjectedEntry[]): Map { + const calls = new Map(); + for (const entry of entries) { + for (const message of entry.messages ?? []) { + if (message?.role !== "assistant" || !Array.isArray(message.content)) continue; + for (const block of message.content) { + if (block?.type === "toolCall" && typeof block.id === "string") { + calls.set(block.id, { + toolName: String(block.name ?? "unknown"), + args: block.arguments ?? block.input ?? block.args ?? {}, + }); + } + } + } + } + return calls; +} + +export function planMasking(params: { + contextEntries: readonly MaskingProjectedEntry[]; + currentTokens: number; + contextWindow: number; + masking: MaskingConfig; + estimate: (message: any) => number; +}): MaskingPlan { + const { masking } = params; + const triggerEff = effectiveThreshold(masking.triggerTokens, masking.triggerRatio, params.contextWindow); + if (!masking.enabled || params.currentTokens < triggerEff) return EMPTY_PLAN; + const targetEff = effectiveThreshold(masking.targetTokens, masking.targetRatio, params.contextWindow); + const minClear = Math.min(masking.minClearTokens, Math.max(0, triggerEff - targetEff)); + + const entries = params.contextEntries; + // The last keepRecentTurns assistant messages and everything after them stay + // untouched; when fewer turns exist, the whole context is protected. + let protectedIndex = 0; + let turns = 0; + for (let index = entries.length - 1; index >= 0; index--) { + const entry = entries[index]; + if (entry?.sourceEntry?.type === "message" && (entry.messages ?? []).some((m) => m?.role === "assistant")) { + turns++; + if (turns >= masking.keepRecentTurns) { + protectedIndex = index; + break; + } + } + } + + const calls = toolResultCalls(entries); + const needed = params.currentTokens - targetEff; + const masked: MaskedResult[] = []; + let clearedTokens = 0; + for (let index = 0; index < protectedIndex && clearedTokens < needed; index++) { + const entry = entries[index]; + const entryId = entry?.sourceEntry?.id; + if (entry?.sourceEntry?.type !== "message" || typeof entryId !== "string") continue; + for (const message of entry.messages ?? []) { + if (clearedTokens >= needed) break; + if (message?.role !== "toolResult" || typeof message.toolCallId !== "string") continue; + const call = calls.get(message.toolCallId); + const toolName = call?.toolName ?? String(message.toolName ?? "unknown"); + if (masking.excludeTools.includes(toolName)) continue; + const args = call?.args ?? {}; + if (isSkillRead(toolName, args)) continue; + if (contentText(message.content).startsWith(MASK_PREFIX)) continue; + const originalTokens = params.estimate(message); + if (originalTokens < masking.minEntryTokens) continue; + const placeholder = maskPlaceholder(toolName, args, message.toolCallId, originalTokens); + const placeholderTokens = params.estimate({ ...message, content: [{ type: "text", text: placeholder }] }); + clearedTokens += originalTokens - placeholderTokens; + masked.push({ + toolCallId: message.toolCallId, + entryId, + placeholder, + originalTokens, + placeholderTokens, + resultTimestamp: timestampOf(message.timestamp), + }); + } + } + if (clearedTokens < minClear) return EMPTY_PLAN; + return { + drafts: masked.map((result) => ({ + type: "context_edit", + targetId: result.entryId, + replacement: { content: result.placeholder }, + })), + clearedTokens, + masked, + }; +} + +/** + * Rebuild the masked-results ledger from a session branch after hydration: + * the latest context_edit per target wins, only edits carrying this + * extension's placeholder prefix count as masked, and only targets that are + * still part of the active context (post-compaction kept set) are counted. + */ +export function rebuildMaskedLedger( + branch: readonly any[], + leafId: string | null | undefined, + estimate: (message: any) => number, +): Map { + const latestEdits = new Map(); + const targets = new Map(); + for (const entry of branch) { + if (entry?.type === "context_edit" && typeof entry.targetId === "string") { + latestEdits.set(entry.targetId, entry); + } else if (entry?.type === "message" && typeof entry.id === "string") { + targets.set(entry.id, entry); + } + } + const contextIds = new Set( + buildContextEntries(branch as any, leafId).map((entry) => entry.id), + ); + const masked = new Map(); + for (const [targetId, edit] of latestEdits) { + if (!contextIds.has(targetId)) continue; + const text = contentText(edit.replacement?.content); + if (!text.startsWith(MASK_PREFIX)) continue; + const message = targets.get(targetId)?.message; + if (message?.role !== "toolResult" || typeof message.toolCallId !== "string") continue; + masked.set(message.toolCallId, { + placeholder: text, + originalTokens: estimate(message), + placeholderTokens: estimate({ ...message, content: [{ type: "text", text }] }), + resultTimestamp: timestampOf(message.timestamp), + }); + } + return masked; +} + +/** + * Apply the latest context_edit per target the way buildSessionProjection does + * for message entries `sessionEntryToContextMessages` alone would miss: null + * omits the entry, a string replacement becomes one text block for + * assistant/toolResult roles. + */ +export function applyContextEdits(entries: readonly any[]): any[] { + const edits = new Map(); + for (const entry of entries) { + if (entry?.type === "context_edit" && typeof entry.targetId === "string") { + edits.set(entry.targetId, entry.replacement ?? null); + } + } + const projected: any[] = []; + for (const entry of entries) { + const messages = sessionEntryToContextMessages(entry); + const edit = edits.get(entry?.id); + if (edit === undefined) { + projected.push(...messages); + continue; + } + if (edit === null) continue; + for (const message of messages) { + if (!EDITABLE_ROLES.has(message?.role)) { + projected.push(message); + continue; + } + const content = (message.role === "assistant" || message.role === "toolResult") + && typeof edit.content === "string" + ? [{ type: "text", text: edit.content }] + : edit.content; + projected.push({ ...message, content }); + } + } + return projected; +} diff --git a/pi-compaction/src/segment-index.ts b/pi-compaction/src/segment-index.ts index 9ceb6e4..b5e2ec2 100644 --- a/pi-compaction/src/segment-index.ts +++ b/pi-compaction/src/segment-index.ts @@ -1,16 +1,8 @@ import { createHash } from "node:crypto"; import { estimateTokens } from "@earendil-works/pi-coding-agent"; -import { - COMPACTION_CUSTOM_TYPE, - COMPACTION_VERSION, - type PiCompactionConfig, - type ProjectionSnapshot, - type SegmentSummaryRecord, - type ToolHistoryRecord, - type UsageBaseline, -} from "./types.js"; +import type { ProjectionSnapshot, UsageBaseline } from "./types.js"; -function timestampOf(value: unknown): number { +export function timestampOf(value: unknown): number { if (typeof value === "number" && Number.isFinite(value)) return value; if (typeof value === "string") { const parsed = Date.parse(value); @@ -19,10 +11,6 @@ function timestampOf(value: unknown): number { return 0; } -export function sourceKey(toolCallId: string, timestamp: unknown): string { - return `${toolCallId}@${timestampOf(timestamp)}`; -} - export function contentText(content: unknown): string { if (typeof content === "string") return content; if (!Array.isArray(content)) return ""; @@ -37,42 +25,6 @@ function hash(value: string): string { return createHash("sha256").update(value).digest("hex").slice(0, 20); } -function json(value: unknown): string { - try { - return JSON.stringify(value) ?? "undefined"; - } catch { - return "[unserializable]"; - } -} - -function boundedEvidence(text: string, maxChars: number): string { - const normalized = text.replace(/\u001b\[[0-9;]*m/g, "").trim(); - if (normalized.length <= maxChars) return normalized; - const marker = `\n… ${normalized.length - maxChars} chars omitted …\n`; - const available = Math.max(0, maxChars - marker.length); - const head = Math.ceil(available * 0.7); - const tail = Math.max(0, available - head); - return `${normalized.slice(0, head)}${marker}${tail ? normalized.slice(-tail) : ""}`; -} - -export function deterministicSummary(params: { - toolName: string; - args: unknown; - rawText: string; - isError: boolean; - maxStubTokens: number; -}): string { - const maxChars = params.maxStubTokens * 4; - const argsText = boundedEvidence(json(params.args), Math.min(600, Math.floor(maxChars * 0.25))); - const evidenceBudget = Math.max(128, maxChars - argsText.length - 120); - const evidence = boundedEvidence(params.rawText || "(no output)", evidenceBudget); - return [ - `[pi-compaction tool=${params.toolName} status=${params.isError ? "error" : "ok"}]`, - `Args: ${argsText}`, - `Evidence:\n${evidence}`, - ].join("\n"); -} - function messageKey(message: any): string { const role = String(message?.role ?? "unknown"); const timestamp = timestampOf(message?.timestamp); @@ -85,24 +37,14 @@ function messageKey(message: any): string { return `${role}:${timestamp}:${typeof message?.content === "string" ? message.content.length : (message?.content?.length ?? 0)}`; } -function isSummaryRecord(value: unknown): value is SegmentSummaryRecord { - if (!value || typeof value !== "object") return false; - const record = value as Partial; - return record.version === COMPACTION_VERSION - && record.kind === "segment-summary" - && typeof record.sourceKey === "string" - && typeof record.toolCallId === "string" - && typeof record.resultTimestamp === "number" - && typeof record.toolName === "string" - && typeof record.sourceHash === "string" - && typeof record.deterministicSummary === "string"; -} - +/** + * Incremental context projection: tracks the branch's message list, estimates + * tokens, and maintains the usage-backed effective-token baseline. Projection + * is identity — pi-compaction never rewrites messages on this path; context + * pressure reduction happens through append-only context_edit entries. + */ export class SegmentIndex { - private readonly records = new Map(); - private readonly ids = new Map(); private generation = 0; - private semanticJobs = 0; private usageBaseline?: UsageBaseline; private sourceMessages: any[] = []; private sourceKeys: string[] = []; @@ -117,67 +59,13 @@ export class SegmentIndex { this.sessionId = sessionId; } - hydrate(branch: readonly any[], config: PiCompactionConfig): void { + hydrate(): void { this.reset(); - if (!config.enabled) return; - const summaries = new Map(); - for (const entry of branch) { - if (entry?.type === "custom" && entry.customType === COMPACTION_CUSTOM_TYPE && isSummaryRecord(entry.data)) { - summaries.set(entry.data.sourceKey, entry.data); - } - } - - const pendingCalls = new Map(); - for (const entry of branch) { - if (entry?.type !== "message") continue; - const message = entry.message; - if (message?.role === "assistant" && Array.isArray(message.content)) { - for (const block of message.content) { - if (block?.type === "toolCall" && typeof block.id === "string") { - pendingCalls.set(block.id, { - toolName: String(block.name ?? "unknown"), - args: block.input ?? block.arguments ?? block.args ?? {}, - }); - } - } - } else if (message?.role === "toolResult" && typeof message.toolCallId === "string") { - const call = pendingCalls.get(message.toolCallId) ?? { - toolName: String(message.toolName ?? "unknown"), - args: {}, - }; - const key = sourceKey(message.toolCallId, message.timestamp); - const rawText = contentText(message.content); - const persisted = summaries.get(key); - const record: ToolHistoryRecord = { - sourceKey: key, - toolCallId: message.toolCallId, - resultTimestamp: timestampOf(message.timestamp), - toolName: call.toolName, - args: call.args, - isError: Boolean(message.isError), - rawText, - sourceHash: hash(`${call.toolName}\0${json(call.args)}\0${rawText}`), - deterministicSummary: persisted?.deterministicSummary ?? deterministicSummary({ - toolName: call.toolName, - args: call.args, - rawText, - isError: Boolean(message.isError), - maxStubTokens: config.deterministic.maxStubTokens, - }), - semanticSummary: persisted?.semanticSummary, - }; - this.put(record); - pendingCalls.delete(message.toolCallId); - } - } this.generation++; } reset(): void { - this.records.clear(); - this.ids.clear(); this.generation = 0; - this.semanticJobs = 0; this.usageBaseline = undefined; this.sourceMessages = []; this.sourceKeys = []; @@ -187,86 +75,6 @@ export class SegmentIndex { this.cached = undefined; } - private put(record: ToolHistoryRecord): void { - this.records.set(record.sourceKey, record); - const keys = this.ids.get(record.toolCallId) ?? []; - if (!keys.includes(record.sourceKey)) keys.push(record.sourceKey); - this.ids.set(record.toolCallId, keys); - } - - captureTurn(message: any, toolResults: any[], config: PiCompactionConfig): ToolHistoryRecord[] { - if (!config.enabled || !config.deterministic.enabled || !Array.isArray(message?.content)) return []; - const calls = new Map(); - for (const block of message.content) { - if (block?.type === "toolCall" && typeof block.id === "string") calls.set(block.id, block); - } - const captured: ToolHistoryRecord[] = []; - for (const result of toolResults ?? []) { - if (!result || typeof result.toolCallId !== "string") continue; - const call = calls.get(result.toolCallId); - if (!call) continue; - const rawText = contentText(result.content); - const toolName = String(call.name ?? result.toolName ?? "unknown"); - const args = call.input ?? call.arguments ?? call.args ?? {}; - const key = sourceKey(result.toolCallId, result.timestamp); - const record: ToolHistoryRecord = { - sourceKey: key, - toolCallId: result.toolCallId, - resultTimestamp: timestampOf(result.timestamp), - toolName, - args, - isError: Boolean(result.isError), - rawText, - sourceHash: hash(`${toolName}\0${json(args)}\0${rawText}`), - deterministicSummary: deterministicSummary({ - toolName, - args, - rawText, - isError: Boolean(result.isError), - maxStubTokens: config.deterministic.maxStubTokens, - }), - }; - this.put(record); - captured.push(record); - } - if (captured.length) { - this.generation++; - this.cached = undefined; - } - return captured; - } - - applySemantic(sourceKeyValue: string, sourceHash: string, summary: string): SegmentSummaryRecord | undefined { - const record = this.records.get(sourceKeyValue); - if (!record || record.sourceHash !== sourceHash || !summary.trim()) return undefined; - record.semanticSummary = summary.trim(); - this.generation++; - this.cached = undefined; - return this.toPersisted(record); - } - - toPersisted(record: ToolHistoryRecord): SegmentSummaryRecord { - const effective = record.semanticSummary ?? record.deterministicSummary; - return { - version: COMPACTION_VERSION, - kind: "segment-summary", - sourceKey: record.sourceKey, - toolCallId: record.toolCallId, - resultTimestamp: record.resultTimestamp, - toolName: record.toolName, - sourceHash: record.sourceHash, - deterministicSummary: record.deterministicSummary, - ...(record.semanticSummary ? { semanticSummary: record.semanticSummary } : {}), - rawChars: record.rawText.length, - effectiveChars: effective.length, - }; - } - - setSemanticJobs(count: number): void { - this.semanticJobs = Math.max(0, count); - if (this.cached) this.cached = { ...this.cached, semanticJobs: this.semanticJobs }; - } - recordUsage(tokens: number, messageEstimate: number, environmentHash: string): void { if (!Number.isFinite(tokens) || tokens <= 0) return; this.usageBaseline = { tokens, messageEstimate, environmentHash }; @@ -277,11 +85,27 @@ export class SegmentIndex { environmentHash: string; environmentTokens: number; contextWindow: number; + maskedResults: number; + reclaimedTokens: number; }): ProjectionSnapshot { const sameTail = messages.length === this.sourceMessages.length && messages.length > 0 && messageKey(messages[messages.length - 1]) === this.sourceKeys[this.sourceKeys.length - 1]; - if (this.cached && sameTail && this.cached.environmentHash === params.environmentHash) return this.cached; + if (this.cached && sameTail && this.cached.environmentHash === params.environmentHash) { + if (this.cached.maskedResults === params.maskedResults + && this.cached.reclaimedTokens === params.reclaimedTokens) { + return this.cached; + } + // Masking stats changed but the message list did not: publish a new + // immutable snapshot object instead of re-projecting or mutating. + this.cached = { + ...this.cached, + maskedResults: params.maskedResults, + reclaimedTokens: params.reclaimedTokens, + rawTokens: this.cached.effectiveTokens + params.reclaimedTokens, + }; + return this.cached; + } const appendOnly = this.sourceMessages.length > 0 && messages.length >= this.sourceMessages.length @@ -301,17 +125,14 @@ export class SegmentIndex { for (let index = start; index < messages.length; index++) { const message = messages[index]; - const projected = this.projectMessage(message); this.sourceMessages.push(message); this.sourceKeys.push(messageKey(message)); - this.projectedMessages.push(projected); + this.projectedMessages.push(message); this.rawEstimates.push(estimateTokens(message as any)); - this.effectiveEstimates.push(estimateTokens(projected as any)); + this.effectiveEstimates.push(estimateTokens(message as any)); } - const rawMessageTokens = this.rawEstimates.reduce((sum, value) => sum + value, 0); const effectiveMessageTokens = this.effectiveEstimates.reduce((sum, value) => sum + value, 0); - const rawTokens = rawMessageTokens + params.environmentTokens; const baseline = this.usageBaseline; const usageBacked = baseline?.environmentHash === params.environmentHash; const effectiveTokens = usageBacked @@ -329,55 +150,19 @@ export class SegmentIndex { environmentHash: params.environmentHash, messages: this.projectedMessages, messageEstimate: effectiveMessageTokens, - rawTokens, + rawTokens: effectiveTokens + params.reclaimedTokens, effectiveTokens, - reclaimedTokens: Math.max(0, rawTokens - effectiveTokens), + reclaimedTokens: params.reclaimedTokens, contextWindow: params.contextWindow, confidence, lowerBound, upperBound, segmentCount: this.sourceMessages.length, - semanticJobs: this.semanticJobs, + maskedResults: params.maskedResults, }; return this.cached; } - private projectMessage(message: any): any { - if (message?.role !== "toolResult" || typeof message.toolCallId !== "string") return message; - const record = this.records.get(sourceKey(message.toolCallId, message.timestamp)); - if (!record) return message; - const original = contentText(message.content); - const summary = record.semanticSummary ?? record.deterministicSummary; - const projected = `${summary}\nSource: ${record.sourceKey}; recover with context_tree_query.`; - if (projected.length >= original.length) return message; - return { - ...message, - content: [{ type: "text", text: projected }], - }; - } - - query(ids: string[]): ToolHistoryRecord[] { - const found: ToolHistoryRecord[] = []; - const seen = new Set(); - for (const id of ids) { - const direct = this.records.get(id); - const keys = direct ? [id] : this.ids.get(id) ?? []; - for (const key of keys) { - if (seen.has(key)) continue; - const record = this.records.get(key); - if (record) { - found.push(record); - seen.add(key); - } - } - } - return found; - } - - allRecords(): ToolHistoryRecord[] { - return [...this.records.values()]; - } - getSnapshot(): ProjectionSnapshot | undefined { return this.cached; } diff --git a/pi-compaction/src/semantic-worker.ts b/pi-compaction/src/semantic-worker.ts deleted file mode 100644 index 53d564e..0000000 --- a/pi-compaction/src/semantic-worker.ts +++ /dev/null @@ -1,189 +0,0 @@ -import type { ExtensionContext } from "@earendil-works/pi-coding-agent"; -import type { PiCompactionConfig, ToolHistoryRecord } from "./types.js"; - -interface Job { - sessionId: string; - sourceKey: string; - sourceHash: string; - toolName: string; - args: unknown; - isError: boolean; - rawText: string; - ctx: ExtensionContext; - config: PiCompactionConfig; - controller: AbortController; -} - -export interface SemanticResult { - sessionId: string; - sourceKey: string; - sourceHash: string; - summary: string; -} - -function modelFor(config: PiCompactionConfig, ctx: ExtensionContext): any { - if (config.semantic.model === "default") return ctx.model; - const slash = config.semantic.model.indexOf("/"); - if (slash <= 0) return ctx.model; - return ctx.modelRegistry.find( - config.semantic.model.slice(0, slash), - config.semantic.model.slice(slash + 1), - ) ?? ctx.model; -} - -function serialize(job: Job): string { - const args = (() => { - try { return JSON.stringify(job.args); } catch { return "[unserializable]"; } - })(); - const max = job.config.semantic.maxInputChars; - const raw = job.rawText.length <= max - ? job.rawText - : `${job.rawText.slice(0, Math.floor(max * 0.7))}\n… omitted …\n${job.rawText.slice(-Math.floor(max * 0.3))}`; - return `Summarize this completed coding-agent tool call for future context.\nKeep concrete outcomes, paths, identifiers, values, and exact errors.\nUse at most 6 concise bullets. Do not give advice and do not call tools.\n\nTool: ${job.toolName}\nStatus: ${job.isError ? "ERROR" : "OK"}\nArgs: ${args}\nOutput:\n${raw}`; -} - -async function run(job: Job): Promise { - const model = modelFor(job.config, job.ctx); - if (!model) throw new Error("No summarizer model selected"); - const message = await job.ctx.modelRegistry.complete(model, { - messages: [{ - role: "user", - content: [{ type: "text", text: serialize(job) }], - timestamp: Date.now(), - }], - } as any, { - signal: job.controller.signal, - reasoningEffort: job.config.semantic.thinking === "off" ? undefined : job.config.semantic.thinking, - } as any); - if (message.stopReason === "length" || message.stopReason === "error" || message.stopReason === "aborted") { - throw new Error(message.errorMessage ?? `Semantic compaction stopped: ${message.stopReason}`); - } - const text = message.content - .flatMap((part: any) => part.type === "text" && typeof part.text === "string" ? [part.text] : []) - .join("\n") - .trim(); - if (!text) throw new Error("Semantic compaction returned an empty summary"); - return text; -} - -function abortableDelay(ms: number, signal: AbortSignal): Promise { - if (signal.aborted) return Promise.reject(new Error("Semantic compaction aborted")); - return new Promise((resolve, reject) => { - const timer = setTimeout(done, ms); - function done(): void { - signal.removeEventListener("abort", aborted); - resolve(); - } - function aborted(): void { - clearTimeout(timer); - signal.removeEventListener("abort", aborted); - reject(new Error("Semantic compaction aborted")); - } - signal.addEventListener("abort", aborted, { once: true }); - }); -} - -export class SemanticWorker { - private readonly queue: Job[] = []; - private readonly active = new Map(); - private readonly ready: SemanticResult[] = []; - private readonly queuedKeys = new Set(); - - enqueue(sessionId: string, record: ToolHistoryRecord, ctx: ExtensionContext, config: PiCompactionConfig): void { - if (!config.semantic.enabled || record.semanticSummary || record.rawText.length < config.semantic.minInputChars) return; - const key = `${sessionId}:${record.sourceKey}:${record.sourceHash}`; - if (this.queuedKeys.has(key)) return; - this.queuedKeys.add(key); - this.queue.push({ - sessionId, - sourceKey: record.sourceKey, - sourceHash: record.sourceHash, - toolName: record.toolName, - args: record.args, - isError: record.isError, - rawText: record.rawText, - ctx, - config, - controller: new AbortController(), - }); - this.pump(config.semantic.concurrency); - } - - private pump(limit: number): void { - while (this.active.size < limit && this.queue.length > 0) { - const job = this.queue.shift()!; - const key = `${job.sessionId}:${job.sourceKey}:${job.sourceHash}`; - this.active.set(key, job); - void this.execute(key, job, limit); - } - } - - private async execute(key: string, job: Job, limit: number): Promise { - let lastError: unknown; - try { - for (let attempt = 0; attempt <= job.config.semantic.maxRetries; attempt++) { - try { - const summary = await run(job); - if (!job.controller.signal.aborted) { - this.ready.push({ - sessionId: job.sessionId, - sourceKey: job.sourceKey, - sourceHash: job.sourceHash, - summary, - }); - } - return; - } catch (error) { - lastError = error; - if (job.controller.signal.aborted || attempt === job.config.semantic.maxRetries) break; - await abortableDelay(Math.min(8_000, 1_000 * 2 ** attempt), job.controller.signal); - } - } - if (!job.controller.signal.aborted) { - console.error(`pi-compaction: semantic summary failed for ${job.sourceKey}:`, lastError); - } - } finally { - this.active.delete(key); - this.queuedKeys.delete(key); - this.pump(limit); - } - } - - drain(sessionId: string): SemanticResult[] { - const drained: SemanticResult[] = []; - for (let index = this.ready.length - 1; index >= 0; index--) { - if (this.ready[index]!.sessionId !== sessionId) continue; - drained.push(this.ready[index]!); - this.ready.splice(index, 1); - } - return drained.reverse(); - } - - count(sessionId: string): number { - return this.queue.filter((job) => job.sessionId === sessionId).length - + [...this.active.values()].filter((job) => job.sessionId === sessionId).length; - } - - cancelSession(sessionId: string): void { - for (const job of this.active.values()) { - if (job.sessionId === sessionId) job.controller.abort(); - } - for (let index = this.queue.length - 1; index >= 0; index--) { - if (this.queue[index]!.sessionId === sessionId) { - const job = this.queue[index]!; - this.queuedKeys.delete(`${job.sessionId}:${job.sourceKey}:${job.sourceHash}`); - this.queue.splice(index, 1); - } - } - for (let index = this.ready.length - 1; index >= 0; index--) { - if (this.ready[index]!.sessionId === sessionId) this.ready.splice(index, 1); - } - } - - cancelAll(): void { - for (const job of this.active.values()) job.controller.abort(); - this.queue.length = 0; - this.ready.length = 0; - this.queuedKeys.clear(); - } -} diff --git a/pi-compaction/src/service.ts b/pi-compaction/src/service.ts index 171e437..b3c764e 100644 --- a/pi-compaction/src/service.ts +++ b/pi-compaction/src/service.ts @@ -2,8 +2,8 @@ import { SERVICE_SYMBOL, type CompactionMetrics, type CompactionService, + type MaskedResultRecord, type ProjectionSnapshot, - type ToolHistoryRecord, } from "./types.js"; interface GlobalWithCompactionService { @@ -14,11 +14,11 @@ const root = globalThis as GlobalWithCompactionService; export class CompactionServiceRegistry implements CompactionService { private readonly snapshots = new Map(); - private readonly queryHandlers = new Map ToolHistoryRecord[]>(); + private readonly queryHandlers = new Map MaskedResultRecord[]>(); setSession( sessionId: string, - query: (ids: string[]) => ToolHistoryRecord[], + query: (ids: string[]) => MaskedResultRecord[], ): void { this.queryHandlers.set(sessionId, query); } @@ -58,11 +58,11 @@ export class CompactionServiceRegistry implements CompactionService { percent: snapshot.contextWindow > 0 ? 100 * snapshot.effectiveTokens / snapshot.contextWindow : 0, confidence: snapshot.confidence, segmentCount: snapshot.segmentCount, - semanticJobs: snapshot.semanticJobs, + maskedResults: snapshot.maskedResults, }; } - query(sessionId: string, ids: string[]): ToolHistoryRecord[] { + query(sessionId: string, ids: string[]): MaskedResultRecord[] { return this.queryHandlers.get(sessionId)?.(ids) ?? []; } } diff --git a/pi-compaction/src/token-accounting.ts b/pi-compaction/src/token-accounting.ts index 16f39b8..732fc3b 100644 --- a/pi-compaction/src/token-accounting.ts +++ b/pi-compaction/src/token-accounting.ts @@ -1,13 +1,5 @@ -import type { PiCompactionConfig, ProjectionSnapshot } from "./types.js"; - -export type CompactionReason = "manual" | "threshold" | "overflow"; - -export function shouldCompact( - reason: CompactionReason, - snapshot: Pick, - config: Pick, -): boolean { - if (reason === "manual" || reason === "overflow") return true; - if (snapshot.contextWindow <= 0) return true; - return snapshot.lowerBound / snapshot.contextWindow >= config.hardRatio; +// Effective trigger for one configured (tokens, ratio) pair: the absolute +// token figure wins on large windows, the ratio caps it on small ones. +export function effectiveThreshold(tokens: number, ratio: number, contextWindow: number): number { + return contextWindow > 0 ? Math.min(tokens, Math.floor(ratio * contextWindow)) : tokens; } diff --git a/pi-compaction/src/types.ts b/pi-compaction/src/types.ts index b842c19..0573200 100644 --- a/pi-compaction/src/types.ts +++ b/pi-compaction/src/types.ts @@ -5,58 +5,33 @@ export const COMPACTION_VERSION = 1 as const; export const QUERY_TOOL_NAME = "context_tree_query"; export const SERVICE_SYMBOL = Symbol.for("my-pi.pi-compaction.service.v1"); +export interface MaskingConfig { + enabled: boolean; + triggerTokens: number; + triggerRatio: number; + targetTokens: number; + targetRatio: number; + minClearTokens: number; + keepRecentTurns: number; + minEntryTokens: number; + excludeTools: string[]; +} + +export interface SummaryConfig { + triggerTokens: number; + triggerRatio: number; + thinking: "off" | "minimal" | "low" | "medium" | "high" | "xhigh"; +} + export interface PiCompactionConfig { enabled: boolean; - softRatio: number; - hardRatio: number; - targetRatio: number; - deterministic: { - enabled: boolean; - maxStubTokens: number; - }; - semantic: { - enabled: boolean; - model: string; - thinking: "off" | "minimal" | "low" | "medium" | "high" | "xhigh"; - concurrency: number; - minInputChars: number; - maxInputChars: number; - maxRetries: number; - }; + masking: MaskingConfig; + summary: SummaryConfig; native: { enabled: boolean; }; } -export interface SegmentSummaryRecord { - version: typeof COMPACTION_VERSION; - kind: "segment-summary"; - sourceKey: string; - toolCallId: string; - resultTimestamp: number; - toolName: string; - sourceHash: string; - deterministicSummary: string; - semanticSummary?: string; - rawChars: number; - effectiveChars: number; -} - -export type PiCompactionRecord = SegmentSummaryRecord; - -export interface ToolHistoryRecord { - sourceKey: string; - toolCallId: string; - resultTimestamp: number; - toolName: string; - args: unknown; - isError: boolean; - rawText: string; - sourceHash: string; - deterministicSummary: string; - semanticSummary?: string; -} - export interface UsageBaseline { tokens: number; messageEstimate: number; @@ -78,7 +53,7 @@ export interface ProjectionSnapshot { lowerBound: number; upperBound: number; segmentCount: number; - semanticJobs: number; + maskedResults: number; codexItems?: ResponseItem[]; } @@ -99,6 +74,13 @@ export interface PortableCheckpointDetails { sourceFingerprint: string; } +/** What the service returns for one masked tool result — the same placeholder the model sees. */ +export interface MaskedResultRecord { + toolCallId: string; + resultTimestamp: number; + deterministicSummary: string; +} + export interface CompactionMetrics { sessionId: string; generation: number; @@ -109,11 +91,11 @@ export interface CompactionMetrics { percent: number; confidence: ProjectionSnapshot["confidence"]; segmentCount: number; - semanticJobs: number; + maskedResults: number; } export interface CompactionService { getSnapshot(sessionId: string): ProjectionSnapshot | undefined; getMetrics(sessionId: string): CompactionMetrics | undefined; - query(sessionId: string, ids: string[]): ToolHistoryRecord[]; + query(sessionId: string, ids: string[]): MaskedResultRecord[]; } diff --git a/pi-compaction/test/config.test.ts b/pi-compaction/test/config.test.ts index e9e8330..10d9dbf 100644 --- a/pi-compaction/test/config.test.ts +++ b/pi-compaction/test/config.test.ts @@ -2,34 +2,82 @@ import test from "node:test"; import assert from "node:assert/strict"; import { DEFAULT_CONFIG, normalizeConfig } from "../src/config.ts"; -test("normalizes one greenfield piCompaction config without legacy aliases", () => { +test("normalizes a full pi-compaction config", () => { const config = normalizeConfig({ enabled: false, - softRatio: 0.7, - hardRatio: 0.91, - targetRatio: 0.5, - deterministic: { enabled: false, maxStubTokens: 512 }, - semantic: { enabled: false, concurrency: 2, maxRetries: 0 }, + masking: { + enabled: false, + triggerTokens: 150_000, + triggerRatio: 0.5, + targetTokens: 60_000, + targetRatio: 0.2, + minClearTokens: 30_000, + keepRecentTurns: 5, + minEntryTokens: 512, + excludeTools: ["bash", "read"], + }, + summary: { triggerTokens: 300_000, triggerRatio: 0.8, thinking: "medium" }, native: { enabled: false }, }); assert.equal(config.enabled, false); - assert.equal(config.deterministic.enabled, false); - assert.equal(config.deterministic.maxStubTokens, 512); - assert.equal(config.semantic.enabled, false); - assert.equal(config.semantic.maxRetries, 0); + assert.deepEqual(config.masking, { + enabled: false, + triggerTokens: 150_000, + triggerRatio: 0.5, + targetTokens: 60_000, + targetRatio: 0.2, + minClearTokens: 30_000, + keepRecentTurns: 5, + minEntryTokens: 512, + excludeTools: ["bash", "read"], + }); + assert.equal(config.summary.triggerTokens, 300_000); + assert.equal(config.summary.triggerRatio, 0.8); + assert.equal(config.summary.thinking, "medium"); assert.equal(config.native.enabled, false); }); -test("invalid ratios and bounds fail safe to bounded defaults", () => { +test("invalid fields fail safe to bounded defaults", () => { const config = normalizeConfig({ - softRatio: 2, - hardRatio: -1, - targetRatio: 0.99, - semantic: { concurrency: 50, maxRetries: -1 }, + enabled: "yes", + masking: { + triggerTokens: -1, + triggerRatio: 2, + targetTokens: "a lot", + targetRatio: 0, + minClearTokens: 0, + keepRecentTurns: 500, + minEntryTokens: -3, + excludeTools: "read", + }, + summary: { triggerTokens: Number.NaN, triggerRatio: 1.5, thinking: "insane" }, + native: { enabled: 1 }, }); - assert.equal(config.softRatio, DEFAULT_CONFIG.softRatio); - assert.ok(config.targetRatio < config.softRatio); - assert.ok(config.hardRatio > config.softRatio); - assert.equal(config.semantic.concurrency, DEFAULT_CONFIG.semantic.concurrency); - assert.equal(config.semantic.maxRetries, DEFAULT_CONFIG.semantic.maxRetries); + assert.equal(config.enabled, DEFAULT_CONFIG.enabled); + assert.deepEqual(config.masking, DEFAULT_CONFIG.masking); + assert.equal(config.summary.triggerTokens, DEFAULT_CONFIG.summary.triggerTokens); + assert.equal(config.summary.triggerRatio, DEFAULT_CONFIG.summary.triggerRatio); + assert.equal(config.summary.thinking, DEFAULT_CONFIG.summary.thinking); + assert.equal(config.native.enabled, DEFAULT_CONFIG.native.enabled); +}); + +test("masking target is corrected below trigger and summary above masking", () => { + const config = normalizeConfig({ + masking: { triggerTokens: 80_000, triggerRatio: 0.5, targetTokens: 120_000, targetRatio: 0.9 }, + summary: { triggerTokens: 50_000, triggerRatio: 0.4 }, + }); + assert.equal(config.masking.targetTokens, 40_000); + assert.equal(config.masking.targetRatio, 0.25); + assert.equal(config.summary.triggerTokens, 160_000); + assert.ok(config.summary.triggerRatio > config.masking.triggerRatio); +}); + +test("minClearTokens and excludeTools accept valid overrides", () => { + const config = normalizeConfig({ + masking: { minClearTokens: 1, keepRecentTurns: 1, minEntryTokens: 0, excludeTools: [] }, + }); + assert.equal(config.masking.minClearTokens, 1); + assert.equal(config.masking.keepRecentTurns, 1); + assert.equal(config.masking.minEntryTokens, 0); + assert.deepEqual(config.masking.excludeTools, []); }); diff --git a/pi-compaction/test/extension-runtime.test.ts b/pi-compaction/test/extension-runtime.test.ts index 56589e5..1208d24 100644 --- a/pi-compaction/test/extension-runtime.test.ts +++ b/pi-compaction/test/extension-runtime.test.ts @@ -1,7 +1,8 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { performance } from "node:perf_hooks"; import piCompaction from "../index.ts"; +import { getCompactionService } from "../src/service.ts"; +import { MASK_PREFIX } from "../src/masking.ts"; function harness() { const handlers = new Map(); @@ -24,90 +25,253 @@ function harness() { return { pi, handlers, tools, commands, entries }; } -function context(branch: any[] = []) { - return { +function context(branch: any[] = [], tokens = 1_000) { + const compactCalls: any[] = []; + const sessionId = `session-${Math.random().toString(36).slice(2)}`; + const ctx = { sessionManager: { - getSessionId: () => "session-test", + getSessionId: () => sessionId, getBranch: () => branch, + getLeafId: () => "leaf-1", }, model: { provider: "test", api: "test", id: "model", contextWindow: 100_000 }, modelRegistry: {}, getSystemPrompt: () => "system", - getContextUsage: () => ({ tokens: 1_000, contextWindow: 100_000, percent: 1 }), + getContextUsage: () => ({ tokens, contextWindow: 100_000, percent: 1 }), + compact(options: unknown) { compactCalls.push(options); }, + isIdle: () => true, hasUI: false, abort() {}, } as any; + return { ctx, compactCalls }; } -test("one runtime projects tool evidence and recovers it through context_tree_query", async () => { +function projectedEntry(id: string, message: any) { + return { sourceEntry: { id, type: "message", message }, messages: [message] }; +} + +function assistantMessage(id: string, calls: Array<{ id: string; name: string; args?: unknown }>) { + return { + role: "assistant", + content: calls.map((call) => ({ type: "toolCall", id: call.id, name: call.name, arguments: call.args ?? {} })), + timestamp: id, + }; +} + +function toolResultMessage(id: string, toolCallId: string, text: string, toolName = "bash") { + return { + role: "toolResult", + toolCallId, + toolName, + content: [{ type: "text", text }], + timestamp: id, + }; +} + +// Default config: masking trigger = min(200K, 60% of window); at the harness +// 100K window that is 60K tokens and the target is 35K, so 60K used needs +// ~25K tokens of maskable results outside the 10-turn protected window. +function pressurizedContextEntries() { + const entries = [ + projectedEntry("a1", assistantMessage("a1", [{ id: "c1", name: "bash", args: { command: "npm test" } }])), + projectedEntry("r1", toolResultMessage("r1", "c1", `failure evidence\n${"x".repeat(120_000)}\nstack tail`)), + ]; + for (let index = 0; index < 10; index++) { + entries.push(projectedEntry(`keep-${index}`, assistantMessage(`keep-${index}`, []))); + } + return entries; +} + +test("turn_end masks old tool results under pressure and preserves prior drafts", async () => { const h = harness(); piCompaction(h.pi); - const ctx = context(); + const { ctx } = context([], 60_000); await h.handlers.get("session_start")![0]!({}, ctx); - const assistant = { - role: "assistant", - content: [{ type: "toolCall", id: "call-1", name: "bash", arguments: { command: "npm test" } }], - timestamp: 10, - }; - const result = { - role: "toolResult", - toolCallId: "call-1", - toolName: "bash", - content: [{ type: "text", text: `failure evidence\n${"x".repeat(12_000)}\nstack tail` }], - isError: true, - timestamp: 11, - }; - await h.handlers.get("turn_end")![0]!({ message: assistant, toolResults: [result] }, ctx); - const projected = await h.handlers.get("context")![0]!({ messages: [assistant, result] }, ctx); + const prior = { type: "custom", customType: "other-extension", data: {} }; + const result = await h.handlers.get("turn_end")![0]!({ + entries: [prior], + message: assistantMessage("a1", []), + toolResults: [], + context: { contextEntries: pressurizedContextEntries(), contextMessages: [] }, + }, ctx); - assert.match(projected.messages[1].content[0].text, /pi-compaction tool=bash/); - const cacheStart = performance.now(); - for (let iteration = 0; iteration < 1_000; iteration++) { - await h.handlers.get("context")![0]!({ messages: [assistant, result] }, ctx); - } - const cacheAverageMs = (performance.now() - cacheStart) / 1_000; - assert.ok(cacheAverageMs < 2, `full context handler cache-hit average ${cacheAverageMs.toFixed(3)}ms exceeded 2ms`); - assert.equal(h.entries.length, 1); - assert.equal(h.entries[0]!.customType, "pi-compaction"); - - const query = h.tools.get("context_tree_query"); - const recovered = await query.execute("q", { toolCallIds: ["call-1"] }, new AbortController().signal, undefined, ctx); - assert.match(recovered.content[0].text, /failure evidence/); + assert.ok(result?.entries); + assert.strictEqual(result.entries[0], prior); + const edits = result.entries.slice(1); + assert.equal(edits.length, 1); + assert.equal(edits[0]!.type, "context_edit"); + assert.equal(edits[0]!.targetId, "r1"); + assert.ok(edits[0]!.replacement.content.startsWith(MASK_PREFIX)); + // Masking is append-only context_edit; nothing is written as custom entries. + assert.equal(h.entries.length, 0); await h.handlers.get("session_shutdown")![0]!({}, ctx); }); -test("small tool results are passed through without a stub wrapper", async () => { +test("turn_end below the masking trigger returns undefined", async () => { const h = harness(); piCompaction(h.pi); - const ctx = context(); + const { ctx } = context([], 40_000); await h.handlers.get("session_start")![0]!({}, ctx); - const assistant = { - role: "assistant", - content: [{ type: "toolCall", id: "call-small", name: "bash", arguments: { command: "ls" } }], - timestamp: 20, - }; - const result = { - role: "toolResult", - toolCallId: "call-small", - toolName: "bash", - content: [{ type: "text", text: "ok\nfile-a.ts\nfile-b.ts" }], - timestamp: 21, - }; + const result = await h.handlers.get("turn_end")![0]!({ + entries: [], + message: assistantMessage("a1", []), + toolResults: [], + context: { contextEntries: pressurizedContextEntries(), contextMessages: [] }, + }, ctx); + assert.equal(result, undefined); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); - await h.handlers.get("turn_end")![0]!({ message: assistant, toolResults: [result] }, ctx); - const projected = await h.handlers.get("context")![0]!({ messages: [assistant, result] }, ctx); +test("context hook returns messages untouched on the non-checkpoint path", async () => { + const h = harness(); + piCompaction(h.pi); + const { ctx } = context(); + await h.handlers.get("session_start")![0]!({}, ctx); - assert.strictEqual(projected.messages[1].content[0].text, "ok\nfile-a.ts\nfile-b.ts"); - assert.doesNotMatch(projected.messages[1].content[0].text, /pi-compaction tool=/); + const messages = [assistantMessage("a1", []), toolResultMessage("r1", "c1", "visible\nfile-a.ts")]; + const projected = await h.handlers.get("context")![0]!({ messages }, ctx); + assert.equal(projected, undefined); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); + +test("agent_settled compacts once above the summary threshold and never repeats on the same leaf", async () => { + const h = harness(); + piCompaction(h.pi); + const { ctx, compactCalls } = context([], 95_000); + await h.handlers.get("session_start")![0]!({}, ctx); + + h.handlers.get("agent_settled")![0]!({ type: "agent_settled", aborted: false }, ctx); + assert.equal(compactCalls.length, 1); + assert.equal(typeof compactCalls[0].onError, "function"); + + h.handlers.get("agent_settled")![0]!({ type: "agent_settled", aborted: false }, ctx); + assert.equal(compactCalls.length, 1); + + ctx.sessionManager.getLeafId = () => "leaf-2"; + h.handlers.get("agent_settled")![0]!({ type: "agent_settled", aborted: false }, ctx); + assert.equal(compactCalls.length, 2); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); + +test("agent_settled stays idle below the summary threshold", async () => { + const h = harness(); + piCompaction(h.pi); + const { ctx, compactCalls } = context([], 50_000); + await h.handlers.get("session_start")![0]!({}, ctx); + + h.handlers.get("agent_settled")![0]!({ type: "agent_settled", aborted: false }, ctx); + assert.equal(compactCalls.length, 0); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); + +test("context_tree_query recovers the original tool output from the branch", async () => { + const h = harness(); + piCompaction(h.pi); + const branch = [ + { type: "message", id: "a1", parentId: null, timestamp: "t", message: assistantMessage("a1", [{ id: "c1", name: "bash", args: { command: "npm test" } }]) }, + { type: "message", id: "r1", parentId: "a1", timestamp: "t", message: toolResultMessage("r1", "c1", "failure evidence\nstack tail") }, + ]; + const { ctx } = context(branch); + await h.handlers.get("session_start")![0]!({}, ctx); - // Original evidence stays recoverable even when the message is passed through. const query = h.tools.get("context_tree_query")!; - const recovered = await query.execute("q", { toolCallIds: ["call-small"] }, new AbortController().signal, undefined, ctx); - assert.match(recovered.content[0].text, /file-a\.ts/); + const recovered = await query.execute("q", { toolCallIds: ["c1"] }, new AbortController().signal, undefined, ctx); + assert.match(recovered.content[0].text, /failure evidence/); + assert.match(recovered.content[0].text, /stack tail/); + assert.deepEqual(recovered.details.missing, []); + const missing = await query.execute("q", { toolCallIds: ["nope"] }, new AbortController().signal, undefined, ctx); + assert.match(missing.content[0].text, /not found/); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); + +test("service query returns placeholders only for masked tool call ids", async () => { + const h = harness(); + piCompaction(h.pi); + const { ctx } = context([], 60_000); + await h.handlers.get("session_start")![0]!({}, ctx); + const sessionId = ctx.sessionManager.getSessionId(); + + await h.handlers.get("turn_end")![0]!({ + entries: [], + message: assistantMessage("a1", []), + toolResults: [], + context: { contextEntries: pressurizedContextEntries(), contextMessages: [] }, + }, ctx); + + const records = getCompactionService()!.query(sessionId, ["c1", "never-masked"]); + assert.equal(records.length, 1); + assert.equal(records[0]!.toolCallId, "c1"); + assert.ok(records[0]!.deterministicSummary.startsWith(MASK_PREFIX)); + assert.equal(typeof records[0]!.resultTimestamp, "number"); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); + +test("hydrating a branch with this extension's context_edit rebuilds masking metrics", async () => { + const h = harness(); + piCompaction(h.pi); + const placeholder = `${MASK_PREFIX}bash result cleared to save context (~1_000 tokens). Call args: {}. Re-run the call if still needed, or recover the original with context_tree_query {"toolCallIds":["c1"]}.`; + const branch = [ + { type: "message", id: "a1", parentId: null, timestamp: "t", message: assistantMessage("a1", [{ id: "c1", name: "bash", args: { command: "npm test" } }]) }, + { type: "message", id: "r1", parentId: "a1", timestamp: "t", message: toolResultMessage("r1", "c1", "x".repeat(8_000)) }, + { type: "context_edit", id: "e1", parentId: "r1", timestamp: "t", targetId: "r1", replacement: { content: placeholder } }, + ]; + const { ctx } = context(branch); + await h.handlers.get("session_start")![0]!({}, ctx); + const sessionId = ctx.sessionManager.getSessionId(); + + await h.handlers.get("context")![0]!({ messages: [toolResultMessage("r1", "c1", "x".repeat(8_000))] }, ctx); + const metrics = getCompactionService()!.getMetrics(sessionId); + assert.equal(metrics!.maskedResults, 1); + assert.ok(metrics!.reclaimedTokens > 0); + assert.ok(metrics!.rawTokens > metrics!.effectiveTokens); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); + +test("session_before_compact no longer cancels threshold compactions", async () => { + const h = harness(); + piCompaction(h.pi); + const { ctx } = context(); + await h.handlers.get("session_start")![0]!({}, ctx); + ctx.modelRegistry.getApiKeyAndHeaders = async () => ({ ok: false, error: "no credentials" }); + + const result = await h.handlers.get("session_before_compact")![0]!({ + reason: "threshold", branchEntries: [], preparation: { firstKeptEntryId: "u1" }, + signal: new AbortController().signal, + }, ctx); + assert.equal(result, undefined); + await h.handlers.get("session_shutdown")![0]!({}, ctx); +}); + +test("native checkpoint projection applies context edits made after the checkpoint", async () => { + const h = harness(); + piCompaction(h.pi); + const details = { + kind: "pi-compaction", + version: 1, + checkpoint: "native", + modelKey: "test:test:model", + replacementHistory: [{ type: "compaction", encrypted_content: "opaque" }], + portableSummary: "checkpoint summary", + sourceFingerprint: "fp", + }; + const branch = [ + { type: "compaction", id: "c1", parentId: null, timestamp: "t", summary: "marker", firstKeptEntryId: "r1", tokensBefore: 1_000, details }, + { type: "message", id: "r1", parentId: "c1", timestamp: "t", message: toolResultMessage("r1", "c1", "ORIGINAL-CONTENT-HERE") }, + { type: "context_edit", id: "e1", parentId: "r1", timestamp: "t", targetId: "r1", replacement: { content: `${MASK_PREFIX}bash result cleared to save context (~5 tokens). Call args: {}.` } }, + { type: "message", id: "u1", parentId: "e1", timestamp: "t", message: { role: "user", content: "next question", timestamp: 30 } }, + ] as any[]; + const { ctx } = context(branch); + await h.handlers.get("session_start")![0]!({}, ctx); + + const projected = await h.handlers.get("context")![0]!({ messages: [] }, ctx); + const serialized = JSON.stringify(projected.messages); + assert.doesNotMatch(serialized, /ORIGINAL-CONTENT-HERE/); + assert.match(serialized, /result cleared to save context/); + assert.match(serialized, /next question/); await h.handlers.get("session_shutdown")![0]!({}, ctx); }); @@ -127,7 +291,7 @@ test("model switching after a native checkpoint projects the portable checkpoint { type: "compaction", id: "c1", parentId: null, timestamp: new Date().toISOString(), summary: "marker", firstKeptEntryId: "m0", tokensBefore: 1_000, details }, { type: "message", id: "m1", parentId: "c1", timestamp: new Date().toISOString(), message: { role: "user", content: "continue", timestamp: 20 } }, ] as any[]; - const ctx = context(branch); + const { ctx } = context(branch); await h.handlers.get("session_start")![0]!({}, ctx); const projected = await h.handlers.get("context")![0]!({ messages: [] }, ctx); assert.match(projected.messages[0].content[0].text, /decisions and files/); @@ -143,7 +307,7 @@ for (const provider of ["sub-api", "openai-codex"]) { type: "message", id: "u1", parentId: null, message: { role: "user", content: "Remember the test decision.", timestamp: 1 }, }]; - const ctx = context(branch); + const { ctx } = context(branch); ctx.model = { provider, api: provider === "sub-api" ? "openai-responses" : "openai-codex-responses", id: "selected-model", input: ["text"], contextWindow: 100_000, @@ -231,7 +395,7 @@ test("SubAPI native failures and cancellation return control to Pi without a che type: "message", id: "u1", parentId: null, message: { role: "user", content: "Test only.", timestamp: 1 }, }]; - const ctx = context(branch); + const { ctx } = context(branch); ctx.model = { provider: "sub-api", api: "openai-responses", id: "test-model", input: ["text"], baseUrl: "https://gateway.example/v1" }; ctx.modelRegistry.getApiKeyAndHeaders = async () => ({ ok: true, apiKey: "test-key" }); const responses = [ @@ -266,7 +430,7 @@ test("SubAPI native failures and cancellation return control to Pi without a che test("other providers and SubAPI Chat Completions keep their provider requests unchanged", async () => { const h = harness(); piCompaction(h.pi); - const ctx = context(); + const { ctx } = context(); try { for (const model of [ { provider: "other", api: "openai-responses", id: "test" }, @@ -302,7 +466,7 @@ test("native runtime delegates authentication to each provider without blocking type: "message", id: "u1", parentId: null, message: { role: "user", content: "Synthetic authentication test.", timestamp: 1 }, }]; - const ctx = context(branch); + const { ctx } = context(branch); const provider = scenario.provider ?? "sub-api"; ctx.model = { provider, api: provider === "sub-api" ? "openai-responses" : "openai-codex-responses", diff --git a/pi-compaction/test/masking.test.ts b/pi-compaction/test/masking.test.ts new file mode 100644 index 0000000..69731c5 --- /dev/null +++ b/pi-compaction/test/masking.test.ts @@ -0,0 +1,188 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { MASK_PREFIX, planMasking, rebuildMaskedLedger } from "../src/masking.ts"; +import { effectiveThreshold } from "../src/token-accounting.ts"; +import type { MaskingConfig } from "../src/types.ts"; + +const MASKING: MaskingConfig = { + enabled: true, + triggerTokens: 1_000, + triggerRatio: 0.9, + targetTokens: 400, + targetRatio: 0.35, + minClearTokens: 200, + keepRecentTurns: 2, + minEntryTokens: 10, + excludeTools: ["ask_user_question"], +}; + +const estimate = (message: any): number => Math.ceil(JSON.stringify(message).length / 4); + +function assistantMessage(id: string, calls: Array<{ id: string; name: string; args?: unknown }>) { + return { + role: "assistant", + content: calls.map((call) => ({ type: "toolCall", id: call.id, name: call.name, arguments: call.args ?? {} })), + timestamp: id, + }; +} + +function toolResultMessage(id: string, toolCallId: string, text: string, toolName = "bash") { + return { + role: "toolResult", + toolCallId, + toolName, + content: [{ type: "text", text }], + timestamp: id, + }; +} + +function entry(id: string, message: any, messages = [message]) { + return { sourceEntry: { id, type: "message", message }, messages }; +} + +function protect(turns = MASKING.keepRecentTurns) { + return Array.from({ length: turns }, (_unused, index) => + entry(`recent-${index}`, { role: "assistant", content: [{ type: "text", text: "ok" }], timestamp: `recent-${index}` })); +} + +test("below the effective trigger plans no drafts", () => { + const entries = [ + entry("e1", assistantMessage("a1", [{ id: "c1", name: "bash", args: { command: "ls" } }])), + entry("e2", toolResultMessage("r1", "c1", "x".repeat(8_000))), + ...protect(), + ]; + const plan = planMasking({ contextEntries: entries, currentTokens: 999, contextWindow: 0, masking: MASKING, estimate }); + assert.equal(plan.drafts.length, 0); + assert.equal(plan.masked.length, 0); +}); + +test("above the trigger masks oldest first until context fits under target", () => { + const entries = [ + entry("e1", assistantMessage("a1", [ + { id: "c1", name: "bash", args: { command: "one" } }, + { id: "c2", name: "bash", args: { command: "two" } }, + { id: "c3", name: "bash", args: { command: "three" } }, + ])), + entry("e2", toolResultMessage("r1", "c1", "a".repeat(4_000))), + entry("e3", toolResultMessage("r2", "c2", "b".repeat(4_000))), + entry("e4", toolResultMessage("r3", "c3", "c".repeat(4_000))), + entry("e5", toolResultMessage("r4", "c4", "recent result".repeat(200))), + ...protect(), + ]; + // needed = 2_000 - 400 = 1_600; each result clears ~980 tokens. + const plan = planMasking({ contextEntries: entries, currentTokens: 2_000, contextWindow: 0, masking: MASKING, estimate }); + assert.deepEqual(plan.drafts.map((draft) => draft.targetId), ["e2", "e3"]); +}); + +test("results inside the recent keepRecentTurns window are never masked", () => { + const entries = [ + entry("e1", toolResultMessage("r1", "c1", "a".repeat(4_000))), + entry("recent-a", assistantMessage("ra", [{ id: "rc", name: "bash", args: {} }])), + entry("recent-r", toolResultMessage("rr", "rc", "b".repeat(4_000))), + entry("recent-b", assistantMessage("rb", [])), + ]; + const plan = planMasking({ contextEntries: entries, currentTokens: 3_500, contextWindow: 0, masking: MASKING, estimate }); + assert.deepEqual(plan.drafts.map((draft) => draft.targetId), ["e1"]); +}); + +test("excluded tools, SKILL.md reads, already-masked and small results are skipped", () => { + const entries = [ + entry("e1", assistantMessage("a1", [ + { id: "c1", name: "ask_user_question", args: { question: "x" } }, + { id: "c2", name: "read", args: { path: "/repo/skills/foo/SKILL.md" } }, + { id: "c3", name: "bash", args: { command: "old" } }, + { id: "c4", name: "bash", args: { command: "tiny" } }, + ])), + entry("e2", toolResultMessage("r1", "c1", "a".repeat(4_000), "ask_user_question")), + entry("e3", toolResultMessage("r2", "c2", "b".repeat(4_000), "read")), + entry("e4", toolResultMessage("r3", "c3", `${MASK_PREFIX}bash result cleared to save context (~900 tokens).`), ), + entry("e5", toolResultMessage("r4", "c4", "tiny")), + entry("e6", toolResultMessage("r5", "c5", "d".repeat(4_000))), + ...protect(), + ]; + const masking = { ...MASKING, minClearTokens: 100, minEntryTokens: 100 }; + const plan = planMasking({ contextEntries: entries, currentTokens: 3_500, contextWindow: 0, masking, estimate }); + assert.deepEqual(plan.drafts.map((draft) => draft.targetId), ["e6"]); + assert.equal(plan.masked[0]!.toolCallId, "c5"); +}); + +test("when clearable volume stays under minClear nothing is masked", () => { + const entries = [ + entry("e1", toolResultMessage("r1", "c1", "a".repeat(600))), + ...protect(), + ]; + const plan = planMasking({ contextEntries: entries, currentTokens: 3_500, contextWindow: 0, masking: MASKING, estimate }); + // ~130 tokens clearable, below minClearTokens (200) → not worth invalidating cache. + assert.equal(plan.drafts.length, 0); +}); + +test("placeholder is deterministic, names the tool and call id, and truncates args at 300 chars", () => { + const longArgs = { command: `run ${"v".repeat(600)}` }; + const entries = [ + entry("e1", assistantMessage("a1", [{ id: "c1", name: "bash", args: longArgs }])), + entry("e2", toolResultMessage("r1", "c1", "a".repeat(4_000))), + ...protect(), + ]; + const masking = { ...MASKING, minClearTokens: 1 }; + const first = planMasking({ contextEntries: entries, currentTokens: 3_500, contextWindow: 0, masking, estimate }); + const second = planMasking({ contextEntries: entries, currentTokens: 3_500, contextWindow: 0, masking, estimate }); + assert.equal(first.drafts.length, 1); + const placeholder = first.masked[0]!.placeholder; + assert.equal(placeholder, second.masked[0]!.placeholder); + assert.ok(placeholder.startsWith(MASK_PREFIX)); + assert.match(placeholder, /bash result cleared/); + assert.match(placeholder, /"toolCallIds":\["c1"\]/); + const argsText = placeholder.match(/Call args: (.*)\. Re-run/)?.[1] ?? ""; + assert.ok(argsText.length <= 301, `args excerpt ${argsText.length} exceeded 300+ellipsis`); + assert.ok(placeholder.includes("…")); +}); + +test("the small-window ratio caps both trigger and target", () => { + const masking: MaskingConfig = { + ...MASKING, + triggerTokens: 200_000, + triggerRatio: 0.6, + targetTokens: 100_000, + targetRatio: 0.35, + minClearTokens: 50_000, + }; + assert.equal(effectiveThreshold(masking.triggerTokens, masking.triggerRatio, 272_000), 163_200); + assert.equal(effectiveThreshold(masking.targetTokens, masking.targetRatio, 272_000), 95_200); + const entries = [ + entry("e1", toolResultMessage("r1", "c1", "a".repeat(300_000))), + ...protect(), + ]; + const below = planMasking({ contextEntries: entries, currentTokens: 163_199, contextWindow: 272_000, masking, estimate }); + assert.equal(below.drafts.length, 0); + const above = planMasking({ contextEntries: entries, currentTokens: 163_200, contextWindow: 272_000, masking, estimate }); + assert.equal(above.drafts.length, 1); +}); + +test("ledger rebuild counts only masked targets still inside the active context", () => { + const placeholder = (id: string) => + `${MASK_PREFIX}bash result cleared to save context (~100 tokens). Call args: {}. Re-run the call if still needed, or recover the original with context_tree_query {"toolCallIds":["${id}"]}.`; + const branch = [ + { type: "message", id: "u1", parentId: null, timestamp: "t", message: { role: "user", content: "start", timestamp: 1 } }, + { type: "message", id: "r-old", parentId: "u1", timestamp: "t", message: toolResultMessage("m-old", "c-old", "o".repeat(2_000)) }, + { type: "context_edit", id: "e-old", parentId: "r-old", timestamp: "t", targetId: "r-old", replacement: { content: placeholder("c-old") } }, + { type: "message", id: "m-keep", parentId: "e-old", timestamp: "t", message: { role: "user", content: "kept question", timestamp: 2 } }, + { type: "compaction", id: "c1", parentId: "m-keep", timestamp: "t", summary: "summary", firstKeptEntryId: "m-keep", tokensBefore: 9_999 }, + { type: "message", id: "a-keep", parentId: "c1", timestamp: "t", message: assistantMessage("a-keep", []) }, + { type: "message", id: "r-new", parentId: "a-keep", timestamp: "t", message: toolResultMessage("m-new", "c-new", "n".repeat(2_000)) }, + { type: "context_edit", id: "e-new", parentId: "r-new", timestamp: "t", targetId: "r-new", replacement: { content: placeholder("c-new") } }, + ]; + const ledger = rebuildMaskedLedger(branch, "e-new", estimate); + assert.equal(ledger.size, 1); + assert.ok(ledger.has("c-new")); + assert.ok(!ledger.has("c-old")); +}); + +test("user and assistant messages are never masked", () => { + const entries = [ + entry("e1", { role: "user", content: "u".repeat(8_000), timestamp: "u1" }), + entry("e2", { role: "assistant", content: [{ type: "text", text: "a".repeat(8_000) }], timestamp: "a1" }), + ...protect(), + ]; + const plan = planMasking({ contextEntries: entries, currentTokens: 9_000, contextWindow: 0, masking: { ...MASKING, keepRecentTurns: 1, minClearTokens: 1 }, estimate }); + assert.equal(plan.drafts.length, 0); +}); diff --git a/pi-compaction/test/performance.test.ts b/pi-compaction/test/performance.test.ts index 0e2f38f..7694024 100644 --- a/pi-compaction/test/performance.test.ts +++ b/pi-compaction/test/performance.test.ts @@ -10,9 +10,9 @@ function message(index: number) { test("cached projection and append-only delta stay inside the hot-path budget", () => { const index = new SegmentIndex("perf"); - index.hydrate([], DEFAULT_CONFIG); + index.hydrate(); const messages = Array.from({ length: 10_000 }, (_, value) => message(value)); - const params = { environmentHash: "env", environmentTokens: 2_000, contextWindow: 300_000 }; + const params = { environmentHash: "env", environmentTokens: 2_000, contextWindow: 300_000, maskedResults: 0, reclaimedTokens: 0 }; index.project(messages, params); const cacheStart = performance.now(); @@ -24,6 +24,7 @@ test("cached projection and append-only delta stay inside the hot-path budget", index.project(appended, params); const deltaMs = performance.now() - deltaStart; + console.log(`cache-hit average ${cacheAverageMs.toFixed(3)}ms; delta projection ${deltaMs.toFixed(3)}ms`); assert.ok(cacheAverageMs < 2, `cache-hit average ${cacheAverageMs.toFixed(3)}ms exceeded 2ms`); assert.ok(deltaMs < 10, `delta projection ${deltaMs.toFixed(3)}ms exceeded 10ms`); }); diff --git a/pi-compaction/test/segment-index.test.ts b/pi-compaction/test/segment-index.test.ts index 099b76c..1facbb6 100644 --- a/pi-compaction/test/segment-index.test.ts +++ b/pi-compaction/test/segment-index.test.ts @@ -1,7 +1,6 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { DEFAULT_CONFIG } from "../src/config.ts"; -import { SegmentIndex, deterministicSummary } from "../src/segment-index.ts"; +import { SegmentIndex } from "../src/segment-index.ts"; const assistant = { role: "assistant", @@ -17,66 +16,96 @@ const result = { timestamp: 11, }; -test("deterministic summary is bounded and preserves head/tail evidence", () => { - const summary = deterministicSummary({ - toolName: "bash", - args: { command: "npm test" }, - rawText: result.content[0]!.text, - isError: true, - maxStubTokens: 256, +// Regression: a ~14.5K-char read result projected under a 1M window must reach +// the model byte-for-byte unchanged — masking only happens via context_edit +// drafts under real pressure, never on the hot projection path. +test("projection is byte-for-byte identity for a 300-line read result under a 1M window", () => { + const readBody = Array.from({ length: 300 }, (_unused, index) => `line ${index + 1}: ${"y".repeat(38)}`).join("\n"); + const readResult = { + role: "toolResult", + toolCallId: "call-read", + toolName: "read", + content: [{ type: "text", text: readBody }], + timestamp: 12, + }; + assert.ok(readBody.length > 14_000 && readBody.length < 15_000, `fixture ${readBody.length} chars`); + + const index = new SegmentIndex("session-1"); + index.hydrate(); + const snapshot = index.project([assistant, readResult], { + environmentHash: "env", + environmentTokens: 100, + contextWindow: 1_000_000, + maskedResults: 0, + reclaimedTokens: 0, }); - assert.match(summary, /tool=bash status=error/); - assert.match(summary, /failure/); - assert.match(summary, /stack tail/); - assert.ok(summary.length <= 1_100); + + assert.strictEqual(snapshot.messages[1], readResult); + assert.strictEqual(snapshot.messages[1]!.content[0].text, readBody); + assert.ok(snapshot.messages[1]!.content[0].text.includes("line 150:")); + assert.equal(snapshot.reclaimedTokens, 0); + assert.equal(snapshot.maskedResults, 0); }); -test("projection incrementally replaces tool results and caches identical input", () => { - const index = new SegmentIndex("session-1"); - index.hydrate([], DEFAULT_CONFIG); - const records = index.captureTurn(assistant, [result], DEFAULT_CONFIG); - assert.equal(records.length, 1); - +test("projection caches identical input and handles append-only deltas", () => { + const index = new SegmentIndex("session-2"); + index.hydrate(); const messages = [assistant, result]; - const first = index.project(messages, { - environmentHash: "env", - environmentTokens: 100, - contextWindow: 300_000, - }); - const second = index.project(messages, { - environmentHash: "env", - environmentTokens: 100, - contextWindow: 300_000, - }); - + const params = { environmentHash: "env", environmentTokens: 100, contextWindow: 300_000, maskedResults: 0, reclaimedTokens: 0 }; + const first = index.project(messages, params); + const second = index.project(messages, params); assert.strictEqual(first, second); - assert.ok(first.effectiveTokens < first.rawTokens); - assert.match(first.messages[1]!.content[0].text, /context_tree_query/); + assert.ok(first.rawTokens > 0); - const record = records[0]!; - assert.ok(index.applySemantic(record.sourceKey, record.sourceHash, "- npm test failed\n- stack tail")); - const semantic = index.project(messages, { + const appended = index.project([...messages, { role: "user", content: "next", timestamp: 13 }], params); + assert.notStrictEqual(appended, first); + assert.equal(appended.messages.length, 3); + assert.strictEqual(appended.messages[1], result); +}); + +test("changed masking stats publish a new snapshot without mutating the old one", () => { + const index = new SegmentIndex("session-stats"); + index.hydrate(); + const params = { environmentHash: "env", environmentTokens: 100, contextWindow: 300_000, - }); - assert.notStrictEqual(semantic, first); - assert.match(semantic.messages[1]!.content[0].text, /npm test failed/); + maskedResults: 0, + reclaimedTokens: 0, + }; + const first = index.project([assistant, result], params); + const same = index.project([assistant, result], params); + assert.strictEqual(same, first); + + const updated = index.project([assistant, result], { ...params, maskedResults: 1, reclaimedTokens: 500 }); + assert.notStrictEqual(updated, first); + assert.equal(updated.maskedResults, 1); + assert.equal(updated.reclaimedTokens, 500); + assert.equal(updated.rawTokens, updated.effectiveTokens + 500); + // The previously published object stays immutable. + assert.equal(first.maskedResults, 0); + assert.equal(first.reclaimedTokens, 0); + // Unchanged stats keep returning the same cached object. + assert.strictEqual(index.project([assistant, result], { ...params, maskedResults: 1, reclaimedTokens: 500 }), updated); }); test("usage-backed baseline is adjusted only by projected delta", () => { - const index = new SegmentIndex("session-2"); - index.hydrate([], DEFAULT_CONFIG); + const index = new SegmentIndex("session-3"); + index.hydrate(); const initial = index.project([assistant], { environmentHash: "env", environmentTokens: 50, contextWindow: 100_000, + maskedResults: 0, + reclaimedTokens: 0, }); index.recordUsage(1_000, initial.messageEstimate, "env"); const next = index.project([assistant, { role: "user", content: "next", timestamp: 12 }], { environmentHash: "env", environmentTokens: 50, contextWindow: 100_000, + maskedResults: 0, + reclaimedTokens: 0, }); assert.equal(next.confidence, "usage-backed"); assert.ok(next.effectiveTokens >= 1_000); diff --git a/pi-compaction/test/token-accounting.test.ts b/pi-compaction/test/token-accounting.test.ts index a23763d..702fa35 100644 --- a/pi-compaction/test/token-accounting.test.ts +++ b/pi-compaction/test/token-accounting.test.ts @@ -1,19 +1,17 @@ import test from "node:test"; import assert from "node:assert/strict"; -import { shouldCompact } from "../src/token-accounting.ts"; +import { effectiveThreshold } from "../src/token-accounting.ts"; -test("raw pressure cannot trigger compaction while effective lower-bound stays below 90%", () => { - const config = { hardRatio: 0.9 }; - const snapshot = { lowerBound: 231_000, contextWindow: 300_000 }; - assert.equal(shouldCompact("threshold", snapshot, config), false); - assert.equal(shouldCompact("manual", snapshot, config), true); - assert.equal(shouldCompact("overflow", snapshot, config), true); +test("absolute token threshold wins on large windows", () => { + assert.equal(effectiveThreshold(200_000, 0.6, 1_000_000), 200_000); }); -test("effective lower-bound at the configured threshold compacts", () => { - assert.equal(shouldCompact( - "threshold", - { lowerBound: 270_000, contextWindow: 300_000 }, - { hardRatio: 0.9 }, - ), true); +test("window ratio caps the threshold on small windows", () => { + assert.equal(effectiveThreshold(200_000, 0.6, 272_000), 163_200); + assert.equal(effectiveThreshold(100_000, 0.35, 272_000), 95_200); +}); + +test("a missing context window falls back to the absolute threshold", () => { + assert.equal(effectiveThreshold(200_000, 0.6, 0), 200_000); + assert.equal(effectiveThreshold(200_000, 0.6, -1), 200_000); });