diff --git a/devlog/_plan/260812_websearch_sidecar_live_streaming/000_findings_and_design.md b/devlog/_plan/260812_websearch_sidecar_live_streaming/000_findings_and_design.md new file mode 100644 index 0000000000..11e3c90103 --- /dev/null +++ b/devlog/_plan/260812_websearch_sidecar_live_streaming/000_findings_and_design.md @@ -0,0 +1,62 @@ +# Web-search sidecar: opt-in live streaming (`streamRoutedModelOutput`) + +Date: 2026-08-12. Follow-up to `260806_codex_desktop_streaming/000_findings.md` — this identifies +the concrete cause of the "chat answers arrive as one end-of-turn burst" symptom that investigation +left open, and lands the fix. + +## Root cause (reproduced end-to-end) + +Codex CLI/Desktop sends a hosted `web_search` tool on every real turn. For routed (non-passthrough) +models with a usable ChatGPT credential, `planWebSearch` engages the web-search sidecar, and +`runWithWebSearch` → `consumeIterationEvents` **fully buffers** every semantic adapter event of an +iteration before scanning for `web_search` calls. Client-visible output therefore arrives only at +turn end — 6–50 s of silence on reasoning-heavy turns, then a burst. + +Evidence chain (all on one machine, same provider `opencode-go/deepseek-v4-flash`, same key): + +- Two bit-identical proxy installs behaved differently: the instance with ChatGPT auth buffered + (`firstOutputMs ≈ durationMs` on 99/103 conversation requests), the instance with an EMPTY + `CODEX_HOME` streamed (`firstOutputMs ≈ 1.5–3 s`) — because only the former could engage the + sidecar. +- Byte-identical replay of a captured real `codex exec` request: buffered on the sidecar-enabled + instance, streamed on the other. Field bisect: removing only the `web_search` tool made the + sidecar-enabled instance stream (first delta 3.5 s, 448 deltas); removing `tool_search` / + `namespace` tools did not. + +## Why buffering exists, and what the fix preserves + +Buffering keeps two invariants: (1) the synthetic `web_search` tool call must never leak to Codex, +and (2) preliminary output from a pre-search iteration must not surface as the answer. The fix +keeps both by construction: + +- Live delivery is **opt-in** (`webSearchSidecar.streamRoutedModelOutput`, default `false`). +- Only event types the sidecar-less path would deliver identically may leave the live window: + `text_delta`, `thinking_delta`, `reasoning_raw_delta`, `thinking_signature`, + `redacted_thinking`, `kiro_redacted_reasoning` (allowlist in `loop.ts`). +- The window closes permanently at the first buffer-only event — tool calls above all — so the + `web_search` interception decision stays atomic and live events are exactly the first N + passthrough entries. The terminal replay skips them by count; nothing is delivered twice. +- Scanner semantics are unchanged: live events are still buffered for `extractIterationThinking` + and the forced-answer output check (#1001 behavior intact). + +Accepted tradeoff (documented in `docs-site/.../sidecars.md`): text the model emits before deciding +to search — which buffered mode silently drops — becomes visible and may partially repeat in the +post-search answer. Reasoning-first models (the common case) avoid the text-repetition case, though +their leading reasoning deltas become client-visible too — that visibility is the point of the +option. + +## Verification + +- `bun test tests/web-search.test.ts` — 55 pass, including 4 new tests: a gated adapter proves + deltas reach the client while the adapter is still mid-turn (buffered mode would deadlock the + gate); default-off buffering; window close at `tool_call_start` with exactly-once replay of the + tail; search-loop pass with pre-search text delivered exactly once. +- `bun test tests/web-search-*.test.ts` — 78 pass. `bun x tsc --noEmit` clean. +- Live replay of the captured Codex request through a patched instance (sidecar-less path): + 893 deltas, first at 2.6–3.3 s, unchanged totals. +- Sidecar-ACTIVE live E2E (patched build running as the native-main owner with a real ChatGPT + credential, identical text-forcing request, routed `opencode-go/deepseek-v4-flash`): toggle off → + 18.3 s silence then 2829 deltas in one 0.02 s burst; toggle on → first delta at 4.0 s, 2538 + deltas over 9.9 s. The toggle applied without restart via `PUT /api/sidecar-settings`. + Field note: a paused ChatGPT account (`pausedCodexAccountIds`) silently disables the sidecar and + masks both bug and fix — everything streams because the sidecar-less path runs. diff --git a/docs-site/src/content/docs/guides/sidecars.md b/docs-site/src/content/docs/guides/sidecars.md index e62e55c3ef..d0e487a47f 100644 --- a/docs-site/src/content/docs/guides/sidecars.md +++ b/docs-site/src/content/docs/guides/sidecars.md @@ -29,10 +29,25 @@ When Codex requests hosted `web_search` for a non-passthrough routed model, open (default 3), then removes the search tool and forces a final answer. Real client tools such as `apply_patch` or shell finalize the turn so those calls reach Codex. -Every routed-model iteration requests upstream `stream: true`, but opencodex fully buffers semantic -events internally before deciding whether to search or return the final answer. Only the first -iteration's final headers/status and 429 key rotations are acquired eagerly. Thus synthetic search -calls and preliminary output are never exposed as client-visible model output. +Every routed-model iteration requests upstream `stream: true`, but by default opencodex fully +buffers semantic events internally before deciding whether to search or return the final answer. +Only the first iteration's final headers/status and 429 key rotations are acquired eagerly. Thus +synthetic search calls and preliminary output are never exposed as client-visible model output. + +Opt-in `webSearchSidecar.streamRoutedModelOutput` (default `false`) streams each iteration's +leading text/thinking deltas live instead — the client sees output as soon as the model produces +it, exactly like the sidecar-less path. The live window closes permanently at the first tool-call +boundary, so the decision to intercept `web_search` stays atomic and nothing is ever delivered +twice (the terminal replay skips what already streamed). Tradeoff: text the model emits *before* +deciding to search — which buffered mode silently drops — becomes visible and may partially repeat +in the post-search answer. The Dashboard overview page exposes this as the **Stream answers live** +toggle on the web-search sidecar card (`PUT /api/sidecar-settings` with +`webSearch.streamRoutedModelOutput`). + +Kiro commentary is independent of this option: commentary-phase text already streams ahead of the +terminal event in buffered mode, and that bypass is unchanged — with or without +`streamRoutedModelOutput`, only search-decision events (tool calls and everything after the first +tool-call boundary) remain buffered for the atomic `web_search` decision. The injected result is wrapped in an untrusted-data boundary, length-capped, and de-duplicated by source URL. In structured-output turns (`json_schema` / `json_object`) it is handed over as compact @@ -48,7 +63,8 @@ relevant images in words and include their source URLs. "reasoning": "low", "maxSearchesPerTurn": 3, "routedModelStallTimeoutMs": 200000, - "timeoutMs": 200000 + "timeoutMs": 200000, + "streamRoutedModelOutput": false } } ``` diff --git a/gui/src/i18n/de.ts b/gui/src/i18n/de.ts index 98c476c617..189d1d27bb 100644 --- a/gui/src/i18n/de.ts +++ b/gui/src/i18n/de.ts @@ -269,6 +269,8 @@ export const de: Record = { "dash.visionModelHint": "Modell zur Beschreibung von Bildern für nur-Text-Routen. Erfordert ChatGPT-Login.", "dash.webSearchSidecar": "Websuche-Sidecar", "dash.webSearchSidecarHint": "Backend und Modell für die Websuche gerouteter Modelle auswählen.", + "dash.webSearchStream": "Antworten live streamen", + "dash.webSearchStreamHint": "Führenden Text und Reasoning live streamen, bis das Modell über einen Tool-Aufruf entscheidet; der Rest bleibt für das Abfangen der Suche gepuffert. Text vor einer Suche kann sich teilweise wiederholen.", "dash.visionSidecar": "Vision-Sidecar", "dash.visionSidecarHint": "Backend und Modell zur Bildbeschreibung für reine Textmodelle auswählen.", "dash.shadowCallIntercept": "Shadow-Call-Abfangen", diff --git a/gui/src/i18n/en.ts b/gui/src/i18n/en.ts index 77648b63ca..c25243a935 100644 --- a/gui/src/i18n/en.ts +++ b/gui/src/i18n/en.ts @@ -281,6 +281,8 @@ export const en = { "dash.visionModelHint": "Model used to describe images for text-only routed models. Requires ChatGPT login.", "dash.webSearchSidecar": "Web search sidecar", "dash.webSearchSidecarHint": "Choose the backend and model used for web search on routed models.", + "dash.webSearchStream": "Stream answers live", + "dash.webSearchStreamHint": "Stream the model’s leading text and reasoning live until it decides on a tool call; the rest of the turn stays buffered for search interception. Text written before a search may partially repeat.", "dash.visionSidecar": "Vision sidecar", "dash.visionSidecarHint": "Choose the backend and model used to describe images for text-only routed models.", "dash.shadowCallIntercept": "Shadow Call Intercept", diff --git a/gui/src/i18n/ja.ts b/gui/src/i18n/ja.ts index 4e56f55d38..2e5b277dc6 100644 --- a/gui/src/i18n/ja.ts +++ b/gui/src/i18n/ja.ts @@ -278,6 +278,8 @@ export const ja: Record = { "dash.visionModelHint": "テキスト専用ルーティングモデルで画像を説明するために使うモデル。ChatGPT ログインが必要です。", "dash.webSearchSidecar": "ウェブ検索サイドカー", "dash.webSearchSidecarHint": "ルーティングモデルでウェブ検索に使うバックエンドとモデルを選択します。", + "dash.webSearchStream": "回答をライブ配信", + "dash.webSearchStreamHint": "モデルがツール呼び出しを決定するまで、先頭のテキストと推論をライブ配信します。以降は検索インターセプトのためバッファされます。検索前のテキストは一部繰り返される場合があります。", "dash.visionSidecar": "ビジョンサイドカー", "dash.visionSidecarHint": "テキスト専用ルーティングモデルで画像を説明するために使うバックエンドとモデルを選択します。", "dash.shadowCallIntercept": "シャドウコール傍受", diff --git a/gui/src/i18n/ko.ts b/gui/src/i18n/ko.ts index c59cfde7fb..96d0f226fe 100644 --- a/gui/src/i18n/ko.ts +++ b/gui/src/i18n/ko.ts @@ -273,6 +273,8 @@ export const ko: Record = { "dash.visionModelHint": "텍스트 전용 라우팅 모델에 이미지를 설명하는 데 사용되는 모델입니다. ChatGPT 로그인 필요.", "dash.webSearchSidecar": "웹 검색 사이드카", "dash.webSearchSidecarHint": "라우팅 모델의 웹 검색에 쓸 백엔드와 모델을 고릅니다.", + "dash.webSearchStream": "응답 실시간 스트리밍", + "dash.webSearchStreamHint": "모델이 도구 호출을 결정할 때까지 앞부분 텍스트와 추론을 실시간 스트리밍합니다. 이후는 검색 가로채기를 위해 버퍼링됩니다. 검색 전 텍스트가 일부 반복될 수 있습니다.", "dash.visionSidecar": "비전 사이드카", "dash.visionSidecarHint": "텍스트 전용 라우팅 모델이 이미지를 읽을 때 쓸 백엔드와 모델을 고릅니다.", "dash.shadowCallIntercept": "쉐도우 호출 가로채기", diff --git a/gui/src/i18n/ru.ts b/gui/src/i18n/ru.ts index ed86f7eb3d..d1d8429662 100644 --- a/gui/src/i18n/ru.ts +++ b/gui/src/i18n/ru.ts @@ -278,6 +278,8 @@ export const ru: Record = { "dash.visionModelHint": "Модель, которая описывает изображения для маршрутизируемых моделей, работающих только с текстом. Требуется вход в аккаунт ChatGPT.", "dash.webSearchSidecar": "Сайдкар веб-поиска", "dash.webSearchSidecarHint": "Выберите бэкенд и модель, используемые для веб-поиска на маршрутизируемых моделях.", + "dash.webSearchStream": "Стримить ответы вживую", + "dash.webSearchStreamHint": "Транслировать начальный текст и рассуждения вживую, пока модель не решит вызвать инструмент; остальное буферизуется для перехвата поиска. Текст до поиска может частично повторяться.", "dash.visionSidecar": "Сайдкар для изображений", "dash.visionSidecarHint": "Выберите бэкенд и модель, которые описывают изображения для маршрутизируемых моделей, работающих только с текстом.", "dash.shadowCallIntercept": "Перехват теневых вызовов", diff --git a/gui/src/i18n/tr.ts b/gui/src/i18n/tr.ts index 3aa8275305..77c81d666a 100644 --- a/gui/src/i18n/tr.ts +++ b/gui/src/i18n/tr.ts @@ -279,6 +279,8 @@ export const tr: Record = { "dash.visionModelHint": "Salt metin yönlendirilen modeller için görselleri tanımlamakta kullanılan model. ChatGPT girişi gerektirir.", "dash.webSearchSidecar": "Web arama yan aracı (sidecar)", "dash.webSearchSidecarHint": "Yönlendirilen modellerde web araması için kullanılan arka ucu ve modeli seçin.", + "dash.webSearchStream": "Yanıtları canlı akıt", + "dash.webSearchStreamHint": "Model bir araç çağrısına karar verene kadar baştaki metni ve akıl yürütmeyi canlı akıtır; kalanı arama yakalama için arabelleğe alınır. Aramadan önce yazılan metin kısmen tekrarlanabilir.", "dash.visionSidecar": "Görsel yan aracı (sidecar)", "dash.visionSidecarHint": "Salt metin modeller için görselleri tanımlamakta kullanılan arka ucu ve modeli seçin.", "dash.shadowCallIntercept": "Gölge Çağrı Yakalama", diff --git a/gui/src/i18n/zh-TW.ts b/gui/src/i18n/zh-TW.ts index 09782ffe73..7a119a6667 100644 --- a/gui/src/i18n/zh-TW.ts +++ b/gui/src/i18n/zh-TW.ts @@ -172,6 +172,8 @@ export const zhTW: Record = { "dash.visionModelHint": "為純文字路由模型描述圖像的模型。需要 ChatGPT 登入。", "dash.webSearchSidecar": "網頁搜尋附屬服務", "dash.webSearchSidecarHint": "選擇路由模型進行網頁搜尋時使用的後端和模型。", + "dash.webSearchStream": "即時串流輸出回答", + "dash.webSearchStreamHint": "即時串流輸出開頭的文字和推理，直到模型決定呼叫工具；其餘部分為攔截搜尋而保持緩衝。搜尋前的文字可能會部分重複。", "dash.visionSidecar": "視覺附屬服務", "dash.visionSidecarHint": "選擇純文字路由模型描述圖像時使用的後端和模型。", "dash.shadowCallIntercept": "影子呼叫攔截", diff --git a/gui/src/i18n/zh.ts b/gui/src/i18n/zh.ts index ede0ffcef2..9825780ec2 100644 --- a/gui/src/i18n/zh.ts +++ b/gui/src/i18n/zh.ts @@ -273,6 +273,8 @@ export const zh: Record = { "dash.visionModelHint": "为纯文本路由模型描述图像的模型。需要 ChatGPT 登录。", "dash.webSearchSidecar": "网页搜索附属服务", "dash.webSearchSidecarHint": "选择路由模型进行网页搜索时使用的后端和模型。", + "dash.webSearchStream": "实时流式输出回答", + "dash.webSearchStreamHint": "实时流式输出开头的文本和推理，直到模型决定调用工具；其余部分为拦截搜索而保持缓冲。搜索前的文本可能会部分重复。", "dash.visionSidecar": "视觉附属服务", "dash.visionSidecarHint": "选择纯文本路由模型描述图像时使用的后端和模型。", "dash.shadowCallIntercept": "影子调用拦截", diff --git a/gui/src/pages/dashboard-overview-sections.tsx b/gui/src/pages/dashboard-overview-sections.tsx index 61ac28f8be..bf22ff796a 100644 --- a/gui/src/pages/dashboard-overview-sections.tsx +++ b/gui/src/pages/dashboard-overview-sections.tsx @@ -284,6 +284,21 @@ export function DashboardSidecarPanels({ d }: { d: Dash }) { disabled={!sidecar || sidecarSaving} label={t("dash.sidecarModel")} /> +
+ {t("dash.webSearchStream")} + +
diff --git a/gui/src/pages/dashboard-shared.ts b/gui/src/pages/dashboard-shared.ts index fe906724b3..faa99d4d43 100644 --- a/gui/src/pages/dashboard-shared.ts +++ b/gui/src/pages/dashboard-shared.ts @@ -56,7 +56,7 @@ export interface SettingsData { } export type SidecarBackend = "openai" | "anthropic"; export type VisionReasoning = "low" | "medium" | "high" | "xhigh" | "max"; -export interface SidecarSetting { backend?: SidecarBackend; model: string; reasoning?: VisionReasoning } +export interface SidecarSetting { backend?: SidecarBackend; model: string; reasoning?: VisionReasoning; streamRoutedModelOutput?: boolean } export interface VisionModelOption { value: string; label: string; backend: SidecarBackend; baseline?: boolean } export interface SidecarData { webSearch: SidecarSetting; @@ -67,7 +67,7 @@ export interface SidecarData { visionModels?: VisionModelOption[]; } export interface SidecarPatch { - webSearch?: { backend?: SidecarBackend | null; model?: string }; + webSearch?: { backend?: SidecarBackend | null; model?: string; streamRoutedModelOutput?: boolean }; vision?: { backend?: SidecarBackend | null; model?: string; reasoning?: VisionReasoning }; } export interface ShadowCallData { enabled: boolean; model: string; sourceModels?: string[] } @@ -151,13 +151,14 @@ export function updateJobLabel(status: UpdateJobStatus, t: (key: TKey) => string export function mergeSidecarSetting( current: SidecarSetting, - update?: { backend?: SidecarBackend | null; model?: string; reasoning?: VisionReasoning }, + update?: { backend?: SidecarBackend | null; model?: string; reasoning?: VisionReasoning; streamRoutedModelOutput?: boolean }, ): SidecarSetting { const merged = { ...current }; if (update?.model !== undefined) merged.model = update.model; if (update?.backend === null) delete merged.backend; else if (update?.backend !== undefined) merged.backend = update.backend; if (update?.reasoning !== undefined) merged.reasoning = update.reasoning; + if (update?.streamRoutedModelOutput !== undefined) merged.streamRoutedModelOutput = update.streamRoutedModelOutput; return merged; } diff --git a/src/server/management/config-routes.ts b/src/server/management/config-routes.ts index 0e3cffc67e..ca37248f36 100644 --- a/src/server/management/config-routes.ts +++ b/src/server/management/config-routes.ts @@ -410,7 +410,11 @@ export async function handleConfigRoutes(ctx: ManagementContext): Promise { const rotated = rotateProviderTransportOn429(config, route.providerName, route.provider, { retryAfter, diff --git a/src/types.ts b/src/types.ts index d0f9fcb21b..d24811f430 100644 --- a/src/types.ts +++ b/src/types.ts @@ -1161,6 +1161,14 @@ export interface OcxWebSearchSidecarConfig { * during a web-search turn. Default 200000. Must be an integer from 1 through 2147483647. */ routedModelStallTimeoutMs?: number; + /** + * Stream the routed model's leading output (text/thinking deltas) live instead of buffering the + * whole iteration. Live delivery stops at the first tool-call boundary so web_search interception + * stays atomic. Tradeoff: text the model emits BEFORE deciding to search — which buffered mode + * silently drops — becomes visible to the client and may partially repeat in the post-search + * answer. Default: false (buffered, previous behavior). + */ + streamRoutedModelOutput?: boolean; } export interface OpenRouterProviderRouting { diff --git a/src/web-search/index.ts b/src/web-search/index.ts index e902828bdc..f79787059d 100644 --- a/src/web-search/index.ts +++ b/src/web-search/index.ts @@ -120,6 +120,8 @@ export interface SidecarPlan { routedModelStallTimeoutMs: number; /** Effective bridge stall deadline for the sidecar turn (see webSearchStallTimeoutSec). */ stallTimeoutSec: number; + /** Stream leading routed-model output live until the first tool-call boundary (opt-in). */ + streamRoutedModelOutput: boolean; } export function shouldResolveOpenAiWebSearchSidecar( @@ -166,6 +168,7 @@ export function planWebSearch( // The routed model being text-only means the search model must verbalize image results (either backend). const describeImages = modelInList(provider.noVisionModels, modelId); const reasoning = cfg.reasoning ?? DEFAULT_SIDECAR_REASONING; + const streamRoutedModelOutput = cfg.streamRoutedModelOutput === true; // Anthropic backend authenticates with the STORED credential — no forward provider or ChatGPT login gate. // resolveSidecarBackend only returns "anthropic" when it was explicitly configured OR a usable credential @@ -181,6 +184,7 @@ export function planWebSearch( maxSearches, routedModelStallTimeoutMs, stallTimeoutSec, + streamRoutedModelOutput, }; } @@ -194,5 +198,6 @@ export function planWebSearch( maxSearches, routedModelStallTimeoutMs, stallTimeoutSec, + streamRoutedModelOutput, }; } diff --git a/src/web-search/loop.ts b/src/web-search/loop.ts index 0460a98cf5..4be7bdbbc6 100644 --- a/src/web-search/loop.ts +++ b/src/web-search/loop.ts @@ -267,6 +267,12 @@ export interface WebSearchLoopDeps { * sidecar search, so a legitimately slow-but-progressing unit never trips the bridge watchdog. */ stallTimeoutSec?: number; + /** + * Opt-in: stream the routed model's leading text/thinking deltas live instead of holding the whole + * iteration back. The live window closes at the first buffer-only event (tool calls above all) so + * the web_search interception decision stays atomic; everything after replays in order at the end. + */ + streamRoutedModelOutput?: boolean; /** One-shot TTFT callback: first non-empty model output observed (WP4). */ onFirstOutput?: () => void; /** Raw adapter usage at the terminal event, pre wire-normalization (see bridgeToResponsesSSE onUsage). */ @@ -336,7 +342,14 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise; + type IterationSplit = ReturnType & { + /** + * How many leading passthrough events were already delivered live this iteration. They are + * exactly the first N passthrough entries (live delivery stops before the first event that + * scanEventsForWebSearch could group or reorder), so the terminal replay skips them by count. + */ + streamedPassthroughCount: number; + }; // Same-target 429 budget is per REQUEST, not per model iteration: later search rounds inherit // what earlier rounds left of `attempts`, so a bounded multi-round turn can never exceed the @@ -531,10 +544,23 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise([ + "text_delta", "thinking_delta", "reasoning_raw_delta", + "thinking_signature", "redacted_thinking", "kiro_redacted_reasoning", + ]); + // Consume and validate one successful response body under a resettable raw-byte inactivity guard. - // Only invisible heartbeat events escape while semantic output remains buffered for safe scanning. + // By default only invisible heartbeat events escape while semantic output remains buffered for + // safe scanning; with `streamRoutedModelOutput` the leading text/thinking deltas stream live and + // the live window closes permanently at the first buffer-only event (see LIVE_STREAMABLE). const consumeIterationEvents = async function* (prepared: IterationResponse): AsyncGenerator { const events: AdapterEvent[] = []; + let liveWindowOpen = deps.streamRoutedModelOutput === true; + let streamedPassthroughCount = 0; try { const parse = prepared.responseAdapter.parseStream.bind(prepared.responseAdapter); for await (const event of parseStreamWithProgress(prepared.response, parse, { @@ -550,7 +576,16 @@ export async function runWithWebSearch(deps: WebSearchLoopDeps): Promise 0 ? "s" : ""}, ${Date.now() - loopT0}ms`, ); } - yield* replay(split.passthrough); + // Live-streamed leading events are exactly the first N passthrough entries — replay + // only the buffered tail so nothing reaches the client twice. + yield* replay(split.passthrough.slice(split.streamedPassthroughCount)); return; } // The thinking that led to the search belongs to the FIRST call's assistant replay turn. diff --git a/tests/sidecar-settings-web-search-stream.test.ts b/tests/sidecar-settings-web-search-stream.test.ts new file mode 100644 index 0000000000..19847d848a --- /dev/null +++ b/tests/sidecar-settings-web-search-stream.test.ts @@ -0,0 +1,107 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { loadConfig } from "../src/config"; +import { handleManagementAPI } from "../src/server/management-api"; +import type { OcxConfig } from "../src/types"; +import { ManagementRequest as Request } from "./helpers/management-auth"; + +async function getSidecarSettings(config: OcxConfig): Promise { + const url = new URL("http://localhost/api/sidecar-settings"); + const response = await handleManagementAPI(new Request(url), url, config); + if (!response) throw new Error("sidecar settings route did not handle GET"); + return response; +} + +async function putSidecarSettings(config: OcxConfig, webSearch: Record): Promise { + const url = new URL("http://localhost/api/sidecar-settings"); + const response = await handleManagementAPI( + new Request(url, { + method: "PUT", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ webSearch }), + }), + url, + config, + ); + if (!response) throw new Error("sidecar settings route did not handle PUT"); + return response; +} + +function emptyConfig(overrides: Partial = {}): OcxConfig { + // A schema-valid provider setup: for an invalid file, loadConfig() first retries with + // defaults merged in and falls back to backup + pure defaults only when that repair also + // fails validation — either path would silently void the reload assertions below. + return { + port: 10100, + defaultProvider: "dummy", + providers: { dummy: { adapter: "openai-chat", baseUrl: "https://example.test/v1" } }, + ...overrides, + } as OcxConfig; +} + +describe("sidecar-settings webSearch.streamRoutedModelOutput", () => { + let previousHome: string | undefined; + let isolatedHome: string | undefined; + + beforeEach(() => { + previousHome = process.env.OPENCODEX_HOME; + isolatedHome = mkdtempSync(join(tmpdir(), "ocx-sidecar-ws-stream-")); + process.env.OPENCODEX_HOME = isolatedHome; + }); + + afterEach(() => { + if (previousHome === undefined) delete process.env.OPENCODEX_HOME; + else process.env.OPENCODEX_HOME = previousHome; + if (isolatedHome) rmSync(isolatedHome, { recursive: true, force: true }); + isolatedHome = undefined; + }); + + test("GET reports false when unset and true when configured", async () => { + const off = await getSidecarSettings(emptyConfig()); + expect(off.status).toBe(200); + expect(((await off.json()) as { webSearch: { streamRoutedModelOutput: boolean } }) + .webSearch.streamRoutedModelOutput).toBe(false); + + const on = await getSidecarSettings(emptyConfig({ + webSearchSidecar: { streamRoutedModelOutput: true }, + })); + expect(((await on.json()) as { webSearch: { streamRoutedModelOutput: boolean } }) + .webSearch.streamRoutedModelOutput).toBe(true); + }); + + test("PUT true persists the flag and echoes it; PUT false removes the key", async () => { + const config = emptyConfig(); + const enable = await putSidecarSettings(config, { streamRoutedModelOutput: true }); + expect(enable.status).toBe(200); + expect(((await enable.json()) as { webSearch: { streamRoutedModelOutput: boolean } }) + .webSearch.streamRoutedModelOutput).toBe(true); + expect(config.webSearchSidecar?.streamRoutedModelOutput).toBe(true); + // Durable persistence: the flag must survive a config reload from disk. + expect(loadConfig().webSearchSidecar?.streamRoutedModelOutput).toBe(true); + + const disable = await putSidecarSettings(config, { streamRoutedModelOutput: false }); + expect(disable.status).toBe(200); + expect(((await disable.json()) as { webSearch: { streamRoutedModelOutput: boolean } }) + .webSearch.streamRoutedModelOutput).toBe(false); + // false is the default — the key is dropped so config files stay minimal. + expect("streamRoutedModelOutput" in (config.webSearchSidecar ?? {})).toBe(false); + expect("streamRoutedModelOutput" in (loadConfig().webSearchSidecar ?? {})).toBe(false); + }); + + test("PUT rejects a non-boolean value and leaves other fields untouched", async () => { + const config = emptyConfig({ webSearchSidecar: { model: "gpt-5.6-luna" } }); + const response = await putSidecarSettings(config, { streamRoutedModelOutput: "yes" }); + expect(response.status).toBe(400); + expect(config.webSearchSidecar?.streamRoutedModelOutput).toBeUndefined(); + expect(config.webSearchSidecar?.model).toBe("gpt-5.6-luna"); + }); + + test("PUT that omits the flag does not disturb an enabled value", async () => { + const config = emptyConfig({ webSearchSidecar: { streamRoutedModelOutput: true } }); + const response = await putSidecarSettings(config, { model: "gpt-5.6-luna" }); + expect(response.status).toBe(200); + expect(config.webSearchSidecar?.streamRoutedModelOutput).toBe(true); + }); +}); diff --git a/tests/vision-anthropic.test.ts b/tests/vision-anthropic.test.ts index ad93e6856a..2fbb3d141b 100644 --- a/tests/vision-anthropic.test.ts +++ b/tests/vision-anthropic.test.ts @@ -254,7 +254,7 @@ describe("Anthropic vision planning and management config", () => { config, ); const getBody = await get!.json() as Record; - expect(getBody.webSearch).toEqual({ model: "claude-search", backend: "anthropic" }); + expect(getBody.webSearch).toEqual({ model: "claude-search", backend: "anthropic", streamRoutedModelOutput: false }); expect(getBody.vision).toEqual({ model: "claude-sonnet-5", backend: "anthropic", @@ -276,7 +276,7 @@ describe("Anthropic vision planning and management config", () => { ); expect(clear.status).toBe(200); const clearBody = await clear.json() as Record; - expect(clearBody.webSearch).toEqual({ model: "gpt-5.6-luna" }); + expect(clearBody.webSearch).toEqual({ model: "gpt-5.6-luna", streamRoutedModelOutput: false }); expect(clearBody.vision).toEqual({ model: "gpt-5.4-mini", reasoning: "low", maxDescriptionsPerTurn: 4 }); expect(config.webSearchSidecar).toEqual({ reasoning: "high" }); expect(config.visionSidecar).toEqual({ maxDescriptionsPerTurn: 4 }); diff --git a/tests/web-search.test.ts b/tests/web-search.test.ts index 062fa6f050..c8e6d68471 100644 --- a/tests/web-search.test.ts +++ b/tests/web-search.test.ts @@ -2094,3 +2094,310 @@ describe("#398 sidecar failure degradation", () => { expect(raw.includes("LEAKMARKER_should_not_appear")).toBe(false); }); }); + +describe("web-search sidecar live streaming (streamRoutedModelOutput)", () => { + /** Incremental SSE frame reader so tests can observe delivery ORDER relative to adapter progress. */ + function frameReader(stream: ReadableStream) { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + let buffered = ""; + const frames: { event?: string; data: Record }[] = []; + const parse = (frame: string) => { + const lines = frame.split("\n"); + const event = lines.find(line => line.startsWith("event: "))?.slice(7); + const dataLine = lines.find(line => line.startsWith("data: ")); + if (dataLine?.slice(6) === "[DONE]") return undefined; + return { event, data: JSON.parse(dataLine?.slice(6) ?? "{}") as Record }; + }; + return { + frames, + /** Read until a frame matches, or the stream ends. Returns the matching frame or undefined. */ + async readUntil(match: (f: { event?: string; data: Record }) => boolean) { + for (const f of frames) if (match(f)) return f; + while (true) { + const { done, value } = await reader.read(); + if (done) return undefined; + buffered += decoder.decode(value, { stream: true }); + const parts = buffered.split("\n\n"); + buffered = parts.pop() ?? ""; + for (const part of parts) { + const trimmed = part.trim(); + if (!trimmed) continue; + const parsed = parse(trimmed); + if (!parsed) continue; + frames.push(parsed); + if (match(parsed)) return parsed; + } + } + }, + async drain() { + await this.readUntil(() => false); + return frames; + }, + }; + } + + const outputTextOf = (frames: { event?: string; data: Record }[]): string => + frames + .filter(f => f.data.type === "response.output_text.delta") + .map(f => String(f.data.delta ?? "")) + .join(""); + + /** + * Bound a readUntil wait with a deadline that REJECTS. The deadline must never release an + * adapter gate: doing so would let a buffered implementation pass via the terminal replay. + */ + const within = async (wait: Promise, what: string): Promise => { + let timer!: ReturnType; + const deadline = new Promise((_, reject) => { + timer = setTimeout( + () => reject(new Error(`timed out waiting for ${what} — output was not delivered live`)), + 5_000, + ); + }); + try { + return await Promise.race([wait, deadline]); + } finally { + clearTimeout(timer); + } + }; + + test("leading text deltas stream live: the client sees them while the adapter is still mid-turn", async () => { + // The adapter blocks after its first delta until the TEST has observed that delta on the wire. + // Buffered delivery would deadlock here; the rejecting 5s deadline turns that into a failure. + let releaseAdapter!: () => void; + const clientSawFirstDelta = new Promise(resolve => { releaseAdapter = resolve; }); + const adapter: ProviderAdapter = { + name: "gated", + buildRequest: () => ({ url: "https://routed.test/v1", method: "POST", headers: {}, body: "{}" }), + fetchResponse: async () => new Response("wire", { status: 200 }), + async *parseStream() { + yield { type: "text_delta", text: "Hello " } satisfies AdapterEvent; + await clientSawFirstDelta; + yield { type: "text_delta", text: "World" } satisfies AdapterEvent; + yield { type: "done" } satisfies AdapterEvent; + }, + async parseResponse() { + throw new Error("parseResponse must be unreachable"); + }, + }; + const response = await runWithWebSearch({ + parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }] }), + adapter, + forwardProvider, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + streamRoutedModelOutput: true, + }); + const sse = frameReader(response.body!); + const first = await within( + sse.readUntil(f => f.data.type === "response.output_text.delta"), + "the first live text delta", + ); + expect(first?.data.delta).toBe("Hello "); + releaseAdapter(); + const frames = await sse.drain(); + // Every delta exactly once — the terminal replay must skip what already streamed. + expect(outputTextOf(frames)).toBe("Hello World"); + expect(frames.some(f => f.event === "response.completed")).toBe(true); + }); + + test("leading reasoning deltas stream live: the client sees them while the adapter is still mid-turn", async () => { + // Same gate as the text test, but for the reasoning path: thinking_delta must reach the + // client as response.reasoning_summary_text.delta before the adapter is allowed to finish. + let releaseAdapter!: () => void; + const clientSawFirstReasoning = new Promise(resolve => { releaseAdapter = resolve; }); + const adapter: ProviderAdapter = { + name: "gated-reasoning", + buildRequest: () => ({ url: "https://routed.test/v1", method: "POST", headers: {}, body: "{}" }), + fetchResponse: async () => new Response("wire", { status: 200 }), + async *parseStream() { + yield { type: "thinking_delta", thinking: "Considering " } satisfies AdapterEvent; + await clientSawFirstReasoning; + yield { type: "thinking_delta", thinking: "options" } satisfies AdapterEvent; + yield { type: "text_delta", text: "Answer" } satisfies AdapterEvent; + yield { type: "done" } satisfies AdapterEvent; + }, + async parseResponse() { + throw new Error("parseResponse must be unreachable"); + }, + }; + const response = await runWithWebSearch({ + // Without reasoning.summary the parser sets hideThinkingSummary and no reasoning frame is + // ever client-visible; "auto" matches what Codex sends on real turns. + parsed: parseRequest({ + model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }], + reasoning: { summary: "auto" }, + }), + adapter, + forwardProvider, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + streamRoutedModelOutput: true, + }); + const sse = frameReader(response.body!); + const first = await within( + sse.readUntil(f => f.data.type === "response.reasoning_summary_text.delta"), + "the first live reasoning delta", + ); + expect(first?.data.delta).toBe("Considering "); + releaseAdapter(); + const frames = await sse.drain(); + // Each reasoning delta exactly once — the terminal replay must not duplicate the streamed head. + const reasoning = frames + .filter(f => f.data.type === "response.reasoning_summary_text.delta") + .map(f => String(f.data.delta ?? "")) + .join(""); + expect(reasoning).toBe("Considering options"); + expect(outputTextOf(frames)).toBe("Answer"); + expect(frames.some(f => f.event === "response.completed")).toBe(true); + }); + + test("default (flag unset) keeps full buffering: no text reaches the client before the adapter finishes", async () => { + let adapterFinished = false; + const adapter: ProviderAdapter = { + name: "paced", + buildRequest: () => ({ url: "https://routed.test/v1", method: "POST", headers: {}, body: "{}" }), + fetchResponse: async () => new Response("wire", { status: 200 }), + async *parseStream() { + yield { type: "text_delta", text: "Hello " } satisfies AdapterEvent; + await new Promise(resolve => setTimeout(resolve, 100)); + yield { type: "text_delta", text: "World" } satisfies AdapterEvent; + adapterFinished = true; + yield { type: "done" } satisfies AdapterEvent; + }, + async parseResponse() { + throw new Error("parseResponse must be unreachable"); + }, + }; + const response = await runWithWebSearch({ + parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }] }), + adapter, + forwardProvider, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + }); + const sse = frameReader(response.body!); + const first = await sse.readUntil(f => f.data.type === "response.output_text.delta"); + // By the time the FIRST delta is visible, the adapter must already be past its last delta. + expect(adapterFinished).toBe(true); + expect(first).toBeDefined(); + const frames = await sse.drain(); + expect(outputTextOf(frames)).toBe("Hello World"); + }); + + test("the live window closes at the first tool_call_start; the buffered tail replays once, in order", async () => { + // The adapter withholds the tool call until the TEST has seen "prefix " on the wire, so a + // buffered implementation (which delivers nothing before the terminal replay) deadlocks the + // gate instead of passing on identical final frames. + let releaseToolCall!: () => void; + const clientSawPrefix = new Promise(resolve => { releaseToolCall = resolve; }); + const adapter: ProviderAdapter = { + name: "tool-tail", + buildRequest: () => ({ url: "https://routed.test/v1", method: "POST", headers: {}, body: "{}" }), + fetchResponse: async () => new Response("wire", { status: 200 }), + async *parseStream() { + yield { type: "text_delta", text: "prefix " } satisfies AdapterEvent; + await clientSawPrefix; + yield { type: "tool_call_start", id: "call_1", name: "shell" } satisfies AdapterEvent; + yield { type: "tool_call_delta", arguments: "{\"cmd\":\"ls\"}" } satisfies AdapterEvent; + yield { type: "tool_call_end" } satisfies AdapterEvent; + yield { type: "text_delta", text: "suffix" } satisfies AdapterEvent; + yield { type: "done" } satisfies AdapterEvent; + }, + async parseResponse() { + throw new Error("parseResponse must be unreachable"); + }, + }; + const response = await runWithWebSearch({ + parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }] }), + adapter, + forwardProvider, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + streamRoutedModelOutput: true, + }); + const sse = frameReader(response.body!); + const prefixDelta = await within( + sse.readUntil(f => f.data.type === "response.output_text.delta"), + "the live prefix delta before the tool call", + ); + expect(prefixDelta?.data.delta).toBe("prefix "); + releaseToolCall(); + const frames = await sse.drain(); + expect(outputTextOf(frames)).toBe("prefix suffix"); + // The real tool call still reaches the client exactly once, and the replayed tail keeps + // wire order: prefix delta → function_call item → suffix delta. + const isCallAdd = (f: { data: Record }) => + f.data.type === "response.output_item.added" + && (f.data.item as Record | undefined)?.type === "function_call"; + expect(frames.filter(isCallAdd).length).toBe(1); + const prefixIdx = frames.findIndex(f => f.data.type === "response.output_text.delta" && f.data.delta === "prefix "); + const callIdx = frames.findIndex(isCallAdd); + const suffixIdx = frames.findIndex(f => f.data.type === "response.output_text.delta" && f.data.delta === "suffix"); + expect(prefixIdx).toBeGreaterThanOrEqual(0); + expect(callIdx).toBeGreaterThan(prefixIdx); + expect(suffixIdx).toBeGreaterThan(callIdx); + expect(frames.some(f => f.event === "response.completed")).toBe(true); + }); + + test("search loop: pre-search text streams live (documented tradeoff), the final answer arrives once", async () => { + // The first pass withholds its web_search call until the TEST has seen "Let me check. " on + // the wire — a buffered implementation would deadlock the gate rather than pass on final + // frames alone. + let releaseWebSearch!: () => void; + const clientSawPreSearchText = new Promise(resolve => { releaseWebSearch = resolve; }); + let pass = 0; + const adapter: ProviderAdapter = { + name: "search-then-answer", + buildRequest: () => ({ url: "https://routed.test/v1", method: "POST", headers: {}, body: "{}" }), + fetchResponse: async () => new Response("wire", { status: 200 }), + async *parseStream() { + if (pass++ === 0) { + yield { type: "text_delta", text: "Let me check. " } satisfies AdapterEvent; + await clientSawPreSearchText; + yield { type: "tool_call_start", id: "ws1", name: "web_search" } satisfies AdapterEvent; + yield { type: "tool_call_delta", arguments: "{\"query\":\"docs\"}" } satisfies AdapterEvent; + yield { type: "tool_call_end" } satisfies AdapterEvent; + yield { type: "done" } satisfies AdapterEvent; + } else { + yield { type: "text_delta", text: "Final answer." } satisfies AdapterEvent; + yield { type: "done" } satisfies AdapterEvent; + } + }, + async parseResponse() { + throw new Error("parseResponse must be unreachable"); + }, + }; + const response = await runWithWebSearch({ + parsed: parseRequest({ model: "routed/model", input: "hi", stream: true, tools: [{ type: "web_search" }] }), + adapter, + forwardProvider: { ...forwardProvider, baseUrl: "https://chatgpt.test/v1" }, + hostedTool: { type: "web_search" }, + selectedForwardHeaders: new Headers({ authorization: "Bearer token" }), + settings: { model: "gpt-5.6-luna", reasoning: "low", timeoutMs: 30_000 }, + maxSearches: 1, + streamRoutedModelOutput: true, + }); + const sse = frameReader(response.body!); + const preSearchDelta = await within( + sse.readUntil(f => f.data.type === "response.output_text.delta"), + "the live pre-search text delta", + ); + expect(preSearchDelta?.data.delta).toBe("Let me check. "); + releaseWebSearch(); + const frames = await sse.drain(); + const text = outputTextOf(frames); + // Pre-search text is visible exactly once, then the post-search answer exactly once. + expect(text).toBe("Let me check. Final answer."); + expect(frames.some(f => f.event === "response.completed")).toBe(true); + }); +});