diff --git a/docs/architecture/external-thread-queue.md b/docs/architecture/external-thread-queue.md index e40cf750fb..0556f0161c 100644 --- a/docs/architecture/external-thread-queue.md +++ b/docs/architecture/external-thread-queue.md @@ -20,6 +20,7 @@ Desktop 在 Turn 运行中提交的跟进消息，经官方 app-server 的 `thre - Turn 正常完成（`completed`）后，Host 自动开始队首提交，对应官方服务端队列的自动推进。若入队请求晚于完成事件到达，Host 在回复入队后补查：仅当前进程已投影的最近一次 Turn 正常完成、Thread 空闲且没有待处理命令时推进；恢复出的历史不作为自动执行依据。 - Turn 被中断或失败时不自动推进；Desktop 显示“队列已暂停”，用户点击“继续”时发送 `thread/queue/start`。 - 运行中点击“引导”时，Desktop 先提交引导，再删除接受的队列消息；Host 按 `clientUserMessageId` 和规范化文本预留该条目，失败恢复原位置，成功则为稍后的 `thread/queue/delete` 保留删除回执（Host 内合计最多 128 条）。即使替换 Turn 在 Desktop 删除前快速完成，自动推进也不会重复执行这条输入。重复身份或输入变化在停止／发送前拒绝，不按文本单独删除其他消息。 +- 原生引导调用抛错时，输入可能已产生副作用；恢复的条目保持可见，但不再由 Host 自动推进，即使编辑、重排或之后另一个 Turn 正常完成也不解除该限制。队首为此类条目时不自动跳过它执行后项。用户检查影响后可明确选择手动发送或删除；普通拒绝仍保留原有恢复语义。该内部限制不增加官方队列响应字段。 - 主进程 Manager 的新轮展示、请求标记和暂停结算见[外部 Thread 的「调整方向」](external-thread-steering.md#主进程-manager-的原生-rpc-路径)。存在其他中断暂停消息时，提交前要求先处理，不能用 resume-all 顺带恢复它。 - 开始失败时，提交保留在原位置；引导预留失败时恢复原位置并再次发出 `thread/queue/changed`。失败、中断及结果未知时暂停队列，不自动重发。原生引导交付结果未知的条目还按稳定 ID 禁止自动启动；恢复队列、编辑、重排或明确发送另一条消息都不解除此限制。该条目仍可查看、删除或由用户明确选择发送；Harness 明确拒绝、保证未接纳的输入不受此限制。 - Desktop 只为自己 `turn/start` 的输入做本地乐观展示。由队列开始的 Turn 由 Host 投影带 `clientId` 的 userMessage，否则该轮用户消息不显示。 diff --git a/docs/architecture/external-thread-steering.md b/docs/architecture/external-thread-steering.md index b30d88fd4a..a0000a72b6 100644 --- a/docs/architecture/external-thread-steering.md +++ b/docs/architecture/external-thread-steering.md @@ -3,9 +3,9 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当前 Session 的能力选择送达方式： - **原生同轮插入**：Session 此刻声明 `capabilities.steer` 时，Host 发送公共命令 `turn.steer`，由 Adapter 调用 Harness 自己的插入原语，在原生安全边界把输入送入**当前 Turn**，不取消、不另起 Turn。 -- **停止后重发（退路）**：其余情况下，Host **取消当前 Turn，等待它终结，再自动执行本次新输入**。任何不确定能并入当前 Turn 的情况都走这条路径，保证不比没有原生插入时差。 +- **停止后重发（退路）**：其余情况下，Host **取消当前 Turn，等待它终结，再自动执行本次新输入**。写入前不能并入当前 Turn 或 Harness 明确未接纳时走这条路径；写入后交付结果不确定则报错，不自动重发。 -用户继续使用 Codex Desktop 原有按钮、跟进处理方式设置和单条消息反向操作，不需要手动停止后重发。官方 Codex Thread 仍透传原生 `turn/steer`，不改变同轮追加输入的语义。当前只有 Claude Code 会声明 `capabilities.steer`，并且只在确定插入会并入当前 Turn 时声明；其条件与原生依据见 [`../harnesses/claude-code/claude-code-steer.md`](../harnesses/claude-code/claude-code-steer.md)；其他 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。 +用户继续使用 Codex Desktop 原有按钮、跟进处理方式设置和单条消息反向操作，不需要手动停止后重发。官方 Codex Thread 仍透传原生 `turn/steer`，不改变同轮追加输入的语义。Claude Code、Pi 和 OMP 按各自原生执行窗口动态声明 `capabilities.steer`；具体条件见 [Claude](../harnesses/claude-code/claude-code-steer.md)、[Pi](../harnesses/pi/pi-steer.md)、[OMP](../harnesses/omp/omp-steer.md)。其余 Adapter 对 `turn.steer` 返回 `unsupported`，保持停止后重发。 ## 原生同轮插入 @@ -17,6 +17,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当 1. `turn/steer` 带有 `codexhostSteerDelivery`（Renderer 已经按该方式展示）时照它执行：`newTurn` 停止后重发；`activeTurn` 原生插入，Harness 没收下时报错、不改走退路，避免和 Desktop 已有的展示不一致。 2. 输入是 Harness 命令（Session 提供命令、文本以 `/命令` 开头）时停止后重发，命令照常解析和执行。 3. 其余情况读取实时的 `capabilities.steer`：为 `true` 时原生插入，Harness 没收下就改走停止后重发；如果当前 Turn 在此期间已经结束，直接把输入作为下一个 Turn 启动。 +- 原生调用抛错表示交付结果不确定，不触发停止后重发。Host 保留有界的失败回执，防止同一消息身份被自动重试；若输入来自队列，恢复原位置并禁止 Host 自动推进它，包括后续其他 Turn 完成之后。输入钩子的副作用不能假定已撤销；用户应检查影响后再决定手动发送或删除，见[队列执行语义](external-thread-queue.md#执行语义)。 - `turn/steer` 回执包含 `turnId` 和 `delivery`：`activeTurn` 表示送入当前 Turn，`newTurn` 表示停止旧 Turn 后开始了新 Turn。 - `codexhost/thread/steering/inspect { threadId, text? }` 按同样的规则（第 2、3 步）预先给出送达方式；读取失败返回 `-32081`。 @@ -24,7 +25,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当 ### 后续 Adapter 接入 -已有原生同轮插入能力的 Harness，只需在自己的 Adapter 中声明能力、处理 `turn.steer`、保证写入前校验与失败语义，并适配本 Harness 的原生历史读取。Host、Broker 和 Renderer 不按 Harness ID 增加分支。公共输入目前仅支持文本；非同轮的排队原语不能声明为此能力。真实原生执行、取消和历史行为仍需由各 Adapter 验证。 +已有原生同轮插入能力的 Harness，只需在自己的 Adapter 中声明能力、处理 `turn.steer`、保证写入前校验与失败语义，并适配本 Harness 的原生历史读取。Host、Broker 和 Renderer 不按 Harness ID 增加分支。公共输入目前仅支持文本；只有原生控制流能证明输入会进入尚未结束的当前 Turn 时，才可声明此能力；单纯 queued／入队回执不足。真实原生执行、取消和历史行为仍需由各 Adapter 验证。 ### Fork 与撤销窗口 @@ -45,7 +46,7 @@ codexhost 对外部 Harness Thread 保留一个「调整方向」操作，按当 - 取消 acknowledgement 不等于完成。等待旧 Turn 的 Interaction/Item 关闭、终态身份持久化及 Desktop 终态通知写出后，复用普通 `turn/start` 的底层启动函数，分配真实的新 Host Turn ID，返回 `{turnId}`。 - 同 Thread 的其他 start、Harness command、委派 start 不能抢占这个替换过程。若原生自主 Turn 已先开始，替换失败，不取消这个意外的新 Turn。 - 取消与旧轮终态等待合计最多 20 秒，短于当前 Desktop 的 30 秒提交超时。超时、Session fault、输出流结束、Host 关闭/断连、显式停止替换、旧轮失败或持久化失败均不再自动启动新轮。新轮启动本身仍遵循原有 Adapter admission 语义；客户端超时不意味着它没有被接受。 -- 同一连接内以 `threadId + clientUserMessageId` 合并相同请求，并保留有界的成功交付回执供 outcome-unknown 重试使用；同一消息 ID 不得携带不同的旧轮 ID 或输入。不承诺跨 Host 重启或回执淘汰后的 exactly-once。 +- 同一连接内以 `threadId + clientUserMessageId` 合并相同请求，并保留有界的成功回执与结果不确定的失败回执；同一消息 ID 不得携带不同的旧轮 ID 或输入。不承诺跨 Host 重启或回执淘汰后的 exactly-once。 普通停止仍调用 Adapter 原来的取消实现。原生进程、工具和子任务的退出速度及恢复策略属于 Adapter；本功能没有增加统一的强杀策略，也不会回滚旧轮已经完成的文件修改。 @@ -124,7 +125,7 @@ Shared Host 是外部 Thread 队列的唯一所有者，并实现 Codex Desktop - `packages/renderer-extension/test/renderer-external-steering.test.ts`（`steer delivery handshake`）：inspect 带文本、两种展示方式分别标记 `codexhostSteerDelivery`、Host 不支持 inspect 时改用 Thread 所有权。 - `packages/protocol-core/test/codex-ui-projector.test.ts`：插入的 `userMessage` 在实时与历史投影中留在所属 Turn。 - `packages/adapters/claude-code/test/*.test.ts`：Claude Code 的 `priority: "next"` 推入、接受条件、错过 Turn 与历史分轮。 -- `packages/host-runtime/test/external-turn-steering.test.ts`：原生插入不取消、失败释放回执、没收下时退回（当前 Turn 已结束则直接启动下一个、正在结束则等待其终态），以及延迟/同步终态、去重、冲突、过期目标、输入预检、超时迟到、fault、关闭、显式停止、自主 Turn 及启动失败。 +- `packages/host-runtime/test/external-turn-steering.test.ts`：原生插入不取消、明确拒绝释放回执、结果不确定保留失败回执与禁止自动重试、没收下时退回（当前 Turn 已结束则直接启动下一个、正在结束则等待其终态），以及延迟/同步终态、去重、冲突、过期目标、输入预检、超时迟到、fault、关闭、显式停止、自主 Turn 及启动失败。 - `packages/host-runtime/test/app-server-host*.test.ts`：真实 Host 路由、外部不泄漏官方流、取消与启动顺序、响应 gate、同 Thread start 竞争、官方透传。 - `packages/renderer-extension/test/renderer-external-steering.test.ts`：正常 start 占位、唯一输入、无旧轮 steer Item、去重、队列暂停恢复、owner/follower、失败和卸载。 - `packages/renderer-extension/test/renderer-external-steering-rpc.test.ts`：类方法的 RPC 可访问性、模型／权限列表读取、官方与外部 steer、实例隔离、卸载和重复安装，以及卸载期间待定请求的清理。 @@ -137,3 +138,13 @@ Shared Host 是外部 Thread 队列的唯一所有者，并实现 Codex Desktop 针对 RpcTarget 实例属性回归，还在运行中的上述版本 Desktop 内通过其真实 `RpcStub` 调用了 `model/list` 和 `permissionProfile/list`，验证原型挂接修复后两者成功返回，并重试了此前失败的只读查询。该检查没有发送新消息、修改权限选择或验证原生 Harness 执行。 以上合成测试和只读 RPC 检查不替代真实 Desktop 与各原生 Harness 验收。发布验收应覆盖流式文本、工具运行、Question/Approval、连续提交、已有队列、取消失败、刷新历史，以及多窗口和远端连接。不得把类型检查或合成测试通过称为所有 Harness 的实机兼容性证明。 + +### Pi / OMP Adapter 实机 Gate + +`tools/gate-steer/adapter.real.test.mjs` 默认跳过。先运行 `npm run typecheck`，再用 `CODEXHOST_STEER_REAL_HARNESSES=pi,omp npx vitest run --config tests/vitest.config.js tools/gate-steer/adapter.real.test.mjs` 选择实际安装的 Harness；`CODEXHOST_STEER_REAL_MODEL` 可指定原生目录中的完整模型显示名。Gate 不修改模型配置，但会使用现有登录和模型额度，留下原生 Session。 + +Pi 的异步 input hook 跨越原执行终态时，恢复与队列隔离由 `packages/adapters/pi/test/pi-steer-delayed-input.real.test.ts` 验证；同样用 `CODEXHOST_STEER_REAL_HARNESSES=pi` 显式运行。该测试还恢复同一个 Native Session 并发送下一轮，确认晚到输入没有被重放；它清理自建 Session，不操作已有历史。 + +默认执行 `single`、`double`、`near-terminal` 三种场景，也可用 `CODEXHOST_STEER_REAL_SCENARIOS` 选择。测试只要求在临时目录执行前台 sleep 命令，并检查单次/连续插入、原工具完成、错误目标/空文本/终态后的拒绝及原生历史身份。工具结束附近的场景继续观察 30 秒：接纳的输入必须在目标 Host Turn 回复一次；明确拒绝的输入不能执行。原生 Session 和记录保留在 `.codexhost/native-steer-real//`，临时工作目录清理。未安装、缺凭据或超时都不能算作通过，且该 Gate 不验证 Desktop GUI。 + +macOS / Desktop 26.1103.31037 的独立 GUI 验收通过真实 Composer 的「排队」「引导」与删除按钮执行。Pi 覆盖工具中正常插话、异步 hook 跨越终态、原轮成功、同一 Native Session 恢复、未确认队列不自动重放，以及删除后普通发送。OMP 在 `openai-codex / gpt-5.6-sol`、High 配置下验证实际 shell 工具完成、一个 Host Turn 内接纳与回复；其他 Model / Provider 组合不能由此推定通过。以上不覆盖多窗口、远端和其他 OS。 diff --git a/docs/harnesses/omp/omp-steer.md b/docs/harnesses/omp/omp-steer.md new file mode 100644 index 0000000000..8a6b065c95 --- /dev/null +++ b/docs/harnesses/omp/omp-steer.md @@ -0,0 +1,17 @@ +# OMP 原生插话 + +RPC `{type:"steer", message}` 将输入交给原生 steering 队列，工具批次结束后、下一次模型请求前注入。原生接口及工具边界见 [OMP RPC](https://github.com/can1357/oh-my-pi/blob/main/docs/rpc.md) 与 [agent loop 契约](https://github.com/can1357/oh-my-pi/blob/main/packages/agent/src/types.ts)。Host 的 `accepted: true` 承诺接纳到当前 Host Turn，不能仅凭一个成功入队回执在所有阶段声明能力。 + +Session 只在当前请求的 Turn 有根级 Tool 未结束、且没有开始取消或终态结算时声明 `capabilities.steer=true`。`turn.steer` 在调用 RPC 前重新检查同一窗口和目标 Turn ID；没有跨 await 的预检查缓存。文本输出、空闲和终态阶段不声明，Host 沿既有停止后重发路径处理。Adapter 不取消工具、不自行开始替代 Turn，也不重复发布 Host 已负责投影的用户输入。 + +原生明确返回相关 `steer` 请求的 `success:false` 时，Adapter 才返回未接纳错误，允许 Host 使用原有退路。传输断开、回执超时或输入钩子已经处理但未确认入队等未知结果继续抛出，不能包装成未接纳，防止已消费的输入被重发。取消期间不接纳新的 steer。 + +## 历史与完成 + +现有历史解析按原生用户条目分轮，实时同轮插入可能在刷新后表现为后续用户条目；本补丁不另存一份分组记录。Turn 开始终态身份结算后关闭新输入窗口，并等待已发出的 RPC 结算，再读取历史。新用户条目数必须等于原提问加明确接纳的插入数，原提问仍是该 Host Turn 的身份；数量不符继续报错，不能悄悄改绑到插入消息。 + +## 验证边界 + +定向测试覆盖根工具窗口的开闭、取消期间拒绝、明确拒绝与未知回执的区别、连续输入、终态与回执竞态、原生历史数量及用户消息投影归属。真实 Gate 位于 `tools/gate-steer/adapter.real.test.mjs`，默认跳过；用 `CODEXHOST_STEER_REAL_HARNESSES=omp` 显式选择。它检查单次、连续两次和工具结束附近的输入；被接纳的输入必须在原 Host Turn 恰好回复一次，不能以排到下一轮冒充通过。Gate 使用已有登录和所选模型，留下原生测试 Session，并清理临时工作目录。 + +macOS / OMP 18.12.6 已运行单次、连续两次和工具结束附近的真实 Gate。Desktop 26.1103.31037 的隔离窗口使用既有 `openai-codex / gpt-5.6-sol`、High 配置，在前台 sleep 命令执行时通过 Composer「排队」「引导」完成同轮插话；原工具成功完成，只有一个 Host Turn，回复遵循补充要求。验证必须识别实际 shell 命令，不能把 OMP 投影到同一 commandExecution 通道的 thinking 工具当作该命令。该结果不覆盖所有 Model / Provider 组合、延迟 input hook、多窗口、远端或其他 OS。 diff --git a/docs/harnesses/pi/pi-steer.md b/docs/harnesses/pi/pi-steer.md new file mode 100644 index 0000000000..9b3aa53d72 --- /dev/null +++ b/docs/harnesses/pi/pi-steer.md @@ -0,0 +1,21 @@ +# Pi 原生插话 + +RPC `{type:"steer", message}` 将输入交给原生 steering 队列，工具批次结束后、下一次模型请求前注入。原生接口及工具边界见 [Pi RPC Commands](https://github.com/earendil-works/pi/blob/main/packages/coding-agent/docs/rpc-commands.md)。Host 的 `accepted: true` 承诺接纳到当前 Host Turn，不能仅凭一个成功入队回执在所有阶段声明能力。 + +Session 只在当前请求的 Turn 有根级 Tool 未结束、且没有开始取消或终态结算时声明 `capabilities.steer=true`。`turn.steer` 在调用 RPC 前重新检查同一窗口和目标 Turn ID；Transport 在收到原生成功回执的那个事件帧再次确认目标执行尚未终结，不使用 await 恢复后的状态替代原生帧顺序。文本输出、空闲和终态阶段不声明，Host 沿既有停止后重发路径处理。正常插话不取消工具、不自行开始替代 Turn，也不重复发布 Host 已负责投影的用户输入。 + +写入前已确认目标不再接纳，或原生明确返回相关 `steer` 请求的 `success:false` 时，Adapter 返回未接纳错误，允许 Host 使用原有退路。传输断开、回执超时或输入钩子已经处理但未确认入队等未知结果继续抛出，不能包装成未接纳，防止已消费的输入被重发。取消期间不接纳新的 steer。 + +Pi 的异步 input hook 可以在原执行结束之后才完成；之后的 `queued` 只表示进入空闲 Session 的队列，不能确认送达原 Turn。此时 Transport 先读取原生终态历史，再关闭该 Thread 专属的 Pi 进程，防止未执行队列混入下一次发送。插话返回结果不确定错误，不自动重发，也不声称能撤销 input hook 已产生的副作用。来自 Desktop 队列的输入恢复原位置，但不会因原轮或后续 Turn 完成而由 Host 自动执行；用户检查影响后再决定手动发送或删除。Adapter 用保留的原生历史结算旧 Turn，不把该插话计为已接纳；旧 Turn 的真实终态发布后，Session 才报告故障，由现有恢复路径重新打开同一个 Native Session。不编辑或清空共享原生队列，不重启其他 Thread。 + +## 历史与完成 + +现有历史解析按原生用户条目分轮，实时同轮插入可能在刷新后表现为后续用户条目；本补丁不另存一份分组记录。Turn 开始终态身份结算后关闭新输入窗口，并等待已发出的 RPC 结算，再读取历史。新用户条目数必须等于原提问加明确接纳的插入数，原提问仍是该 Host Turn 的身份；数量不符继续报错，不能悄悄改绑到插入消息。 + +## 验证边界 + +定向测试覆盖根工具窗口的开闭、取消期间拒绝、明确拒绝与未知回执的区别、连续输入、终态与回执竞态、原生历史数量及用户消息投影归属。真实 Gate 位于 `tools/gate-steer/adapter.real.test.mjs`，默认跳过；用 `CODEXHOST_STEER_REAL_HARNESSES=pi` 显式选择。它检查单次、连续两次和工具结束附近的输入；被接纳的输入必须在原 Host Turn 恰好回复一次，不能以排到下一轮冒充通过。Gate 使用已有登录和所选模型，留下原生测试 Session，并清理临时工作目录。 + +`packages/adapters/pi/test/pi-steer-delayed-input.real.test.ts` 使用同一开关，在真实 CLI 中额外加载临时 input hook，直到原执行的 `agent_settled` 之后才释放输入。检查旧 Turn 保持成功、晚到回执不报接纳、专属进程已退出、同一 Native Session 可以恢复，以及下一轮不会执行遗留输入；只删除测试自建目录和 Session，报告保存在 `.codexhost/native-steer-real/pi-delayed-input/`。 + +原生依据为随包 `docs/rpc-commands.md`、`dist/core/agent-session.js`、`node_modules/@earendil-works/pi-agent-core/dist/agent-loop.js`。macOS / Pi 1.4.0 已实测正常单次和连续插话，以及延迟 input hook 后恢复同一 Session 且不重放输入；单次实测不能证明所有临界时序或自定义扩展行为。macOS / Desktop 26.1103.31037 的隔离窗口实测覆盖正常工具中插话、异步 hook 晚到、原轮成功、同一 Native Session 后续输入不重放旧队列，以及删除未确认输入后普通发送。操作通过真实 Composer 的「排队」「引导」与删除按钮完成。多窗口、远端和其他 OS 另行验收。 diff --git a/docs/index.md b/docs/index.md index 9b2e0255f2..9e0f9b8000 100644 --- a/docs/index.md +++ b/docs/index.md @@ -98,12 +98,14 @@ | [`harnesses/pi/pi-subagents.md`](harnesses/pi/pi-subagents.md) | Pi 的 gotgenes / nicobailon 子代理身份、实时状态、只读子会话与适配边界。 | | [`harnesses/pi/pi-fast.md`](harnesses/pi/pi-fast.md) | Pi Codex Fast 闪电开关、别名 Provider 能力判定、自动加载和 priority 请求边界。 | | [`harnesses/pi/pi-progress.md`](harnesses/pi/pi-progress.md) | Pi 可见通知、重试状态、流式完成态核对及历史恢复边界。 | +| [`harnesses/pi/pi-steer.md`](harnesses/pi/pi-steer.md) | Pi 原生插话窗口、拒绝与未知回执、历史和实机 Gate。 | ### OMP | 文档 | 内容与阅读时机 | | --- | --- | | [`harnesses/omp/omp-interactions.md`](harnesses/omp/omp-interactions.md) | OMP 原生提问、审批、选项说明和超时语义；维护 OMP 交互时阅读。 | +| [`harnesses/omp/omp-steer.md`](harnesses/omp/omp-steer.md) | OMP 原生插话窗口、拒绝与未知回执、历史和实机 Gate。 | ### Grok diff --git a/openspec/specs/harness-adapter-text-session/spec.md b/openspec/specs/harness-adapter-text-session/spec.md index a47b336f57..3e8e0e294a 100644 --- a/openspec/specs/harness-adapter-text-session/spec.md +++ b/openspec/specs/harness-adapter-text-session/spec.md @@ -185,3 +185,45 @@ A concrete Adapter SHALL publish a Native Session Ref only when the native Harne - **WHEN** 具体 Harness 不提供可靠的 Usage Telemetry - **THEN** 其 Session MUST 将 `initialUsage` 暴露为 `null`，并且不发出 Usage 快照 - **AND** 所有现有文本、cancel、state、fault 和 close 行为 MUST 保持不变 + +### Requirement: Pi and OMP admit native steering only into the current Turn + +Pi and OMP SHALL advertise native steering only while their current Turn has an unfinished root Tool and is not cancelling, compacting, or settling. They SHALL recheck the target and admission window before submission. Acceptance SHALL confirm current-Turn delivery, not merely native queue admission. Completed history SHALL bind the original prompt and account for accepted steers without manufacturing a new Host Turn. + +#### Scenario: Input is accepted at a Tool boundary + +- **WHEN** steering is accepted while an unfinished root Tool keeps the current Turn open +- **THEN** the Tool continues without cancellation and the input belongs to that Host Turn +- **AND** history settlement waits for in-flight steering receipts and counts each accepted input once + +#### Scenario: The current Turn cannot accept input + +- **WHEN** the target changed, the admission window closed before writing, or the native Harness explicitly refused the input +- **THEN** the Adapter reports that the input was not taken +- **AND** the Host may use its existing stop-and-start path unless the client already selected active-Turn delivery + +### Requirement: Late Pi queue admission preserves the completed Turn + +If Pi confirms queue admission only after the target execution ended, the Adapter SHALL report an unknown delivery outcome, retain the original Turn's native history and outcome, and close only that Thread's Pi process before recovery. It SHALL NOT count the input as accepted, replay it, or claim to undo input-hook side effects. The same Native Session SHALL remain recoverable for later explicit input. + +#### Scenario: An asynchronous input hook outlives the execution + +- **WHEN** the original execution settles before the input hook produces a queue receipt +- **THEN** the original Turn completes from its native history before the Session faults +- **AND** recovery does not run the leftover queued input as a later Turn + +### Requirement: Unknown steering outcomes cannot trigger automatic replay + +The Host SHALL retain unknown steering outcomes in its bounded message-identity receipts, so a retry with the same Thread, identity, target and input returns that outcome without resubmission. Explicitly rejected input MAY release its identity for retry. Queued input with an unknown outcome SHALL remain visible but ineligible for automatic advancement, including after another queued input is sent explicitly; the user may explicitly send or remove it. + +#### Scenario: A steering receipt is lost + +- **WHEN** a native steering call may have accepted input but its outcome is unknown +- **THEN** retrying the same message identity does not invoke the native insertion or a replacement Turn again +- **AND** the Host does not claim exactly-once delivery across receipt eviction or Host restart + +#### Scenario: A later explicit queue send resumes the queue + +- **WHEN** input A has an unknown steering outcome and the user explicitly starts queued input B +- **THEN** completing B does not automatically start A +- **AND** A can still be sent or removed explicitly by the user diff --git a/packages/adapters/omp/src/omp-adapter.ts b/packages/adapters/omp/src/omp-adapter.ts index a6b3af996c..d1f5f76caa 100644 --- a/packages/adapters/omp/src/omp-adapter.ts +++ b/packages/adapters/omp/src/omp-adapter.ts @@ -68,7 +68,6 @@ import { type TurnStartCommand, type TurnSteerAccepted, type TurnSteerCommand, - unsupportedSteer, } from "@codexhost/harness-adapter"; import { harnessCommandCatalogSchema, @@ -104,6 +103,7 @@ import { rollbackOmpLastTurn } from "./omp-last-turn-rollback.js"; import { OmpRpcFaultError, OmpRpcSession, + OmpRpcSteerRejectedError, OmpRpcUnsupportedCommandError, type OmpInteractionRequest, type OmpInteractionResponse, @@ -165,6 +165,7 @@ export interface OmpTurnTransport { onEvent: (event: OmpTurnEvent) => void, ): Promise; runTurn(text: string, onEvent: (event: OmpTurnEvent) => void): Promise; + steer(message: string): Promise; respondToInteraction(response: OmpInteractionResponse): Promise; abort(): Promise; close(): Promise; @@ -217,6 +218,9 @@ interface ActiveTurn { agentMessageId: string | null; compactionItem: HostContextCompactionItem | null; sawAssistantMessage: boolean; + acceptedSteerCount: number; + acceptingSteers: boolean; + pendingSteers: Set>; reasoningItem: HostReasoningItem | null; tools: Map; interactions: Map; @@ -723,16 +727,19 @@ class OmpHarnessSession implements HarnessSession { this.#toolOutputLimit = options.toolOutputLimit; this.#permissionMode = options.permissionMode; this.#permissionModeId = options.permissionModeId; - this.capabilities = { + this.capabilities = ((session) => ({ configuration: { selectModel: true, selectThinkingOption: options.supportsThinkingSelection, selectPermissionMode: true, - permissionModeScope: "live", + permissionModeScope: "live" as const, }, history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true }, subagents: { observe: true, readTranscript: true }, - }; + get steer(): boolean { + return session.#canSteerActiveTurn(); + }, + }))(this); this.commands = { list: async () => ({ ok: true, value: this.#liveCommandCatalog() }), execute: (command) => this.#executeHarnessCommand(command), @@ -921,7 +928,7 @@ class OmpHarnessSession implements HarnessSession { return { ok: false, error: invalidState("Omp Session is not open") }; } if (command.type === "turn.cancel") return this.#cancel(command); - if (command.type === "turn.steer") return unsupportedSteer("OMP"); + if (command.type === "turn.steer") return this.#steer(command); if (command.type === "interaction.respond") return this.#respond(command); if (command.type === "model.select") return this.#selectModel(command); if (command.type === "thinking.select") return this.#selectThinking(command); @@ -995,6 +1002,9 @@ class OmpHarnessSession implements HarnessSession { agentMessageId: null, compactionItem: null, sawAssistantMessage: false, + acceptedSteerCount: 0, + acceptingSteers: true, + pendingSteers: new Set(), reasoningItem: null, tools: new Map(), interactions: new Map(), @@ -1395,6 +1405,59 @@ class OmpHarnessSession implements HarnessSession { } } + #canSteerActiveTurn(): boolean { + const active = this.#active; + return Boolean( + this.#phase === "open" && + active && + !active.cancellationRequested && + active.acceptingSteers && + active.tools.size > 0 && + this.#transport, + ); + } + + async #steer(command: TurnSteerCommand): Promise> { + const text = command.input.map((input) => input.text).join("\n"); + if (text.length === 0) { + return { + ok: false, + error: { + code: "invalidRequest", + message: "Omp steer text must not be empty", + retryable: false, + }, + }; + } + const active = this.#active; + const transport = this.#transport; + if ( + !active || + !transport || + active.command.turnId !== command.turnId || + !this.#canSteerActiveTurn() + ) { + return { ok: false, error: invalidState("Omp Turn is not accepting native input") }; + } + let accepted: Promise | undefined; + try { + accepted = transport.steer(text).then(() => { + active.acceptedSteerCount += 1; + }); + active.pendingSteers.add(accepted); + await accepted; + return { ok: true, value: { accepted: true } }; + } catch (error) { + if (error instanceof OmpRpcSteerRejectedError) { + return { ok: false, error: invalidState(error.message) }; + } + // A missing acknowledgement may follow native acceptance; never invite stop-and-resend. + throw error; + } finally { + if (accepted) active.pendingSteers.delete(accepted); + } + } + async #cancel(command: TurnCancelCommand): Promise> { const active = this.#active; if (!active || active.command.turnId !== command.turnId) { @@ -1519,6 +1582,9 @@ class OmpHarnessSession implements HarnessSession { agentMessageId: null, compactionItem: null, sawAssistantMessage: false, + acceptedSteerCount: 0, + acceptingSteers: false, + pendingSteers: new Set(), reasoningItem: null, tools: new Map(), interactions: new Map(), @@ -2095,6 +2161,8 @@ class OmpHarnessSession implements HarnessSession { active: ActiveTurn, transport: OmpTurnTransport, ): Promise<{ nativeTurnRef: NativeTurnRef; checkpoint: NativeCheckpointRef }> { + active.acceptingSteers = false; + await Promise.allSettled([...active.pendingSteers]); const snapshot = mapOmpSnapshot(await transport.getEntries(), { sessionId: transport.state.sessionId, model: nativeModelForHistory(transport.state), @@ -2102,9 +2170,11 @@ class OmpHarnessSession implements HarnessSession { const created = snapshot.turns.filter( (turn) => !active.beforeNativeTurnKeys.has(turn.nativeTurnRef.nativeTurnKey), ); - if (created.length !== 1) { + const expected = 1 + active.acceptedSteerCount; + if (created.length !== expected) { + const countText = expected === 1 ? "one" : `${expected}`; throw new Error( - `Omp Turn persisted ${created.length} new User Entries; exactly one is required`, + `Omp Turn persisted ${created.length} new User Entries; exactly ${countText} is required`, ); } const turn = created[0]; diff --git a/packages/adapters/omp/src/omp-rpc-session.ts b/packages/adapters/omp/src/omp-rpc-session.ts index 085898f51f..6754fa8252 100644 --- a/packages/adapters/omp/src/omp-rpc-session.ts +++ b/packages/adapters/omp/src/omp-rpc-session.ts @@ -169,6 +169,9 @@ export class OmpRpcFaultError extends Error { } } +/** A correlated native response explicitly refused steer before acceptance. */ +export class OmpRpcSteerRejectedError extends Error {} + export class OmpRpcUnsupportedCommandError extends Error { constructor(readonly command: string) { super(`Omp RPC does not support '${command}'`); @@ -934,6 +937,22 @@ export class OmpRpcSession { } } + async steer(message: string): Promise { + if (!this.#child || !this.#state || this.#closed || this.#failed) { + throw new Error("Omp RPC Session is unavailable"); + } + if (message.length === 0) throw new Error("Omp steer text must not be empty"); + const response = await this.#send("steer", { message }); + if ( + isRecord(response.data) && + response.data.disposition !== undefined && + response.data.disposition !== "queued" + ) { + // An input hook may have handled it already; retrying could repeat its effects. + throw new Error("Omp steer was handled without confirming same-Turn queue admission"); + } + } + async runTurn(text: string, onEvent: (event: OmpTurnEvent) => void): Promise { if (!this.#child || !this.#state || this.#closed || this.#failed) { throw new Error("Omp RPC Session is unavailable"); @@ -1655,7 +1674,12 @@ export class OmpRpcSession { pending.resolve(value); } else { const error = typeof value.error === "string" ? value.error : "Omp RPC command failed"; - if (value.command === pending.command && error === `Unknown command: ${pending.command}`) { + if (pending.command === "steer" && value.command === "steer" && value.success === false) { + pending.reject(new OmpRpcSteerRejectedError(error)); + } else if ( + value.command === pending.command && + error === `Unknown command: ${pending.command}` + ) { pending.reject(new OmpRpcUnsupportedCommandError(pending.command)); } else { pending.reject(new Error(error)); diff --git a/packages/adapters/omp/test/omp-adapter.test.ts b/packages/adapters/omp/test/omp-adapter.test.ts index d0a58eeb5d..bfd8237817 100644 --- a/packages/adapters/omp/test/omp-adapter.test.ts +++ b/packages/adapters/omp/test/omp-adapter.test.ts @@ -213,6 +213,8 @@ class FakeOmpTransport implements OmpTurnTransport { }); } + readonly steer = vi.fn(async () => undefined); + async abort(): Promise { this.#resolveTurn?.({ text: "", cancelled: true, agentInvoked: true }); } diff --git a/packages/adapters/omp/test/omp-steer.test.ts b/packages/adapters/omp/test/omp-steer.test.ts new file mode 100644 index 0000000000..d4a8137b74 --- /dev/null +++ b/packages/adapters/omp/test/omp-steer.test.ts @@ -0,0 +1,555 @@ +import { EventEmitter } from "node:events"; +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { PassThrough } from "node:stream"; + +import type { HarnessOutput, HarnessSession, HostEvent } from "@codexhost/harness-adapter"; +import { harnessThinkingOptionIdSchema, hostTurnIdSchema } from "@codexhost/shared-contracts"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { OmpAdapter, type OmpTurnTransport } from "../src/omp-adapter.js"; +import type { OmpSessionHistory } from "../src/omp-history.js"; +import { + OmpRpcSession, + OmpRpcSteerRejectedError, + type OmpRpcProcessAdapter, + type OmpSessionState, + type OmpTurnEvent, + type OmpTurnResult, +} from "../src/omp-rpc-session.js"; + +const adapters: OmpAdapter[] = []; + +afterEach(async () => { + await Promise.all(adapters.splice(0).map((adapter) => adapter.close())); +}); + +class SteerOmpTransport implements OmpTurnTransport { + readonly abort = vi.fn(async () => { + this.resolveTurn?.({ text: "", cancelled: true, agentInvoked: true }); + }); + readonly steer = vi.fn(async () => undefined); + state: OmpSessionState = { + sessionId: "omp-session-1", + sessionFile: "/synthetic/omp-session.jsonl", + provider: "synthetic", + modelId: "model", + thinkingLevel: harnessThinkingOptionIdSchema.parse("high"), + contextUsage: null, + availableThinkingLevels: [harnessThinkingOptionIdSchema.parse("high")], + }; + history: OmpSessionHistory = { entries: [], leafId: null }; + onEvent: ((event: OmpTurnEvent) => void) | null = null; + resolveTurn: ((result: OmpTurnResult) => void) | null = null; + + async start(): Promise {} + async getAvailableModels() { + return [{ provider: "synthetic", id: "model", reasoning: true }]; + } + async getAvailableThinkingLevels() { + return [harnessThinkingOptionIdSchema.parse("high")]; + } + async getEntries(): Promise { + return this.history; + } + async getSubagentMessages() { + return { + sessionFile: "/synthetic/subagent.jsonl", + fromByte: 0, + nextByte: 0, + reset: false, + entries: [], + messages: [], + }; + } + async getSessionUsage() { + return null; + } + async fork() { + return this.state; + } + async verifySessionCwd(): Promise {} + async selectModel() { + return this.state; + } + async selectThinkingOption() { + return this.state; + } + async compact() { + return { outcome: "succeeded" as const }; + } + runTurn(_text: string, onEvent: (event: OmpTurnEvent) => void): Promise { + this.onEvent = onEvent; + onEvent({ + type: "tool.started", + callId: "call-1", + toolName: "bash", + arguments: { command: "echo hello" }, + }); + return new Promise((resolve) => { + this.resolveTurn = (result) => { + onEvent({ + type: "tool.completed", + callId: "call-1", + toolName: "bash", + result: "hello", + isError: false, + }); + resolve(result); + }; + }); + } + async respondToInteraction(): Promise {} + async close(): Promise {} +} + +function openAdapter() { + const transports: SteerOmpTransport[] = []; + const adapter = new OmpAdapter( + {}, + { + createTransport: () => { + const transport = new SteerOmpTransport(); + transports.push(transport); + return transport; + }, + }, + ); + adapters.push(adapter); + return { adapter, transports }; +} + +async function openSession() { + const opened = openAdapter(); + const result = await opened.adapter.open({ kind: "create", cwd: "/synthetic" }); + if (!result.ok) throw new Error(result.error.message); + const outputs: HarnessOutput[] = []; + void (async () => { + for await (const output of result.value.outputs) outputs.push(output); + })(); + return { ...opened, session: result.value, outputs }; +} + +function hostEvents(outputs: HarnessOutput[]): HostEvent[] { + return outputs.flatMap((output) => (output.kind === "event" ? [output.event] : [])); +} + +function userMessageEvents(outputs: HarnessOutput[]) { + return hostEvents(outputs).filter( + (event) => + (event.type === "item.started" && event.item.type === "userMessage") || + (event.type === "item.completed" && event.snapshot.item.type === "userMessage"), + ); +} + +function historyOf(userTexts: string[]): OmpSessionHistory { + let parentId: string | null = null; + const entries: OmpSessionHistory["entries"] = []; + userTexts.forEach((text, index) => { + const userId = `user-${index + 1}`; + const assistantId = `assistant-${index + 1}`; + entries.push( + { + id: userId, + parentId, + type: "message", + message: { role: "user", content: [{ type: "text", text }] }, + }, + { + id: assistantId, + parentId: userId, + type: "message", + message: { + role: "assistant", + stopReason: "stop", + content: [{ type: "text", text: "answer" }], + }, + }, + ); + parentId = assistantId; + }); + return { entries, leafId: parentId }; +} + +describe("Omp steer capability", () => { + it("declares steer on the session and on inspect", async () => { + const opened = await openSession(); + expect(opened.session.capabilities.steer).toBe(false); + const inspected = await opened.adapter.inspect({ cwd: "/synthetic", refresh: true }); + expect(inspected.status === "ready" ? inspected.capabilities.steer : false).toBeUndefined(); + const turnId = hostTurnIdSchema.parse("capability-turn"); + await opened.session.execute({ + type: "turn.start", + turnId, + input: [{ type: "text", text: "run" }], + }); + expect(opened.session.capabilities.steer).toBe(true); + opened.transports[0]?.resolveTurn?.({ text: "done", cancelled: true, agentInvoked: true }); + }); +}); + +describe("Omp turn.steer", () => { + async function startTurn(session: HarnessSession, turnId = "turn-1") { + const id = hostTurnIdSchema.parse(turnId); + await expect( + session.execute({ + type: "turn.start", + turnId: id, + input: [{ type: "text", text: "run the tool" }], + }), + ).resolves.toMatchObject({ ok: true, value: { turnId: id } }); + return id; + } + + it("inserts into the active turn without publishing a user message", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [ + { type: "text", text: "line one" }, + { type: "text", text: "line two" }, + ], + }), + ).resolves.toEqual({ ok: true, value: { accepted: true } }); + expect(transport.steer).toHaveBeenCalledWith("line one\nline two"); + expect(transport.abort).not.toHaveBeenCalled(); + expect(userMessageEvents(outputs)).toEqual([]); + + await expect( + session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-2"), + input: [{ type: "text", text: "another" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "sessionBusy" } }); + + transport.history = historyOf(["run the tool", "line one\nline two"]); + transport.resolveTurn?.({ text: "pong", cancelled: false, agentInvoked: true }); + await vi.waitFor(() => { + expect(hostEvents(outputs).some((event) => event.type === "turn.completed")).toBe(true); + }); + expect(hostEvents(outputs).find((event) => event.type === "turn.completed")).toMatchObject({ + turnId, + nativeTurnRef: { nativeTurnKey: "user-1" }, + outcome: { status: "succeeded", checkpoint: { checkpointId: "user-1" } }, + }); + expect(userMessageEvents(outputs)).toEqual([]); + }); + + it("preserves acceptance when the Turn finishes before the steer receipt", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + const receipt = Promise.withResolvers(); + transport.steer.mockImplementationOnce(() => receipt.promise); + const steering = session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }); + transport.history = historyOf(["run the tool", "reply pong"]); + transport.resolveTurn?.({ text: "pong", cancelled: false, agentInvoked: true }); + await new Promise((resolve) => setImmediate(resolve)); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "too late" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + receipt.resolve(undefined); + await expect(steering).resolves.toEqual({ ok: true, value: { accepted: true } }); + await vi.waitFor(() => + expect(hostEvents(outputs).find((event) => event.type === "turn.completed")).toMatchObject({ + turnId, + nativeTurnRef: { nativeTurnKey: "user-1" }, + outcome: { status: "succeeded" }, + }), + ); + expect(transport.steer).toHaveBeenCalledOnce(); + }); + + it("still fails when the new user entry count does not match accepted steers", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + await session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }); + transport.history = historyOf(["run the tool", "reply pong", "other client"]); + transport.resolveTurn?.({ text: "pong", cancelled: false, agentInvoked: true }); + await vi.waitFor(() => { + expect(hostEvents(outputs).some((event) => event.type === "turn.completed")).toBe(true); + }); + expect(hostEvents(outputs).find((event) => event.type === "turn.completed")).toMatchObject({ + outcome: { + status: "failed", + error: { code: "protocolError", message: expect.stringContaining("exactly 2 is required") }, + }, + }); + }); + + it("stops offering steer when the last root tool ends", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + transport.onEvent?.({ + type: "tool.completed", + callId: "call-1", + toolName: "bash", + result: "done", + isError: false, + }); + expect(session.capabilities.steer).toBe(false); + await expect( + session.execute({ type: "turn.steer", turnId, input: [{ type: "text", text: "late" }] }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + expect(transport.steer).not.toHaveBeenCalled(); + }); + + it("rejects new input while cancellation is pending", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + transport.abort.mockImplementationOnce(async () => undefined); + await session.execute({ type: "turn.cancel", turnId }); + expect(session.capabilities.steer).toBe(false); + await expect( + session.execute({ type: "turn.steer", turnId, input: [{ type: "text", text: "late" }] }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + expect(transport.steer).not.toHaveBeenCalled(); + }); + + it("does not turn a lost native receipt into a safe-to-resend refusal", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + transport.steer.mockRejectedValueOnce(new Error("steer receipt timed out")); + await expect( + session.execute({ type: "turn.steer", turnId, input: [{ type: "text", text: "once" }] }), + ).rejects.toThrow("steer receipt timed out"); + expect(transport.steer).toHaveBeenCalledOnce(); + expect(transport.abort).not.toHaveBeenCalled(); + }); + + it("rejects an inactive turn, an empty steer, a native rejection, and a closed session", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + await expect( + session.execute({ + type: "turn.steer", + turnId: hostTurnIdSchema.parse("other-turn"), + input: [{ type: "text", text: "reply pong" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidRequest" } }); + expect(transports[0]?.steer).not.toHaveBeenCalled(); + transports[0]?.steer.mockRejectedValueOnce( + new OmpRpcSteerRejectedError("Omp RPC steer was not accepted"), + ); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }), + ).resolves.toMatchObject({ + ok: false, + error: { code: "invalidState", message: "Omp RPC steer was not accepted" }, + }); + await session.close(); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + }); +}); + +describe("Omp RPC steer", () => { + it.each(["missing", "handled"])( + "does not classify a %s native receipt as rejection", + async (receipt) => { + const fake = new FakeOmpProcess(); + fake.steerReceipt = receipt; + const rpc = new OmpRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 50, closeTimeoutMs: 500 }, + { spawn: () => fake as unknown as ChildProcessWithoutNullStreams }, + ); + await rpc.start(); + const failure = await rpc.steer("once").catch((error: unknown) => error); + expect(failure).toBeInstanceOf(Error); + expect(failure).not.toBeInstanceOf(OmpRpcSteerRejectedError); + expect(fake.steered).toEqual(["once"]); + await rpc.close(); + }, + ); + + it("marks only an explicit native rejection as not accepted", async () => { + const fake = new FakeOmpProcess(); + fake.rejectSteer = "Agent is idle"; + const rpc = new OmpRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 2_000, closeTimeoutMs: 500 }, + { spawn: () => fake as unknown as ChildProcessWithoutNullStreams }, + ); + await rpc.start(); + await expect(rpc.steer("reply pong")).rejects.toBeInstanceOf(OmpRpcSteerRejectedError); + await rpc.close(); + }); + + it("sends steer and ignores the user echo", async () => { + const fake = new FakeOmpProcess(); + const processAdapter: OmpRpcProcessAdapter = { + spawn: () => fake as unknown as ChildProcessWithoutNullStreams, + }; + const rpc = new OmpRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 2_000, closeTimeoutMs: 500 }, + processAdapter, + ); + const events: OmpTurnEvent[] = []; + await rpc.start(); + const running = rpc.runTurn("run the tool", (event) => events.push(event)); + await vi.waitFor(() => { + expect(events.some((event) => event.type === "tool.completed")).toBe(true); + }); + await rpc.steer("reply pong"); + await expect(running).resolves.toMatchObject({ text: "pong", cancelled: false }); + expect(fake.steered).toEqual(["reply pong"]); + const text = events.flatMap((event) => (event.type === "text.delta" ? [event.delta] : [])); + expect(text.join("")).toBe("pong"); + await rpc.close(); + }); +}); + +class FakeOmpProcess extends EventEmitter { + readonly stdin = new PassThrough(); + readonly stdout = new PassThrough(); + readonly stderr = new PassThrough(); + readonly pid = 42_002; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + readonly steered: string[] = []; + rejectSteer: string | null = null; + steerReceipt = "queued"; + #buffer = Buffer.alloc(0); + + constructor() { + super(); + this.stdin.on("data", (chunk: Buffer) => this.#push(chunk)); + this.stdin.once("finish", () => { + this.exitCode = 0; + this.stdout.end(); + this.stderr.end(); + this.emit("exit", 0, null); + }); + queueMicrotask(() => { + this.emit("spawn"); + this.#output({ type: "ready" }); + }); + } + + #push(chunk: Buffer): void { + this.#buffer = Buffer.concat([this.#buffer, chunk]); + let newline = this.#buffer.indexOf(0x0a); + while (newline >= 0) { + const line = this.#buffer.subarray(0, newline).toString("utf8"); + this.#buffer = this.#buffer.subarray(newline + 1); + this.#onCommand(JSON.parse(line) as Record); + newline = this.#buffer.indexOf(0x0a); + } + } + + #onCommand(command: Record): void { + if (command.type === "prompt") { + this.#respond(command); + this.#output({ + type: "tool_execution_start", + toolCallId: "call-1", + toolName: "bash", + args: { command: "echo hello" }, + }); + this.#output({ + type: "tool_execution_end", + toolCallId: "call-1", + toolName: "bash", + result: { output: "hello" }, + isError: false, + }); + return; + } + if (command.type === "steer") { + if (this.rejectSteer) { + this.#output({ + id: command.id, + type: "response", + command: "steer", + success: false, + error: this.rejectSteer, + }); + return; + } + if (typeof command.message === "string") this.steered.push(command.message); + if (this.steerReceipt === "missing") return; + if (this.steerReceipt === "handled") { + this.#respond(command, { disposition: "handled" }); + return; + } + this.#respond(command); + const user = { + role: "user", + steering: true, + content: [{ type: "text", text: "reply pong" }], + }; + const assistant = { + role: "assistant", + content: [{ type: "text", text: "pong" }], + stopReason: "stop", + }; + this.#output({ type: "message_start", message: user }); + this.#output({ type: "message_end", message: user }); + this.#output({ type: "message_end", message: assistant }); + this.#output({ type: "agent_end", isTerminal: true, messages: [assistant] }); + return; + } + this.#respond(command, { + sessionId: "omp-session", + sessionFile: null, + model: { provider: "synthetic", id: "model", thinking: { efforts: ["high"] } }, + thinkingLevel: "high", + isStreaming: false, + }); + } + + #respond(command: Record, data?: unknown): void { + this.#output({ + id: command.id, + type: "response", + command: command.type, + success: true, + ...(data === undefined ? {} : { data }), + }); + } + + #output(value: unknown): void { + this.stdout.write(`${JSON.stringify(value)}\n`); + } +} diff --git a/packages/adapters/pi/CAPABILITIES.md b/packages/adapters/pi/CAPABILITIES.md index 6da0ca0aeb..5341b2e8f3 100644 --- a/packages/adapters/pi/CAPABILITIES.md +++ b/packages/adapters/pi/CAPABILITIES.md @@ -22,9 +22,9 @@ | A5 | 执行 Turn | 必做 | P0 | 受限 | Pi RPC / adapter | 请求与自主回合组合回归 | 本轮验证准入、输出和终态；Desktop 应用待验收 | | A6 | 取消 | 必做 | P0 | 受限 | Pi RPC / adapter | 首消息前取消、旧轮收尾、close 回归 | 不误取消后继自主回合 | | A7 | 只读快照 | 必做 | P0 | 受限 | pi-history | 真实历史候选回放 | 可见通知、phase 和命令元数据；不宣称所有快照形态已覆盖 | -| A8 | 持久身份 + resume | 必做 | P0 | 待验证 | 既有 Adapter | 本次未重验 | 只验 B11 及受影响路径 | -| A9 | 错误与并发 | 必做 | P0 | 受限 | Pi RPC / adapter | 异步预检竞争、typed sessionBusy 回归 | 未接纳输入不 dispatch、不自动重试 | -| A10 | 关闭 | 必做 | P0 | 待验证 | 既有 Adapter | 本次未重验 | 只验 B11 及受影响路径 | +| A8 | 持久身份 + resume | 必做 | P0 | 受限 | pi-adapter.ts | pi-steer-delayed-input.real.test.ts；pi-adapter.test.ts | 晚到插话故障后恢复同一 Session，历史身份不变；其他平台待验收 | +| A9 | 错误与并发 | 必做 | P0 | 受限 | Pi RPC / adapter | pi-steer.test.ts；异步预检竞争、typed sessionBusy 回归 | 写入前拒绝与结果未知分开；晚到输入不报成功、不自动重试 | +| A10 | 关闭 | 必做 | P0 | 受限 | pi-rpc-session.ts / pi-adapter.ts | pi-steer.test.ts；pi-steer-delayed-input.real.test.ts | 等待专属进程退出后完成晚到插话恢复；未验所有 OS | | B1 | 切换 Model | 原生有则必做 | P1 | 待验证 | 既有 Adapter | 本次未重验 | 只验 B11 及受影响路径 | | B2 | 切换 Thinking | 原生有则必做 | P1 | 待验证 | 既有 Adapter | 本次未重验 | 只验 B11 及受影响路径 | | B3 | 权限模式 | 原生有则必做 | P1 | 待验证 | 既有 Adapter | 本次未重验 | 只验 B11 及受影响路径 | @@ -55,6 +55,14 @@ | D4 | 文档 | 预装时必做 | P1 | 待验证 | 既有 Adapter | 本次未重验 | 只验 B11 及受影响路径 | | D5 | 其他 | 可选 | P2 | 待验证 | 既有 Adapter | 本次未重验 | 只验 B11 及受影响路径 | +## 原生插话边界 + +- 范围：A5、A7、A9、A10；优先级 P0，结论为受限。实现位于 `pi-adapter.ts`、`pi-rpc-session.ts`，使用 Pi 原生 RPC，不新增公共契约。 +- 工具执行中的正常插话保持原生入队。成功回执必须在目标原生执行尚未终结时收到；终结之后才收到的入队回执按结果不确定处理，不触发停止后重发。 +- 晚到回执的恢复保留已完成 Turn 的原生历史，关闭该 Thread 专属的 Pi 进程以丢弃未执行队列，再由现有 Session 故障恢复路径恢复同一历史。不修改全局配置或其他 Thread，不声称可以撤销 input hook 已产生的副作用。 +- 验证：`pi-steer.test.ts` 覆盖回执顺序、拒绝与未知结果、历史身份及清理；`test/pi-steer-delayed-input.real.test.ts` 在 macOS / Pi 1.4.0 上验证晚到输入不报成功、原 Turn 成功、专属进程退出、同一 Session 恢复后可继续且无遗留输入重放。该用例在未修复 PR 上失败，在修复版本上通过。macOS / Desktop 26.1103.31037 的隔离窗口经真实 Composer 验证正常插话、晚到输入、原轮成功、同一 Session 恢复、旧队列不自动重放及删除后普通发送；多窗口、远端和其他 OS 未验收。 +- 检查记录：Pi、OMP 及 Host steering、未知交付队列与回执的定向合成测试通过；默认跳过的两个原生文件中，Pi 延迟输入用例已单独显式运行通过，OMP 原生 ask 未运行。Pi / OMP 的正常原生插话 Gate 均通过。类型检查、定向 ESLint / Prettier、boundary 与 diff 检查通过；插件重新打包后，经仓库外临时根的公共 Loader 冷加载、独立实例及关闭验证通过。 + ## Windows UNC 启动验证范围 - 范围：A4、A8、B5、C5 的进程启动边界；仅处理 Windows UNC 工作目录与标准 Pi CMD launcher 的组合，不改 RPC、Model、Thinking、扩展或历史语义。 @@ -66,11 +74,11 @@ ## 受限与已知缺口 -本清单补记于既有 Adapter 修复；未影响的能力本轮不重新验收，因此明确列为待验证，不代表原实现不支持。声明子代理修复及下述输出修复实际覆盖范围。Desktop 应用待空闲时重启并验收，不把候选插件通过验证当作已生效。 +本清单补记于既有 Adapter 修复；未影响的能力本轮不重新验收，因此明确列为待验证，不代表原实现不支持。声明子代理修复及下述输出修复实际覆盖范围。Desktop 插话已在隔离窗口验收；其他能力仍按各项记录界定，不把候选插件通过验证当作已在日常窗口生效。 ## 插件外改动 -对应 docs/harnesses/pi/pi-subagents.md、pi-progress.md 功能说明及 docs/index.md 链接描述；不新增公共合同或 Harness 专用 Host/Renderer 分支。 +对应 docs/harnesses/pi/pi-subagents.md、pi-progress.md 功能说明及 docs/index.md 链接描述；原生插话边界补充 `docs/harnesses/pi/pi-steer.md`、`docs/architecture/external-thread-steering.md`。不新增公共契约或 Harness 专用 Host/Renderer 分支。 ## 验证记录 diff --git a/packages/adapters/pi/src/pi-adapter.ts b/packages/adapters/pi/src/pi-adapter.ts index a8db28374c..11b89a803d 100644 --- a/packages/adapters/pi/src/pi-adapter.ts +++ b/packages/adapters/pi/src/pi-adapter.ts @@ -12,7 +12,6 @@ import { randomUUID } from "node:crypto"; import path from "node:path"; import { - unsupportedSteer, HarnessOutputChannel, validateHostQuestionResponse, type HarnessAdapter, @@ -112,6 +111,8 @@ import { PiRpcFaultError, PiRpcBusyError, PiRpcSession, + PiRpcSteerRejectedError, + PiRpcSteerTargetEndedError, PiRpcUnsupportedCommandError, type PiInteractionRequest, type PiInteractionResponse, @@ -180,6 +181,7 @@ export interface PiTurnTransport { onEvent: (event: PiTurnEvent) => void, ): Promise; runTurn(text: string, onEvent: (event: PiTurnEvent) => void): Promise; + steer(message: string): Promise; respondToInteraction(response: PiInteractionResponse): Promise; abort(): Promise; close(): Promise; @@ -207,6 +209,10 @@ interface ActiveTurn { agentMessageId: string | null; compactionItem: HostContextCompactionItem | null; sawAssistantMessage: boolean; + acceptedSteerCount: number; + acceptingSteers: boolean; + pendingSteers: Set>; + steerFailure?: PiRpcSteerTargetEndedError; reasoningItem: HostReasoningItem | null; tools: Map; interactions: Map; @@ -458,17 +464,21 @@ class PiHarnessSession implements HarnessSession { this.#requestedThinkingOptionId = options.thinkingOptionId; this.#toolOutputLimit = options.toolOutputLimit; this.#isFastSupported = options.isFastSupported; - this.capabilities = { + this.capabilities = ((session) => ({ configuration: { selectModel: true, selectThinkingOption: options.supportsThinkingSelection, selectPermissionMode: false, - permissionModeScope: "live", + permissionModeScope: "live" as const, }, history: { fork: true, forkAcrossCwd: true, rollbackLastTurn: true }, autonomousTurns: { observe: true }, subagents: { observe: true, readTranscript: true }, - }; + // Root Tools provide an admission window; the transport also checks native receipt order. + get steer(): boolean { + return session.#canSteerActiveTurn(); + }, + }))(this); this.commands = { list: async () => ({ ok: true, value: await this.#refreshLiveCommands() }), execute: (command) => this.#executeHarnessCommand(command), @@ -610,7 +620,6 @@ class PiHarnessSession implements HarnessSession { if (this.#phase !== "open") { return { ok: false, error: invalidState("Pi Session is not open") }; } - if (command.type === "turn.steer") return unsupportedSteer("Pi"); if (command.type === "turn.cancel") return this.#cancel(command); if (command.type === "interaction.respond") return this.#respond(command); if (command.type === "model.select") return this.#selectModel(command); @@ -625,6 +634,7 @@ class PiHarnessSession implements HarnessSession { }, }; } + if (command.type === "turn.steer") return this.#steer(command); if (this.#acceptingTurn || this.#active || this.#configuring || this.#transport?.isBusy) { return { ok: false, @@ -695,6 +705,9 @@ class PiHarnessSession implements HarnessSession { agentMessageId: null, compactionItem: null, sawAssistantMessage: false, + acceptedSteerCount: 0, + acceptingSteers: true, + pendingSteers: new Set(), reasoningItem: null, tools: new Map(), interactions: new Map(), @@ -984,6 +997,60 @@ class PiHarnessSession implements HarnessSession { } } + #canSteerActiveTurn(): boolean { + const active = this.#active; + return Boolean( + this.#phase === "open" && + active && + !active.cancellationRequested && + active.acceptingSteers && + active.tools.size > 0 && + this.#transport, + ); + } + + async #steer(command: TurnSteerCommand): Promise> { + const text = command.input.map((input) => input.text).join("\n"); + if (text.length === 0) { + return { + ok: false, + error: { + code: "invalidRequest", + message: "Pi steer text must not be empty", + retryable: false, + }, + }; + } + const active = this.#active; + const transport = this.#transport; + if ( + !active || + !transport || + active.command.turnId !== command.turnId || + !this.#canSteerActiveTurn() + ) { + return { ok: false, error: invalidState("Pi Turn is not accepting native input") }; + } + let accepted: Promise | undefined; + try { + accepted = transport.steer(text).then(() => { + active.acceptedSteerCount += 1; + }); + active.pendingSteers.add(accepted); + await accepted; + return { ok: true, value: { accepted: true } }; + } catch (error) { + if (error instanceof PiRpcSteerRejectedError) { + return { ok: false, error: invalidState(error.message) }; + } + if (error instanceof PiRpcSteerTargetEndedError) active.steerFailure = error; + // A missing acknowledgement may follow native acceptance; never invite stop-and-resend. + throw error; + } finally { + if (accepted) active.pendingSteers.delete(accepted); + } + } + async #cancel(command: TurnCancelCommand): Promise> { const active = this.#active; if (!active || active.command.turnId !== command.turnId) { @@ -1123,6 +1190,9 @@ class PiHarnessSession implements HarnessSession { agentMessageId: null, compactionItem: null, sawAssistantMessage: false, + acceptedSteerCount: 0, + acceptingSteers: false, + pendingSteers: new Set(), reasoningItem: null, tools: new Map(), interactions: new Map(), @@ -1430,6 +1500,9 @@ class PiHarnessSession implements HarnessSession { agentMessageId: null, compactionItem: null, sawAssistantMessage: false, + acceptedSteerCount: 0, + acceptingSteers: false, + pendingSteers: new Set(), reasoningItem: null, tools: new Map(), interactions: new Map(), @@ -1978,7 +2051,9 @@ class PiHarnessSession implements HarnessSession { active: ActiveTurn, transport: PiTurnTransport, ): Promise<{ nativeTurnRef: NativeTurnRef; checkpoint?: NativeCheckpointRef }> { - const snapshot = mapPiSnapshot(await transport.getEntries(), { + active.acceptingSteers = false; + await Promise.allSettled([...active.pendingSteers]); + const snapshot = mapPiSnapshot(active.steerFailure?.history ?? (await transport.getEntries()), { sessionId: transport.state.sessionId, model: nativeModelForHistory(transport.state), }); @@ -1986,9 +2061,10 @@ class PiHarnessSession implements HarnessSession { (turn) => turn.input.length > 0 && !active.beforeNativeTurnKeys.has(turn.nativeTurnRef.nativeTurnKey), ); - if (created.length !== 1) { + const expected = 1 + active.acceptedSteerCount; + if (created.length !== expected) { throw new Error( - `Pi Turn persisted ${created.length} new User Entries; exactly one is required`, + `Pi Turn persisted ${created.length} new User Entries; exactly ${expected} is required`, ); } const turn = created[0]; @@ -2036,6 +2112,11 @@ class PiHarnessSession implements HarnessSession { ...(nativeTurnRef ? { nativeTurnRef } : {}), }); active.resolveCompletion(); + if (active.steerFailure) { + // The completed Turn keeps its real outcome; Host recovery reopens the same history. + this.#fault(active.steerFailure); + return; + } queueMicrotask(() => { if (this.#phase !== "open") return; if (this.#transport) this.#drainPendingAutonomousTurns(this.#transport); diff --git a/packages/adapters/pi/src/pi-rpc-session.ts b/packages/adapters/pi/src/pi-rpc-session.ts index c3ed39cc17..a2ea822e67 100644 --- a/packages/adapters/pi/src/pi-rpc-session.ts +++ b/packages/adapters/pi/src/pi-rpc-session.ts @@ -189,6 +189,19 @@ export class PiRpcFaultError extends Error { } } +/** A correlated native response explicitly refused steer before acceptance. */ +export class PiRpcSteerRejectedError extends Error {} + +/** An idle queue cannot establish delivery to the requested Turn. Never replay this input. */ +export class PiRpcSteerTargetEndedError extends PiRpcFaultError { + constructor(readonly history: PiSessionHistory) { + super( + "protocolError", + "Pi Turn ended before steer admission could be confirmed; the isolated runtime was closed. Input hook effects are unknown, so do not automatically resend the input", + ); + } +} + export class PiRpcUnsupportedCommandError extends Error { constructor(readonly command: string) { super(`Pi RPC does not support '${command}'`); @@ -1047,6 +1060,37 @@ export class PiRpcSession { } } + async steer(message: string): Promise { + if (!this.#child || !this.#state || this.#closed || this.#failed) { + throw new Error("Pi RPC Session is unavailable"); + } + if (message.length === 0) throw new Error("Pi steer text must not be empty"); + const target = this.#activeTurn; + if (!target || target.origin !== "requested" || target.settlement !== "pending") { + throw new PiRpcSteerRejectedError("Pi Turn is not accepting native input"); + } + let admitted = false; + const response = await this.#send("steer", { message }, () => { + // Capture native frame order, not the state after an awaiting continuation runs. + admitted = this.#activeTurn === target && target.settlement === "pending"; + }); + if ( + isRecord(response.data) && + response.data.disposition !== undefined && + response.data.disposition !== "queued" + ) { + // An input hook may have handled it already; retrying could repeat its effects. + throw new Error("Pi steer was handled without confirming same-Turn queue admission"); + } + if (!admitted) { + // Native input hooks may outlive the run and then enqueue into an idle Session. + // Preserve the terminal history before closing this owned process and its queue. + const history = await this.getEntries(); + await this.close(); + throw new PiRpcSteerTargetEndedError(history); + } + } + async runTurn(text: string, onEvent: (event: PiTurnEvent) => void): Promise { if (!this.#child || !this.#state || this.#closed || this.#failed) { throw new Error("Pi RPC Session is unavailable"); @@ -1732,7 +1776,12 @@ export class PiRpcSession { if (value.success === true) pending.resolve(value); else { const error = typeof value.error === "string" ? value.error : "Pi RPC command failed"; - if (value.command === pending.command && error === `Unknown command: ${pending.command}`) { + if (pending.command === "steer" && value.command === "steer" && value.success === false) { + pending.reject(new PiRpcSteerRejectedError(error)); + } else if ( + value.command === pending.command && + error === `Unknown command: ${pending.command}` + ) { pending.reject(new PiRpcUnsupportedCommandError(pending.command)); } else { pending.reject(new Error(error)); @@ -2092,7 +2141,11 @@ export class PiRpcSession { }); } - #send(type: string, payload: Record): Promise> { + #send( + type: string, + payload: Record, + onResponse?: () => void, + ): Promise> { const child = this.#child; if (!child?.stdin.writable || this.#closed || this.#failed) { return Promise.reject(new Error("Pi RPC stdin is unavailable")); @@ -2101,7 +2154,10 @@ export class PiRpcSession { return new Promise((resolve, reject) => { const pending: PendingCommand = { command: type, - resolve, + resolve: (response) => { + onResponse?.(); + resolve(response); + }, reject, timeout: null, }; diff --git a/packages/adapters/pi/test/pi-adapter.test.ts b/packages/adapters/pi/test/pi-adapter.test.ts index 728270cffa..74529b655a 100644 --- a/packages/adapters/pi/test/pi-adapter.test.ts +++ b/packages/adapters/pi/test/pi-adapter.test.ts @@ -180,6 +180,7 @@ class FakePiTransport implements PiTurnTransport { return { outcome: "succeeded" }; }, ); + readonly steer = vi.fn(async () => undefined); readonly runTurn = vi.fn((text: string, onEvent: (event: PiTurnEvent) => void) => { this.text = text; this.onEvent = onEvent; diff --git a/packages/adapters/pi/test/pi-steer-delayed-input.real.test.ts b/packages/adapters/pi/test/pi-steer-delayed-input.real.test.ts new file mode 100644 index 0000000000..b3c116b1b7 --- /dev/null +++ b/packages/adapters/pi/test/pi-steer-delayed-input.real.test.ts @@ -0,0 +1,278 @@ +import assert from "node:assert/strict"; +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { it } from "vitest"; +import type { HarnessSession, HostEvent } from "@codexhost/harness-adapter"; +import { hostTurnIdSchema } from "@codexhost/shared-contracts"; +import { PiAdapter } from "../src/pi-adapter.js"; +import { + PiRpcSession, + PiRpcSteerRejectedError, + piRpcProcessCommand, +} from "../src/pi-rpc-session.js"; + +const enabled = (process.env.CODEXHOST_STEER_REAL_HARNESSES ?? "").split(",").includes("pi"); + +// An opt-in native regression. Only the extra input hook is controlled; the Adapter, +// CLI, Model and persisted history are real. Uses existing auth and Model quota. +it.runIf(enabled)( + "preserves the completed Pi Turn and recovers without replaying late input", + async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "codexhost-steer-delayed-")); + const log = path.join(cwd, "native-hook.jsonl"); + const extension = path.join(cwd, "delayed-input.ts"); + const marker = `CODEXHOST_DELAYED_STEER_${randomUUID().replaceAll("-", "")}`; + await writeFile( + extension, + `import { appendFileSync } from "node:fs"; +export default function (pi) { + let release = null; + const log = (event) => appendFileSync(${JSON.stringify(log)}, JSON.stringify({ event, at: Date.now() }) + "\\n"); + pi.on("agent_settled", () => { + log("agent_settled"); + if (release) setTimeout(() => release?.(), 100); + }); + pi.on("input", async (event) => { + if (!event.text.startsWith(${JSON.stringify(marker)})) return { action: "continue" }; + log("steer_input_enter"); + await new Promise((resolve) => { + const timeout = setTimeout(() => { log("hook_timeout"); resolve(); }, 20000); + release = () => { clearTimeout(timeout); resolve(); }; + }); + log("steer_input_release"); + release = null; + return { action: "continue" }; + }); +} +`, + ); + const children: ChildProcessWithoutNullStreams[] = []; + const consumers: Promise[] = []; + const sessionsDirectory = path.join(cwd, "sessions"); + const adapter = new PiAdapter( + {}, + { + createTransport: (options) => + new PiRpcSession( + { + ...options, + environment: { + ...process.env, + ...options.environment, + PI_CODING_AGENT_SESSION_DIR: sessionsDirectory, + }, + commandTimeoutMs: 45000, + }, + { + spawn(processOptions) { + const invocation = piRpcProcessCommand(processOptions); + const arguments_ = [...invocation.arguments]; + const mode = arguments_.indexOf("--mode"); + assert.ok(mode >= 0, "Delayed-input Gate requires direct Pi CLI arguments"); + arguments_.splice(mode, 0, "--extension", extension); + const child = spawn(invocation.command, arguments_, { + cwd: processOptions.cwd, + env: processOptions.environment, + stdio: ["pipe", "pipe", "pipe"], + windowsHide: true, + windowsVerbatimArguments: invocation.windowsVerbatimArguments, + }); + children.push(child); + return child; + }, + }, + ), + }, + ); + const report: Record = { status: "running", harness: "pi" }; + const collect = (session: HarnessSession): HostEvent[] => { + const events: HostEvent[] = []; + consumers.push( + (async () => { + for await (const output of session.outputs) { + if (output.kind === "event") events.push(output.event); + } + })(), + ); + return events; + }; + const wait = async ( + events: HostEvent[], + predicate: (event: HostEvent) => boolean, + label: string, + ): Promise => { + const deadline = Date.now() + 90000; + while (Date.now() < deadline) { + const event = events.find(predicate); + if (event) return event; + assert.ok( + !events.some((event) => event.type === "session.faulted"), + `Fault before ${label}`, + ); + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`Timed out: ${label}`); + }; + try { + const opened = await adapter.open({ kind: "create", cwd }); + assert.ok(opened.ok, opened.ok ? "" : opened.error.message); + const session = opened.value; + const events = collect(session); + const turnId = hostTurnIdSchema.parse(randomUUID()); + assert.equal( + ( + await session.execute({ + type: "turn.start", + turnId, + input: [ + { + type: "text", + text: "Run exactly one foreground shell command: sleep 3; printf 'ORIGINAL_TOOL_DONE\\n'. Wait for it, then reply exactly ORIGINAL_FINISHED. Do not run any other tools or read or modify files.", + }, + ], + }) + ).ok, + true, + ); + await wait( + events, + (event) => + event.type === "item.started" && + ["commandExecution", "toolExecution"].includes(event.item.type), + "foreground tool", + ); + assert.equal(session.capabilities.steer, true); + const failure = await session + .execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: `${marker}. Reply exactly DELAYED_STEER_REACHED.` }], + }) + .catch((error) => error); + assert.ok( + failure instanceof Error && !(failure instanceof PiRpcSteerRejectedError), + "Late queue admission must not report acceptance or a safe-to-replay refusal", + ); + const terminal = await wait( + events, + (event) => event.type === "turn.completed" && event.turnId === turnId, + "original completion", + ); + assert.ok(terminal.type === "turn.completed"); + assert.equal(terminal.outcome.status, "succeeded", JSON.stringify(terminal.outcome)); + assert.ok(terminal.nativeTurnRef); + await consumers[0]; + const faultIndex = events.findIndex((event) => event.type === "session.faulted"); + assert.ok( + faultIndex > events.indexOf(terminal), + "Fault must follow, not replace, the completed Turn", + ); + const originalProcess = children[0]; + assert.ok(originalProcess); + assert.ok( + originalProcess.exitCode !== null || originalProcess.signalCode !== null, + "Retired runtime must have exited", + ); + const nativeRef = events.findLast( + (event): event is Extract => + event.type === "session.state.changed" && event.state.nativeRef !== undefined, + )?.state.nativeRef; + assert.ok(nativeRef, "Original native Session must remain resumable"); + const hookEvents = (await readFile(log, "utf8")) + .trim() + .split("\n") + .map((line) => JSON.parse(line)); + assert.deepEqual( + hookEvents.map((event) => event.event), + ["steer_input_enter", "agent_settled", "steer_input_release"], + ); + await session.close(); + const resumed = await adapter.open({ kind: "resume", nativeRef, cwd }); + assert.ok(resumed.ok, resumed.ok ? "" : resumed.error.message); + const recovered = resumed.value; + const recoveredRef = recovered.initialState.nativeRef; + assert.ok(recoveredRef); + assert.equal(recoveredRef.nativeSessionId, nativeRef.nativeSessionId); + const recoveredEvents = collect(recovered); + const followUpId = hostTurnIdSchema.parse(randomUUID()); + assert.equal( + ( + await recovered.execute({ + type: "turn.start", + turnId: followUpId, + input: [ + { + type: "text", + text: "Reply exactly RECOVERY_OK. Do not run any tools or read or modify files.", + }, + ], + }) + ).ok, + true, + ); + const followUp = await wait( + recoveredEvents, + (event) => event.type === "turn.completed" && event.turnId === followUpId, + "recovered completion", + ); + assert.ok(followUp.type === "turn.completed"); + assert.equal(followUp.outcome.status, "succeeded", JSON.stringify(followUp.outcome)); + const snapshot = await recovered.readSnapshot(); + assert.ok(snapshot.ok); + assert.equal(snapshot.value.turns.length, 2); + assert.deepEqual(snapshot.value.turns[0]?.nativeTurnRef, terminal.nativeTurnRef); + assert.ok( + snapshot.value.turns.every((turn) => + turn.input.every((input) => !input.text.includes(marker)), + ), + "The retired queue must not leak into a later Turn", + ); + const replies = recoveredEvents + .flatMap((event) => + event.type === "item.completed" && event.snapshot.item.type === "agentMessage" + ? [event.snapshot.item.text] + : [], + ) + .join("\n"); + assert.ok(replies.includes("RECOVERY_OK")); + assert.ok(!replies.includes("DELAYED_STEER_REACHED")); + Object.assign(report, { + status: "passed", + originalOutcome: terminal.outcome.status, + recoveredOutcome: followUp.outcome.status, + sameSession: true, + staleInputReplayed: false, + hookEvents, + }); + } catch (error) { + Object.assign(report, { + status: "failed", + error: error instanceof Error ? error.message : String(error), + }); + throw error; + } finally { + try { + await adapter.close(); + await Promise.all(consumers); + await rm(cwd, { recursive: true, force: true }); + } catch (error) { + Object.assign(report, { + status: "failed", + cleanupError: error instanceof Error ? error.message : String(error), + }); + throw error; + } finally { + const reportDirectory = path.resolve(".codexhost", "native-steer-real", "pi-delayed-input"); + await mkdir(reportDirectory, { recursive: true }); + await writeFile( + path.join(reportDirectory, "result.json"), + `${JSON.stringify(report, null, 2)}\n`, + ); + } + } + }, + 180000, +); diff --git a/packages/adapters/pi/test/pi-steer.test.ts b/packages/adapters/pi/test/pi-steer.test.ts new file mode 100644 index 0000000000..e2960f90f4 --- /dev/null +++ b/packages/adapters/pi/test/pi-steer.test.ts @@ -0,0 +1,696 @@ +import { EventEmitter } from "node:events"; +import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { PassThrough } from "node:stream"; + +import type { HarnessOutput, HarnessSession, HostEvent } from "@codexhost/harness-adapter"; +import { harnessThinkingOptionIdSchema, hostTurnIdSchema } from "@codexhost/shared-contracts"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { PiAdapter, type PiTurnTransport } from "../src/pi-adapter.js"; +import type { PiSessionHistory } from "../src/pi-history.js"; +import { + PiRpcSession, + PiRpcSteerRejectedError, + PiRpcSteerTargetEndedError, + type PiRpcProcessAdapter, + type PiSessionState, + type PiTurnEvent, + type PiTurnResult, +} from "../src/pi-rpc-session.js"; + +const adapters: PiAdapter[] = []; + +afterEach(async () => { + await Promise.all(adapters.splice(0).map((adapter) => adapter.close())); +}); + +class SteerPiTransport implements PiTurnTransport { + readonly abort = vi.fn(async () => { + this.resolveTurn?.({ text: "", cancelled: true }); + }); + readonly steer = vi.fn(async () => undefined); + readonly setAutonomousTurnHandler = vi.fn(); + state: PiSessionState = { + sessionId: "pi-session-1", + sessionFile: "/synthetic/pi-session.jsonl", + provider: "synthetic-provider", + modelId: "synthetic-model", + thinkingLevel: harnessThinkingOptionIdSchema.parse("off"), + contextUsage: null, + }; + history: PiSessionHistory = { entries: [], leafId: null }; + onEvent: ((event: PiTurnEvent) => void) | null = null; + resolveTurn: ((result: PiTurnResult) => void) | null = null; + + async start(): Promise {} + async getAvailableModels() { + return [{ provider: "synthetic-provider", id: "synthetic-model", reasoning: false }]; + } + async getAvailableThinkingLevels() { + return [harnessThinkingOptionIdSchema.parse("off")]; + } + async getEntries(): Promise { + return this.history; + } + async getSessionUsage() { + return null; + } + async fork() { + return this.state; + } + async clone() { + return this.state; + } + async verifySessionCwd(): Promise {} + async selectModel() { + return this.state; + } + async selectThinkingOption() { + return this.state; + } + async compact() { + return { outcome: "succeeded" as const }; + } + runTurn(_text: string, onEvent: (event: PiTurnEvent) => void): Promise { + this.onEvent = onEvent; + onEvent({ + type: "tool.started", + callId: "call-1", + toolName: "bash", + arguments: { command: "echo hello" }, + }); + return new Promise((resolve) => { + this.resolveTurn = (result) => { + onEvent({ + type: "tool.completed", + callId: "call-1", + toolName: "bash", + result: "hello", + isError: false, + }); + resolve(result); + }; + }); + } + async respondToInteraction(): Promise {} + async close(): Promise {} +} + +function openAdapter() { + const transports: SteerPiTransport[] = []; + const adapter = new PiAdapter( + {}, + { + createTransport: () => { + const transport = new SteerPiTransport(); + transports.push(transport); + return transport; + }, + }, + ); + adapters.push(adapter); + return { adapter, transports }; +} + +async function openSession() { + const opened = openAdapter(); + const result = await opened.adapter.open({ kind: "create", cwd: "/synthetic" }); + if (!result.ok) throw new Error(result.error.message); + const outputs: HarnessOutput[] = []; + void (async () => { + for await (const output of result.value.outputs) outputs.push(output); + })(); + return { ...opened, session: result.value, outputs }; +} + +function hostEvents(outputs: HarnessOutput[]): HostEvent[] { + return outputs.flatMap((output) => (output.kind === "event" ? [output.event] : [])); +} + +function userMessageEvents(outputs: HarnessOutput[]) { + return hostEvents(outputs).filter( + (event) => + (event.type === "item.started" && event.item.type === "userMessage") || + (event.type === "item.completed" && event.snapshot.item.type === "userMessage"), + ); +} + +function historyOf(userTexts: string[]): PiSessionHistory { + let parentId: string | null = null; + const entries: PiSessionHistory["entries"] = []; + userTexts.forEach((text, index) => { + const userId = `user-${index + 1}`; + const assistantId = `assistant-${index + 1}`; + entries.push( + { + id: userId, + parentId, + type: "message", + message: { role: "user", content: [{ type: "text", text }] }, + }, + { + id: assistantId, + parentId: userId, + type: "message", + message: { + role: "assistant", + stopReason: "stop", + content: [{ type: "text", text: "answer" }], + }, + }, + ); + parentId = assistantId; + }); + return { entries, leafId: parentId }; +} + +describe("Pi steer capability", () => { + it("declares steer on the session and on inspect", async () => { + const opened = await openSession(); + expect(opened.session.capabilities.steer).toBe(false); + const inspected = await opened.adapter.inspect({ cwd: "/synthetic", refresh: true }); + expect(inspected.status === "ready" ? inspected.capabilities.steer : false).toBeUndefined(); + const turnId = hostTurnIdSchema.parse("capability-turn"); + await opened.session.execute({ + type: "turn.start", + turnId, + input: [{ type: "text", text: "run" }], + }); + expect(opened.session.capabilities.steer).toBe(true); + opened.transports[0]?.resolveTurn?.({ text: "done", cancelled: true }); + }); +}); + +describe("Pi turn.steer", () => { + async function startTurn(session: HarnessSession, turnId = "turn-1") { + const id = hostTurnIdSchema.parse(turnId); + await expect( + session.execute({ + type: "turn.start", + turnId: id, + input: [{ type: "text", text: "run the tool" }], + }), + ).resolves.toMatchObject({ ok: true, value: { turnId: id } }); + return id; + } + + it("inserts into the active turn without publishing a user message", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [ + { type: "text", text: "line one" }, + { type: "text", text: "line two" }, + ], + }), + ).resolves.toEqual({ ok: true, value: { accepted: true } }); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "second" }], + }), + ).resolves.toEqual({ ok: true, value: { accepted: true } }); + expect(transport.steer).toHaveBeenNthCalledWith(1, "line one\nline two"); + expect(transport.steer).toHaveBeenNthCalledWith(2, "second"); + expect(transport.abort).not.toHaveBeenCalled(); + expect(userMessageEvents(outputs)).toEqual([]); + + await expect( + session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("turn-2"), + input: [{ type: "text", text: "another" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "sessionBusy" } }); + + transport.history = historyOf(["run the tool", "line one\nline two", "second"]); + transport.resolveTurn?.({ text: "pong", cancelled: false }); + await vi.waitFor(() => { + expect(hostEvents(outputs).some((event) => event.type === "turn.completed")).toBe(true); + }); + expect(hostEvents(outputs).find((event) => event.type === "turn.completed")).toMatchObject({ + turnId, + nativeTurnRef: { nativeTurnKey: "user-1" }, + outcome: { status: "succeeded", checkpoint: { checkpointId: "user-1" } }, + }); + expect(userMessageEvents(outputs)).toEqual([]); + }); + + it("waits for an accepted steer before completing and closes steer admission", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + const accepted = Promise.withResolvers(); + transport.steer.mockImplementationOnce(() => accepted.promise); + const steering = session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }); + await vi.waitFor(() => expect(transport.steer).toHaveBeenCalledOnce()); + + transport.history = historyOf(["run the tool", "reply pong"]); + transport.resolveTurn?.({ text: "pong", cancelled: false }); + await Promise.resolve(); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "too late" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + + accepted.resolve(undefined); + await expect(steering).resolves.toEqual({ ok: true, value: { accepted: true } }); + await vi.waitFor(() => { + expect(hostEvents(outputs).some((event) => event.type === "turn.completed")).toBe(true); + }); + expect(hostEvents(outputs).find((event) => event.type === "turn.completed")).toMatchObject({ + turnId, + nativeTurnRef: { nativeTurnKey: "user-1" }, + outcome: { status: "succeeded", checkpoint: { checkpointId: "user-1" } }, + }); + expect(transport.steer).toHaveBeenCalledOnce(); + }); + + it("preserves a completed Turn before faulting a runtime retired by late steering", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + const receipt = Promise.withResolvers(); + transport.steer.mockReturnValueOnce(receipt.promise); + const steering = session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "too late to queue" }], + }); + const rejection = expect(steering).rejects.toBeInstanceOf(PiRpcSteerTargetEndedError); + transport.resolveTurn?.({ text: "original answer", cancelled: false }); + const history = historyOf(["run the tool"]); + const late = new PiRpcSteerTargetEndedError(history); + const read = vi.spyOn(transport, "getEntries").mockRejectedValue(new Error("Runtime closed")); + receipt.reject(late); + await rejection; + await vi.waitFor(() => + expect(hostEvents(outputs).some((event) => event.type === "session.faulted")).toBe(true), + ); + const events = hostEvents(outputs); + expect(events.find((event) => event.type === "turn.completed")).toMatchObject({ + turnId, + nativeTurnRef: { nativeTurnKey: "user-1" }, + outcome: { status: "succeeded" }, + }); + expect(events.filter((event) => event.type === "turn.completed")).toHaveLength(1); + expect(events.findIndex((event) => event.type === "session.faulted")).toBeGreaterThan( + events.findIndex((event) => event.type === "turn.completed"), + ); + expect(read).not.toHaveBeenCalled(); + expect(transport.abort).not.toHaveBeenCalled(); + expect(userMessageEvents(outputs)).toEqual([]); + await expect( + session.execute({ + type: "turn.start", + turnId: hostTurnIdSchema.parse("after-runtime-retirement"), + input: [{ type: "text", text: "must reopen the native history first" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + }); + + it("keeps the original prompt when one steer adds exactly one user entry", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + await session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }); + transport.history = historyOf(["run the tool", "reply pong"]); + transport.resolveTurn?.({ text: "pong", cancelled: false }); + await vi.waitFor(() => { + expect(hostEvents(outputs).some((event) => event.type === "turn.completed")).toBe(true); + }); + expect(hostEvents(outputs).find((event) => event.type === "turn.completed")).toMatchObject({ + nativeTurnRef: { nativeTurnKey: "user-1" }, + outcome: { status: "succeeded" }, + }); + }); + + it("still fails when the new user entry count does not match accepted steers", async () => { + const { session, transports, outputs } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + await session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }); + transport.history = historyOf(["run the tool"]); + transport.resolveTurn?.({ text: "pong", cancelled: false }); + await vi.waitFor(() => { + expect(hostEvents(outputs).some((event) => event.type === "turn.completed")).toBe(true); + }); + expect(hostEvents(outputs).find((event) => event.type === "turn.completed")).toMatchObject({ + outcome: { + status: "failed", + error: { code: "protocolError", message: expect.stringContaining("exactly 2 is required") }, + }, + }); + + const extra = await openSession(); + const extraTurn = await startTurn(extra.session, "turn-extra"); + const extraTransport = extra.transports[0]; + if (!extraTransport) throw new Error("Missing transport"); + extraTransport.history = historyOf(["run the tool", "other client"]); + extraTransport.resolveTurn?.({ text: "pong", cancelled: false }); + await vi.waitFor(() => { + expect(hostEvents(extra.outputs).some((event) => event.type === "turn.completed")).toBe(true); + }); + expect( + hostEvents(extra.outputs).find((event) => event.type === "turn.completed"), + ).toMatchObject({ + turnId: extraTurn, + outcome: { + status: "failed", + error: { code: "protocolError", message: expect.stringContaining("exactly 1 is required") }, + }, + }); + }); + + it("stops offering steer when the last root tool ends", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + transport.onEvent?.({ + type: "tool.completed", + callId: "call-1", + toolName: "bash", + result: "done", + isError: false, + }); + expect(session.capabilities.steer).toBe(false); + await expect( + session.execute({ type: "turn.steer", turnId, input: [{ type: "text", text: "late" }] }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + expect(transport.steer).not.toHaveBeenCalled(); + }); + + it("rejects new input while cancellation is pending", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + transport.abort.mockImplementationOnce(async () => undefined); + await session.execute({ type: "turn.cancel", turnId }); + expect(session.capabilities.steer).toBe(false); + await expect( + session.execute({ type: "turn.steer", turnId, input: [{ type: "text", text: "late" }] }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + expect(transport.steer).not.toHaveBeenCalled(); + }); + + it("does not turn a lost native receipt into a safe-to-resend refusal", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + if (!transport) throw new Error("Missing transport"); + transport.steer.mockRejectedValueOnce(new Error("steer receipt timed out")); + await expect( + session.execute({ type: "turn.steer", turnId, input: [{ type: "text", text: "once" }] }), + ).rejects.toThrow("steer receipt timed out"); + expect(transport.steer).toHaveBeenCalledOnce(); + expect(transport.abort).not.toHaveBeenCalled(); + }); + + it("rejects an inactive turn, an empty steer, a native rejection, and a closed session", async () => { + const { session, transports } = await openSession(); + const turnId = await startTurn(session); + const transport = transports[0]; + await expect( + session.execute({ + type: "turn.steer", + turnId: hostTurnIdSchema.parse("other-turn"), + input: [{ type: "text", text: "reply pong" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidRequest" } }); + expect(transport?.steer).not.toHaveBeenCalled(); + transport?.steer.mockRejectedValueOnce( + new PiRpcSteerRejectedError("Pi RPC steer was not accepted"), + ); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }), + ).resolves.toMatchObject({ + ok: false, + error: { code: "invalidState", message: "Pi RPC steer was not accepted" }, + }); + await session.close(); + await expect( + session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "reply pong" }], + }), + ).resolves.toMatchObject({ ok: false, error: { code: "invalidState" } }); + }); +}); + +describe("Pi RPC steer", () => { + it.each(["missing", "handled"])( + "does not classify a %s native receipt as rejection", + async (receipt) => { + const fake = new FakePiProcess(); + fake.steerReceipt = receipt; + const rpc = new PiRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 50, closeTimeoutMs: 500 }, + { spawn: () => fake as unknown as ChildProcessWithoutNullStreams }, + ); + await rpc.start(); + const running = rpc.runTurn("run the tool", () => undefined).catch(() => undefined); + const failure = await rpc.steer("once").catch((error: unknown) => error); + expect(failure).toBeInstanceOf(Error); + expect(failure).not.toBeInstanceOf(PiRpcSteerRejectedError); + expect(fake.steered).toEqual(["once"]); + await rpc.close(); + await running; + }, + ); + + it("sends steer and ignores the user echo", async () => { + const fake = new FakePiProcess(); + const processAdapter: PiRpcProcessAdapter = { + spawn: () => fake as unknown as ChildProcessWithoutNullStreams, + }; + const rpc = new PiRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 2_000, closeTimeoutMs: 500 }, + processAdapter, + ); + const events: PiTurnEvent[] = []; + await rpc.start(); + const running = rpc.runTurn("run the tool", (event) => events.push(event)); + await vi.waitFor(() => { + expect(events.some((event) => event.type === "tool.completed")).toBe(true); + }); + await rpc.steer("reply pong"); + await expect(running).resolves.toMatchObject({ text: "pong", cancelled: false }); + expect(fake.steered).toEqual(["reply pong"]); + const text = events.flatMap((event) => (event.type === "text.delta" ? [event.delta] : [])); + expect(text.join("")).toBe("pong"); + await rpc.close(); + }); + + it("closes a late native queue without reporting acceptance or corrupting the completed run", async () => { + const fake = new FakePiProcess(); + fake.steerReceipt = "after-settled"; + const rpc = new PiRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 2_000, closeTimeoutMs: 500 }, + { spawn: () => fake as unknown as ChildProcessWithoutNullStreams }, + ); + await rpc.start(); + const running = rpc.runTurn("run the tool", () => undefined); + const failure = await rpc.steer("late input").catch((error: unknown) => error); + expect(failure).toBeInstanceOf(PiRpcSteerTargetEndedError); + expect(failure).not.toBeInstanceOf(PiRpcSteerRejectedError); + expect((failure as PiRpcSteerTargetEndedError).history).toEqual(historyOf(["run the tool"])); + await expect(running).resolves.toMatchObject({ text: "original answer", cancelled: false }); + expect(fake.stdin.writable).toBe(false); + expect(fake.exitCode).toBe(0); + await expect(rpc.runTurn("do not consume the stale queue", () => undefined)).rejects.toThrow( + /unavailable/, + ); + await rpc.close(); + }); + + it("does not dispatch native steering without an active requested run", async () => { + const fake = new FakePiProcess(); + const rpc = new PiRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 2_000, closeTimeoutMs: 500 }, + { spawn: () => fake as unknown as ChildProcessWithoutNullStreams }, + ); + await rpc.start(); + await expect(rpc.steer("idle input")).rejects.toBeInstanceOf(PiRpcSteerRejectedError); + expect(fake.steered).toEqual([]); + await rpc.close(); + }); + + it("rejects a steer the native session does not accept", async () => { + const fake = new FakePiProcess(); + fake.rejectSteer = "Agent is idle"; + const rpc = new PiRpcSession( + { cwd: process.cwd(), commandTimeoutMs: 2_000, closeTimeoutMs: 500 }, + { spawn: () => fake as unknown as ChildProcessWithoutNullStreams }, + ); + await rpc.start(); + const running = rpc.runTurn("run the tool", () => undefined).catch(() => undefined); + await expect(rpc.steer("reply pong")).rejects.toBeInstanceOf(PiRpcSteerRejectedError); + await rpc.close(); + await running; + }); +}); + +class FakePiProcess extends EventEmitter { + readonly stdin = new PassThrough(); + readonly stdout = new PassThrough(); + readonly stderr = new PassThrough(); + readonly pid = 42_001; + exitCode: number | null = null; + signalCode: NodeJS.Signals | null = null; + readonly steered: string[] = []; + rejectSteer: string | null = null; + steerReceipt = "queued"; + history: PiSessionHistory = { entries: [], leafId: null }; + #buffer = Buffer.alloc(0); + + constructor() { + super(); + this.stdin.on("data", (chunk: Buffer) => this.#push(chunk)); + this.stdin.once("finish", () => { + this.exitCode = 0; + this.stdout.end(); + this.stderr.end(); + this.emit("exit", 0, null); + }); + queueMicrotask(() => this.emit("spawn")); + } + + #push(chunk: Buffer): void { + this.#buffer = Buffer.concat([this.#buffer, chunk]); + let newline = this.#buffer.indexOf(0x0a); + while (newline >= 0) { + const line = this.#buffer.subarray(0, newline).toString("utf8"); + this.#buffer = this.#buffer.subarray(newline + 1); + this.#onCommand(JSON.parse(line) as Record); + newline = this.#buffer.indexOf(0x0a); + } + } + + #onCommand(command: Record): void { + if (command.type === "get_entries") { + this.#respond(command, this.history); + return; + } + if (command.type === "prompt") { + this.history = historyOf([String(command.message)]); + this.#respond(command); + this.#output({ + type: "tool_execution_start", + toolCallId: "call-1", + toolName: "bash", + args: { command: "echo hello" }, + }); + this.#output({ + type: "tool_execution_end", + toolCallId: "call-1", + toolName: "bash", + result: { output: "hello" }, + isError: false, + }); + return; + } + if (command.type === "steer") { + if (this.rejectSteer) { + this.#output({ + id: command.id, + type: "response", + command: "steer", + success: false, + error: this.rejectSteer, + }); + return; + } + if (typeof command.message === "string") this.steered.push(command.message); + if (this.steerReceipt === "missing") return; + if (this.steerReceipt === "handled") { + this.#respond(command, { disposition: "handled" }); + return; + } + if (this.steerReceipt === "after-settled") { + this.#output({ + type: "message_end", + message: { + role: "assistant", + content: [{ type: "text", text: "original answer" }], + stopReason: "stop", + }, + }); + this.#output({ type: "agent_settled" }); + setTimeout(() => this.#respond(command, { disposition: "queued" }), 10); + return; + } + this.#respond(command); + const user = { role: "user", content: [{ type: "text", text: command.message }] }; + const assistant = { + role: "assistant", + content: [{ type: "text", text: "pong" }], + stopReason: "stop", + }; + this.#output({ type: "message_start", message: user }); + this.#output({ type: "message_end", message: user }); + this.#output({ type: "message_end", message: assistant }); + this.#output({ type: "agent_settled" }); + return; + } + this.#respond(command, { + sessionId: "pi-session", + sessionFile: null, + model: { provider: "synthetic", id: "model" }, + thinkingLevel: "off", + isStreaming: false, + }); + } + + #respond(command: Record, data?: unknown): void { + this.#output({ + id: command.id, + type: "response", + command: command.type, + success: true, + ...(data === undefined ? {} : { data }), + }); + } + + #output(value: unknown): void { + this.stdout.write(`${JSON.stringify(value)}\n`); + } +} diff --git a/packages/host-runtime/src/external-turn-steering.ts b/packages/host-runtime/src/external-turn-steering.ts index 2d6a61a5be..b055b9a552 100644 --- a/packages/host-runtime/src/external-turn-steering.ts +++ b/packages/host-runtime/src/external-turn-steering.ts @@ -271,20 +271,19 @@ export class ExternalTurnSteering { if (key) { const receipt = { fingerprint, result, settled: false }; this.#receipts.set(key, receipt); - void result.then( - () => { - receipt.settled = true; - // Bound completed receipts; never evict an in-flight delivery reservation. - while (this.#receipts.size > 128) { - const oldest = [...this.#receipts].find(([, entry]) => entry.settled)?.[0]; - if (oldest === undefined) break; - this.#receipts.delete(oldest); - } - }, - () => { - this.#receipts.delete(key); - }, - ); + const settle = (): void => { + receipt.settled = true; + // Bound completed receipts; never evict an in-flight delivery reservation. + while (this.#receipts.size > 128) { + const oldest = [...this.#receipts].find(([, entry]) => entry.settled)?.[0]; + if (oldest === undefined) break; + this.#receipts.delete(oldest); + } + }; + void result.then(settle, (error: unknown) => { + if (error instanceof ExternalSteerOutcomeUnknownError) settle(); + else this.#receipts.delete(key); + }); } return result; } catch (error) { diff --git a/packages/host-runtime/test/app-server-host.external-queue.test.ts b/packages/host-runtime/test/app-server-host.external-queue.test.ts index 5adcd1aa9b..ea313d4493 100644 --- a/packages/host-runtime/test/app-server-host.external-queue.test.ts +++ b/packages/host-runtime/test/app-server-host.external-queue.test.ts @@ -296,6 +296,153 @@ describe("External Thread follow-up queue", () => { } }); + it("does not automatically replay unknown steering after the original or a later Turn", async () => { + const fixture = createFixture(); + fixture.adapter.supportsSteer = true; + try { + const threadId = await startPiThread(fixture); + const turnId = await startPiTurn(fixture, threadId); + const session = fixture.adapter.sessions[0]; + if (!session) throw new Error("Fake Session was not opened"); + const execute = vi.spyOn(session, "execute"); + writeRequest(fixture.desktopInput, { + id: 10, + method: "thread/queue/add", + params: { threadId, input: text("uncertain input"), clientUserMessageId: "uncertain" }, + }); + const added = await fixture.collector.waitFor((message) => requestId(message, 10)); + const queued = (added.result as { queuedSubmission: JsonObject & { id: string } }) + .queuedSubmission; + execute.mockRejectedValueOnce(new Error("Native input outcome is unknown")); + writeRequest(fixture.desktopInput, { + id: 20, + method: "turn/steer", + params: { + threadId, + expectedTurnId: turnId, + input: text("uncertain input"), + clientUserMessageId: "uncertain", + codexhostSteerDelivery: "activeTurn", + }, + }); + await expect( + fixture.collector.waitFor((message) => requestId(message, 20)), + ).resolves.toMatchObject({ error: { code: -32074 } }); + session.succeedTurn(); + await fixture.collector.waitFor((message) => turnEvent(message, "turn/completed", turnId)); + await new Promise((resolve) => setImmediate(resolve)); + writeRequest(fixture.desktopInput, { + id: 21, + method: "thread/queue/list", + params: { threadId }, + }); + await expect( + fixture.collector.waitFor((message) => requestId(message, 21)), + ).resolves.toMatchObject({ result: { data: [queued], nextCursor: null } }); + writeRequest(fixture.desktopInput, { + id: 22, + method: "turn/start", + params: { threadId, input: text("explicit follow-up"), clientUserMessageId: "follow-up" }, + }); + const started = await fixture.collector.waitFor((message) => requestId(message, 22)); + expect(started).not.toHaveProperty("error"); + const followUpId = (started.result as { turn: { id: string } }).turn.id; + session.succeedTurn(); + await fixture.collector.waitFor((message) => + turnEvent(message, "turn/completed", followUpId), + ); + await new Promise((resolve) => setImmediate(resolve)); + expect(execute).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "turn.start", input: text("uncertain input") }), + ); + writeRequest(fixture.desktopInput, { + id: 24, + method: "thread/queue/add", + params: { threadId, input: text("another queued input"), clientUserMessageId: "another" }, + }); + const another = await fixture.collector.waitFor((message) => requestId(message, 24)); + const anotherItem = (another.result as { queuedSubmission: JsonObject & { id: string } }) + .queuedSubmission; + writeRequest(fixture.desktopInput, { + id: 25, + method: "thread/queue/start", + params: { threadId, queuedSubmissionId: anotherItem.id }, + }); + const anotherStarted = await fixture.collector.waitFor((message) => requestId(message, 25)); + expect(anotherStarted).not.toHaveProperty("error"); + const anotherTurnId = (anotherStarted.result as { turn: { id: string } }).turn.id; + session.succeedTurn(); + await fixture.collector.waitFor((message) => + turnEvent(message, "turn/completed", anotherTurnId), + ); + await new Promise((resolve) => setImmediate(resolve)); + expect(execute).not.toHaveBeenCalledWith( + expect.objectContaining({ type: "turn.start", input: text("uncertain input") }), + ); + writeRequest(fixture.desktopInput, { + id: 23, + method: "thread/queue/start", + params: { threadId, queuedSubmissionId: queued.id }, + }); + await expect( + fixture.collector.waitFor((message) => requestId(message, 23)), + ).resolves.toHaveProperty("result.turn.id"); + expect(execute).toHaveBeenCalledWith( + expect.objectContaining({ type: "turn.start", input: text("uncertain input") }), + ); + } finally { + await stopFixture(fixture); + } + }); + + it.each(["completed", "interrupted", "failed"])( + "handles a queue/add arriving after the Turn is already %s", + async (outcome) => { + const fixture = createFixture(); + try { + const threadId = await startPiThread(fixture); + const turnId = await startPiTurn(fixture, threadId); + const session = fixture.adapter.sessions[0]; + if (!session) throw new Error("Fake Session was not opened"); + const execute = vi.spyOn(session, "execute"); + if (outcome === "completed") session.succeedTurn(); + else if (outcome === "failed") + session.failTurn({ code: "internalError", message: "Fixture failure", retryable: false }); + else { + writeRequest(fixture.desktopInput, { + id: 9, + method: "turn/interrupt", + params: { threadId, turnId }, + }); + await fixture.collector.waitFor((message) => requestId(message, 9)); + session.completeCancellation(); + } + await fixture.collector.waitFor((message) => turnEvent(message, "turn/completed", turnId)); + // Let the completion handler finish its empty-queue check before the delayed request. + await new Promise((resolve) => setImmediate(resolve)); + writeRequest(fixture.desktopInput, { + id: 10, + method: "thread/queue/add", + params: { threadId, input: text("late follow-up"), clientUserMessageId: "late-message" }, + }); + const added = await fixture.collector.waitFor((message) => requestId(message, 10)); + expect(added).not.toHaveProperty("error"); + if (outcome === "completed") { + await vi.waitFor(() => + expect(execute).toHaveBeenCalledWith( + expect.objectContaining({ type: "turn.start", input: text("late follow-up") }), + ), + ); + } else { + await new Promise((resolve) => setImmediate(resolve)); + expect(execute).not.toHaveBeenCalledWith(expect.objectContaining({ type: "turn.start" })); + } + } finally { + await stopFixture(fixture); + } + }, + ); + it("restores and pauses the only queued submission when its replacement fails to start", async () => { const fixture = createFixture(); try { diff --git a/packages/host-runtime/test/external-thread-queue.test.ts b/packages/host-runtime/test/external-thread-queue.test.ts index 4b5c0b4c51..cfa539ff36 100644 --- a/packages/host-runtime/test/external-thread-queue.test.ts +++ b/packages/host-runtime/test/external-thread-queue.test.ts @@ -198,3 +198,29 @@ it("rejects an ambiguous client identity after its add receipt expires", () => { expect(() => queue.takeSteered("thread", "same-message", input)).toThrow("changed"); expect(queue.list("thread", {}).data).toHaveLength(2); }); + +it("keeps unknown input visible but blocks automatic replay across edits and later Turns", async () => { + const queue = new ExternalThreadQueue(); + const input = [{ type: "text", text: "uncertain" }]; + const target = await queue.add("thread", input, "uncertain-message", undefined, "source"); + const taken = queue.takeSteered("thread", "uncertain-message", input); + if (!taken) throw new Error("Missing reservation"); + taken.automaticStartAllowed = false; + queue.restore("thread", taken); + const next = await queue.add("thread", [{ type: "text", text: "next" }], "next-message"); + expect(queue.snapshot("thread").items).toEqual([target, next]); + expect(queue.canStartAutomatically("thread")).toBe(false); + expect(await queue.update("thread", target.id, input)).toEqual(target); + expect(queue.canStartAutomatically("thread")).toBe(false); + const explicitNext = queue.takeSteered("thread", "next-message", next.input); + expect(explicitNext?.automaticStartAllowed).toBe(true); + expect(queue.canStartAutomatically("thread")).toBe(false); + const retry = queue.takeSteered("thread", "uncertain-message", input); + if (!retry) throw new Error("Missing explicit retry"); + expect(retry.sourceHostId).toBe("source"); + queue.restore("thread", retry); + expect(queue.canStartAutomatically("thread")).toBe(false); + expect(await queue.delete("thread", target.id)).toBe(true); + await queue.add("thread", input, "fresh-message"); + expect(queue.canStartAutomatically("thread")).toBe(true); +}); diff --git a/packages/host-runtime/test/external-turn-steering.test.ts b/packages/host-runtime/test/external-turn-steering.test.ts index 0bfca9e8be..6eaad16b51 100644 --- a/packages/host-runtime/test/external-turn-steering.test.ts +++ b/packages/host-runtime/test/external-turn-steering.test.ts @@ -11,8 +11,10 @@ import { harnessIdSchema, hostTurnIdSchema } from "@codexhost/shared-contracts"; import { ExternalSteerNotTakenError, + ExternalSteerOutcomeUnknownError, ExternalTurnSteering, externalSteeringDelivery, + insertNativeSteer, } from "../src/external-turn-steering.js"; type Cancel = (command: TurnCancelCommand) => Promise>; @@ -338,6 +340,36 @@ describe("Host native steering", () => { }); expect(f.insert).toHaveBeenCalledTimes(2); }); + it("retains an unknown receipt so retrying the same identity cannot send it twice", async () => { + const f = fixture(); + f.thread.session.capabilities.steer = true; + const unknown = new ExternalSteerOutcomeUnknownError(new Error("native reply lost")); + f.insert.mockRejectedValueOnce(unknown); + const first = f.coordinator.run(f.thread, f.params, f.start, f.insert); + await expect(first).rejects.toBe(unknown); + const retry = f.coordinator.run(f.thread, f.params, f.start, f.insert); + expect(retry).toBe(first); + await expect(retry).rejects.toBe(unknown); + expect(f.insert).toHaveBeenCalledOnce(); + expect(f.start).not.toHaveBeenCalled(); + expect(f.execute).not.toHaveBeenCalled(); + }); + + it("classifies an execute exception as an unknown outcome and releases projection ordering", async () => { + const thread = { + pendingSteerProjections: new Set>(), + steeredTurnIds: new Set>(), + session: { execute: vi.fn().mockRejectedValue(new Error("native reply lost")) }, + }; + const host = { project: vi.fn(), diagnose: vi.fn() }; + await expect( + insertNativeSteer(thread, { turnId: hostTurnIdSchema.parse("old"), text: "input" }, host), + ).rejects.toBeInstanceOf(ExternalSteerOutcomeUnknownError); + expect(thread.pendingSteerProjections.size).toBe(0); + expect(thread.steeredTurnIds.size).toBe(0); + expect(host.project).not.toHaveBeenCalled(); + }); + it("starts the next Turn when untaken input arrives after the active Turn ended", async () => { const f = fixture(); f.thread.session.capabilities.steer = true; diff --git a/tools/gate-steer/adapter.real.test.mjs b/tools/gate-steer/adapter.real.test.mjs new file mode 100644 index 0000000000..0ed6b84315 --- /dev/null +++ b/tools/gate-steer/adapter.real.test.mjs @@ -0,0 +1,315 @@ +import assert from "node:assert/strict"; +import { randomUUID } from "node:crypto"; +import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { describe, it } from "vitest"; +import { observeSteerDelivery } from "./delivery.mjs"; +import { OmpAdapter } from "@codexhost/adapter-omp"; +import { PiAdapter } from "@codexhost/adapter-pi"; + +// Opt in per Harness. Uses existing authentication and the selected Model, real model quota, +// and native Session storage. Only the synthetic working directory is removed after the test. +const selected = new Set((process.env.CODEXHOST_STEER_REAL_HARNESSES ?? "").split(",")); +const scenarios = ( + process.env.CODEXHOST_STEER_REAL_SCENARIOS ?? "single,double,near-terminal" +).split(","); +for (const name of scenarios) + assert.ok(["single", "double", "near-terminal"].includes(name), `Unknown scenario: ${name}`); + +const adapters = { + omp: () => new OmpAdapter(), + pi: () => new PiAdapter(), +}; + +async function bounded(promise, label, timeoutMs = 90_000) { + let timer; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error(`Timed out: ${label}`)), timeoutMs); + }), + ]); + } finally { + clearTimeout(timer); + } +} + +function succeeded(result) { + assert.equal(result.ok, true, JSON.stringify(result.ok ? {} : result.error)); + return result.value; +} + +class Collector { + outputs = []; + failure; + constructor(session) { + this.done = (async () => { + for await (const output of session.outputs) { + this.outputs.push({ ...output, observedAtMs: Date.now() }); + if (output.kind === "interaction") { + const interaction = output.interaction; + assert.equal(interaction.type, "approval", "Unexpected native question"); + const action = interaction.actions.find(({ effect }) => effect === "allowOnce"); + assert.ok(action, "Native approval has no allow-once action"); + succeeded( + await session.execute({ + type: "interaction.respond", + interactionId: interaction.interactionId, + response: { type: "approval", actionId: action.id }, + }), + ); + } + } + })().catch((error) => { + this.failure = error; + }); + } + + events(from) { + return this.outputs + .slice(from) + .flatMap((output) => (output.kind === "event" ? [output.event] : [])); + } + + async wait(from, predicate, label) { + const deadline = Date.now() + 90_000; + while (Date.now() < deadline) { + if (this.failure) throw this.failure; + const events = this.events(from); + const found = events.find(predicate); + if (found) return found; + const fault = events.find(({ type }) => type === "session.faulted"); + if (fault) throw new Error(JSON.stringify(fault.error)); + const terminal = events.find(({ type }) => type === "turn.completed"); + if (terminal) + throw new Error(`Turn ended before ${label}: ${JSON.stringify(terminal.outcome)}`); + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw new Error(`Timed out: ${label}`); + } +} + +async function scenario(session, collector, receipt) { + const { name } = receipt; + const from = collector.outputs.length; + const turnId = randomUUID(); + const inputMarker = `START_${randomUUID().replaceAll("-", "")}`; + const toolMarker = `TOOL_DONE_${randomUUID().replaceAll("-", "")}`; + const answerMarker = `STEER_${randomUUID().replaceAll("-", "")}`; + const prompt = `${inputMarker}. Run exactly one foreground shell command: sleep 8; printf '${toolMarker}\\n'. Wait for it to finish. Then follow the latest user instruction. Do not read or modify files or use other tools.`; + Object.assign(receipt, { turnId, inputMarker, toolMarker, answerMarker, steers: [] }); + succeeded( + await bounded( + session.execute({ type: "turn.start", turnId, input: [{ type: "text", text: prompt }] }), + "turn.start", + ), + ); + const tool = await collector.wait( + from, + (event) => + event.type === "item.started" && + (event.item.type === "commandExecution" || event.item.type === "toolExecution") && + JSON.stringify(event.item).includes(toolMarker), + "foreground sleep tool", + ); + assert.equal(session.capabilities.steer, true, "Adapter did not confirm the active steer window"); + + for (const input of [ + { turnId: randomUUID(), text: "Do not deliver this wrong-target message." }, + { turnId, text: "" }, + ]) { + const rejected = await session.execute({ + type: "turn.steer", + turnId: input.turnId, + input: [{ type: "text", text: input.text }], + }); + assert.equal(rejected.ok, false, "Invalid steer was accepted"); + assert.equal(rejected.error.code, input.text ? "invalidState" : "invalidRequest"); + } + + const toolDone = () => + collector.wait( + from, + (event) => event.type === "item.completed" && event.snapshot.item.itemId === tool.item.itemId, + "foreground tool completion", + ); + if (name === "near-terminal") await toolDone(); + const count = name === "double" ? 2 : 1; + for (let index = 0; index < count; index += 1) { + const text = + index === count - 1 + ? `After the current shell command finishes, reply with exactly ${answerMarker}. Do not cancel it or run more tools.` + : "Keep waiting for the current shell command. A second instruction follows; do not run more tools."; + const result = await bounded( + session.execute({ type: "turn.steer", turnId, input: [{ type: "text", text }] }), + "turn.steer receipt", + 30_000, + ); + receipt.steers.push({ ...result, observedAtMs: Date.now() }); + if (name === "near-terminal" && !result.ok) { + assert.equal(result.error.code, "invalidState"); + } else succeeded(result); + } + const completedTool = await toolDone(); + assert.equal(completedTool.snapshot.outcome.status, "succeeded", "Steering interrupted the tool"); + assert.ok( + JSON.stringify(completedTool.snapshot.item).includes(toolMarker), + "Tool did not finish its command", + ); + const terminal = await collector.wait( + from, + (event) => event.type === "turn.completed" && event.turnId === turnId, + "Turn completion", + ); + assert.equal(terminal.outcome.status, "succeeded", JSON.stringify(terminal.outcome)); + if (name === "near-terminal") { + receipt.delivery = await observeSteerDelivery( + () => collector.events(from), + answerMarker, + turnId, + ); + if (collector.failure) throw collector.failure; + } + const events = collector.events(from); + assert.equal( + events.filter( + (event) => + event.type === "turn.started" && (name !== "near-terminal" || event.turnId === turnId), + ).length, + 1, + "Steer opened a second Host Turn", + ); + assert.equal( + events.filter( + (event) => + event.type === "turn.completed" && (name !== "near-terminal" || event.turnId === turnId), + ).length, + 1, + ); + const replies = events + .filter( + (event) => event.type === "item.completed" && event.snapshot.item.type === "agentMessage", + ) + .map((event) => event.snapshot.item.text) + .join("\n"); + if (name !== "near-terminal" && receipt.steers.every((result) => result.ok)) { + assert.ok(replies.includes(answerMarker), "Accepted input was not used in the same Turn"); + } + const snapshot = succeeded(await session.readSnapshot()); + receipt.history = snapshot; + if (name === "near-terminal") { + const expected = receipt.steers.every((result) => result.ok) ? 1 : 0; + assert.equal( + receipt.delivery.replies.length, + expected, + "Accepted input must execute exactly once in the observation window", + ); + for (const reply of receipt.delivery.replies) { + assert.equal(reply.placement, "targetTurn", "Accepted steer ran outside its target Turn"); + assert.equal( + reply.outcome?.status, + "succeeded", + "The delivery Turn has not completed successfully", + ); + } + } + const original = snapshot.turns.find((turn) => + turn.input.some(({ text }) => text.includes(inputMarker)), + ); + assert.ok(original, "Original input missing from native history"); + assert.deepEqual( + terminal.nativeTurnRef, + original.nativeTurnRef, + "Turn identity bound to steered input", + ); + const late = await session.execute({ + type: "turn.steer", + turnId, + input: [{ type: "text", text: "Do not deliver after completion." }], + }); + assert.equal(late.ok, false, "Completed Turn accepted a steer"); + assert.equal(late.error.code, "invalidState"); + Object.assign(receipt, { status: "passed", nativeTurnRef: terminal.nativeTurnRef }); +} + +for (const [id, createAdapter] of Object.entries(adapters)) { + describe.runIf(selected.has(id))(`${id} real native steering`, () => { + it("accepts single and consecutive input, exercises terminal timing and rejects invalid targets", async (context) => { + const cwd = await mkdtemp(path.join(os.tmpdir(), `codexhost-steer-${id}-`)); + const reportDirectory = path.resolve(".codexhost", "native-steer-real", id); + await mkdir(reportDirectory, { recursive: true }); + const report = { + harness: id, + cwd, + startedAt: new Date().toISOString(), + status: "running", + scenarios: [], + }; + const adapter = createAdapter(); + let collector; + try { + const inspection = await bounded(adapter.inspect({ cwd }), "Harness inspect", 45_000); + if (inspection.status !== "ready") { + report.status = "blocked"; + report.error = inspection.error; + context.skip(JSON.stringify(inspection.error)); + } + const requestedModel = process.env.CODEXHOST_STEER_REAL_MODEL; + const selectedModel = requestedModel + ? inspection.catalog.models.find( + (model) => model.nativeId === requestedModel || model.label === requestedModel, + ) + : undefined; + if (requestedModel) + assert.ok(selectedModel, `Model not in native catalog: ${requestedModel}`); + report.model = selectedModel?.ref ?? inspection.catalog.defaultModel; + report.requestedModel = requestedModel; + const opened = await bounded( + adapter.open({ + kind: "create", + cwd, + ...(selectedModel ? { model: selectedModel.ref } : {}), + }), + "Harness open", + 45_000, + ); + if (!opened.ok) { + report.status = "blocked"; + report.error = opened.error; + context.skip(JSON.stringify(opened.error)); + } + const session = opened.value; + collector = new Collector(session); + for (const name of scenarios) { + report.activeScenario = name; + const receipt = { name, status: "running" }; + report.scenarios.push(receipt); + await scenario(session, collector, receipt); + } + report.status = "passed"; + } catch (error) { + if (report.status !== "blocked") { + report.status = "failed"; + const active = report.scenarios.at(-1); + if (active?.status === "running") active.status = "failed"; + report.error = error instanceof Error ? error.message : String(error); + } + throw error; + } finally { + try { + await bounded(adapter.close(), "Adapter close", 30_000); + if (collector) await bounded(collector.done, "Output consumer close", 5_000); + } finally { + report.outputs = collector?.outputs ?? []; + await writeFile( + path.join(reportDirectory, `${Date.now()}.json`), + `${JSON.stringify(report, null, 2)}\n`, + ); + await rm(cwd, { recursive: true, force: true }); + } + } + }, 360_000); + }); +} diff --git a/tools/gate-steer/delivery.mjs b/tools/gate-steer/delivery.mjs new file mode 100644 index 0000000000..458c58be02 --- /dev/null +++ b/tools/gate-steer/delivery.mjs @@ -0,0 +1,27 @@ +/** Observe the entire window, including after a following Turn completes, to detect duplicates. */ +export async function observeSteerDelivery(events, marker, targetTurnId, durationMs = 30_000) { + const startedAtMs = Date.now(); + await new Promise((resolve) => setTimeout(resolve, durationMs)); + const observed = events(); + const replies = observed.flatMap((event) => { + if ( + event.type !== "item.completed" || + event.snapshot.item.type !== "agentMessage" || + !event.snapshot.item.text.includes(marker) + ) + return []; + const terminal = observed.find( + (candidate) => candidate.type === "turn.completed" && candidate.turnId === event.turnId, + ); + return [ + { + turnId: event.turnId, + itemId: event.snapshot.item.itemId, + placement: event.turnId === targetTurnId ? "targetTurn" : "followingTurn", + outcome: terminal?.outcome, + nativeTurnRef: terminal?.nativeTurnRef, + }, + ]; + }); + return { startedAtMs, endedAtMs: Date.now(), replies }; +} diff --git a/tools/gate-steer/delivery.test.mjs b/tools/gate-steer/delivery.test.mjs new file mode 100644 index 0000000000..6e42b6fbc9 --- /dev/null +++ b/tools/gate-steer/delivery.test.mjs @@ -0,0 +1,54 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { observeSteerDelivery } from "./delivery.mjs"; + +afterEach(() => vi.useRealTimers()); +const answer = (turnId, itemId = "reply") => ({ + type: "item.completed", + turnId, + snapshot: { item: { type: "agentMessage", itemId, text: "STEER_MARKER" } }, +}); +const terminal = (turnId) => ({ type: "turn.completed", turnId, outcome: { status: "succeeded" } }); + +describe("native steer delivery observation", () => { + it("records a following-Turn delivery distinctly without stopping observation early", async () => { + vi.useFakeTimers(); + const events = [terminal("original")]; + const result = observeSteerDelivery(() => events, "STEER_MARKER", "original"); + await vi.advanceTimersByTimeAsync(10_000); + events.push(answer("next"), terminal("next")); + let finished = false; + void result.then(() => { + finished = true; + }); + await vi.advanceTimersByTimeAsync(19_999); + expect(finished).toBe(false); + await vi.advanceTimersByTimeAsync(1); + expect((await result).replies).toEqual([ + expect.objectContaining({ + turnId: "next", + placement: "followingTurn", + outcome: { status: "succeeded" }, + }), + ]); + }); + + it("retains duplicate replies across Turns and does not infer delivery from completion alone", async () => { + vi.useFakeTimers(); + const events = [answer("original"), terminal("original")]; + const result = observeSteerDelivery(() => events, "STEER_MARKER", "original"); + await vi.advanceTimersByTimeAsync(10_000); + events.push(answer("next"), terminal("next")); + await vi.advanceTimersByTimeAsync(20_000); + expect((await result).replies).toHaveLength(2); + const missing = observeSteerDelivery(() => [terminal("original")], "STEER_MARKER", "original"); + await vi.advanceTimersByTimeAsync(30_000); + expect((await missing).replies).toEqual([]); + }); + + it("does not treat an unfinished reply as a completed delivery", async () => { + vi.useFakeTimers(); + const result = observeSteerDelivery(() => [answer("original")], "STEER_MARKER", "original"); + await vi.advanceTimersByTimeAsync(30_000); + expect((await result).replies).toEqual([expect.objectContaining({ outcome: undefined })]); + }); +});