From c4677a66d217e3ad8ecb8481ef66fdce0e528546 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Thu, 17 Sep 2026 06:16:57 +0000 Subject: [PATCH 1/2] feat(responses): complete safe steering overrides, API transport and executable probe --- .../content/docs/guides/codex-integration.md | 85 +++++++++- .../docs/reference/configuration/server.md | 9 +- scripts/steering-probe.ts | 123 ++++++++++++++ scripts/steering-smoke.ts | 100 +++++++++++ scripts/test-layout/layout.json | 4 +- src/server/index/websocket-handler.ts | 5 +- src/server/responses/codex-ws-exchange.ts | 16 +- .../responses/native-response-control.ts | 19 ++- .../responses/native-steering-policy.ts | 49 ++++++ .../responses/native-steering-settings.ts | 76 +++++++++ src/server/responses/native-steering.ts | 11 +- src/server/responses/passthrough-dispatch.ts | 4 + src/server/responses/ws-upstream.ts | 2 +- src/server/ws-bridge.ts | 2 + structure/adapters/registry.md | 2 + structure/catalog.md | 2 + structure/clients/claude-desktop.md | 2 + structure/data-planes/images.md | 2 + structure/data-planes/inbound-compat.md | 2 + structure/gui-and-management-api.md | 2 + structure/ops/docs-and-release.md | 2 + structure/ops/service-and-sidecars.md | 2 + structure/overview.md | 2 + structure/providers/xai-grok.md | 2 + structure/runtime.md | 2 + structure/subagents.md | 2 + structure/transports/byte-accounting.md | 2 + structure/transports/inventory.md | 2 + structure/transports/responses.md | 2 + structure/transports/streaming-health.md | 50 +++++- tests/fixtures/test-layout-expected.json | 4 +- tests/helpers/responses-core-source.ts | 2 + .../responses/ws-steering-completion.test.ts | 160 ++++++++++++++++++ tests/responses/ws-steering-smoke.test.ts | 127 ++++++++++++++ 34 files changed, 852 insertions(+), 26 deletions(-) create mode 100644 scripts/steering-probe.ts create mode 100644 scripts/steering-smoke.ts create mode 100644 src/server/responses/native-steering-policy.ts create mode 100644 src/server/responses/native-steering-settings.ts create mode 100644 tests/responses/ws-steering-completion.test.ts create mode 100644 tests/responses/ws-steering-smoke.test.ts diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index baafd4afad..a98f5f4f05 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -896,7 +896,7 @@ options in `~/.opencodex/config.json` and restart OpenCodex before starting a fr ``` 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 +This option is off by default. It forwards steering to the same explicitly configured native WebSocket connection and selected account, preserving automatic successor responses and pending saved-tool-result continuations. Acceptance means queued, not yet applied. @@ -905,15 +905,15 @@ Results can arrive before `response.steer.pending`: the relay also matches the c 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. +Do not rerun tools or resend accepted steering text. Model, account, tool declarations and routing stay unchanged. Validated generation settings +may change in an explicit saved-result continuation as described below. Other changes +require 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 +HTTP fallback, noncanonical gateways, 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 have fixed 90-second acknowledgement or successor deadlines; +resubmit tools or steering text. Pending controls have per-submission absolute 90-second confirmation deadlines; saved-tool-result waits have a 30-minute cap. The implementation has synthetic protocol and regression coverage, not live Astra/client @@ -1066,3 +1066,76 @@ result remains available for a later explicit continuation. There is no automati conversion, retry, tool rerun or account/API switch. A single-agent steering turn can follow a completed multi-agent turn as a new explicit request using ordinary routing. Client support and backend entitlement still require live verification. + + +## Steering continuation settings and public API + +An explicit saved-result `response.create` may override `reasoning` (effort and +summary), `text` (verbosity and supported structured-output format), and +`stream_options`. On an explicitly configured public API route it may also +change `max_output_tokens`. Subscription routes refuse that token-limit override +instead of silently ignoring it. Normal provider pins, subagent caps, effort +mapping and summary/verbosity capability exclusions still apply. + +Omitted settings retain the current effective values; explicit null resets that +setting where the upstream accepts null. Overrides replace the supplied setting +object, not individual nested fields. Changed values carry into later explicit +continuations. A rejected override does not reserve the saved result, so a +corrected request can be submitted without rerunning its tool. The server still +decides which settings the chosen model accepts. Changes to model, account, +provider, tools, instructions or service tier require a separate ordinary turn. + +For public API steering, configure an `openai-responses` provider with exactly +`https://api.openai.com/v1`, its API key and `upstreamWebsocket: true`, then use its +normal prefixed model selector with `websockets: true` and +`codexNativeSteering: true`. This does not buy API credit or redirect a ChatGPT +subscription to separately billed usage. A supporting single-agent model/execution +mode is still required. Conversation-bound responses and API automatic compaction +are not steerable; their ordinary responses are preserved and a steering attempt +receives an explanatory error. The multi-agent injection path stays separate. + +### Executable direct-versus-proxy wire probe + +From a source checkout, run the offline positive control: + +```sh +bun scripts/steering-smoke.ts --self-test +``` + +Plan a comparison without reading tokens or opening any connection: + +```sh +bun scripts/steering-smoke.ts --direct wss://api.openai.com/v1/responses \ + --proxy ws://127.0.0.1:1455/v1/responses --model \ + --proxy-model +``` + +For a subscription comparison the direct URL is +`wss://chatgpt.com/backend-api/codex/responses`. Select the same actual model and +account on both routes; the script cannot prove that a proxy configuration selected +the same account. The proxy URL must be a loopback Responses endpoint and must not +contain credentials, query parameters or a fragment. + +Only after reviewing the plan, supply `STEERING_DIRECT_TOKEN` and +`STEERING_PROXY_TOKEN` through your shell environment and add **both** `--live` +and `--allow-model-requests`. A direct ChatGPT connection may additionally need +`STEERING_DIRECT_ACCOUNT_ID`; that header is never copied to the public API or the +proxy. Do not put credentials in command arguments, logs, screenshots or PRs. +The script does not read your saved Codex login, refresh tokens or change settings. + +Live execution sends four synthetic initial requests (two scenarios per route), +plus any resulting successors or required-result continuations, and **can consume +model usage**. One scenario checks an automatic successor; the other returns a +fixed synthetic result only for the script's own advertised function and changes +reasoning/verbosity on its explicit continuation. No external tool is executed and +no approval is inferred. There are no retries or automatic recovery requests. +Each scenario is limited to 120 seconds, 5,000 events and 2 MiB received data. + +The JSON report contains only outcomes, timing and boolean checkpoints. A pass +requires queued acceptance, a created successor and the synthetic marker in its +completed output. Missing confirmations are `unknown`; if the model never enters +the required-input path the result is `not_exercised`. Neither is counted as pass. +The process exits 0 only if all four live scenarios pass, 1 otherwise, and 2 for +invalid arguments or missing credentials. This is a **wire diagnostic**, not an +end-to-end Codex App/CLI interface test, live certification or instruction to enable +the experimental feature for production work. diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index ab2b96e2a2..06b1898af5 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -21,7 +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). | +| `codexNativeSteering?` | `boolean` | `false` | Experimental, native-only mid-turn steering on the Responses WebSocket endpoint. Requires `websockets: true`, a compatible upstream/client, and a pinned account/model/tool surface. Validated generation settings can change in explicit saved-result continuations. Does not enable translated models or HTTP fallback. See [native steering](/guides/codex-integration/#experimental-native-mid-turn-steering). | | `codexNativeInjection?` | `boolean` | `false` | Experimental saved function-result injection on compatible native multi-agent WebSocket turns. Requires `websockets: true`, explicit `multi_agent.enabled`, and an eligible provider. Separate from steering; no automatic tool rerun or recovery create. See [native injection](/guides/codex-integration/#experimental-native-function-result-injection). | | `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. | @@ -585,3 +585,10 @@ See [the continuation contract](/guides/codex-integration/#rich-tool-results-and completed output for local continuation history. See [steering confirmation deadlines and retained context](/guides/codex-integration/#steering-confirmation-deadlines-and-retained-context) for phase timing, unknown-delivery recovery and live-comparison precautions. + + +Native steering also supports explicitly opted-in canonical OpenAI API key-mode +WebSockets; it never moves subscription traffic to API billing. Its saved-result +continuations allow validated generation settings, while normal provider pins, +subagent caps and capability restrictions remain effective. See +[settings and the executable consent-gated probe](/guides/codex-integration/#steering-continuation-settings-and-public-api). diff --git a/scripts/steering-probe.ts b/scripts/steering-probe.ts new file mode 100644 index 0000000000..007d5bc219 --- /dev/null +++ b/scripts/steering-probe.ts @@ -0,0 +1,123 @@ +type Frame = Record; +export type ProbeScenario = "automatic" | "required-input"; +export type ProbeReport = { + scenario: ProbeScenario; outcome: "passed" | "failed" | "unknown" | "not_exercised"; + accepted: boolean; successorCreated: boolean; markerObserved: boolean; explicitContinuation: boolean; + sentControls: number; elapsedMs: number; code?: string; +}; +const MARKER = "STEERING_PROBE_OK"; +const safeCodes = new Set(["steering_not_supported", "response_not_active", "response_already_completed", + "invalid_input", "steering_settings_changed", "steering_settings_unsupported", "too_many_pending_steers"]); + +/** Content-free, single-attempt probe state. It never executes external tools or approval decisions. */ +export class SteeringProbe { + private base?: Frame; + private root?: string; + private successor?: string; + private steerId?: string; + private callId?: string; + private rootEnded = false; + private sentSteer = false; + private reportValue?: ProbeReport; + private bytes = 0; + private frames = 0; + private markerObserved = false; + private explicit = false; + private sentControls = 0; + private textTail = ""; + private started = performance.now(); + constructor(readonly scenario: ProbeScenario, private readonly send: (frame: Frame) => void) {} + + /** Only fixed synthetic prompts and a non-executing tool are sent by this harness. */ + request(model: string): Frame { + return this.base = { type: "response.create", model, store: false, reasoning: { effort: "low" }, + input: this.scenario === "automatic" + ? "Explain five techniques for organizing a fictional book collection. Work through each in detail." + : "Call steering_probe once, then use its saved result to answer briefly.", + ...(this.scenario === "required-input" ? { + tools: [{ type: "function", name: "steering_probe", description: "Returns a fixed synthetic fixture; performs no external action.", + parameters: { type: "object", properties: {}, required: [], additionalProperties: false }, strict: true }], + tool_choice: "auto", + } : {}), + }; + } + private steer(): void { + if (!this.root || this.sentSteer || this.rootEnded) return; + this.sentSteer = true; this.sentControls++; + this.send({ type: "response.steer", previous_response_id: this.root, input: `Change the answer: respond only with ${MARKER}. Do not run more tools.` }); + } + /** Stop with sanitized state, never returning IDs, model output, tokens or endpoint paths. */ + finish(outcome: ProbeReport["outcome"], code?: string): ProbeReport { + return this.reportValue ??= { scenario: this.scenario, outcome, accepted: !!this.steerId, + successorCreated: !!this.successor, markerObserved: this.markerObserved, explicitContinuation: this.explicit, + sentControls: this.sentControls, elapsedMs: Math.max(0, Math.round(performance.now() - this.started)), ...(code ? { code } : {}) }; + } + get report(): ProbeReport | undefined { return this.reportValue; } + + /** Observe a bounded wire stream. Acceptance alone is never a passing probe. */ + receive(raw: string): ProbeReport | undefined { + if (this.reportValue) return this.reportValue; + this.bytes += Buffer.byteLength(raw); + if (++this.frames > 5000 || this.bytes > 2 * 1024 * 1024) return this.finish("unknown", "probe_budget_exceeded"); + let event: Frame; + try { event = JSON.parse(raw); } catch { return this.finish("failed", "invalid_event"); } + if (!event || typeof event !== "object" || Array.isArray(event)) return this.finish("failed", "invalid_event"); + const response = event.response; + if (event.type === "response.created") { + if (!response || typeof response.id !== "string" || (!response.id.length || response.id.length > 512)) return this.finish("failed", "invalid_identity"); + if (!this.root) { this.root = response.id; if (this.scenario === "automatic") this.steer(); } + else { + if (this.successor || response.id === this.root || !this.rootEnded || !this.steerId + || (response.previous_response_id != null && response.previous_response_id !== this.root)) return this.finish("failed", "unexpected_successor"); + this.successor = response.id; + } + } else if (event.type === "response.output_item.done" && !this.successor && this.scenario === "required-input") { + const item = event.item; + if (event.response_id != null && event.response_id !== this.root) return this.finish("failed", "output_identity_mismatch"); + if (item?.type === "function_call" && item.name === "steering_probe" && typeof item.call_id === "string") { + if (this.callId && item.call_id !== this.callId) return this.finish("failed", "unexpected_tool"); + this.callId = item.call_id; this.steer(); + } + } else if (event.type === "response.steer.accepted") { + if (!this.sentSteer || this.steerId || event.steer?.previous_response_id !== this.root || typeof event.steer?.id !== "string") { + return this.finish("failed", "unexpected_acceptance"); + } + this.steerId = event.steer.id; + } else if (event.type === "response.steer.pending") { + if (!this.steerId || event.steer?.id !== this.steerId || event.steer?.previous_response_id !== this.root || !this.rootEnded) { + return this.finish("failed", "unexpected_pending"); + } + if (this.explicit) return this.finish("failed", "duplicate_pending"); + const stubs = event.required_input; + if (event.reason !== "waiting_for_required_input" || !Array.isArray(stubs) || stubs.length !== 1 + || stubs[0]?.type !== "function_call_output" || stubs[0]?.call_id !== this.callId || !this.callId) { + return this.finish("not_exercised", "unsupported_required_input"); + } + this.explicit = true; this.sentControls++; + this.send({ ...this.base, type: "response.create", previous_response_id: this.root, + ...(event.stream_id !== undefined ? { stream_id: event.stream_id } : {}), + input: [{ type: "function_call_output", call_id: this.callId, output: "synthetic saved result; no action was executed" }], + reasoning: { effort: "medium" }, text: { verbosity: "low" } }); + } else if (event.type === "response.steer.failed" || event.type === "error") { + const code = event.error?.code; + return this.finish("failed", safeCodes.has(code) ? code : "upstream_rejection"); + } else if (["response.completed", "response.incomplete", "response.failed"].includes(event.type)) { + if (this.root && response?.id === this.root) { + this.rootEnded = true; + if (!this.sentSteer) return this.finish("not_exercised", "no_steering_window"); + } else if (this.successor && response?.id === this.successor) { + for (const item of Array.isArray(response.output) ? response.output : []) { + for (const part of Array.isArray(item?.content) ? item.content : []) if (typeof part?.text === "string" && part.text.includes(MARKER)) this.markerObserved = true; + } + if (event.type !== "response.completed") return this.finish("failed", "successor_not_completed"); + if (this.scenario === "required-input" && !this.explicit) return this.finish("not_exercised", "required_input_not_observed"); + return this.finish(this.markerObserved ? "passed" : "failed", this.markerObserved ? undefined : "marker_missing"); + } else return this.finish("failed", "terminal_identity_mismatch"); + } else if (event.type === "response.output_text.delta" && this.successor && typeof event.delta === "string") { + if (event.response_id != null && event.response_id !== this.successor) return this.finish("failed", "output_identity_mismatch"); + const text = this.textTail + event.delta; + this.markerObserved ||= text.includes(MARKER); this.textTail = text.slice(-MARKER.length); + } + return undefined; + } +} diff --git a/scripts/steering-smoke.ts b/scripts/steering-smoke.ts new file mode 100644 index 0000000000..cdd80cb5aa --- /dev/null +++ b/scripts/steering-smoke.ts @@ -0,0 +1,100 @@ +import { SteeringProbe, type ProbeReport, type ProbeScenario } from "./steering-probe"; + +type Target = { url: string; model: string; headers: Record }; +const API = "wss://api.openai.com/v1/responses"; +const CHATGPT = "wss://chatgpt.com/backend-api/codex/responses"; +const USAGE = "bun scripts/steering-smoke.ts --self-test | --direct --proxy --model [--proxy-model ] [--live --allow-model-requests]"; + +/** Validate destinations before reading credentials. Default invocation never sends a model request. */ +export function probeTargets(args: string[], env: Record): { live: boolean; direct: Target; proxy: Target } { + const values = new Map(); let live = false; let consent = false; + for (let i = 0; i < args.length; i++) { + const key = args[i]; + if (key === "--live") { if (live) throw new Error(USAGE); live = true; continue; } + if (key === "--allow-model-requests") { if (consent) throw new Error(USAGE); consent = true; continue; } + if (!["--direct", "--proxy", "--model", "--proxy-model"].includes(key) || values.has(key) || !args[i + 1] || args[i + 1].startsWith("--")) throw new Error(USAGE); + values.set(key, args[++i]); + } + const directUrl = values.get("--direct"); + if (directUrl !== API && directUrl !== CHATGPT) throw new Error("Direct destination must be the canonical OpenAI API or ChatGPT Responses WebSocket."); + let proxyUrl: URL; + try { proxyUrl = new URL(values.get("--proxy") ?? ""); } catch { throw new Error("A loopback proxy URL is required."); } + if (!["ws:", "wss:"].includes(proxyUrl.protocol) || !["127.0.0.1", "[::1]", "localhost"].includes(proxyUrl.hostname) + || proxyUrl.username || proxyUrl.password || proxyUrl.search || proxyUrl.hash + || !["/responses", "/v1/responses", "/backend-api/codex/responses"].includes(proxyUrl.pathname)) throw new Error("Proxy must be a loopback Responses URL without credentials, query or fragment."); + const model = values.get("--model"); const proxyModel = values.get("--proxy-model") ?? model; + if (!model || !proxyModel || [model, proxyModel].some(value => value.length > 256 || /[\u0000-\u0020\u007f]/.test(value))) throw new Error("Explicit valid model selectors are required."); + if (live !== consent) throw new Error("Live execution requires both --live and --allow-model-requests; it can consume model usage."); + const headers = (kind: "DIRECT" | "PROXY") => { + const token = live ? env[`STEERING_${kind}_TOKEN`] : undefined; + if (live && (!token || /[\r\n\0]/.test(token))) throw new Error(`Set STEERING_${kind}_TOKEN in the environment, never on the command line.`); + return { "OpenAI-Beta": "responses_websockets=2026-02-06", ...(token ? { Authorization: `Bearer ${token}` } : {}) }; + }; + const directHeaders: Record = headers("DIRECT"); + const account = live && directUrl === CHATGPT ? env.STEERING_DIRECT_ACCOUNT_ID : undefined; + if (account) { if (/[\r\n\0]/.test(account)) throw new Error("Invalid account header."); directHeaders["chatgpt-account-id"] = account; } + return { live, direct: { url: directUrl, model, headers: directHeaders }, proxy: { url: proxyUrl.href, model: proxyModel, headers: headers("PROXY") } }; +} + +/** One root request per scenario, bounded socket, no retries and no arbitrary tool execution. */ +export function runSteeringProbe(target: Target, scenario: ProbeScenario): Promise { + return new Promise(resolve => { + let socket: WebSocket | undefined; let timer: ReturnType | undefined; let done = false; + const settle = (report: ProbeReport) => { if (done) return; done = true; clearTimeout(timer); try { socket?.close(); } catch { /* reporting never retries transport */ } resolve(report); }; + const probe = new SteeringProbe(scenario, frame => { + try { if (socket?.readyState !== WebSocket.OPEN) throw new Error(); socket.send(JSON.stringify(frame)); } + catch { settle(probe.finish("unknown", "send_outcome_unknown")); } + }); + try { socket = new WebSocket(target.url, { headers: target.headers, maxPayloadLength: 2 * 1024 * 1024 } as unknown as string[]); } + catch { settle(probe.finish("unknown", "connection_failed")); return; } + timer = setTimeout(() => settle(probe.finish("unknown", "probe_deadline")), 120_000); + socket.addEventListener("open", () => { + try { socket!.send(JSON.stringify(probe.request(target.model))); } + catch { settle(probe.finish("unknown", "send_outcome_unknown")); } + }); + socket.addEventListener("message", event => { + if (done) return; + if (typeof event.data !== "string") { settle(probe.finish("failed", "unsupported_wire_frame")); return; } + try { const report = probe.receive(event.data); if (report) settle(report); } + catch { settle(probe.finish("unknown", "probe_processing_failed")); } + }); + socket.addEventListener("error", () => settle(probe.finish("unknown", "connection_error"))); + socket.addEventListener("close", () => settle(probe.finish("unknown", "connection_closed"))); + }); +} + +/** Offline positive control; no socket, credential lookup or model request. */ +export function steeringProbeSelfTest(): ProbeReport { + const sent: Record[] = []; + const probe = new SteeringProbe("automatic", frame => sent.push(frame)); + for (const frame of [ + { type: "response.created", response: { id: "fixture-root" } }, + { type: "response.steer.accepted", steer: { id: "fixture-steer", previous_response_id: "fixture-root" } }, + { type: "response.incomplete", response: { id: "fixture-root" } }, + { type: "response.created", response: { id: "fixture-next", previous_response_id: "fixture-root" } }, + { type: "response.output_text.delta", delta: "STEERING_PROBE_OK" }, + { type: "response.completed", response: { id: "fixture-next", output: [] } }, + ]) probe.receive(JSON.stringify(frame)); + if (sent.length !== 1 || probe.report?.outcome !== "passed") throw new Error("Offline steering probe control failed."); + return probe.report; +} + +if (import.meta.main) { + try { + const args = process.argv.slice(2); + if (args.length === 1 && args[0] === "--self-test") console.log(JSON.stringify({ mode: "offline-fixture", result: steeringProbeSelfTest() }, null, 2)); + else if (!args.length || args.includes("--help")) console.log(USAGE); + else { + const targets = probeTargets(args, process.env); + if (!targets.live) console.log(JSON.stringify({ mode: "plan-only", modelRequestsSent: 0, scenarios: ["automatic", "required-input"], targets: ["direct", "proxy"] }, null, 2)); + else { + const reports = []; + for (const label of ["direct", "proxy"] as const) for (const scenario of ["automatic", "required-input"] as const) { + reports.push({ target: label, ...await runSteeringProbe(targets[label], scenario) }); + } + console.log(JSON.stringify({ mode: "live-single-attempt", reports }, null, 2)); + process.exitCode = reports.every(report => report.outcome === "passed") ? 0 : 1; + } + } + } catch (error) { console.error(error instanceof Error ? error.message : USAGE); process.exitCode = 2; } +} diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index d4a52ed85c..adebdb38bb 100644 --- a/scripts/test-layout/layout.json +++ b/scripts/test-layout/layout.json @@ -1492,7 +1492,9 @@ "ws-native-result-continuations.test.ts": "responses", "ws-native-injection.test.ts": "responses", "ws-native-steering.test.ts": "responses", - "ws-steering-stability.test.ts": "responses" + "ws-steering-stability.test.ts": "responses", + "ws-steering-completion.test.ts": "responses", + "ws-steering-smoke.test.ts": "responses" }, "migrated": [ "adapters", diff --git a/src/server/index/websocket-handler.ts b/src/server/index/websocket-handler.ts index c45eafda6d..17fbb6e30b 100644 --- a/src/server/index/websocket-handler.ts +++ b/src/server/index/websocket-handler.ts @@ -1,4 +1,4 @@ -import { nativeResponseControlMode, type NativeResponseControl } from "../responses/native-response-control"; +import { nativeSteeringUnavailableReason, nativeResponseControlMode, type NativeResponseControl } from "../responses/native-response-control"; import { NativeInjectionChannel } from "../responses/native-injection"; import { NativeSteeringChannel, NativeSteeringError } from "../responses/native-steering"; import { createNativeSteeringLogObserver } from "../responses/native-steering-log"; @@ -200,7 +200,7 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { return; } if (frame.type === "response.steer") { - if (!ws.data.nativeControl) throw new NativeSteeringError("steering_not_supported", "Native steering is disabled or unavailable on this route."); + if (!ws.data.nativeControl) throw new NativeSteeringError("steering_not_supported", ws.data.nativeSteeringUnavailable ?? "Native steering transport is unavailable; the route may be unsupported or using HTTP fallback."); ws.data.nativeControl.steer(frame); return; } @@ -232,6 +232,7 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) { ws.data.cancel?.(); // A superseded turn must not keep ownership during warmup or refusal. ws.data.nativeControl = undefined; + ws.data.nativeSteeringUnavailable = nativeSteeringUnavailableReason(frame, config.codexNativeSteering); const turnId = (ws.data.turnId ?? 0) + 1; ws.data.turnId = turnId; const isCurrent = () => ws.data.turnId === turnId; diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index c79e1cef54..d95be865fc 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -1,3 +1,4 @@ +import { mergeSteeringContinuation } from "./native-steering-settings"; import { markNativeControlResponse } from "./native-response-control"; import type { NativeResponseControl } from "./native-response-control"; import { MAX_CLIENT_SSE_FRAME_BYTES } from "../sse-frame-buffer"; @@ -121,6 +122,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise { const correlation = session.retainable ? new CodexWsCorrelation(session.reused, id => session.hasCompleted(id)) : null; let detachOwner = () => {}; let detachSteering = () => {}; + let continuationBase: Record | undefined; // 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. @@ -339,7 +341,8 @@ export function codexWsExchange(options: ExchangeOptions): Promise { try { if (nativeControl) { // Parsed once: the base body is immutable for this exchange, and a - // full-replay frame runs to megabytes. + // full-replay frame runs to megabytes. It seeds continuationBase on + // the first create frame. let base: Record | undefined; detachSteering = nativeControl.attach(frame => { const sendControl = () => { @@ -349,16 +352,19 @@ export function codexWsExchange(options: ExchangeOptions): Promise { 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. + // Generation overrides have passed route policy; identity/tools remain pinned. + // Keep the last explicit wire settings for later explicit and automatic successors. base ??= JSON.parse(frameText) as Record; - outgoing = { ...base, input: frame.input, previous_response_id: frame.previous_response_id }; + continuationBase ??= base; + outgoing = nativeControl.kind === "steering" + ? mergeSteeringContinuation(continuationBase, frame) + : { ...continuationBase, 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"); } + if (frame.type === "response.create") continuationBase = outgoing; try { ws.send(text); } catch { // A send failure has unknown delivery. Never replay or fall back. failStream("Native steering send failed; delivery is unknown"); diff --git a/src/server/responses/native-response-control.ts b/src/server/responses/native-response-control.ts index 290452a2eb..afd1558d7f 100644 --- a/src/server/responses/native-response-control.ts +++ b/src/server/responses/native-response-control.ts @@ -8,6 +8,7 @@ import { isInjectionRequest } from "./native-injection-protocol"; export interface NativeResponseControl { readonly kind?: "steering" | "injection"; relayActive: boolean; + normalizeContinuation?: (frame: Record) => Record; replayFactory?: () => NativeSteeringReplayObserver; readonly attached: boolean; readonly ended: boolean; @@ -26,10 +27,10 @@ export function markNativeControlResponse(response: Response): Response { native /** Recognize a marked native response by identity, not by caller-controlled content. */ export function isNativeControlResponse(response: Response): boolean { return nativeControlResponses.has(response); } -/** Preserve canonical ChatGPT eligibility; public API injection is separately opted in. */ +/** Preserve canonical ChatGPT eligibility; public API controls require an explicit provider WebSocket opt-in. */ export function nativeResponseControlEligible(provider: OcxProviderConfig, control?: NativeResponseControl): boolean { if (isCanonicalOpenAiForwardProvider(provider)) return true; - return control?.kind === "injection" && provider.adapter === "openai-responses" + return (control?.kind === "injection" || control?.kind === "steering") && provider.adapter === "openai-responses" && provider.upstreamWebsocket === true && provider.authMode !== "forward" && provider.baseUrl?.replace(/\/+$/, "") === "https://api.openai.com/v1"; } @@ -39,5 +40,17 @@ export function nativeResponseControlMode(frame: Record, flags: codexNativeInjection?: boolean; codexNativeSteering?: boolean; }): "injection" | "steering" | undefined { if (isInjectionRequest(frame)) return flags.codexNativeInjection === true ? "injection" : undefined; - return flags.codexNativeSteering === true ? "steering" : undefined; + return nativeSteeringUnavailableReason(frame, flags.codexNativeSteering) === undefined ? "steering" : undefined; +} + +/** Explain documented execution-mode exclusions without claiming model entitlement. */ +export function nativeSteeringUnavailableReason(frame: Record, enabled?: boolean): string | undefined { + if (enabled !== true) return "Native steering is disabled; enable codexNativeSteering and WebSockets for a supported route."; + if (isInjectionRequest(frame)) return "Multi-agent execution does not support single-agent response.steer; use a later client request."; + if (frame.conversation != null) return "Conversation-bound responses do not support native steering."; + if (Array.isArray(frame.context_management) && frame.context_management.some(item => + item && typeof item === "object" && (item as Record).type === "compaction")) { + return "Automatic API compaction and native steering cannot share an active response."; + } + return undefined; } diff --git a/src/server/responses/native-steering-policy.ts b/src/server/responses/native-steering-policy.ts new file mode 100644 index 0000000000..ab0329e36a --- /dev/null +++ b/src/server/responses/native-steering-policy.ts @@ -0,0 +1,49 @@ +import type { OcxConfig, OcxParsedRequest, OcxProviderConfig } from "../../types"; +import { nativeEffortClamp, shouldApplyNativeEffortClamp } from "../../codex/catalog"; +import { applyEffortCap, applyPinnedEffort, effortCapAppliesTo, prepareEffortNormalization, stripEmptyLadderEffort, supportedLadderFor } from "../effort-policy"; +import { collabSurface } from "./collaboration"; +import { mapRoutedResponsesReasoningEffort, normalizeConfiguredReasoningSummaryDelivery, + stripDisabledReasoningSummaries, stripDisabledVerbosity, stripUnsupportedReasoningSummaryDelivery } from "../../adapters/openai-responses/reasoning"; +import { NativeSteeringError } from "./native-steering"; +import { nativeResponseRecord as record } from "./native-response-json"; +import { STEERING_MUTABLE_SETTINGS } from "./native-steering-settings"; + +type Frame = Record; + +/** Reuse normal route-specific policy on a private override, never reroute or rebuild saved tool results. */ +export function createSteeringSettingsNormalizer( + parsed: OcxParsedRequest, + route: { provider: OcxProviderConfig; providerName: string; modelId: string }, + config: OcxConfig, + headers: Headers, +): (frame: Frame) => Frame { + const selector = prepareEffortNormalization(parsed, route); + const surface = collabSurface(parsed); + return (frame: Frame): Frame => { + if (route.provider.authMode === "forward" && Object.hasOwn(frame, "max_output_tokens")) { + throw new NativeSteeringError("steering_settings_unsupported", "This subscription route does not accept max_output_tokens; omit that override."); + } + // The frame is cloned by the channel. Policy receives only generation keys, + // not saved results, tool declarations, credentials or caller response IDs. + let body: Frame = Object.fromEntries(STEERING_MUTABLE_SETTINGS.filter(key => Object.hasOwn(frame, key)).map(key => [key, frame[key]])); + if (Object.hasOwn(body, "reasoning")) { + const candidate = { ...parsed, options: { ...parsed.options, + reasoning: record(body.reasoning) && typeof body.reasoning.effort === "string" ? body.reasoning.effort : undefined }, _rawBody: body }; + applyPinnedEffort(candidate, route, config, selector); + if (effortCapAppliesTo(surface, headers, config, parsed._compactionRequest === true)) { + applyEffortCap(candidate, headers, config, supportedLadderFor(route)); + } + const clamp = shouldApplyNativeEffortClamp(route.providerName, route.provider, route.modelId) + ? nativeEffortClamp(route.modelId, candidate.options.reasoning) : null; + if (clamp && record(body.reasoning)) body.reasoning.effort = clamp; + body = mapRoutedResponsesReasoningEffort(body, route.provider, route.modelId) as Frame; + body.reasoning = stripEmptyLadderEffort(body.reasoning, supportedLadderFor(route)); + } + body = stripDisabledVerbosity(stripDisabledReasoningSummaries( + normalizeConfiguredReasoningSummaryDelivery(stripUnsupportedReasoningSummaryDelivery(body, route.modelId), route.provider, route.modelId), + route.provider, route.modelId), route.provider, route.modelId) as Frame; + const next = { ...frame }; + for (const key of STEERING_MUTABLE_SETTINGS) if (Object.hasOwn(frame, key)) next[key] = body[key]; + return next; + }; +} diff --git a/src/server/responses/native-steering-settings.ts b/src/server/responses/native-steering-settings.ts new file mode 100644 index 0000000000..d58083396e --- /dev/null +++ b/src/server/responses/native-steering-settings.ts @@ -0,0 +1,76 @@ +import { REASONING_SUMMARY_DELIVERY_VALUES } from "../../types/wire"; +import { nativeResponseRecord as record } from "./native-response-json"; + +type Frame = Record; +/** Only generation settings may change without selecting a new route or tool surface. */ +export const STEERING_MUTABLE_SETTINGS = ["reasoning", "text", "max_output_tokens", "stream_options"] as const; +export const isSteeringMutableSetting = (key: string): boolean => + (STEERING_MUTABLE_SETTINGS as readonly string[]).includes(key); + +/** Bound nested structured-output schemas before fingerprinting or copying them. */ +function boundedJson(value: unknown): boolean { + const pending = [{ value, depth: 0 }]; + let nodes = 0; + while (pending.length) { + const current = pending.pop()!; + if (++nodes > 20_000 || current.depth > 64) return false; + if (current.value && typeof current.value === "object") { + for (const value of Object.values(current.value)) pending.push({ value, depth: current.depth + 1 }); + } + } + return true; +} +const only = (value: Frame, keys: readonly string[]) => Object.keys(value).every(key => keys.includes(key)); +const optionalEnum = (value: unknown, choices: readonly string[]) => value === undefined || value === null + || (typeof value === "string" && choices.includes(value)); + +/** Reject malformed overrides rather than treating them as omitted settings. */ +export function validSteeringSettings(frame: Frame): boolean { + for (const key of STEERING_MUTABLE_SETTINGS) { + if (!Object.hasOwn(frame, key)) continue; + const value = frame[key]; + if (!boundedJson(value)) return false; + if (value === null) continue; + if (key === "max_output_tokens") { + if (!Number.isSafeInteger(value) || (value as number) < 1) return false; + continue; + } + if (!record(value)) return false; + if (key === "reasoning") { + if (!only(value, ["effort", "summary", "generate_summary"]) + || !optionalEnum(value.effort, ["none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra"]) + || !optionalEnum(value.summary, ["auto", "concise", "detailed", "none"]) + || !optionalEnum(value.generate_summary, ["auto", "concise", "detailed", "none"])) return false; + } else if (key === "stream_options") { + if (!only(value, ["reasoning_summary_delivery", "include_usage", "include_obfuscation"]) + || !optionalEnum(value.reasoning_summary_delivery, REASONING_SUMMARY_DELIVERY_VALUES) + || [value.include_usage, value.include_obfuscation].some(item => item !== undefined && typeof item !== "boolean")) return false; + } else { + if (!only(value, ["format", "verbosity"]) || !optionalEnum(value.verbosity, ["low", "medium", "high"])) return false; + if (value.format === undefined || value.format === null) continue; + const format = value.format; + if (!record(format)) return false; + if (format.type === "text" || format.type === "json_object") { + if (!only(format, ["type"])) return false; + } else if (format.type === "json_schema") { + if (!only(format, ["type", "name", "schema", "strict", "description"]) + || typeof format.name !== "string" || !/^[a-zA-Z0-9_-]{1,64}$/.test(format.name) + || !record(format.schema) + || (format.strict != null && typeof format.strict !== "boolean") + || (format.description !== undefined && typeof format.description !== "string")) return false; + } else return false; + } + } + return true; +} + +/** Overlay only validated generation keys on the current, already authorized wire settings. */ +export function mergeSteeringContinuation(base: Frame, frame: Frame): Frame { + const outgoing: Frame = { ...base, input: frame.input, previous_response_id: frame.previous_response_id }; + for (const key of STEERING_MUTABLE_SETTINGS) { + if (!Object.hasOwn(frame, key)) continue; + if (frame[key] === undefined) delete outgoing[key]; + else outgoing[key] = frame[key]; + } + return outgoing; +} diff --git a/src/server/responses/native-steering.ts b/src/server/responses/native-steering.ts index 61bf13a871..b0ef4d68d4 100644 --- a/src/server/responses/native-steering.ts +++ b/src/server/responses/native-steering.ts @@ -1,3 +1,4 @@ +import { isSteeringMutableSetting, validSteeringSettings } from "./native-steering-settings"; import type { NativeSteeringReplayObserver } from "./native-steering-replay"; import { createHash } from "node:crypto"; import { CODEX_WS_ID_MAX_BYTES, CodexWsCorrelation } from "./codex-ws-correlation"; @@ -94,9 +95,11 @@ function matchesRequirement(item: Frame, stub: Frame): boolean { * 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. + * and validated generation overrides, but cannot change routing or tools. General new turns still use normal dispatch. */ export class NativeSteeringChannel { + readonly kind = "steering" as const; + normalizeContinuation?: (frame: Frame) => Frame; relayActive = false; replayFactory?: () => NativeSteeringReplayObserver; private replay?: NativeSteeringReplayObserver; @@ -288,8 +291,9 @@ export class NativeSteeringChannel { 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."); + if (!isSteeringMutableSetting(key) && this.settings.get(key) !== fingerprint(value)) throw new NativeSteeringError("steering_settings_changed", "The native steering continuation cannot change routing, tools or non-generation settings; start a separate turn instead."); } + if (!validSteeringSettings(frame)) throw new NativeSteeringError("invalid_input", "Invalid or oversized native steering generation settings."); 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(); @@ -305,6 +309,9 @@ export class NativeSteeringChannel { used.add(match); } if (used.size !== required.length) throw new NativeSteeringError("invalid_input", "Every required tool output or approval must be supplied exactly once."); + // Snapshot before asynchronous pacing; later caller mutation must not alter the authorized frame. + frame = structuredClone(frame); + if (this.normalizeContinuation) frame = this.normalizeContinuation(frame); this.continuationSent = true; this.continuationDeadline = performance.now() + NATIVE_STEERING_WAIT_MS; this.armTimer(); diff --git a/src/server/responses/passthrough-dispatch.ts b/src/server/responses/passthrough-dispatch.ts index f38f3a02af..56bca0310d 100644 --- a/src/server/responses/passthrough-dispatch.ts +++ b/src/server/responses/passthrough-dispatch.ts @@ -1,3 +1,4 @@ +import { createSteeringSettingsNormalizer } from "./native-steering-policy"; import { nativeResponseControlEligible } from "./native-response-control"; import { NativeInjectionReplay } from "./native-injection-replay"; import { NativeSteeringReplay } from "./native-steering-replay"; @@ -257,6 +258,9 @@ export async function preparePassthroughExchange( if (options.nativeControl && nativeResponseControlEligible(route.provider, options.nativeControl) && options.inboundTransport === "websocket" && !options.comboAttempt) { const body = parsed._rawBody as Record; + if (options.nativeControl.kind === "steering") { + options.nativeControl.normalizeContinuation = createSteeringSettingsNormalizer(parsed, route, config, req.headers); + } const Replay = options.nativeControl.kind === "injection" ? NativeInjectionReplay : NativeSteeringReplay; options.nativeControl.replayFactory = () => new Replay(body.input, (input, response) => { if (passthroughRecordEligible && !isBodyNonPersistable(body)) { diff --git a/src/server/responses/ws-upstream.ts b/src/server/responses/ws-upstream.ts index 69b5539b16..4359f46801 100644 --- a/src/server/responses/ws-upstream.ts +++ b/src/server/responses/ws-upstream.ts @@ -150,7 +150,7 @@ export function codexWsUpstreamFetch( // Never infer backend support from a model name or enable controls on a gateway. const control = nativeControl?.kind === "injection" ? ((prepared.canonical || url === OPENAI_API_RESPONSES_URL) && isInjectionRequest(JSON.parse(frameText)) ? nativeControl : undefined) - : prepared.canonical ? nativeControl : undefined; + : (prepared.canonical || url === OPENAI_API_RESPONSES_URL) ? nativeControl : undefined; if (control?.kind === "injection" && url === OPENAI_API_RESPONSES_URL) { const beta = headers["openai-beta"]; if (!beta?.split(",").some(value => value.trim() === "responses_multi_agent=v1")) { diff --git a/src/server/ws-bridge.ts b/src/server/ws-bridge.ts index eb562b43ae..e6782bc235 100644 --- a/src/server/ws-bridge.ts +++ b/src/server/ws-bridge.ts @@ -19,6 +19,8 @@ type ResponsesPayloadObserver = (payload: string) => void; export interface WsData { nativeControl?: NativeResponseControl; + /** Content-free per-turn explanation; never a model capability assertion. */ + nativeSteeringUnavailable?: string; 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 diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md index 8645581cde..44ecf77b6f 100644 --- a/structure/adapters/registry.md +++ b/structure/adapters/registry.md @@ -191,3 +191,5 @@ Canonical Responses identity sanitation and narrowly scoped pre-output combo rec Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](../transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](../transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/catalog.md b/structure/catalog.md index 2d0c1f6d92..4926f34917 100644 --- a/structure/catalog.md +++ b/structure/catalog.md @@ -442,3 +442,5 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index 6ccee591de..a1cdfad714 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -181,3 +181,5 @@ Provider-scoped approval reviewer settings are projected by the [catalog owner]( Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](../transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](../transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md index 9b80866978..58e6227af5 100644 --- a/structure/data-planes/images.md +++ b/structure/data-planes/images.md @@ -129,3 +129,5 @@ Provider-scoped approval reviewer settings are projected by the [catalog owner]( Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](../transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](../transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index 34cd677849..03e343c54f 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -339,3 +339,5 @@ Canonical Responses identity sanitation and narrowly scoped pre-output combo rec Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](../transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](../transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index 44eafc575b..8d7f5edf98 100644 --- a/structure/gui-and-management-api.md +++ b/structure/gui-and-management-api.md @@ -660,3 +660,5 @@ The raw provider editor round-trips `autoReviewModel` and `autoReviewModelOverri Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/ops/docs-and-release.md b/structure/ops/docs-and-release.md index 03854a7b06..1aa51abe58 100644 --- a/structure/ops/docs-and-release.md +++ b/structure/ops/docs-and-release.md @@ -428,3 +428,5 @@ Provider-scoped approval reviewer settings are projected by the [catalog owner]( Renamed fixed-key providers receive [missing reasoning metadata](../catalog.md#renamed-destination-reasoning-metadata) during derivation; explicit per-model entries and provider defaults retain precedence. Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](../transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](../transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md index c8b30cfb5c..c6597b7a8e 100644 --- a/structure/ops/service-and-sidecars.md +++ b/structure/ops/service-and-sidecars.md @@ -190,3 +190,5 @@ Provider-scoped approval reviewer settings are projected by the [catalog owner]( Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](../transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](../transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/overview.md b/structure/overview.md index c0b95b8c7c..695869ab63 100644 --- a/structure/overview.md +++ b/structure/overview.md @@ -160,3 +160,5 @@ The [explicit model-capability contract](config.md#explicit-per-model-capability 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. Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md index a4050bfa33..1043931311 100644 --- a/structure/providers/xai-grok.md +++ b/structure/providers/xai-grok.md @@ -157,3 +157,5 @@ Translated audio/file admission follows the [final-adapter input contract](../ad Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](../transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](../transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/runtime.md b/structure/runtime.md index 5961fb2444..94c8ad9fc8 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -533,3 +533,5 @@ retains per-key attempt usage, and `src/usage/log.ts` validates and persists lab defines identity, unknown records, and aggregation boundaries. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/subagents.md b/structure/subagents.md index 49d53eea2f..3ee59b82c6 100644 --- a/structure/subagents.md +++ b/structure/subagents.md @@ -392,3 +392,5 @@ Renamed fixed-key providers receive [missing reasoning metadata](catalog.md#rena Shared response-log retention and native SSE inspection pacing follow the [bounded inspection contract](transports/byte-accounting.md#response-log-inspection); other subsystem behavior remains unchanged. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](transports/streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 909824b37e..a229de333b 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -102,3 +102,5 @@ its active source; cancellation at an assistant boundary does not start the cont The same focused tests cover these lifecycle paths and Unicode code-unit limit boundaries. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index bf85742689..10670c2eb1 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -158,3 +158,5 @@ Canonical Responses identity sanitation and narrowly scoped pre-output combo rec 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. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/transports/responses.md b/structure/transports/responses.md index d23ae44f9a..9963bc058e 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -1092,3 +1092,5 @@ Regression coverage: `tests/server/input-admission.test.ts` and `tests/helpers/combo-context-headroom-cases.ts`. Native steering retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). + +Native steering generation overrides, explicit public-API eligibility and the consent-gated wire probe follow the [shared control contract](streaming-health.md#steering-settings-public-api-and-diagnostic-probe); this owner does not change routing or execute diagnostic tools. diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index cec983ca54..c4e6e1f7a7 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -256,9 +256,9 @@ Provider-scoped approval reviewer settings are projected by the [catalog owner]( ## 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 +Responses WebSocket endpoint. It requires `websockets: true`, a canonical ChatGPT forward +route or explicitly opted-in canonical OpenAI API route, an eligible Bun runtime, and a +supporting model/execution mode. HTTP fallback and translated/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 @@ -277,7 +277,8 @@ the saved results. Results may arrive before the pending event: completed output 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 +implementation pins routing, models and tools; validated generation overrides follow the +[continuation-setting contract](#steering-settings-public-api-and-diagnostic-probe). 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. @@ -428,3 +429,44 @@ contracts to deterministic clocks and a synthetic real-handler continuation test `src/server/responses/native-response-json.ts` owns content comparison without importing either control owner, keeping the replay dependency graph acyclic. Injection retains its existing helper export names and comparison semantics. + +## Steering settings, public API and diagnostic probe + +`native-steering-settings.ts` validates a bounded allowlist for explicit saved-result +continuations: `reasoning`, `text` (including structured-output format), +`stream_options` and public-API `max_output_tokens`. Unknown/malformed overrides +fail before result reservation. Null resets the supplied setting; omission keeps +the current authorized wire value. Models, tools, instructions, account, lane, +service tier, execution mode and other settings remain pinned. The schema uses +`REASONING_SUMMARY_DELIVERY_VALUES`, not a second invented enum. + +`native-steering-policy.ts` reuses normal selector pins, subagent caps, native +clamps, provider effort mapping, empty-ladder handling and summary/verbosity +capabilities on private generation-only data. Subscription output-token overrides +are explicitly refused. `codex-ws-exchange.ts` overlays normalized keys on the +current wire base, retaining new values across later explicit continuations. +Normal pacing and captured account/dispatch guards still run before physical send. +No tool results are transformed by generation normalization or rerun on rejection. + +Public API steering requires `openai-responses`, key-mode authentication, +`upstreamWebsocket: true` and exactly `https://api.openai.com/v1`. It uses its own +configured API key; subscription traffic is never migrated there. Injection-only +beta metadata is not attached to steering. Initial mode selection explains disabled, +multi-agent, conversation-bound and automatic-compaction exclusions without breaking +ordinary creates or inventing model entitlement. HTTP fallback remains non-steerable. + +`scripts/steering-probe.ts` and `scripts/steering-smoke.ts` provide a bounded, +content-free direct/proxy wire check. Default operation is plan-only; `--self-test` +is offline. Live runs require both consent flags and distinct explicit environment +credentials. Destinations are canonical upstream plus loopback, with no URL secrets, +query or fragments. The script never discovers stored credentials, modifies config, +executes tools/approvals, retries sends, or logs payloads/IDs. It checks acceptance, +successor creation and a synthetic result marker separately; an unobserved required- +input path is `not_exercised`, not pass. The live run uses at most four initial +synthetic requests plus resulting continuations, each bounded to 120 seconds, +5,000 events and 2 MiB received bytes. It can consume model usage and is not a +Codex App/CLI UI certification. The fixture suite also exercises real loopback sockets. + +`tests/responses/ws-steering-completion.test.ts` and `ws-steering-smoke.test.ts` +cover effective wire settings, immutable-route refusals, policy preservation, +independent API credentials, unavailable-mode diagnostics and safe probe outcomes. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index 2cc60235a0..5ba90d98db 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1324,5 +1324,7 @@ "ws-native-result-continuations.test.ts": "responses", "ws-native-injection.test.ts": "responses", "ws-native-steering.test.ts": "responses", - "ws-steering-stability.test.ts": "responses" + "ws-steering-stability.test.ts": "responses", + "ws-steering-completion.test.ts": "responses", + "ws-steering-smoke.test.ts": "responses" } diff --git a/tests/helpers/responses-core-source.ts b/tests/helpers/responses-core-source.ts index 6abfaca99a..a4f9ae1a9a 100644 --- a/tests/helpers/responses-core-source.ts +++ b/tests/helpers/responses-core-source.ts @@ -15,6 +15,8 @@ export const RESPONSES_CORE_MODULES = [ "native-injection-protocol.ts", "native-injection-replay.ts", "native-steering.ts", + "native-steering-settings.ts", + "native-steering-policy.ts", "native-steering-replay.ts", "codex-ws-correlation.ts", "core-lifetime.ts", diff --git a/tests/responses/ws-steering-completion.test.ts b/tests/responses/ws-steering-completion.test.ts new file mode 100644 index 0000000000..d0dd450c4e --- /dev/null +++ b/tests/responses/ws-steering-completion.test.ts @@ -0,0 +1,160 @@ +import { expect, test } from "bun:test"; +import { NativeSteeringChannel } from "../../src/server/responses/native-steering"; +import { createSteeringSettingsNormalizer } from "../../src/server/responses/native-steering-policy"; +import { validSteeringSettings } from "../../src/server/responses/native-steering-settings"; +import { nativeResponseControlEligible } from "../../src/server/responses/native-response-control"; +import { beginInjection, injectionConfig, installInjectionFixture, waitForInjection, InjectionSocket, + advertiseInjection, savedResult, type Frame } from "../helpers/native-injection-fixture"; +import type { OcxConfig, OcxParsedRequest, OcxProviderConfig } from "../../src/types"; + +installInjectionFixture(); +const config = (api = false) => ({ ...injectionConfig(api), codexNativeSteering: true }); +const begin = (api = false, fields: Frame = {}, settings = config(api)) => + beginInjection({ multi_agent: { enabled: false }, reasoning: { effort: "low" }, ...fields }, settings); +function accept(socket: InjectionSocket, id: string, number = 1) { + socket.emit({ type: "response.steer.accepted", steer: { id: `steer-${number}`, previous_response_id: id } }); +} +function pending(socket: InjectionSocket, id: string, number = 1) { + const call = advertiseInjection(socket, `call-${number}`); + socket.emit({ type: "response.completed", response: { id, status: "completed", output: [call] } }); + socket.emit({ type: "response.steer.pending", steer: { id: `steer-${number}`, previous_response_id: id }, + reason: "waiting_for_required_input", required_input: [{ type: "function_call_output", call_id: `call-${number}` }] }); +} + +test("public API steering uses only its explicit API-key route and preserves its beta tokens", async () => { + const c = await begin(true); + c.send({ type: "response.steer", previous_response_id: c.id, input: "new constraint" }); + expect(c.socket.frames.at(-1)?.type).toBe("response.steer"); + expect(c.socket.url).toBe("wss://api.openai.com/v1/responses"); + expect(c.socket.options.headers.authorization).toBe("Bearer fixture-public-key"); + expect(c.socket.options.headers["chatgpt-account-id"]).toBeUndefined(); + expect(c.socket.options.headers["openai-beta"]).toContain("fixture_beta=v1"); + expect(c.socket.options.headers["openai-beta"]).not.toContain("responses_multi_agent"); + accept(c.socket, c.id); + c.socket.emit({ type: "response.incomplete", response: { id: c.id, output: [], incomplete_details: { reason: "steered" } } }); + c.socket.emit({ type: "response.created", response: { id: "successor", previous_response_id: c.id } }); + c.socket.emit({ type: "response.completed", response: { id: "successor", status: "completed", output: [] } }); + await waitForInjection(() => !c.ws.data.nativeControl); + expect(InjectionSocket.all).toHaveLength(1); + expect(c.socket.frames).toHaveLength(2); +}); + +for (const api of [false, true]) test(`explicit settings survive two same-socket continuations (${api ? "API" : "subscription"})`, async () => { + const c = await begin(api); + c.send({ type: "response.steer", previous_response_id: c.id, input: "update" }); + accept(c.socket, c.id); pending(c.socket, c.id); + const override = { reasoning: { effort: "high", summary: "detailed" }, text: { verbosity: "low", format: { type: "json_object" } }, + ...(api ? { max_output_tokens: 256 } : {}) }; + c.send({ type: "response.create", previous_response_id: c.id, input: [savedResult()], ...override }); + await waitForInjection(() => c.socket.frames.length === 3); + expect(c.socket.frames[2]).toMatchObject(override); + expect(c.socket.frames[2].model).toBe("gpt-5.6-sol"); + expect(c.socket.frames[2].previous_response_id).toBe(c.id); + c.socket.emit({ type: "response.created", response: { id: "second", previous_response_id: c.id } }); + c.send({ type: "response.steer", previous_response_id: "second", input: "another update" }); + accept(c.socket, "second", 2); pending(c.socket, "second", 2); + c.send({ type: "response.create", previous_response_id: "second", input: [savedResult("call-2")] }); + await waitForInjection(() => c.socket.frames.length === 5); + expect(c.socket.frames[4]).toMatchObject(override); + expect(InjectionSocket.all).toHaveLength(1); +}); + +test("provider-pinned effort still wins over an explicit continuation override", async () => { + const settings = config(); settings.providers.openai.pinnedReasoningEffort = "low"; + const c = await begin(false, {}, settings); + c.send({ type: "response.steer", previous_response_id: c.id, input: "update" }); + accept(c.socket, c.id); pending(c.socket, c.id); + c.send({ type: "response.create", previous_response_id: c.id, input: [savedResult()], reasoning: { effort: "high" } }); + await waitForInjection(() => c.socket.frames.length === 3); + expect(c.socket.frames[2].reasoning.effort).toBe("low"); +}); + +test("an unsupported subscription output limit fails before reservation, allowing correction", async () => { + const c = await begin(); + c.send({ type: "response.steer", previous_response_id: c.id, input: "update" }); accept(c.socket, c.id); pending(c.socket, c.id); + c.send({ type: "response.create", previous_response_id: c.id, input: [savedResult()], max_output_tokens: 50 }); + expect(c.sent.at(-1)?.error.code).toBe("steering_settings_unsupported"); + expect(c.socket.frames).toHaveLength(2); + c.send({ type: "response.create", previous_response_id: c.id, input: [savedResult()], reasoning: { effort: "medium" } }); + await waitForInjection(() => c.socket.frames.length === 3); + expect(c.socket.frames[2].reasoning.effort).toBe("medium"); +}); + +for (const override of [ + { reasoning: { effort: "invented" } }, { reasoning: [] }, { reasoning: { injected: "no" } }, + { text: { verbosity: "huge" } }, { text: { format: { type: "json_schema", name: "invalid name", schema: {} } } }, + { max_output_tokens: 0 }, { max_output_tokens: 1.5 }, { stream_options: { include_usage: "yes" } }, +]) test(`malformed generation override is rejected: ${JSON.stringify(override)}`, () => { + expect(validSteeringSettings(override)).toBe(false); +}); + +for (const change of [{ model: "another-model" }, { tools: [] }, { service_tier: "priority" }, + { instructions: "replace policy" }, { multi_agent: { enabled: true } }, { conversation: "foreign" }]) { + test(`immutable continuation setting stays pinned: ${Object.keys(change)[0]}`, async () => { + const c = await begin(); + c.send({ type: "response.steer", previous_response_id: c.id, input: "update" }); accept(c.socket, c.id); pending(c.socket, c.id); + c.send({ type: "response.create", previous_response_id: c.id, input: [savedResult()], ...change }); + expect(c.sent.at(-1)?.error.code).toBe("steering_settings_changed"); + expect(c.socket.frames).toHaveLength(2); + }); +} + +test("normal policy retains subagent effort caps and configured capability exclusions", () => { + const provider = { ...config().providers.openai, modelSupportsVerbosity: { "gpt-5.6-sol": false }, modelSupportsReasoningSummaries: { "gpt-5.6-sol": false } }; + const parsed = { modelId: "gpt-5.6-sol", options: {}, context: {}, _rawBody: {} } as unknown as OcxParsedRequest; + const normalize = createSteeringSettingsNormalizer(parsed, { provider, modelId: parsed.modelId, providerName: "openai" }, + { ...config(), subagentEffortCap: "low" } as OcxConfig, new Headers({ "x-openai-subagent": "collab_spawn" })); + const result = normalize({ reasoning: { effort: "high", summary: "detailed" }, text: { verbosity: "high", format: { type: "text" } } }); + expect(result.reasoning).toEqual({ effort: "low" }); + expect(result.text).toEqual({ format: { type: "text" } }); + expect(parsed._rawBody).toEqual({}); +}); + +test("API effort mapping is the same as normal routed requests", () => { + const provider = { ...config(true).providers.api, reasoningEfforts: ["low", "medium", "high"] }; + const parsed = { modelId: "gpt-5.6-sol", options: {}, context: {}, _rawBody: {} } as unknown as OcxParsedRequest; + const normalize = createSteeringSettingsNormalizer(parsed, { provider, modelId: parsed.modelId, providerName: "api" }, config(true), new Headers()); + expect(normalize({ reasoning: { effort: "ultra" } }).reasoning).toEqual({ effort: "high" }); +}); + +test("queued continuation owns a private copy of both result and settings", () => { + const channel = new NativeSteeringChannel({ model: "fixture" }); + const sent: Frame[] = []; const detach = channel.attach(frame => sent.push(frame), () => {}); + try { + channel.observe({ type: "response.created", response: { id: "r" } }); + channel.steer({ type: "response.steer", previous_response_id: "r", input: "update" }); + channel.observe({ type: "response.steer.accepted", steer: { id: "s", previous_response_id: "r" } }); + channel.observe({ type: "response.completed", response: { id: "r", output: [{ type: "function_call", call_id: "c" }] } }); + const frame = { type: "response.create", previous_response_id: "r", input: [savedResult("c")], reasoning: { effort: "high" } }; + channel.continue(frame); frame.reasoning.effort = "low"; frame.input[0].output = "modified"; + expect(sent[1].reasoning.effort).toBe("high"); expect(sent[1].input[0].output).toBe("saved result"); + } finally { detach(); } +}); + +for (const override of [{ upstreamWebsocket: false }, { baseUrl: "https://gateway.example/v1" }, { authMode: "forward" }, { adapter: "openai-chat" }] as Partial[]) { + test(`public API eligibility does not widen other routes: ${Object.keys(override)[0]}`, () => { + const provider = { ...config(true).providers.api, ...override }; + expect(nativeResponseControlEligible(provider, new NativeSteeringChannel({}))).toBe(false); + }); +} + +for (const [fields, flags, reason] of [ + [{ conversation: "fixture-conversation" }, {}, "Conversation-bound"], + [{ context_management: [{ type: "compaction" }] }, {}, "compaction"], + [{ multi_agent: { enabled: true } }, { codexNativeInjection: false }, "Multi-agent"], + [{}, { codexNativeSteering: false }, "disabled"], +] as Array<[Frame, Frame, string]>) test(`handler explains unavailable steering without cancelling normal output: ${reason}`, async () => { + const c = await begin(false, fields, { ...config(), ...flags }); + c.send({ type: "response.steer", previous_response_id: c.id, input: "update" }); + expect(c.sent.at(-1)?.error.code).toBe("steering_not_supported"); + expect(c.sent.at(-1)?.error.message).toContain(reason); + expect(c.socket.frames).toHaveLength(1); expect(c.socket.readyState).toBe(1); +}); +for (const value of ["sequential", "sequential_cutoff", "concurrent", "concurrent_cutoff"]) { + test(`summary delivery uses the repository-owned wire enum: ${value}`, () => { + expect(validSteeringSettings({ stream_options: { reasoning_summary_delivery: value, include_obfuscation: true } })).toBe(true); + }); +} +test("invented summary-delivery enum is refused", () => { + expect(validSteeringSettings({ stream_options: { reasoning_summary_delivery: "buffered" } })).toBe(false); +}); diff --git a/tests/responses/ws-steering-smoke.test.ts b/tests/responses/ws-steering-smoke.test.ts new file mode 100644 index 0000000000..6fdf1b1661 --- /dev/null +++ b/tests/responses/ws-steering-smoke.test.ts @@ -0,0 +1,127 @@ +import { expect, test } from "bun:test"; +import { SteeringProbe } from "../../scripts/steering-probe"; +import { probeTargets, runSteeringProbe, steeringProbeSelfTest } from "../../scripts/steering-smoke"; +import { nativeResponseControlMode, nativeSteeringUnavailableReason } from "../../src/server/responses/native-response-control"; +import { createSteeringSettingsNormalizer } from "../../src/server/responses/native-steering-policy"; +import { validSteeringSettings } from "../../src/server/responses/native-steering-settings"; +import { injectionConfig } from "../helpers/native-injection-fixture"; + +const args = ["--direct", "wss://api.openai.com/v1/responses", "--proxy", "ws://127.0.0.1:1455/v1/responses", "--model", "fixture-model"]; +const created = (id: string, parent?: string) => ({ type: "response.created", response: { id, ...(parent ? { previous_response_id: parent } : {}) } }); +const accepted = { type: "response.steer.accepted", steer: { id: "s", previous_response_id: "r" } }; +const ended = { type: "response.completed", response: { id: "r", output: [] } }; +const pending = { type: "response.steer.pending", steer: accepted.steer, reason: "waiting_for_required_input", required_input: [{ type: "function_call_output", call_id: "c" }] }; +const finish = { type: "response.completed", response: { id: "n", output: [{ content: [{ type: "output_text", text: "STEERING_PROBE_OK" }] }] } }; +function fixture(mode: "automatic" | "required-input" = "automatic") { + const sent: any[] = []; const probe = new SteeringProbe(mode, frame => sent.push(frame)); probe.request("fixture-model"); + const emit = (frame: unknown) => probe.receive(JSON.stringify(frame)); emit(created("r")); + if (mode === "required-input") emit({ type: "response.output_item.done", item: { type: "function_call", name: "steering_probe", call_id: "c" } }); + return { probe, emit, sent }; +} + +test("offline positive control confirms acceptance, successor and marker separately", () => { + expect(steeringProbeSelfTest()).toMatchObject({ outcome: "passed", accepted: true, successorCreated: true, markerObserved: true, sentControls: 1 }); +}); +test("acceptance without a created and completed successor never passes", () => { + const f = fixture(); f.emit(accepted); + expect(f.probe.report).toBeUndefined(); expect(f.probe.finish("unknown", "connection_closed").outcome).toBe("unknown"); + expect(f.sent).toHaveLength(1); +}); +test("required-input probe sends full settings and saved synthetic result once", () => { + const f = fixture("required-input"); f.emit(accepted); f.emit(ended); f.emit(pending); + expect(f.sent).toHaveLength(2); expect(f.sent[1]).toMatchObject({ model: "fixture-model", store: false, tool_choice: "auto", reasoning: { effort: "medium" }, text: { verbosity: "low" } }); + expect(f.sent[1].input).toEqual([{ type: "function_call_output", call_id: "c", output: "synthetic saved result; no action was executed" }]); + f.emit(created("n", "r")); f.emit(finish); + expect(f.probe.report).toMatchObject({ outcome: "passed", explicitContinuation: true }); +}); +test("probe cannot invent approval or execute a foreign tool", () => { + const f = fixture("required-input"); f.emit(accepted); f.emit(ended); + f.emit({ ...pending, required_input: [{ type: "mcp_approval_response", approval_request_id: "approval" }] }); + expect(f.probe.report?.outcome).toBe("not_exercised"); expect(f.sent).toHaveLength(1); +}); +test("diagnostic report never leaks provider errors, response IDs or bodies", () => { + const f = fixture(); f.emit({ type: "error", error: { code: "private-secret-code", message: "secret-token-and-body" } }); + const report = JSON.stringify(f.probe.report); + expect(report).not.toContain("private-secret"); expect(report).not.toContain("secret-token"); + expect(f.probe.report?.code).toBe("upstream_rejection"); +}); +test("wrong-response marker cannot turn a probe green", () => { + const f = fixture(); f.emit(accepted); f.emit(ended); f.emit(created("n", "r")); + f.emit({ type: "response.output_text.delta", response_id: "foreign", delta: "STEERING_PROBE_OK" }); + expect(f.probe.report?.outcome).toBe("failed"); +}); +test("missing and reused response identity cannot be mistaken for a successor", () => { + const a = fixture(); a.emit({ type: "response.completed", response: {} }); expect(a.probe.report?.outcome).toBe("failed"); + const b = fixture(); b.emit(accepted); b.emit(ended); b.emit(created("r", "r")); expect(b.probe.report?.outcome).toBe("failed"); +}); +test("multiple pending notifications cannot produce duplicate continuations", () => { + const f = fixture("required-input"); f.emit(accepted); f.emit(ended); f.emit(pending); f.emit(pending); + expect(f.sent).toHaveLength(2); expect(f.probe.report?.code).toBe("duplicate_pending"); +}); +test("probe budgets cap data and prevent parsing arbitrary large output", () => { + const f = fixture(); f.probe.receive("x".repeat(2 * 1024 * 1024)); + expect(f.probe.report?.code).toBe("probe_budget_exceeded"); expect(f.sent).toHaveLength(1); +}); +test("plan-only mode does not read any token", () => { + const env = new Proxy({}, { get() { throw new Error("credential lookup was attempted"); } }); + const plan = probeTargets(args, env); expect(plan.live).toBe(false); expect(plan.direct.headers.Authorization).toBeUndefined(); +}); +for (const partial of [["--live"], ["--allow-model-requests"]]) test(`live consent pair required (${partial[0]})`, () => { + expect(() => probeTargets([...args, ...partial], {})).toThrow("both"); +}); +for (const url of ["wss://untrusted.example/v1/responses", "https://127.0.0.1/responses", "ws://user:secret@localhost/responses", "ws://localhost/responses?key=secret", "ws://localhost/private", "ws://localhost/responses#secret"]) { + test(`nonlocal or credential-bearing proxy refused ${url.split(":")[0]} ${url.length}`, () => { + expect(() => probeTargets(["--direct", args[1], "--proxy", url, "--model", "fixture"], {})).toThrow(); + }); +} +test("direct API and proxy use independent explicitly supplied headers", () => { + const plan = probeTargets([...args, "--live", "--allow-model-requests"], { STEERING_DIRECT_TOKEN: "fixture-direct", STEERING_PROXY_TOKEN: "fixture-proxy", STEERING_DIRECT_ACCOUNT_ID: "must-not-go-to-api" }); + expect(plan.direct.headers.Authorization).toBe("Bearer fixture-direct"); expect(plan.proxy.headers.Authorization).toBe("Bearer fixture-proxy"); + expect(plan.direct.headers["chatgpt-account-id"]).toBeUndefined(); expect(plan.proxy.headers["chatgpt-account-id"]).toBeUndefined(); +}); +test("executable self-test and plan mode succeed without credentials or sockets", async () => { + for (const parameters of [["--self-test"], args]) { + const child = Bun.spawn([process.execPath, "scripts/steering-smoke.ts", ...parameters], { stdout: "pipe", stderr: "pipe", env: { ...process.env, STEERING_DIRECT_TOKEN: "", STEERING_PROXY_TOKEN: "" } }); + const output = await new Response(child.stdout).text(); const err = await new Response(child.stderr).text(); + expect(await child.exited).toBe(0); expect(err).toBe(""); expect(JSON.parse(output).mode).toBe(parameters.length === 1 ? "offline-fixture" : "plan-only"); + } +}); +for (const scenario of ["automatic", "required-input"] as const) test(`actual bounded WebSocket probe works on isolated loopback (${scenario})`, async () => { + let roots = 0; let steers = 0; let continuations = 0; let authorization = ""; + const server = Bun.serve({ hostname: "127.0.0.1", port: 0, + fetch(req, server) { authorization = req.headers.get("authorization") ?? ""; if (server.upgrade(req)) return; return new Response(null, { status: 400 }); }, + websocket: { message(ws, raw) { + const frame = JSON.parse(String(raw)); const emit = (event: unknown) => ws.send(JSON.stringify(event)); + if (frame.type === "response.steer") { + steers++; emit(accepted); emit(ended); + if (scenario === "required-input") emit(pending); else { emit(created("n", "r")); emit(finish); } + } else if (!frame.previous_response_id) { + roots++; emit(created("r")); + if (scenario === "required-input") emit({ type: "response.output_item.done", item: { type: "function_call", name: "steering_probe", call_id: "c" } }); + } else { continuations++; emit(created("n", "r")); emit(finish); } + } }, + }); + try { + const report = await runSteeringProbe({ url: `ws://127.0.0.1:${server.port}/responses`, model: "fixture", headers: { Authorization: "Bearer fixture-loopback" } }, scenario); + expect(report.outcome).toBe("passed"); expect(roots).toBe(1); expect(steers).toBe(1); expect(continuations).toBe(scenario === "automatic" ? 0 : 1); expect(authorization).toBe("Bearer fixture-loopback"); + } finally { await server.stop(true); } +}); +for (const [frame, text] of [[{}, "disabled"], [{ multi_agent: { enabled: true } }, "multi-agent"], [{ conversation: "fixture" }, "conversation"], [{ context_management: [{ type: "compaction" }] }, "compaction"]] as const) { + test(`explicit unavailable reason (${text}) does not promise steering`, () => { + expect(nativeSteeringUnavailableReason(frame, text !== "disabled")?.toLowerCase()).toContain(text); + expect(nativeResponseControlMode(frame, { codexNativeSteering: text !== "disabled" })).toBeUndefined(); + }); +} +test("nullable conversation and non-compaction context keep supported steering selectable", () => { + expect(nativeResponseControlMode({ conversation: null, context_management: [] }, { codexNativeSteering: true })).toBe("steering"); +}); +test("no-effort-control policy strips only effort from mutable reasoning", () => { + const cfg = injectionConfig(true); const provider = { ...cfg.providers.api, noReasoningModels: ["fixture"] }; + const normalize = createSteeringSettingsNormalizer({ modelId: "fixture", options: {}, context: {}, _rawBody: {} } as any, { providerName: "api", provider, modelId: "fixture" }, cfg, new Headers()); + expect(normalize({ reasoning: { effort: "high", summary: "detailed" } }).reasoning).toEqual({ summary: "detailed" }); +}); +test("structured schemas are bounded, preserve field names, and allow explicit summary none", () => { + expect(validSteeringSettings({ reasoning: { summary: "none" }, text: { format: { type: "json_schema", name: "fixture", strict: true, schema: { type: "object", properties: { model: { type: "string" } } } } } })).toBe(true); + let schema: any = {}; for (let i = 0; i < 70; i++) schema = { nested: schema }; + expect(validSteeringSettings({ text: { format: { type: "json_schema", name: "fixture", schema } } })).toBe(false); +}); From 5c79b218a6cdd999839e242905b421359ffbe2f0 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Thu, 17 Sep 2026 06:36:47 +0000 Subject: [PATCH 2/2] docs: clarify steering routes and consolidate control guidance --- .../content/docs/guides/codex-integration.md | 6 +++-- .../docs/reference/configuration/server.md | 23 +++++-------------- 2 files changed, 10 insertions(+), 19 deletions(-) diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index a98f5f4f05..b007408cb9 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -885,8 +885,10 @@ Do not rewrite an active paginated rollout or thread row to migrate those conver ## Experimental native mid-turn steering -For a compatible model on the canonical ChatGPT forward route and a client that sends `response.steer`, enable both -options in `~/.opencodex/config.json` and restart OpenCodex before starting a fresh turn: +For a compatible model on the canonical ChatGPT forward route or an explicitly configured +[OpenAI API WebSocket route](#steering-continuation-settings-and-public-api), and a client +that sends `response.steer`, enable both options in `~/.opencodex/config.json` and restart +OpenCodex before starting a fresh turn: ```json { diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 06b1898af5..3aaba75fc0 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -574,21 +574,10 @@ companion form is what makes a hub a single-port deployment, and it is refused o wildcard `hostname`, where the public listener already holds `127.0.0.1:`. -The opt-in `codexNativeInjection` owner also accepts typed saved-result -continuations on the original account/socket. It does not widen -`response.inject` beyond string-valued function results, and multi-agent requests -never acquire the single-agent steering owner merely because injection is disabled. -See [the continuation contract](/guides/codex-integration/#rich-tool-results-and-explicit-approvals-after-response-completion). +## Experimental native response controls - -`codexNativeSteering` confirmation uses fixed absolute deadlines and retains -completed output for local continuation history. See -[steering confirmation deadlines and retained context](/guides/codex-integration/#steering-confirmation-deadlines-and-retained-context) -for phase timing, unknown-delivery recovery and live-comparison precautions. - - -Native steering also supports explicitly opted-in canonical OpenAI API key-mode -WebSockets; it never moves subscription traffic to API billing. Its saved-result -continuations allow validated generation settings, while normal provider pins, -subagent caps and capability restrictions remain effective. See -[settings and the executable consent-gated probe](/guides/codex-integration/#steering-continuation-settings-and-public-api). +`codexNativeSteering` and `codexNativeInjection` enable separate, default-off native +WebSocket control paths. See the canonical guide for +[supported steering routes and settings](../../guides/codex-integration.md#steering-continuation-settings-and-public-api), +[typed result and approval continuations](../../guides/codex-integration.md#rich-tool-results-and-explicit-approvals-after-response-completion), +and [confirmation deadlines and retained context](../../guides/codex-integration.md#steering-confirmation-deadlines-and-retained-context).