fix(sse): preserve captured requests on streaming errors

changelog.d/fixes/preserve-captured-provider-request.md

@@ -0,0 +1 @@
+- **fix(sse):** Preserve the captured outgoing request when streaming provider execution fails, and leave provider-request diagnostics absent when no outgoing request was captured.

open-sse/handlers/chatCore/streamingResponse.ts

@@ -28,8 +28,6 @@ export {
   stripStaleForwardingHeaders,
 };
 
-import { normalizeHeaders } from "../../utils/headers.ts";
-
 import { FORMATS } from "../../translator/formats.ts";
 
 import { COLORS } from "../../utils/stream.ts";
@@ -315,9 +313,10 @@ export async function runStreamingResponse(deps: StreamingDeps) {
     upstreamDiagnostic = pipelineOutcome.upstreamDiagnostic;
     if (pipelineOutcome.kind === "error") {
       providerResponse = pipelineOutcome.result.response;
-      providerUrl = "";
-      providerHeaders = normalizeHeaders(pipelineOutcome.result.response.headers);
-      finalBody = translatedBody;
+      const lastRequest = providerRequestCapture.latest();
+      providerUrl = lastRequest?.url;
+      providerHeaders = lastRequest?.headers;
+      finalBody = lastRequest?.body;
     } else {
       const result = {
         response: pipelineOutcome.response,
@@ -339,13 +338,17 @@ export async function runStreamingResponse(deps: StreamingDeps) {
       clientRawRequest?.headers
     );
 
-    // Log target request (final request to provider)
-    reqLogger.logTargetRequest(providerUrl, providerHeaders, finalBody);
-    updatePendingScope(pendingScope, {
-      providerRequest: finalBody,
-      providerUrl,
-      stage: "provider_response_started",
-    });
+    // Keep the captured request on failure; outcome metadata may describe the response.
+    if (pipelineOutcome.kind === "response") {
+      reqLogger.logTargetRequest(providerUrl, providerHeaders, finalBody);
+      updatePendingScope(pendingScope, {
+        providerRequest: finalBody,
+        providerUrl,
+        stage: "provider_response_started",
+      });
+    } else {
+      updatePendingScope(pendingScope, { stage: "provider_response_started" });
+    }
     // Update rate limiter from response headers (learn limits dynamically)
     updateFromHeaders(
       provider,
@@ -405,7 +408,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
       persistAttemptLogs({
         status: HTTP_STATUS.RATE_LIMITED,
         error: failureMessage,
-        providerRequest: finalBody || translatedBody,
+        providerRequest: providerRequestCapture.latest()?.body,
         clientResponse: buildErrorBody(HTTP_STATUS.RATE_LIMITED, failureMessage),
         claudeCacheMeta: claudePromptCacheLogMeta,
         cacheSource: "upstream",
@@ -500,7 +503,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
     persistAttemptLogs({
       status: failureStatus,
       error: safeFailureMessage,
-      providerRequest: finalBody || translatedBody,
+      providerRequest: providerRequestCapture.latest()?.body,
       // On a client-abort (AbortError), the client already disconnected before
       // we ever got here — this body is what we WOULD have sent, not what was
       // actually delivered. Logging it as `clientResponse` is misleading (the
@@ -1002,8 +1005,8 @@ export async function runStreamingResponse(deps: StreamingDeps) {
       log?.debug?.("RETRY", `Antigravity quota reset in ${retrySeconds}s (${retryAfterMs}ms)`);
     }
 
-    // Log error with full request body for debugging
-    reqLogger.logError(new Error(message), finalBody || translatedBody);
+    // Log only a captured outgoing request body with provider errors.
+    reqLogger.logError(new Error(message), providerRequestCapture.latest()?.body);
     const persistedProviderErrorBody = logTerminalProviderResponse(
       reqLogger,
       provider,
@@ -1050,7 +1053,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
             persistAttemptLogs({
               status: statusCode,
               error: safeErrMsg,
-              providerRequest: finalBody || translatedBody,
+              providerRequest: providerRequestCapture.latest()?.body,
               providerResponse: persistedProviderErrorBody,
               clientResponse: buildErrorBody(statusCode, errMsg),
               cacheSource: "upstream",
@@ -1082,7 +1085,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
           persistAttemptLogs({
             status: statusCode,
             error: safeErrMsg,
-            providerRequest: finalBody || translatedBody,
+            providerRequest: providerRequestCapture.latest()?.body,
             providerResponse: persistedProviderErrorBody,
             clientResponse: buildErrorBody(statusCode, errMsg),
             cacheSource: "upstream",
@@ -1114,7 +1117,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
         persistAttemptLogs({
           status: statusCode,
           error: safeErrMsg,
-          providerRequest: finalBody || translatedBody,
+          providerRequest: providerRequestCapture.latest()?.body,
           providerResponse: persistedProviderErrorBody,
           clientResponse: buildErrorBody(statusCode, errMsg),
           cacheSource: "upstream",
@@ -1171,7 +1174,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
             persistAttemptLogs({
               status: statusCode,
               error: safeErrMsg,
-              providerRequest: finalBody || translatedBody,
+              providerRequest: providerRequestCapture.latest()?.body,
               providerResponse: persistedProviderErrorBody,
               clientResponse: buildErrorBody(statusCode, errMsg),
               cacheSource: "upstream",
@@ -1203,7 +1206,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
           persistAttemptLogs({
             status: statusCode,
             error: safeErrMsg,
-            providerRequest: finalBody || translatedBody,
+            providerRequest: providerRequestCapture.latest()?.body,
             providerResponse: persistedProviderErrorBody,
             clientResponse: buildErrorBody(statusCode, errMsg),
             cacheSource: "upstream",
@@ -1235,7 +1238,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
         persistAttemptLogs({
           status: statusCode,
           error: safeErrMsg,
-          providerRequest: finalBody || translatedBody,
+          providerRequest: providerRequestCapture.latest()?.body,
           providerResponse: persistedProviderErrorBody,
           clientResponse: buildErrorBody(statusCode, errMsg),
           cacheSource: "upstream",
@@ -1267,7 +1270,7 @@ export async function runStreamingResponse(deps: StreamingDeps) {
       persistAttemptLogs({
         status: statusCode,
         error: safeErrMsg,
-        providerRequest: finalBody || translatedBody,
+        providerRequest: providerRequestCapture.latest()?.body,
         providerResponse: persistedProviderErrorBody,
         clientResponse: buildErrorBody(statusCode, errMsg),
         cacheSource: "upstream",

open-sse/utils/providerRequestLogging.ts

@@ -97,10 +97,21 @@ async function capturePreparedRequest(
   log?: WarnLog | null
 ) {
   if (!requestCapture) return;
-  const latest = requestCapture.latest?.();
-  if (latest?.url === url && latest.bodyString === bodyString) return;
-
   try {
+    const headerEntries = Array.from(new Headers(headers).entries());
+    const latest = requestCapture.latest?.();
+    if (latest?.url === url && latest.bodyString === bodyString) {
+      const latestEntries = Array.from(new Headers(latest.headers).entries());
+      if (
+        headerEntries.length === latestEntries.length &&
+        headerEntries.every(
+          ([name, value], index) =>
+            name === latestEntries[index][0] && value === latestEntries[index][1]
+        )
+      ) {
+        return;
+      }
+    }
     await requestCapture.capture({ url, headers, body, bodyString });
   } catch {
     // Capture hooks can fail while processing a provider payload. The thrown
@@ -290,11 +301,11 @@ export function createPreparedRequestLogger(
       // a direct call from a settled dispatch's async resource must not overwrite the
       // body of a fresh retry on this same Capture.
       if (dispatchContext.getStore()?.settled) return;
-      latest = request;
-      reqLogger.logTargetRequest(request.url, request.headers, request.body);
+      latest = { ...request, headers: { ...request.headers } };
+      reqLogger.logTargetRequest(latest.url, latest.headers, latest.body);
       updatePendingScope(scope, {
-        providerRequest: request.body,
-        providerUrl: request.url,
+        providerRequest: latest.body,
+        providerUrl: latest.url,
         stage: "sending_to_provider",
       });
     },