diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index 65e1201acc..774a4dc5e4 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -882,3 +882,41 @@ When returning to the root-override form, OpenCodex retains an existing `[model_ `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. + +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 +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 4d1fcd20da..7b93c8f3f7 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 aeb49ef74a..e9e550717d 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1487,7 +1487,8 @@ "codex-pool-refresh-backoff.test.ts": "codex-integration", "responses-account-change-scrub.test.ts": "responses", "response-log-inspection.test.ts": "server", - "request-log-nonstream.test.ts": "usage" + "request-log-nonstream.test.ts": "usage", + "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..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,27 +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 rememberResponseState( requestBody: unknown, response: { id?: unknown; output?: unknown; status?: unknown; incomplete_details?: unknown }, @@ -1264,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..ef4d608a06 --- /dev/null +++ b/src/responses/state/body-policy.ts @@ -0,0 +1,25 @@ +/** + * 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); +} + +/** 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 2a23f12762..f1c564c65a 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, @@ -188,11 +190,39 @@ 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?.(); + // 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) + ? 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; + } const turnId = (ws.data.turnId ?? 0) + 1; ws.data.turnId = turnId; const isCurrent = () => ws.data.turnId === turnId; @@ -227,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); @@ -267,6 +299,7 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { ...(wsAdmission ? { admission: wsAdmission } : {}), forceEmptyResponseId: true, inboundTransport: "websocket", + nativeSteering, abortSignal: turnAbort.signal, turnAdmissionLease, onFirstOutput: () => recordFirstOutput(logCtx, start), @@ -277,7 +310,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), { @@ -313,6 +349,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..213a91fd79 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, 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; + 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..f300c55e64 --- /dev/null +++ b/src/server/responses/native-steering-replay.ts @@ -0,0 +1,124 @@ +/** + * 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; +/** 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 : []; +} +export interface NativeSteeringReplayObserver { + submitted(frame: Frame): () => void; + observe(frame: Frame): void; + dispose(): void; +} + +/** Keep bounded, connection-local input until a validated successor commits it. */ +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; + + /** 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)); + 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; }; + } + /** 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; + 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); + // 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 = []; + 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 }); + } + } + /** Release retained input, output and queued submissions when the owner detaches. */ + 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..741606ad43 --- /dev/null +++ b/src/server/responses/native-steering.ts @@ -0,0 +1,320 @@ +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(); +/** 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"; } +} + +/** 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 }; + +/** 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 }; +} + +/** 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. + 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 and new user messages + * 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[] = []; + 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. */ + 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.advertised.clear(); + this.advertisedBytes = 0; + this.replay?.dispose(); + this.replay = undefined; + }; + } + + /** Replace the unrefed watchdog; timeout reports uncertainty instead of replaying. */ + 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?.(); + } + /** 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."); + const rollback = this.replay?.submitted(frame); + 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; } + 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; + } + } + + /** Retain bounded call or approval identities that authorize early saved results. */ + 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; + 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."); + } + const input = frame.input; + 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) { + // 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; } + 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.advertised.clear(); + this.advertisedBytes = 0; + 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.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 && !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/src/server/responses/passthrough-delivery.ts b/src/server/responses/passthrough-delivery.ts index 03e5aa0242..b45701dbcf 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"; @@ -312,6 +313,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 19a4de0bac..91633ce2d5 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, @@ -251,6 +252,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 ` @@ -772,6 +782,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, @@ -868,6 +881,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, @@ -974,6 +990,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, @@ -1095,6 +1114,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, @@ -1217,6 +1239,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 17bf93218b..3c427e808f 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 1eaa48e016..be6ebdaeec 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. diff --git a/structure/catalog.md b/structure/catalog.md index b6cea5318b..b42a190e69 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. diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index a2025bdb16..94caaf1068 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. diff --git a/structure/config.md b/structure/config.md index 7320aa8202..53a6112a9d 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. diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md index 99d4a68584..a621e17546 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. diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index c4acea5037..07eba5f5b9 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. diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index b4d9490772..cb3b7b23f3 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. diff --git a/structure/ops/docs-and-release.md b/structure/ops/docs-and-release.md index 74267d090f..03854a7b06 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. diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md index 2095aa257f..c9aced60cd 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. diff --git a/structure/overview.md b/structure/overview.md index 3b076ea986..c0b95b8c7c 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. diff --git a/structure/providers/chat-compat.md b/structure/providers/chat-compat.md index 380165c705..95c6c4e022 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). diff --git a/structure/providers/kiro.md b/structure/providers/kiro.md index 5132d9c550..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. diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md index 8fc9514b1a..950942980b 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. diff --git a/structure/runtime.md b/structure/runtime.md index fcb75a3e11..a5b1fe7df0 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. diff --git a/structure/subagents.md b/structure/subagents.md index bebd8fe79b..fdb108d743 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. diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 72ed6119ce..0c2e59e2ba 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. diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index 6b55bbaf0d..e820658bb9 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. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 14181d5b8c..b847b8bba6 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). diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index 6b901c9db4..c06436dd01 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -253,4 +253,58 @@ The [explicit model-capability contract](../config.md#explicit-per-model-capabil 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. +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 +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. 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 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. 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 +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. + Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index cede781c20..d0a98dd762 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1319,5 +1319,6 @@ "codex-pool-refresh-backoff.test.ts": "codex-integration", "responses-account-change-scrub.test.ts": "responses", "response-log-inspection.test.ts": "server", - "request-log-nonstream.test.ts": "usage" + "request-log-nonstream.test.ts": "usage", + "ws-native-steering.test.ts": "responses" } diff --git a/tests/helpers/responses-core-source.ts b/tests/helpers/responses-core-source.ts index 996d66ccdf..5131188bd3 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 new file mode 100644 index 0000000000..67708cea6c --- /dev/null +++ b/tests/responses/ws-native-steering.test.ts @@ -0,0 +1,473 @@ +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 { MAX_ACTIVE_TURNS, tryAdmitTurn } from "../../src/server/lifecycle"; +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(); +}); + +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(); +}); + + +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(); } +});