Merge pull request #546 from ndycode/claude/audit-30-stream-stall-fix

fix(request): stream-stall failover, private-header prefix block, pinned-index message

Author: ndycode

lib/request/rate-limit-decision.ts

@@ -195,9 +195,14 @@ export function buildPinnedUnavailableErrorBody(
 			? accountSkipReasons.get(normalizedPinnedIndex) ?? null
 			: null;
 	const reasonSuffix = skipReason ? ` (${skipReason})` : "";
-	const displayIndex = (normalizedPinnedIndex ?? 0) + 1;
+	// On the desync path the pin index is unknown (null); claiming "account 1"
+	// there would contradict the machine-readable pinnedAccountIndex: null.
+	const accountPhrase =
+		normalizedPinnedIndex === null
+			? "The pinned account"
+			: `Pinned account ${normalizedPinnedIndex + 1}`;
 	return {
-		message: `Pinned account ${displayIndex} is currently unavailable${reasonSuffix}; run \`codex-multi-auth status\` for details, or \`codex-multi-auth unpin\` to allow rotation.`,
+		message: `${accountPhrase} is currently unavailable${reasonSuffix}; run \`codex-multi-auth status\` for details, or \`codex-multi-auth unpin\` to allow rotation.`,
 		code: "codex_pinned_account_unavailable",
 		pinnedAccountIndex: normalizedPinnedIndex,
 		reason: skipReason,

lib/request/stream-failover-runtime.ts

@@ -13,12 +13,11 @@ export const HOP_BY_HOP_HEADERS = new Set([
 	"transfer-encoding",
 	"upgrade",
 ]);
-const PRIVATE_CLIENT_RESPONSE_HEADERS = new Set([
-	"x-codex-multi-auth-account-index",
-	"x-codex-multi-auth-account-label",
-	"x-codex-multi-auth-account-email",
-	"x-codex-multi-auth-account-id",
-]);
+// Any header under this prefix carries account-identifying rotation metadata
+// (index/label/email/id today) and must never reach the client; matching by
+// prefix means a future header added under it is blocked by default instead
+// of leaking until someone remembers to extend an allowlist.
+const PRIVATE_CLIENT_RESPONSE_HEADER_PREFIX = "x-codex-multi-auth-account-";
 const DECODED_UPSTREAM_RESPONSE_HEADERS = new Set([
 	// Node fetch returns decoded bytes while preserving the upstream encoding header.
 	"content-encoding",
@@ -29,7 +28,7 @@ export function responseHeadersForClient(upstreamHeaders: Headers): Record<strin
 	for (const [key, value] of upstreamHeaders.entries()) {
 		const normalizedKey = key.toLowerCase();
 		if (HOP_BY_HOP_HEADERS.has(normalizedKey)) continue;
-		if (PRIVATE_CLIENT_RESPONSE_HEADERS.has(normalizedKey)) continue;
+		if (normalizedKey.startsWith(PRIVATE_CLIENT_RESPONSE_HEADER_PREFIX)) continue;
 		if (DECODED_UPSTREAM_RESPONSE_HEADERS.has(normalizedKey)) continue;
 		headers[key] = value;
 	}
@@ -48,8 +47,12 @@ export async function withTimeout<T>(
 			promise,
 			new Promise<T>((_resolve, reject) => {
 				timeout = setTimeout(() => {
-					onTimeout();
+					// Reject BEFORE onTimeout: onTimeout side effects (e.g. cancelling a
+					// stream reader) can settle `promise` with a clean value, and a
+					// settlement enqueued ahead of this rejection would win the race —
+					// turning a stall into a silent success.
 					reject(new Error(message));
+					onTimeout();
 				}, Math.max(1, timeoutMs));
 			}),
 		]);

test/issue-474-pin-honored.test.ts

@@ -719,7 +719,10 @@ describe("issue #474 — manual pin honored by runtime proxy", () => {
 			);
 			expect(body.pinnedAccountIndex).toBeNull();
 			expect(body.reason).toBeNull();
-			expect(body.message).toContain("Pinned account 1");
+			// The desync path must not fabricate an index that contradicts the
+			// machine-readable pinnedAccountIndex: null.
+			expect(body.message).toContain("The pinned account is currently unavailable;");
+			expect(body.message).not.toContain("Pinned account 1");
 		});
 
 		it("mirrors the full accountSkipReasons map even when the pinned entry is unknown", () => {

test/rate-limit-decision.test.ts

@@ -275,11 +275,14 @@ describe("buildPinnedUnavailableErrorBody", () => {
 		expect(body.account_skip_reasons).toEqual({ "0": "rate-limited", "1": "disabled" });
 	});
 
-	it("omits the parenthetical and reason on the null-index desync path", () => {
+	it("omits the index, parenthetical, and reason on the null-index desync path", () => {
+		// Regression: the message used to render the null index as "Pinned
+		// account 1", contradicting the machine-readable pinnedAccountIndex.
 		const body = buildPinnedUnavailableErrorBody(null, new Map());
 		expect(body.pinnedAccountIndex).toBeNull();
 		expect(body.reason).toBeNull();
-		expect(body.message).toContain("Pinned account 1 is currently unavailable;");
+		expect(body.message).toContain("The pinned account is currently unavailable;");
+		expect(body.message).not.toContain("Pinned account 1");
 		expect(body.message).not.toContain("(");
 		expect(body.account_skip_reasons).toEqual({});
 	});

test/stream-failover-runtime.test.ts

@@ -96,6 +96,19 @@ describe("responseHeadersForClient", () => {
 		});
 	});
 
+	it("blocks any header under the private account prefix, not just known names", () => {
+		// Regression: the filter used to be an exact-name allowlist, so a future
+		// x-codex-multi-auth-account-* header would have leaked by default.
+		const upstream = new Headers({
+			"content-type": "application/json",
+			"x-codex-multi-auth-account-plan": "pro",
+			"X-Codex-Multi-Auth-Account-Future-Field": "secret",
+		});
+		expect(responseHeadersForClient(upstream)).toEqual({
+			"content-type": "application/json",
+		});
+	});
+
 	it("covers every hop-by-hop header in the exported set", () => {
 		const upstream = new Headers();
 		for (const name of HOP_BY_HOP_HEADERS) {
@@ -232,14 +245,12 @@ describe("forwardStreamingResponse", () => {
 		expect(res.chunks).toEqual([]);
 	});
 
-	it("currently ends a stalled stream cleanly instead of failing it", async () => {
-		// Pins a BUG (not the intended design): on a stall, withTimeout's
-		// onTimeout cancels the reader, which settles the pending read() with
-		// {done: true} BEFORE the rejection lands, so Promise.race resolves.
-		// The stall is reported as a clean end: truncated body, res.end(), no
-		// lastError, and onStreamError (the failover hook) never fires. The
-		// stall branch of the catch block is unreachable today. Fixed in the
-		// stacked follow-up PR, which flips these expectations.
+	it("records the stall error, destroys the response, and reports failure on timeout", async () => {
+		// Regression: withTimeout used to invoke onTimeout (which cancels the
+		// reader and thereby settles the pending read() with {done: true})
+		// before rejecting, so the race resolved and a stalled upstream was
+		// forwarded as a clean end-of-stream — truncated body, res.end(), no
+		// lastError, and onStreamError (the failover hook) never fired.
 		const res = new FakeServerResponse();
 		const status = createStatus();
 		const upstream = new Response(streamOf(
@@ -250,13 +261,14 @@ describe("forwardStreamingResponse", () => {
 		const onStreamError = vi.fn();
 		await expect(
 			forwardStreamingResponse(upstream, res.asServerResponse(), status, onStreamError, 25),
-		).resolves.toBe(true);
+		).resolves.toBe(false);
 
-		expect(onStreamError).not.toHaveBeenCalled();
-		expect(status.lastError).toBeNull();
-		expect(res.destroyed).toBe(false);
+		expect(onStreamError).toHaveBeenCalledTimes(1);
+		expect(status.lastError).toBe("upstream stream stalled after 25ms");
+		expect(res.destroyed).toBe(true);
+		expect(res.destroyError).toBeInstanceOf(Error);
 		expect(Buffer.concat(res.chunks).toString("utf8")).toBe("data: a\n\n");
-		expect(res.ended).toBe(true);
+		expect(res.ended).toBe(false);
 	});
 
 	it("fails the stream when the upstream read rejects mid-stream", async () => {