test(sse): pin the pipeline error-classification and Kiro stream contracts

Regression guards for the two defects above: the pipeline must surface the
upstream error code/type, and a Kiro error frame now terminates the stream
after forwarding response.failed (the #12506 contract the sibling
stream-passthrough suite already encodes).

Co-authored-by: diegosouzapw <8016841+diegosouzapw@users.noreply.github.com>

tests/unit/kiro-tool-call-validation.test.ts

@@ -236,23 +236,44 @@ test("Kiro stream errors become Responses response.failed events", async () => {
     null,
     "kiro-model"
   );
-  const writer = transform.writable.getWriter();
-  const responseText = new Response(transform.readable).text();
+  const source = new ReadableStream<Uint8Array>({
+    start(controller) {
+      controller.enqueue(
+        textEncoder.encode(
+          `data: ${JSON.stringify({
+            error: {
+              message: "Invalid Kiro tool_call payload: missing nested MCP tool name at input.name",
+              type: "invalid_request_error",
+              code: "invalid_kiro_tool_call",
+            },
+          })}\n\n`
+        )
+      );
+      controller.close();
+    },
+  });
 
-  await writer.write(
-    textEncoder.encode(
-      `data: ${JSON.stringify({
-        error: {
-          message: "Invalid Kiro tool_call payload: missing nested MCP tool name at input.name",
-          type: "invalid_request_error",
-          code: "invalid_kiro_tool_call",
-        },
-      })}\n\n`
-    )
-  );
-  await writer.close();
-  const text = await responseText;
+  // #12506 turned an upstream error frame into a TERMINAL stream failure: the
+  // failure boundary forwards the projected event and then calls
+  // controller.error(), so the readable rejects right after the bytes land.
+  // Read it with a reader (the pattern the sibling boundary suite
+  // stream-passthrough-error-redaction.test.ts uses) — Response#text() can
+  // never resolve on a body that ends in an error.
+  const reader = source.pipeThrough(transform).getReader();
+  const decoder = new TextDecoder();
+  let text = "";
+  let streamError: unknown = null;
+  try {
+    for (;;) {
+      const chunk = await reader.read();
+      if (chunk.done) break;
+      text += decoder.decode(chunk.value);
+    }
+  } catch (caught) {
+    streamError = caught;
+  }
 
+  assert.ok(streamError, "a Kiro error frame must terminate the stream, not end it cleanly");
   assert.match(text, /event: response\.failed/);
   assert.match(text, /invalid_kiro_tool_call/);
   assert.match(text, /missing nested MCP tool name/);

tests/unit/provider-execution-pipeline.test.ts

@@ -119,13 +119,16 @@ function makeInput(opts: {
 }
 
 test("initial Codex 429: rotation resolver>=1 and successful retry", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   let resolverCallCount = 0;
   const input = makeInput({
-    policy: { allowAccountRotation: true, allowModelFallback: true, expectedConnectionId: undefined },
+    policy: {
+      allowAccountRotation: true,
+      allowModelFallback: true,
+      expectedConnectionId: undefined,
+    },
     provider: "codex",
     connectionId: "conn-a",
     send: async () => {
@@ -135,10 +138,13 @@ test("initial Codex 429: rotation resolver>=1 and successful retry", async () =>
           headers: { "retry-after": "1" },
         });
       }
-      return makeAttempt({
-        id: "chatcmpl-ok",
-        choices: [{ message: { role: "assistant", content: "rotated" }, finish_reason: "stop" }],
-      }, 200);
+      return makeAttempt(
+        {
+          id: "chatcmpl-ok",
+          choices: [{ message: { role: "assistant", content: "rotated" }, finish_reason: "stop" }],
+        },
+        200
+      );
     },
     getProviderCredentials: (async () => {
       resolverCallCount += 1;
@@ -157,9 +163,8 @@ test("initial Codex 429: rotation resolver>=1 and successful retry", async () =>
 });
 
 test("initial Antigravity 422 gcp_project_required: rotation resolver>=1 and successful retry", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   let resolverCallCount = 0;
   const input = makeInput({
@@ -169,7 +174,10 @@ test("initial Antigravity 422 gcp_project_required: rotation resolver>=1 and suc
     send: async () => {
       sendCount += 1;
       if (sendCount === 1) {
-        return makeAttempt({ error: { message: "gcp_project_required", type: "invalid_request" } }, 422);
+        return makeAttempt(
+          { error: { message: "gcp_project_required", type: "invalid_request" } },
+          422
+        );
       }
       return makeAttempt(
         {
@@ -196,9 +204,8 @@ test("initial Antigravity 422 gcp_project_required: rotation resolver>=1 and suc
 });
 
 test("follow-up rotation blocks resolver on Antigravity 422", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   let resolverCallCount = 0;
   const input = makeInput({
@@ -211,7 +218,10 @@ test("follow-up rotation blocks resolver on Antigravity 422", async () => {
     connectionId: "agy-a",
     send: async () => {
       sendCount += 1;
-      return makeAttempt({ error: { message: "gcp_project_required", type: "invalid_request" } }, 422);
+      return makeAttempt(
+        { error: { message: "gcp_project_required", type: "invalid_request" } },
+        422
+      );
     },
     getProviderCredentials: (async () => {
       resolverCallCount += 1;
@@ -230,9 +240,8 @@ test("follow-up rotation blocks resolver on Antigravity 422", async () => {
 });
 
 test("follow-up rotation blocks resolver on Codex 429", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   let resolverCallCount = 0;
   const input = makeInput({
@@ -264,9 +273,8 @@ test("follow-up rotation blocks resolver on Codex 429", async () => {
 });
 
 test("401 refresh succeeds then retries once on same connection", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   let refreshCount = 0;
   let persistCount = 0;
@@ -278,12 +286,17 @@ test("401 refresh succeeds then retries once on same connection", async () => {
     send: async () => {
       sendCount += 1;
       if (sendCount === 1) {
-        return makeAttempt({ error: { message: "invalid_api_key", type: "authentication_error" } }, 401);
+        return makeAttempt(
+          { error: { message: "invalid_api_key", type: "authentication_error" } },
+          401
+        );
       }
       return makeAttempt(
         {
           id: "chatcmpl-ok",
-          choices: [{ message: { role: "assistant", content: "refreshed" }, finish_reason: "stop" }],
+          choices: [
+            { message: { role: "assistant", content: "refreshed" }, finish_reason: "stop" },
+          ],
         },
         200
       );
@@ -314,9 +327,8 @@ test("401 refresh succeeds then retries once on same connection", async () => {
 });
 
 test("status restatement rewrites agentrouter 403 quota exhaustion to 429 before classification", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   const input = makeInput({
     policy: { allowAccountRotation: true, allowModelFallback: true },
@@ -338,9 +350,8 @@ test("status restatement rewrites agentrouter 403 quota exhaustion to 429 before
 });
 
 test("thinking-signature recovery returns winning response", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   const input = makeInput({
     policy: { allowAccountRotation: true, allowModelFallback: true },
@@ -350,7 +361,12 @@ test("thinking-signature recovery returns winning response", async () => {
       sendCount += 1;
       if (sendCount === 1) {
         return makeAttempt(
-          { error: { message: "invalid signature in thinking block", type: "invalid_request_error" } },
+          {
+            error: {
+              message: "invalid signature in thinking block",
+              type: "invalid_request_error",
+            },
+          },
           400
         );
       }
@@ -390,9 +406,8 @@ test("thinking-signature recovery returns winning response", async () => {
 });
 
 test("initial model-unavailable falls back to sibling model", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   let fallbackLookupCount = 0;
   const sentModels: string[] = [];
@@ -436,9 +451,8 @@ test("initial model-unavailable falls back to sibling model", async () => {
 });
 
 test("follow-up allowModelFallback=false blocks model-unavailable fallback", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   let sendCount = 0;
   let fallbackLookupCount = 0;
   const input = makeInput({
@@ -474,9 +488,8 @@ test("follow-up allowModelFallback=false blocks model-unavailable fallback", asy
 });
 
 test("Codex 429 rotation calls scope-rate-limit, affinity-clear, and audit hooks", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   const rateLimited: Array<Record<string, unknown>> = [];
   const affinityCleared: string[] = [];
   const audits: Array<Record<string, unknown>> = [];
@@ -529,9 +542,8 @@ test("Codex 429 rotation calls scope-rate-limit, affinity-clear, and audit hooks
 });
 
 test("Codex 429 cooldown reads Retry-After from the response, not request headers", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   const rateLimited: Array<Record<string, unknown>> = [];
   let sendCount = 0;
   const input = makeInput({
@@ -586,9 +598,8 @@ test("Codex 429 cooldown reads Retry-After from the response, not request header
 });
 
 test("Antigravity BYOP 422 rotation persists cooldown via setConnectionRateLimitedUntil", async () => {
-  const { runProviderExecutionPipeline } = await import(
-    "../../open-sse/handlers/chatCore/providerExecutionPipeline.ts"
-  );
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
   const cooldowns: Array<{ id: string; untilMs: number | null }> = [];
   let sendCount = 0;
   const input = makeInput({
@@ -630,3 +641,40 @@ test("Antigravity BYOP 422 rotation persists cooldown via setConnectionRateLimit
   assert.equal(typeof cooldowns[0]?.untilMs, "number");
   assert.equal((cooldowns[0]?.untilMs ?? 0) > Date.now(), true);
 });
+
+test("upstream error code/type survive into the error outcome", async () => {
+  const { runProviderExecutionPipeline } =
+    await import("../../open-sse/handlers/chatCore/providerExecutionPipeline.ts");
+  const input = makeInput({
+    policy: {
+      allowAccountRotation: false,
+      allowModelFallback: false,
+      expectedConnectionId: "agy-a",
+    },
+    provider: "antigravity",
+    connectionId: "agy-a",
+    send: async () =>
+      makeAttempt(
+        {
+          error: {
+            message: "Missing Google projectId for Antigravity account.",
+            type: "oauth_missing_project_id",
+            code: "missing_project_id",
+          },
+        },
+        422
+      ),
+  });
+
+  const outcome = await runProviderExecutionPipeline(input);
+  assert.equal(outcome.kind, "error");
+  if (outcome.kind === "error") {
+    // #12867 dropped this pair when the leg moved into the pipeline, so
+    // downstream gates that key on BOTH fields (e.g.
+    // isAntigravityMissingProjectError) silently stopped firing and a
+    // config-class 422 degraded into a generic account cooldown.
+    assert.equal(outcome.result.errorCode, "missing_project_id");
+    assert.equal(outcome.result.errorType, "oauth_missing_project_id");
+    assert.equal(outcome.result.status, 422);
+  }
+});