fix: stop stream failover replay after output

Author: ndycode

lib/request/stream-failover.ts

@@ -105,7 +105,11 @@ async function readChunkWithSoftHardTimeout(
  *
  * The returned Response streams bytes from the initialResponse body and, when the stream stalls or errors, will attempt up to `maxFailovers` failovers by calling `getFallbackResponse(attempt, emittedBytes)`. On each successful failover a textual marker is injected into the stream identifying the failover attempt (and `requestInstanceId` when provided). The function performs best-effort cleanup of underlying readers and enforces soft/hard read timeouts as configured via `options`.
  *
- * Concurrency assumptions: the implementation expects a single consumer reading the returned Response body; callers must not concurrently read the same stream body from multiple consumers. Filesystem/platform note: behavior is platform-agnostic; no filesystem access is performed (Windows-specific filesystem semantics do not apply). Token redaction: any request identifiers embedded in the injected marker are limited to the normalized `requestInstanceId` (trimmed and truncated to 64 chars) to avoid leaking long tokens.
+	 * Concurrency assumptions: the implementation expects a single consumer reading the returned Response body; callers must not concurrently read the same stream body from multiple consumers. Filesystem/platform note: behavior is platform-agnostic; no filesystem access is performed (Windows-specific filesystem semantics do not apply). Token redaction: any request identifiers embedded in the injected marker are limited to the normalized `requestInstanceId` (trimmed and truncated to 64 chars) to avoid leaking long tokens.
+	 *
+	 * Failover safety: once the wrapper has already emitted bytes from the primary stream,
+	 * it will stop attempting fallback replays. Reissuing the upstream request after partial
+	 * output can duplicate streamed text and any side-effectful tool activity.
  *
  * @param initialResponse - The original Response whose body will be streamed and monitored for stalls/errors.
  * @param getFallbackResponse - Async function invoked for each failover attempt with the 1-based attempt number and total emitted bytes; should return a Response with a streaming body to switch to, or `null`/a Response without a body to indicate no fallback.
@@ -162,6 +166,9 @@ export function withStreamingFailover(
 				if (failoverAttempt >= maxFailovers) {
 					return false;
 				}
+				if (emittedBytes > 0) {
+					return false;
+				}
 				failoverAttempt += 1;
 				const fallback = await getFallbackResponse(failoverAttempt, emittedBytes);
 				if (!fallback?.body) {

test/stream-failover.test.ts

@@ -18,6 +18,21 @@ function makeStallingResponse(): Response {
 	);
 }
 
+function makeIdleResponse(): Response {
+	return new Response(
+		new ReadableStream<Uint8Array>({
+			start() {
+				// Intentionally idle until timeout.
+			},
+		}),
+		{
+			headers: {
+				"content-type": "text/event-stream",
+			},
+		},
+	);
+}
+
 function makeSseResponse(payload: string): Response {
 	return new Response(
 		new ReadableStream<Uint8Array>({
@@ -52,15 +67,14 @@ describe("stream failover", () => {
 	it("switches to fallback stream when primary stalls", async () => {
 		vi.useFakeTimers();
 		const fallback = vi.fn(async () => makeSseResponse("data: second\n\n"));
-		const response = withStreamingFailover(makeStallingResponse(), fallback, {
+		const response = withStreamingFailover(makeIdleResponse(), fallback, {
 			maxFailovers: 1,
 			stallTimeoutMs: 10,
 		});
 
 		const textPromise = response.text();
 		await vi.advanceTimersByTimeAsync(1_200);
 		const text = await textPromise;
-		expect(text).toContain("data: first");
 		expect(text).toContain("codex-multi-auth failover 1");
 		expect(text).toContain("data: second");
 		expect(fallback).toHaveBeenCalledTimes(1);
@@ -69,7 +83,7 @@ describe("stream failover", () => {
 	it("includes request id marker when provided", async () => {
 		vi.useFakeTimers();
 		const response = withStreamingFailover(
-			makeStallingResponse(),
+			makeIdleResponse(),
 			async () => makeSseResponse("data: fallback\n\n"),
 			{
 				maxFailovers: 1,
@@ -87,7 +101,7 @@ describe("stream failover", () => {
 	it("errors when fallback is unavailable", async () => {
 		vi.useFakeTimers();
 		const response = withStreamingFailover(
-			makeStallingResponse(),
+			makeIdleResponse(),
 			async () => null,
 			{ maxFailovers: 1, stallTimeoutMs: 10 },
 		);
@@ -101,7 +115,7 @@ describe("stream failover", () => {
 	it("propagates fallback provider exceptions deterministically", async () => {
 		vi.useFakeTimers();
 		const response = withStreamingFailover(
-			makeStallingResponse(),
+			makeIdleResponse(),
 			async () => {
 				throw new Error("fallback exploded");
 			},
@@ -114,7 +128,7 @@ describe("stream failover", () => {
 		await assertion;
 	});
 
-	it("calls fallback exactly once when read-error and timeout race", async () => {
+	it("does not trigger fallback when read-error and timeout race after bytes emitted", async () => {
 		vi.useFakeTimers();
 		const raceResponse = new Response(
 			new ReadableStream<Uint8Array>({
@@ -139,13 +153,27 @@ describe("stream failover", () => {
 		});
 
 		const textPromise = response.text();
+		const assertion = expect(textPromise).rejects.toThrow("primary read failure");
 		await vi.advanceTimersByTimeAsync(1_200);
-		const text = await textPromise;
+		await assertion;
 
-		expect(fallback).toHaveBeenCalledTimes(1);
-		expect((text.match(/codex-multi-auth failover 1/g) ?? []).length).toBe(1);
-		expect(text).toContain("data: first");
-		expect(text).toContain("data: fallback");
+		expect(fallback).not.toHaveBeenCalled();
+	});
+
+	it("does not replay after bytes have already been emitted", async () => {
+		vi.useFakeTimers();
+		const fallback = vi.fn(async () => makeSseResponse("data: fallback\n\n"));
+		const response = withStreamingFailover(makeStallingResponse(), fallback, {
+			maxFailovers: 1,
+			stallTimeoutMs: 10,
+		});
+
+		const textPromise = response.text();
+		const assertion = expect(textPromise).rejects.toThrow("SSE stream stalled");
+		await vi.advanceTimersByTimeAsync(1_200);
+		await assertion;
+
+		expect(fallback).not.toHaveBeenCalled();
 	});
 
 	it("releases underlying reader when wrapped stream is cancelled", async () => {