diff --git a/docs-site/src/content/docs/fr/reference/configuration/advisor.md b/docs-site/src/content/docs/fr/reference/configuration/advisor.md index 1fa62b02611..3c65d98c1d9 100644 --- a/docs-site/src/content/docs/fr/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/fr/reference/configuration/advisor.md @@ -54,7 +54,7 @@ Sur la page **Advisor**, choisissez le modèle dans le catalogue. Le sélecteur Gérez-le via la page **Advisor** du tableau de bord ou `ocx advisor status|on|off|set --model --effort --policy `. -Sans consentement courant, `ocx advisor on` n'active pas le partage inter-fournisseurs : il affiche cette divulgation et s'arrête. `ocx advisor on --ack-context-sharing` et `ocx advisor consent` enregistrent `v1`. `ocx advisor consent --revoke` retire le consentement et arrête immédiatement l'envoi. `ocx advisor set` n'accorde pas le consentement. La case du tableau de bord n'est pas précochée. +Sans consentement courant, `ocx advisor on` n'active pas le partage inter-fournisseurs : il affiche cette divulgation et s'arrête. `ocx advisor on --ack-context-sharing` et `ocx advisor consent` enregistrent `v1`. `ocx advisor consent --revoke` retire le consentement. Les consultations qui n'ont pas encore été envoyées sont bloquées, mais une consultation déjà envoyée ne peut pas être annulée, et le contexte déjà transmis hors du proxy ne peut pas être retiré. `ocx advisor set` n'accorde pas le consentement. La case du tableau de bord n'est pas précochée. ## Politiques @@ -125,6 +125,13 @@ codage, et une consultation ne change jamais le modèle principal de la session. - Les tours natifs OpenAI en passthrough (workers du pool ChatGPT) ne reçoivent pas l'outil synthétique ; le conseiller couvre les fournisseurs routés (traduits). La consultation preflight s'applique aux adaptateurs run-turn ; l'outil non. +- Les requêtes `/v1/messages` traduites (par exemple Claude Code routé vers un fournisseur + traduit) reçoivent l'outil synthétique et la consultation preflight comme `/v1/responses` et + `/v1/chat/completions`. Les mêmes règles de `policy` et de consentement au partage de contexte + s'appliquent, et les réglages sont lus dans la configuration active : une révocation du + consentement empêche donc les requêtes déjà en cours de lancer une consultation qui n'a pas + encore été envoyée. Elle n'annule pas une consultation déjà envoyée et ne retire pas le contexte + déjà sorti du proxy. - Pas de détection sémantique de blocage, de revue finale, de niveaux d’escalade, de conseillers multiples ou de vote. - Le registre de déduplication preflight vit dans le processus ; après un redémarrage du proxy, une tâche en cours peut recevoir une tentative preflight de plus. diff --git a/docs-site/src/content/docs/ja/reference/configuration/advisor.md b/docs-site/src/content/docs/ja/reference/configuration/advisor.md index 6193ce3b4dd..3a537f51df5 100644 --- a/docs-site/src/content/docs/ja/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/ja/reference/configuration/advisor.md @@ -46,7 +46,7 @@ ocx advisor set --context-mode question-only ダッシュボードの **Advisor** ページまたは `ocx advisor status|on|off|set --model --effort --policy ` で管理します。 -現在の同意がないとき、`ocx advisor on` はプロバイダー間送信を有効にしません。開示を表示して停止します。`ocx advisor on --ack-context-sharing` と `ocx advisor consent` が `v1` を記録します。`ocx advisor consent --revoke` は同意を消し、送信を直ちに止めます。`ocx advisor set` は同意を与えません。ダッシュボードの同意チェックは初期状態でオフです。 +現在の同意がないとき、`ocx advisor on` はプロバイダー間送信を有効にしません。開示を表示して停止します。`ocx advisor on --ack-context-sharing` と `ocx advisor consent` が `v1` を記録します。`ocx advisor consent --revoke` は同意を消します。まだディスパッチされていない相談はブロックされますが、すでにディスパッチされた相談は取り消せず、すでにプロキシを離れた文脈も取り戻せません。`ocx advisor set` は同意を与えません。ダッシュボードの同意チェックは初期状態でオフです。 ## ポリシー @@ -98,6 +98,7 @@ fetch ベースの Web 検索ターンでは、実際に適用される `tool_ch ## PR1 の制限 - ネイティブ OpenAI パススルーのターン（ChatGPT プールのワーカー）には合成ツールが注入されません。アドバイザーはルーティング（翻訳）プロバイダーを対象とします。preflight 相談は run-turn アダプターに適用されますが、ツールは適用されません。 +- 翻訳された `/v1/messages` リクエスト（例: 翻訳プロバイダーへルーティングした Claude Code）も、`/v1/responses` や `/v1/chat/completions` と同じく合成ツールと preflight 相談を受け取ります。`policy` と文脈共有の同意は同じ規則に従い、設定は現在の構成から読み取ります。そのため同意を取り消すと、進行中のリクエストがまだ送信されていない相談を開始することはブロックされますが、送信済みの相談は取り消されず、すでにプロキシを離れた文脈も取り戻せません。 - 意味的な停滞検出、完了前レビュー、段階的エスカレーション、複数専門家、投票は行いません。 - preflight の重複排除台帳はプロセス内です。プロキシ再起動後、進行中のタスクはもう一度 preflight 相談を受けることがあります。 diff --git a/docs-site/src/content/docs/ko/reference/configuration/advisor.md b/docs-site/src/content/docs/ko/reference/configuration/advisor.md index eed48010a75..269993bdf88 100644 --- a/docs-site/src/content/docs/ko/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/ko/reference/configuration/advisor.md @@ -46,7 +46,7 @@ ocx advisor set --context-mode question-only 대시보드 **Advisor** 페이지 또는 `ocx advisor status|on|off|set --model --effort --policy `로 관리합니다. -현재 동의가 없으면 `ocx advisor on`은 프로바이더 간 전송을 켜지 않습니다. 고지를 출력하고 멈춥니다. `ocx advisor on --ack-context-sharing`와 `ocx advisor consent`가 `v1`을 기록합니다. `ocx advisor consent --revoke`는 동의를 지우고 전송을 즉시 멈춥니다. `ocx advisor set`은 동의를 부여하지 않습니다. 대시보드의 동의 확인란은 미리 선택되지 않습니다. +현재 동의가 없으면 `ocx advisor on`은 프로바이더 간 전송을 켜지 않습니다. 고지를 출력하고 멈춥니다. `ocx advisor on --ack-context-sharing`와 `ocx advisor consent`가 `v1`을 기록합니다. `ocx advisor consent --revoke`는 동의를 지웁니다. 아직 디스패치되지 않은 상담은 차단되지만, 이미 디스패치된 상담은 취소할 수 없고 이미 프록시를 떠난 맥락도 되돌릴 수 없습니다. `ocx advisor set`은 동의를 부여하지 않습니다. 대시보드의 동의 확인란은 미리 선택되지 않습니다. ## 정책 @@ -98,6 +98,7 @@ fetch 기반 웹 검색 턴에서는 실제 적용되는 `tool_choice`가 합성 ## PR1 제한 - 네이티브 OpenAI 패스스루 턴(ChatGPT 풀 워커)에는 합성 도구가 주입되지 않습니다. 어드바이저는 라우팅(번역) 제공자를 대상으로 합니다. preflight 상담은 run-turn 어댑터에 적용되지만 도구는 적용되지 않습니다. +- 번역된 `/v1/messages` 요청(예: 번역 제공자로 라우팅한 Claude Code)도 `/v1/responses`, `/v1/chat/completions`와 마찬가지로 합성 도구와 preflight 상담을 받습니다. `policy`와 맥락 공유 동의는 같은 규칙을 따르며, 설정은 현재 구성에서 읽습니다. 따라서 동의를 철회하면 진행 중인 요청이 아직 디스패치되지 않은 상담을 시작하는 것은 차단되지만, 이미 디스패치된 상담은 취소되지 않고 이미 프록시를 떠난 맥락도 되돌릴 수 없습니다. - 의미 기반 막힘 감지, 완료 검토, 단계별 에스컬레이션, 다중 전문가 또는 투표는 제공하지 않습니다. - preflight 중복 제거 원장은 프로세스 내에 있습니다. 프록시 재시작 후 진행 중인 작업은 preflight 상담을 한 번 더 받을 수 있습니다. diff --git a/docs-site/src/content/docs/reference/configuration/advisor.md b/docs-site/src/content/docs/reference/configuration/advisor.md index 6f67ab9e406..5576824297a 100644 --- a/docs-site/src/content/docs/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/reference/configuration/advisor.md @@ -217,6 +217,12 @@ consultation never switches the session's main model. - Native OpenAI passthrough turns (ChatGPT-pool workers) do not get the synthetic tool; advisor support covers routed (translated) providers. Preflight consultation applies to run-turn adapters; the tool does not. +- Translated `/v1/messages` requests (for example Claude Code routed to a translated provider) + get the synthetic tool and preflight like `/v1/responses` and `/v1/chat/completions`. The + same `policy` and context-sharing consent rules apply, and the settings are read from the live + config, so a consent revocation blocks consultations for requests already in flight that have + not yet been dispatched. It does not cancel a dispatched consultation or retract context that + already left the proxy. - No semantic stuck detection, completion review, escalation tiers, multiple advisors or voting. Adaptive consultation requires the structured validation/edit/retry evidence described above. - The preflight dedup ledger is process-local; after a proxy restart, a task in progress may diff --git a/docs-site/src/content/docs/ru/reference/configuration/advisor.md b/docs-site/src/content/docs/ru/reference/configuration/advisor.md index 12dbc3b0026..bf3c3012d14 100644 --- a/docs-site/src/content/docs/ru/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/ru/reference/configuration/advisor.md @@ -46,7 +46,7 @@ ocx advisor set --context-mode question-only Управляйте через страницу **Advisor** на дашборде или `ocx advisor status|on|off|set --model --effort --policy `. -Без текущего согласия `ocx advisor on` не включает межпровайдерную отправку: команда печатает раскрытие и останавливается. `ocx advisor on --ack-context-sharing` и `ocx advisor consent` записывают `v1`. `ocx advisor consent --revoke` снимает согласие и сразу прекращает отправку. `ocx advisor set` согласие не выдаёт. Флажок на дашборде изначально снят. +Без текущего согласия `ocx advisor on` не включает межпровайдерную отправку: команда печатает раскрытие и останавливается. `ocx advisor on --ack-context-sharing` и `ocx advisor consent` записывают `v1`. `ocx advisor consent --revoke` снимает согласие. Ещё не отправленные консультации блокируются, но уже отправленную консультацию нельзя отменить, а ушедший из прокси контекст нельзя отозвать. `ocx advisor set` согласие не выдаёт. Флажок на дашборде изначально снят. ## Политики @@ -98,6 +98,7 @@ OpenCodex не вставляет в этот запрос ключи API про ## Ограничения PR1 - Нативные passthrough-ходы OpenAI (воркеры пула ChatGPT) не получают синтетический инструмент; поддержка консультанта покрывает маршрутизируемых (переведённых) провайдеров. Preflight-консультация применяется к run-turn-адаптерам; инструмент — нет. +- Переведённые запросы `/v1/messages` (например, Claude Code, направленный к переведённому провайдеру) получают синтетический инструмент и preflight-консультацию так же, как `/v1/responses` и `/v1/chat/completions`. К ним применяются те же правила `policy` и согласия на передачу контекста, а настройки читаются из текущей конфигурации: поэтому отзыв согласия не даёт уже выполняющимся запросам начать ещё не отправленную консультацию. Он не отменяет отправленную консультацию и не возвращает контекст, который уже покинул прокси. - Нет семантической детекции застревания, итоговой проверки, уровней эскалации, нескольких консультантов или голосования. - Учётная книга дедупликации preflight живёт в процессе; после перезапуска прокси задача в работе может получить ещё одну preflight-консультацию. diff --git a/docs-site/src/content/docs/tr/reference/configuration/advisor.md b/docs-site/src/content/docs/tr/reference/configuration/advisor.md index 0d28eb2c35e..b7e056d275d 100644 --- a/docs-site/src/content/docs/tr/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/tr/reference/configuration/advisor.md @@ -46,7 +46,7 @@ Kontrol panelindeki **Advisor** sayfasında katalogdan model seçin. Yerel, sağ Panodaki **Advisor** sayfası veya `ocx advisor status|on|off|set --model --effort --policy ` ile yönetin. -Güncel onay yokken `ocx advisor on` sağlayıcılar arası gönderimi açmaz: açıklamayı basar ve durur. `ocx advisor on --ack-context-sharing` ile `ocx advisor consent` `v1` kaydeder. `ocx advisor consent --revoke` onayı kaldırır ve gönderimi hemen durdurur. `ocx advisor set` onay vermez. Panodaki onay kutusu önceden işaretli değildir. +Güncel onay yokken `ocx advisor on` sağlayıcılar arası gönderimi açmaz: açıklamayı basar ve durur. `ocx advisor on --ack-context-sharing` ile `ocx advisor consent` `v1` kaydeder. `ocx advisor consent --revoke` onayı kaldırır. Henüz gönderilmemiş danışmalar engellenir, ancak gönderilmiş bir danışma iptal edilemez ve proxy'den ayrılmış bağlam geri alınamaz. `ocx advisor set` onay vermez. Panodaki onay kutusu önceden işaretli değildir. ## Politikalar @@ -98,6 +98,7 @@ Danışman fail-open davranır: gönderilmiş bir danışma başarısız olursa ## PR1 sınırlamaları - Yerel OpenAI passthrough turları (ChatGPT havuzu worker'ları) sentetik aracı almaz; danışman desteği yönlendirilen (çevrilen) sağlayıcıları kapsar. Preflight danışması run-turn bağdaştırıcılarına uygulanır; araç uygulanmaz. +- Çevrilen `/v1/messages` istekleri (örneğin çevrilen bir sağlayıcıya yönlendirilen Claude Code), `/v1/responses` ve `/v1/chat/completions` gibi sentetik aracı ve preflight danışmasını alır. Aynı `policy` ve bağlam paylaşımı onayı kuralları geçerlidir ve ayarlar güncel yapılandırmadan okunur; bu yüzden onayın geri alınması, devam eden isteklerin henüz gönderilmemiş bir danışma başlatmasını engeller. Gönderilmiş bir danışmayı iptal etmez ve proxy'den ayrılmış bağlamı geri almaz. - Anlamsal takılma algılama, tamamlanma incelemesi, yükseltme katmanları, çoklu danışman veya oylama yoktur. - Preflight tekilleştirme defteri süreç içindedir; proxy yeniden başlatıldıktan sonra devam eden bir görev bir preflight danışması daha alabilir. diff --git a/docs-site/src/content/docs/zh-cn/reference/configuration/advisor.md b/docs-site/src/content/docs/zh-cn/reference/configuration/advisor.md index 54d6bac562d..9b7c9752d8c 100644 --- a/docs-site/src/content/docs/zh-cn/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/zh-cn/reference/configuration/advisor.md @@ -46,7 +46,7 @@ ocx advisor set --context-mode question-only 通过仪表盘的 **Advisor** 页面或 `ocx advisor status|on|off|set --model --effort --policy ` 管理。 -`ocx advisor on` 在没有当前同意时不会开启跨 provider 发送：它会打印披露并停止。`ocx advisor on --ack-context-sharing` 与 `ocx advisor consent` 记录 `v1`。`ocx advisor consent --revoke` 会移除同意并立即停止发送。`ocx advisor set` 不授予同意。仪表盘上的同意复选框默认不勾选。 +`ocx advisor on` 在没有当前同意时不会开启跨 provider 发送：它会打印披露并停止。`ocx advisor on --ack-context-sharing` 与 `ocx advisor consent` 记录 `v1`。`ocx advisor consent --revoke` 会移除同意。尚未发出的咨询会被阻断，但已经发出的咨询无法取消，已经离开代理的上下文也无法收回。`ocx advisor set` 不授予同意。仪表盘上的同意复选框默认不勾选。 ## 策略 @@ -98,6 +98,7 @@ Advisor 失败是 fail-open 的：已经发出的咨询若失败（模型不可 ## PR1 限制 - 原生 OpenAI passthrough 回合（ChatGPT 池 Worker）不会获得合成工具；advisor 支持覆盖路由（translated）provider。preflight 咨询适用于 run-turn 适配器；工具不适用。 +- 翻译后的 `/v1/messages` 请求（例如将 Claude Code 路由到翻译 provider）与 `/v1/responses`、`/v1/chat/completions` 一样获得合成工具和 preflight 咨询。`policy` 与上下文共享同意遵循相同规则，设置从当前配置读取，因此撤销同意会阻止进行中的请求发起尚未发出的咨询；它不会取消已经发出的咨询，也无法收回已经离开代理的上下文。 - 不提供语义卡住检测、任务完成复核、升级分层、多专家或投票；自适应触发只使用已观察到的工具证据。 - preflight 去重账本是进程内的；代理重启后，进行中的任务可能再收到一次 preflight 咨询。 diff --git a/docs-site/src/content/docs/zh-tw/reference/configuration/advisor.md b/docs-site/src/content/docs/zh-tw/reference/configuration/advisor.md index 37ad6ee9b2e..486a1fe30d8 100644 --- a/docs-site/src/content/docs/zh-tw/reference/configuration/advisor.md +++ b/docs-site/src/content/docs/zh-tw/reference/configuration/advisor.md @@ -46,7 +46,7 @@ ocx advisor set --context-mode question-only 透過儀表板的 **Advisor** 頁面或 `ocx advisor status|on|off|set --model --effort --policy ` 管理。 -沒有目前同意時，`ocx advisor on` 不會開啟跨 provider 傳送：它會印出揭露並停止。`ocx advisor on --ack-context-sharing` 與 `ocx advisor consent` 記錄 `v1`。`ocx advisor consent --revoke` 會移除同意並立即停止傳送。`ocx advisor set` 不授予同意。儀表板上的同意核取方塊預設不勾選。 +沒有目前同意時，`ocx advisor on` 不會開啟跨 provider 傳送：它會印出揭露並停止。`ocx advisor on --ack-context-sharing` 與 `ocx advisor consent` 記錄 `v1`。`ocx advisor consent --revoke` 會移除同意。尚未發出的諮詢會被阻擋，但已經發出的諮詢無法取消，已經離開代理的上下文也無法收回。`ocx advisor set` 不授予同意。儀表板上的同意核取方塊預設不勾選。 ## 策略 @@ -98,6 +98,7 @@ Advisor 失敗是 fail-open 的：已經發出的諮詢若失敗（模型不可 ## PR1 限制 - 原生 OpenAI passthrough 回合（ChatGPT 池 Worker）不會獲得合成工具；advisor 支援覆蓋路由（translated）provider。preflight 諮詢適用於 run-turn 介面卡；工具不適用。 +- 翻譯後的 `/v1/messages` 請求（例如將 Claude Code 路由至翻譯 provider）與 `/v1/responses`、`/v1/chat/completions` 一樣取得合成工具與 preflight 諮詢。`policy` 與上下文共享同意遵循相同規則，設定從目前組態讀取，因此撤銷同意會阻止進行中的請求發起尚未送出的諮詢；它不會取消已送出的諮詢，也無法收回已離開代理的上下文。 - 不提供語義卡住偵測、任務完成複核、升級分層、多專家或投票；自適應觸發只使用已觀察到的工具證據。 - preflight 去重帳本是行程內的；代理重啟後，進行中的任務可能再收到一次 preflight 諮詢。 diff --git a/scripts/test-layout/layout-agent-assertion.json b/scripts/test-layout/layout-agent-assertion.json index 190baeb209e..0fd249faf55 100644 --- a/scripts/test-layout/layout-agent-assertion.json +++ b/scripts/test-layout/layout-agent-assertion.json @@ -10,6 +10,7 @@ "advisor-caller-policy-routing.test.ts": "advisor", "advisor-caller-policy-host-key.test.ts": "advisor", "advisor-caller-policy-self-fetch.test.ts": "advisor", + "advisor-messages-replay-config.test.ts": "advisor", "windows-native-principal.test.ts": "windows", "fixture-windows-runtime-preload.test.ts": "ci-workflows", "agent-assertion-auth.test.ts": "server", diff --git a/scripts/test-layout/layout-providers.json b/scripts/test-layout/layout-providers.json index 8e325ae8a29..386573c0d70 100644 --- a/scripts/test-layout/layout-providers.json +++ b/scripts/test-layout/layout-providers.json @@ -13,5 +13,6 @@ "mistral-image-limit.test.ts": "providers", "meta-muse-plan-refusal.test.ts": "providers", "model-discovery-skip-invalid-rows.test.ts": "providers", - "management-provider-model-discovery.test.ts": "server" + "management-provider-model-discovery.test.ts": "server", + "claude-cli-fixture-stdout-flush.test.ts": "providers" } diff --git a/skills/ocx/references/01_surface_agents-routing.md b/skills/ocx/references/01_surface_agents-routing.md index fbf56be29b3..873ca9a7be6 100644 --- a/skills/ocx/references/01_surface_agents-routing.md +++ b/skills/ocx/references/01_surface_agents-routing.md @@ -220,7 +220,7 @@ State-changing: yes. JSON mode: `payload`. | `--context-mode` | string | With set: context (default) or question-only; question-only pauses automatic consultations. | - `status` (the default) reads the resolved settings; `on`/`off` toggle the sidecar; `consent` records or revokes context-sharing consent; `set` updates model, effort, policy, context mode, task limit, or timeout. -- `on` does not grant consent. Without current consent it refuses and prints the disclosure. `on --ack-context-sharing` records consent v1 and enables. `consent --revoke` removes consent and stops task-context transfer. +- `on` does not grant consent. Without current consent it refuses and prints the disclosure. `on --ack-context-sharing` records consent v1 and enables. `consent --revoke` removes consent: consultations not yet dispatched are blocked, but it cannot cancel a dispatched consultation or retract task context already sent. - The advisor model may be any routable model string: a bare native model, an explicit `provider/model`, or an account-qualified native model. - `policy: preflight` makes OpenCodex attempt one automatic consultation per task with a stable conversation identity once the task shows orientation evidence (an assistant tool call or a tool result after the latest user message). Without a stable identity, each eligible request may trigger another consultation. `policy: manual` consults only when the worker calls the synthetic `advisor` tool. Neither path sends task context without current context-sharing consent. - `policy: adaptive` adds evidence-based failed-repair consultation after the same validation fails, a file edit completes, and validation fails again. Automatic calls require a stable conversation identity. `set --max-consultations <1-3>` caps adaptive task attempts, including manual calls, in a durable 24-hour budget. No semantic stuck detection is claimed. diff --git a/src/advisor/consult.ts b/src/advisor/consult.ts index 902120c905f..629b7823d8c 100644 --- a/src/advisor/consult.ts +++ b/src/advisor/consult.ts @@ -20,7 +20,6 @@ import type { OcxConfig } from "../types"; import { localInferenceDestination, selfFetchAdmissionToken, type LocalInferenceDestination } from "../lib/local-destinations"; import { signalWithTimeout, cancelBodyOnAbort } from "../lib/abort"; -import { redactSecretString } from "../lib/redact"; import { sidecarEnter } from "../lib/sidecar-tracker"; import { ADVISOR_INTERNAL_CAPABILITY_HEADER, @@ -258,7 +257,7 @@ export async function consultAdvisor( const kind = e instanceof Error && e.name === "TimeoutError" ? "timeout" : "connect_error"; return { ok: false, advice: "", advisorModel: input.advisorModel, - error: `advisor ${kind}: ${redactSecretString(e instanceof Error ? e.message : String(e))}`, + error: `advisor ${kind}`, durationMs, }; } finally { diff --git a/src/advisor/runtime.ts b/src/advisor/runtime.ts index 871e87492d2..0a57468e85c 100644 --- a/src/advisor/runtime.ts +++ b/src/advisor/runtime.ts @@ -36,6 +36,7 @@ import { } from "./state"; import { formatAdvisorAdvice, advisorPreflightMessages, formatAdvisorUnavailable } from "./context"; import { advisorActivityStore, type AdvisorActivityStore, type AdvisorTrigger } from "./activity"; +import { advisorPlanRootConfig } from "../server/responses/advisor-plan-slot"; import { createAdvisorTaskBudget, type AdvisorTaskBudget } from "./budget"; import { collectAdvisorRepairEvidence, type AdvisorRepairEvidence } from "./evidence"; import type { AdmissionModelScope } from "../server/admission-model-scope"; @@ -95,7 +96,10 @@ export function createAdvisorRuntimePlan(deps: AdvisorRuntimeDeps): AdvisorRunti const ledger = deps.ledger ?? sharedPreflightLedger; const now = deps.now ?? (() => Date.now()); const budget = deps.budget ?? createAdvisorTaskBudget(); - const activities = deps.activities ?? advisorActivityStore(deps.config); + // Identity-scoped state must key on the LIVE config: `GET /api/advisor/activity` reads the store + // of the management context's own config, so a request-scoped view (the Claude replay config) + // would record consultations the dashboard and CLI never see. + const activities = deps.activities ?? advisorActivityStore(advisorPlanRootConfig(deps.config)); // Management PUT mutates this same config object (`config.advisor = ...`). Re-resolving // at dispatch is what makes a mid-request consent revocation stop outbound transfer. const liveSettings = () => resolveAdvisorSettings(deps.config); diff --git a/src/cli/advisor.ts b/src/cli/advisor.ts index bdfa4ea08c7..4ffd5320a51 100644 --- a/src/cli/advisor.ts +++ b/src/cli/advisor.ts @@ -117,7 +117,7 @@ async function consent(argv: string[], deps: RuntimeApiDeps): Promise { contextSharingConsent: revoke ? null : ADVISOR_CONTEXT_SHARING_CONSENT_VERSION, }), }, deps), wantsJson, [revoke - ? "Advisor context-sharing consent removed. Consultations will not send task content." + ? "Advisor context-sharing consent removed. Consultations not yet dispatched will not send task content; a dispatched consultation cannot be cancelled and task content already sent cannot be retracted." : "Advisor context-sharing consent v1 recorded.", ]); } diff --git a/src/cli/capabilities-agents-routing.ts b/src/cli/capabilities-agents-routing.ts index 6d7c479585d..28e2553f7a8 100644 --- a/src/cli/capabilities-agents-routing.ts +++ b/src/cli/capabilities-agents-routing.ts @@ -55,7 +55,7 @@ export const AGENT_ROUTING_CAPABILITIES: readonly Capability[] = [ json: "payload", details: [ "`status` (the default) reads the resolved settings; `on`/`off` toggle the sidecar; `consent` records or revokes context-sharing consent; `set` updates model, effort, policy, context mode, task limit, or timeout.", - "`on` does not grant consent. Without current consent it refuses and prints the disclosure. `on --ack-context-sharing` records consent v1 and enables. `consent --revoke` removes consent and stops task-context transfer.", + "`on` does not grant consent. Without current consent it refuses and prints the disclosure. `on --ack-context-sharing` records consent v1 and enables. `consent --revoke` removes consent: consultations not yet dispatched are blocked, but it cannot cancel a dispatched consultation or retract task context already sent.", "The advisor model may be any routable model string: a bare native model, an explicit `provider/model`, or an account-qualified native model.", "`policy: preflight` makes OpenCodex attempt one automatic consultation per task with a stable conversation identity once the task shows orientation evidence (an assistant tool call or a tool result after the latest user message). Without a stable identity, each eligible request may trigger another consultation. `policy: manual` consults only when the worker calls the synthetic `advisor` tool. Neither path sends task context without current context-sharing consent.", "`policy: adaptive` adds evidence-based failed-repair consultation after the same validation fails, a file edit completes, and validation fails again. Automatic calls require a stable conversation identity. `set --max-consultations <1-3>` caps adaptive task attempts, including manual calls, in a durable 24-hour budget. No semantic stuck detection is claimed.", diff --git a/src/cli/registry.ts b/src/cli/registry.ts index 75b76122639..5b96125bff7 100644 --- a/src/cli/registry.ts +++ b/src/cli/registry.ts @@ -416,7 +416,7 @@ export const CLI_COMMANDS: CliCommandEntry[] = [ details: [ "ocx advisor and ocx advisor status read the resolved settings; use --json for machine-readable output.", "ocx advisor on requires current context-sharing consent. Pass --ack-context-sharing to record consent v1 and enable. ocx advisor off disables the sidecar without granting consent.", - "ocx advisor consent records consent v1 after printing the disclosure. ocx advisor consent --revoke removes it and stops task-context transfer.", + "ocx advisor consent records consent v1 after printing the disclosure. ocx advisor consent --revoke removes it: consultations not yet dispatched are blocked, while a dispatched consultation cannot be cancelled and task content already sent cannot be retracted.", "ocx advisor set updates --model, --effort, --policy, --context-mode, --max-consultations and --timeout-ms; the model may be any routable model string (bare native, provider/model, or account-qualified). set does not grant consent.", "In the default context mode, policy manual consults only when the worker calls the synthetic advisor tool; policy preflight also attempts one automatic consultation per task with a stable conversation identity, once the task shows orientation evidence. Without a stable identity, each eligible request may trigger another consultation. Both require current consent before any task context is sent. The question-only mode pauses automatic consultations.", ], diff --git a/src/server/claude-messages.ts b/src/server/claude-messages.ts index d77104fcf3e..9933e1c0f32 100644 --- a/src/server/claude-messages.ts +++ b/src/server/claude-messages.ts @@ -46,6 +46,7 @@ import { captureAdmissionPolicy } from "./capture-admission-policy"; import type { CapturedAdmissionPolicy } from "../types/admission-policy"; import { captureClaudeInbound } from "../claude/inbound-debug"; import { declaredAnthropicServerTools, replayedAnthropicServerBlocks, systemCachePrefix } from "../claude/inbound-content-options"; +import { linkAdvisorPlanConfig } from "./responses/advisor-plan-slot"; import { claudeCodeForIngress } from "../claude/intercept/model-bindings"; import { classifyInterceptClient } from "../claude/intercept/client-class"; import { analyzeClaudeCompatibility, isClaudeCompatibilityMode } from "../claude/compatibility"; @@ -1265,12 +1266,13 @@ async function handleClaudeMessagesWithBudget( // BEFORE translation (the translated Anthropic stream has no response.completed // frame, so tapping it records a bogus 502 with no usage/cache detail). const finalizeNativeLog = createFinalRequestLog(logIds, logCtx).finish; - const upstream = await handleResponses(internalReq, { + // The replay view is a new object; link it so the Advisor factory registered on `config` applies. + const upstream = await handleResponses(internalReq, linkAdvisorPlanConfig(config, { ...buildClaudeReplayConfig(config), // Operator config wins; otherwise the caller's explicit TTLs (all 1h, or a pinned 5m) carry // over, and TTL-less breakpoints take the destination default. cacheRetention: config.cacheRetention ?? callerCacheRetention(anthropicBody), - }, logCtx, { + }), logCtx, { ...(nativeComboSource && envelope && settledRoute?.combo ? { protocolSource: (await import("./messages-native")).createNativeMessagesComboSource({ req, config, logCtx, envelope, requestedModel, translatorBudget, selector: nativeSelector, diff --git a/src/server/responses/advisor-plan-slot.ts b/src/server/responses/advisor-plan-slot.ts index 55174d00c6e..eea097b8ebf 100644 --- a/src/server/responses/advisor-plan-slot.ts +++ b/src/server/responses/advisor-plan-slot.ts @@ -29,6 +29,38 @@ export function setAdvisorPlanFactory(config: OcxConfig, factory: AdvisorPlanFac else factories.delete(config); } +/** Request-scoped config views (e.g. the Claude replay config) -> the live config they were built from. */ +type AdvisorLinkedConfig = Pick; +const views = new WeakMap(); + +/** + * Record that `view` is a request-scoped copy of `live`. A view resolves the live config's + * factory at lookup time, so a later disable or re-activation of the live config still applies. + * The view's `advisor` reads through to the live config: management writes replace that object + * (`config.advisor = ...`), so a copied reference would keep a revoked consent alive for the + * plan. Every other request-scoped override on the view is kept. Nothing is stored on the live + * config, and an accessor is still cloneable (#7974). + */ +export function linkAdvisorPlanConfig(live: OcxConfig, view: T): T { + const root = views.get(live) ?? live; + views.set(view, root); + if (view !== root) { + Object.defineProperty(view, "advisor", { get: () => root.advisor, enumerable: true, configurable: true }); + } + return view; +} + +/** + * The live config a request-scoped view was linked to, or the config itself when it was not + * linked. Identity-scoped state that must outlive one request (the activity feed the management + * route reads from its own config object) resolves through the same link as the factory. + */ +export function advisorPlanRootConfig(config: T): T | OcxConfig { + return views.get(config) ?? config; +} + +/** Create the plan from the factory registered for this config, or for the live config it was linked to. */ export function createRegisteredAdvisorPlan(input: AdvisorPlanInput): Promise | null { - return factories.get(input.config)?.(input) ?? null; + const live = views.get(input.config); + return (factories.get(input.config) ?? (live && factories.get(live)))?.(input) ?? null; } diff --git a/structure/advisor.md b/structure/advisor.md index 511a85e7d74..ff8712c3f89 100644 --- a/structure/advisor.md +++ b/structure/advisor.md @@ -19,6 +19,13 @@ The server composition root registers it synchronously at startup; a successful updates activation, including enable after startup and disable/reset. The runtime is loaded only when an active factory is used, so its process-local ledger is not constructed by an inactive install. `sidecar-execution.ts` reads the core-owned slot, not Advisor settings or runtime. +Request-scoped copies of the live config, such as the Claude replay config `/v1/messages` passes to +the Responses pipeline, are linked with `linkAdvisorPlanConfig` so they resolve the live config's +factory at lookup time. The view's `advisor` reads through to the live config, so a consent +revocation after the copy still blocks outbound transfer up to dispatch: the runtime rechecks +consent immediately before the loopback call, and a revocation after that call has been dispatched +neither cancels it nor retracts context already sent +(`tests/advisor/advisor-messages-replay-config.test.ts`). Advisor settings admit the complete synchronous candidate/read/patch/save/activation phase through the [config admission contract](config.md). Production uses the current disk candidate when present, preserving unrelated disk edits during admission. Successful saves adopt the committed Advisor subtree and its field-scoped live baseline together, including known absence after reset, without advancing unrelated pending live edits. Fallback or error-bearing diagnostics are refused before mutation. Deadline refusal leaves live and durable settings untouched and retains the existing `write_failed` response. Successful receipts are serialized in that callback before another settings mutation can run. Management loads `advisor-routes.ts` only for the Advisor namespace. The guard is attached through `_advisorGuard`. `tests/advisor/advisor-core-boundary.test.ts` enforces the transitive load-time @@ -231,9 +238,8 @@ substring search. Automatic preflight is not decided from developer-message text manual tool result is a separate provenance check; every other history form — ordinary tool output, developer text, user text, and failure notices (``) — matches nothing. The guard never composes failure prose itself: `AdvisorPlan.formatUnavailable` -owns that text and neutralizes untrusted fragments. Upstream HTTP failures are logged as a -status code only, so an error body that echoes the prompt is not written to the worker context -or the log line. +owns that text and neutralizes untrusted fragments. Upstream HTTP failures use a status code only; transport exceptions use a fixed failure category. +Neither upstream error bodies nor exception messages enter worker context or consultation logs. Keys are SHA-256 digests, never a short fold and never raw text: one domain-separated digest over `conversation identity + task boundary + worker model`, where the task boundary digests the @@ -367,7 +373,9 @@ request under the advisor model. Consultation usage is never merged into the wor usage; intercepted worker legs are, via the same usage-merge rule the terminal guard applies. `src/advisor/activity.ts` holds config-scoped process memory: at most 100 records, at most 24 -hours old, with pruning on access and no timer. `GET /api/advisor/activity` and `ocx advisor +hours old, with pruning on access and no timer. The store is keyed by the LIVE config object, so a +request that runs on a linked view (the Claude replay config) records into the feed the management +route reads. `GET /api/advisor/activity` and `ocx advisor activity` expose trigger, status, models, requested/mapped effort, duration and reported tokens. There are no task bodies, commands, advice, conversation identifiers or provider error bodies in these records. Dashboard polling runs only while its page is visible. Restart clears activity, diff --git a/tests/advisor/advisor-consult.test.ts b/tests/advisor/advisor-consult.test.ts index 5910fe6baf7..94568ba9384 100644 --- a/tests/advisor/advisor-consult.test.ts +++ b/tests/advisor/advisor-consult.test.ts @@ -116,6 +116,18 @@ describe("consultAdvisor", () => { expect(result.error).toContain("connect_error"); }); + test("timeout and caller cancellation retain distinct safe outcomes", async () => { + globalThis.fetch = (async () => { throw new DOMException("private timeout text", "TimeoutError"); }) as typeof fetch; + const timeout = await consultAdvisor(baseInput, {}, "max", 5_000, undefined, "http://advisor.test"); + expect(timeout.ok).toBe(false); + expect(timeout.error).toBe("advisor timeout"); + const controller = new AbortController(); + globalThis.fetch = (async () => { controller.abort(); throw new Error("private cancellation text"); }) as typeof fetch; + const cancelled = await consultAdvisor(baseInput, {}, "max", 5_000, controller.signal, "http://advisor.test"); + expect(cancelled.cancelled).toBe(true); + expect(cancelled.error).toBe("advisor cancelled by the caller"); + }); + test("non-JSON and empty responses fail open without throwing", async () => { globalThis.fetch = (async () => new Response("not json", { status: 200 })) as typeof fetch; const nonJson = await consultAdvisor(baseInput, {}, "max", 5_000, undefined, "http://advisor.test"); diff --git a/tests/advisor/advisor-messages-replay-config.test.ts b/tests/advisor/advisor-messages-replay-config.test.ts new file mode 100644 index 00000000000..cce4ff167c8 --- /dev/null +++ b/tests/advisor/advisor-messages-replay-config.test.ts @@ -0,0 +1,434 @@ +import { afterEach, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { createAdvisorRuntimePlan, type AdvisorRuntimeDeps } from "../../src/advisor/runtime"; +import { createAdvisorTaskBudget } from "../../src/advisor/budget"; +import { createAdvisorActivityStore } from "../../src/advisor/activity"; +import { createAdvisorPreflightLedger } from "../../src/advisor/state"; +import { parseRequest } from "../../src/responses/parser"; +import { handleClaudeMessages } from "../../src/server/claude-messages"; +import { handleChatCompletions } from "../../src/server/chat-completions"; +import { handleAdvisorRoutes } from "../../src/server/management/advisor-routes"; +import type { ManagementContext } from "../../src/server/management/context"; +import { + createRegisteredAdvisorPlan, linkAdvisorPlanConfig, setAdvisorPlanFactory, type AdvisorPlanInput, +} from "../../src/server/responses/advisor-plan-slot"; +import { acquireOwnedSpendHome } from "../helpers/owned-spend-home"; +import type { OcxConfig, OcxParsedRequest } from "../../src/types"; + +// The Advisor factory is registered on the live config object, but /v1/messages hands the +// Responses pipeline a request-scoped replay view (`buildClaudeReplayConfig` plus cacheRetention). +// Looking the factory up by that view alone found nothing, so Claude Code never saw the advisor +// tool and preflight never ran while /v1/chat/completions on the same config worked. + +const workerStream = 'data: {"choices":[{"delta":{"content":"fixture"},"finish_reason":null}]}\n\n' + + 'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n'; + +/** Distinctive advice so the dispatched worker body proves what the preflight pass delivered. */ +const ADVICE = "advise from the translated messages fixture"; +const TASK = "synthetic-private-task-marker"; +const ERROR_BODY = "synthetic-private-error-body-marker"; +const CREDENTIAL = "synthetic-private-credential-marker"; + +/** Distinctive advice so the worker body proves it carried the preflight payload. */ +const STUB_ADVICE = '{"advisor_result":{"status":"advice","advice":"stub-advice-marker"}}'; + +const originalFetch = globalThis.fetch; +const roots: string[] = []; +afterEach(() => { + globalThis.fetch = originalFetch; + for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true }); +}); + +const plan = (seen: AdvisorPlanInput[], preflights: OcxParsedRequest[] = []) => async (input: AdvisorPlanInput) => { + seen.push(input); + return { + tool: { name: "advisor", description: "fixture", parameters: { type: "object", properties: {} }, advisor: true }, + preflightInject: async (parsed: OcxParsedRequest) => { + preflights.push(parsed); + parsed.context.messages = [ + ...parsed.context.messages, + { role: "user", content: STUB_ADVICE, timestamp: Date.now() }, + ]; + return true; + }, + attachGuard: () => {}, + }; +}; + +test("a linked request view resolves the live factory at lookup time", async () => { + const live = { port: 10100, providers: {} } as OcxConfig; + const view = linkAdvisorPlanConfig(live, { ...live, cacheRetention: "short" as const }); + const nested = linkAdvisorPlanConfig(view, { ...view }); + const seen: AdvisorPlanInput[] = []; + expect(createRegisteredAdvisorPlan({ config: view, workerIdentity: "w", workerModelId: "w" })).toBeNull(); + setAdvisorPlanFactory(live, plan(seen)); + try { + expect((await createRegisteredAdvisorPlan({ config: view, workerIdentity: "w", workerModelId: "w" }))?.tool.name).toBe("advisor"); + expect((await createRegisteredAdvisorPlan({ config: nested, workerIdentity: "w", workerModelId: "w" }))?.tool.name).toBe("advisor"); + // The plan still receives the view it was asked for, not the live object. + expect(seen.map(input => input.config)).toEqual([view, nested]); + expect(seen[0]!.config).toBe(view); + // The read-through accessor keeps the view cloneable (#7974). + expect(structuredClone(view).cacheRetention).toBe("short"); + // An unrelated copy never inherits a factory it was not linked to. + expect(createRegisteredAdvisorPlan({ config: { ...live }, workerIdentity: "w", workerModelId: "w" })).toBeNull(); + } finally { setAdvisorPlanFactory(live, null); } + // Disabling the live config detaches every view linked to it. + expect(createRegisteredAdvisorPlan({ config: view, workerIdentity: "w", workerModelId: "w" })).toBeNull(); +}); + +test("/v1/messages reaches the factory registered on the live config and sends the advisor tool", async () => { + const release = acquireOwnedSpendHome(); + const workerBodies: Record[] = []; + const seen: AdvisorPlanInput[] = []; + const preflights: OcxParsedRequest[] = []; + const config: OcxConfig = { + defaultProvider: "worker", + providers: { worker: { adapter: "openai-chat", apiKey: "fixture", models: ["m"], liveModels: false, + baseUrl: "https://worker.test/v1", + fetch: (async (_url, init) => { + workerBodies.push(JSON.parse(String(init?.body))); + return new Response(workerStream, { headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch } }, + }; + setAdvisorPlanFactory(config, plan(seen, preflights)); + try { + const response = await handleClaudeMessages(new Request("http://localhost/v1/messages", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "worker/m", max_tokens: 100, stream: false, messages: [{ role: "user", content: "fixture" }] }), + }), config, { model: "", provider: "" }, { requestId: "advisor-messages", start: Date.now() }); + await response.text(); + expect(response.status).toBe(200); + expect(seen).toHaveLength(1); + expect(seen[0]!.config).not.toBe(config); + // The Messages turn runs the automatic preflight pass, not only the tool declaration. + expect(preflights).toHaveLength(1); + expect(JSON.stringify(preflights[0]!.context.messages)).toContain("fixture"); + expect(workerBodies).toHaveLength(1); + const tools = (workerBodies[0]!.tools ?? []) as { function?: { name?: string } }[]; + expect(tools.map(tool => tool.function?.name)).toContain("advisor"); + // The advice the preflight pass injected reached the dispatch, not just the tool declaration. + expect(JSON.stringify(workerBodies[0])).toContain("stub-advice-marker"); + } finally { setAdvisorPlanFactory(config, null); release(); } +}); + +test("consent revoked on the live config after the view was copied blocks the consultation", async () => { + const root = mkdtempSync(join(tmpdir(), "advisor-messages-consent-")); roots.push(root); + const calls: unknown[] = []; + globalThis.fetch = (async (_input: RequestInfo | URL, init?: RequestInit) => { + calls.push(init?.body); + return Response.json({ choices: [{ message: { content: "advice" } }] }); + }) as typeof fetch; + const live = { port: 10100, providers: {}, advisor: { + enabled: true, model: "expert", policy: "manual", effort: "high", contextSharingConsent: "v1", + } } as OcxConfig; + const view = linkAdvisorPlanConfig(live, { ...live, cacheRetention: "short" as const }); + setAdvisorPlanFactory(live, async input => createAdvisorRuntimePlan({ + ...input, baseUrlOverride: "http://expert.test", ledger: createAdvisorPreflightLedger(), + budget: createAdvisorTaskBudget(join(root, "budget.sqlite")), activities: createAdvisorActivityStore(), + })); + try { + const registered = await createRegisteredAdvisorPlan({ config: view, workerIdentity: "w", workerModelId: "w" }); + expect(registered?.consult).toBeDefined(); + // Management writes replace the live object (`config.advisor = ...`) rather than mutating it. + const { contextSharingConsent: _revoked, ...rest } = live.advisor!; + live.advisor = rest; + const parsed = parseRequest({ model: "w", stream: false, input: [{ role: "user", content: "private-task-text" }] }); + const result = await registered!.consult!(parsed, "manual", "question"); + expect(result.blocked).toBe("consent"); + expect(calls).toHaveLength(0); + // Request-scoped overrides on the view are untouched. + expect(view.cacheRetention).toBe("short"); + } finally { setAdvisorPlanFactory(live, null); } +}); + +/** The management activity route as the dashboard and CLI reach it: it reads its own config object. */ +async function managementActivity(config: OcxConfig) { + const ctx: ManagementContext = { + req: new Request("http://localhost/api/advisor/activity"), + url: new URL("http://localhost/api/advisor/activity"), + config, + deps: {}, + version: "test", + trustedLoopbackIngress: false, + guiSessionIssuance: null, + convergeCodexCatalog: async () => ({ status: "skipped", reason: "not-requested", retryable: false }), + syncClaudeAgentDefsBestEffort: async () => {}, + }; + const response = await handleAdvisorRoutes(ctx); + return (await response!.json()) as { entries: { trigger: string; status: string; advisorModel: string; requestedEffort: string }[] }; +} + +/** A real translated /v1/messages turn: worker provider "worker", Advisor routed to "expert". */ +function translatedAdvisorFixture(options: { revokeOnDispatch?: boolean; failExpert?: boolean; throwTransport?: boolean; advice?: string } = {}) { + const root = mkdtempSync(join(tmpdir(), "advisor-messages-fixture-")); roots.push(root); + const workerBodies: Record[] = []; + const expertBodies: { url: string; body: Record }[] = []; + let revoked = false; + const config: OcxConfig = { + defaultProvider: "worker", + advisor: { enabled: true, model: "expert/worker", policy: "preflight", effort: "high", contextSharingConsent: "v1" }, + providers: { + worker: { adapter: "openai-chat", apiKey: "fixture", models: ["m"], liveModels: false, + baseUrl: "https://worker.test/v1", + fetch: (async (_url, init) => { + workerBodies.push(JSON.parse(String(init?.body))); + return new Response(workerStream, { headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch }, + expert: { adapter: "openai-chat", apiKey: CREDENTIAL, models: ["worker"], liveModels: false, + baseUrl: "https://expert.test/v1", + fetch: (async (url, init) => { + // Provider-facing model id is the target model ("worker"); the destination proves the route. + expertBodies.push({ url: String(url), body: JSON.parse(String(init?.body)) }); + // A revocation that lands only after the outbound hop has begun. + if (options.revokeOnDispatch) revoke(); + if (options.failExpert) return new Response(ERROR_BODY, { status: 400 }); + return Response.json({ choices: [{ message: { content: options.advice ?? ADVICE } }], + usage: { prompt_tokens: 11, completion_tokens: 7, total_tokens: 18 } }); + }) as typeof fetch }, + }, + } as OcxConfig; + const revoke = () => { + if (revoked) return; + revoked = true; + // Management writes replace the object; nothing mutates the copied reference in place. + const { contextSharingConsent: _revokedConsent, ...rest } = config.advisor!; + config.advisor = rest; + }; + const register = (extra: Partial = {}) => setAdvisorPlanFactory(config, async input => createAdvisorRuntimePlan({ + ...input, + ledger: createAdvisorPreflightLedger(), + budget: createAdvisorTaskBudget(join(root, "budget.sqlite")), + ...extra, + })); + // The Advisor self-fetches over loopback; re-enter the real data plane so the expert provider runs. + const activate = () => { globalThis.fetch = (async (input: RequestInfo | URL, init?: RequestInit) => { + const url = String(input); + if (url.includes("/v1/chat/completions")) { + if (options.throwTransport) throw new Error("exception-task-marker exception-advice-marker exception-credential-marker"); + const response = await handleChatCompletions(new Request(url, init), config, { model: "", provider: "" }); + // Exercise a loopback error that echoes private content even if the provider adapter + // itself sanitized its error. The actual expert dispatch still ran above. + if (options.failExpert && !response.ok) return new Response(ERROR_BODY, { status: response.status }); + return response; + } + throw new Error("unexpected external fetch: " + url); + }) as typeof fetch; + }; + activate(); + return { config, workerBodies, expertBodies, revoke, register, activate, revoked: () => revoked }; +} + +/** The real Messages request, with orientation evidence after the latest user turn. */ +function messagesRequest(task = TASK): Request { + return new Request("http://localhost/v1/messages", { + method: "POST", headers: { "content-type": "application/json" }, + body: JSON.stringify({ model: "worker/m", max_tokens: 100, stream: false, messages: [ + { role: "user", content: [{ type: "text", text: `repair the token store ${task}` }, + { type: "tool_result", tool_use_id: "call_1", content: "3 tests failed" }] }, + ] }), + signal: AbortSignal.timeout(10_000), + }); +} + +// Regression: the plan used to take its activity store from the replay view. Stores are keyed by +// config identity, and the management route reads the live config's store, so a consultation made +// through /v1/messages was invisible in the dashboard and the CLI activity feed. +test("/v1/messages records its consultation in the management-visible activity feed", async () => { + const release = acquireOwnedSpendHome(); + const fixture = translatedAdvisorFixture(); + fixture.register(); + try { + const response = await handleClaudeMessages(messagesRequest(), + fixture.config, { model: "", provider: "" }, { requestId: "advisor-messages-activity", start: Date.now() }); + await response.text(); + expect(response.status).toBe(200); + expect(fixture.expertBodies).toHaveLength(1); + expect(fixture.expertBodies[0]!.url).toBe("https://expert.test/v1/chat/completions"); + + const feed = await managementActivity(fixture.config); + expect(feed.entries).toHaveLength(1); + expect(feed.entries[0]).toMatchObject({ trigger: "preflight", status: "ok", advisorModel: "expert/worker" }); + // The feed's record is the consultation whose advice reached the worker. + expect(JSON.stringify(fixture.workerBodies[0])).toContain(ADVICE); + } finally { setAdvisorPlanFactory(fixture.config, null); release(); } +}); + +test("translated consultations stay in their live config feed and omit private content from metadata and logs", async () => { + const release = acquireOwnedSpendHome(); + // Construct B first: the loopback fetch installed by A then routes A's actual consultation. + const b = translatedAdvisorFixture(); + const a = translatedAdvisorFixture(); + b.config.advisor = { ...b.config.advisor!, effort: "low" }; + b.register(); + a.register(); + const originalWarn = console.warn; + const logs: string[] = []; + console.warn = (...args: unknown[]) => { logs.push(args.map(String).join(" ")); }; + try { + const response = await handleClaudeMessages(messagesRequest(), a.config, + { model: "", provider: "" }, { requestId: "advisor-isolation-success", start: Date.now() }); + await response.text(); + expect(response.status).toBe(200); + expect(a.expertBodies).toHaveLength(1); + expect(JSON.stringify(a.expertBodies[0]!.body)).toContain(TASK); + expect(JSON.stringify(a.workerBodies[0])).toContain(ADVICE); + const feedA = await managementActivity(a.config); + const feedB = await managementActivity(b.config); + expect(feedA.entries).toHaveLength(1); + expect(feedA.entries[0]).toMatchObject({ trigger: "preflight", status: "ok" }); + expect(feedB.entries).toHaveLength(0); + expect(b.expertBodies).toHaveLength(0); + expect(logs.some(line => line.includes("[advisor]"))).toBe(true); + for (const marker of [TASK, ADVICE, ERROR_BODY, CREDENTIAL]) { + expect(JSON.stringify([feedA, feedB, logs.filter(line => line.includes("[advisor]"))])).not.toContain(marker); + } + } finally { + console.warn = originalWarn; + setAdvisorPlanFactory(a.config, null); + setAdvisorPlanFactory(b.config, null); + release(); + } +}); + +test("translated expert HTTP errors do not expose their body in activity or advisor logs", async () => { + const release = acquireOwnedSpendHome(); + const fixture = translatedAdvisorFixture({ failExpert: true }); + fixture.register(); + const originalWarn = console.warn; + const logs: string[] = []; + console.warn = (...args: unknown[]) => { logs.push(args.map(String).join(" ")); }; + try { + const response = await handleClaudeMessages(messagesRequest(), fixture.config, + { model: "", provider: "" }, { requestId: "advisor-isolation-error", start: Date.now() }); + await response.text(); + expect(response.status).toBe(200); + expect(fixture.expertBodies).toHaveLength(1); + const feed = await managementActivity(fixture.config); + expect(feed.entries).toHaveLength(1); + expect(feed.entries[0]!.status).toBe("failed"); + expect(logs.some(line => line.includes("[advisor]"))).toBe(true); + for (const marker of [TASK, ADVICE, ERROR_BODY, CREDENTIAL]) { + expect(JSON.stringify([feed, logs.filter(line => line.includes("[advisor]"))])).not.toContain(marker); + } + } finally { + console.warn = originalWarn; + setAdvisorPlanFactory(fixture.config, null); + release(); + } +}); + +// Consent is rechecked before the outbound hop. A revocation in the claim-to-dispatch window +// blocks the transfer; the worker still runs without advisory context. +test("revocation after the view was copied but before dispatch blocks the /v1/messages consultation", async () => { + const release = acquireOwnedSpendHome(); + const fixture = translatedAdvisorFixture(); + fixture.register({ afterPreflightClaim: fixture.revoke }); + try { + const response = await handleClaudeMessages(messagesRequest(), + fixture.config, { model: "", provider: "" }, { requestId: "advisor-messages-revoke-before", start: Date.now() }); + await response.text(); + expect(fixture.revoked()).toBe(true); + expect(response.status).toBe(200); + expect(fixture.expertBodies).toHaveLength(0); + expect(fixture.workerBodies).toHaveLength(1); + expect(JSON.stringify(fixture.workerBodies[0])).not.toContain(ADVICE); + } finally { setAdvisorPlanFactory(fixture.config, null); release(); } +}); + +// The documented boundary: once dispatched, a revocation neither cancels the consultation nor +// retracts context already sent. This is what the docs now say, instead of promising a retraction. +test("revocation after dispatch does not cancel the /v1/messages consultation", async () => { + const release = acquireOwnedSpendHome(); + const fixture = translatedAdvisorFixture({ revokeOnDispatch: true }); + fixture.register(); + try { + const response = await handleClaudeMessages(messagesRequest(), + fixture.config, { model: "", provider: "" }, { requestId: "advisor-messages-revoke-after", start: Date.now() }); + await response.text(); + expect(fixture.revoked()).toBe(true); + expect(response.status).toBe(200); + // Dispatched: the context left the proxy, and the consultation completed. + expect(fixture.expertBodies).toHaveLength(1); + expect(JSON.stringify(fixture.expertBodies[0]!.body)).toContain(TASK); + expect(JSON.stringify(fixture.workerBodies[0])).toContain(ADVICE); + } finally { setAdvisorPlanFactory(fixture.config, null); release(); } +}); + + +test("translated transport exceptions preserve safe failure without leaking exception contents", async () => { + const release = acquireOwnedSpendHome(); + const fixture = translatedAdvisorFixture({ throwTransport: true }); + fixture.register(); + const originalWarn = console.warn; + const logs: string[] = []; + console.warn = (...args: unknown[]) => { logs.push(args.map(String).join(" ")); }; + try { + const response = await handleClaudeMessages(messagesRequest(), fixture.config, + { model: "", provider: "" }, { requestId: "advisor-transport-error", start: Date.now() }); + const body = await response.text(); + expect(response.status).toBe(200); + const feed = await managementActivity(fixture.config); + expect(feed.entries).toHaveLength(1); + expect(feed.entries[0]!.status).toBe("failed"); + expect(logs.some(line => line.includes("advisor connect_error"))).toBe(true); + expect(JSON.stringify(fixture.workerBodies)).toContain("advisor connect_error"); + for (const marker of ["exception-task-marker", "exception-advice-marker", "exception-credential-marker"]) { + expect(JSON.stringify([feed, logs, fixture.workerBodies, body])).not.toContain(marker); + } + } finally { + console.warn = originalWarn; + setAdvisorPlanFactory(fixture.config, null); + release(); + } +}); + +test("two live translated configurations isolate activity and consent", async () => { + const release = acquireOwnedSpendHome(); + const a = translatedAdvisorFixture({ advice: "advice-for-config-a" }); + const b = translatedAdvisorFixture({ advice: "advice-for-config-b" }); + b.config.advisor = { ...b.config.advisor!, effort: "low" }; + a.register(); b.register(); + const run = async (fixture: typeof a, task: string) => { + fixture.activate(); + const response = await handleClaudeMessages(messagesRequest(task), fixture.config, + { model: "", provider: "" }, { requestId: task, start: Date.now() }); + await response.text(); + expect(response.status).toBe(200); + }; + try { + await run(a, "task-for-config-a"); + await run(b, "task-for-config-b"); + expect(a.expertBodies).toHaveLength(1); + expect(b.expertBodies).toHaveLength(1); + expect(JSON.stringify(a.expertBodies)).toContain("task-for-config-a"); + expect(JSON.stringify(b.expertBodies)).toContain("task-for-config-b"); + expect(JSON.stringify(a.workerBodies)).toContain("advice-for-config-a"); + expect(JSON.stringify(b.workerBodies)).toContain("advice-for-config-b"); + expect(JSON.stringify([a.expertBodies, a.workerBodies])).not.toContain("config-b"); + expect(JSON.stringify([b.expertBodies, b.workerBodies])).not.toContain("config-a"); + const feedA = await managementActivity(a.config); + const feedB = await managementActivity(b.config); + expect(feedA.entries).toHaveLength(1); + expect(feedB.entries).toHaveLength(1); + expect(feedA.entries[0]!.requestedEffort).toBe("high"); + expect(feedB.entries[0]!.requestedEffort).toBe("low"); + for (const marker of ["task-for-config-a", "task-for-config-b", "advice-for-config-a", "advice-for-config-b", CREDENTIAL]) { + expect(JSON.stringify([feedA, feedB])).not.toContain(marker); + } + a.revoke(); + await run(a, "after-revoke-a"); + await run(b, "after-revoke-b"); + expect(a.expertBodies).toHaveLength(1); + expect(b.expertBodies).toHaveLength(2); + expect((await managementActivity(a.config)).entries).toHaveLength(1); + expect((await managementActivity(b.config)).entries).toHaveLength(2); + b.revoke(); + await run(b, "after-revoke-b-again"); + expect(b.expertBodies).toHaveLength(2); + } finally { + setAdvisorPlanFactory(a.config, null); setAdvisorPlanFactory(b.config, null); release(); + } +}); diff --git a/tests/advisor/advisor-settings.test.ts b/tests/advisor/advisor-settings.test.ts index 75d996df42b..30eb8c1337f 100644 --- a/tests/advisor/advisor-settings.test.ts +++ b/tests/advisor/advisor-settings.test.ts @@ -185,6 +185,36 @@ describe("ocx advisor consent", () => { { model: "expert/m" }, ]); }); + + // The human line is the operator's only confirmation. An unconditional "stops transfer" claim + // would promise cancellation the runtime cannot perform: consent is re-read before dispatch, so + // a revocation reaches pending consultations and nothing already sent. + test("the human revoke line names the undispatched-only boundary", async () => { + const lines: string[] = []; + const original = console.log; + console.log = (line?: unknown) => { lines.push(String(line)); }; + try { + expect(await handleAdvisorCommand(["consent", "--revoke"], depsWith([]))).toBe(0); + } finally { + console.log = original; + } + const output = lines.join("\n"); + expect(output).toContain("not yet dispatched"); + expect(output).toContain("cannot be cancelled"); + expect(output).toContain("cannot be retracted"); + }); + + test("revoke JSON remains the settings payload without the human line", async () => { + const lines: string[] = []; + const original = console.log; + console.log = (line?: unknown) => { lines.push(String(line)); }; + try { + expect(await handleAdvisorCommand(["consent", "--revoke", "--json"], depsWith([]))).toBe(0); + } finally { + console.log = original; + } + expect(JSON.parse(lines.join("\n"))).toEqual({ settings: { contextSharingConsent: null }, runnable: false }); + }); }); describe("ocx advisor set — clearing the model", () => { diff --git a/tests/fixtures/claude-efficiency-cli.ts b/tests/fixtures/claude-efficiency-cli.ts index 7274f3a9cf5..521384ba1fe 100644 --- a/tests/fixtures/claude-efficiency-cli.ts +++ b/tests/fixtures/claude-efficiency-cli.ts @@ -1,18 +1,123 @@ -import { readFileSync } from "node:fs"; +import { readFileSync, writeSync } from "node:fs"; const args = JSON.parse(process.argv[2]!) as string[]; const fixture = process.argv[3]!; // Only the owned synthetic probe uses this deliberate defect to calibrate its HTTP assertions. const cacheWrite = process.argv[4] === "omit-cache-creation" ? 0 : 80; + +/** + * Probe plumbing for the stdout-flush regression, inert in every other run. + * + * That regression leaves stdout unread on purpose, so the child cannot report what its own stdout + * write did by writing to stdout: the report would join the buffer under observation. It reports on + * stderr behind a sentinel prefix instead, which needs no descriptor beyond the three the adapter + * already gives it. + * + * The channel is one-way on purpose. A command descriptor one fd up was tried and dropped: the + * child's read of it intermittently never saw a command, and commands sent over stdin cannot be + * serviced while a multi-megabyte stdin write holds that same pipe. So the test does not ask the + * child for its state; the child states it as it happens, and the test waits for that mark. + * `OCX_REPLAY_PROBE` is never set by the adapter, a direct run, or any other test, so the default + * path below is the production one. + */ +const probeMode = process.env.OCX_REPLAY_PROBE; +const mark = (value: Record): void => { + if (!probeMode) return; + try { writeSync(2, `OCXPROBE ${JSON.stringify(value)}\n`); } catch { /* the test closed its end */ } +}; +// This mode exists to make the parent's stdin write fail. It runs before stdin is touched, so the +// child is provably alive and not reading while the parent's write is outstanding. +if (probeMode === "no-read") { + mark({ phase: "ready" }); + await new Promise(() => { /* the test kills this child and reads the write's failure */ }); +} +// This mode exits before stdin is touched, which is the early-exit shape the parent's request write +// has to survive: the write it issued in the spawn tick has no reader left, so the pipe closes under +// it. The mark is written synchronously first, so the parent has a record that this child reached the +// exit point rather than failing to start. +if (probeMode === "exit-before-read") { + mark({ phase: "exiting" }); + process.exit(9); +} let input = ""; for await (const chunk of process.stdin) input += chunk; const frame = JSON.parse(input.trim()); const emit = (value: unknown) => process.stdout.write(JSON.stringify(value) + "\n"); + +interface ReplayHooks { + /** Called in the write's own tick with its return value and the frame's byte count. */ + onReplayReturned?: (accepted: boolean, bytes: number) => void; + /** Called once the event loop has had one turn to run the write's callback. */ + onReplayTurn?: (settled: boolean) => void; + /** Called from inside the replay write's own completion callback. */ + onReplaySettled?: (error: string | null) => void; +} + +/** + * The replay writer, and the only place the exit sequence exists. + * + * Production calls this with no hooks, so the bytes a real turn depends on travel through the same + * statements the probe observes. A probe that reimplemented the sequence could keep passing while + * this one regressed, which is exactly what let an earlier version of this regression stay green + * after the production path was reverted. + */ +async function writeReplayAndExit(replayText: string, hooks: ReplayHooks): Promise { + const replayFrame = `${JSON.stringify({ type: "stream_event", event: { type: "content_block_delta", delta: { type: "text_delta", text: replayText } } })}\n`; + const resultFrame = `${JSON.stringify({ type: "result", subtype: "success", usage: { input_tokens: 2, output_tokens: 4 } })}\n`; + // The replay frame is megabytes. Bun buffers what it cannot hand to the destination immediately, so + // this write is not proof about reader progress on its own; the completion callback is. `process.exit` + // terminates without draining stdout still queued, which discarded the tail of this frame. + // Awaiting the result write's callback below orders every earlier write ahead of it and returns + // only once the frame is out of the process, so the exit cannot cut it. + let replaySettled = false; + const accepted = process.stdout.write(replayFrame, error => { + replaySettled = true; + hooks.onReplaySettled?.(error ? String((error as { code?: string }).code ?? error) : null); + }); + // `accepted` is the call's own return value. `false` says the runtime took the bytes into its own + // queue instead of handing them all to the destination within this call; it does not by itself say + // anything about the kernel pipe's capacity, because the runtime buffers internally first. + hooks.onReplayReturned?.(accepted, Buffer.byteLength(replayFrame)); + if (hooks.onReplayTurn) { + // One turn of the event loop, not a delay. This reports only whether the callback had run by the + // end of that single turn. It is not a statement that no later turn can change the answer: a + // reader that drains the destination later does complete the pending write. + await new Promise(resolve => setImmediate(resolve)); + hooks.onReplayTurn(replaySettled); + } + await new Promise((resolve, reject) => { + process.stdout.write(resultFrame, err => err ? reject(err) : resolve()); + }); + process.exit(0); +} + // Echo only the owned synthetic replay for the request-parser regression. No vendor account or socket. if (process.argv[4] === "image-replay") { - emit({ type: "system", subtype: "init", mcp_servers: [{ name: "opencodex", status: "connected" }] }); - emit({ type: "stream_event", event: { type: "content_block_delta", delta: { type: "text_delta", text: JSON.stringify(frame.message.content) } } }); - emit({ type: "result", subtype: "success", usage: { input_tokens: 2, output_tokens: 4 } }); - process.exit(0); + const replayText = JSON.stringify(frame.message.content); + const init = { type: "system", subtype: "init", mcp_servers: [{ name: "opencodex", status: "connected" }] }; + // The one defect this regression contrasts against stays outside the writer above: the pre-fix + // exit drops the tail by leaving it queued, which is not the production sequence. + if (probeMode === "exit-before-flush") { + emit(init); + emit({ type: "stream_event", event: { type: "content_block_delta", delta: { type: "text_delta", text: replayText } } }); + emit({ type: "result", subtype: "success", usage: { input_tokens: 2, output_tokens: 4 } }); + process.exit(0); + } + if (probeMode === "malformed-hang") { + emit(init); + process.stdout.write("this is not a stream-json frame\n"); + mark({ phase: "malformed-written" }); + await new Promise(() => { /* reaping is the test's job */ }); + } + emit(init); + // The probe observes the production writer; it never supplies a different exit sequence. + if (probeMode === "observe") { + await writeReplayAndExit(replayText, { + onReplayReturned: (accepted, bytes) => mark({ phase: "replay-returned", accepted, bytes }), + onReplayTurn: settled => mark({ phase: "replay-turn", settled }), + onReplaySettled: error => mark({ phase: "replay-settled", replayError: error }), + }); + } + await writeReplayAndExit(replayText, {}); } const current = frame.message.content.at(-1).text as string; const mcpPath = args[args.indexOf("--mcp-config") + 1]; diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index c069bb9bd9c..749963f0b0f 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -22,6 +22,7 @@ "advisor-caller-policy-routing.test.ts": "advisor", "advisor-caller-policy-host-key.test.ts": "advisor", "advisor-caller-policy-self-fetch.test.ts": "advisor", + "advisor-messages-replay-config.test.ts": "advisor", "graceful-max-output-tokens.test.ts": "responses", "graceful-max-output-tokens-exclusions.test.ts": "responses", "windows-config-root-async-writers.test.ts": "windows", @@ -485,6 +486,7 @@ "claude-auth-mode.test.ts": "claude-integration", "claude-authmode-migration.test.ts": "claude-integration", "claude-cli-adapter.test.ts": "providers", "claude-cli.test.ts": "claude-integration", "claude-cli-client-http.test.ts": "providers", "claude-cli-efficiency.test.ts": "providers", + "claude-cli-fixture-stdout-flush.test.ts": "providers", "claude-cli-tool-capture.test.ts": "providers", "claude-cli-usage-admission.test.ts": "providers", "claude-config-first-party.test.ts": "cli", "claude-code-thought-signature-scope.test.ts": "claude-integration", "claude-compatibility.test.ts": "claude-integration", "claude-context-windows.test.ts": "claude-integration", diff --git a/tests/helpers/messages-overflow-differential.ts b/tests/helpers/messages-overflow-differential.ts index 94ba3473029..d1c0e796f7d 100644 --- a/tests/helpers/messages-overflow-differential.ts +++ b/tests/helpers/messages-overflow-differential.ts @@ -26,8 +26,8 @@ function replaceOnce(source: string, needle: string, replacement: string): strin for (const path of changed) { let source = base ? git("show", `72e80d280:${path}`) : readFileSync(repoPath(path), "utf8"); if (path === "src/server/claude-messages.ts") source = replaceOnce(source, - "const upstream = await handleResponses(internalReq, {", - "const upstream = globalThis.__messagesReplayResponse ? globalThis.__messagesReplayResponse() : await handleResponses(internalReq, {"); + "const upstream = await handleResponses(internalReq, ", + "const upstream = globalThis.__messagesReplayResponse ? globalThis.__messagesReplayResponse() : await handleResponses(internalReq, "); if (path === "src/server/claude-messages.ts") source = replaceOnce(source, "message = await collectAnthropicMessage(anthropicSse,", "if (globalThis.__messagesCollectorStream) await anthropicSse.cancel();\n message = await collectAnthropicMessage(globalThis.__messagesCollectorStream ?? anthropicSse,"); diff --git a/tests/helpers/messages-proxy-refusal-case.ts b/tests/helpers/messages-proxy-refusal-case.ts index 0d03d70f10b..5a44b2c0f58 100644 --- a/tests/helpers/messages-proxy-refusal-case.ts +++ b/tests/helpers/messages-proxy-refusal-case.ts @@ -48,8 +48,8 @@ Bun.plugin({ name: "Messages refusal boundary fixture", setup(build) { build.onLoad({ filter: filters.upstream }, ({ path }) => { hooks.upstreamLoads++; return { loader: "ts", contents: replaceOnce(readOriginal(path), - "const upstream = await handleResponses(internalReq, {", - "const upstream = globalThis.__localRefusalReply ? (globalThis.__fixtureHooks.upstreamHits++, globalThis.__localRefusalReply) : await handleResponses(internalReq, {") }; + "const upstream = await handleResponses(internalReq, ", + "const upstream = globalThis.__localRefusalReply ? (globalThis.__fixtureHooks.upstreamHits++, globalThis.__localRefusalReply) : await handleResponses(internalReq, ") }; }); build.onLoad({ filter: filters.native }, ({ path }) => { hooks.nativeLoads++; diff --git a/tests/helpers/responses-core-source.ts b/tests/helpers/responses-core-source.ts index 354980b3552..22bb3c14c67 100644 --- a/tests/helpers/responses-core-source.ts +++ b/tests/helpers/responses-core-source.ts @@ -105,7 +105,7 @@ export const RESPONSES_CORE_MODULES = [ "adapter-delivery.ts", "anthropic-thinking-recovery.ts", "advisor-slot.ts", - // Reached from advisor-slot.ts: clips the opt-in question-only consultation text (#7526). + // Reached from advisor-slot.ts: shared question bounds/clip helper for opt-in question-only consultation text (#7526). "advisor-question.ts", "advisor-retention.ts", "advisor-plan-slot.ts", diff --git a/tests/providers/claude-cli-fixture-stdout-flush.test.ts b/tests/providers/claude-cli-fixture-stdout-flush.test.ts new file mode 100644 index 00000000000..c96dd97db3c --- /dev/null +++ b/tests/providers/claude-cli-fixture-stdout-flush.test.ts @@ -0,0 +1,438 @@ +import { expect, test } from "bun:test"; +import { spawn } from "node:child_process"; +import type { ChildProcess } from "node:child_process"; +import { readJsonLines } from "../../src/adapters/coding-agent/protocol"; +import { fixturePath } from "../helpers/repo-root"; + +/** + * The owned Claude CLI fixture answers the synthetic image replay with one JSONL frame whose text + * is the whole replayed conversation. That frame is megabytes, which is the shape the fixture's + * original bug needed: it wrote the frame and then called `process.exit(0)`, and `process.exit` + * terminates without draining stdout, so whatever the runtime still held was discarded and the + * reader saw a malformed frame. The size alone proves nothing here — Bun buffers internally — which + * is why the observations below come from the write's own callback rather than from byte counts. + * + * The fix and the observation share one function. `writeReplayAndExit` in the fixture is the only + * place the exit sequence exists, and the production run calls it with no hooks. An earlier version + * of this regression kept a separate probe copy of that sequence, so reverting the production path + * left every case green: the tests were watching code the adapter never ran. The default-path case + * below therefore drives production `readJsonLines` against a fixture spawned exactly as the adapter + * spawns it, with no probe environment at all, and the probe cases observe the same statements + * rather than a twin. + * + * Claims measured here, kept separate on purpose: + * + * - The contract, on the default path: all three frames and the whole replayed text arrive through + * the production reader, with a zero exit. This is the case that carries the fix: reverting only + * the production write-and-exit fails it, and also fails the reader-release case below, which + * waits for a mark the reverted sequence never emits. + * - The unsettled state is *observed*, not inferred from payload size. The fixture reports, in the + * write's own tick and again after one loop turn, whether the runtime took the bytes into its own + * queue and whether the completion callback had run by then. With stdout deliberately unread a + * multi-megabyte write has neither at that point; a small control has both. That contrast is + * buffering and size behavior on this platform — it is not a claim about a pipe threshold, and no + * later-turn behavior is asserted. + * - The narrowness is deliberate: the multi-megabyte case says the callback had not run within one + * turn, not that it can never run without a reader. Releasing the reader later does settle it, + * and that is the next case. + * - The failure paths fail: the pre-fix exit drops the frame, a malformed frame followed by a child + * that never exits surfaces the protocol error and is reaped, and a child that never reads stdin + * fails the parent's write with EPIPE instead of aborting an unrelated test. + * + * There is no command channel and no deadline. Marks ride stderr, so no spawn capability beyond the + * three standard descriptors is involved and no case is skipped on a platform that cannot pass a + * fourth one. The child's marks, its `close`, and the parent write's own failure are the only + * signals; a child that dies early fails the wait that expected its mark instead of stalling. + */ +const REPLAY_CHARS = 3_000_000; +/** + * The control is deliberately tiny: the point is only that a small write settles while the large one + * has not, through the same statements. It is kept at a byte count no plausible destination can fail + * to take in one piece, and the guard below applies only to the fixture's own reported byte count. + * It is *not* a measured pipe threshold, and nothing here asserts that a write above it blocks or + * that a write below it cannot: Bun's internal buffering means the runtime's return value describes + * its own queue, not the kernel destination. + */ +const CONTROL_SIZE_GUARD_BYTES = 512; +/** Frame overhead is fixed, so this many characters stays well inside that guard. */ +const CONTROL_CHARS = 64; +/** Marks are newline-delimited on stderr behind a sentinel, since stderr also carries diagnostics. */ +const MARK_PREFIX = "OCXPROBE "; + +const replayContent = (chars: number) => [{ type: "text", text: "x".repeat(chars) }]; + +interface Tracked { + child: ChildProcess; + /** Resolves at the child's `close`; subscribed at spawn, so a fast exit cannot be missed. */ + closed: Promise; + isClosed: () => boolean; + /** Exit code once closed, matching `close`'s argument. */ + exitCode: () => number | null; + spawnError: () => Error | undefined; + /** + * Stdin write failures recorded since spawn. A case that expects its request to arrive asserts + * this is empty; the case that deliberately provokes EPIPE asserts the code it contains. + */ + stdinErrors: () => NodeJS.ErrnoException[]; +} + +/** + * Subscribe to termination immediately after `spawn`. + * + * Registering `close` later — in a cleanup helper — is racy: a child that has already closed never + * emits it again, so the cleanup would wait forever for an event that is in the past. + */ +function track(child: ChildProcess): Tracked { + let closed = false; + let code: number | null = null; + let spawnError: Error | undefined; + let settle!: () => void; + const done = new Promise(resolve => { settle = resolve; }); + const finish = () => { + if (closed) return; + closed = true; + settle(); + }; + child.once("close", exitCode => { + code = exitCode; + finish(); + }); + child.once("error", error => { + spawnError = error; + // A launch failure has no process to reap and is not guaranteed to emit `close`. + if (child.pid === undefined) finish(); + }); + /** + * Subscribe to stdin failures here, before any caller writes. + * + * A child that exits without draining stdin fails the outstanding write with EPIPE. With no + * listener, Bun raises that as an uncaught error against whichever test happens to be running when + * it fires — which can be a later, unrelated one. Subscribing at spawn keeps the failure + * attributed to the test that owns the child, and the cases that expect their request to arrive + * assert the list is empty, so this listener records rather than swallows. + */ + const stdinErrors: NodeJS.ErrnoException[] = []; + child.stdin?.on("error", error => { stdinErrors.push(error as NodeJS.ErrnoException); }); + return { + child, + closed: done, + isClosed: () => closed, + exitCode: () => code ?? child.exitCode, + spawnError: () => spawnError, + stdinErrors: () => stdinErrors, + }; +} + +interface Probe extends Tracked { + marks: Record[]; + waitForMark: (phase: string) => Promise>; +} + +/** + * Spawn the fixture in a probe mode and collect its stderr marks. + * + * `OCX_REPLAY_PROBE` selects the mode and turns the marks on; the fixture ignores it when absent, so + * the adapter, a direct run and every other test run the production path unchanged. The request is + * written and stdin closed here, exactly as the adapter does it, so each probe case also exercises + * production stdin handling. + */ +function spawnProbe(mode: string, content: Array<{ type: string; text: string }>): Probe { + const child = spawn(process.execPath, [fixturePath("claude-efficiency-cli.ts"), "[]", "", "image-replay"], { + stdio: ["pipe", "pipe", "pipe"], + env: { ...process.env, OCX_REPLAY_PROBE: mode }, + }); + const tracked = track(child); + const marks: Record[] = []; + /** + * Waiters that have not settled yet. A check removes itself when it resolves or rejects; dropping + * the whole list on every event would lose a waiter whose mark is still in a later chunk, and the + * marks of one write do arrive in separate chunks. + */ + const waiting = new Set<() => void>(); + let buffered = ""; + let exited: Error | undefined; + + const notify = () => { + for (const check of [...waiting]) check(); + }; + + child.stderr!.on("data", (chunk: Buffer) => { + buffered += chunk.toString(); + for (let newline = buffered.indexOf("\n"); newline >= 0; newline = buffered.indexOf("\n")) { + const line = buffered.slice(0, newline); + buffered = buffered.slice(newline + 1); + if (!line.startsWith(MARK_PREFIX)) continue; // ordinary CLI diagnostics, not this probe + marks.push(JSON.parse(line.slice(MARK_PREFIX.length)) as Record); + } + notify(); + }); + + void tracked.closed.then(() => { + exited = tracked.spawnError() ?? new Error( + `fixture exited before the expected mark (exit=${String(tracked.exitCode())}); marks=${JSON.stringify(marks)}`, + ); + notify(); + }); + + const waitForMark = (phase: string): Promise> => new Promise((resolve, reject) => { + const check = () => { + const found = marks.find(mark => mark.phase === phase); + if (found) { + waiting.delete(check); + return resolve(found); + } + if (exited) { + waiting.delete(check); + return reject(exited); + } + }; + waiting.add(check); + check(); + }); + + // The fixture reads stdin to EOF before it answers, and `no-read` never reads it at all; that case + // deliberately leaves the request unwritten. Every other mode gets the request and EOF here. + if (mode !== "no-read") child.stdin!.end(`${JSON.stringify({ message: { content } })}\n`); + + return { ...tracked, marks, waitForMark }; +} + +/** Drain a child through the production reader, collecting the frames or the failure. */ +async function drain(child: ChildProcess) { + const frames: Record[] = []; + let readError: Error | undefined; + try { + for await (const frame of readJsonLines(child.stdout!)) frames.push(frame); + } catch (error) { + readError = error as Error; + } + return { frames, readError }; +} + +/** Resolve the exit code from the `close` subscription taken at spawn. */ +async function exitCodeOf(tracked: Tracked): Promise { + if (!tracked.isClosed()) await tracked.closed; + return tracked.exitCode(); +} + +/** Terminate and reap, so a failing assertion cannot leave a live child behind. */ +async function reap(tracked: Tracked): Promise { + if (tracked.isClosed()) return; + tracked.child.kill("SIGKILL"); + await tracked.closed; +} + +const replayText = (frames: Record[]): string | undefined => + (frames.find(frame => frame.type === "stream_event") as + { event?: { delta?: { text?: string } } } | undefined)?.event?.delta?.text; + +/** + * The production path, spawned the way `src/adapters/coding-agent/turn.ts` spawns it: three standard + * descriptors, no probe environment, request on stdin, frames read through `readJsonLines`. + * + * This is the case the review asked for. It exercises no probe at all, so it cannot pass while the + * statements a real turn depends on are reverted. + */ +test("the default fixture path delivers every frame and the whole replay through the production reader", async () => { + const content = replayContent(REPLAY_CHARS); + const expectedText = JSON.stringify(content); + const env = { ...process.env }; + delete env.OCX_REPLAY_PROBE; + const child = spawn(process.execPath, [fixturePath("claude-efficiency-cli.ts"), "[]", "", "image-replay"], { + stdio: ["pipe", "pipe", "pipe"], + env, + }); + const tracked = track(child); + child.stdin!.end(`${JSON.stringify({ message: { content } })}\n`); + try { + const { frames, readError } = await drain(child); + const exitCode = await exitCodeOf(tracked); + + expect(readError).toBeUndefined(); + expect(exitCode).toBe(0); + // The request reached the child: a write the child never read would have failed here instead of + // producing these frames, and this case would be reporting a pipe error as a parse error. + expect(tracked.stdinErrors()).toEqual([]); + expect(frames.map(frame => frame.type)).toEqual(["system", "stream_event", "result"]); + // Exact equality against the requested text is the completeness proof: a truncated frame either + // fails to parse or carries a prefix of it, and both are distinguished here. + expect(replayText(frames)).toBe(expectedText); + } finally { + await reap(tracked); + } +}, 60_000); + +test("the fixture's replay write is still pending while stdout has no reader", async () => { + const probe = spawnProbe("observe", replayContent(REPLAY_CHARS)); + try { + const returned = await probe.waitForMark("replay-returned"); + const turn = await probe.waitForMark("replay-turn"); + + // The payload is the same multi-megabyte frame the contract case uses; this is the child's own + // byte count, not the test's. + expect(returned.bytes as number).toBeGreaterThan(CONTROL_SIZE_GUARD_BYTES * 1000); + // The runtime took the bytes into its own queue instead of handing them all over, and one loop + // turn later the write's own callback still had not run. Both are the child's state, observed in + // the child; neither is inferred from the payload size, and neither claims more than one turn. + expect(returned.accepted).toBe(false); + expect(turn.settled).toBe(false); + } finally { + await reap(probe); + } +}, 60_000); + +test("a small payload settles with no reader, which is the buffering and size contrast", async () => { + const probe = spawnProbe("observe", replayContent(CONTROL_CHARS)); + try { + const returned = await probe.waitForMark("replay-returned"); + const turn = await probe.waitForMark("replay-turn"); + const settled = await probe.waitForMark("replay-settled"); + + // The byte count comes from the fixture, so a framing change cannot quietly push the control up + // into the size range of the case it contrasts with. This is a guard on that count only. + expect(returned.bytes as number).toBeLessThanOrEqual(CONTROL_SIZE_GUARD_BYTES); + // Same statements, same code path, same absent reader; only the size changed. Observed on this + // platform: the runtime accepted the bytes and the callback had run within one turn. The + // specific size at which that flips is not asserted — Bun's buffering, not the pipe bound, + // decides it. + expect(returned.accepted).toBe(true); + expect(turn.settled).toBe(true); + // The state in the mark is the write's own callback result, not a field this test invented. + expect(settled.replayError).toBeNull(); + } finally { + await reap(probe); + } +}, 60_000); + +/** + * Two independently observed things in one child: the frames the production reader receives, and + * the mark the child writes from inside the replay write's completion callback. Nothing orders the + * assertion on the mark against the reader, because the reader is the one draining the buffer that + * keeps the write alive. + */ +test("releasing the reader delivers the whole replay frame and the callback orders the exit", async () => { + const content = replayContent(REPLAY_CHARS); + const expectedText = JSON.stringify(content); + const probe = spawnProbe("observe", content); + try { + const returned = await probe.waitForMark("replay-returned"); + expect(returned.accepted).toBe(false); + + // The reader starts here, so the write that was queued now has somewhere to go. Its completion + // callback runs only after every earlier byte has left the process, which is what lets the + // fixture's `process.exit` follow without cutting the frame. + const drained = drain(probe.child); + const settled = await probe.waitForMark("replay-settled"); + const { frames, readError } = await drained; + const exitCode = await exitCodeOf(probe); + + expect(settled.replayError).toBeNull(); + expect(readError).toBeUndefined(); + expect(exitCode).toBe(0); + expect(frames.map(frame => frame.type)).toEqual(["system", "stream_event", "result"]); + // Exact equality: a truncated frame either fails to parse or carries a prefix. + expect(replayText(frames)).toBe(expectedText); + } finally { + await reap(probe); + } +}, 60_000); + +test("exiting before the replay write drains drops the frame, which the reader reports as malformed", async () => { + const probe = spawnProbe("exit-before-flush", replayContent(REPLAY_CHARS)); + try { + const { frames, readError } = await drain(probe.child); + await probe.closed; + + // The pre-fix behavior, kept as the red case. Only the small init frame survives: the replay + // frame is cut mid-JSON, so the reader rejects it rather than yielding a prefix, and the result + // frame after it never arrives at all. + expect(readError?.name).toBe("CodingAgentProtocolError"); + expect(readError?.message).toContain("Malformed stream-json frame"); + expect(frames.map(frame => frame.type)).toEqual(["system"]); + expect(replayText(frames)).toBeUndefined(); + } finally { + await reap(probe); + } +}, 60_000); + +test("a malformed frame from a child that never exits surfaces the error and is reaped", async () => { + const probe = spawnProbe("malformed-hang", replayContent(REPLAY_CHARS)); + try { + // The child's own mark says the corrupt frame reached the pipe; the failure below is then the + // reader rejecting it rather than the read having started before anything was written. + await probe.waitForMark("malformed-written"); + const { readError } = await drain(probe.child); + + // The read fails on its own: no deadline is involved, and the child is still running because + // nothing has asked it to stop. That is what makes the reap below the thing under test. + expect(readError?.name).toBe("CodingAgentProtocolError"); + expect(probe.isClosed()).toBe(false); + + await reap(probe); + expect(probe.isClosed()).toBe(true); + } finally { + await reap(probe); + } +}, 60_000); + +test("a child that never reads stdin fails the parent's write with EPIPE instead of aborting an unrelated test", async () => { + const probe = spawnProbe("no-read", replayContent(REPLAY_CHARS)); + const streamErrors: NodeJS.ErrnoException[] = []; + try { + // The child is alive and provably not reading stdin: it marked readiness before touching the + // pipe. The write below therefore fails through the child's exit, not through a race with its + // startup. + await probe.waitForMark("ready"); + + // The write is megabytes against a pipe nobody is draining, so the runtime queues it instead of + // handing it all over. That return value is asserted, not assumed: it is the precondition for the + // failure below coming from the pipe closing rather than the write having already finished. + const accepted = probe.child.stdin!.write("z".repeat(REPLAY_CHARS) + "\n", () => {}); + expect(accepted).toBe(false); + + const writeFailed = new Promise(resolve => { + // An unhandled stdin `error` does not abort the runner; Bun fails whichever test is running + // when it fires, which can be a later, unrelated one. Handling it here keeps the failure + // attributed to this test, and the error is still asserted below rather than swallowed. + probe.child.stdin!.on("error", error => { + streamErrors.push(error as NodeJS.ErrnoException); + resolve(error as NodeJS.ErrnoException); + }); + }); + + probe.child.kill("SIGKILL"); + const writeError = await writeFailed; + await probe.closed; + + // The queued write failed, and it failed as a closed pipe, not as an arbitrary error a + // condition free assertion would also accept. + expect(writeError.code).toBe("EPIPE"); + expect(streamErrors.map(error => error.code)).toContain("EPIPE"); + } finally { + await reap(probe); + } +}, 60_000); + +/** + * The early exit the writer cannot prevent: this child never reads stdin and leaves immediately, so + * the request written in the spawn tick has no reader and the pipe closes under it. The failure is + * asserted here, as a code, rather than left to surface wherever it happens to be noticed — an + * unhandled stdin `error` fails whichever test is running when it fires, which can be a later one. + * + * The exit status is checked alongside the pipe error so the two claims stay separate: the child + * really reached its exit point (9), and the outstanding write really failed as a closed pipe. + */ +test("an early child exit fails the request write with EPIPE in the case that wrote it", async () => { + const probe = spawnProbe("exit-before-read", replayContent(REPLAY_CHARS)); + try { + // The fixture marks the exit point before leaving, so an arriving mark rules out a child that + // died on startup, which would be a different failure with its own listener. + await probe.waitForMark("exiting"); + await probe.closed; + + expect(probe.exitCode()).toBe(9); + expect(probe.stdinErrors().map(error => error.code)).toContain("EPIPE"); + } finally { + await reap(probe); + } +}, 60_000);