fix(responses): restore plaintext V2 calls from verified unlabeled SSE

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.

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<Response> {
+  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<Uint8Array>({
+        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<ResponsesRequestContext, "logCtx" | "config" | "options" | "req">,
@@ -180,7 +254,6 @@ export async function deliverPassthroughResponse(
 ): Promise<Response> {
   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);

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

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
 

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<Record<string, unknown>>;
+      };
+      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<Uint8Array>({
+      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(