feat(responses): preserve native mid-turn steering over WebSocket

docs-site/src/content/docs/guides/codex-integration.md

@@ -882,3 +882,41 @@ When returning to the root-override form, OpenCodex retains an existing `[model_
 `ocx restore` and Codex config removal still refuse on `history_paginated_requires_native_writer`. Stripping the `[model_providers.opencodex]` definition while thread rows still reference it would make those conversations unresolvable, and the restore path has no way to keep a compatibility provider table. A home that is already paginated cannot currently be uninstalled through the product; that is known open work rather than intended behaviour.
 
 Do not rewrite an active paginated rollout or thread row to migrate those conversations yourself. Close the affected conversation before any recovery, and report the exact error and versions without uploading private history. A backup or a successful script alone does not prove the conversation is visible again. Check the restored conversation in Codex after reopening.
+
+## Experimental native mid-turn steering
+
+For a compatible native OpenAI model and a client that sends `response.steer`, enable both
+options in `~/.opencodex/config.json` and restart OpenCodex before starting a fresh turn:
+
+```json
+{
+  "websockets": true,
+  "codexNativeSteering": true
+}
+```
+
+Merge these keys into the existing configuration; do not replace your provider/account settings.
+This option is off by default. It forwards steering to the same native ChatGPT WebSocket
+connection and selected account, preserving automatic successor responses and pending
+saved-tool-result continuations. Acceptance means queued, not yet applied.
+
+Supply the required tool results or approval decisions **once per parent**, on the same lane.
+Results can arrive before `response.steer.pending`: the relay also matches the completed
+parent's advertised calls and approvals. A `name` on a pending function-output stub is
+optional on the result, as in the native schema. Additional user messages may accompany
+these results; system/developer messages, duplicate results and unrelated call IDs are refused.
+Do not rerun tools or resend accepted steering text. This first implementation requires
+unchanged model and request settings. A changed model/settings requires an explicitly stopped or finished turn
+and normal new dispatch. Multiple independent conversations use independent connections.
+
+HTTP fallback, other providers, translated models, sidecars, Combo attempts and plaintext V2
+restoration do not support this option. It does not add steering capability to a model or
+a client that lacks it. Unsupported routes return a protocol error rather than silently
+ignoring input. Disconnected or timed-out delivery may be unknown: never automatically
+resubmit tools or steering text. Pending controls time out after 90 seconds of inactivity;
+saved-tool-result waits have a 30-minute cap.
+
+The implementation has synthetic protocol and regression coverage, not live Astra/client
+certification. Keep the option disabled for production work until your client/model path
+has been verified. Set `codexNativeSteering` to `false` and restart to restore the existing
+single-response relay; no account or conversation files need to be deleted.

docs-site/src/content/docs/reference/configuration/server.md

@@ -21,6 +21,7 @@ runs helper features around provider requests.
 | `connectTimeoutMs?` | `number` | `200000` | Per-attempt DNS/TCP/TLS/final-header deadline; it ends before body generation. |
 | `shutdownTimeoutMs?` | `number` | `5000` | Graceful drain deadline before active turns are aborted. |
 | `websockets?` | `boolean` | `false` | Advertise and admit the client-facing Responses WebSocket path. False keeps clients on HTTP/SSE; it does not disable an eligible canonical ChatGPT upstream WS optimization. Complete-input requests may reuse an upstream connection within the same selected credential, account, thread and turn; changed handshake policy or missing identity keeps requests on separate connections. This does not trim HTTP input or create previous-response IDs. |
+| `codexNativeSteering?` | `boolean` | `false` | Experimental, native-only mid-turn steering on the Responses WebSocket endpoint. Requires `websockets: true`, a compatible upstream/client, and unchanged model/settings for saved-tool-result continuations. Does not enable translated models or HTTP fallback. See [native steering](/guides/codex-integration/#experimental-native-mid-turn-steering). |
 | `corsAllowOrigins?` | `string[]` | `[]` | Additional exact origins allowed by CORS. Loopback origins are always allowed. Authority-based browser extension origins such as `chrome-extension://<extension-id>` are supported; `*` is not a wildcard. Firefox and Safari regenerate the extension UUID (per install / per browser launch), so update the entry when the origin changes. |
 | `apiKeys?` | `OcxApiKey[]` | `[]` | Generated `ocx_…` credentials accepted by management and data-plane auth on non-loopback binds. Dashboard-managed. |
 | `storageCleanupPolicy?` | `StorageCleanupPolicy` | disabled | Opt-in archived-session cleanup policy. Never enabled implicitly. |

scripts/test-layout/layout.json

@@ -1487,7 +1487,8 @@
     "codex-pool-refresh-backoff.test.ts": "codex-integration",
     "responses-account-change-scrub.test.ts": "responses",
     "response-log-inspection.test.ts": "server",
-    "request-log-nonstream.test.ts": "usage"
+    "request-log-nonstream.test.ts": "usage",
+    "ws-native-steering.test.ts": "responses"
   },
   "migrated": [
     "adapters",

src/config/schema/config-schema.ts

@@ -59,6 +59,7 @@ import { parseDesktopProfile } from "../../claude/desktop-profile";
 import { DEFAULT_APP_OWNED_MEMORY_BUDGET_BYTES, MAX_APP_OWNED_MEMORY_BUDGET_MB, MIN_APP_OWNED_MEMORY_BUDGET_MB } from "../../lib/app-owned-memory";
 
 export const configSchema = z.object({
+  codexNativeSteering: z.boolean().optional().catch(false),
   port: z.number().int().min(0).max(65535).default(10100),
   // A malformed hand edit must disable only remote-role behavior, not discard
   // providers or data-plane keys. Live writes are rejected explicitly below.

src/responses/state.ts

@@ -23,6 +23,8 @@ import type { ResponseSpillWriteFailureCode, ResponseSpillWriteStatus, ResponseS
 export { responseAdmissionCountersForTests } from "./state/spill-failure";
 import { admissionCounters, noteSpillWriteFailure, noteSpillWriteSuccess, spillCounters, spillWriteHealth } from "./state/spill-failure";
 import { loadSnapshotEntry } from "./state/snapshot-codec";
+import { isBodyNonPersistable } from "./state/body-policy";
+export { isBodyNonPersistable, markBodyNonPersistable } from "./state/body-policy";
 export { flushPendingResponseSpillsForTests, awaitResponseSpillPublicationTailForTests, pendingResponseSpillMetricsForTests, setResponseSpillShutdownBudgetForTests, setResponseSpillAsyncAclAttemptBudgetForTests, setResponseSpillShutdownTerminalizationPassLimitForTests } from "./state/spill-queue";
 import {
   bindSpillQueueStore,
@@ -1235,27 +1237,6 @@ export function responseStateMetrics(): ResponseStateMetrics {
  * Cache completed output and max_output_tokens partial output for previous_response_id replay.
  * Content-filtered incomplete and failed output are not authoritative replay history.
  */
-/**
- * Request bodies that must never enter the continuation cache.
- *
- * The cache is persisted to `responses-state.json`, so anything recorded here reaches disk.
- * Encrypted-agent-task recovery decrypts task text into the request body and promises
- * in-memory, TTL-bounded retention; recording that body would put the plaintext on disk with
- * no TTL and break the promise.
- *
- * A WeakSet rather than a body field on purpose: `_rawBody` is serialized verbatim by the
- * native passthrough, so any marker written into the body itself would be sent upstream.
- * Marking is enforced once here rather than at each call site, because every recording path
- * (streaming, non-streaming, passthrough, forced) funnels through `rememberResponseState` —
- * a new call site cannot reintroduce the leak by forgetting a guard.
- */
-const nonPersistableBodies = new WeakSet<object>();
-
-/** Bar this exact request body from the continuation cache, and therefore from disk. */
-export function markBodyNonPersistable(body: unknown): void {
-  if (body && typeof body === "object") nonPersistableBodies.add(body as object);
-}
-
 export function rememberResponseState(
   requestBody: unknown,
   response: { id?: unknown; output?: unknown; status?: unknown; incomplete_details?: unknown },
@@ -1264,7 +1245,7 @@ export function rememberResponseState(
 ): void {
   if (!requestBody || typeof requestBody !== "object" || Array.isArray(requestBody)) return;
   const request = requestBody as Record<string, unknown>;
-  if (nonPersistableBodies.has(request)) return;
+  if (isBodyNonPersistable(request)) return;
   // `force` bypasses only the store:false skip: Codex sends `store:false` on every non-Azure
   // HTTP request (and WS inherits it), yet its WS turns still chain with previous_response_id.
   // The passthrough branch records with force so those chains can be expanded locally; the

src/responses/state/body-policy.ts

@@ -0,0 +1,25 @@
+/**
+ * Request bodies that must never enter the continuation cache.
+ *
+ * The cache is persisted to `responses-state.json`, so anything recorded here reaches disk.
+ * Encrypted-agent-task recovery decrypts task text into the request body and promises
+ * in-memory, TTL-bounded retention; recording that body would put the plaintext on disk with
+ * no TTL and break the promise.
+ *
+ * A WeakSet rather than a body field on purpose: `_rawBody` is serialized verbatim by the
+ * native passthrough, so any marker written into the body itself would be sent upstream.
+ * Marking is enforced once here rather than at each call site, because every recording path
+ * (streaming, non-streaming, passthrough, forced) funnels through `rememberResponseState` —
+ * a new call site cannot reintroduce the leak by forgetting a guard.
+ */
+const nonPersistableBodies = new WeakSet<object>();
+
+/** Bar this exact request body from the continuation cache, and therefore from disk. */
+export function markBodyNonPersistable(body: unknown): void {
+  if (body && typeof body === "object") nonPersistableBodies.add(body as object);
+}
+
+/** Test the body's in-memory persistence restriction without adding a wire marker. */
+export function isBodyNonPersistable(body: unknown): boolean {
+  return !!body && typeof body === "object" && nonPersistableBodies.has(body);
+}

src/server/index/websocket-handler.ts

@@ -1,3 +1,5 @@
+import { NativeSteeringChannel, NativeSteeringError } from "../responses/native-steering";
+import { createNativeSteeringLogObserver } from "../responses/native-steering-log";
 import type { Server, ServerWebSocket } from "bun";
 import {
   LIVE_SIDEBAND_UPSTREAM_OPEN_TIMEOUT_MS,
@@ -188,11 +190,39 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) {
         } catch {
           return; // text-only contract; ignore unparseable frames
         }
+        if (frame.type === "response.steer" || (frame.type === "response.create" && ws.data.nativeSteering)) {
+          try {
+            if (frame.type === "response.steer") {
+              if (!ws.data.nativeSteering) throw new NativeSteeringError("steering_not_supported", "Native steering is disabled or unavailable on this route.");
+              ws.data.nativeSteering.steer(frame);
+              return;
+            }
+            if (ws.data.nativeSteering?.continue(frame)) return;
+          } catch (error) {
+            sendJsonFrame(ws, buildWsErrorFrame(400, {
+              type: "invalid_request_error",
+              code: error instanceof NativeSteeringError ? error.code : "native_steering_error",
+              message: error instanceof NativeSteeringError ? error.message : "Native steering transport failed; delivery may be unknown. Do not automatically replay input.",
+            }));
+            return;
+          }
+        }
         if (frame.type === "response.processed") return; // ack — no-op
         if (frame.type !== "response.create") return;
         markActivity("ws response.create");
 
         ws.data.cancel?.();
+        // A superseded turn must not keep ownership during warmup or refusal.
+        ws.data.nativeSteering = undefined;
+        let nativeSteering: NativeSteeringChannel | undefined;
+        try {
+          const idleMs = typeof config.stallTimeoutSec === "number" && Number.isFinite(config.stallTimeoutSec)
+            ? Math.max(1, config.stallTimeoutSec) * 1000 : 300_000;
+          nativeSteering = config.codexNativeSteering === true ? new NativeSteeringChannel(frame, idleMs) : undefined;
+        } catch {
+          sendJsonFrame(ws, buildWsErrorFrame(400, { type: "invalid_request_error", message: "Invalid native steering request settings" }));
+          return;
+        }
         const turnId = (ws.data.turnId ?? 0) + 1;
         ws.data.turnId = turnId;
         const isCurrent = () => ws.data.turnId === turnId;
@@ -227,6 +257,8 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) {
           return;
         }
 
+        // Only a genuinely admitted turn may receive steering or continuations.
+        ws.data.nativeSteering = nativeSteering;
         const payload: Record<string, unknown> = { ...frame };
         delete payload.type;
         turnAdmissionLease.bindAbortController(turnAbort);
@@ -267,6 +299,7 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) {
               ...(wsAdmission ? { admission: wsAdmission } : {}),
               forceEmptyResponseId: true,
               inboundTransport: "websocket",
+              nativeSteering,
               abortSignal: turnAbort.signal,
               turnAdmissionLease,
               onFirstOutput: () => recordFirstOutput(logCtx, start),
@@ -277,7 +310,10 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) {
               },
             });
             await sendResponseToWebSocket(ws, response, isCurrent, {
-              onSsePayload: payload => inspectResponseLogSsePayload(logCtx, payload),
+              untilEof: nativeSteering?.relayActive === true,
+              onSsePayload: nativeSteering?.relayActive
+                ? createNativeSteeringLogObserver(logCtx, () => recordFirstOutput(logCtx, start))
+                : payload => inspectResponseLogSsePayload(logCtx, payload),
               onTerminal: status => {
                 terminalRecorder?.(status, logCtx.terminalHttpStatus);
                 finalizeLog(httpStatusForRequestLogTerminal(status, logCtx), {
@@ -313,6 +349,7 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) {
             }
           } finally {
             turnAdmissionLease.release();
+            if (ws.data.nativeSteering === nativeSteering) ws.data.nativeSteering = undefined;
             if (!logged && turnAbort.signal.aborted) finalizeLog(499);
             if (ws.data.cancel === cancelTurn) ws.data.cancel = undefined;
           }