From 978b27c2478e5dabc93829a1d920b84a09614fae Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 16 Sep 2026 02:12:22 +0000 Subject: [PATCH 1/3] feat(responses): add opt-in native WebSocket steering --- .../content/docs/guides/codex-integration.md | 35 ++ .../docs/reference/configuration/server.md | 1 + scripts/test-layout/layout.json | 3 +- src/config/schema/config-schema.ts | 1 + src/responses/state.ts | 4 + src/server/index/websocket-handler.ts | 36 ++- src/server/responses/codex-ws-exchange.ts | 50 ++- src/server/responses/core-options.ts | 3 + src/server/responses/fetch-helpers.ts | 5 +- src/server/responses/native-steering-log.ts | 44 +++ .../responses/native-steering-replay.ts | 111 +++++++ src/server/responses/native-steering.ts | 248 ++++++++++++++ src/server/responses/passthrough-delivery.ts | 11 + src/server/responses/passthrough-dispatch.ts | 27 +- src/server/responses/ws-upstream.ts | 9 +- src/server/ws-bridge.ts | 15 +- src/types/config.ts | 2 + structure/adapters/registry.md | 4 + structure/catalog.md | 4 + structure/clients/claude-desktop.md | 4 + structure/config.md | 4 + structure/data-planes/images.md | 4 + structure/data-planes/inbound-compat.md | 4 + structure/gui-and-management-api.md | 4 + structure/ops/docs-and-release.md | 4 + structure/ops/service-and-sidecars.md | 4 + structure/overview.md | 4 + structure/providers/chat-compat.md | 4 + structure/providers/kiro.md | 4 + structure/providers/xai-grok.md | 4 + structure/runtime.md | 4 + structure/subagents.md | 4 + structure/transports/byte-accounting.md | 4 + structure/transports/inventory.md | 4 + structure/transports/responses.md | 4 + structure/transports/streaming-health.md | 46 +++ tests/fixtures/test-layout-expected.json | 3 +- tests/responses/ws-native-steering.test.ts | 304 ++++++++++++++++++ 38 files changed, 1019 insertions(+), 11 deletions(-) create mode 100644 src/server/responses/native-steering-log.ts create mode 100644 src/server/responses/native-steering-replay.ts create mode 100644 src/server/responses/native-steering.ts create mode 100644 tests/responses/ws-native-steering.test.ts diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index 877117babe..1ed04eb2ba 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -878,3 +878,38 @@ When an affected history store supports paginated records, a provider transition `ocx restore` and Codex config removal still refuse on `history_paginated_requires_native_writer`. Stripping the `[model_providers.opencodex]` definition while thread rows still reference it would make those conversations unresolvable, and the restore path has no way to keep a compatibility provider table. A home that is already paginated cannot currently be uninstalled through the product; that is known open work rather than intended behaviour. Do not rewrite an active paginated rollout or thread row to migrate those conversations yourself. Close the affected conversation before any recovery, and report the exact error and versions without uploading private history. A backup or a successful script alone does not prove the conversation is visible again. Check the restored conversation in Codex after reopening. + +## Experimental native mid-turn steering + +For a compatible native OpenAI model and a client that sends `response.steer`, enable both +options in `~/.opencodex/config.json` and restart OpenCodex before starting a fresh turn: + +```json +{ + "websockets": true, + "codexNativeSteering": true +} +``` + +Merge these keys into the existing configuration; do not replace your provider/account settings. +This option is off by default. It forwards steering to the same native ChatGPT WebSocket +connection and selected account, preserving automatic successor responses and pending +saved-tool-result continuations. Acceptance means queued, not yet applied. + +For `response.steer.pending`, supply the reported required tool results or approval decisions +**once per parent**, on the same lane. Do not rerun tools or resend accepted steering text. +This first implementation accepts saved-result-only continuations with unchanged model and +request settings. A changed model/settings requires an explicitly stopped or finished turn +and normal new dispatch. Multiple independent conversations use independent connections. + +HTTP fallback, other providers, translated models, sidecars, Combo attempts and plaintext V2 +restoration do not support this option. It does not add steering capability to a model or +a client that lacks it. Unsupported routes return a protocol error rather than silently +ignoring input. Disconnected or timed-out delivery may be unknown: never automatically +resubmit tools or steering text. Pending controls time out after 90 seconds of inactivity; +saved-tool-result waits have a 30-minute cap. + +The implementation has synthetic protocol and regression coverage, not live Astra/client +certification. Keep the option disabled for production work until your client/model path +has been verified. Set `codexNativeSteering` to `false` and restart to restore the existing +single-response relay; no account or conversation files need to be deleted. diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 3fbf619df4..2f9939bb01 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -21,6 +21,7 @@ runs helper features around provider requests. | `connectTimeoutMs?` | `number` | `200000` | Per-attempt DNS/TCP/TLS/final-header deadline; it ends before body generation. | | `shutdownTimeoutMs?` | `number` | `5000` | Graceful drain deadline before active turns are aborted. | | `websockets?` | `boolean` | `false` | Advertise and admit the client-facing Responses WebSocket path. False keeps clients on HTTP/SSE; it does not disable an eligible canonical ChatGPT upstream WS optimization. Complete-input requests may reuse an upstream connection within the same selected credential, account, thread and turn; changed handshake policy or missing identity keeps requests on separate connections. This does not trim HTTP input or create previous-response IDs. | +| `codexNativeSteering?` | `boolean` | `false` | Experimental, native-only mid-turn steering on the Responses WebSocket endpoint. Requires `websockets: true`, a compatible upstream/client, and unchanged model/settings for saved-tool-result continuations. Does not enable translated models or HTTP fallback. See [native steering](/guides/codex-integration/#experimental-native-mid-turn-steering). | | `corsAllowOrigins?` | `string[]` | `[]` | Additional exact origins allowed by CORS. Loopback origins are always allowed. Authority-based browser extension origins such as `chrome-extension://` are supported; `*` is not a wildcard. Firefox and Safari regenerate the extension UUID (per install / per browser launch), so update the entry when the origin changes. | | `apiKeys?` | `OcxApiKey[]` | `[]` | Generated `ocx_…` credentials accepted by management and data-plane auth on non-loopback binds. Dashboard-managed. | | `storageCleanupPolicy?` | `StorageCleanupPolicy` | disabled | Opt-in archived-session cleanup policy. Never enabled implicitly. | diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index 885d2abcd4..a8eb9e98fe 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1456,7 +1456,8 @@ "execution-budget-permits.test.ts": "lib", "spend-instrumentation-log.test.ts": "server", "codex-pool-refresh-backoff.test.ts": "codex-integration", - "responses-account-change-scrub.test.ts": "responses" + "responses-account-change-scrub.test.ts": "responses", + "ws-native-steering.test.ts": "responses" }, "migrated": [ "adapters", diff --git a/src/config/schema/config-schema.ts b/src/config/schema/config-schema.ts index da959ce60d..8d66ebe732 100644 --- a/src/config/schema/config-schema.ts +++ b/src/config/schema/config-schema.ts @@ -59,6 +59,7 @@ import { parseDesktopProfile } from "../../claude/desktop-profile"; import { DEFAULT_APP_OWNED_MEMORY_BUDGET_BYTES, MAX_APP_OWNED_MEMORY_BUDGET_MB, MIN_APP_OWNED_MEMORY_BUDGET_MB } from "../../lib/app-owned-memory"; export const configSchema = z.object({ + codexNativeSteering: z.boolean().optional().catch(false), port: z.number().int().min(0).max(65535).default(10100), // A malformed hand edit must disable only remote-role behavior, not discard // providers or data-plane keys. Live writes are rejected explicitly below. diff --git a/src/responses/state.ts b/src/responses/state.ts index 43a1e3e43e..d48e7dcad0 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -1256,6 +1256,10 @@ export function markBodyNonPersistable(body: unknown): void { if (body && typeof body === "object") nonPersistableBodies.add(body as object); } +export function isBodyNonPersistable(body: unknown): boolean { + return !!body && typeof body === "object" && nonPersistableBodies.has(body); +} + export function rememberResponseState( requestBody: unknown, response: { id?: unknown; output?: unknown; status?: unknown; incomplete_details?: unknown }, diff --git a/src/server/index/websocket-handler.ts b/src/server/index/websocket-handler.ts index f94ca32e9c..584ee93557 100644 --- a/src/server/index/websocket-handler.ts +++ b/src/server/index/websocket-handler.ts @@ -1,3 +1,5 @@ +import { NativeSteeringChannel, NativeSteeringError } from "../responses/native-steering"; +import { createNativeSteeringLogObserver } from "../responses/native-steering-log"; import type { Server, ServerWebSocket } from "bun"; import { LIVE_SIDEBAND_UPSTREAM_OPEN_TIMEOUT_MS, @@ -187,11 +189,38 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { } catch { return; // text-only contract; ignore unparseable frames } + if (frame.type === "response.steer" || (frame.type === "response.create" && ws.data.nativeSteering)) { + try { + if (frame.type === "response.steer") { + if (!ws.data.nativeSteering) throw new NativeSteeringError("steering_not_supported", "Native steering is disabled or unavailable on this route."); + ws.data.nativeSteering.steer(frame); + return; + } + if (ws.data.nativeSteering?.continue(frame)) return; + } catch (error) { + sendJsonFrame(ws, buildWsErrorFrame(400, { + type: "invalid_request_error", + code: error instanceof NativeSteeringError ? error.code : "native_steering_error", + message: error instanceof NativeSteeringError ? error.message : "Native steering transport failed; delivery may be unknown. Do not automatically replay input.", + })); + return; + } + } if (frame.type === "response.processed") return; // ack — no-op if (frame.type !== "response.create") return; markActivity("ws response.create"); ws.data.cancel?.(); + let nativeSteering: NativeSteeringChannel | undefined; + try { + const idleMs = typeof config.stallTimeoutSec === "number" && Number.isFinite(config.stallTimeoutSec) + ? Math.max(1, config.stallTimeoutSec) * 1000 : 300_000; + nativeSteering = config.codexNativeSteering === true ? new NativeSteeringChannel(frame, idleMs) : undefined; + } catch { + sendJsonFrame(ws, buildWsErrorFrame(400, { type: "invalid_request_error", message: "Invalid native steering request settings" })); + return; + } + ws.data.nativeSteering = nativeSteering; const turnId = (ws.data.turnId ?? 0) + 1; ws.data.turnId = turnId; const isCurrent = () => ws.data.turnId === turnId; @@ -266,6 +295,7 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { ...(wsAdmission ? { admission: wsAdmission } : {}), forceEmptyResponseId: true, inboundTransport: "websocket", + nativeSteering, abortSignal: turnAbort.signal, turnAdmissionLease, onFirstOutput: () => recordFirstOutput(logCtx, start), @@ -276,7 +306,10 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { }, }); await sendResponseToWebSocket(ws, response, isCurrent, { - onSsePayload: payload => inspectResponseLogSsePayload(logCtx, payload), + untilEof: nativeSteering?.relayActive === true, + onSsePayload: nativeSteering?.relayActive + ? createNativeSteeringLogObserver(logCtx, () => recordFirstOutput(logCtx, start)) + : payload => inspectResponseLogSsePayload(logCtx, payload), onTerminal: status => { terminalRecorder?.(status, logCtx.terminalHttpStatus); finalizeLog(httpStatusForRequestLogTerminal(status, logCtx), { @@ -312,6 +345,7 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { } } finally { turnAdmissionLease.release(); + if (ws.data.nativeSteering === nativeSteering) ws.data.nativeSteering = undefined; if (!logged && turnAbort.signal.aborted) finalizeLog(499); if (ws.data.cancel === cancelTurn) ws.data.cancel = undefined; } diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 8073aff9d6..4b7b55f95d 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -1,3 +1,4 @@ +import { markNativeSteeringResponse, type NativeSteeringChannel } from "./native-steering"; import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer"; import { isSafeResponseHeader } from "../safe-response-headers"; import { CodexWsMetadata, type CodexWsQuotaObserver } from "./codex-ws-metadata"; @@ -6,10 +7,12 @@ import { CodexWsCorrelation } from "./codex-ws-correlation"; import type { CodexWsSession } from "./codex-ws-session"; import { UPGRADE_DEADLINE_MS, CODEX_WS_LIVENESS_PING_INTERVAL_MS, CODEX_WS_RESPONSE_PRELUDE_TIMEOUT_MS, MAX_CODEX_WS_FRAME_BYTES, MAX_CODEX_WS_QUEUE_BYTES, markCodexWsResponse, normalizeResponsesWsRelayEvent, closedBeforeTerminalMessage, - codexWsFailureDetail, codexWsPreResponseFailure, markCodexWsStage, codexWsOcxVersion, + codexWsCreateFrameExceedsLimit, codexWsFailureDetail, codexWsPreResponseFailure, markCodexWsStage, codexWsOcxVersion, type CodexWsFailureStage, type CodexWsStageRecord } from "./codex-ws-wire"; interface ExchangeOptions { + nativeSteering?: NativeSteeringChannel; + beforeContinuation?: () => Promise; session: CodexWsSession; url: string; init: RequestInit; @@ -86,7 +89,7 @@ function wrappedRejectionResponse(payload: Record, prelude: Hea /** The sole SSE exchange state machine for both one-shot and retained sockets. */ export function codexWsExchange(options: ExchangeOptions): Promise { - const { session, url, init, prepared, sseFallback, onQuota, beforeDispatch, bunVersion } = options; + const { session, url, init, prepared, sseFallback, onQuota, beforeDispatch, bunVersion, nativeSteering, beforeContinuation } = options; const { frameText, headers } = prepared; const signal = init.signal ?? undefined; return new Promise((resolve, reject) => { @@ -116,6 +119,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { const metadata = url === CODEX_RESPONSES_HTTP_URL ? new CodexWsMetadata(onQuota) : null; const correlation = session.retainable ? new CodexWsCorrelation(session.reused, id => session.hasCompleted(id)) : null; let detachOwner = () => {}; + let detachSteering = () => {}; // Liveness while waiting for the first response event (metadata path only): the // silence timer is re-armed by every inbound frame or pong; the pinger runs on a fixed // interval so a peer that answers pings can never trip the silence bound while alive. @@ -142,6 +146,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { metadata?.finish(); correlation?.finish(); detachOwner(); + detachSteering(); ws.removeEventListener("open", onOpen); ws.removeEventListener("message", onMessage); ws.removeEventListener("close", onClose); @@ -198,6 +203,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { const response = new Response(stream, { status: 200, headers: responseHeaders }); metadata?.commit(); markCodexWsResponse(response, Boolean(metadata && onQuota)); + if (nativeSteering) markNativeSteeringResponse(response); markCodexWsStage(response, stageRecord(null)); committedResponse = response; resolve(response); @@ -330,6 +336,38 @@ export function codexWsExchange(options: ExchangeOptions): Promise { if (terminal || settledPreOpen || signal?.aborted) return; sent = true; try { + if (nativeSteering) { + detachSteering = nativeSteering.attach(frame => { + const sendControl = () => { + if (terminal || signal?.aborted || session.closed || ws.readyState !== WebSocket.OPEN) { + throw new Error("Native steering connection is no longer available"); + } + beforeDispatch?.(new Headers(headers)); + let outgoing = frame; + if (frame.type === "response.create") { + // The channel validates same settings/lane and saved-result-only input. + // Reuse the already-routed/authorized native settings; never feed a + // previous_response_id through the REST sanitizer or account selector. + const base = JSON.parse(frameText) as Record; + outgoing = { ...base, input: frame.input, previous_response_id: frame.previous_response_id }; + } + const text = JSON.stringify(outgoing); + if (codexWsCreateFrameExceedsLimit(text)) { + throw new Error("Native steering frame exceeds the transport byte limit"); + } + try { ws.send(text); } catch { + // A send failure has unknown delivery. Never replay or fall back. + failStream("Native steering send failed; delivery is unknown"); + throw new Error("Native steering send failed; delivery is unknown"); + } + }; + if (frame.type === "response.create" && beforeContinuation) { + // Explicit tool-result continuations are physical request starts; + // they keep provider pacing and revalidate auth AFTER the wait. + void beforeContinuation().then(sendControl).catch(() => failStream("Native steering continuation could not be dispatched; do not automatically replay queued input")); + } else sendControl(); + }, error => failStream(error)); + } ws.send(frameText); sentAt = Date.now(); } catch { @@ -400,8 +438,12 @@ export function codexWsExchange(options: ExchangeOptions): Promise { return; } if (!controlFrame && !type.startsWith("response.") && type !== "error") return; + let steeringEnded = false; if (!controlFrame) { - try { correlation?.accept(normalized.payload); } catch (error) { failStream(error); return; } + try { + if (nativeSteering) steeringEnded = nativeSteering.observe(normalized.payload); + else correlation?.accept(normalized.payload); + } catch (error) { failStream(error); return; } // Correlation must run first: a reused socket's foreign-stream error settles as a // non-replayable 502 above, never as the refused-create 4xx projection below, which // is the one status family that could authorize an account replay. @@ -443,7 +485,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { return; } if (!controlFrame) relayedEvents += 1; - if (type === "response.completed" || type === "response.failed" || type === "response.incomplete" || type === "error") { + if (nativeSteering ? steeringEnded : (type === "response.completed" || type === "response.failed" || type === "response.incomplete" || type === "error")) { const completedId = correlation?.completed(normalized.payload) ?? null; terminal = true; cleanup(); diff --git a/src/server/responses/core-options.ts b/src/server/responses/core-options.ts index 19adf73412..0c84494a28 100644 --- a/src/server/responses/core-options.ts +++ b/src/server/responses/core-options.ts @@ -1,3 +1,4 @@ +import type { NativeSteeringChannel } from "./native-steering"; import type { OcxUsage, OcxProviderContinuationState, OcxConfig } from "../../types"; import type { CodexAuthPolicyConfig, CodexAuthContext } from "../../codex/auth-context"; import type { AdmissionLease } from "../../lib/admission"; @@ -51,6 +52,8 @@ export interface HandleResponsesOptions { /** Called at most once after the complete client body is read and accepted for dispatch. */ onRequestBodyRead?: () => void; forceEmptyResponseId?: boolean; + /** Internal, connection-owned control channel; never reconstructed from headers. */ + nativeSteering?: NativeSteeringChannel; abortSignal?: AbortSignal; /** One-shot TTFT callback: first non-empty model output observed (WP4). */ onFirstOutput?: () => void; diff --git a/src/server/responses/fetch-helpers.ts b/src/server/responses/fetch-helpers.ts index 00bdbdc0f2..2c867ba0d9 100644 --- a/src/server/responses/fetch-helpers.ts +++ b/src/server/responses/fetch-helpers.ts @@ -1,3 +1,4 @@ +import type { NativeSteeringChannel } from "./native-steering"; import type { Server } from "bun"; import { codexWsUpstreamFetch, @@ -53,6 +54,7 @@ export interface PaceAwareFetch { export type ProviderFetch = typeof globalThis.fetch & PaceAwareFetch; export interface ProviderFetchOptions { + nativeSteering?: NativeSteeringChannel; providerName?: string; modelId?: string; /** One pacing slot was acquired immediately before this fetch wrapper was created. */ @@ -101,7 +103,8 @@ export function providerFetch( // used, protocol pin included: a WS turn that falls back is serving the // request over HTTP, and dropping the provider's `upstreamHttpVersion` // there would silently negotiate a transport the operator ruled out. - return codexWsUpstreamFetch(input, init, httpFetch, runtime, options.onCodexWsQuota, options.beforeDispatch); + return codexWsUpstreamFetch(input, init, httpFetch, runtime, options.onCodexWsQuota, options.beforeDispatch, options.nativeSteering, + () => waitForPacing(init.signal ?? undefined)); } return httpFetch(input, init); }; diff --git a/src/server/responses/native-steering-log.ts b/src/server/responses/native-steering-log.ts new file mode 100644 index 0000000000..d36911d127 --- /dev/null +++ b/src/server/responses/native-steering-log.ts @@ -0,0 +1,44 @@ +import { inspectResponseLogSsePayload, usageFromResponsesPayload, type RequestLogContext } from "../request-log"; +import type { OcxUsage } from "../../types"; +import { MAX_NATIVE_STEERING_RESPONSES } from "./native-steering"; + +/** Count every terminal once. Control frames may echo user input: never sample them. */ +export function createNativeSteeringLogObserver(logCtx: RequestLogContext, onFirstOutput?: () => void): (payload: string) => void { + let outputSeen = false; + const usages = new Map(); + return payload => { + let event: { type?: string; delta?: unknown; response?: { id?: string; usage?: unknown; incomplete_details?: { reason?: string } } }; + try { event = JSON.parse(payload); } catch { return; } + if (event.type?.startsWith("response.steer.")) return; + if (!outputSeen && event.type?.endsWith(".delta") && typeof event.delta === "string" && event.delta.length) { + outputSeen = true; onFirstOutput?.(); + } + // A steered parent is not a failed logical turn. Keep its usage, but never + // label an eventual successful successor as an upstream failure. + if (!(event.type === "response.incomplete" && event.response?.incomplete_details?.reason === "steered")) { + inspectResponseLogSsePayload(logCtx, payload); + } + const terminal = ["response.completed", "response.failed", "response.incomplete"].includes(event.type ?? ""); + if (terminal && typeof event.response?.id === "string") { + const usage = usageFromResponsesPayload(event.response.usage); + if (usage && (usages.has(event.response.id) || usages.size < MAX_NATIVE_STEERING_RESPONSES)) { + // Retain numeric counters only, never arbitrary rawUsage metadata. + const counters: OcxUsage = { inputTokens: usage.inputTokens, outputTokens: usage.outputTokens }; + for (const key of ["totalTokens", "cachedInputTokens", "cacheReadInputTokens", "cacheCreationInputTokens", "reasoningOutputTokens"] as const) { + if (typeof usage[key] === "number") counters[key] = usage[key]; + } + usages.set(event.response.id, counters); + } + } + if (!usages.size) return; + const total: OcxUsage = { inputTokens: 0, outputTokens: 0 }; + for (const usage of usages.values()) { + for (const key of ["inputTokens", "outputTokens", "totalTokens", "cachedInputTokens", "cacheReadInputTokens", "cacheCreationInputTokens", "reasoningOutputTokens"] as const) { + const value = usage[key]; + if (typeof value === "number" && Number.isFinite(value)) total[key] = (total[key] ?? 0) + value; + } + } + logCtx.usage = total; + if (logCtx.activeAttempt) logCtx.activeAttempt.usage = total; + }; +} diff --git a/src/server/responses/native-steering-replay.ts b/src/server/responses/native-steering-replay.ts new file mode 100644 index 0000000000..f4149cef00 --- /dev/null +++ b/src/server/responses/native-steering-replay.ts @@ -0,0 +1,111 @@ +/** + * Connection-local replay journal. Only input committed by response.created enters + * a successor's prefix. Uncommitted/rejected steering never enters the shared + * continuation cache. Bodies are bounded and discarded at connection teardown. + */ +export const MAX_NATIVE_STEERING_REPLAY_BYTES = 32 * 1024 * 1024; +type Frame = Record; +function record(value: unknown): value is Frame { + return value !== null && typeof value === "object" && !Array.isArray(value); +} +function inputItems(input: unknown): unknown[] { + if (typeof input === "string") return [{ type: "message", role: "user", content: [{ type: "input_text", text: input }] }]; + return Array.isArray(input) ? input : []; +} +export interface NativeSteeringReplayObserver { + submitted(frame: Frame): () => void; + observe(frame: Frame): void; + dispose(): void; +} + +export class NativeSteeringReplay implements NativeSteeringReplayObserver { + private prefix: unknown[]; + private bytes: number; + private current?: string; + private previousOutput: unknown[] = []; + private outputItems = new Map(); + private submissions: Array<{ parent: string; input: unknown[]; id?: string; bytes: number }> = []; + private explicitInput: unknown[] = []; + private explicitBytes = 0; + + constructor(input: unknown, private readonly remember: (input: unknown[], response: Frame) => void) { + this.prefix = [...inputItems(input)]; + this.bytes = Buffer.byteLength(JSON.stringify(this.prefix)); + this.check(); + } + private check(): void { + if (this.bytes > MAX_NATIVE_STEERING_REPLAY_BYTES) throw new Error("Native steering replay exceeded its bounded history budget; input was not silently truncated."); + } + submitted(frame: Frame): () => void { + const input = inputItems(frame.input); + const bytes = Buffer.byteLength(JSON.stringify(input)); + this.bytes += bytes; + try { this.check(); } catch (error) { this.bytes -= bytes; throw error; } + if (frame.type === "response.steer") { + const submission = { parent: String(frame.previous_response_id), input, bytes }; + this.submissions.push(submission); + return () => { + const index = this.submissions.indexOf(submission); + if (index >= 0) { this.submissions.splice(index, 1); this.bytes -= bytes; } + }; + } + this.explicitInput = input; + this.explicitBytes = bytes; + return () => { this.explicitInput = []; this.bytes -= this.explicitBytes; this.explicitBytes = 0; }; + } + observe(frame: Frame): void { + const response = record(frame.response) ? frame.response : undefined; + const steer = record(frame.steer) ? frame.steer : undefined; + if (frame.type === "response.steer.accepted") { + const first = this.submissions.find(item => item.parent === steer?.previous_response_id && item.id === undefined); + if (!first || typeof steer?.id !== "string") throw new Error("Native steering replay acceptance does not match submitted input"); + first.id = steer.id; + } else if (frame.type === "response.steer.failed") { + const index = this.submissions.findIndex(item => steer?.id !== undefined + ? item.id === steer.id + : item.parent === steer?.previous_response_id && item.id === undefined); + if (index >= 0) { + const [failed] = this.submissions.splice(index, 1); + this.bytes -= failed.bytes; + } + } else if (frame.type === "response.created") { + if (this.current) { + const committed = this.submissions.filter(item => item.parent === this.current && item.id !== undefined); + this.prefix.push(...this.previousOutput, ...committed.flatMap(item => item.input), ...this.explicitInput); + this.submissions = this.submissions.filter(item => !committed.includes(item)); + } + this.explicitInput = []; + this.explicitBytes = 0; + this.previousOutput = []; + this.outputItems.clear(); + this.current = String(response?.id); + } else if (frame.type === "response.output_item.done" && Number.isSafeInteger(frame.output_index)) { + const index = frame.output_index as number; + if (index < 0 || index > 10_000 || !record(frame.item)) throw new Error("Native steering replay output identity is invalid"); + const previous = this.outputItems.get(index); + if (previous !== undefined) this.bytes -= Buffer.byteLength(JSON.stringify(previous)); + this.bytes += Buffer.byteLength(JSON.stringify(frame.item)); + this.check(); + this.outputItems.set(index, frame.item); + } else if (response && ["response.completed", "response.incomplete", "response.failed"].includes(String(frame.type))) { + const doneItems = [...this.outputItems.entries()].sort((a, b) => a[0] - b[0]).map(([, item]) => item); + const output = Array.isArray(response.output) && response.output.length ? response.output : doneItems; + for (const item of doneItems) this.bytes -= Buffer.byteLength(JSON.stringify(item)); + this.bytes += Buffer.byteLength(JSON.stringify(output)); + this.check(); + this.outputItems.clear(); + this.previousOutput = output; + // Failed and steered parents are never presented to shared state as completed. + // Their output is used only when a validated successor commits that prefix. + if (frame.type === "response.completed") this.remember(this.prefix, { ...response, output }); + } + } + dispose(): void { + this.prefix = []; + this.previousOutput = []; + this.submissions = []; + this.explicitInput = []; + this.outputItems.clear(); + this.bytes = 0; + } +} diff --git a/src/server/responses/native-steering.ts b/src/server/responses/native-steering.ts new file mode 100644 index 0000000000..3c5e658fee --- /dev/null +++ b/src/server/responses/native-steering.ts @@ -0,0 +1,248 @@ +import type { NativeSteeringReplayObserver } from "./native-steering-replay"; +import { createHash } from "node:crypto"; +import { CODEX_WS_ID_MAX_BYTES, CodexWsCorrelation } from "./codex-ws-correlation"; +import { codexWsCreateFrameExceedsLimit } from "./codex-ws-wire"; + +export const MAX_NATIVE_STEERS = 32; +export const MAX_NATIVE_STEERING_RESPONSES = 128; +export const NATIVE_STEERING_WAIT_MS = 90_000; +export const NATIVE_STEERING_TOOL_WAIT_MS = 30 * 60_000; + +type Frame = Record; +type Send = (frame: Frame) => void; +const responses = new WeakSet(); +export function markNativeSteeringResponse(response: Response): Response { responses.add(response); return response; } +export function isNativeSteeringResponse(response: Response): boolean { return responses.has(response); } + +function record(value: unknown): value is Frame { + return value !== null && typeof value === "object" && !Array.isArray(value); +} +function validId(value: unknown): value is string { + return typeof value === "string" && value.length > 0 && Buffer.byteLength(value) <= CODEX_WS_ID_MAX_BYTES + && !/[\u0000-\u001f\u007f]/.test(value); +} +function stable(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(stable).join(",")}]`; + if (record(value)) return `{${Object.keys(value).sort().map(key => `${JSON.stringify(key)}:${stable(value[key])}`).join(",")}}`; + return JSON.stringify(value) ?? "null"; +} +function fingerprint(value: unknown): string { return createHash("sha256").update(stable(value)).digest("hex"); } + +/** Typed, content-free local protocol rejection. */ +export class NativeSteeringError extends Error { + constructor(readonly code: string, message: string) { super(message); this.name = "NativeSteeringError"; } +} + +/** Validate the narrow public steer envelope, not the much wider create schema. */ +export function validateSteeringFrame(frame: Frame): void { + const fail = () => { throw new NativeSteeringError("invalid_input", "Steering requires user-only input and a previous_response_id; extra envelope fields are not supported."); }; + if (frame.type !== "response.steer" || !validId(frame.previous_response_id) + || Object.keys(frame).some(key => !["type", "previous_response_id", "input"].includes(key))) fail(); + if (typeof frame.input === "string") return; + if (!Array.isArray(frame.input) || !frame.input.length) { fail(); return; } + for (const item of frame.input) { + if (!record(item) || item.role !== "user" || (item.type !== undefined && item.type !== "message") + || Object.keys(item).some(key => !["type", "role", "content"].includes(key))) { fail(); return; } + if (typeof item.content === "string") continue; + if (!Array.isArray(item.content) || !item.content.length) { fail(); return; } + for (const part of item.content) { + if (!record(part) || !["input_text", "input_image", "input_file"].includes(String(part.type)) + || (part.type === "input_text" && typeof part.text !== "string")) fail(); + } + } +} + +type Parent = { unacknowledged: number; accepted: Set; ended: boolean }; + +/** + * One downstream turn owns one dedicated native upstream socket, including all + * automatic successors and required-input continuations. Never registered by a + * caller-supplied response ID in global state; never lent to another account. + * + * Opt-in single-lane implementation. Continuations may supply saved tool results + * but cannot change settings/routing. General new turns still use normal dispatch. + */ +export class NativeSteeringChannel { + relayActive = false; + replayFactory?: () => NativeSteeringReplayObserver; + private replay?: NativeSteeringReplayObserver; + private send?: Send; + private onFailure?: (error: Error) => void; + private timer?: ReturnType; + private readonly parents = new Map(); + private readonly settings = new Map(); + private currentId?: string; + private correlation?: CodexWsCorrelation; + private pendingParent?: string; + private continuationSent = false; + private lane: unknown; + private finished = false; + private everAttached = false; + private required: Frame[] = []; + + constructor(initial: Frame, private readonly idleMs = 300_000) { + this.lane = initial.stream_id; + for (const [key, value] of Object.entries(initial)) { + if (!["type", "input", "previous_response_id", "stream", "stream_id"].includes(key)) this.settings.set(key, fingerprint(value)); + } + } + get attached(): boolean { return this.everAttached; } + get ended(): boolean { return this.finished; } + get hasOutstanding(): boolean { + return [...this.parents.values()].some(parent => parent.unacknowledged > 0 || parent.accepted.size > 0); + } + get awaitingContinuation(): boolean { return this.pendingParent !== undefined; } + + /** Bind only after native credentials have been selected and admission passed. */ + attach(send: Send, onFailure: (error: Error) => void): () => void { + if (this.send || this.currentId) throw new Error("native steering transport already owned"); + this.replay = this.replayFactory?.(); + this.everAttached = true; + this.finished = false; + this.send = send; + this.onFailure = onFailure; + return () => { + if (this.send !== send) return; + this.send = undefined; + this.onFailure = undefined; + clearTimeout(this.timer); + this.timer = undefined; + this.correlation?.finish(); + this.correlation = undefined; + this.parents.clear(); + this.required = []; + this.replay?.dispose(); + this.replay = undefined; + }; + } + + private wait(ms = NATIVE_STEERING_WAIT_MS): void { + clearTimeout(this.timer); + this.timer = setTimeout(() => { + this.onFailure?.(new Error("Native steering continuation timed out; queued-input delivery is unknown. Do not automatically replay tools or steering input.")); + }, ms); + this.timer.unref?.(); + } + private liveSend(frame: Frame): void { + if (!this.send || this.finished) throw new NativeSteeringError("steering_not_supported", "No active native WebSocket steering transport; this route may be disabled or using HTTP fallback."); + if (codexWsCreateFrameExceedsLimit(JSON.stringify(frame))) throw new NativeSteeringError("invalid_input", "Native steering frame exceeds the upstream byte limit."); + const rollback = this.replay?.submitted(frame); + try { this.send(frame); } catch (error) { rollback?.(); throw error; } + } + + steer(frame: Frame): void { + validateSteeringFrame(frame); + if (!this.send || this.finished) { this.liveSend(frame); return; } + const target = this.parents.get(frame.previous_response_id as string); + if (!target || frame.previous_response_id !== this.currentId || target.ended) { + throw new NativeSteeringError("response_not_active", "The target response is not active on this connection."); + } + if ([...this.parents.values()].reduce((n, p) => n + p.unacknowledged + p.accepted.size, 0) >= MAX_NATIVE_STEERS) { + throw new NativeSteeringError("too_many_pending_steers", "The native steering pending-submission limit was reached."); + } + // Count before send: fake transports, and some runtimes, deliver synchronously. + target.unacknowledged += 1; + this.wait(); + try { this.liveSend(frame); } catch (error) { + target.unacknowledged -= 1; + if (!this.hasOutstanding) { clearTimeout(this.timer); this.timer = undefined; } + throw error; + } + } + + /** Returns false only for an unrelated ordinary create after this chain ends. */ + continue(frame: Frame): boolean { + if (this.finished || !this.send) return false; + if (!this.pendingParent || !this.currentId || frame.previous_response_id !== this.pendingParent) { + if (this.hasOutstanding) throw new NativeSteeringError("steering_continuation_required", "Queued steering owns this connection; wait for the successor or send the required-input continuation, or explicitly stop the turn."); + return false; + } + if (this.continuationSent) throw new NativeSteeringError("duplicate_continuation", "A required-input continuation was already sent for this parent."); + if (frame.stream_id !== this.lane || frame.generate === false) throw new NativeSteeringError("invalid_input", "The continuation must use the same WebSocket lane and generate a response."); + for (const [key, value] of Object.entries(frame)) { + if (["type", "input", "previous_response_id", "stream", "stream_id"].includes(key)) continue; + if (this.settings.get(key) !== fingerprint(value)) throw new NativeSteeringError("steering_settings_changed", "The experimental native steering continuation cannot change model or request settings; start a separate turn instead."); + } + // Restrict this bypass to exactly the server-identified outputs/approvals. + // General messages/settings continue through ordinary validation and routing. + const input = frame.input; + if (!Array.isArray(input) || input.length !== this.required.length) throw new NativeSteeringError("invalid_input", "Supply the saved results for the required_input stubs exactly once; do not resend steering text."); + const used = new Set(); + for (const item of input) { + const match = record(item) ? this.required.findIndex((stub, i) => !used.has(i) + && Object.entries(stub).every(([key, value]) => stable(item[key]) === stable(value))) : -1; + if (match < 0) throw new NativeSteeringError("invalid_input", "Continuation input must match the pending tool-output or approval stubs."); + used.add(match); + } + this.continuationSent = true; + this.wait(); + try { this.liveSend(frame); } catch (error) { this.continuationSent = false; throw error; } + return true; + } + + /** Called on the ordered upstream wire, BEFORE the event is published to SSE. */ + observe(event: Frame): boolean { + const type = event.type; + if (!(type === "error" && event.stream_id == null) && (event.stream_id ?? undefined) !== (this.lane ?? undefined)) throw new Error("native steering WebSocket lane mismatch"); + const response = record(event.response) ? event.response : undefined; + if (type === "response.created") { + const id = response?.id; + if (!validId(id) || this.parents.has(id) || this.parents.size >= MAX_NATIVE_STEERING_RESPONSES) throw new Error("native steering response identity or chain limit violated"); + if (this.currentId) { + const parent = this.parents.get(this.currentId)!; + if (!parent.ended || (!parent.accepted.size && !this.continuationSent)) throw new Error("unexpected native steering successor"); + if (response?.previous_response_id != null && response.previous_response_id !== this.currentId) throw new Error("native steering successor parent mismatch"); + parent.accepted.clear(); // response.created, not accepted, is the commit point. + } + this.currentId = id; + this.parents.set(id, { unacknowledged: 0, accepted: new Set(), ended: false }); + this.pendingParent = undefined; + this.required = []; + this.continuationSent = false; + this.correlation?.finish(); + this.correlation = new CodexWsCorrelation(true, () => false); + } + if (typeof type === "string" && type.startsWith("response.steer.")) { + const steer = record(event.steer) ? event.steer : undefined; + const parent = typeof steer?.previous_response_id === "string" ? this.parents.get(steer.previous_response_id) : undefined; + if (!parent) throw new Error("native steering acknowledgement has an unknown parent"); + if (type === "response.steer.accepted") { + if (!validId(steer?.id) || parent.unacknowledged < 1 || parent.accepted.has(steer.id)) throw new Error("unexpected native steering acceptance"); + parent.unacknowledged -= 1; + parent.accepted.add(steer.id); + } else if (type === "response.steer.failed") { + if (steer?.id !== undefined) { + if (!validId(steer.id) || !parent.accepted.delete(steer.id)) throw new Error("unexpected native steering failure"); + } else { + if (parent.unacknowledged < 1) throw new Error("unexpected native steering rejection"); + parent.unacknowledged -= 1; + } + } else if (type === "response.steer.pending") { + if (!validId(steer?.id) || !parent.accepted.has(steer.id) || !parent.ended + || steer.previous_response_id !== this.currentId) throw new Error("unexpected native steering pending event"); + if (event.reason === "waiting_for_required_input") { + if (!Array.isArray(event.required_input) || !event.required_input.length || event.required_input.length > 1024 + || event.required_input.some(item => !record(item) || typeof item.type !== "string" || item.type === "message") + || Buffer.byteLength(JSON.stringify(event.required_input)) > 256 * 1024) throw new Error("native steering required-input budget or schema violated"); + if (this.pendingParent && stable(this.required) !== stable(event.required_input)) throw new Error("native steering required-input stubs changed"); + this.pendingParent = this.currentId; + this.required = event.required_input as Frame[]; + } + // Unknown reasons are preserved, not converted into a create or success. + } else throw new Error("unsupported native steering control event"); + } else { + this.correlation?.accept({ ...event, stream_id: undefined }); + if (type === "response.completed" || type === "response.failed" || type === "response.incomplete") { + if (!this.currentId || response?.id !== this.currentId) throw new Error("native steering terminal identity mismatch"); + this.parents.get(this.currentId)!.ended = true; + } + } + if (type === "error") this.finished = true; + else this.finished = this.currentId !== undefined && this.parents.get(this.currentId)!.ended && !this.hasOutstanding; + if (this.finished || !this.hasOutstanding) { clearTimeout(this.timer); this.timer = undefined; } + else this.wait(this.pendingParent && !this.continuationSent ? NATIVE_STEERING_TOOL_WAIT_MS : NATIVE_STEERING_WAIT_MS); + if (!this.finished && !this.hasOutstanding && this.currentId) this.wait(this.idleMs); + this.replay?.observe(event); + return this.finished; + } +} diff --git a/src/server/responses/passthrough-delivery.ts b/src/server/responses/passthrough-delivery.ts index 79e6ca3d46..b6c5c2912b 100644 --- a/src/server/responses/passthrough-delivery.ts +++ b/src/server/responses/passthrough-delivery.ts @@ -1,3 +1,4 @@ +import { isNativeSteeringResponse } from "./native-steering"; import type { ResponsesRequestContext, ResponsesAdmissionState } from "./core-options"; import type { PreparedResponsesRequest } from "./request-prepare"; import type { ResponsesTransport } from "./request-transport"; @@ -301,6 +302,16 @@ export async function deliverPassthroughResponse( }); } + if (options.nativeSteering && isNativeSteeringResponse(upstreamResponse) && upstreamResponse.body) { + // A native chain carries several response terminals. Ordinary SSE repair, + // cancellation-on-terminal and local previous-response replay are single-response + // contracts and would truncate it. Keep the bounded upstream as the sole reader. + options.nativeSteering.relayActive = true; + commitReasoningReplayServingRoute(nativeExchange.request.headers); + const body = trackStreamLifetime(upstreamResponse.body, upstream, undefined, options.turnAdmissionLease); + return new Response(body, { status: upstreamResponse.status, headers }); + } + // Bun#32111 workaround: passthrough SSE uses tee()+native relay to avoid the // async-pull segfault on Windows. Branch[0] goes directly to the Response (Bun // native relay, never enters JS Sink.write); branch[1] is consumed in the diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index f5c5b94694..5b5e03b779 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -1,3 +1,4 @@ +import { NativeSteeringReplay } from "./native-steering-replay"; import type { ResponsesRequestContext, ResponsesAdmissionState, @@ -10,7 +11,7 @@ import type { ResponsesSendBudget } from "./request-send-budget"; import { isCanonicalOpenAiForwardProvider } from "../../providers/openai-tiers"; import { codexSafetyBufferingFilterOptions, terminalStatusFromParsed } from "../relay"; import { imageGenToolCallAliases } from "../responses-image-gen-repair"; -import { rememberResponseState } from "../../responses/state"; +import { rememberResponseState, isBodyNonPersistable } from "../../responses/state"; import { currentTurnWireToolCatalogBody, hasExplicitWireToolCatalog, @@ -248,6 +249,15 @@ export async function preparePassthroughExchange( ? (response: { id?: unknown; output?: unknown; status?: unknown }) => rememberResponseState(parsed._rawBody, response, undefined, responseStateOptions(true)) : undefined; + if (options.nativeSteering && isCanonicalOpenAiForwardProvider(route.provider) + && options.inboundTransport === "websocket" && !options.comboAttempt) { + const body = parsed._rawBody as Record; + options.nativeSteering.replayFactory = () => new NativeSteeringReplay(body.input, (input, response) => { + if (passthroughRecordEligible && !isBodyNonPersistable(body)) { + rememberResponseState({ ...body, input }, response, undefined, responseStateOptions(true)); + } + }); + } if (parsed.previousResponseId && !parsed._previousResponseInputExpanded) { console.warn( `[responses] previous_response_id ${parsed.previousResponseId} not found in local replay state ` @@ -759,6 +769,9 @@ export async function preparePassthroughExchange( body: request.body, }, recovery), upstream.signal, connectMs, parsed.stream, providerFetch(route.provider, options.codexWsRuntimeIdentity, { + nativeSteering: isCanonicalOpenAiForwardProvider(route.provider) && options.inboundTransport === "websocket" && !options.comboAttempt + && responseEffects.plaintextV2AgentMessageToolNames.size === 0 + ? options.nativeSteering : undefined, dispatchOverride: oauthDispatch(request), providerName: route.providerName, modelId: route.modelId, @@ -855,6 +868,9 @@ export async function preparePassthroughExchange( body: request.body, }, innerRecovery), upstream.signal, connectMs, parsed.stream, providerFetch(route.provider, options.codexWsRuntimeIdentity, { + nativeSteering: isCanonicalOpenAiForwardProvider(route.provider) && options.inboundTransport === "websocket" && !options.comboAttempt + && responseEffects.plaintextV2AgentMessageToolNames.size === 0 + ? options.nativeSteering : undefined, dispatchOverride: oauthDispatch(request), providerName: route.providerName, modelId: route.modelId, @@ -961,6 +977,9 @@ export async function preparePassthroughExchange( // here on is a genuine transport attempt. storedPoolReplayDispatchNotifier( providerFetch(route.provider, options.codexWsRuntimeIdentity, { + nativeSteering: isCanonicalOpenAiForwardProvider(route.provider) && options.inboundTransport === "websocket" && !options.comboAttempt + && responseEffects.plaintextV2AgentMessageToolNames.size === 0 + ? options.nativeSteering : undefined, dispatchOverride: oauthDispatch(request), providerName: route.providerName, modelId: route.modelId, @@ -1082,6 +1101,9 @@ export async function preparePassthroughExchange( body: request.body, }, recovery), upstream.signal, connectMs, parsed.stream, providerFetch(route.provider, options.codexWsRuntimeIdentity, { + nativeSteering: isCanonicalOpenAiForwardProvider(route.provider) && options.inboundTransport === "websocket" && !options.comboAttempt + && responseEffects.plaintextV2AgentMessageToolNames.size === 0 + ? options.nativeSteering : undefined, dispatchOverride: oauthDispatch(request), providerName: route.providerName, modelId: route.modelId, @@ -1199,6 +1221,9 @@ export async function preparePassthroughExchange( body: request.body, }, recovery), upstream.signal, connectMs, parsed.stream, providerFetch(route.provider, options.codexWsRuntimeIdentity, { + nativeSteering: isCanonicalOpenAiForwardProvider(route.provider) && options.inboundTransport === "websocket" && !options.comboAttempt + && responseEffects.plaintextV2AgentMessageToolNames.size === 0 + ? options.nativeSteering : undefined, dispatchOverride: oauthDispatch(request), providerName: route.providerName, modelId: route.modelId, diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index efa5dbb466..31857e38d5 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -1,3 +1,4 @@ +import type { NativeSteeringChannel } from "./native-steering"; // Upstream WebSocket transport for the ChatGPT Codex backend. // // Why this exists: the Codex backend serves the responses_websockets path from @@ -129,6 +130,8 @@ export function codexWsUpstreamFetch( runtime: BunRuntimeGateInput = currentBunRuntimeIdentity(), onQuota?: CodexWsQuotaObserver, beforeDispatch?: (headers: Headers) => void, + nativeSteering?: NativeSteeringChannel, + beforeContinuation?: () => Promise, ): Promise { const prepared = prepareCodexWsRequest(url, init); if (!prepared) return sseFallback(url, prepareCodexHttpInit(url, init)); @@ -169,7 +172,9 @@ export function codexWsUpstreamFetch( } let session: CodexWsSession; try { - const identity = codexWsReuseIdentity(url, headers, frameText, proxy); + // Steering keeps a private physical connection across successor responses; it + // must never enter the idle-socket pool or move to a different credential. + const identity = nativeSteering && prepared.canonical ? null : codexWsReuseIdentity(url, headers, frameText, proxy); session = (identity ? codexWsPool.acquire(identity, wsUrl, headers, proxy) : null) ?? new CodexWsSession(wsUrl, headers, false, undefined, proxy); if (!session.busy && !session.reserve()) { @@ -181,6 +186,8 @@ export function codexWsUpstreamFetch( } return codexWsExchange({ session, url, init, prepared, sseFallback, onQuota, beforeDispatch, + nativeSteering: prepared.canonical ? nativeSteering : undefined, + beforeContinuation, bunVersion: typeof runtime === "string" ? runtime : runtime.version, }); } diff --git a/src/server/ws-bridge.ts b/src/server/ws-bridge.ts index d401ddeba8..451272e707 100644 --- a/src/server/ws-bridge.ts +++ b/src/server/ws-bridge.ts @@ -1,3 +1,4 @@ +import type { NativeSteeringChannel } from "./responses/native-steering"; import type { ServerWebSocket } from "bun"; import { responsesJsonEventSequence } from "./responses-json-events"; import { FORWARD_HEADERS } from "../adapters/openai-responses"; @@ -17,6 +18,7 @@ type ResponsesTerminalReporter = (status: ResponsesTerminalStatus) => void; type ResponsesPayloadObserver = (payload: string) => void; export interface WsData { + nativeSteering?: NativeSteeringChannel; headers?: Headers; // base inbound forward headers only; per-turn auth refresh injects current pool tokens /** * Resolved once at the handshake. Auth is handshake-time only on this path, so @@ -229,6 +231,7 @@ export async function pumpResponsesSseToWebSocket( sseStream: ReadableStream, options: { isCurrent?: () => boolean; + untilEof?: boolean; onTerminal?: ResponsesTerminalReporter; onSsePayload?: ResponsesPayloadObserver; } = {}, @@ -251,6 +254,7 @@ export async function pumpResponsesSseToWebSocket( const decoder = new TextDecoder(); const framer = new BoundedSseFrameBuffer(); let terminalSeen = false; + let lastTerminal: ResponsesTerminalStatus | undefined; const handlePayload = (payload: string): boolean => { if (!isCurrent()) return true; @@ -270,8 +274,10 @@ export async function pumpResponsesSseToWebSocket( } if (terminalSeen) return true; sendTextFrame(ws, payload); - const terminalStatus = terminalStatusFromType(type); + if (options.untilEof && type === "response.created") lastTerminal = undefined; + const terminalStatus = type === "error" && options.untilEof ? "failed" : terminalStatusFromType(type); if (terminalStatus) { + if (options.untilEof) { lastTerminal = terminalStatus; return false; } reportTerminal(terminalStatus); terminalSeen = true; void reader.cancel().catch(() => {}); @@ -294,6 +300,10 @@ export async function pumpResponsesSseToWebSocket( const payload = parseSseBlock(decoder.decode(tail)); if (payload) handlePayload(payload); } + if (options.untilEof && lastTerminal && isCurrent() && !clientCancelled) { + reportTerminal(lastTerminal); + terminalSeen = true; + } if (!terminalSeen && isCurrent() && !clientCancelled) { reportTerminal("incomplete"); sendProtocolError(ws, 502, "Upstream stream ended before response terminal event"); @@ -368,6 +378,7 @@ export async function sendResponseToWebSocket( response: Response, isCurrent: () => boolean, options: { + untilEof?: boolean; onTerminal?: ResponsesTerminalReporter; onSsePayload?: ResponsesPayloadObserver; } = {}, @@ -398,6 +409,7 @@ export async function sendResponseToWebSocket( if (contentType.includes("text/event-stream")) { await pumpResponsesSseToWebSocket(ws, response.body, { isCurrent, + untilEof: options.untilEof, onTerminal: options.onTerminal, onSsePayload: options.onSsePayload, }); @@ -420,6 +432,7 @@ export async function sendResponseToWebSocket( if (looksLikeSse(prefix)) { await pumpResponsesSseToWebSocket(ws, stream, { isCurrent, + untilEof: options.untilEof, onTerminal: options.onTerminal, onSsePayload: options.onSsePayload, }); diff --git a/src/types/config.ts b/src/types/config.ts index bcc21c825d..4512f254cd 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -749,6 +749,8 @@ export interface OcxConfig { shutdownTimeoutMs?: number; /** Advertise supports_websockets so Codex opens the WS endpoint. Default false; set true to opt in. */ websockets?: boolean; + /** Experimental single-lane native OpenAI WebSocket steering; default off. */ + codexNativeSteering?: boolean; /** * Opt-in auto-cleanup policy for archived Codex sessions (issue #42 Phase 3). * Default OFF (`enabled` false / unset). Never enabled implicitly. diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md index 83e8f4f466..c99d2358ae 100644 --- a/structure/adapters/registry.md +++ b/structure/adapters/registry.md @@ -183,3 +183,7 @@ implement legacy call/result pairing. Modern tool-image carriers are unchanged. raw passthrough; `tests/responses/chat-media-translation.test.ts` reaches the real HTTP translation boundary and verifies that rejection sends no upstream request. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/catalog.md b/structure/catalog.md index bf9d6e3751..d55c40cecb 100644 --- a/structure/catalog.md +++ b/structure/catalog.md @@ -395,3 +395,7 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara ## Renamed destination reasoning metadata `src/providers/derive.ts` fills missing reasoning tables for renamed providers accepted by the existing fixed-key destination matcher. Model entries are cloned and explicit user entries (including empty arrays) win. Provider-wide effort defaults fill only when undefined; Command Code unknown models therefore keep the registry's empty picker policy unless overridden. Identity, transport and other capability axes are unchanged. The gathered row drives client exports; this metadata contract does not prove arbitrary gateway routing. + +The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 01a583c182..9d7d47fee2 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -157,3 +157,7 @@ The [explicit model-capability contract](../config.md#explicit-per-model-capabil Exact [model input declarations](../config.md#explicit-per-model-capability-declarations) now feed text-only eligibility and catalog hints; existing image-description/omission handling consumes them before the main upstream send. Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/config.md b/structure/config.md index d00b36b06b..7d46729dc7 100644 --- a/structure/config.md +++ b/structure/config.md @@ -310,3 +310,7 @@ The text-only consumer reads exact inputModalities declarations before legacy hi ## Catalog auto-refresh `catalogAutoRefresh` on `src/types/config.ts` stores an optional `enabled` / `intervalMinutes` section that defaults off: an absent key, an explicit false, and a malformed value all leave the scheduler dormant. `src/config/feature-flags.ts` resolves the cadence; an explicit `intervalMinutes: 0` keeps the unref'd timer idle, and any other value is clamped up to 15 minutes because upstream `/models` caches have not moved below that and a shorter tick only multiplies rate-limit exposure. `src/codex/catalog-auto-refresh.ts` is the module-singleton interval `src/server/background-lifecycle.ts` starts beside the quota reset poller; a tick that is enabled and non-dormant drives the same catalog-only converge funnel management mutations drive. The last-outcome record lives in `src/codex/catalog-refresh-status.ts` (when the tick finished, the normalized `CatalogDisposition`, whether the served model set changed, consecutive failures) and carries no provider or account detail. + +The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md index a7420072bf..ce42ea81b4 100644 --- a/structure/data-planes/images.md +++ b/structure/data-planes/images.md @@ -121,3 +121,7 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior. Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index 33ca8c76b0..159ec056c3 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -331,3 +331,7 @@ Modern `tool` images continue through the existing following-user carrier. These an OpenCodex conversion limit, not a provider capability claim. Final Responses-to-adapter admission follows the [registry contract](../adapters/registry.md#untranslated-input-media). Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index 4628ca5e50..64a2221dbb 100644 --- a/structure/gui-and-management-api.md +++ b/structure/gui-and-management-api.md @@ -632,3 +632,7 @@ The [explicit model-capability contract](config.md#explicit-per-model-capability Exact [model input declarations](config.md#explicit-per-model-capability-declarations) now feed text-only eligibility and catalog hints; existing image-description/omission handling consumes them before the main upstream send. The raw provider editor round-trips `autoReviewModel` and `autoReviewModelOverrides` through editor-owned DTO fields. POST/PATCH/PUT share validation; PUT copies schema-normalized values into the persisted and live candidate before adoption. Canonical `openai` rejects these fields, including clear forms. Field-masked writes (PATCH, editor PUT, reload) pin every registry-seed key and ignore operator overlays the seed never defines, most commonly `selectedModels`; POST keeps the exact-key comparison. Canonical `openai` still rejects `allowPrivateNetwork`, which must not short-circuit destination DNS checks on the ChatGPT forward row. Existing authentication, origin checks and stale-baseline protection still govern the writes. See [reviewer projection](catalog.md#provider-scoped-approval-reviewer). + +The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/ops/docs-and-release.md b/structure/ops/docs-and-release.md index 0c0ffe38d0..ec6227c4ac 100644 --- a/structure/ops/docs-and-release.md +++ b/structure/ops/docs-and-release.md @@ -384,3 +384,7 @@ Exact [model input declarations](../config.md#explicit-per-model-capability-decl Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. Renamed fixed-key providers receive [missing reasoning metadata](../catalog.md#renamed-destination-reasoning-metadata) during derivation; explicit per-model entries and provider defaults retain precedence. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md index e28575b432..761024b18c 100644 --- a/structure/ops/service-and-sidecars.md +++ b/structure/ops/service-and-sidecars.md @@ -182,3 +182,7 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior. Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/overview.md b/structure/overview.md index 0151115adc..d412e5c6da 100644 --- a/structure/overview.md +++ b/structure/overview.md @@ -147,3 +147,7 @@ Translated Chat request construction uses the [inline-image budget](transports/s The [explicit model-capability contract](config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior. Provider-scoped approval reviewer settings are projected by the [catalog owner](catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. + +The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/providers/chat-compat.md b/structure/providers/chat-compat.md index 5ee17ed807..acad53fd97 100644 --- a/structure/providers/chat-compat.md +++ b/structure/providers/chat-compat.md @@ -339,3 +339,7 @@ prose the model reads beside them. Vendor tool execution stays disabled on both adapters, and Qoder's explicit refusal of original images is unchanged. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/providers/kiro.md b/structure/providers/kiro.md index 80f765adae..266162a2ce 100644 --- a/structure/providers/kiro.md +++ b/structure/providers/kiro.md @@ -85,3 +85,7 @@ request on a request path. The marker carries a count and no URL, because a remo image URL can carry a signed token. Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md index dc982c5639..ac4041d3f3 100644 --- a/structure/providers/xai-grok.md +++ b/structure/providers/xai-grok.md @@ -141,3 +141,7 @@ Account quota surfaces use [safe probe diagnostics](../transports/inventory.md#a Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. + +The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/runtime.md b/structure/runtime.md index bd9ebbd561..4e3d849492 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -430,3 +430,7 @@ Translated audio/file admission follows the [final-adapter input contract](adapt The combo may advance to its next eligible unattempted target before output commitment. It records no target/provider cooldown for these request-local mismatches and does not silently drop reasoning controls or raise `none` to a supported rung. Cancellation, origin/cyber-policy rejection, non-replayable post-send errors and the existing streaming commit boundary stay authoritative. Other invalid requests remain terminal. Regression coverage: `tests/responses/responses-forward-prompt-envelope.test.ts`, `tests/routing/router-combo-failover-classification.test.ts`, and `tests/server/server-combo-failover-e2e.test.ts`. + +The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/subagents.md b/structure/subagents.md index c0c92891f7..d3a17f7a39 100644 --- a/structure/subagents.md +++ b/structure/subagents.md @@ -374,3 +374,7 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara Provider-scoped approval reviewer settings are projected by the [catalog owner](catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. Renamed fixed-key providers receive [missing reasoning metadata](catalog.md#renamed-destination-reasoning-metadata) during derivation; explicit per-model entries and provider defaults retain precedence. + +The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 7f01dee197..70dad64e07 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -39,3 +39,7 @@ These optimizations do not add request queues, retry policies, or RSS-based admi Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. + +The opt-in [native mid-turn steering contract](streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index ca80373a8e..08c32dc3ef 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -148,3 +148,7 @@ Renamed fixed-key providers receive [missing reasoning metadata](../catalog.md#r Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. + +The opt-in [native mid-turn steering contract](streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 9af5e63100..305745bcd3 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -702,3 +702,7 @@ What must not happen is a ladder that charges and then returns through a path th nor releases. That is not a lost send; it is a send the request never made, spending an allowance a later recovery in the same request then cannot have. `tests/lib/execution-budget-permits.test.ts` pins both ladder shapes against exactly that. + +The opt-in [native mid-turn steering contract](streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary +authentication, routing and non-native behavior; its connection-local controls, replay journal +and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index 682fbb2ca2..4b0f024b50 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -252,3 +252,49 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior. Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. + +## Experimental native mid-turn steering + +`codexNativeSteering: true` is an independent, default-off opt-in for the client-facing +Responses WebSocket endpoint. It requires `websockets: true`, the canonical ChatGPT forward +route, an eligible Bun runtime, and an upstream model/execution mode that supports steering. +HTTP fallback and translated/provider/sidecar/Combo paths do not gain steering. Plaintext V2 +restoration is excluded because it is not a transparent native event stream. + +`src/server/responses/native-steering.ts` owns one downstream turn and one private physical +upstream connection. The connection remains bound to the credential selected by the ordinary +dispatch path and never enters the idle reuse pool. The normal authentication, admission, +quota observation and pre-dispatch guard remain in force. `response.steer` accepts user-only +input, preserves its target response ID, and cannot select another account or lane. + +An acceptance acknowledges queued input, not application. The parent terminal is relayed, +but the native chain ends only after outstanding submissions settle and the last response +ends. Automatic successors are relayed without an extra create. A pending event preserves +its `required_input` stubs; exactly one explicit same-parent/lane continuation may provide +the saved results. This initial implementation pins model/settings to the initial request +and refuses unrelated input or changed settings on that continuation. Explicit continuations +are paced and recheck the captured dispatch guard after waiting. No tools, accepted input +or ambiguously delivered sends are automatically replayed. + +`src/server/responses/native-steering-replay.ts` journals only committed native input into +the existing thread-scoped replay cache. Rejected/uncommitted steer text is excluded. Sparse +terminal outputs are reconstructed from completed output-item events. Derived state inherits +the original non-persistable-body restriction; the original request is never mutated. The +bounded journal is discarded at teardown. This keeps subsequent ordinary delta-input turns +working without inventing IDs or silently dropping the steering instruction. + +Native chains bypass single-response SSE repair/terminal truncation. Wire IDs, lane IDs and +control events are preserved. A single bounded reader owns delivery; client cancellation, +account invalidation and shutdown abort its upstream. Numeric usage is summed once per +response; steering control frames (which can contain returned user input) are not log samples. + +Bounds: 32 outstanding submissions, 128 response IDs per chain, 32 MiB replay journal, +256 KiB / 1,024 required-input stubs, existing WS frame/queue byte limits, a 90-second control +wait, and a 30-minute saved-tool-result wait. Ordinary active-response silence uses the +configured stall deadline. Unsupported routes return explicit errors rather than discarding +steers. Unknown or mismatched protocol identities fail closed without replay. + +The regression fixture is derived from the pinned OpenAI Python SDK response-steering +schemas at commit `98e1d24f4902ab58830adf0e2b6a729a5d5429b1`; it is not a live Astra +compatibility certification. End-to-end live client/backend verification remains required +before promoting this experimental option to a default. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d2eb5d244b..b6fc1a75a6 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1288,5 +1288,6 @@ "execution-budget-permits.test.ts": "lib", "spend-instrumentation-log.test.ts": "server", "codex-pool-refresh-backoff.test.ts": "codex-integration", - "responses-account-change-scrub.test.ts": "responses" + "responses-account-change-scrub.test.ts": "responses", + "ws-native-steering.test.ts": "responses" } diff --git a/tests/responses/ws-native-steering.test.ts b/tests/responses/ws-native-steering.test.ts new file mode 100644 index 0000000000..4b4da29da5 --- /dev/null +++ b/tests/responses/ws-native-steering.test.ts @@ -0,0 +1,304 @@ +import { afterEach, beforeEach, expect, test } from "bun:test"; +import type { ServerWebSocket } from "bun"; +import type { OcxConfig } from "../../src/types"; +import { createWebsocketHandler } from "../../src/server/index/websocket-handler"; +import type { ServeOptionsContext } from "../../src/server/index/serve-options"; +import { NativeSteeringChannel, MAX_NATIVE_STEERS, validateSteeringFrame } from "../../src/server/responses/native-steering"; +import { NativeSteeringReplay, MAX_NATIVE_STEERING_REPLAY_BYTES } from "../../src/server/responses/native-steering-replay"; +import { type WsData } from "../../src/server/ws-bridge"; +import { getRequestLogEntries, clearRequestLogsForTests } from "../../src/server/request-log"; +import { runOptionalShutdownHooks } from "../../src/lib/optional-shutdown-hooks"; +import { configSchema } from "../../src/config/schema/config-schema"; + +type Frame = Record; +const realSocket = globalThis.WebSocket; +const realFetch = globalThis.fetch; +const proxyKeys = ["HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY", "http_proxy", "https_proxy", "all_proxy", "no_proxy"]; +let savedProxy: Record; +let fallbackCalls = 0; +let nextId = 0; + +class Socket extends EventTarget { + static OPEN = 1; + static all: Socket[] = []; + readyState = 0; + frames: Frame[] = []; + readonly root = `native-${++nextId}`; + constructor(readonly url: string, readonly options: { headers: Record }) { + super(); Socket.all.push(this); + queueMicrotask(() => { this.readyState = 1; this.dispatchEvent(new Event("open")); }); + } + send(text: string) { + const frame = JSON.parse(text); + this.frames.push(frame); + if (this.frames.length === 1) queueMicrotask(() => this.emit({ type: "response.created", response: { id: this.root, status: "in_progress", output: [] } })); + } + emit(frame: Frame) { + const lane = this.frames[0]?.stream_id; + this.dispatchEvent(new MessageEvent("message", { data: JSON.stringify({ ...(lane !== undefined ? { stream_id: lane } : {}), ...frame }) })); + } + close() { if (this.readyState === 3) return; this.readyState = 3; this.dispatchEvent(new Event("close")); } +} +const config = (): OcxConfig => ({ port: 0, defaultProvider: "openai", websockets: true, codexNativeSteering: true, + providers: { openai: { adapter: "openai-responses", baseUrl: "https://chatgpt.com/backend-api/codex", authMode: "forward", codexAccountMode: "direct" } }, +} as OcxConfig); +const waitFor = async (condition: () => boolean) => { + for (let i = 0; i < 1000; i++) { if (condition()) return; await Bun.sleep(1); } + throw new Error("fixture condition timed out"); +}; +function downstream(fields: Frame = {}, settings = config(), credential = "test") { + const handler = createWebsocketHandler({ config: settings, deps: {} } as ServeOptionsContext); + const sent: Frame[] = []; + const ws = { readyState: 1, data: { headers: new Headers({ authorization: `Bearer ${credential}`, "thread-id": `fixture-${credential}`, session_id: `fixture-${credential}` }) } as WsData, + send: (text: string) => { sent.push(JSON.parse(text)); return 1; }, close() { handler.close(ws); }, + } as unknown as ServerWebSocket; + const send = (frame: Frame) => handler.message(ws, JSON.stringify(frame)); + send({ type: "response.create", model: "gpt-5.5", input: "initial", ...fields }); + return { ws, sent, send, handler }; +} +async function begin(fields: Frame = {}, credential = "test") { + const client = downstream(fields, config(), credential); + await waitFor(() => client.sent.some(frame => frame.type === "response.created")); + const socket = Socket.all.find(s => s.options.headers.authorization === `Bearer ${credential}`)!; + expect(socket).toBeDefined(); + return { ...client, socket, id: socket.root }; +} +function accept(socket: Socket, id: string, steerId = "s1") { + socket.emit({ type: "response.steer.accepted", steer: { id: steerId, previous_response_id: id } }); +} +function complete(socket: Socket, id: string, extra: Frame = {}) { + socket.emit({ type: "response.completed", response: { id, status: "completed", output: [], ...extra } }); +} +beforeEach(() => { + nextId = 0; fallbackCalls = 0; + savedProxy = Object.fromEntries(proxyKeys.map(key => [key, process.env[key]])); + for (const key of proxyKeys) delete process.env[key]; + globalThis.WebSocket = Socket as unknown as typeof WebSocket; + globalThis.fetch = (async () => { fallbackCalls++; throw new Error("unexpected network/fallback in native steering fixture"); }) as typeof fetch; + clearRequestLogsForTests(); +}); +afterEach(() => { + for (const socket of Socket.all) socket.close(); + Socket.all = []; + runOptionalShutdownHooks(); + globalThis.WebSocket = realSocket; + globalThis.fetch = realFetch; + for (const key of proxyKeys) { delete process.env[key]; if (savedProxy[key] !== undefined) process.env[key] = savedProxy[key]; } +}); + +test("configuration is explicit opt-in and malformed values fail closed", () => { + const value = config(); + expect(configSchema.parse(value).codexNativeSteering).toBe(true); + expect(configSchema.parse({ ...value, codexNativeSteering: "true" }).codexNativeSteering).toBe(false); + delete value.codexNativeSteering; + expect(configSchema.parse(value).codexNativeSteering).not.toBe(true); +}); + +test("real handler -> auth/dispatch -> native exchange -> downstream preserves automatic successor and aggregate usage", async () => { + const { ws, socket, send, sent, id } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "do not edit" }); + expect(socket.frames[1]).toEqual({ type: "response.steer", previous_response_id: id, input: "do not edit" }); + accept(socket, id); + socket.emit({ type: "response.incomplete", response: { id, status: "incomplete", output: [], incomplete_details: { reason: "steered" }, usage: { input_tokens: 10, output_tokens: 2 } } }); + socket.emit({ type: "response.created", response: { id: "successor", previous_response_id: id, output: [] } }); + complete(socket, "successor", { usage: { input_tokens: 20, output_tokens: 3 } }); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.map(frame => frame.type)).toEqual(["response.created", "response.steer.accepted", "response.incomplete", "response.created", "response.completed"]); + expect(sent.at(-1)?.response.id).toBe("successor"); + expect(Socket.all).toHaveLength(1); + expect(socket.frames).toHaveLength(2); // no synthetic create for an automatic successor + expect(socket.readyState).toBe(3); + expect(fallbackCalls).toBe(0); + const log = getRequestLogEntries().at(-1)!; + expect(log.usage).toMatchObject({ inputTokens: 30, outputTokens: 5 }); + expect(log.terminalStatus).toBe("completed"); + expect(log.upstreamError).toBeUndefined(); +}); + +test("normal completion before acceptance still retains the socket and successor", async () => { + const { ws, socket, send, sent, id } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "new constraint" }); + complete(socket, id); + accept(socket, id); + socket.emit({ type: "response.created", response: { id: "r2", previous_response_id: id } }); + complete(socket, "r2"); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.filter(frame => frame.type === "response.completed").map(frame => frame.response.id)).toEqual([id, "r2"]); +}); + +test("pending results use one same-account/lane create and never replay accepted user text", async () => { + const { ws, socket, send, sent, id } = await begin({ stream_id: "lane" }); + send({ type: "response.steer", previous_response_id: id, input: "keep files" }); + send({ type: "response.steer", previous_response_id: id, input: "only report" }); + accept(socket, id); accept(socket, id, "s2"); + complete(socket, id); + const stub = { type: "function_call_output", call_id: "call-1" }; + for (const steerId of ["s1", "s2"]) socket.emit({ type: "response.steer.pending", steer: { id: steerId, previous_response_id: id }, reason: "waiting_for_required_input", required_input: [stub] }); + await waitFor(() => sent.some(frame => frame.type === "response.steer.pending")); + const continuation = { type: "response.create", previous_response_id: id, stream_id: "lane", model: "gpt-5.5", input: [{ ...stub, output: "saved result" }] }; + send(continuation); send(continuation); + await waitFor(() => socket.frames.length === 4); + expect(socket.frames).toHaveLength(4); // initial, two steers, exactly one continuation + expect(socket.frames[3].previous_response_id).toBe(id); + expect(socket.frames[3].stream_id).toBe("lane"); + expect(socket.frames[3].input).toEqual(continuation.input); + expect(sent.at(-1)?.error.code).toBe("duplicate_continuation"); + socket.emit({ type: "response.created", response: { id: "r2", previous_response_id: id } }); + complete(socket, "r2"); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.at(-1)?.response.id).toBe("r2"); + expect(Socket.all).toHaveLength(1); +}); + +test("subsequent ordinary turns retain committed steering through the scoped replay cache", async () => { + const { ws, socket, send, id, sent } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "committed instruction" }); + accept(socket, id); complete(socket, id); + socket.emit({ type: "response.created", response: { id: "cached-successor", previous_response_id: id } }); + complete(socket, "cached-successor"); + await waitFor(() => !ws.data.nativeSteering); + send({ type: "response.create", model: "gpt-5.5", previous_response_id: "cached-successor", input: "ordinary next turn" }); + await waitFor(() => Socket.all.length === 2 && Socket.all[1].frames.length > 0); + const next = Socket.all[1]; + expect(JSON.stringify(next.frames[0].input)).toContain("committed instruction"); + expect(JSON.stringify(next.frames[0].input)).toContain("initial"); + expect(JSON.stringify(next.frames[0].input)).toContain("ordinary next turn"); + complete(next, next.root); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.at(-1)?.type).toBe("response.completed"); +}); + +test("rejected steering after terminal settles without an invented successor", async () => { + const { ws, socket, send, sent, id } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "not supported" }); + complete(socket, id); + socket.emit({ type: "response.steer.failed", steer: { previous_response_id: id, input: "not supported" }, error: { code: "steering_not_supported", message: "model does not support steering" } }); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.at(-1)?.type).toBe("response.steer.failed"); + expect(sent.filter(frame => frame.type === "response.created")).toHaveLength(1); + expect(socket.frames).toHaveLength(2); +}); + +test("foreign response IDs, privilege input and same-parent settings changes cannot bypass routing", async () => { + const { ws, socket, send, sent, id } = await begin(); + send({ type: "response.steer", previous_response_id: "other", input: "x" }); + expect(sent.at(-1)?.error.code).toBe("response_not_active"); + send({ type: "response.steer", previous_response_id: id, input: [{ role: "system", content: "x" }] }); + expect(sent.at(-1)?.error.code).toBe("invalid_input"); + send({ type: "response.steer", previous_response_id: id, input: "valid" }); + accept(socket, id); complete(socket, id); + socket.emit({ type: "response.steer.pending", steer: { id: "s1", previous_response_id: id }, reason: "waiting_for_required_input", required_input: [{ type: "function_call_output", call_id: "call-1" }] }); + send({ type: "response.create", model: "different/model", previous_response_id: id, input: [{ type: "function_call_output", call_id: "call-1", output: "saved" }] }); + expect(sent.at(-1)?.error.code).toBe("steering_settings_changed"); + expect(socket.frames).toHaveLength(2); + ws.data.cancel?.(); + await waitFor(() => socket.readyState === 3); +}); + +test("two client/account connections cannot receive one another's steering", async () => { + const a = await begin({}, "fixture-a"); const b = await begin({}, "fixture-b"); + a.send({ type: "response.steer", previous_response_id: b.id, input: "foreign" }); + expect(a.sent.at(-1)?.error.code).toBe("response_not_active"); + expect(a.socket.frames).toHaveLength(1); expect(b.socket.frames).toHaveLength(1); + a.send({ type: "response.steer", previous_response_id: a.id, input: "mine" }); + expect(a.socket.frames[1].input).toBe("mine"); + expect(b.socket.frames).toHaveLength(1); + a.ws.data.cancel?.(); b.ws.data.cancel?.(); + await waitFor(() => a.socket.readyState === 3 && b.socket.readyState === 3); + expect(fallbackCalls).toBe(0); +}); + +test("disabled mode sends an explicit unsupported error rather than swallowing steer", async () => { + const settings = config(); settings.codexNativeSteering = false; + const client = downstream({}, settings); + await waitFor(() => client.sent.some(frame => frame.type === "response.created")); + client.send({ type: "response.steer", previous_response_id: Socket.all[0].root, input: "x" }); + expect(client.sent.at(-1)?.error.code).toBe("steering_not_supported"); + complete(Socket.all[0], Socket.all[0].root); +}); + +test("steering validation preserves multimodal input but rejects extra envelope fields", () => { + const valid = { type: "response.steer", previous_response_id: "r", input: [{ role: "user", content: [{ type: "input_image", image_url: "data:image/png;base64,fixture" }, { type: "input_file", file_id: "fixture-file" }] }] }; + expect(() => validateSteeringFrame(valid)).not.toThrow(); + for (const extra of [{ stream_id: "lane" }, { model: "other" }, { authorization: "not-a-credential" }]) expect(() => validateSteeringFrame({ ...valid, ...extra })).toThrow(); + expect(() => validateSteeringFrame({ ...valid, input: [] })).toThrow(); +}); + +test("pending submissions have a hard count bound and disconnect releases them", () => { + const channel = new NativeSteeringChannel({ model: "fixture" }); + const detach = channel.attach(() => {}, () => {}); + channel.observe({ type: "response.created", response: { id: "r" } }); + for (let i = 0; i < MAX_NATIVE_STEERS; i++) channel.steer({ type: "response.steer", previous_response_id: "r", input: "x" }); + expect(() => channel.steer({ type: "response.steer", previous_response_id: "r", input: "x" })).toThrow("limit"); + detach(); expect(channel.hasOutstanding).toBe(false); +}); + +test("foreign lane or successor parent is a non-replayable protocol failure", () => { + const channel = new NativeSteeringChannel({ stream_id: "one" }); + const detach = channel.attach(() => {}, () => {}); + expect(() => channel.observe({ type: "response.created", stream_id: "two", response: { id: "r" } })).toThrow("lane mismatch"); + detach(); +}); + +test("replay budget refuses overflow instead of silently losing context", () => { + expect(() => new NativeSteeringReplay("x".repeat(MAX_NATIVE_STEERING_REPLAY_BYTES), () => {})).toThrow("budget"); +}); + +test("HTTP upgrade fallback keeps ordinary streaming and rejects steering explicitly", async () => { + globalThis.WebSocket = class { constructor() { throw new Error("fixture unavailable upgrade"); } } as unknown as typeof WebSocket; + let finish!: () => void; + globalThis.fetch = (async () => { + fallbackCalls++; + const encoder = new TextEncoder(); + const stream = new ReadableStream({ start(controller) { + const event = (value: Frame) => controller.enqueue(encoder.encode(`data: ${JSON.stringify(value)}\n\n`)); + event({ type: "response.created", response: { id: "http-response", status: "in_progress", output: [] } }); + finish = () => { event({ type: "response.completed", response: { id: "http-response", status: "completed", output: [] } }); controller.close(); }; + } }); + return new Response(stream, { headers: { "content-type": "text/event-stream" } }); + }) as typeof fetch; + const { ws, send, sent } = downstream(); + await waitFor(() => sent.some(frame => frame.type === "response.created")); + send({ type: "response.steer", previous_response_id: "http-response", input: "not delivered" }); + expect(sent.at(-1)?.error.code).toBe("steering_not_supported"); + finish(); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.at(-1)?.type).toBe("response.completed"); + expect(fallbackCalls).toBe(1); + expect(Socket.all).toHaveLength(0); +}); + +test("post-send disconnect never replays accepted steering through HTTP or another socket", async () => { + const { ws, socket, send, sent, id } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "delivery unknown" }); + accept(socket, id); + socket.close(); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.at(-1)?.type).toBe("error"); + expect(fallbackCalls).toBe(0); + expect(Socket.all).toHaveLength(1); + expect(socket.frames).toHaveLength(2); +}); + +test("downstream disconnect closes the dedicated upstream while steering is pending", async () => { + const { ws, socket, handler, send, id } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "only report" }); + accept(socket, id); complete(socket, id); + socket.emit({ type: "response.steer.pending", steer: { id: "s1", previous_response_id: id }, reason: "waiting_for_required_input", required_input: [{ type: "function_call_output", call_id: "saved-call" }] }); + handler.close(ws); + await waitFor(() => socket.readyState === 3 && !ws.data.nativeSteering); + expect(fallbackCalls).toBe(0); + expect(socket.frames).toHaveLength(2); +}); + +test("idle deadline is bounded and reports uncertainty without inventing a continuation", async () => { + const channel = new NativeSteeringChannel({ type: "response.create", model: "fixture" }, 1); + const sent: Frame[] = []; + let failure: Error | undefined; + const detach = channel.attach(frame => sent.push(frame), error => { failure = error; }); + channel.observe({ type: "response.created", response: { id: "idle" } }); + await waitFor(() => failure !== undefined); + expect(failure?.message).toContain("timed out"); + expect(sent).toHaveLength(0); + detach(); +}); From 710cbffac9f6a48a8605cc6ed76d490651c4bdb4 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 16 Sep 2026 02:45:00 +0000 Subject: [PATCH 2/3] fix(responses): preserve steering continuations and register bounded owners --- .../content/docs/guides/codex-integration.md | 11 ++- src/responses/state.ts | 29 +------ src/responses/state/body-policy.ts | 24 ++++++ src/server/responses/codex-ws-exchange.ts | 2 +- src/server/responses/native-steering.ts | 81 ++++++++++++++++--- structure/transports/streaming-health.md | 11 ++- tests/helpers/responses-core-source.ts | 3 + tests/responses/ws-native-steering.test.ts | 78 ++++++++++++++++++ 8 files changed, 192 insertions(+), 47 deletions(-) create mode 100644 src/responses/state/body-policy.ts diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index 1ed04eb2ba..534c35bca7 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -896,10 +896,13 @@ This option is off by default. It forwards steering to the same native ChatGPT W connection and selected account, preserving automatic successor responses and pending saved-tool-result continuations. Acceptance means queued, not yet applied. -For `response.steer.pending`, supply the reported required tool results or approval decisions -**once per parent**, on the same lane. Do not rerun tools or resend accepted steering text. -This first implementation accepts saved-result-only continuations with unchanged model and -request settings. A changed model/settings requires an explicitly stopped or finished turn +Supply the required tool results or approval decisions **once per parent**, on the same lane. +Results can arrive before `response.steer.pending`: the relay also matches the completed +parent's advertised calls and approvals. A `name` on a pending function-output stub is +optional on the result, as in the native schema. Additional user messages may accompany +these results; system/developer messages, duplicate results and unrelated call IDs are refused. +Do not rerun tools or resend accepted steering text. This first implementation requires +unchanged model and request settings. A changed model/settings requires an explicitly stopped or finished turn and normal new dispatch. Multiple independent conversations use independent connections. HTTP fallback, other providers, translated models, sidecars, Combo attempts and plaintext V2 diff --git a/src/responses/state.ts b/src/responses/state.ts index d48e7dcad0..35237de6db 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -23,6 +23,8 @@ import type { ResponseSpillWriteFailureCode, ResponseSpillWriteStatus, ResponseS export { responseAdmissionCountersForTests } from "./state/spill-failure"; import { admissionCounters, noteSpillWriteFailure, noteSpillWriteSuccess, spillCounters, spillWriteHealth } from "./state/spill-failure"; import { loadSnapshotEntry } from "./state/snapshot-codec"; +import { isBodyNonPersistable } from "./state/body-policy"; +export { isBodyNonPersistable, markBodyNonPersistable } from "./state/body-policy"; export { flushPendingResponseSpillsForTests, awaitResponseSpillPublicationTailForTests, pendingResponseSpillMetricsForTests, setResponseSpillShutdownBudgetForTests, setResponseSpillAsyncAclAttemptBudgetForTests, setResponseSpillShutdownTerminalizationPassLimitForTests } from "./state/spill-queue"; import { bindSpillQueueStore, @@ -1235,31 +1237,6 @@ export function responseStateMetrics(): ResponseStateMetrics { * Cache completed output and max_output_tokens partial output for previous_response_id replay. * Content-filtered incomplete and failed output are not authoritative replay history. */ -/** - * Request bodies that must never enter the continuation cache. - * - * The cache is persisted to `responses-state.json`, so anything recorded here reaches disk. - * Encrypted-agent-task recovery decrypts task text into the request body and promises - * in-memory, TTL-bounded retention; recording that body would put the plaintext on disk with - * no TTL and break the promise. - * - * A WeakSet rather than a body field on purpose: `_rawBody` is serialized verbatim by the - * native passthrough, so any marker written into the body itself would be sent upstream. - * Marking is enforced once here rather than at each call site, because every recording path - * (streaming, non-streaming, passthrough, forced) funnels through `rememberResponseState` — - * a new call site cannot reintroduce the leak by forgetting a guard. - */ -const nonPersistableBodies = new WeakSet(); - -/** Bar this exact request body from the continuation cache, and therefore from disk. */ -export function markBodyNonPersistable(body: unknown): void { - if (body && typeof body === "object") nonPersistableBodies.add(body as object); -} - -export function isBodyNonPersistable(body: unknown): boolean { - return !!body && typeof body === "object" && nonPersistableBodies.has(body); -} - export function rememberResponseState( requestBody: unknown, response: { id?: unknown; output?: unknown; status?: unknown; incomplete_details?: unknown }, @@ -1268,7 +1245,7 @@ export function rememberResponseState( ): void { if (!requestBody || typeof requestBody !== "object" || Array.isArray(requestBody)) return; const request = requestBody as Record; - if (nonPersistableBodies.has(request)) return; + if (isBodyNonPersistable(request)) return; // `force` bypasses only the store:false skip: Codex sends `store:false` on every non-Azure // HTTP request (and WS inherits it), yet its WS turns still chain with previous_response_id. // The passthrough branch records with force so those chains can be expanded locally; the diff --git a/src/responses/state/body-policy.ts b/src/responses/state/body-policy.ts new file mode 100644 index 0000000000..062f49c463 --- /dev/null +++ b/src/responses/state/body-policy.ts @@ -0,0 +1,24 @@ +/** + * Request bodies that must never enter the continuation cache. + * + * The cache is persisted to `responses-state.json`, so anything recorded here reaches disk. + * Encrypted-agent-task recovery decrypts task text into the request body and promises + * in-memory, TTL-bounded retention; recording that body would put the plaintext on disk with + * no TTL and break the promise. + * + * A WeakSet rather than a body field on purpose: `_rawBody` is serialized verbatim by the + * native passthrough, so any marker written into the body itself would be sent upstream. + * Marking is enforced once here rather than at each call site, because every recording path + * (streaming, non-streaming, passthrough, forced) funnels through `rememberResponseState` — + * a new call site cannot reintroduce the leak by forgetting a guard. + */ +const nonPersistableBodies = new WeakSet(); + +/** Bar this exact request body from the continuation cache, and therefore from disk. */ +export function markBodyNonPersistable(body: unknown): void { + if (body && typeof body === "object") nonPersistableBodies.add(body as object); +} + +export function isBodyNonPersistable(body: unknown): boolean { + return !!body && typeof body === "object" && nonPersistableBodies.has(body); +} diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 4b7b55f95d..213a91fd79 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -345,7 +345,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { beforeDispatch?.(new Headers(headers)); let outgoing = frame; if (frame.type === "response.create") { - // The channel validates same settings/lane and saved-result-only input. + // The channel validates same settings/lane, saved results and user-only additions. // Reuse the already-routed/authorized native settings; never feed a // previous_response_id through the REST sanitizer or account selector. const base = JSON.parse(frameText) as Record; diff --git a/src/server/responses/native-steering.ts b/src/server/responses/native-steering.ts index 3c5e658fee..3190948684 100644 --- a/src/server/responses/native-steering.ts +++ b/src/server/responses/native-steering.ts @@ -54,12 +54,37 @@ export function validateSteeringFrame(frame: Frame): void { type Parent = { unacknowledged: number; accepted: Set; ended: boolean }; +/** Only client-owned results can use the early-continuation path. */ +function outputRequirement(item: unknown): Frame | undefined { + if (!record(item) || typeof item.type !== "string") return; + if (item.type === "mcp_approval_request") { + if (!validId(item.id)) throw new Error("Native steering approval identity is invalid"); + return { type: "mcp_approval_response", approval_request_id: item.id }; + } + const types: Record = { + function_call: "function_call_output", custom_tool_call: "custom_tool_call_output", + local_shell_call: "local_shell_call_output", shell_call: "shell_call_output", + computer_call: "computer_call_output", apply_patch_call: "apply_patch_call_output", + }; + const type = Object.hasOwn(types, item.type) ? types[item.type] : undefined; + if (!type) return; + if (!validId(item.call_id)) throw new Error("Native steering tool-call identity is invalid"); + return { type, call_id: item.call_id }; +} + +function matchesRequirement(item: Frame, stub: Frame): boolean { + // The wire may label a required result with its tool name, but the ordinary + // function/custom output schema identifies the result by call_id, not name. + return Object.entries(stub).every(([key, value]) => + key === "name" && item[key] === undefined || stable(item[key]) === stable(value)); +} + /** * One downstream turn owns one dedicated native upstream socket, including all * automatic successors and required-input continuations. Never registered by a * caller-supplied response ID in global state; never lent to another account. * - * Opt-in single-lane implementation. Continuations may supply saved tool results + * Opt-in single-lane implementation. Continuations may supply saved tool results and new user messages * but cannot change settings/routing. General new turns still use normal dispatch. */ export class NativeSteeringChannel { @@ -79,6 +104,8 @@ export class NativeSteeringChannel { private finished = false; private everAttached = false; private required: Frame[] = []; + private readonly advertised = new Map(); + private advertisedBytes = 0; constructor(initial: Frame, private readonly idleMs = 300_000) { this.lane = initial.stream_id; @@ -111,6 +138,8 @@ export class NativeSteeringChannel { this.correlation = undefined; this.parents.clear(); this.required = []; + this.advertised.clear(); + this.advertisedBytes = 0; this.replay?.dispose(); this.replay = undefined; }; @@ -150,30 +179,52 @@ export class NativeSteeringChannel { } } - /** Returns false only for an unrelated ordinary create after this chain ends. */ + private advertise(item: unknown): void { + const stub = outputRequirement(item); + if (!stub) return; + const key = JSON.stringify(stub); + if (this.advertised.has(key)) return; + const bytes = Buffer.byteLength(key); + if (this.advertised.size >= 1024 || this.advertisedBytes + bytes > 256 * 1024) { + throw new Error("Native steering advertised-input budget exceeded"); + } + this.advertised.set(key, stub); + this.advertisedBytes += bytes; + } + + /** Returns false only when an ordinary create may use normal dispatch. */ continue(frame: Frame): boolean { if (this.finished || !this.send) return false; - if (!this.pendingParent || !this.currentId || frame.previous_response_id !== this.pendingParent) { - if (this.hasOutstanding) throw new NativeSteeringError("steering_continuation_required", "Queued steering owns this connection; wait for the successor or send the required-input continuation, or explicitly stop the turn."); + const parent = this.currentId ? this.parents.get(this.currentId) : undefined; + if (!parent?.ended || !this.currentId || frame.previous_response_id !== this.currentId) { + if (this.hasOutstanding || this.continuationSent) throw new NativeSteeringError("steering_continuation_required", "Queued steering owns this connection; wait for the successor or send the required-input continuation, or explicitly stop the turn."); return false; } if (this.continuationSent) throw new NativeSteeringError("duplicate_continuation", "A required-input continuation was already sent for this parent."); + // The client is allowed to return saved results before response.steer.pending. + // In that case only calls actually advertised by this response authorize it. + const required = this.pendingParent ? this.required : [...this.advertised.values()]; + if (!required.length) throw new NativeSteeringError("steering_continuation_required", "Wait for the automatic successor or server-identified required input."); if (frame.stream_id !== this.lane || frame.generate === false) throw new NativeSteeringError("invalid_input", "The continuation must use the same WebSocket lane and generate a response."); for (const [key, value] of Object.entries(frame)) { if (["type", "input", "previous_response_id", "stream", "stream_id"].includes(key)) continue; if (this.settings.get(key) !== fingerprint(value)) throw new NativeSteeringError("steering_settings_changed", "The experimental native steering continuation cannot change model or request settings; start a separate turn instead."); } - // Restrict this bypass to exactly the server-identified outputs/approvals. - // General messages/settings continue through ordinary validation and routing. const input = frame.input; - if (!Array.isArray(input) || input.length !== this.required.length) throw new NativeSteeringError("invalid_input", "Supply the saved results for the required_input stubs exactly once; do not resend steering text."); + if (!Array.isArray(input) || !input.length) throw new NativeSteeringError("invalid_input", "Supply the saved results for the required_input stubs exactly once; do not resend steering text."); const used = new Set(); for (const item of input) { - const match = record(item) ? this.required.findIndex((stub, i) => !used.has(i) - && Object.entries(stub).every(([key, value]) => stable(item[key]) === stable(value))) : -1; + // An explicit continuation may carry new user input after its saved results. + // Reuse the narrow user-only validator so privileged roles cannot bypass routing. + if (record(item) && item.role === "user") { + validateSteeringFrame({ type: "response.steer", previous_response_id: this.currentId, input: [item] }); + continue; + } + const match = record(item) ? required.findIndex((stub, i) => !used.has(i) && matchesRequirement(item, stub)) : -1; if (match < 0) throw new NativeSteeringError("invalid_input", "Continuation input must match the pending tool-output or approval stubs."); used.add(match); } + if (used.size !== required.length) throw new NativeSteeringError("invalid_input", "Every required tool output or approval must be supplied exactly once."); this.continuationSent = true; this.wait(); try { this.liveSend(frame); } catch (error) { this.continuationSent = false; throw error; } @@ -198,6 +249,8 @@ export class NativeSteeringChannel { this.parents.set(id, { unacknowledged: 0, accepted: new Set(), ended: false }); this.pendingParent = undefined; this.required = []; + this.advertised.clear(); + this.advertisedBytes = 0; this.continuationSent = false; this.correlation?.finish(); this.correlation = new CodexWsCorrelation(true, () => false); @@ -232,16 +285,18 @@ export class NativeSteeringChannel { } else throw new Error("unsupported native steering control event"); } else { this.correlation?.accept({ ...event, stream_id: undefined }); + if (type === "response.output_item.done") this.advertise(event.item); if (type === "response.completed" || type === "response.failed" || type === "response.incomplete") { if (!this.currentId || response?.id !== this.currentId) throw new Error("native steering terminal identity mismatch"); this.parents.get(this.currentId)!.ended = true; + if (Array.isArray(response.output)) for (const item of response.output) this.advertise(item); } } if (type === "error") this.finished = true; - else this.finished = this.currentId !== undefined && this.parents.get(this.currentId)!.ended && !this.hasOutstanding; - if (this.finished || !this.hasOutstanding) { clearTimeout(this.timer); this.timer = undefined; } - else this.wait(this.pendingParent && !this.continuationSent ? NATIVE_STEERING_TOOL_WAIT_MS : NATIVE_STEERING_WAIT_MS); - if (!this.finished && !this.hasOutstanding && this.currentId) this.wait(this.idleMs); + else this.finished = this.currentId !== undefined && this.parents.get(this.currentId)!.ended && !this.hasOutstanding && !this.continuationSent; + if (this.finished) { clearTimeout(this.timer); this.timer = undefined; } + else if (this.hasOutstanding || this.continuationSent) this.wait(this.pendingParent && !this.continuationSent ? NATIVE_STEERING_TOOL_WAIT_MS : NATIVE_STEERING_WAIT_MS); + else if (this.currentId) this.wait(this.idleMs); this.replay?.observe(event); return this.finished; } diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index 4b0f024b50..53ad76c0ee 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -271,15 +271,20 @@ An acceptance acknowledges queued input, not application. The parent terminal is but the native chain ends only after outstanding submissions settle and the last response ends. Automatic successors are relayed without an extra create. A pending event preserves its `required_input` stubs; exactly one explicit same-parent/lane continuation may provide -the saved results. This initial implementation pins model/settings to the initial request -and refuses unrelated input or changed settings on that continuation. Explicit continuations +the saved results. Results may arrive before the pending event: completed output items and +terminal output advertise the permitted call/approval IDs. Stub `name` is optional on a +returned function output; a different supplied name is still refused. New user messages may +accompany results, but privileged messages, unrelated IDs and duplicate results cannot. This +initial implementation pins model/settings to the initial request. A failed steer does not +cancel an explicit continuation already dispatched. Explicit continuations are paced and recheck the captured dispatch guard after waiting. No tools, accepted input or ambiguously delivered sends are automatically replayed. `src/server/responses/native-steering-replay.ts` journals only committed native input into the existing thread-scoped replay cache. Rejected/uncommitted steer text is excluded. Sparse terminal outputs are reconstructed from completed output-item events. Derived state inherits -the original non-persistable-body restriction; the original request is never mutated. The +the original non-persistable-body restriction from `src/responses/state/body-policy.ts`; +`state.ts` keeps its existing public exports. The original request is never mutated. The bounded journal is discarded at teardown. This keeps subsequent ordinary delta-input turns working without inventing IDs or silently dropping the steering instruction. diff --git a/tests/helpers/responses-core-source.ts b/tests/helpers/responses-core-source.ts index 35fea3be44..984faf3213 100644 --- a/tests/helpers/responses-core-source.ts +++ b/tests/helpers/responses-core-source.ts @@ -8,6 +8,9 @@ import { repoPath } from "./repo-root"; export const RESPONSES_CORE_MODULES = [ "core.ts", "core-options.ts", + "native-steering.ts", + "native-steering-replay.ts", + "codex-ws-correlation.ts", "core-lifetime.ts", "core-replay.ts", "core-errors.ts", diff --git a/tests/responses/ws-native-steering.test.ts b/tests/responses/ws-native-steering.test.ts index 4b4da29da5..558a348174 100644 --- a/tests/responses/ws-native-steering.test.ts +++ b/tests/responses/ws-native-steering.test.ts @@ -302,3 +302,81 @@ test("idle deadline is bounded and reports uncertainty without inventing a conti expect(sent).toHaveLength(0); detach(); }); + +test("saved tool results may arrive before pending and retain extra user input without replaying accepted steering", async () => { + const { ws, socket, send, sent, id } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "accepted constraint" }); + accept(socket, id); + complete(socket, id, { output: [{ type: "function_call", call_id: "early-call", name: "lookup", arguments: "{}" }] }); + const input = [ + { type: "function_call_output", call_id: "early-call", output: "saved result" }, + { role: "user", content: "Show the revised plan first." }, + ]; + send({ type: "response.create", previous_response_id: id, model: "gpt-5.5", input }); + await waitFor(() => socket.frames.length === 3); + expect(socket.frames[2].input).toEqual(input); + expect(sent.some(frame => frame.type === "error")).toBe(false); + socket.emit({ type: "response.created", response: { id: "early-successor", previous_response_id: id } }); + complete(socket, "early-successor"); + await waitFor(() => !ws.data.nativeSteering); + expect(Socket.all).toHaveLength(1); + expect(fallbackCalls).toBe(0); +}); + +test("pending stub name is optional on a function output but a different supplied name is rejected", () => { + const channel = new NativeSteeringChannel({ model: "fixture" }); + const sent: Frame[] = []; + const detach = channel.attach(frame => sent.push(frame), () => {}); + channel.observe({ type: "response.created", response: { id: "r" } }); + channel.steer({ type: "response.steer", previous_response_id: "r", input: "constraint" }); + channel.observe({ type: "response.steer.accepted", steer: { id: "s", previous_response_id: "r" } }); + channel.observe({ type: "response.completed", response: { id: "r", output: [] } }); + channel.observe({ type: "response.steer.pending", steer: { id: "s", previous_response_id: "r" }, reason: "waiting_for_required_input", + required_input: [{ type: "function_call_output", call_id: "c", name: "lookup" }] }); + const continuation = { type: "response.create", previous_response_id: "r", input: [{ type: "function_call_output", call_id: "c", output: "saved" }] }; + expect(() => channel.continue({ ...continuation, input: [{ ...continuation.input[0], name: "other" }] })).toThrow(); + expect(channel.continue(continuation)).toBe(true); + expect(sent).toHaveLength(2); + detach(); +}); + +test("a steering failure cannot close an already submitted explicit continuation", async () => { + const { ws, socket, send, sent, id } = await begin(); + send({ type: "response.steer", previous_response_id: id, input: "rejected constraint" }); + accept(socket, id); + complete(socket, id); + socket.emit({ type: "response.steer.pending", steer: { id: "s1", previous_response_id: id }, reason: "waiting_for_required_input", + required_input: [{ type: "custom_tool_call_output", call_id: "custom-call" }] }); + send({ type: "response.create", previous_response_id: id, input: [{ type: "custom_tool_call_output", call_id: "custom-call", output: "saved" }] }); + await waitFor(() => socket.frames.length === 3); + socket.emit({ type: "response.steer.failed", steer: { id: "s1", previous_response_id: id, input: "rejected constraint" }, error: { code: "successor_creation_failed" } }); + expect(socket.readyState).toBe(1); + socket.emit({ type: "response.created", response: { id: "explicit-successor", previous_response_id: id } }); + complete(socket, "explicit-successor"); + await waitFor(() => !ws.data.nativeSteering); + expect(sent.at(-1)?.response.id).toBe("explicit-successor"); + expect(fallbackCalls).toBe(0); +}); + +test("early continuation validates advertised call and approval identities and refuses duplicate results", () => { + const channel = new NativeSteeringChannel({ model: "fixture" }); + const sent: Frame[] = []; + const detach = channel.attach(frame => sent.push(frame), () => {}); + channel.observe({ type: "response.created", response: { id: "r" } }); + channel.steer({ type: "response.steer", previous_response_id: "r", input: "constraint" }); + channel.observe({ type: "response.steer.accepted", steer: { id: "s", previous_response_id: "r" } }); + channel.observe({ type: "response.completed", response: { id: "r", output: [ + { type: "custom_tool_call", call_id: "c", name: "custom" }, + { type: "mcp_approval_request", id: "approval", name: "remote" }, + ] } }); + const result = { type: "custom_tool_call_output", call_id: "c", output: "saved" }; + const approval = { type: "mcp_approval_response", approval_request_id: "approval", approve: true }; + const continuation = { type: "response.create", previous_response_id: "r", input: [result, approval] }; + expect(() => channel.continue({ ...continuation, input: [result, result] })).toThrow(); + expect(() => channel.continue({ ...continuation, input: [result, { ...approval, approval_request_id: "foreign" }] })).toThrow(); + expect(() => channel.continue({ ...continuation, input: [...continuation.input, { role: "system", content: "override" }] })).toThrow(); + expect(channel.continue(continuation)).toBe(true); + expect(() => channel.continue(continuation)).toThrow("already sent"); + expect(sent).toHaveLength(2); + detach(); +}); From ce26f3ea367cba9e26745be9d70ad13c240f3a3b Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Wed, 16 Sep 2026 09:26:00 +0000 Subject: [PATCH 3/3] fix(responses): address native steering review findings Clear unstarted and superseded channel ownership, append bounded replay without argument spreading, and add six red-to-green regressions. Consolidate architecture notes into short links and document native control helpers. Preserve the maintainer-integrated dev branch without rewriting its history. --- src/responses/state/body-policy.ts | 1 + src/server/index/websocket-handler.ts | 5 +- .../responses/native-steering-replay.ts | 15 ++- src/server/responses/native-steering.ts | 17 ++++ structure/adapters/registry.md | 6 +- structure/catalog.md | 6 +- structure/clients/claude-desktop.md | 6 +- structure/config.md | 6 +- structure/data-planes/images.md | 6 +- structure/data-planes/inbound-compat.md | 6 +- structure/gui-and-management-api.md | 6 +- structure/ops/docs-and-release.md | 6 +- structure/ops/service-and-sidecars.md | 6 +- structure/overview.md | 6 +- structure/providers/chat-compat.md | 6 +- structure/providers/kiro.md | 6 +- structure/providers/xai-grok.md | 6 +- structure/runtime.md | 6 +- structure/subagents.md | 6 +- structure/transports/byte-accounting.md | 6 +- structure/transports/inventory.md | 6 +- structure/transports/responses.md | 6 +- structure/transports/streaming-health.md | 5 +- tests/responses/ws-native-steering.test.ts | 91 +++++++++++++++++++ 24 files changed, 167 insertions(+), 75 deletions(-) diff --git a/src/responses/state/body-policy.ts b/src/responses/state/body-policy.ts index 062f49c463..ef4d608a06 100644 --- a/src/responses/state/body-policy.ts +++ b/src/responses/state/body-policy.ts @@ -19,6 +19,7 @@ export function markBodyNonPersistable(body: unknown): void { if (body && typeof body === "object") nonPersistableBodies.add(body as object); } +/** Test the body's in-memory persistence restriction without adding a wire marker. */ export function isBodyNonPersistable(body: unknown): boolean { return !!body && typeof body === "object" && nonPersistableBodies.has(body); } diff --git a/src/server/index/websocket-handler.ts b/src/server/index/websocket-handler.ts index 81f07767a6..f1c564c65a 100644 --- a/src/server/index/websocket-handler.ts +++ b/src/server/index/websocket-handler.ts @@ -212,6 +212,8 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { markActivity("ws response.create"); ws.data.cancel?.(); + // A superseded turn must not keep ownership during warmup or refusal. + ws.data.nativeSteering = undefined; let nativeSteering: NativeSteeringChannel | undefined; try { const idleMs = typeof config.stallTimeoutSec === "number" && Number.isFinite(config.stallTimeoutSec) @@ -221,7 +223,6 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { sendJsonFrame(ws, buildWsErrorFrame(400, { type: "invalid_request_error", message: "Invalid native steering request settings" })); return; } - ws.data.nativeSteering = nativeSteering; const turnId = (ws.data.turnId ?? 0) + 1; ws.data.turnId = turnId; const isCurrent = () => ws.data.turnId === turnId; @@ -256,6 +257,8 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { return; } + // Only a genuinely admitted turn may receive steering or continuations. + ws.data.nativeSteering = nativeSteering; const payload: Record = { ...frame }; delete payload.type; turnAdmissionLease.bindAbortController(turnAbort); diff --git a/src/server/responses/native-steering-replay.ts b/src/server/responses/native-steering-replay.ts index f4149cef00..f300c55e64 100644 --- a/src/server/responses/native-steering-replay.ts +++ b/src/server/responses/native-steering-replay.ts @@ -5,9 +5,11 @@ */ export const MAX_NATIVE_STEERING_REPLAY_BYTES = 32 * 1024 * 1024; type Frame = Record; +/** Accept JSON object envelopes without treating arrays as records. */ function record(value: unknown): value is Frame { return value !== null && typeof value === "object" && !Array.isArray(value); } +/** Normalize string input to one user message while preserving array order. */ function inputItems(input: unknown): unknown[] { if (typeof input === "string") return [{ type: "message", role: "user", content: [{ type: "input_text", text: input }] }]; return Array.isArray(input) ? input : []; @@ -18,6 +20,7 @@ export interface NativeSteeringReplayObserver { dispose(): void; } +/** Keep bounded, connection-local input until a validated successor commits it. */ export class NativeSteeringReplay implements NativeSteeringReplayObserver { private prefix: unknown[]; private bytes: number; @@ -28,14 +31,17 @@ export class NativeSteeringReplay implements NativeSteeringReplayObserver { private explicitInput: unknown[] = []; private explicitBytes = 0; + /** Capture the initial prefix and reject over-budget history before dispatch. */ constructor(input: unknown, private readonly remember: (input: unknown[], response: Frame) => void) { this.prefix = [...inputItems(input)]; this.bytes = Buffer.byteLength(JSON.stringify(this.prefix)); this.check(); } + /** Reject overflow rather than silently truncating retained conversation input. */ private check(): void { if (this.bytes > MAX_NATIVE_STEERING_REPLAY_BYTES) throw new Error("Native steering replay exceeded its bounded history budget; input was not silently truncated."); } + /** Reserve replay bytes before send and return a rollback for synchronous failure. */ submitted(frame: Frame): () => void { const input = inputItems(frame.input); const bytes = Buffer.byteLength(JSON.stringify(input)); @@ -53,6 +59,7 @@ export class NativeSteeringReplay implements NativeSteeringReplayObserver { this.explicitBytes = bytes; return () => { this.explicitInput = []; this.bytes -= this.explicitBytes; this.explicitBytes = 0; }; } + /** Apply ordered upstream events; only created successors commit queued input. */ observe(frame: Frame): void { const response = record(frame.response) ? frame.response : undefined; const steer = record(frame.steer) ? frame.steer : undefined; @@ -71,7 +78,12 @@ export class NativeSteeringReplay implements NativeSteeringReplayObserver { } else if (frame.type === "response.created") { if (this.current) { const committed = this.submissions.filter(item => item.parent === this.current && item.id !== undefined); - this.prefix.push(...this.previousOutput, ...committed.flatMap(item => item.input), ...this.explicitInput); + // Byte-valid histories may exceed the runtime's positional-argument limit. + for (const item of this.previousOutput) this.prefix.push(item); + for (const submission of committed) { + for (const item of submission.input) this.prefix.push(item); + } + for (const item of this.explicitInput) this.prefix.push(item); this.submissions = this.submissions.filter(item => !committed.includes(item)); } this.explicitInput = []; @@ -100,6 +112,7 @@ export class NativeSteeringReplay implements NativeSteeringReplayObserver { if (frame.type === "response.completed") this.remember(this.prefix, { ...response, output }); } } + /** Release retained input, output and queued submissions when the owner detaches. */ dispose(): void { this.prefix = []; this.previousOutput = []; diff --git a/src/server/responses/native-steering.ts b/src/server/responses/native-steering.ts index 3190948684..741606ad43 100644 --- a/src/server/responses/native-steering.ts +++ b/src/server/responses/native-steering.ts @@ -11,25 +11,32 @@ export const NATIVE_STEERING_TOOL_WAIT_MS = 30 * 60_000; type Frame = Record; type Send = (frame: Frame) => void; const responses = new WeakSet(); +/** Mark the exact response for multi-response delivery without serializing a wire field. */ export function markNativeSteeringResponse(response: Response): Response { responses.add(response); return response; } +/** Recognize a marked native response by identity, not by caller-controlled content. */ export function isNativeSteeringResponse(response: Response): boolean { return responses.has(response); } +/** Narrow JSON object envelopes while excluding arrays and null. */ function record(value: unknown): value is Frame { return value !== null && typeof value === "object" && !Array.isArray(value); } +/** Require a bounded, nonempty protocol identity without control characters. */ function validId(value: unknown): value is string { return typeof value === "string" && value.length > 0 && Buffer.byteLength(value) <= CODEX_WS_ID_MAX_BYTES && !/[\u0000-\u001f\u007f]/.test(value); } +/** Serialize JSON settings deterministically so key order cannot change equality. */ function stable(value: unknown): string { if (Array.isArray(value)) return `[${value.map(stable).join(",")}]`; if (record(value)) return `{${Object.keys(value).sort().map(key => `${JSON.stringify(key)}:${stable(value[key])}`).join(",")}}`; return JSON.stringify(value) ?? "null"; } +/** Retain only a digest of pinned settings instead of their request payloads. */ function fingerprint(value: unknown): string { return createHash("sha256").update(stable(value)).digest("hex"); } /** Typed, content-free local protocol rejection. */ export class NativeSteeringError extends Error { + /** Create a content-free protocol error with a stable downstream rejection code. */ constructor(readonly code: string, message: string) { super(message); this.name = "NativeSteeringError"; } } @@ -72,6 +79,7 @@ function outputRequirement(item: unknown): Frame | undefined { return { type, call_id: item.call_id }; } +/** Match a saved result to its required stub, allowing an omitted optional name. */ function matchesRequirement(item: Frame, stub: Frame): boolean { // The wire may label a required result with its tool name, but the ordinary // function/custom output schema identifies the result by call_id, not name. @@ -107,17 +115,22 @@ export class NativeSteeringChannel { private readonly advertised = new Map(); private advertisedBytes = 0; + /** Pin the initial lane and setting digests without opening a transport. */ constructor(initial: Frame, private readonly idleMs = 300_000) { this.lane = initial.stream_id; for (const [key, value] of Object.entries(initial)) { if (!["type", "input", "previous_response_id", "stream", "stream_id"].includes(key)) this.settings.set(key, fingerprint(value)); } } + /** Report whether native dispatch ever bound a physical connection to this owner. */ get attached(): boolean { return this.everAttached; } + /** Report whether ordered terminal handling has finished the native chain. */ get ended(): boolean { return this.finished; } + /** Count unacknowledged or accepted submissions that still own the response chain. */ get hasOutstanding(): boolean { return [...this.parents.values()].some(parent => parent.unacknowledged > 0 || parent.accepted.size > 0); } + /** Report whether the server has requested a saved-result continuation. */ get awaitingContinuation(): boolean { return this.pendingParent !== undefined; } /** Bind only after native credentials have been selected and admission passed. */ @@ -145,6 +158,7 @@ export class NativeSteeringChannel { }; } + /** Replace the unrefed watchdog; timeout reports uncertainty instead of replaying. */ private wait(ms = NATIVE_STEERING_WAIT_MS): void { clearTimeout(this.timer); this.timer = setTimeout(() => { @@ -152,6 +166,7 @@ export class NativeSteeringChannel { }, ms); this.timer.unref?.(); } + /** Check the live owner and byte limit before journaling and sending one control. */ private liveSend(frame: Frame): void { if (!this.send || this.finished) throw new NativeSteeringError("steering_not_supported", "No active native WebSocket steering transport; this route may be disabled or using HTTP fallback."); if (codexWsCreateFrameExceedsLimit(JSON.stringify(frame))) throw new NativeSteeringError("invalid_input", "Native steering frame exceeds the upstream byte limit."); @@ -159,6 +174,7 @@ export class NativeSteeringChannel { try { this.send(frame); } catch (error) { rollback?.(); throw error; } } + /** Send user-only input to this connection's active response under the pending cap. */ steer(frame: Frame): void { validateSteeringFrame(frame); if (!this.send || this.finished) { this.liveSend(frame); return; } @@ -179,6 +195,7 @@ export class NativeSteeringChannel { } } + /** Retain bounded call or approval identities that authorize early saved results. */ private advertise(item: unknown): void { const stub = outputRequirement(item); if (!stub) return; diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md index 286806f825..c7e7820763 100644 --- a/structure/adapters/registry.md +++ b/structure/adapters/registry.md @@ -1,5 +1,7 @@ # Adapter Registry Authority +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Request-local adapter bindings are separate from registry authority in the Responses [core module ownership](../transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -183,7 +185,3 @@ implement legacy call/result pairing. Modern tool-image carriers are unchanged. raw passthrough; `tests/responses/chat-media-translation.test.ts` reaches the real HTTP translation boundary and verifies that rejection sends no upstream request. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/catalog.md b/structure/catalog.md index dcbca135aa..899c959dfa 100644 --- a/structure/catalog.md +++ b/structure/catalog.md @@ -1,5 +1,7 @@ # Model Catalog +Native steering follows [the shared WebSocket contract](transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Catalog discovery remains separate from the Responses final-route [core module ownership](transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -430,7 +432,3 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara ## Renamed destination reasoning metadata `src/providers/derive.ts` fills missing reasoning tables for renamed providers accepted by the existing fixed-key destination matcher. Model entries are cloned and explicit user entries (including empty arrays) win. Provider-wide effort defaults fill only when undefined; Command Code unknown models therefore keep the registry's empty picker policy unless overridden. Identity, transport and other capability axes are unchanged. The gathered row drives client exports; this metadata contract does not prove arbitrary gateway routing. - -The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 84d7036634..2a2db3e4f7 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -1,5 +1,7 @@ # Claude Desktop Integration +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Desktop callers retain their existing ingress through the Responses [core module ownership](../transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -168,7 +170,3 @@ The [explicit model-capability contract](../config.md#explicit-per-model-capabil Exact [model input declarations](../config.md#explicit-per-model-capability-declarations) now feed text-only eligibility and catalog hints; existing image-description/omission handling consumes them before the main upstream send. Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/config.md b/structure/config.md index f15e1aecf3..0a3c41bb1d 100644 --- a/structure/config.md +++ b/structure/config.md @@ -1,5 +1,7 @@ # Config Surface +Native steering follows [the shared WebSocket contract](transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Catalog HTTP acquisition follows the [proxy-routing contract](catalog.md#remote-catalog-http-proxy-routing). Configuration consumers retain the [refresh-lock ownership boundary](catalog.md#accounts-namespaces-and-pool-rotation); failing to establish a usable matching lock identity does not authorize deleting its path or replacing the refresh callback outcome with a path-probe error. Cooperating lock metadata changes serialize through the existing SQLite mutation transaction; release keeps the descriptor open through identity comparison and any unlink, then closes it. Failed metadata writes remove only a matching owned path after successful coordination; unknown identity, failed probes or unavailable coordination retain the path for stale recovery. Async refresh work holds no metadata transaction. @@ -322,7 +324,3 @@ The text-only consumer reads exact inputModalities declarations before legacy hi ## Catalog auto-refresh `catalogAutoRefresh` on `src/types/config.ts` stores an optional `enabled` / `intervalMinutes` section that defaults off: an absent key, an explicit false, and a malformed value all leave the scheduler dormant. `src/config/feature-flags.ts` resolves the cadence; an explicit `intervalMinutes: 0` keeps the unref'd timer idle, and any other value is clamped up to 15 minutes because upstream `/models` caches have not moved below that and a shorter tick only multiplies rate-limit exposure. `src/codex/catalog-auto-refresh.ts` is the module-singleton interval `src/server/background-lifecycle.ts` starts beside the quota reset poller; a tick that is enabled and non-dormant drives the same catalog-only converge funnel management mutations drive. The last-outcome record lives in `src/codex/catalog-refresh-status.ts` (when the tick finished, the normalized `CatalogDisposition`, whether the served model set changed, consecutive failures) and carries no provider or account detail. - -The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md index 14c6d53a31..721c4c6c04 100644 --- a/structure/data-planes/images.md +++ b/structure/data-planes/images.md @@ -1,5 +1,7 @@ # Images Data Plane +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Vision preprocessing and image/video/search execution use the Responses [core module ownership](../transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -121,7 +123,3 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior. Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index bbd15ccf43..5ab8ac3ca3 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -1,5 +1,7 @@ # Inbound Compatibility Surfaces +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Compatibility callers retain the public Responses ingress described by the [core module ownership](../transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -331,7 +333,3 @@ Modern `tool` images continue through the existing following-user carrier. These an OpenCodex conversion limit, not a provider capability claim. Final Responses-to-adapter admission follows the [registry contract](../adapters/registry.md#untranslated-input-media). Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index dddba453b2..98944d1122 100644 --- a/structure/gui-and-management-api.md +++ b/structure/gui-and-management-api.md @@ -1,5 +1,7 @@ # GUI And Management API +Native steering follows [the shared WebSocket contract](transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + The shared server request path follows the Responses [core module ownership](transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -652,7 +654,3 @@ The [explicit model-capability contract](config.md#explicit-per-model-capability Exact [model input declarations](config.md#explicit-per-model-capability-declarations) now feed text-only eligibility and catalog hints; existing image-description/omission handling consumes them before the main upstream send. The raw provider editor round-trips `autoReviewModel` and `autoReviewModelOverrides` through editor-owned DTO fields. POST/PATCH/PUT share validation; PUT copies schema-normalized values into the persisted and live candidate before adoption. Canonical `openai` rejects these fields, including clear forms. Field-masked writes (PATCH, editor PUT, reload) pin every registry-seed key and ignore operator overlays the seed never defines, most commonly `selectedModels`; POST keeps the exact-key comparison. Canonical `openai` still rejects `allowPrivateNetwork`, which must not short-circuit destination DNS checks on the ChatGPT forward row. Existing authentication, origin checks and stale-baseline protection still govern the writes. See [reviewer projection](catalog.md#provider-scoped-approval-reviewer). - -The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/ops/docs-and-release.md b/structure/ops/docs-and-release.md index 18da048626..ba26107c37 100644 --- a/structure/ops/docs-and-release.md +++ b/structure/ops/docs-and-release.md @@ -1,5 +1,7 @@ # Docs And Release +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Catalog HTTP acquisition follows the [proxy-routing contract](../catalog.md#remote-catalog-http-proxy-routing). Refresh-lock validation covers fresh unreadable locks, descriptor-matched release, path-probe failures preserving callback outcomes, and confirmed-owner unlink error handling in `tests/codex-integration/codex-account-store.test.ts`; the [catalog contract](../catalog.md#accounts-namespaces-and-pool-rotation) explicitly does not promise atomic compare-and-delete. Cooperating lock metadata changes serialize through the existing SQLite mutation transaction; release keeps the descriptor open through identity comparison and any unlink, then closes it. Failed metadata writes remove only a matching owned path after successful coordination; unknown identity, failed probes or unavailable coordination retain the path for stale recovery. Async refresh work holds no metadata transaction. @@ -392,7 +394,3 @@ Exact [model input declarations](../config.md#explicit-per-model-capability-decl Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. Renamed fixed-key providers receive [missing reasoning metadata](../catalog.md#renamed-destination-reasoning-metadata) during derivation; explicit per-model entries and provider defaults retain precedence. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md index 1089c1789f..8d11161221 100644 --- a/structure/ops/service-and-sidecars.md +++ b/structure/ops/service-and-sidecars.md @@ -1,5 +1,7 @@ # Background Service And Sidecars +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Service endpoints are unchanged by the Responses [core module ownership](../transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -182,7 +184,3 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c The [explicit model-capability contract](../config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior. Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/overview.md b/structure/overview.md index d412e5c6da..28345816b3 100644 --- a/structure/overview.md +++ b/structure/overview.md @@ -1,5 +1,7 @@ # Overview +Native steering follows [the shared WebSocket contract](transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + The configuration-only [plaintext V2 contract](subagents.md#plaintext-v2-agent-messages) is scoped to canonical ChatGPT Responses forwarding; other source-area behavior described here is unchanged. @@ -147,7 +149,3 @@ Translated Chat request construction uses the [inline-image budget](transports/s The [explicit model-capability contract](config.md#explicit-per-model-capability-declarations) preserves operator declarations through provider storage and catalog capture; it does not infer upstream capability or change this surface's routing behavior. Provider-scoped approval reviewer settings are projected by the [catalog owner](catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. - -The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/providers/chat-compat.md b/structure/providers/chat-compat.md index 11efc4dffe..58e2126278 100644 --- a/structure/providers/chat-compat.md +++ b/structure/providers/chat-compat.md @@ -1,5 +1,7 @@ # Chat Provider Compatibility +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + The configuration-only [plaintext V2 contract](../subagents.md#plaintext-v2-agent-messages) is scoped to canonical ChatGPT Responses forwarding; other source-area behavior described here is unchanged. Cursor's localized native-shell names follow the [routing-commentary guard contract](cursor.md#cursor-native-exec). @@ -363,7 +365,3 @@ refusal of original images is unchanged. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. Upstream API-key usage follows the [physical-attempt account attribution contract](../gui-and-management-api.md#upstream-key-account-attribution), independently of subscription quota observations. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/providers/kiro.md b/structure/providers/kiro.md index 132b4e5646..62d9919b8e 100644 --- a/structure/providers/kiro.md +++ b/structure/providers/kiro.md @@ -1,5 +1,7 @@ # Kiro Provider +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + The configuration-only [plaintext V2 contract](../subagents.md#plaintext-v2-agent-messages) is scoped to canonical ChatGPT Responses forwarding; other source-area behavior described here is unchanged. @@ -106,7 +108,3 @@ request on a request path. The marker carries a count and no URL, because a remo image URL can carry a signed token. Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md index fb0983baad..ada7431cca 100644 --- a/structure/providers/xai-grok.md +++ b/structure/providers/xai-grok.md @@ -1,5 +1,7 @@ # xAI Grok Provider +Native steering follows [the shared WebSocket contract](../transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + xAI uses the same shared credential and delivery policies through the Responses [core module ownership](../transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -143,7 +145,3 @@ Account quota surfaces use [safe probe diagnostics](../transports/inventory.md#a Live sideband admission and its bounded upstream handshake follow the [runtime contract](../runtime.md#live-sideband-handshake); the ordinary Responses WebSocket exchange remains separate. Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. - -The opt-in [native mid-turn steering contract](../transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/runtime.md b/structure/runtime.md index 553534da23..29dc811b30 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -1,5 +1,7 @@ # Runtime +Native steering follows [the shared WebSocket contract](transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Responses admission and finalization are composed through the [core module ownership](transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -492,7 +494,3 @@ stamps the configured key selected for the physical request. `src/server/request retains per-key attempt usage, and `src/usage/log.ts` validates and persists labels. The [account attribution contract](gui-and-management-api.md#upstream-key-account-attribution) defines identity, unknown records, and aggregation boundaries. - -The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/subagents.md b/structure/subagents.md index b0ef2fcb3c..cb75c5b715 100644 --- a/structure/subagents.md +++ b/structure/subagents.md @@ -1,5 +1,7 @@ # Subagents And Multi-Agent Surface +Native steering follows [the shared WebSocket contract](transports/streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Encrypted-task and fallback request handling follow the Responses [core module ownership](transports/responses.md#core-module-ownership). This surface retains its existing behavior. @@ -384,7 +386,3 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara Provider-scoped approval reviewer settings are projected by the [catalog owner](catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. Renamed fixed-key providers receive [missing reasoning metadata](catalog.md#renamed-destination-reasoning-metadata) during derivation; explicit per-model entries and provider defaults retain precedence. - -The opt-in [native mid-turn steering contract](transports/streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 90b0e25654..3ac73d1121 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -1,5 +1,7 @@ # Byte Accounting +Native steering follows [the shared WebSocket contract](streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + Responses body-reader limits and lifetime handling follow the [core module ownership](responses.md#core-module-ownership). This surface retains its existing behavior. @@ -41,7 +43,3 @@ Translated audio/file admission follows the [final-adapter input contract](../ad Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. Upstream API-key usage follows the [physical-attempt account attribution contract](../gui-and-management-api.md#upstream-key-account-attribution), independently of subscription quota observations. - -The opt-in [native mid-turn steering contract](streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index 256e682a09..f4cb1c6fa9 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -1,5 +1,7 @@ # Transport Inventory +Native steering follows [the shared WebSocket contract](streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + The existing Responses transport is divided by responsibility in the [core module ownership](responses.md#core-module-ownership). This surface retains its existing behavior. @@ -150,7 +152,3 @@ Renamed fixed-key providers receive [missing reasoning metadata](../catalog.md#r Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. - -The opt-in [native mid-turn steering contract](streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index dccc2c8ab7..b1b5b31b42 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -1,5 +1,7 @@ # Responses Transport +Native steering follows [the shared WebSocket contract](streaming-health.md#experimental-native-mid-turn-steering); this surface's defaults remain unchanged. + The configuration-only [plaintext V2 contract](../subagents.md#plaintext-v2-agent-messages) is scoped to canonical ChatGPT Responses forwarding; other source-area behavior described here is unchanged. Cursor's localized native-shell names follow the [routing-commentary guard contract](../providers/cursor.md#cursor-native-exec). @@ -904,7 +906,3 @@ route where this was first observed; explicit provider and operator caps may onl Regression coverage: `tests/server/input-admission.test.ts` and `tests/helpers/combo-context-headroom-cases.ts`. - -The opt-in [native mid-turn steering contract](streaming-health.md#experimental-native-mid-turn-steering) preserves this area's ordinary -authentication, routing and non-native behavior; its connection-local controls, replay journal -and multi-response lifetime are owned by the native Responses transport. diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index caac5bfe87..d3ab7f6b59 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -266,6 +266,8 @@ upstream connection. The connection remains bound to the credential selected by dispatch path and never enters the idle reuse pool. The normal authentication, admission, quota observation and pre-dispatch guard remain in force. `response.steer` accepts user-only input, preserves its target response ID, and cannot select another account or lane. +A steering owner is installed only after turn admission; warmup and capacity refusal leave +no retained owner. Superseding a turn clears its old owner before any early return. An acceptance acknowledges queued input, not application. The parent terminal is relayed, but the native chain ends only after outstanding submissions settle and the last response @@ -285,7 +287,8 @@ the existing thread-scoped replay cache. Rejected/uncommitted steer text is excl terminal outputs are reconstructed from completed output-item events. Derived state inherits the original non-persistable-body restriction from `src/responses/state/body-policy.ts`; `state.ts` keeps its existing public exports. The original request is never mutated. The -bounded journal is discarded at teardown. This keeps subsequent ordinary delta-input turns +bounded journal is discarded at teardown. Prefix arrays are appended iteratively, so a +byte-valid history cannot overflow the runtime's positional-argument stack. This keeps subsequent ordinary delta-input turns working without inventing IDs or silently dropping the steering instruction. Native chains bypass single-response SSE repair/terminal truncation. Wire IDs, lane IDs and diff --git a/tests/responses/ws-native-steering.test.ts b/tests/responses/ws-native-steering.test.ts index 558a348174..67708cea6c 100644 --- a/tests/responses/ws-native-steering.test.ts +++ b/tests/responses/ws-native-steering.test.ts @@ -8,6 +8,7 @@ import { NativeSteeringReplay, MAX_NATIVE_STEERING_REPLAY_BYTES } from "../../sr import { type WsData } from "../../src/server/ws-bridge"; import { getRequestLogEntries, clearRequestLogsForTests } from "../../src/server/request-log"; import { runOptionalShutdownHooks } from "../../src/lib/optional-shutdown-hooks"; +import { MAX_ACTIVE_TURNS, tryAdmitTurn } from "../../src/server/lifecycle"; import { configSchema } from "../../src/config/schema/config-schema"; type Frame = Record; @@ -380,3 +381,93 @@ test("early continuation validates advertised call and approval identities and r expect(sent).toHaveLength(2); detach(); }); + + +test("warmup leaves no steering owner and the next ordinary turn gets a fresh channel", async () => { + const { ws, sent, send } = downstream({ generate: false }); + expect(sent.map(frame => frame.type)).toEqual(["response.created", "response.completed"]); + expect(ws.data.nativeSteering).toBeUndefined(); + expect(ws.data.cancel).toBeUndefined(); + expect(Socket.all).toHaveLength(0); + send({ type: "response.steer", previous_response_id: sent[0].response.id, input: "not a running turn" }); + expect(sent.at(-1)?.error.code).toBe("steering_not_supported"); + send({ type: "response.create", model: "gpt-5.5", input: "real turn" }); + await waitFor(() => Socket.all.length === 1 && sent.filter(frame => frame.type === "response.created").length === 2); + expect(ws.data.nativeSteering?.attached).toBe(true); + const socket = Socket.all[0]; + expect(socket.frames[0].input).toBe("real turn"); + complete(socket, socket.root); + await waitFor(() => !ws.data.nativeSteering); +}); + +test("admission refusal leaves no steering owner and a later admitted turn is independent", async () => { + const leases: NonNullable>[] = []; + try { + for (let i = 0; i < MAX_ACTIVE_TURNS; i++) { + const lease = tryAdmitTurn(); + if (lease) leases.push(lease); + } + expect(leases.length).toBeGreaterThan(0); + const { ws, sent, send } = downstream(); + expect(sent.at(-1)?.error.code).toBe("server_busy"); + expect(ws.data.nativeSteering).toBeUndefined(); + expect(ws.data.cancel).toBeUndefined(); + expect(Socket.all).toHaveLength(0); + for (const lease of leases) lease.release(); + send({ type: "response.create", model: "gpt-5.5", input: "after admission" }); + await waitFor(() => sent.some(frame => frame.type === "response.created")); + expect(Socket.all).toHaveLength(1); + const socket = Socket.all[0]; + expect(socket.frames[0].input).toBe("after admission"); + expect(ws.data.nativeSteering?.attached).toBe(true); + complete(socket, socket.root); + await waitFor(() => !ws.data.nativeSteering); + } finally { + for (const lease of leases) lease.release(); + } +}); + +test("superseding an active turn with warmup clears its steering owner immediately", async () => { + const { ws, socket, send } = await begin(); + expect(ws.data.nativeSteering?.attached).toBe(true); + send({ type: "response.create", model: "gpt-5.5", input: "warmup", generate: false }); + expect(ws.data.nativeSteering).toBeUndefined(); + expect(ws.data.cancel).toBeUndefined(); + await waitFor(() => socket.readyState === 3); +}); + +test.each(["output", "steer", "continuation"] as const)("large %s arrays stay ordered below the replay byte limit", (source) => { + // 750,000 small, valid messages exceed the runtime argument-count limit while + // remaining within the unchanged 32 MiB history budget. + const items = Array.from({ length: 750_000 }, (_, i) => ({ + role: source === "output" ? "assistant" : "user", content: String(i), + })); + expect(Buffer.byteLength(JSON.stringify(items))).toBeLessThan(MAX_NATIVE_STEERING_REPLAY_BYTES - 1024); + let prefix: unknown[] = []; + const replay = new NativeSteeringReplay("initial", (input, response) => { + if (response.id === "large-successor") prefix = input.slice(); + }); + try { + replay.observe({ type: "response.created", response: { id: "large-parent" } }); + replay.submitted({ type: "response.steer", previous_response_id: "large-parent", + input: source === "steer" ? items : "committed steer" }); + replay.observe({ type: "response.steer.accepted", steer: { id: "large-steer", previous_response_id: "large-parent" } }); + const output = source === "output" ? items : [{ role: "assistant", content: "parent output" }]; + replay.observe({ type: "response.completed", response: { id: "large-parent", output } }); + replay.submitted({ type: "response.create", previous_response_id: "large-parent", + input: source === "continuation" ? items : "explicit continuation" }); + replay.observe({ type: "response.created", response: { id: "large-successor", previous_response_id: "large-parent" } }); + replay.observe({ type: "response.completed", response: { id: "large-successor", output: [] } }); + expect(prefix).toHaveLength(items.length + 3); + expect(prefix[0]).toEqual({ type: "message", role: "user", content: [{ type: "input_text", text: "initial" }] }); + const offset = source === "output" ? 1 : source === "steer" ? 2 : 3; + expect(prefix.slice(offset, offset + items.length)).toEqual(items); + if (source !== "output") expect(prefix[1]).toEqual(output[0]); + if (source !== "steer") expect(prefix[source === "output" ? items.length + 1 : 2]).toEqual({ + type: "message", role: "user", content: [{ type: "input_text", text: "committed steer" }], + }); + if (source !== "continuation") expect(prefix.at(-1)).toEqual({ + type: "message", role: "user", content: [{ type: "input_text", text: "explicit continuation" }], + }); + } finally { replay.dispose(); } +});