fix(kiro): treat pre-output stream socket closes as retryable

src/adapters/kiro.ts

@@ -586,18 +586,34 @@ function retryableKiroIncomplete(
   message: string,
   usage: OcxUsage,
   providerState: { kiro: { conversationId: string } } | undefined,
+  retryable = true,
 ): AdapterEvent {
   return {
     type: "incomplete",
     reason,
     message,
     usage,
-    retryable: true,
+    retryable,
     endTurn: false,
     ...(providerState ? { providerState } : {}),
   };
 }
 
+/**
+ * Catch-path retryability for #519: only transport/socket failures with no emitted output
+ * are replay-safe. Malformed event payloads (`invalid Kiro …`) and any post-output failure
+ * stay terminal — same spirit as cursor's emittedOutput gate.
+ */
+export function isRetryableKiroStreamCatchError(err: unknown, emittedOutput: boolean): boolean {
+  if (emittedOutput) return false;
+  const message = err instanceof Error ? err.message : String(err);
+  if (/^invalid Kiro\b/i.test(message)) return false;
+  // Include Smithy/eventstream truncation (`eventstream: truncated message at end of stream`):
+  // partial frame + clean EOF with zero output is the same replay-safe class as a socket close.
+  return /socket connection was closed|connection(?: was)? closed unexpectedly|ECONNRESET|EPIPE|UND_ERR_|fetch failed|decoder failed|premature close|other side closed|unexpected EOF|network connection lost|terminated|truncated message at end of stream|eventstream:\s*truncated/i
+    .test(message);
+}
+
 /**
  * Suppress only a whitespace-normalized exact repeat. Semantic/fuzzy matching was rejected
  * during review: two long near-identical messages can differ by a single status word
@@ -629,6 +645,8 @@ async function* parseKiroAttempt(
   conversationId: string | undefined,
   previousAssistantText?: string,
   contextInputEstimate?: number,
+  /** True when an earlier attempt already flushed visible content to the client (#520). */
+  priorEmittedOutput = false,
 ): AsyncGenerator<AdapterEvent, KiroAttemptResult> {
   // `required` mode holds staged commentary here so a terminal END_TURN can relabel it as the final
   // answer instead of paying for another inference request. Anything the inner parser leaves behind
@@ -646,6 +664,7 @@ async function* parseKiroAttempt(
     deferred,
     previousAssistantText,
     contextInputEstimate,
+    priorEmittedOutput,
   );
   let next = await attempt.next();
   while (!next.done) {
@@ -667,6 +686,7 @@ async function* parseKiroAttemptEvents(
   deferred: AdapterEvent[],
   previousAssistantText?: string,
   contextInputEstimate?: number,
+  priorEmittedOutput = false,
 ): AsyncGenerator<AdapterEvent, KiroAttemptResult> {
   const emptyResult = (): KiroAttemptResult => ({ assistantText: "", sawReasoning: false });
   if (!response.body) {
@@ -717,15 +737,29 @@ async function* parseKiroAttemptEvents(
     return contextTotal > 0 ? { ...base, contextTotalTokens: contextTotal } : base;
   };
 
-  const classifiedTerminal = (failure: KiroErrorClassification): AdapterEvent => ({
-    type: "error",
-    message: failure.message,
-    status: failure.status,
-    errorType: failure.errorType,
-    code: failure.code,
-    retryable: failure.retryable,
-    usage: usage(),
-  });
+  const classifiedTerminal = (failure: KiroErrorClassification): AdapterEvent => {
+    // Upstream exception/error frames can arrive after commentary was already staged (and will be
+    // flushed before this terminal is yielded). Replaying after that content would duplicate it.
+    const emittedOutput = priorEmittedOutput
+      || sawText
+      || sawReasoning
+      || sawRealTool
+      || assistantText.length > 0
+      || deferred.length > 0
+      || completionAnswer !== undefined
+      || completionCalls > 0
+      || open !== null
+      || fallbackEvents.length > 0;
+    return {
+      type: "error",
+      message: failure.message,
+      status: failure.status,
+      errorType: failure.errorType,
+      code: failure.code,
+      retryable: emittedOutput ? false : failure.retryable,
+      usage: usage(),
+    };
+  };
 
   const protocolTerminal = (message: string, malformedCompletion = false): AdapterEvent => {
     if (mode === "text_fallback" && malformedCompletion) {
@@ -734,6 +768,8 @@ async function* parseKiroAttemptEvents(
         message,
         usage(),
         providerState(),
+        // First-attempt progress was already flushed before this bounded fallback (#520).
+        !priorEmittedOutput,
       );
     }
     return {
@@ -1093,6 +1129,8 @@ async function* parseKiroAttemptEvents(
             : "Kiro produced no final answer on its bounded completion retry",
           finalUsage,
           finalProviderState,
+          // First-attempt progress was already flushed before this bounded fallback (#520).
+          !priorEmittedOutput,
         ),
       };
     }
@@ -1195,6 +1233,22 @@ async function* parseKiroAttemptEvents(
       },
     };
   } catch (err) {
+    // Mid-stream socket closes after response.created / heartbeats only must stay retryable:
+    // nothing was relayed to the client, so a string-body replay is safe (see #519 / cursor's
+    // emittedOutput gate). Once any assistant text, reasoning, tool, or deferred content exists
+    // — including content flushed by a prior attempt before a bounded fallback — fail closed;
+    // the client may already have partial output. Protocol parse throws stay non-retryable even
+    // with zero output.
+    const emittedOutput = priorEmittedOutput
+      || sawText
+      || sawReasoning
+      || sawRealTool
+      || assistantText.length > 0
+      || deferred.length > 0
+      || completionAnswer !== undefined
+      || completionCalls > 0
+      || open !== null
+      || fallbackEvents.length > 0;
     return {
       assistantText,
       sawReasoning,
@@ -1204,7 +1258,7 @@ async function* parseKiroAttemptEvents(
         status: 502,
         errorType: "server_error",
         code: "kiro_stream_protocol_error",
-        retryable: false,
+        retryable: isRetryableKiroStreamCatchError(err, emittedOutput),
         usage: usage(),
       },
     };
@@ -1255,6 +1309,10 @@ export async function* parseKiroStream(
   }
 
   yield { type: "heartbeat" };
+  // First attempt already flushed deferred progress before this point. Gate fallback
+  // setup/HTTP failures the same way as the second-stream catch so a replay cannot
+  // duplicate visible commentary (#520).
+  const priorEmittedOutput = Boolean(firstResult.assistantText.trim()) || firstResult.sawReasoning;
   let fallback: KiroFallbackAttempt;
   try {
     fallback = await fallbackFactory(
@@ -1268,7 +1326,7 @@ export async function* parseKiroStream(
       message: safeKiroErrorMessage({}, err instanceof Error ? err.message : String(err)),
       status: err instanceof Error && err.name === "TimeoutError" ? 504 : 502,
       errorType: "upstream_error",
-      retryable: true,
+      retryable: !priorEmittedOutput,
       usage: firstResult.usage,
     };
     return;
@@ -1282,7 +1340,7 @@ export async function* parseKiroStream(
       status: failure.status,
       errorType: failure.errorType,
       code: failure.code,
-      retryable: failure.retryable,
+      retryable: priorEmittedOutput ? false : failure.retryable,
       usage: firstResult.usage,
     };
     return;
@@ -1298,6 +1356,9 @@ export async function* parseKiroStream(
     fallback.conversationId,
     firstResult.assistantText,
     fallback.contextInputEstimate,
+    // First attempt already flushed deferred progress to the client before this fallback.
+    // A zero-output transport failure here must stay non-retryable to avoid duplicating that text.
+    priorEmittedOutput,
   );
   let secondNext = await second.next();
   while (!secondNext.done) {
@@ -1312,12 +1373,17 @@ export async function* parseKiroStream(
       mergeKiroUsage(firstResult.usage, secondResult.usage, Boolean(firstResult.assistantText))
         ?? { inputTokens, outputTokens: 0, estimated: true },
       secondResult.providerState ?? firstResult.providerState,
+      !priorEmittedOutput,
     );
     return;
   }
   if (secondResult.terminal.type === "done" || secondResult.terminal.type === "incomplete") {
     yield {
       ...secondResult.terminal,
+      // Belt-and-suspenders: never advertise a replay-safe incomplete after flushed progress.
+      ...(secondResult.terminal.type === "incomplete" && priorEmittedOutput
+        ? { retryable: false as const }
+        : {}),
       usage: mergeKiroUsage(firstResult.usage, secondResult.terminal.usage, Boolean(firstResult.assistantText)),
       providerState: secondResult.terminal.providerState ?? firstResult.providerState,
     };