fix(responses): combine native steering route restriction with memory and body limits

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

@@ -927,8 +927,7 @@ 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 or an explicitly configured
-[OpenAI API WebSocket route](#steering-continuation-settings-and-public-api), and a client
+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:
 
@@ -953,7 +952,7 @@ Do not rerun tools or resend accepted steering text. Model, account, tool declar
 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, noncanonical gateways, translated models, sidecars, Combo attempts and plaintext V2
+HTTP fallback, noncanonical gateways, public API-key routes, 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
@@ -984,6 +983,16 @@ During that time the turn holds one physical socket and one pinned credential th
 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.
 
+Steering frames also share the proxy's configured body and memory limits. A control frame above
+[`maxInboundBodyBytes`](/reference/inbound-body-admission/) is refused before
+it is parsed on an established control connection (an initial frame is still parsed before
+its type-based limit applies), and the reconstructed body sent upstream is refused when it exceeds
+[`maxUpstreamBodyBytes`](/reference/configuration/providers/). Each
+connection's replay journal is capped at 32 MiB and counted as pinned state against
+[`appOwnedMemoryBudgetMb`](/reference/configuration/server/); admitting a
+journal demotes evictable caches first rather than failing, and the aggregate across live journals
+is capped at 128 MiB regardless of the configured budget.
+
 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
@@ -1112,12 +1121,11 @@ 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
+## Steering continuation settings
 
 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
+`stream_options`. Subscription routes refuse a `max_output_tokens` override
 instead of silently ignoring it. Normal provider pins, subagent caps, effort
 mapping and summary/verbosity capability exclusions still apply.
 
@@ -1129,14 +1137,11 @@ 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.
+Native steering is restricted to the canonical ChatGPT subscription route. Public
+API-key and gateway routes are not steerable; their ordinary responses are preserved
+and a steering attempt receives an explanatory error. This prevents successor
+generations on a retained socket from bypassing normal per-request admission. The
+separately gated public API multi-agent injection path remains available.
 
 ### Executable direct-versus-proxy wire probe
 

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

@@ -714,6 +714,6 @@ wildcard `hostname`, where the public listener already holds `127.0.0.1:<port>`.
 
 `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),
+[supported steering routes and settings](../../guides/codex-integration.md#steering-continuation-settings),
 [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).

src/lib/app-owned-memory-stores.ts

@@ -54,6 +54,7 @@ import { translatorObservedBufferSnapshot } from "./translator-budget";
 import { imageFulfillmentTailSnapshot } from "../images/fulfill";
 import { oauthMutationTailSnapshot } from "../oauth/store";
 import { grokApplyFlightSnapshot } from "../server/management/agent-settings-routes";
+import { nativeControlReplayRetainedStoreSnapshot } from "../server/responses/native-steering-replay";
 
 function ringSnapshot(metrics: { entries: number; bytes: number; oldestAt: number | null }): RetainedStoreSnapshot {
   return {
@@ -187,6 +188,12 @@ export const APP_OWNED_RETAINED_STORE_REGISTRATIONS = [
     snapshot: responseContinuationRetainedStoreSnapshot,
     evictOldest: evictOldestResponseContinuationForBudget,
   },
+  {
+    id: "native_control_replay",
+    category: "continuation",
+    snapshot: nativeControlReplayRetainedStoreSnapshot,
+    evictOldest: () => 0,
+  },
 ] as const satisfies readonly RetainedStoreRegistration[];
 
 export function registerDefaultAppOwnedMemoryStores(): void {

src/lib/app-owned-memory.ts

@@ -209,14 +209,14 @@ function warnPinnedSaturation(): void {
   console.warn("[app-owned-memory] retained state remains over budget with no evictable candidate");
 }
 
-export function enforceAppOwnedMemoryBudget(): AppOwnedBytesSnapshot {
+export function enforceAppOwnedMemoryBudget(reservedPinnedBytes = 0): AppOwnedBytesSnapshot {
   if (isEnforcing) return appOwnedBytesSnapshot();
   isEnforcing = true;
   enforcementCounters.runs += 1;
   try {
     const ineligible = new Set<string>();
     const current = retainedSnapshots();
-    while (current.total > budgetBytes) {
+    while (current.total + reservedPinnedBytes > budgetBytes) {
       const candidate = nextCandidate(current.stores, ineligible);
       if (!candidate) {
         enforcementCounters.noEvictableCandidate += 1;
@@ -250,6 +250,17 @@ export function enforceAppOwnedMemoryBudget(): AppOwnedBytesSnapshot {
   }
 }
 
+/**
+ * Admission for a pinned allocation the owner cannot demote later. The proposal is
+ * counted against the shared target BEFORE the normal eviction pass runs, so
+ * reclaimable logs, caches, blobs and continuations are demoted first and only a
+ * projected total still above budget — pinned state that cannot fit — is refused.
+ */
+export function admitAppOwnedPinnedBytes(proposedPinnedBytes: number): boolean {
+  const snapshot = enforceAppOwnedMemoryBudget(Math.max(0, proposedPinnedBytes));
+  return snapshot.retainedBytes + Math.max(0, proposedPinnedBytes) <= snapshot.budgetBytes;
+}
+
 export function resetAppOwnedMemoryForTests(): void {
   retainedStores.clear();
   observedBuffers.clear();

src/server/index/websocket-handler.ts

@@ -86,6 +86,7 @@ import {
 } from "../live";
 import type { ServeOptionsContext } from "./serve-options";
 import type { RequestMetricsRecorder } from "../request-metrics";
+import { resolveInboundBodyLimitBytes } from "../request-decompress";
 
 /**
  * The WebSocket half of the Bun.serve options, split out of serve-options.ts to keep that file
@@ -190,12 +191,31 @@ export function createWebsocketHandler(
           ws.close(1009, "message too large");
           return;
         }
+        // An established control connection only ever carries control frames, so the
+        // inbound body limit applies to raw bytes before the parse materializes them.
+        if (ws.data.nativeControl && rawBytes > resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)) {
+          sendJsonFrame(ws, buildWsErrorFrame(413, {
+            type: "invalid_request_error",
+            code: "inbound_body_too_large",
+            message: "Native response control frame exceeds the configured inbound body limit.",
+          }));
+          return;
+        }
         let frame: Record<string, unknown>;
         try {
           frame = JSON.parse(typeof raw === "string" ? raw : raw.toString()) as Record<string, unknown>;
         } catch {
           return; // text-only contract; ignore unparseable frames
         }
+        if ((frame.type === "response.inject" || frame.type === "response.steer")
+          && rawBytes > resolveInboundBodyLimitBytes(config.maxInboundBodyBytes)) {
+          sendJsonFrame(ws, buildWsErrorFrame(413, {
+            type: "invalid_request_error",
+            code: "inbound_body_too_large",
+            message: "Native response control frame exceeds the configured inbound body limit.",
+          }));
+          return;
+        }
         if (frame.type === "response.inject" || frame.type === "response.steer" || (frame.type === "response.create" && ws.data.nativeControl)) {
           try {
             if (frame.type === "response.inject") {
@@ -227,8 +247,8 @@ export function createWebsocketHandler(
           const idleMs = typeof config.stallTimeoutSec === "number" && Number.isFinite(config.stallTimeoutSec)
             ? Math.max(1, config.stallTimeoutSec) * 1000 : 300_000;
           const mode = nativeResponseControlMode(frame, config);
-          nativeControl = mode === "injection" ? new NativeInjectionChannel(frame, idleMs)
-            : mode === "steering" ? new NativeSteeringChannel(frame, idleMs) : undefined;
+          nativeControl = mode === "injection" ? new NativeInjectionChannel(frame, idleMs, config.maxUpstreamBodyBytes)
+            : mode === "steering" ? new NativeSteeringChannel(frame, idleMs, config.maxUpstreamBodyBytes) : undefined;
         } catch {
           sendJsonFrame(ws, buildWsErrorFrame(400, { type: "invalid_request_error", message: "Invalid native steering request settings" }));
           return;

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

@@ -1,3 +1,4 @@
+import { NativeSteeringError } from "./native-steering";
 import { mergeSteeringContinuation } from "./native-steering-settings";
 import { markNativeControlResponse } from "./native-response-control";
 import type { NativeResponseControl } from "./native-response-control";
@@ -345,11 +346,11 @@ export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
           // the first create frame.
           let base: Record<string, unknown> | undefined;
           detachSteering = nativeControl.attach(frame => {
-            const sendControl = () => {
-              if (terminal || signal?.aborted || session.closed || ws.readyState !== WebSocket.OPEN) {
-                throw new Error("Native steering connection is no longer available");
-              }
-              beforeDispatch?.(new Headers(headers));
+            // Build, serialize and bound-check the reconstructed frame synchronously in the
+            // channel callback: a typed refusal must reach the channel's synchronous rollback
+            // (continuation slot released, journal unwritten) rather than the asynchronous
+            // failStream path, so a corrected continuation can still retry on this channel.
+            const prepare = () => {
               let outgoing = frame;
               if (frame.type === "response.create") {
                 // Generation overrides have passed route policy; identity/tools remain pinned.
@@ -361,11 +362,20 @@ export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
                   : { ...continuationBase, input: frame.input, previous_response_id: frame.previous_response_id };
               }
               const text = JSON.stringify(outgoing);
+              nativeControl.assertOutboundFrame?.(text);
               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 {
+              return { outgoing, text };
+            };
+            const prepared = prepare();
+            const sendControl = () => {
+              if (terminal || signal?.aborted || session.closed || ws.readyState !== WebSocket.OPEN) {
+                throw new Error("Native steering connection is no longer available");
+              }
+              beforeDispatch?.(new Headers(headers));
+              if (frame.type === "response.create") continuationBase = prepared.outgoing;
+              try { ws.send(prepared.text); } catch {
                 // A send failure has unknown delivery. Never replay or fall back.
                 failStream("Native steering send failed; delivery is unknown");
                 throw new Error("Native steering send failed; delivery is unknown");
@@ -374,7 +384,13 @@ export function codexWsExchange(options: ExchangeOptions): Promise<Response> {
             if (frame.type === "response.create" && beforeContinuation) {
               // Explicit tool-result continuations are physical request starts;
               // they keep provider pacing and revalidate auth AFTER the wait.
-              void beforeContinuation().then(sendControl).catch(() => failStream("Native steering continuation could not be dispatched; do not automatically replay queued input"));
+              // A typed pre-send refusal (e.g. the configured upstream body limit) is a
+              // known non-delivery and keeps its own message; every other failure keeps
+              // the unknown-delivery wording.
+              void beforeContinuation().then(sendControl).catch(error => failStream(
+                error instanceof NativeSteeringError ? error
+                  : new Error("Native steering continuation could not be dispatched; do not automatically replay queued input"),
+              ));
             } else sendControl();
           }, error => failStream(error));
         }

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

@@ -1,4 +1,4 @@
-import { MAX_NATIVE_STEERING_REPLAY_BYTES, type NativeSteeringReplayObserver } from "./native-steering-replay";
+import { MAX_NATIVE_STEERING_REPLAY_BYTES, admitNativeControlReplayJournal, registerNativeControlReplayJournal, type NativeSteeringReplayObserver } from "./native-steering-replay";
 import { injectionRecord as record, type InjectionFrame as Frame, type FunctionResult } from "./native-injection-protocol";
 
 import { nativeResultFingerprint } from "./native-tool-results";
@@ -16,19 +16,29 @@ export class NativeInjectionReplay implements NativeSteeringReplayObserver {
   private acceptedBatchBytes: number[] = [];
   private explicit: unknown[] = [];
   private previous: unknown[] = [];
+  private unregisterAccounting?: () => void;
+
+  get retainedBytes(): number { return this.bytes; }
 
   /** Capture a private initial prefix; existing persistence eligibility is checked by the caller. */
   constructor(input: unknown, private readonly remember: (input: unknown[], response: Frame) => void) {
     this.prefix = typeof input === "string"
       ? [{ type: "message", role: "user", content: [{ type: "input_text", text: input }] }]
       : Array.isArray(input) ? [...input] : [];
     this.reserve(this.prefix);
+    this.unregisterAccounting = registerNativeControlReplayJournal(this);
   }
   /** Charge serialized bytes, refusing rather than truncating an over-budget transcript. */
   private reserve(value: unknown): number {
     const bytes = Buffer.byteLength(JSON.stringify(value));
     if (this.bytes + bytes > MAX_NATIVE_STEERING_REPLAY_BYTES) throw new Error("Native injection replay exceeded its history budget.");
     this.bytes += bytes;
+    try {
+      admitNativeControlReplayJournal(this);
+    } catch (error) {
+      this.bytes -= bytes;
+      throw error;
+    }
     return bytes;
   }
   /** Journal before physical send, with rollback usable only for a known unsent frame. */
@@ -99,6 +109,8 @@ export class NativeInjectionReplay implements NativeSteeringReplayObserver {
   }
   /** Drop all retained bodies at cancellation, connection teardown or unknown delivery. */
   dispose(): void {
+    this.unregisterAccounting?.();
+    this.unregisterAccounting = undefined;
     this.prefix = []; this.output.clear(); this.accepted.clear(); this.pending = undefined; this.pendingBytes = 0; this.acceptedBatchBytes = [];
     this.explicit = []; this.previous = []; this.bytes = 0;
   }