fix(ai): retry pre-output stream failures

.agents/skills/senpi-qa/scripts/mock-loop-stream-retry.mjs

@@ -0,0 +1,167 @@
+/**
+ * Channel 3 proof for an OpenAI-compatible SSE error before the first chunk.
+ *
+ * The first request returns a protocol-level `event: error` carrying the exact
+ * DigitalOcean message. The second request returns one successful completion.
+ * The real source CLI must retry inside the provider adapter and print exactly
+ * one final marker.
+ */
+
+import { createServer } from "node:http";
+import { writeFileSync } from "node:fs";
+import { join } from "node:path";
+import {
+	createChecks,
+	evidenceDir,
+	guardRealAuth,
+	installCleanupHooks,
+	makeSandbox,
+	runCli,
+} from "./lib/common.mjs";
+import {
+	API_PRESETS,
+	checkRealAuthUnchanged,
+	hermeticEnv,
+	writeMockModelsJson,
+} from "./lib/mock-loop-support.mjs";
+
+const ERROR_MESSAGE = "Upstream error from DigitalOcean: stream failed";
+const FINAL_MARKER = "SENPI-QA-DIGITALOCEAN-STREAM-RECOVERED-7a4f";
+const EVIDENCE_SLUG = "digitalocean-stream-retry";
+
+function startServer() {
+	const requests = [];
+	let attempts = 0;
+	const server = createServer((request, response) => {
+		const chunks = [];
+		request.on("data", (chunk) => chunks.push(chunk));
+		request.on("end", () => {
+			attempts++;
+			requests.push({
+				attempt: attempts,
+				url: request.url,
+				body: Buffer.concat(chunks).toString("utf8"),
+			});
+			response.writeHead(200, {
+				"content-type": "text/event-stream",
+				"cache-control": "no-cache",
+				connection: "keep-alive",
+			});
+			if (attempts === 1) {
+				response.end(`event: error\ndata: ${JSON.stringify({ error: { message: ERROR_MESSAGE, type: "server_error" } })}\n\n`);
+				return;
+			}
+			const base = {
+				id: "chatcmpl-digitalocean-retry",
+				object: "chat.completion.chunk",
+				created: 0,
+				model: API_PRESETS["openai-completions"].modelId,
+			};
+			const send = (delta, finishReason = null) => {
+				response.write(
+					`data: ${JSON.stringify({ ...base, choices: [{ index: 0, delta, finish_reason: finishReason }] })}\n\n`,
+				);
+			};
+			send({ role: "assistant", content: "" });
+			send({ content: FINAL_MARKER });
+			send({}, "stop");
+			response.write(
+				`data: ${JSON.stringify({
+					...base,
+					choices: [],
+					usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 },
+				})}\n\n`,
+			);
+			response.end("data: [DONE]\n\n");
+		});
+	});
+	return new Promise((resolve, reject) => {
+		server.once("error", reject);
+		server.listen(0, "127.0.0.1", () => {
+			const address = server.address();
+			if (!address || typeof address === "string") {
+				reject(new Error("Failed to resolve fake server port"));
+				return;
+			}
+			const origin = `http://127.0.0.1:${address.port}`;
+			resolve({
+				url: `${origin}/v1`,
+				origin,
+				port: address.port,
+				requests,
+				stop: () => new Promise((done) => server.close(done)),
+			});
+		});
+	});
+}
+
+async function main() {
+	installCleanupHooks();
+	const checks = createChecks("mock-loop-stream-retry.mjs");
+	const guard = guardRealAuth();
+	const box = makeSandbox("mock-loop-digitalocean-stream-retry");
+	const server = await startServer();
+	try {
+		writeMockModelsJson(box.agentDir, server, "openai-completions", {}, {
+			retry: {
+				enabled: false,
+				maxRetries: 0,
+				baseDelayMs: 0,
+				provider: { maxRetries: 1, maxRetryDelayMs: 60000 },
+			},
+		});
+		const preset = API_PRESETS["openai-completions"];
+		const result = await runCli(
+			[
+				"--provider",
+				preset.provider,
+				"--model",
+				preset.modelId,
+				"--no-context-files",
+				"--no-extensions",
+				"--print",
+				`Return ${FINAL_MARKER} after the transient stream error.`,
+			],
+			{ env: hermeticEnv(box.env), cwd: box.cwd, timeoutMs: 60000 },
+		);
+		const combined = `${result.stdout}\n${result.stderr}`;
+		const finalCount = combined.split(FINAL_MARKER).length - 1;
+		const pass = result.code === 0 && !result.timedOut && server.requests.length === 2 && finalCount === 1;
+		checks.ok(
+			"real CLI retries the literal pre-output stream failure exactly once",
+			pass,
+			`code=${result.code} requests=${server.requests.length} finalCount=${finalCount}`,
+		);
+		checkRealAuthUnchanged(checks, guard);
+		const dir = evidenceDir(EVIDENCE_SLUG);
+		writeFileSync(join(dir, "mock-loop-stream-retry-stdout.txt"), result.stdout);
+		writeFileSync(join(dir, "mock-loop-stream-retry-stderr.txt"), result.stderr);
+		writeFileSync(
+			join(dir, "mock-loop-stream-retry-summary.json"),
+			`${JSON.stringify(
+				{
+					command: "node .agents/skills/senpi-qa/scripts/mock-loop-stream-retry.mjs",
+					literalError: ERROR_MESSAGE,
+					exitCode: result.code,
+					timedOut: result.timedOut,
+					serverPort: server.port,
+					sandboxDir: box.dir,
+					requests: server.requests.length,
+					finalMarkerCount: finalCount,
+					pass,
+				},
+				null,
+				2,
+			)}\n`,
+		);
+		process.exitCode = checks.finish() ? 0 : 1;
+	} finally {
+		await server.stop();
+		box.cleanup();
+	}
+}
+
+main().catch((error) => {
+	process.stderr.write(`${error instanceof Error ? error.stack : String(error)}\n`);
+	process.exit(1);
+});

packages/ai/src/api/openai-completions.ts

@@ -40,7 +40,7 @@ import { shortHash } from "../utils/hash.ts";
 import { headersToRecord } from "../utils/headers.ts";
 import { parseStreamingJson } from "../utils/json-parse.ts";
 import { getProviderEnvValue } from "../utils/provider-env.ts";
-import { retryProviderRequest } from "../utils/provider-retry.ts";
+import { retryProviderStreamRequest } from "../utils/provider-retry.ts";
 import { sanitizeSurrogates } from "../utils/sanitize-unicode.ts";
 import { isForcedToolChoiceUnsupportedError, omitToolChoiceParam } from "../utils/tool-choice-fallback.ts";
 import {
@@ -404,12 +404,17 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio
 					throw error;
 				}
 			};
-			const { data: openaiStream, response } = await retryProviderRequest(createRequest, {
-				maxRetries: options?.maxRetries,
-				maxRetryDelayMs: options?.maxRetryDelayMs,
-				signal: options?.signal,
-			});
-			await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
+			const { stream: openaiStream } = await retryProviderStreamRequest(
+				async () => {
+					const { data, response } = await createRequest();
+					await options?.onResponse?.(
+						{ status: response.status, headers: headersToRecord(response.headers) },
+						model,
+					);
+					return { stream: data, metadata: response };
+				},
+				{ maxRetries: options?.maxRetries, maxRetryDelayMs: options?.maxRetryDelayMs, signal: options?.signal },
+			);
 			stream.push({ type: "start", partial: output });
 
 			interface StreamingToolCallBlock extends ToolCall {

packages/ai/src/changes.md

@@ -1,5 +1,26 @@
 # AI Source Changes
 
+## 2026-07-28 - Retry OpenAI-compatible stream failures before the first chunk
+
+### What changed and why
+
+- `utils/provider-retry.ts` now prefetches the first SDK stream result inside the existing bounded, abortable provider
+  retry policy. A retry creates a fresh request only when stream consumption fails before any wire chunk can reach
+  the public event stream.
+- `api/openai-completions.ts` uses that prefetch wrapper for OpenAI-compatible providers. Once the first chunk exists,
+  the stream is replayed exactly once and any later failure remains terminal, preventing duplicated text or tool
+  effects.
+- The exact property-less gateway error `Upstream error from DigitalOcean: stream failed` is recognized as transient;
+  arbitrary property-less errors remain non-retryable.
+- `../test/openai-completions-retry.test.ts` covers recovery, retry exhaustion, non-retryable failures, and the
+  post-first-chunk no-retry boundary. The isolated mock-loop driver
+  `.agents/skills/senpi-qa/scripts/mock-loop-stream-retry.mjs` proves the same behavior through the real source CLI.
+
+### Expected merge conflict zones
+
+- LOW: the request creation/retry block in `api/openai-completions.ts`.
+- LOW: shared provider retry classification and stream-prefetch helper in `utils/provider-retry.ts`.
+
 ## 2026-07-27 - Codex reasoning summary null omits the field instead of sending "off"
 
 ### What changed and why

packages/ai/src/utils/provider-retry.ts

@@ -1,4 +1,5 @@
 const DEFAULT_MAX_RETRY_DELAY_MS = 60_000;
+const DIGITALOCEAN_STREAM_FAILURE_MESSAGE = "Upstream error from DigitalOcean: stream failed";
 
 interface ProviderRetryOptions {
 	maxRetries?: number;
@@ -12,7 +13,9 @@ interface ProviderError extends Error {
 }
 
 function isProviderError(error: unknown): error is ProviderError {
-	if (!(error instanceof Error) || !("status" in error) || !("headers" in error)) return false;
+	if (!(error instanceof Error)) return false;
+	if (error.message === DIGITALOCEAN_STREAM_FAILURE_MESSAGE) return true;
+	if (!("status" in error) || !("headers" in error)) return false;
 	return (
 		(error.status === undefined || typeof error.status === "number") &&
 		(error.headers === undefined || error.headers instanceof Headers)
@@ -123,3 +126,53 @@ export async function retryProviderRequest<T>(
 		}
 	}
 }
+
+interface ProviderStreamAttempt<TChunk, TMetadata> {
+	stream: AsyncIterable<TChunk>;
+	metadata: TMetadata;
+}
+
+async function* replayPrefetchedStream<TChunk>(
+	first: IteratorResult<TChunk>,
+	iterator: AsyncIterator<TChunk>,
+): AsyncGenerator<TChunk> {
+	if (first.done) return;
+	let completed = false;
+	let providerFailed = false;
+	try {
+		yield first.value;
+		for (;;) {
+			let next: IteratorResult<TChunk>;
+			try {
+				next = await iterator.next();
+			} catch (error) {
+				providerFailed = true;
+				throw error;
+			}
+			if (next.done) {
+				completed = true;
+				return;
+			}
+			yield next.value;
+		}
+	} finally {
+		if (!completed && !providerFailed) {
+			await iterator.return?.();
+		}
+	}
+}
+
+export async function retryProviderStreamRequest<TChunk, TMetadata>(
+	request: () => Promise<ProviderStreamAttempt<TChunk, TMetadata>>,
+	options: ProviderRetryOptions = {},
+): Promise<ProviderStreamAttempt<TChunk, TMetadata>> {
+	return retryProviderRequest(async () => {
+		const attempt = await request();
+		const iterator = attempt.stream[Symbol.asyncIterator]();
+		const first = await iterator.next();
+		return {
+			stream: replayPrefetchedStream(first, iterator),
+			metadata: attempt.metadata,
+		};
+	}, options);
+}