diff --git a/src/adapters/kiro.ts b/src/adapters/kiro.ts index 7dde66b4e9..5e8d5211f7 100644 --- a/src/adapters/kiro.ts +++ b/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 { // `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 { 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, }; diff --git a/tests/kiro-stream.test.ts b/tests/kiro-stream.test.ts index b45086a0cc..8c6536c985 100644 --- a/tests/kiro-stream.test.ts +++ b/tests/kiro-stream.test.ts @@ -2,7 +2,7 @@ import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { createKiroAdapter } from "../src/adapters/kiro"; +import { createKiroAdapter, isRetryableKiroStreamCatchError } from "../src/adapters/kiro"; import { KIRO_COMPLETION_RETRY_MESSAGE, KIRO_COMPLETION_TOOL_NAME, @@ -637,7 +637,29 @@ describe("kiro adapter — parseStream", () => { expect(events.filter(event => event.type === "text_delta")).toEqual([ { type: "text_delta", text: "Partial progress.", phase: "commentary" }, ]); - expect(events.at(-1)).toMatchObject({ type: "error", status: 429, retryable: true }); + // Commentary was already flushed; keep status/code but block replay (#520). + expect(events.at(-1)).toMatchObject({ + type: "error", + status: 429, + code: "rate_limit_exceeded", + retryable: false, + }); + }); + + test("zero-output throttling exception remains retryable (#520)", async () => { + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(streamOf( + encodeMessage( + { ":message-type": "exception", ":exception-type": "ThrottlingException" }, + enc.encode(JSON.stringify({ message: "Too many requests." })), + ), + )))); + expect(events.some(event => event.type === "text_delta")).toBe(false); + expect(events.at(-1)).toMatchObject({ + type: "error", + status: 429, + code: "rate_limit_exceeded", + retryable: true, + }); }); test("normal Responses cancellation aborts the adapter-owned fallback without another replay", async () => { @@ -670,7 +692,8 @@ describe("kiro adapter — parseStream", () => { expect(fetches).toBe(2); expect(fallbackSignal?.aborted).toBe(true); - expect(events.at(-1)).toMatchObject({ type: "error", retryable: true }); + // First attempt already flushed reasoning; aborting the fallback must not look replay-safe. + expect(events.at(-1)).toMatchObject({ type: "error", retryable: false }); }); test("real tools never trigger the fallback and always leave endTurn false", async () => { @@ -714,7 +737,7 @@ describe("kiro adapter — parseStream", () => { test.each([ ["empty", [] as Uint8Array[], "empty_kiro_fallback"], ["reasoning-only", [eventFrame({ content: "still working" })], "reasoning_only_kiro_fallback"], - ])("%s fallback is retryable incomplete and never starts a third attempt", async (_label, fallbackFrames, reason) => { + ])("%s fallback is non-retryable incomplete after first-attempt output (#520)", async (_label, fallbackFrames, reason) => { let fetches = 0; globalThis.fetch = (async () => { fetches++; @@ -726,7 +749,7 @@ describe("kiro adapter — parseStream", () => { eventFrame({ content: "Working." }), )))); expect(fetches).toBe(1); - expect(events.at(-1)).toMatchObject({ type: "incomplete", reason, retryable: true, endTurn: false }); + expect(events.at(-1)).toMatchObject({ type: "incomplete", reason, retryable: false, endTurn: false }); expect(events.some(event => event.type === "done")).toBe(false); }); @@ -746,7 +769,7 @@ describe("kiro adapter — parseStream", () => { test.each([ ["empty answer", JSON.stringify({ answer: " " })], ["malformed JSON", "{\"answer\":"], - ])("fallback rejects %s completion as retryable incomplete", async (_label, input) => { + ])("fallback rejects %s completion as non-retryable incomplete after first-attempt output (#520)", async (_label, input) => { globalThis.fetch = (async () => new Response(streamOf( eventFrame({ name: KIRO_COMPLETION_TOOL_NAME, toolUseId: "complete-bad" }), eventFrame({ input, name: KIRO_COMPLETION_TOOL_NAME, toolUseId: "complete-bad" }), @@ -757,7 +780,12 @@ describe("kiro adapter — parseStream", () => { const events = await collectAdapterEvents(adapter.parseStream(new Response(streamOf( eventFrame({ content: "Working." }), )))); - expect(events.at(-1)).toMatchObject({ type: "incomplete", reason: "malformed_kiro_completion", retryable: true }); + expect(events.at(-1)).toMatchObject({ + type: "incomplete", + reason: "malformed_kiro_completion", + retryable: false, + endTurn: false, + }); expect(JSON.stringify(events)).not.toContain(KIRO_COMPLETION_TOOL_NAME); }); @@ -970,15 +998,162 @@ describe("kiro adapter — parseStream", () => { throw new Error("decoder failed refreshToken=rt-secret clientSecret=client-secret /Users/example/private/file.json"); }, }); - const errors: string[] = []; + const errors: Array<{ message: string; retryable?: boolean }> = []; for await (const e of createKiroAdapter(provider).parseStream(new Response(broken))) { - if (e.type === "error") errors.push(e.message); + if (e.type === "error") errors.push({ message: e.message, retryable: e.retryable }); } expect(errors).toHaveLength(1); - expect(errors[0]).toContain("Kiro upstream error"); - expect(errors[0]).not.toContain("rt-secret"); - expect(errors[0]).not.toContain("client-secret"); - expect(errors[0]).not.toContain("/Users/example"); + expect(errors[0]?.message).toContain("Kiro upstream error"); + expect(errors[0]?.message).not.toContain("rt-secret"); + expect(errors[0]?.message).not.toContain("client-secret"); + expect(errors[0]?.message).not.toContain("/Users/example"); + // No content was emitted — safe to replay (#519). + expect(errors[0]?.retryable).toBe(true); + }); + + test("socket close after heartbeats-only / zero output is retryable (#519)", async () => { + const broken = new ReadableStream({ + start(controller) { + controller.enqueue(eventFrame({ conversationId: "kiro-conv-heartbeat-only" })); + }, + pull() { + throw new Error("The socket connection was closed unexpectedly. For more information, pass verbose: true in the second argument to fetch()"); + }, + }); + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(broken))); + expect(events.at(-1)).toMatchObject({ + type: "error", + code: "kiro_stream_protocol_error", + status: 502, + retryable: true, + usage: expect.objectContaining({ outputTokens: 0 }), + }); + }); + + test("socket close after assistant text is not retryable (#519)", async () => { + const frames = [eventFrame({ content: "partial answer" })]; + let i = 0; + const broken = new ReadableStream({ + pull(controller) { + if (i < frames.length) { + controller.enqueue(frames[i++]!); + return; + } + throw new Error("The socket connection was closed unexpectedly"); + }, + }); + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(broken))); + expect(events.some(event => event.type === "text_delta")).toBe(true); + expect(events.at(-1)).toMatchObject({ + type: "error", + code: "kiro_stream_protocol_error", + retryable: false, + }); + }); + + test("eventstream truncated EOF with zero output is retryable (#520)", async () => { + expect(isRetryableKiroStreamCatchError( + new Error("eventstream: truncated message at end of stream"), + false, + )).toBe(true); + expect(isRetryableKiroStreamCatchError( + new Error("eventstream: truncated message at end of stream"), + true, + )).toBe(false); + + const broken = new ReadableStream({ + pull() { + throw new Error("eventstream: truncated message at end of stream"); + }, + }); + const events = await collectAdapterEvents(createKiroAdapter(provider).parseStream(new Response(broken))); + expect(events.at(-1)).toMatchObject({ + type: "error", + code: "kiro_stream_protocol_error", + retryable: true, + usage: expect.objectContaining({ outputTokens: 0 }), + }); + }); + + test("fallback socket close after first-attempt progress stays non-retryable (#520)", async () => { + globalThis.fetch = (async () => { + const broken = new ReadableStream({ + pull() { + throw new Error("The socket connection was closed unexpectedly"); + }, + }); + return new Response(broken); + }) as typeof fetch; + const adapter = createKiroAdapter(provider); + await adapter.buildRequest(parsedWith([{ role: "user", content: "do it" }], [bashTool])); + + const events = await collectAdapterEvents(adapter.parseStream(new Response(streamOf( + eventFrame({ content: "I am checking." }), + eventFrame({ conversationId: "returned-conversation-fallback-close" }), + )))); + + expect(events.some(event => + event.type === "text_delta" && event.text === "I am checking.", + )).toBe(true); + expect(events.at(-1)).toMatchObject({ + type: "error", + code: "kiro_stream_protocol_error", + retryable: false, + }); + }); + + test("fallback setup throw after first-attempt commentary stays non-retryable (#520)", async () => { + globalThis.fetch = (async () => { + throw new Error("fetch failed refreshToken=rt-secret-fallback"); + }) as typeof fetch; + const adapter = createKiroAdapter(provider); + await adapter.buildRequest(parsedWith([{ role: "user", content: "do it" }], [bashTool])); + + const events = await collectAdapterEvents(adapter.parseStream(new Response(streamOf( + eventFrame({ content: "I am checking." }), + eventFrame({ conversationId: "returned-conversation-fallback-throw" }), + )))); + + expect(events.some(event => + event.type === "text_delta" && event.text === "I am checking.", + )).toBe(true); + const terminal = events.at(-1); + expect(terminal).toMatchObject({ + type: "error", + status: 502, + errorType: "upstream_error", + retryable: false, + }); + if (terminal?.type === "error") { + expect(terminal.message).toContain("Kiro upstream error"); + expect(terminal.message).not.toContain("rt-secret-fallback"); + expect(terminal.usage).toEqual(expect.objectContaining({})); + } + }); + + test("retryable fallback HTTP after first-attempt commentary stays non-retryable (#520)", async () => { + globalThis.fetch = (async () => new Response("{\"message\":\"temporarily unavailable\"}", { + status: 503, + headers: { "content-type": "application/json" }, + })) as typeof fetch; + const adapter = createKiroAdapter(provider); + await adapter.buildRequest(parsedWith([{ role: "user", content: "do it" }], [bashTool])); + + const events = await collectAdapterEvents(adapter.parseStream(new Response(streamOf( + eventFrame({ content: "I am checking." }), + eventFrame({ conversationId: "returned-conversation-fallback-http" }), + )))); + + expect(events.some(event => + event.type === "text_delta" && event.text === "I am checking.", + )).toBe(true); + expect(events.at(-1)).toMatchObject({ + type: "error", + status: 503, + code: "server_is_overloaded", + retryable: false, + usage: expect.objectContaining({}), + }); }); test("leading thinking block is emitted as raw reasoning, not visible text", async () => {