fix(responses): bound steering confirmation waits and preserve sparse replay output

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

@@ -885,7 +885,7 @@ Do not rewrite an active paginated rollout or thread row to migrate those conver
 
 ## Experimental native mid-turn steering
 
-For a compatible native OpenAI model and a client that sends `response.steer`, enable both
+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:
 
 ```json
@@ -913,7 +913,7 @@ HTTP fallback, other providers, translated models, sidecars, Combo attempts and
 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;
+resubmit tools or steering text. Pending controls have fixed 90-second acknowledgement or successor deadlines;
 saved-tool-result waits have a 30-minute cap.
 
 The implementation has synthetic protocol and regression coverage, not live Astra/client
@@ -922,6 +922,40 @@ has been verified. Set `codexNativeSteering` to `false` and restart to restore t
 single-response relay; no account or conversation files need to be deleted.
 
 
+### Steering confirmation deadlines and retained context
+
+Each submitted steer has a fixed 90-second acknowledgement window. Other output
+and additional steers do not extend it. Once accepted, the input remains queued
+while the current response reaches a safe boundary; ordinary stream-idle checks
+still apply. After the response ends, the successor must begin within 90 seconds.
+A request for tool results or approval allows 30 minutes from the first such
+notification. Repeated notices do not renew this wait. Submitting saved results
+starts a new 90-second successor window, including local pacing/auth checks.
+Missing acknowledgements remain subject to their earlier individual deadlines.
+
+The owned connection itself has no absolute lifetime cap. Up to 128 responses may share it, and
+each may legitimately consume its own acknowledgement, successor, stream-idle and required-input
+waits, so the per-stage deadlines above compose to a worst case on the order of tens of hours.
+During that time the turn holds one physical socket and one pinned credential that cannot rotate,
+because the channel deliberately never re-enters account selection. Treat an enabled steering
+connection as a long-lived session resource rather than an ordinary bounded request.
+
+A timeout means **delivery is unknown**, not that the server rejected the input.
+Do not resend an accepted instruction or rerun a tool automatically. Inspect the
+actual task state before deciding how to resume. No account switch or paid API
+fallback is performed. Completed output already received on the wire is retained
+for local continuation history even when the terminal summary omits it. Conflicting
+item content or order causes an explicit failure rather than silent context loss.
+
+For a live comparison, use the same supported client version, model and account
+in isolated test conversations, once without the proxy and once with it enabled.
+Use a read-only task, steer while output is active, and compare acceptance and the
+successor's actual instruction adherence. Repeat while a synthetic tool result or
+approval is pending and after an explicit disconnect. Record only event types,
+relative times and redacted outcomes, not credentials or task bodies. Passing mock
+transport tests does not establish live client/backend support; no real-account
+smoke test is implied by these instructions.
+
 ## Experimental native function-result injection
 
 For a compatible client that sends OpenAI multi-agent `response.inject` messages,

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

@@ -575,8 +575,13 @@ wildcard `hostname`, where the public listener already holds `127.0.0.1:<port>`.
 
 
 The opt-in `codexNativeInjection` owner also accepts typed saved-result
-continuations after the response terminal: rich function/custom outputs and explicit
-MCP approval decisions remain on the original account/socket. This does not widen
-`response.inject` beyond string-valued function results. Multi-agent requests never
-acquire the single-agent steering owner merely because injection is disabled.
+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).
+
+
+`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.

scripts/test-layout/layout.json

@@ -1491,7 +1491,8 @@
     "request-log-nonstream.test.ts": "usage",
     "ws-native-result-continuations.test.ts": "responses",
     "ws-native-injection.test.ts": "responses",
-    "ws-native-steering.test.ts": "responses"
+    "ws-native-steering.test.ts": "responses",
+    "ws-steering-stability.test.ts": "responses"
   },
   "migrated": [
     "adapters",

src/server/index/websocket-handler.ts

@@ -218,9 +218,6 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) {
         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.nativeControl = undefined;
         let nativeControl: NativeResponseControl | undefined;
         try {
           const idleMs = typeof config.stallTimeoutSec === "number" && Number.isFinite(config.stallTimeoutSec)
@@ -232,6 +229,9 @@ export function createWebsocketHandler(ctx: ServeOptionsContext) {
           sendJsonFrame(ws, buildWsErrorFrame(400, { type: "invalid_request_error", message: "Invalid native steering request settings" }));
           return;
         }
+        ws.data.cancel?.();
+        // A superseded turn must not keep ownership during warmup or refusal.
+        ws.data.nativeControl = undefined;
         const turnId = (ws.data.turnId ?? 0) + 1;
         ws.data.turnId = turnId;
         const isCurrent = () => ws.data.turnId === turnId;

src/server/responses/codex-ws-exchange.ts

@@ -338,6 +338,9 @@ export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
       sent = true;
       try {
         if (nativeControl) {
+          // Parsed once: the base body is immutable for this exchange, and a
+          // full-replay frame runs to megabytes.
+          let base: Record<string, unknown> | undefined;
           detachSteering = nativeControl.attach(frame => {
             const sendControl = () => {
               if (terminal || signal?.aborted || session.closed || ws.readyState !== WebSocket.OPEN) {
@@ -349,7 +352,7 @@ export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
                 // The channel validates same settings/lane, saved results and user-only additions.
                 // Reuse the already-routed/authorized native settings; never feed a
                 // previous_response_id through the REST sanitizer or account selector.
-                const base = JSON.parse(frameText) as Record<string, unknown>;
+                base ??= JSON.parse(frameText) as Record<string, unknown>;
                 outgoing = { ...base, input: frame.input, previous_response_id: frame.previous_response_id };
               }
               const text = JSON.stringify(outgoing);

src/server/responses/native-injection-protocol.ts

@@ -1,4 +1,5 @@
-import { createHash } from "node:crypto";
+import { nativeResponseRecord as injectionRecord } from "./native-response-json";
+export { nativeResponseRecord as injectionRecord, nativeResponseFingerprint as injectionFingerprint } from "./native-response-json";
 import { CODEX_WS_ID_MAX_BYTES } from "./codex-ws-correlation";
 import { NativeSteeringError } from "./native-steering";
 
@@ -10,22 +11,11 @@ export const MAX_NATIVE_INJECTION_CALLS = 1024;
 export const NATIVE_INJECTION_ACK_MS = 90_000;
 export const NATIVE_INJECTION_TOOL_MS = 30 * 60_000;
 
-/** Narrow a JSON object without accepting arrays or null. */
-export function injectionRecord(value: unknown): value is InjectionFrame {
-  return value !== null && typeof value === "object" && !Array.isArray(value);
-}
 /** Bound identities and exclude control characters, without changing their spelling. */
 export function injectionId(value: unknown): value is string {
   return typeof value === "string" && value.length > 0 && Buffer.byteLength(value) <= CODEX_WS_ID_MAX_BYTES
     && !/[\u0000-\u001f\u007f]/.test(value);
 }
-/** Stable setting comparison; only the digest is retained by the connection owner. */
-export function injectionFingerprint(value: unknown): string {
-  const canonical = (item: unknown): string => Array.isArray(item) ? `[${item.map(canonical).join(",")}]`
-    : injectionRecord(item) ? `{${Object.keys(item).sort().map(key => `${JSON.stringify(key)}:${canonical(item[key])}`).join(",")}}`
-    : JSON.stringify(item) ?? "null";
-  return createHash("sha256").update(canonical(value)).digest("hex");
-}
 /** Throw only fixed, content-free errors, never tool output or caller identifiers. */
 export function injectionError(code: string, message: string): never {
   throw new NativeSteeringError(code, message);

src/server/responses/native-response-json.ts

@@ -0,0 +1,14 @@
+import { createHash } from "node:crypto";
+
+/** Narrow JSON object envelopes independently of either native control owner. */
+export function nativeResponseRecord(value: unknown): value is Record<string, unknown> {
+  return value !== null && typeof value === "object" && !Array.isArray(value);
+}
+
+/** Compare JSON content by value: object-key order is irrelevant, array order is not. */
+export function nativeResponseFingerprint(value: unknown): string {
+  const canonical = (item: unknown): string => Array.isArray(item) ? `[${item.map(canonical).join(",")}]`
+    : nativeResponseRecord(item) ? `{${Object.keys(item).sort().map(key => `${JSON.stringify(key)}:${canonical(item[key])}`).join(",")}}`
+    : JSON.stringify(item) ?? "null";
+  return createHash("sha256").update(canonical(value)).digest("hex");
+}

src/server/responses/native-response-output.ts

@@ -1,4 +1,5 @@
-import { injectionFingerprint, injectionRecord as record, type InjectionFrame as Frame } from "./native-injection-protocol";
+import { nativeResponseFingerprint as injectionFingerprint, nativeResponseRecord as record } from "./native-response-json";
+type Frame = Record<string, unknown>;
 
 /**
  * Preserve completed wire items missing from a sparse terminal, including hosted

src/server/responses/native-steering-replay.ts

@@ -1,3 +1,5 @@
+import { nativeResponseOutput } from "./native-response-output";
+
 /**
  * Connection-local replay journal. Only input committed by response.created enters
  * a successor's prefix. Uncommitted/rejected steering never enters the shared
@@ -26,7 +28,7 @@ export class NativeSteeringReplay implements NativeSteeringReplayObserver {
   private bytes: number;
   private current?: string;
   private previousOutput: unknown[] = [];
-  private outputItems = new Map<number, unknown>();
+  private outputItems = new Map<number, Frame>();
   private submissions: Array<{ parent: string; input: unknown[]; id?: string; bytes: number }> = [];
   private explicitInput: unknown[] = [];
   private explicitBytes = 0;
@@ -91,17 +93,17 @@ export class NativeSteeringReplay implements NativeSteeringReplayObserver {
       this.previousOutput = [];
       this.outputItems.clear();
       this.current = String(response?.id);
-    } else if (frame.type === "response.output_item.done" && Number.isSafeInteger(frame.output_index)) {
+    } else if (frame.type === "response.output_item.done") {
       const index = frame.output_index as number;
-      if (index < 0 || index > 10_000 || !record(frame.item)) throw new Error("Native steering replay output identity is invalid");
+      if (!Number.isSafeInteger(index) || index < 0 || index > 10_000 || !record(frame.item)) throw new Error("Native steering replay output identity is invalid");
       const previous = this.outputItems.get(index);
       if (previous !== undefined) this.bytes -= Buffer.byteLength(JSON.stringify(previous));
       this.bytes += Buffer.byteLength(JSON.stringify(frame.item));
       this.check();
       this.outputItems.set(index, frame.item);
     } else if (response && ["response.completed", "response.incomplete", "response.failed"].includes(String(frame.type))) {
       const doneItems = [...this.outputItems.entries()].sort((a, b) => a[0] - b[0]).map(([, item]) => item);
-      const output = Array.isArray(response.output) && response.output.length ? response.output : doneItems;
+      const output = nativeResponseOutput(this.outputItems, response.output);
       for (const item of doneItems) this.bytes -= Buffer.byteLength(JSON.stringify(item));
       this.bytes += Buffer.byteLength(JSON.stringify(output));
       this.check();