Skip to content

Commit e59a453

Browse files
TiaraBasoriTiaraBasori
authored
fix: end a turn when the upstream call fails instead of waiting it out (#27)
Squash-merged from #27.
1 parent 9c4c167 commit e59a453

1 file changed

Lines changed: 130 additions & 12 deletions

File tree

‎src/proxy.js‎

Lines changed: 130 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -526,6 +526,17 @@ function retryDelayMs({ attempt, response = null, baseMs = RETRY_BACKOFF_BASE_MS
526526
return Math.min(delay, RETRY_BACKOFF_MAX_MS);
527527
}
528528

529+
// --- Mutex Logic with Timeout ---
530+
/**
531+
* The failure carried by a `session.prompt` result, or null when the turn was
532+
* accepted. The SDK resolves (rather than rejects) for a non-2xx backend
533+
* response, so callers must look at `error` as well as at a rejection.
534+
*/
535+
function promptErrorOf(result) {
536+
if (!result || !result.error) return null;
537+
return result.error.info?.error || result.error;
538+
}
539+
529540
// --- Mutex Logic with Timeout ---
530541
async function getImageDataUri(url) {
531542
if (url.startsWith('data:')) {
@@ -1633,6 +1644,7 @@ export function createApp(config) {
16331644
if (!entry || entry.sessionId !== sessionId) return;
16341645
entry.userTurns = userTurns;
16351646
entry.pending = false;
1647+
entry.failed = false;
16361648
entry.expiresAt = Date.now() + CHAT_SESSION_TTL_MS;
16371649
chatSessionActive.delete(sessionId);
16381650
};
@@ -1721,7 +1733,12 @@ export function createApp(config) {
17211733
const timeoutPromise = new Promise((_, reject) => {
17221734
setTimeout(() => reject(new Error(`Request timeout after ${timeoutMs}ms`)), timeoutMs);
17231735
});
1724-
return Promise.race([client.session.prompt(promptParams), timeoutPromise]);
1736+
const result = await Promise.race([client.session.prompt(promptParams), timeoutPromise]);
1737+
// The SDK resolves for a non-2xx backend response; treat that as the
1738+
// failure it is instead of polling for a reply that will never appear.
1739+
const failure = promptErrorOf(result);
1740+
if (failure) throw failure;
1741+
return result;
17251742
}
17261743

17271744
const getCleanupRoots = () => {
@@ -1885,7 +1902,7 @@ export function createApp(config) {
18851902
* @returns {Promise<object>} `{content, reasoning, usage?, error?}`
18861903
*/
18871904
async function collectFromEvents(sessionId, timeoutMs, onDelta, firstDeltaTimeoutMs, idleTimeoutMs, options = {}) {
1888-
const { signal: clientSignal = null, onUsage = null } = options;
1905+
const { signal: clientSignal = null, onUsage = null, promptFailure = null } = options;
18891906
const controller = new AbortController();
18901907
const eventStreamResult = await client.event.subscribe({ signal: controller.signal });
18911908
const eventStream = eventStreamResult.stream;
@@ -2042,6 +2059,36 @@ export function createApp(config) {
20422059
deltaChars += delta.length;
20432060
};
20442061

2062+
// A prompt call that fails never produces an event, so waiting for one
2063+
// means waiting out the whole window while the caller sees keepalives.
2064+
// The SDK reports a failed turn either by rejecting or by resolving with
2065+
// `{ error }` (a non-2xx from the backend), so both count. A resolved
2066+
// prompt keeps the collector waiting for the events it will produce.
2067+
if (promptFailure) {
2068+
const endOnPromptError = (err) => {
2069+
if (finished) return;
2070+
finished = true;
2071+
clearTimeout(timeoutId);
2072+
if (firstDeltaTimer) clearTimeout(firstDeltaTimer);
2073+
if (idleTimer) clearTimeout(idleTimer);
2074+
releaseClientSignal();
2075+
controller.abort();
2076+
logDebug('Prompt call failed, ending turn', {
2077+
sessionId,
2078+
ms: Date.now() - startedAt,
2079+
error: err?.message || String(err)
2080+
});
2081+
resolve({ content, reasoning, usage: reportedUsage, __error: err, promptFailed: true });
2082+
};
2083+
promptFailure.then(
2084+
(result) => {
2085+
const err = promptErrorOf(result);
2086+
if (err) endOnPromptError(err);
2087+
},
2088+
endOnPromptError
2089+
);
2090+
}
2091+
20452092
(async () => {
20462093
try {
20472094
for await (const event of eventStream) {
@@ -2447,7 +2494,10 @@ export function createApp(config) {
24472494
signal = chatSessionSignalOf(req, req.body);
24482495
if (signal) {
24492496
const entry = chatSessions.get(signal);
2450-
if (entry && entry.expiresAt > Date.now() && !entry.pending
2497+
// A turn that ended on a provider error leaves an errored
2498+
// message in the session; resuming it would replay that error
2499+
// on every later turn of the conversation.
2500+
if (entry && !entry.failed && entry.expiresAt > Date.now() && !entry.pending
24512501
&& entry.model === `${pID}/${mID}`
24522502
&& entry.system === (systemWithGuard || '')
24532503
&& entry.toolPolicy === toolControl.title
@@ -2562,6 +2612,24 @@ export function createApp(config) {
25622612
};
25632613
ensureKeepalive();
25642614

2615+
// Errors are not model output: send them as their own chunk
2616+
// instead of pushing them through the content filter, which can
2617+
// legitimately hold or strip text.
2618+
const sendStreamError = (failure) => {
2619+
if (res.destroyed) return;
2620+
res.write(`data: ${JSON.stringify({
2621+
id,
2622+
object: 'chat.completion.chunk',
2623+
created: Math.floor(Date.now() / 1000),
2624+
model: `${pID}/${mID}`,
2625+
choices: [{
2626+
index: 0,
2627+
delta: { content: `\n\n[Proxy Error] ${failure?.name || 'OpenCodeError'}: ${failure?.data?.message || failure?.message || 'upstream request failed'}` },
2628+
finish_reason: null
2629+
}]
2630+
})}\n\n`);
2631+
};
2632+
25652633
const sendDelta = (delta, isReasoning = false) => {
25662634
if (!delta) return;
25672635
if (isReasoning) rawStreamedReasoning += delta;
@@ -2649,6 +2717,8 @@ export function createApp(config) {
26492717
await sleep(retryDelayMs({ attempt, response: lastAttemptError?.response }));
26502718
}
26512719
try {
2720+
const promptCall = client.session.prompt(promptParams);
2721+
promptCall.catch((err) => logDebug('Prompt error:', err.message));
26522722
const collectPromise = collectFromEvents(
26532723
sessionId,
26542724
REQUEST_TIMEOUT_MS,
@@ -2657,10 +2727,13 @@ export function createApp(config) {
26572727
EVENT_IDLE_TIMEOUT_MS,
26582728
{
26592729
signal: clientAbort.signal,
2660-
onUsage: (usage) => { reportedUsage = usage; }
2730+
onUsage: (usage) => { reportedUsage = usage; },
2731+
promptFailure: promptCall
26612732
}
26622733
);
26632734
const safeCollect = collectPromise.catch((err) => ({ __error: err }));
2735+
// The collector watches this promise, so a prompt that fails
2736+
// ends the turn instead of waiting out its window.
26642737
client.session.prompt(promptParams).catch(err => logDebug('Prompt error:', err.message));
26652738
collected = await safeCollect;
26662739
} catch (e) {
@@ -2689,7 +2762,11 @@ export function createApp(config) {
26892762
// Every fallback below re-reads the session through polling.
26902763
// With the caller gone there is nobody to stream to and the
26912764
// turn was aborted, so re-running it would be pure cost.
2692-
if (collected && collected.__error && !clientGone) {
2765+
// A prompt that failed leaves nothing to poll for: the turn never
2766+
// started, so the polling fallback would only wait out its window
2767+
// and then report an empty result.
2768+
const noTurnToPoll = Boolean(collected && collected.promptFailed);
2769+
if (collected && collected.__error && !clientGone && !noTurnToPoll) {
26932770
logDebug('SSE collect error, falling back to polling', {
26942771
sessionId,
26952772
error: collected.__error?.message
@@ -2738,7 +2815,7 @@ export function createApp(config) {
27382815
if (collected.content) sendDelta(collected.content, false);
27392816
}
27402817

2741-
if (!streamedContent && !streamedReasoning && !clientGone) {
2818+
if (!streamedContent && !streamedReasoning && !clientGone && !noTurnToPoll) {
27422819
logDebug('SSE returned empty, falling back to polling', { sessionId });
27432820
const polled = await pollForAssistantResponse(sessionId, REQUEST_TIMEOUT_MS);
27442821
if (polled.usage) reportedUsage = polled.usage;
@@ -2836,7 +2913,12 @@ export function createApp(config) {
28362913
return;
28372914
}
28382915

2839-
if (affinitySignal && !streamToolCallsSeen) {
2916+
if (noTurnToPoll) {
2917+
// The turn never started: report it on the open stream, then
2918+
// drop the mapping so the next request starts a clean session.
2919+
sendStreamError(collected.__error);
2920+
if (affinitySignal) await chatSessionEntryFail(affinitySignal, sessionId);
2921+
} else if (affinitySignal && !streamToolCallsSeen) {
28402922
chatSessionEntryComplete(affinitySignal, sessionId, [...(reuseUserTurns || []), lastUserMsg]);
28412923
} else if (affinitySignal) {
28422924
chatSessionEntryFail(affinitySignal, sessionId);
@@ -2894,7 +2976,21 @@ export function createApp(config) {
28942976
}
28952977
if (clientGone) break;
28962978
const attemptStart = Date.now();
2897-
await promptWithTimeout(promptParams, REQUEST_TIMEOUT_MS);
2979+
// A rejected prompt means the turn never started; polling for
2980+
// it would wait out the whole request timeout first.
2981+
try {
2982+
await promptWithTimeout(promptParams, REQUEST_TIMEOUT_MS);
2983+
} catch (promptError) {
2984+
if (clientGone) break;
2985+
const failure = promptErrorOf(promptError) || promptError;
2986+
error = failure;
2987+
lastAttemptError = failure;
2988+
if (attempt < RETRY_MAX_ATTEMPTS && isTransientUpstreamError(failure)) {
2989+
console.warn(`[Proxy] Transient upstream error (attempt ${attempt}/${RETRY_MAX_ATTEMPTS}), retrying:`, failure.data?.message || failure.message || failure.name || 'unknown');
2990+
continue;
2991+
}
2992+
break;
2993+
}
28982994
logDebug('Prompt sent', { sessionId, ms: Date.now() - attemptStart, attempt });
28992995
const collected = await pollForAssistantResponse(sessionId, REQUEST_TIMEOUT_MS);
29002996
content = collected.content || '';
@@ -3537,6 +3633,8 @@ export function createApp(config) {
35373633
await sleep(retryDelayMs({ attempt, response: lastAttemptError?.response }));
35383634
}
35393635
try {
3636+
const promptCall = client.session.prompt(promptParams);
3637+
promptCall.catch((err) => logDebug('Prompt error:', err.message));
35403638
const collectPromise = collectFromEvents(
35413639
sessionId,
35423640
REQUEST_TIMEOUT_MS,
@@ -3545,10 +3643,13 @@ export function createApp(config) {
35453643
EVENT_IDLE_TIMEOUT_MS,
35463644
{
35473645
signal: clientAbort.signal,
3548-
onUsage: (usage) => { reportedUsage = usage; }
3646+
onUsage: (usage) => { reportedUsage = usage; },
3647+
promptFailure: promptCall
35493648
}
35503649
);
35513650
const safeCollect = collectPromise.catch((err) => ({ __error: err }));
3651+
// The collector watches this promise, so a prompt that fails
3652+
// ends the turn instead of waiting out its window.
35523653
client.session.prompt(promptParams).catch(err => logDebug('Prompt error:', err.message));
35533654
collected = await safeCollect;
35543655
} catch (e) {
@@ -3778,7 +3879,21 @@ export function createApp(config) {
37783879
await sleep(retryDelayMs({ attempt, response: lastAttemptError?.response }));
37793880
}
37803881
if (clientGone) break;
3781-
await promptWithTimeout(promptParams, REQUEST_TIMEOUT_MS);
3882+
// See the Chat handler: a rejected prompt ends the attempt
3883+
// instead of polling for a reply that cannot appear.
3884+
try {
3885+
await promptWithTimeout(promptParams, REQUEST_TIMEOUT_MS);
3886+
} catch (promptError) {
3887+
if (clientGone) break;
3888+
const failure = promptErrorOf(promptError) || promptError;
3889+
error = failure;
3890+
lastAttemptError = failure;
3891+
if (attempt < RETRY_MAX_ATTEMPTS && isTransientUpstreamError(failure)) {
3892+
console.warn(`[Proxy] Transient upstream error (attempt ${attempt}/${RETRY_MAX_ATTEMPTS}), retrying:`, failure.data?.message || failure.message || failure.name || 'unknown');
3893+
continue;
3894+
}
3895+
break;
3896+
}
37823897
const collected = await pollForAssistantResponse(sessionId, REQUEST_TIMEOUT_MS);
37833898
content = collected.content || '';
37843899
reasoning = collected.reasoning || '';
@@ -4652,6 +4767,9 @@ export function createApp(config) {
46524767

46534768
let collected = null;
46544769
try {
4770+
// See the Chat handler: a prompt that fails ends the turn at once.
4771+
const promptCall = client.session.prompt(promptParams);
4772+
promptCall.catch(err => logDebug('Responses prompt error:', err.message));
46554773
const collectPromise = collectFromEvents(
46564774
sessionId,
46574775
REQUEST_TIMEOUT_MS,
@@ -4660,11 +4778,11 @@ export function createApp(config) {
46604778
EVENT_IDLE_TIMEOUT_MS,
46614779
{
46624780
signal: clientAbort.signal,
4663-
onUsage: (usage) => { reportedUsage = usage; }
4781+
onUsage: (usage) => { reportedUsage = usage; },
4782+
promptFailure: promptCall
46644783
}
46654784
);
46664785
const safeCollect = collectPromise.catch((err) => ({ __error: err }));
4667-
client.session.prompt(promptParams).catch(err => logDebug('Responses prompt error:', err.message));
46684786
collected = await safeCollect;
46694787
} catch (e) {
46704788
collected = { __error: e };

0 commit comments

Comments
 (0)