diff --git a/docs-site/src/content/docs/guides/sub-agent-surface.md b/docs-site/src/content/docs/guides/sub-agent-surface.md index 939d93b4ab2..26a5f8677e0 100644 --- a/docs-site/src/content/docs/guides/sub-agent-surface.md +++ b/docs-site/src/content/docs/guides/sub-agent-surface.md @@ -344,3 +344,6 @@ encrypted, but task text can be retained in Codex history, routed-provider reque response/debug state. Existing ciphertext is unchanged, and the option depends on undocumented ChatGPT and Codex behavior. See [Agent configuration: Plaintext v2 agent messages](/reference/configuration/agents/#plaintext-v2-agent-messages). +If a streamed ChatGPT response omits or mislabels its content type, opencodex checks a bounded +Responses event prefix before restoring the message-tool names. An unrecognized body still fails +closed. diff --git a/src/server/responses/passthrough-delivery.ts b/src/server/responses/passthrough-delivery.ts index 71e19ebd209..25a1e6cc880 100644 --- a/src/server/responses/passthrough-delivery.ts +++ b/src/server/responses/passthrough-delivery.ts @@ -127,6 +127,80 @@ import { restoreRoutedCustomCallsInJson } from "../../responses/custom-tool-comp import { restoreRoutedToolSearchCallsInJson } from "../../responses/tool-search-compat"; import { responsesJsonToSseStream } from "../responses-json-events"; +const PLAINTEXT_V2_SSE_PREFIX_LIMIT = 4096; + +function classifyPlaintextV2SsePrefix(prefix: string): "sse" | "unknown" | "more" { + const lastLineEnd = prefix.lastIndexOf("\n"); + if (lastLineEnd < 0) return "more"; + for (const rawLine of prefix.slice(0, lastLineEnd + 1).split("\n")) { + const line = rawLine.replace(/\r$/, "").trim(); + if (!line || line.startsWith(":")) continue; + if (/^(id|retry):/.test(line)) continue; + if (line.startsWith("event:")) { + return /^event:\s*(?:response\.[\w.-]+|error)$/.test(line) ? "sse" : "unknown"; + } + if (line.startsWith("data:")) { + try { + const value = JSON.parse(line.slice(5).trim()) as { type?: unknown }; + return typeof value.type === "string" && /^(?:response\.[\w.-]+|error)$/.test(value.type) + ? "sse" : "unknown"; + } catch { + return "unknown"; + } + } + return "unknown"; + } + return "more"; +} + +/** Confirm an unlabeled successful body is Responses SSE before alias restoration. */ +async function classifyPlaintextV2SseResponse(response: Response): Promise { + if (!response.body) return response; + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const buffered: Uint8Array[] = []; + let prefix = ""; + let inspectedBytes = 0; + while (inspectedBytes < PLAINTEXT_V2_SSE_PREFIX_LIMIT) { + const next = await reader.read(); + if (next.done) break; + buffered.push(next.value); + const inspected = next.value.subarray(0, PLAINTEXT_V2_SSE_PREFIX_LIMIT - inspectedBytes); + inspectedBytes += inspected.byteLength; + prefix += decoder.decode(inspected, { stream: true }); + const kind = classifyPlaintextV2SsePrefix(prefix); + if (kind === "sse") { + let bufferedIndex = 0; + const body = new ReadableStream({ + async pull(controller) { + if (bufferedIndex < buffered.length) { + controller.enqueue(buffered[bufferedIndex++]!); + return; + } + try { + const result = await reader.read(); + if (result.done) controller.close(); + else controller.enqueue(result.value); + } catch (error) { + controller.error(error); + } + }, + cancel(reason) { return reader.cancel(reason); }, + }); + const headers = new Headers(response.headers); + headers.set("content-type", "text/event-stream"); + return new Response(body, { status: response.status, statusText: response.statusText, headers }); + } + if (kind === "unknown") break; + } + void reader.cancel().catch(() => {}); + return new Response(null, { + status: response.status, + statusText: response.statusText, + headers: response.headers, + }); +} + /** One responsibility of the Responses request pipeline; state owners are explicit. */ export async function deliverPassthroughResponse( requestContext: Pick, @@ -180,7 +254,6 @@ export async function deliverPassthroughResponse( ): Promise { const { logCtx, config, options, req } = requestContext; const { - upstreamResponse, codexSafetyBufferingOptions, upstream, connectMs, @@ -205,18 +278,26 @@ export async function deliverPassthroughResponse( const { openAiSidecar } = sidecarState; const { requestBindings } = transportState; + let upstreamResponse = nativeExchange.upstreamResponse; + const originalContentType = upstreamResponse.headers.get("content-type"); + if (isUsageDebugEnabled() && originalContentType) logCtx.usageDebugContentType = originalContentType; + if (responseEffects.plaintextV2AgentMessageToolNames.size > 0 + && upstreamResponse.ok && upstreamResponse.body && parsed.stream + && !originalContentType?.toLowerCase().includes("text/event-stream") + && !originalContentType?.toLowerCase().includes("application/json") + && !isCodexWsUpstreamResponse(upstreamResponse) + && !(options.nativeControl && isNativeControlResponse(upstreamResponse))) { + upstreamResponse = await classifyPlaintextV2SseResponse(upstreamResponse); + } + const headers = sanitizePassthroughHeaders(upstreamResponse.headers, codexSafetyBufferingOptions); const resolvedModel = headers.get("openai-model")?.trim(); if (resolvedModel) { logCtx.servedModel = resolvedModel; if (!logCtx.preserveResolvedModelFromRoute) logCtx.resolvedModel = resolvedModel; } - if (isUsageDebugEnabled()) { - const upstreamContentType = upstreamResponse.headers.get("content-type"); - if (upstreamContentType) logCtx.usageDebugContentType = upstreamContentType; - } - // The chatgpt backend may omit Content-Type on SSE responses. Fall back to - // treating a successful body as SSE when the caller requested streaming. + // ChatGPT may omit Content-Type on SSE responses. Plaintext V2 responses + // reach this fallback only after their first Responses event is confirmed. const passthroughCt = headers.get("content-type")?.toLowerCase(); const isEventStream = passthroughCt?.includes("text/event-stream") || (responseEffects.plaintextV2AgentMessageToolNames.size === 0 && upstreamResponse.ok && !!upstreamResponse.body && !passthroughCt && parsed.stream); diff --git a/structure/subagents.md b/structure/subagents.md index 345ec4a9681..070f862138d 100644 --- a/structure/subagents.md +++ b/structure/subagents.md @@ -32,6 +32,9 @@ snapshot repair. Malformed, conflicting, unsupported or over-limit responses fai retrying the model. Raw stream inspection cannot publish plaintext continuation state: only restored client blocks reach its dedicated bounded collector. Foreign namespaces and opaque argument/metadata values remain unchanged; the empty encrypted-function-args marker is preserved. +For a streamed response with a missing or incorrect content type, the native passthrough reads at +most the first 4 KiB to confirm a Responses SSE event before restoring aliases. A body that does +not match still fails closed, and its bytes never reach Codex as a successful response. Startup warns that task text can remain in Codex history, selected-provider requests and local response/debug state. This is application-level plaintext over HTTPS, depends on undocumented diff --git a/structure/transports/responses.md b/structure/transports/responses.md index 5674169f73a..9abae99aef6 100644 --- a/structure/transports/responses.md +++ b/structure/transports/responses.md @@ -8,6 +8,7 @@ The configuration-only [plaintext V2 contract](../subagents.md#plaintext-v2-agen is scoped to canonical ChatGPT Responses forwarding; other source-area behavior described here is unchanged. Management provider-validation calls use the [initialization-independent relative send-path validation](../config.md#provider-relative-send-paths) before persistence. Cursor's localized native-shell names follow the [routing-commentary guard contract](../providers/cursor.md#cursor-native-exec). Plaintext collaboration restoration treats a null namespace as absent, rejects non-string namespace types, and restores the native namespace/name pair before HTTP/WS delivery and continuation publication. +When a successful streamed native response has no recognizable content type, the plaintext V2 path confirms a bounded Responses SSE prefix before applying that restoration; unknown bodies retain the fail-closed response. ## Responses HTTP/SSE diff --git a/tests/server/plaintext-v2-agent-messages-server.test.ts b/tests/server/plaintext-v2-agent-messages-server.test.ts index a81a60f1147..a22ff6126ac 100644 --- a/tests/server/plaintext-v2-agent-messages-server.test.ts +++ b/tests/server/plaintext-v2-agent-messages-server.test.ts @@ -274,6 +274,53 @@ describe("plaintext v2 agent messages at the Responses server boundary", () => { expect(clientBody).toContain('"encrypted_function_args":[]'); }); + test.each(["missing", "text/plain"] as const)( + "restores plaintext V2 calls from valid SSE with %s upstream content type", + async contentType => { + takeInheritedSpendHome(); + const payload = completedResponsePayload(`resp-plaintext-v2-${contentType}`); + const wire = `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: payload })}\n\ndata: [DONE]\n\n`; + globalThis.fetch = (async () => new Response(new TextEncoder().encode(wire), { + status: 200, + ...(contentType === "missing" ? {} : { headers: { "content-type": contentType } }), + })) as typeof fetch; + + const response = await handleResponses(collaborationRequest(), config(true), { model: "", provider: "" }); + const clientBody = await response.text(); + + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + expect(clientBody).toContain('"namespace":"collaboration"'); + expect(clientBody).toContain('"name":"spawn_agent"'); + expect(clientBody).not.toContain(PLAINTEXT_V2_COLLABORATION_NAMESPACE); + expect(clientBody).not.toContain('"name":"start_delegated_task"'); + const replay = expandPreviousResponseInput({ previous_response_id: payload.id, input: [] }) as { + input: Array>; + }; + expect(replay.input.find(value => value.type === "function_call")) + .toMatchObject({ namespace: "collaboration", name: "spawn_agent" }); + }, + ); + + test("recognizes a headerless SSE event split across upstream chunks", async () => { + takeInheritedSpendHome(); + const payload = completedResponsePayload("resp-plaintext-v2-split"); + const wire = `event: response.completed\ndata: ${JSON.stringify({ type: "response.completed", response: payload })}\n\ndata: [DONE]\n\n`; + globalThis.fetch = (async () => new Response(new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(wire.slice(0, 12))); + controller.enqueue(new TextEncoder().encode(wire.slice(12))); + controller.close(); + }, + }), { status: 200 })) as typeof fetch; + + const response = await handleResponses(collaborationRequest(), config(true), { model: "", provider: "" }); + const clientBody = await response.text(); + expect(response.status).toBe(200); + expect(clientBody).toContain('"name":"spawn_agent"'); + expect(clientBody).not.toContain(PLAINTEXT_V2_COLLABORATION_NAMESPACE); + }); + test("rejects an unclassified successful response while restoration is required", async () => { takeInheritedSpendHome(); globalThis.fetch = (async () => new Response(