Skip to content

Commit fabdaf7

Browse files
code-yeongyucode-yeongyu
authored
Merge pull request #421 from code-yeongyu/fix/digitalocean-stream-retry
fix(ai): retry pre-output stream failures
2 parents 9c65526 + 6012db5 commit fabdaf7

6 files changed

Lines changed: 369 additions & 10 deletions

File tree

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

Lines changed: 167 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
/**
2+
* Channel 3 proof for an OpenAI-compatible SSE error before the first chunk.
3+
*
4+
* The first request returns a protocol-level `event: error` carrying the exact
5+
* DigitalOcean message. The second request returns one successful completion.
6+
* The real source CLI must retry inside the provider adapter and print exactly
7+
* one final marker.
8+
*/
9+
10+
import { createServer } from "node:http";
11+
import { writeFileSync } from "node:fs";
12+
import { join } from "node:path";
13+
import {
14+
createChecks,
15+
evidenceDir,
16+
guardRealAuth,
17+
installCleanupHooks,
18+
makeSandbox,
19+
runCli,
20+
} from "./lib/common.mjs";
21+
import {
22+
API_PRESETS,
23+
checkRealAuthUnchanged,
24+
hermeticEnv,
25+
writeMockModelsJson,
26+
} from "./lib/mock-loop-support.mjs";
27+
28+
const ERROR_MESSAGE = "Upstream error from DigitalOcean: stream failed";
29+
const FINAL_MARKER = "SENPI-QA-DIGITALOCEAN-STREAM-RECOVERED-7a4f";
30+
const EVIDENCE_SLUG = "digitalocean-stream-retry";
31+
32+
function startServer() {
33+
const requests = [];
34+
let attempts = 0;
35+
const server = createServer((request, response) => {
36+
const chunks = [];
37+
request.on("data", (chunk) => chunks.push(chunk));
38+
request.on("end", () => {
39+
attempts++;
40+
requests.push({
41+
attempt: attempts,
42+
url: request.url,
43+
body: Buffer.concat(chunks).toString("utf8"),
44+
});
45+
response.writeHead(200, {
46+
"content-type": "text/event-stream",
47+
"cache-control": "no-cache",
48+
connection: "keep-alive",
49+
});
50+
if (attempts === 1) {
51+
response.end(`event: error\ndata: ${JSON.stringify({ error: { message: ERROR_MESSAGE, type: "server_error" } })}\n\n`);
52+
return;
53+
}
54+
const base = {
55+
id: "chatcmpl-digitalocean-retry",
56+
object: "chat.completion.chunk",
57+
created: 0,
58+
model: API_PRESETS["openai-completions"].modelId,
59+
};
60+
const send = (delta, finishReason = null) => {
61+
response.write(
62+
`data: ${JSON.stringify({ ...base, choices: [{ index: 0, delta, finish_reason: finishReason }] })}\n\n`,
63+
);
64+
};
65+
send({ role: "assistant", content: "" });
66+
send({ content: FINAL_MARKER });
67+
send({}, "stop");
68+
response.write(
69+
`data: ${JSON.stringify({
70+
...base,
71+
choices: [],
72+
usage: { prompt_tokens: 1, completion_tokens: 1, total_tokens: 2 },
73+
})}\n\n`,
74+
);
75+
response.end("data: [DONE]\n\n");
76+
});
77+
});
78+
return new Promise((resolve, reject) => {
79+
server.once("error", reject);
80+
server.listen(0, "127.0.0.1", () => {
81+
const address = server.address();
82+
if (!address || typeof address === "string") {
83+
reject(new Error("Failed to resolve fake server port"));
84+
return;
85+
}
86+
const origin = `http://127.0.0.1:${address.port}`;
87+
resolve({
88+
url: `${origin}/v1`,
89+
origin,
90+
port: address.port,
91+
requests,
92+
stop: () => new Promise((done) => server.close(done)),
93+
});
94+
});
95+
});
96+
}
97+
98+
async function main() {
99+
installCleanupHooks();
100+
const checks = createChecks("mock-loop-stream-retry.mjs");
101+
const guard = guardRealAuth();
102+
const box = makeSandbox("mock-loop-digitalocean-stream-retry");
103+
const server = await startServer();
104+
try {
105+
writeMockModelsJson(box.agentDir, server, "openai-completions", {}, {
106+
retry: {
107+
enabled: false,
108+
maxRetries: 0,
109+
baseDelayMs: 0,
110+
provider: { maxRetries: 1, maxRetryDelayMs: 60000 },
111+
},
112+
});
113+
const preset = API_PRESETS["openai-completions"];
114+
const result = await runCli(
115+
[
116+
"--provider",
117+
preset.provider,
118+
"--model",
119+
preset.modelId,
120+
"--no-context-files",
121+
"--no-extensions",
122+
"--print",
123+
`Return ${FINAL_MARKER} after the transient stream error.`,
124+
],
125+
{ env: hermeticEnv(box.env), cwd: box.cwd, timeoutMs: 60000 },
126+
);
127+
const combined = `${result.stdout}\n${result.stderr}`;
128+
const finalCount = combined.split(FINAL_MARKER).length - 1;
129+
const pass = result.code === 0 && !result.timedOut && server.requests.length === 2 && finalCount === 1;
130+
checks.ok(
131+
"real CLI retries the literal pre-output stream failure exactly once",
132+
pass,
133+
`code=${result.code} requests=${server.requests.length} finalCount=${finalCount}`,
134+
);
135+
checkRealAuthUnchanged(checks, guard);
136+
const dir = evidenceDir(EVIDENCE_SLUG);
137+
writeFileSync(join(dir, "mock-loop-stream-retry-stdout.txt"), result.stdout);
138+
writeFileSync(join(dir, "mock-loop-stream-retry-stderr.txt"), result.stderr);
139+
writeFileSync(
140+
join(dir, "mock-loop-stream-retry-summary.json"),
141+
`${JSON.stringify(
142+
{
143+
command: "node .agents/skills/senpi-qa/scripts/mock-loop-stream-retry.mjs",
144+
literalError: ERROR_MESSAGE,
145+
exitCode: result.code,
146+
timedOut: result.timedOut,
147+
serverPort: server.port,
148+
sandboxDir: box.dir,
149+
requests: server.requests.length,
150+
finalMarkerCount: finalCount,
151+
pass,
152+
},
153+
null,
154+
2,
155+
)}\n`,
156+
);
157+
process.exitCode = checks.finish() ? 0 : 1;
158+
} finally {
159+
await server.stop();
160+
box.cleanup();
161+
}
162+
}
163+
164+
main().catch((error) => {
165+
process.stderr.write(`${error instanceof Error ? error.stack : String(error)}\n`);
166+
process.exit(1);
167+
});

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

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ import { shortHash } from "../utils/hash.ts";
4040
import { headersToRecord } from "../utils/headers.ts";
4141
import { parseStreamingJson } from "../utils/json-parse.ts";
4242
import { getProviderEnvValue } from "../utils/provider-env.ts";
43-
import { retryProviderRequest } from "../utils/provider-retry.ts";
43+
import { retryProviderStreamRequest } from "../utils/provider-retry.ts";
4444
import { sanitizeSurrogates } from "../utils/sanitize-unicode.ts";
4545
import { isForcedToolChoiceUnsupportedError, omitToolChoiceParam } from "../utils/tool-choice-fallback.ts";
4646
import {
@@ -404,12 +404,17 @@ export const stream: StreamFunction<"openai-completions", OpenAICompletionsOptio
404404
throw error;
405405
}
406406
};
407-
const { data: openaiStream, response } = await retryProviderRequest(createRequest, {
408-
maxRetries: options?.maxRetries,
409-
maxRetryDelayMs: options?.maxRetryDelayMs,
410-
signal: options?.signal,
411-
});
412-
await options?.onResponse?.({ status: response.status, headers: headersToRecord(response.headers) }, model);
407+
const { stream: openaiStream } = await retryProviderStreamRequest(
408+
async () => {
409+
const { data, response } = await createRequest();
410+
await options?.onResponse?.(
411+
{ status: response.status, headers: headersToRecord(response.headers) },
412+
model,
413+
);
414+
return { stream: data, metadata: response };
415+
},
416+
{ maxRetries: options?.maxRetries, maxRetryDelayMs: options?.maxRetryDelayMs, signal: options?.signal },
417+
);
413418
stream.push({ type: "start", partial: output });
414419

415420
interface StreamingToolCallBlock extends ToolCall {

‎packages/ai/src/changes.md‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,26 @@
11
# AI Source Changes
22

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

526
### What changed and why

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

Lines changed: 54 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
const DEFAULT_MAX_RETRY_DELAY_MS = 60_000;
2+
const DIGITALOCEAN_STREAM_FAILURE_MESSAGE = "Upstream error from DigitalOcean: stream failed";
23

34
interface ProviderRetryOptions {
45
maxRetries?: number;
@@ -12,7 +13,9 @@ interface ProviderError extends Error {
1213
}
1314

1415
function isProviderError(error: unknown): error is ProviderError {
15-
if (!(error instanceof Error) || !("status" in error) || !("headers" in error)) return false;
16+
if (!(error instanceof Error)) return false;
17+
if (error.message === DIGITALOCEAN_STREAM_FAILURE_MESSAGE) return true;
18+
if (!("status" in error) || !("headers" in error)) return false;
1619
return (
1720
(error.status === undefined || typeof error.status === "number") &&
1821
(error.headers === undefined || error.headers instanceof Headers)
@@ -123,3 +126,53 @@ export async function retryProviderRequest<T>(
123126
}
124127
}
125128
}
129+
130+
interface ProviderStreamAttempt<TChunk, TMetadata> {
131+
stream: AsyncIterable<TChunk>;
132+
metadata: TMetadata;
133+
}
134+
135+
async function* replayPrefetchedStream<TChunk>(
136+
first: IteratorResult<TChunk>,
137+
iterator: AsyncIterator<TChunk>,
138+
): AsyncGenerator<TChunk> {
139+
if (first.done) return;
140+
let completed = false;
141+
let providerFailed = false;
142+
try {
143+
yield first.value;
144+
for (;;) {
145+
let next: IteratorResult<TChunk>;
146+
try {
147+
next = await iterator.next();
148+
} catch (error) {
149+
providerFailed = true;
150+
throw error;
151+
}
152+
if (next.done) {
153+
completed = true;
154+
return;
155+
}
156+
yield next.value;
157+
}
158+
} finally {
159+
if (!completed && !providerFailed) {
160+
await iterator.return?.();
161+
}
162+
}
163+
}
164+
165+
export async function retryProviderStreamRequest<TChunk, TMetadata>(
166+
request: () => Promise<ProviderStreamAttempt<TChunk, TMetadata>>,
167+
options: ProviderRetryOptions = {},
168+
): Promise<ProviderStreamAttempt<TChunk, TMetadata>> {
169+
return retryProviderRequest(async () => {
170+
const attempt = await request();
171+
const iterator = attempt.stream[Symbol.asyncIterator]();
172+
const first = await iterator.next();
173+
return {
174+
stream: replayPrefetchedStream(first, iterator),
175+
metadata: attempt.metadata,
176+
};
177+
}, options);
178+
}

0 commit comments

Comments
 (0)