From dbb35da2132d0cf9754032cdcaa167fcaa7030d0 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Thu, 17 Sep 2026 05:13:13 +0000 Subject: [PATCH 1/5] fix(responses): bound steering confirmation waits and preserve sparse replay output Separate monotonic acknowledgement, successor and tool deadlines; reconcile steering replay with completed wire items without weakening ownership or retry guards. Follow up on #4861. --- .../content/docs/guides/codex-integration.md | 34 +++ .../docs/reference/configuration/server.md | 7 + scripts/test-layout/layout.json | 3 +- .../responses/native-injection-protocol.ts | 14 +- src/server/responses/native-response-json.ts | 14 + .../responses/native-response-output.ts | 3 +- .../responses/native-steering-replay.ts | 6 +- src/server/responses/native-steering.ts | 123 ++++++-- 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/service-and-sidecars.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 | 38 +++ tests/fixtures/test-layout-expected.json | 3 +- tests/helpers/responses-core-source.ts | 1 + tests/responses/ws-steering-stability.test.ts | 272 ++++++++++++++++++ 25 files changed, 503 insertions(+), 41 deletions(-) create mode 100644 src/server/responses/native-response-json.ts create mode 100644 tests/responses/ws-steering-stability.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 d2c9a25fe2..a9db588d90 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/docs-site/src/content/docs/guides/codex-integration.md @@ -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, diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index bf794541fe..6bb048efaa 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -580,3 +580,10 @@ MCP approval decisions remain on the original account/socket. This does not wide `response.inject` beyond string-valued function results. 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). + + +Native steering confirmation uses per-submission and per-phase absolute deadlines, +not a timeout renewed by unrelated output. Its local continuation history reconciles +sparse terminal summaries with completed output already received. 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. diff --git a/scripts/test-layout/layout.json b/scripts/test-layout/layout.json index f15e51edd6..d4a52ed85c 100644 --- a/scripts/test-layout/layout.json +++ b/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", diff --git a/src/server/responses/native-injection-protocol.ts b/src/server/responses/native-injection-protocol.ts index 6c88e37646..26aeebbde6 100644 --- a/src/server/responses/native-injection-protocol.ts +++ b/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); diff --git a/src/server/responses/native-response-json.ts b/src/server/responses/native-response-json.ts new file mode 100644 index 0000000000..1692c6b25d --- /dev/null +++ b/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 { + 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"); +} diff --git a/src/server/responses/native-response-output.ts b/src/server/responses/native-response-output.ts index 6115c76913..04cce41e54 100644 --- a/src/server/responses/native-response-output.ts +++ b/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; /** * Preserve completed wire items missing from a sparse terminal, including hosted diff --git a/src/server/responses/native-steering-replay.ts b/src/server/responses/native-steering-replay.ts index f300c55e64..fa7950922e 100644 --- a/src/server/responses/native-steering-replay.ts +++ b/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(); + private outputItems = new Map(); private submissions: Array<{ parent: string; input: unknown[]; id?: string; bytes: number }> = []; private explicitInput: unknown[] = []; private explicitBytes = 0; @@ -101,7 +103,7 @@ export class NativeSteeringReplay implements NativeSteeringReplayObserver { 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(); diff --git a/src/server/responses/native-steering.ts b/src/server/responses/native-steering.ts index d5979ffa8e..61bf13a871 100644 --- a/src/server/responses/native-steering.ts +++ b/src/server/responses/native-steering.ts @@ -54,7 +54,13 @@ export function validateSteeringFrame(frame: Frame): void { } } -type Parent = { unacknowledged: number; accepted: Set; ended: boolean }; +type Parent = { + unacknowledged: Array<{ deadline: number }>; + accepted: Set; + ended: boolean; + successorDeadline?: number; + toolDeadline?: number; +}; /** Only client-owned results can use the early-continuation path. */ function outputRequirement(item: unknown): Frame | undefined { @@ -97,6 +103,9 @@ export class NativeSteeringChannel { private send?: Send; private onFailure?: (error: Error) => void; private timer?: ReturnType; + private timerDeadline?: number; + private idleDeadline?: number; + private continuationDeadline?: number; private readonly parents = new Map(); private readonly settings = new Map(); private currentId?: string; @@ -126,7 +135,7 @@ export class NativeSteeringChannel { get ended(): boolean { return this.finished; } /** Count unacknowledged or accepted submissions that still own the response chain. */ get hasOutstanding(): boolean { - return [...this.parents.values()].some(parent => parent.unacknowledged > 0 || parent.accepted.size > 0); + return [...this.parents.values()].some(parent => parent.unacknowledged.length > 0 || parent.accepted.size > 0); } /** Report whether the server has requested a saved-result continuation. */ get awaitingContinuation(): boolean { return this.pendingParent !== undefined; } @@ -143,8 +152,9 @@ export class NativeSteeringChannel { if (this.send !== send) return; this.send = undefined; this.onFailure = undefined; - clearTimeout(this.timer); - this.timer = undefined; + this.clearTimer(); + this.idleDeadline = undefined; + this.continuationDeadline = undefined; this.correlation?.finish(); this.correlation = undefined; this.parents.clear(); @@ -156,12 +166,63 @@ export class NativeSteeringChannel { }; } - /** Replace the unrefed watchdog; timeout reports uncertainty instead of replaying. */ - private wait(ms = NATIVE_STEERING_WAIT_MS): void { + /** Drop the physical timer without altering any protocol-stage deadline. */ + private clearTimer(): void { clearTimeout(this.timer); + this.timer = undefined; + this.timerDeadline = undefined; + } + + /** Earliest absolute control deadline wins, independently of ordinary stream activity. */ + private nextDeadline(): number | undefined { + let deadline: number | undefined; + const include = (value: number | undefined) => { + if (value !== undefined) deadline = deadline === undefined ? value : Math.min(deadline, value); + }; + for (const [id, parent] of this.parents) { + include(parent.unacknowledged[0]?.deadline); + if (parent.ended && (parent.accepted.size || parent.unacknowledged.length)) { + if (id === this.currentId && this.continuationSent) continue; + include(this.pendingParent === id ? parent.toolDeadline : parent.successorDeadline); + } + } + if (this.continuationSent) include(this.continuationDeadline); + if (this.currentId && !this.parents.get(this.currentId)?.ended) include(this.idleDeadline); + return deadline; + } + + /** Settle once as unknown delivery; expiry never retries or invents a server rejection. */ + private expire(): Error { + const error = new Error("Native steering continuation timed out; queued-input delivery is unknown. Do not automatically replay tools or steering input."); + if (!this.finished) { + this.finished = true; + this.clearTimer(); + this.replay?.dispose(); + this.onFailure?.(error); + } + return error; + } + + /** Late wire activity must not win a race with an expired but not-yet-fired timer. */ + private assertTimely(): void { + const deadline = this.nextDeadline(); + if (this.finished || (deadline !== undefined && performance.now() >= deadline)) throw this.expire(); + } + + /** Arm one unrefed timer for the existing deadline, never for now plus a fresh wait. */ + private armTimer(): void { + const deadline = this.finished || !this.send ? undefined : this.nextDeadline(); + if (deadline === this.timerDeadline) return; + this.clearTimer(); + if (deadline === undefined) return; + this.timerDeadline = deadline; this.timer = setTimeout(() => { - this.onFailure?.(new Error("Native steering continuation timed out; queued-input delivery is unknown. Do not automatically replay tools or steering input.")); - }, ms); + this.clearTimer(); + if (this.finished || !this.send) return; + const next = this.nextDeadline(); + if (next !== undefined && performance.now() >= next) this.expire(); + else this.armTimer(); // Early callbacks cannot shorten the monotonic bound. + }, Math.max(0, deadline - performance.now())); this.timer.unref?.(); } /** Check the live owner and byte limit before journaling and sending one control. */ @@ -176,19 +237,22 @@ export class NativeSteeringChannel { steer(frame: Frame): void { validateSteeringFrame(frame); if (!this.send || this.finished) { this.liveSend(frame); return; } + this.assertTimely(); const target = this.parents.get(frame.previous_response_id as string); if (!target || frame.previous_response_id !== this.currentId || target.ended) { throw new NativeSteeringError("response_not_active", "The target response is not active on this connection."); } - if ([...this.parents.values()].reduce((n, p) => n + p.unacknowledged + p.accepted.size, 0) >= MAX_NATIVE_STEERS) { + if ([...this.parents.values()].reduce((n, p) => n + p.unacknowledged.length + p.accepted.size, 0) >= MAX_NATIVE_STEERS) { throw new NativeSteeringError("too_many_pending_steers", "The native steering pending-submission limit was reached."); } // Count before send: fake transports, and some runtimes, deliver synchronously. - target.unacknowledged += 1; - this.wait(); + const submission = { deadline: performance.now() + NATIVE_STEERING_WAIT_MS }; + target.unacknowledged.push(submission); + this.armTimer(); try { this.liveSend(frame); } catch (error) { - target.unacknowledged -= 1; - if (!this.hasOutstanding) { clearTimeout(this.timer); this.timer = undefined; } + const index = target.unacknowledged.indexOf(submission); + if (index >= 0) target.unacknowledged.splice(index, 1); + this.armTimer(); throw error; } } @@ -210,6 +274,7 @@ export class NativeSteeringChannel { /** Returns false only when an ordinary create may use normal dispatch. */ continue(frame: Frame): boolean { if (this.finished || !this.send) return false; + this.assertTimely(); const parent = this.currentId ? this.parents.get(this.currentId) : undefined; if (!parent?.ended || !this.currentId || frame.previous_response_id !== this.currentId) { if (this.hasOutstanding || this.continuationSent) throw new NativeSteeringError("steering_continuation_required", "Queued steering owns this connection; wait for the successor or send the required-input continuation, or explicitly stop the turn."); @@ -241,13 +306,20 @@ export class NativeSteeringChannel { } if (used.size !== required.length) throw new NativeSteeringError("invalid_input", "Every required tool output or approval must be supplied exactly once."); this.continuationSent = true; - this.wait(); - try { this.liveSend(frame); } catch (error) { this.continuationSent = false; throw error; } + this.continuationDeadline = performance.now() + NATIVE_STEERING_WAIT_MS; + this.armTimer(); + try { this.liveSend(frame); } catch (error) { + this.continuationSent = false; + this.continuationDeadline = undefined; + this.armTimer(); + throw error; + } return true; } /** Called on the ordered upstream wire, BEFORE the event is published to SSE. */ observe(event: Frame): boolean { + this.assertTimely(); const type = event.type; if (!(type === "error" && event.stream_id == null) && (event.stream_id ?? undefined) !== (this.lane ?? undefined)) throw new Error("native steering WebSocket lane mismatch"); const response = record(event.response) ? event.response : undefined; @@ -261,12 +333,13 @@ export class NativeSteeringChannel { parent.accepted.clear(); // response.created, not accepted, is the commit point. } this.currentId = id; - this.parents.set(id, { unacknowledged: 0, accepted: new Set(), ended: false }); + this.parents.set(id, { unacknowledged: [], accepted: new Set(), ended: false }); this.pendingParent = undefined; this.required = []; this.advertised.clear(); this.advertisedBytes = 0; this.continuationSent = false; + this.continuationDeadline = undefined; this.correlation?.finish(); this.correlation = new CodexWsCorrelation(true, () => false); } @@ -275,15 +348,15 @@ export class NativeSteeringChannel { const parent = typeof steer?.previous_response_id === "string" ? this.parents.get(steer.previous_response_id) : undefined; if (!parent) throw new Error("native steering acknowledgement has an unknown parent"); if (type === "response.steer.accepted") { - if (!validId(steer?.id) || parent.unacknowledged < 1 || parent.accepted.has(steer.id)) throw new Error("unexpected native steering acceptance"); - parent.unacknowledged -= 1; + if (!validId(steer?.id) || parent.unacknowledged.length < 1 || parent.accepted.has(steer.id)) throw new Error("unexpected native steering acceptance"); + parent.unacknowledged.shift(); parent.accepted.add(steer.id); } else if (type === "response.steer.failed") { if (steer?.id !== undefined) { if (!validId(steer.id) || !parent.accepted.delete(steer.id)) throw new Error("unexpected native steering failure"); } else { - if (parent.unacknowledged < 1) throw new Error("unexpected native steering rejection"); - parent.unacknowledged -= 1; + if (parent.unacknowledged.length < 1) throw new Error("unexpected native steering rejection"); + parent.unacknowledged.shift(); } } else if (type === "response.steer.pending") { if (!validId(steer?.id) || !parent.accepted.has(steer.id) || !parent.ended @@ -293,6 +366,7 @@ export class NativeSteeringChannel { || event.required_input.some(item => !record(item) || typeof item.type !== "string" || item.type === "message") || Buffer.byteLength(JSON.stringify(event.required_input)) > 256 * 1024) throw new Error("native steering required-input budget or schema violated"); if (this.pendingParent && stable(this.required) !== stable(event.required_input)) throw new Error("native steering required-input stubs changed"); + parent.toolDeadline ??= performance.now() + NATIVE_STEERING_TOOL_WAIT_MS; this.pendingParent = this.currentId; this.required = event.required_input as Frame[]; } @@ -303,15 +377,16 @@ export class NativeSteeringChannel { if (type === "response.output_item.done") this.advertise(event.item); if (type === "response.completed" || type === "response.failed" || type === "response.incomplete") { if (!this.currentId || response?.id !== this.currentId) throw new Error("native steering terminal identity mismatch"); - this.parents.get(this.currentId)!.ended = true; + const parent = this.parents.get(this.currentId)!; + parent.ended = true; + parent.successorDeadline ??= performance.now() + NATIVE_STEERING_WAIT_MS; if (Array.isArray(response.output)) for (const item of response.output) this.advertise(item); } } if (type === "error") this.finished = true; else this.finished = this.currentId !== undefined && this.parents.get(this.currentId)!.ended && !this.hasOutstanding && !this.continuationSent; - if (this.finished) { clearTimeout(this.timer); this.timer = undefined; } - else if (this.hasOutstanding || this.continuationSent) this.wait(this.pendingParent && !this.continuationSent ? NATIVE_STEERING_TOOL_WAIT_MS : NATIVE_STEERING_WAIT_MS); - else if (this.currentId) this.wait(this.idleMs); + if (this.currentId && !this.parents.get(this.currentId)!.ended) this.idleDeadline = performance.now() + this.idleMs; + this.armTimer(); this.replay?.observe(event); return this.finished; } diff --git a/structure/adapters/registry.md b/structure/adapters/registry.md index 27a4de9046..8645581cde 100644 --- a/structure/adapters/registry.md +++ b/structure/adapters/registry.md @@ -189,3 +189,5 @@ translation boundary and verifies that rejection sends no upstream request. Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. 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). diff --git a/structure/catalog.md b/structure/catalog.md index 760dcfaf4e..2d0c1f6d92 100644 --- a/structure/catalog.md +++ b/structure/catalog.md @@ -440,3 +440,5 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara `src/providers/derive.ts` fills missing reasoning tables for renamed providers accepted by the existing fixed-key destination matcher. Model entries are cloned and explicit user entries (including empty arrays) win. Provider-wide effort defaults fill only when undefined; Command Code unknown models therefore keep the registry's empty picker policy unless overridden. Identity, transport and other capability axes are unchanged. The gathered row drives client exports; this metadata contract does not prove arbitrary gateway routing. 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). diff --git a/structure/clients/claude-desktop.md b/structure/clients/claude-desktop.md index faf83ef09e..6ccee591de 100644 --- a/structure/clients/claude-desktop.md +++ b/structure/clients/claude-desktop.md @@ -179,3 +179,5 @@ Exact [model input declarations](../config.md#explicit-per-model-capability-decl 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 retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](../transports/streaming-health.md#steering-deadlines-and-replay-completeness). diff --git a/structure/data-planes/images.md b/structure/data-planes/images.md index a5db402c7f..9b80866978 100644 --- a/structure/data-planes/images.md +++ b/structure/data-planes/images.md @@ -127,3 +127,5 @@ The [explicit model-capability contract](../config.md#explicit-per-model-capabil Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. 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). diff --git a/structure/data-planes/inbound-compat.md b/structure/data-planes/inbound-compat.md index f5e00c1468..34cd677849 100644 --- a/structure/data-planes/inbound-compat.md +++ b/structure/data-planes/inbound-compat.md @@ -337,3 +337,5 @@ admission follows the [registry contract](../adapters/registry.md#untranslated-i Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. 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). diff --git a/structure/gui-and-management-api.md b/structure/gui-and-management-api.md index 09264658ba..44eafc575b 100644 --- a/structure/gui-and-management-api.md +++ b/structure/gui-and-management-api.md @@ -658,3 +658,5 @@ Exact [model input declarations](config.md#explicit-per-model-capability-declara The raw provider editor round-trips `autoReviewModel` and `autoReviewModelOverrides` through editor-owned DTO fields. POST/PATCH/PUT share validation; PUT copies schema-normalized values into the persisted and live candidate before adoption. Canonical `openai` rejects these fields, including clear forms. Field-masked writes (PATCH, editor PUT, reload) pin every registry-seed key and ignore operator overlays the seed never defines, most commonly `selectedModels`; POST keeps the exact-key comparison. Canonical `openai` still rejects `allowPrivateNetwork`, which must not short-circuit destination DNS checks on the ChatGPT forward row. Existing authentication, origin checks and stale-baseline protection still govern the writes. See [reviewer projection](catalog.md#provider-scoped-approval-reviewer). 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). diff --git a/structure/ops/service-and-sidecars.md b/structure/ops/service-and-sidecars.md index e3a7ad19e8..c8b30cfb5c 100644 --- a/structure/ops/service-and-sidecars.md +++ b/structure/ops/service-and-sidecars.md @@ -188,3 +188,5 @@ The [explicit model-capability contract](../config.md#explicit-per-model-capabil Provider-scoped approval reviewer settings are projected by the [catalog owner](../catalog.md#provider-scoped-approval-reviewer); this surface retains its existing routing, transport and account-selection behavior. 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). diff --git a/structure/providers/xai-grok.md b/structure/providers/xai-grok.md index 864de99d30..a4050bfa33 100644 --- a/structure/providers/xai-grok.md +++ b/structure/providers/xai-grok.md @@ -155,3 +155,5 @@ Live sideband admission and its bounded upstream handshake follow the [runtime c Translated audio/file admission follows the [final-adapter input contract](../adapters/registry.md#untranslated-input-media); native raw passthrough remains separate. 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). diff --git a/structure/runtime.md b/structure/runtime.md index 1b0dbc06b5..5961fb2444 100644 --- a/structure/runtime.md +++ b/structure/runtime.md @@ -531,3 +531,5 @@ stamps the configured key selected for the physical request. `src/server/request retains per-key attempt usage, and `src/usage/log.ts` validates and persists labels. The [account attribution contract](gui-and-management-api.md#upstream-key-account-attribution) 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). diff --git a/structure/subagents.md b/structure/subagents.md index a68e4f06d0..49d53eea2f 100644 --- a/structure/subagents.md +++ b/structure/subagents.md @@ -390,3 +390,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 retains fixed phase deadlines and reconciled replay output; see the [steering stability contract](transports/streaming-health.md#steering-deadlines-and-replay-completeness). diff --git a/structure/transports/byte-accounting.md b/structure/transports/byte-accounting.md index 65f25b9612..909824b37e 100644 --- a/structure/transports/byte-accounting.md +++ b/structure/transports/byte-accounting.md @@ -100,3 +100,5 @@ invent usage for an unreported failed send, retry a failed factory, or turn fail Source-iteration exceptions still propagate to the caller. Returning the guard iterator closes its active source; cancellation at an assistant boundary does not start the continuation callback. 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). diff --git a/structure/transports/inventory.md b/structure/transports/inventory.md index d7113c550f..bf85742689 100644 --- a/structure/transports/inventory.md +++ b/structure/transports/inventory.md @@ -156,3 +156,5 @@ Translated audio/file admission follows the [final-adapter input contract](../ad Canonical Responses identity sanitation and narrowly scoped pre-output combo recovery follow [request-local target compatibility](../runtime.md#request-local-target-compatibility); other adapter contracts remain unchanged. 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). diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 2833d62b30..d23ae44f9a 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -1090,3 +1090,5 @@ route where this was first observed; explicit provider and operator caps may onl 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). diff --git a/structure/transports/streaming-health.md b/structure/transports/streaming-health.md index d584a33dd3..cec983ca54 100644 --- a/structure/transports/streaming-health.md +++ b/structure/transports/streaming-health.md @@ -390,3 +390,41 @@ The wire relay does not synthesize or modify server-owned events or approvals. including false approval decisions, typed identity, content order, unsupported injection batches, sparse terminals and explicit mode transitions. No test asserts that a live subscription backend accepts these optional execution modes. + +### Steering deadlines and replay completeness + +`src/server/responses/native-steering.ts` uses monotonic, per-submission 90-second +acknowledgement deadlines. Accepting or rejecting a steer removes only that +submission's deadline; later steers or unrelated output never extend another +submission's time. Accepted input can wait for a safe boundary while the active +response retains ordinary sliding idle liveness. At a parent terminal, outstanding +steering gets a fixed 90-second successor deadline. The first valid +`waiting_for_required_input` notification replaces that parent's successor wait +with a 30-minute tool/approval deadline; repeated notifications cannot restart it. +An explicit saved-result continuation starts a fresh 90-second successor bound +at local submission, including any existing pacing/auth wait. Late pending events +or a rejected steer cannot extend or cancel that in-flight continuation's bound. +Unacknowledged steers retain their own earlier deadlines during these phase changes. + +One unrefed timer tracks the earliest deadline. A late control or response event +cannot rescue an expired deadline before the timer callback runs. Expiry settles +once, clears retained replay bodies and follows the existing connection-failure +path. It reports unknown delivery, not a synthesized rejection or success, and +never resends instructions/results, reruns a tool or chooses another account. +Normal completion and detach cancel the timer. Defaults and frame/count limits +remain unchanged; no capability or execution-mode allowance is added. + +`src/server/responses/native-steering-replay.ts` uses the same +`src/server/responses/native-response-output.ts` reconciliation as injection +replay: retain completed wire items omitted by a sparse terminal, match shared +identities by content and relative order, and reject contradictions before calling +the continuation-cache writer. This affects local replay, not the original wire +terminal. Completed parents can be remembered; failed/incomplete parent output +stays private until a validated successor commits the prefix. Merged output is +charged against the unchanged 32 MiB serialized history budget. The existing +body-persistence eligibility and accepted-only steering commit rules still apply. +`tests/responses/ws-steering-stability.test.ts` binds these deadline and replay +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. diff --git a/tests/fixtures/test-layout-expected.json b/tests/fixtures/test-layout-expected.json index d3265ef130..2cc60235a0 100644 --- a/tests/fixtures/test-layout-expected.json +++ b/tests/fixtures/test-layout-expected.json @@ -1323,5 +1323,6 @@ "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" } diff --git a/tests/helpers/responses-core-source.ts b/tests/helpers/responses-core-source.ts index 5ec7a0c6f9..6abfaca99a 100644 --- a/tests/helpers/responses-core-source.ts +++ b/tests/helpers/responses-core-source.ts @@ -11,6 +11,7 @@ export const RESPONSES_CORE_MODULES = [ "native-response-control.ts", "native-tool-results.ts", "native-response-output.ts", + "native-response-json.ts", "native-injection-protocol.ts", "native-injection-replay.ts", "native-steering.ts", diff --git a/tests/responses/ws-steering-stability.test.ts b/tests/responses/ws-steering-stability.test.ts new file mode 100644 index 0000000000..d040484cf5 --- /dev/null +++ b/tests/responses/ws-steering-stability.test.ts @@ -0,0 +1,272 @@ +import { expect, spyOn, test } from "bun:test"; +import { NativeSteeringChannel, NATIVE_STEERING_WAIT_MS as WAIT, + NATIVE_STEERING_TOOL_WAIT_MS as TOOL_WAIT } from "../../src/server/responses/native-steering"; +import { NativeSteeringReplay, MAX_NATIVE_STEERING_REPLAY_BYTES as REPLAY_LIMIT } from "../../src/server/responses/native-steering-replay"; +import { beginInjection, injectionConfig, installInjectionFixture, InjectionSocket, + waitForInjection, fallbackCalls, type Frame } from "../helpers/native-injection-fixture"; + +installInjectionFixture(); + +/** Synchronous monotonic clock: exercise real owner transitions without wall-clock sleeps. */ +function clock() { + let now = 1_000; + const timers = new Map void }>(); + const time = spyOn(performance, "now").mockImplementation(() => now); + const schedule = spyOn(globalThis, "setTimeout").mockImplementation(((run: () => void, delay = 0) => { + const timer = { unref() { return timer; } }; + timers.set(timer, { due: now + delay, run }); + return timer; + }) as unknown as typeof setTimeout); + const cancel = spyOn(globalThis, "clearTimeout").mockImplementation(((timer: object) => { timers.delete(timer); }) as typeof clearTimeout); + return { + get pending() { return timers.size; }, + advance(ms: number, fire = true) { + now += ms; + if (!fire) return; + for (let count = 0; count < 100; count++) { + const ready = [...timers.entries()].find(([, value]) => value.due <= now); + if (!ready) return; + timers.delete(ready[0]); ready[1].run(); + } + throw new Error("fixture timer rescheduled without progress"); + }, + restore() { schedule.mockRestore(); cancel.mockRestore(); time.mockRestore(); }, + }; +} +const steer = (input = "change", id = "root") => ({ type: "response.steer", previous_response_id: id, input }); +const accepted = (id = "s1", parent = "root") => ({ type: "response.steer.accepted", steer: { id, previous_response_id: parent } }); +const terminal = (id = "root", output: Frame[] = []) => ({ type: "response.completed", response: { id, status: "completed", output } }); +const stub = { type: "function_call_output", call_id: "saved-call" }; +const pending = (id = "s1") => ({ type: "response.steer.pending", steer: { id, previous_response_id: "root" }, reason: "waiting_for_required_input", required_input: [stub] }); +const continuation = () => ({ type: "response.create", previous_response_id: "root", input: [{ ...stub, output: "saved result" }] }); + +/** Every fixture restores timer hooks even when demonstrating a pre-fix failure. */ +function unit(run: (value: ReturnType) => void) { + const value = unitValue(); + try { run(value); } finally { value.detach(); value.time.restore(); } +} +function unitValue() { + const time = clock(); + const sent: Frame[] = []; + const failures: Error[] = []; + const channel = new NativeSteeringChannel({}); + const detach = channel.attach(frame => sent.push(frame), error => failures.push(error)); + channel.observe({ type: "response.created", response: { id: "root" } }); + const activity = () => channel.observe({ type: "response.in_progress", response: { id: "root" } }); + return { time, sent, failures, channel, detach, activity }; +} + +test("steer acknowledgement expires despite continuous response activity", () => unit(({ channel, activity, time, failures, sent }) => { + channel.steer(steer()); + for (let i = 0; i < 2; i++) { time.advance(WAIT / 3); activity(); } + time.advance(WAIT / 3); + expect(failures).toHaveLength(1); expect(failures[0].message).toContain("unknown"); + expect(channel.ended).toBe(true); expect(sent).toHaveLength(1); + time.advance(WAIT * 3); expect(failures).toHaveLength(1); +})); + +test("later submissions cannot postpone the oldest unacknowledged steer", () => unit(({ channel, time, failures, sent }) => { + channel.steer(steer("first")); time.advance(WAIT / 2); + channel.steer(steer("second")); time.advance(WAIT / 2); + expect(failures).toHaveLength(1); expect(sent).toHaveLength(2); +})); + +test("acknowledgement removes only its submission deadline", () => unit(({ channel, time, failures, activity }) => { + channel.steer(steer("first")); time.advance(10_000); channel.steer(steer("second")); + time.advance(10_000); channel.observe(accepted()); + time.advance(WAIT - 20_000); activity(); expect(failures).toHaveLength(0); + time.advance(10_000); expect(failures).toHaveLength(1); +})); + +test("an unacknowledged steer keeps its deadline during a tool wait for another steer", () => unit(({ channel, time, failures }) => { + channel.steer(steer("one")); channel.steer(steer("two")); channel.observe(accepted()); + channel.observe(terminal()); time.advance(1000); channel.observe(pending()); + time.advance(WAIT - 1000); expect(failures).toHaveLength(1); +})); + +test("automatic successor has a fixed deadline from parent termination", () => unit(({ channel, time, failures, activity }) => { + channel.steer(steer()); channel.observe(accepted()); + time.advance(20_000); channel.observe(terminal()); + time.advance(40_000); activity(); time.advance(49_999); activity(); + expect(failures).toHaveLength(0); time.advance(1); expect(failures).toHaveLength(1); +})); + +test("acceptance after the terminal cannot restart the successor deadline", () => unit(({ channel, time, failures, activity }) => { + channel.steer(steer()); time.advance(1000); channel.observe(terminal()); + time.advance(40_000); channel.observe(accepted()); + time.advance(49_999); activity(); expect(failures).toHaveLength(0); + time.advance(1); expect(failures).toHaveLength(1); +})); + +test("repeated required-input notifications share the first parent tool deadline", () => unit(({ channel, time, failures }) => { + channel.steer(steer("one")); channel.steer(steer("two")); + channel.observe(accepted()); channel.observe(accepted("s2")); channel.observe(terminal()); + time.advance(1000); channel.observe(pending()); + time.advance(TOOL_WAIT - 1); channel.observe(pending("s2")); + expect(failures).toHaveLength(0); time.advance(1); expect(failures).toHaveLength(1); +})); + +test("saved results get a new successor deadline and late pending cannot extend it", () => unit(({ channel, time, failures, sent }) => { + channel.steer(steer()); channel.observe(accepted()); channel.observe(terminal()); channel.observe(pending()); + time.advance(200_000); expect(channel.continue(continuation())).toBe(true); + time.advance(WAIT - 1); channel.observe(pending()); + expect(failures).toHaveLength(0); time.advance(1); + expect(failures).toHaveLength(1); expect(sent).toHaveLength(2); +})); + +test("early saved results start a successor deadline before required-input notification", () => unit(({ channel, time, failures }) => { + channel.steer(steer()); channel.observe(accepted()); + channel.observe(terminal("root", [{ type: "function_call", call_id: stub.call_id }])); + time.advance(1000); channel.continue(continuation()); + time.advance(WAIT - 1); channel.observe(pending()); + expect(failures).toHaveLength(0); time.advance(1); expect(failures).toHaveLength(1); +})); + +test("a dispatched continuation keeps its deadline when accepted steering fails", () => unit(({ channel, time, failures, sent }) => { + channel.steer(steer()); channel.observe(accepted()); channel.observe(terminal()); channel.observe(pending()); + channel.continue(continuation()); time.advance(20_000); + channel.observe({ type: "response.steer.failed", steer: { id: "s1", previous_response_id: "root" } }); + time.advance(WAIT - 20_000); expect(failures).toHaveLength(1); expect(sent).toHaveLength(2); +})); + +test("an acknowledgement cannot rescue an expired deadline before the timer callback runs", () => unit(({ channel, time, failures }) => { + channel.steer(steer()); time.advance(WAIT, false); + expect(() => channel.observe(accepted())).toThrow("unknown"); + expect(failures).toHaveLength(1); expect(channel.ended).toBe(true); +})); + +test("normal response activity refreshes only idle liveness", () => unit(({ channel, activity, time, failures }) => { + time.advance(250_000); activity(); time.advance(250_000); activity(); + expect(failures).toHaveLength(0); time.advance(300_000); + expect(failures).toHaveLength(1); expect(channel.ended).toBe(true); +})); + +test("a rejected steer removes its hard deadline without ending a live response", () => unit(({ channel, activity, time, failures }) => { + channel.steer(steer()); time.advance(20_000); + channel.observe({ type: "response.steer.failed", steer: { previous_response_id: "root" } }); + time.advance(WAIT); activity(); expect(failures).toHaveLength(0); + expect(channel.observe(terminal())).toBe(true); expect(time.pending).toBe(0); +})); + +test("created successor clears old phase deadlines and receives its own idle interval", () => unit(({ channel, time, failures }) => { + channel.steer(steer()); channel.observe(accepted()); channel.observe(terminal()); + time.advance(WAIT - 1); + channel.observe({ type: "response.created", response: { id: "next", previous_response_id: "root" } }); + time.advance(WAIT * 2); expect(failures).toHaveLength(0); + expect(channel.observe(terminal("next"))).toBe(true); expect(time.pending).toBe(0); +})); + +test("detach cancels pending deadlines and prevents delayed failure callbacks", () => unit(({ channel, time, detach, failures }) => { + channel.steer(steer()); detach(); expect(time.pending).toBe(0); + time.advance(TOOL_WAIT * 2); expect(failures).toHaveLength(0); +})); + +const output = [ + { id: "reason", type: "reasoning", encrypted_content: "fixture-opaque-reasoning", summary: [] }, + { id: "tool", type: "function_call", call_id: "saved-call", name: "read", arguments: "{}" }, + { id: "message", type: "message", role: "assistant", content: [{ type: "output_text", text: "done", annotations: [] }] }, +]; +/** Replay assertions inspect committed cache inputs, not just server events on the wire. */ +function replayFixture() { + const stored: Frame[] = []; + const replay = new NativeSteeringReplay("initial", (input, response) => stored.push(structuredClone({ input, response }))); + replay.observe({ type: "response.created", response: { id: "root" } }); + output.forEach((item, output_index) => replay.observe({ type: "response.output_item.done", output_index, item })); + return { replay, stored }; +} + +test.each(["response.completed", "response.incomplete", "response.failed"])("%s sparse output survives a steering successor prefix", type => { + const { replay, stored } = replayFixture(); + try { + replay.submitted(steer()); replay.observe(accepted()); + replay.observe({ type, response: { id: "root", output: [structuredClone(output[2])] } }); + expect(stored).toHaveLength(type === "response.completed" ? 1 : 0); + replay.observe({ type: "response.created", response: { id: "next", previous_response_id: "root" } }); + replay.observe(terminal("next")); + expect(stored.at(-1)!.input.slice(1, 4)).toEqual(output); + expect(JSON.stringify(stored.at(-1)!.input.at(-1))).toContain("change"); + } finally { replay.dispose(); } +}); + +test("matching terminal echoes appear once even when object-key order differs", () => { + const { replay, stored } = replayFixture(); + try { + replay.observe(terminal("root", [{ ...output[2], id: "message" }])); + expect(stored[0].response.output).toEqual(output); + expect(stored[0].response.output.map((item: Frame) => item.id)).toEqual(["reason", "tool", "message"]); + } finally { replay.dispose(); } +}); + +test.each([ + [{ ...output[0], encrypted_content: "contradiction" }], + [output[2], output[0]], + [output[0], output[0]], +])("conflicting terminal content/order/duplicates never enters the shared cache", terminalOutput => { + const { replay, stored } = replayFixture(); + try { + expect(() => replay.observe(terminal("root", terminalOutput))).toThrow(); + expect(stored).toHaveLength(0); + } finally { replay.dispose(); } +}); + +test("merged sparse output still enforces the unchanged serialized replay budget", () => { + const remembered: unknown[] = []; + const replay = new NativeSteeringReplay([], (_, value) => remembered.push(value)); + try { + replay.observe({ type: "response.created", response: { id: "root" } }); + const one = { id: "large-wire", type: "reasoning", encrypted_content: "x".repeat(REPLAY_LIMIT / 2) }; + replay.observe({ type: "response.output_item.done", output_index: 0, item: one }); + expect(() => replay.observe(terminal("root", [{ ...one, id: "large-terminal" }]))).toThrow("budget"); + expect(remembered).toHaveLength(0); + } finally { replay.dispose(); } +}); + +test("real steering handler preserves sparse output through automatic and ordinary successors", async () => { + const settings = { ...injectionConfig(), codexNativeInjection: false, codexNativeSteering: true }; + const { ws, socket, send, sent, id } = await beginInjection({ multi_agent: undefined }, settings); + for (let index = 0; index < output.length; index++) { + socket.emit({ type: "response.output_item.added", output_index: index, item: output[index] }); + socket.emit({ type: "response.output_item.done", output_index: index, item: output[index] }); + } + send(steer("preserve the reasoning", id)); socket.emit(accepted("s1", id)); + socket.emit({ type: "response.incomplete", response: { id, status: "incomplete", incomplete_details: { reason: "steered" }, output: [output[2]] } }); + socket.emit({ type: "response.created", response: { id: "next", previous_response_id: id } }); + socket.emit(terminal("next")); await waitForInjection(() => !ws.data.nativeControl); + send({ type: "response.create", model: "gpt-5.6-sol", previous_response_id: "next", input: "ordinary followup" }); + await waitForInjection(() => InjectionSocket.all.length === 2 && InjectionSocket.all[1].frames.length > 0); + const next = InjectionSocket.all[1]; + const history = JSON.stringify(next.frames[0].input); + expect(history).toContain("fixture-opaque-reasoning"); expect(history).toContain("saved-call"); + expect(history).toContain("preserve the reasoning"); expect(history).toContain("ordinary followup"); + expect(sent.find(frame => frame.type === "response.incomplete")?.response.output).toEqual([output[2]]); + expect(fallbackCalls).toBe(0); next.emit(terminal(next.root)); + await waitForInjection(() => !ws.data.nativeControl); +}); + +test("accepted input allows active work beyond the acknowledgement window until a safe boundary", () => unit(({ channel, time, failures, activity }) => { + channel.steer(steer()); channel.observe(accepted()); + for (let i = 0; i < 4; i++) { time.advance(WAIT - 1); activity(); } + expect(failures).toHaveLength(0); channel.observe(terminal()); + time.advance(WAIT); expect(failures).toHaveLength(1); +})); + +test("a new steer cannot rescue an expired submission when timers are delayed", () => unit(({ channel, time, failures, sent }) => { + channel.steer(steer()); time.advance(WAIT, false); + expect(() => channel.steer(steer("late"))).toThrow("unknown"); + expect(sent).toHaveLength(1); expect(failures).toHaveLength(1); +})); + +test("a late continuation cannot rescue an expired tool wait", () => unit(({ channel, time, failures, sent }) => { + channel.steer(steer()); channel.observe(accepted()); channel.observe(terminal()); channel.observe(pending()); + time.advance(TOOL_WAIT, false); + expect(() => channel.continue(continuation())).toThrow("unknown"); + expect(sent).toHaveLength(1); expect(failures).toHaveLength(1); +})); + +test("wall-clock corrections cannot change monotonic acknowledgement deadlines", () => unit(({ channel, time, failures, activity }) => { + const wallClock = spyOn(Date, "now").mockReturnValue(0); + try { + channel.steer(steer()); time.advance(WAIT - 1); wallClock.mockReturnValue(10 ** 15); activity(); + expect(failures).toHaveLength(0); time.advance(1); expect(failures).toHaveLength(1); + } finally { wallClock.mockRestore(); } +})); From b7d4979597757ff1d154d16e68cf73aac443e939 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:50:58 +0900 Subject: [PATCH 2/5] fix(server): validate native control settings before superseding the active turn A malformed response.create frame cancelled the live turn before its steering channel was constructed, so a rejected frame could discard active work without recording a replacement. Build the channel first; only cancel after it validates. --- src/server/index/websocket-handler.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/server/index/websocket-handler.ts b/src/server/index/websocket-handler.ts index 20199d9e1f..c45eafda6d 100644 --- a/src/server/index/websocket-handler.ts +++ b/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; From 3dd78fb96b20907892045c9e7728558591a4fe5e Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:51:40 +0900 Subject: [PATCH 3/5] fix(responses): reject malformed output_index in steering replay A response.output_item.done frame with a non-safe-integer index matched no branch and was silently dropped from retained output. Validate inside the branch and throw, matching the injection replay observer. --- src/server/responses/native-steering-replay.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/server/responses/native-steering-replay.ts b/src/server/responses/native-steering-replay.ts index fa7950922e..1b6405f3a0 100644 --- a/src/server/responses/native-steering-replay.ts +++ b/src/server/responses/native-steering-replay.ts @@ -93,9 +93,9 @@ 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)); From 4a1b0ea068ddc8f233ddaf3b4830cc9a1f212df1 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:51:41 +0900 Subject: [PATCH 4/5] perf(responses): cache the parsed base frame across steering continuations sendControl re-parsed the full original frameText for every response.create continuation; a full-replay frame runs to megabytes. Hoist the parse and reuse the immutable base. --- src/server/responses/codex-ws-exchange.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/server/responses/codex-ws-exchange.ts b/src/server/responses/codex-ws-exchange.ts index 89f3096f0c..c79e1cef54 100644 --- a/src/server/responses/codex-ws-exchange.ts +++ b/src/server/responses/codex-ws-exchange.ts @@ -338,6 +338,9 @@ export function codexWsExchange(options: ExchangeOptions): Promise { 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 | 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 { // 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; + base ??= JSON.parse(frameText) as Record; outgoing = { ...base, input: frame.input, previous_response_id: frame.previous_response_id }; } const text = JSON.stringify(outgoing); From 77c65e1a99ddac10343ba2ecc456f6c7a62c5713 Mon Sep 17 00:00:00 2001 From: luvs01 <27862058+luvs01@users.noreply.github.com> Date: Thu, 17 Sep 2026 17:52:07 +0900 Subject: [PATCH 5/5] docs: name the canonical steering route and drop duplicated policy text State that native steering requires the canonical ChatGPT forward route, describe control deadlines as fixed rather than inactivity-based, and reduce the server reference paragraphs to a scope summary with the canonical guide links. --- .../src/content/docs/guides/codex-integration.md | 4 ++-- .../content/docs/reference/configuration/server.md | 12 +++++------- 2 files changed, 7 insertions(+), 9 deletions(-) diff --git a/docs-site/src/content/docs/guides/codex-integration.md b/docs-site/src/content/docs/guides/codex-integration.md index a9db588d90..baafd4afad 100644 --- a/docs-site/src/content/docs/guides/codex-integration.md +++ b/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 diff --git a/docs-site/src/content/docs/reference/configuration/server.md b/docs-site/src/content/docs/reference/configuration/server.md index 6bb048efaa..ab2b96e2a2 100644 --- a/docs-site/src/content/docs/reference/configuration/server.md +++ b/docs-site/src/content/docs/reference/configuration/server.md @@ -575,15 +575,13 @@ wildcard `hostname`, where the public listener already holds `127.0.0.1:`. 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). -Native steering confirmation uses per-submission and per-phase absolute deadlines, -not a timeout renewed by unrelated output. Its local continuation history reconciles -sparse terminal summaries with completed output already received. See +`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.