From 2f7d100d910dabcf5b18da482ff08c4bc00cf815 Mon Sep 17 00:00:00 2001 From: TheRealDinghyDog Date: Fri, 10 Jul 2026 12:54:29 +0200 Subject: [PATCH] Poll for recovery after a stream drop instead of failing at the grace (#30) A closed event stream raced the whole turn: after a 5s grace, captureTurn made one recovery attempt and could fail immediately even while the held-open /message request was still running and about to succeed (the review reproduced a turn failing at ~5.8s while the response completed at 6.2s). Treat stream closure as loss of one observation channel, not turn completion. After the stream drops, wait the grace for the response and trailing events to land, then actively poll the server for the finished message on an interval until the turn completes another way (response fallback, HTTP recovery, a reconnected session.idle) or the OUTER turn timeout fires. The turn is never failed merely because the stream ended. The grace is now a delay before polling starts rather than a deadline, and a bounded poll interval (default 2s) governs subsequent attempts. The loop is drained in finally so it cannot poll past return, and it still honors the outer timeout so a genuinely stalled turn ends promptly instead of spinning. captureTurn is exported for tests, which drive it against a localhost server across three timings: stream drop + slow-but-successful response (previously failed at the grace), stream drop + lost POST recoverable only via GET, and stream drop + response that never lands (must end near the outer timeout). Co-Authored-By: Claude Fable 5 --- plugins/opencode/scripts/lib/opencode.mjs | 48 +++++- tests/turn-capture.test.mjs | 175 ++++++++++++++++++++++ 2 files changed, 216 insertions(+), 7 deletions(-) create mode 100644 tests/turn-capture.test.mjs diff --git a/plugins/opencode/scripts/lib/opencode.mjs b/plugins/opencode/scripts/lib/opencode.mjs index 93a8066..d6b898f 100644 --- a/plugins/opencode/scripts/lib/opencode.mjs +++ b/plugins/opencode/scripts/lib/opencode.mjs @@ -25,7 +25,13 @@ const READ_ONLY_AGENT = "plan"; // (issue #2 / review finding #17). Deep reviews legitimately run many minutes, // so keep it generous; on expiry we still try to recover the final message. const DEFAULT_TURN_TIMEOUT_MS = 30 * 60 * 1000; +// After the event stream drops, wait this long for the held-open /message +// response and any trailing events to land before the first HTTP recovery +// poll. This is a grace before polling STARTS — not a deadline (issue #30). const DEFAULT_STREAM_DROP_GRACE_MS = 5000; +// Interval between HTTP recovery polls while the stream is down and the turn +// has not otherwise completed. Polling continues until the outer turn timeout. +const DEFAULT_STREAM_DROP_POLL_INTERVAL_MS = 2000; const DEFAULT_RECOVERY_TIMEOUT_MS = 5000; function sleep(ms) { @@ -926,6 +932,8 @@ async function captureTurn(client, sessionID, startRequest, options = {}) { const turnTimeoutMs = Math.max(0, Number(options.turnTimeoutMs) || DEFAULT_TURN_TIMEOUT_MS); let timedOut = false; let turnTimer = null; + // Hoisted so the finally block can drain it even if the try throws early. + let streamDropRecovery = Promise.resolve(); try { await Promise.race([ @@ -991,17 +999,38 @@ async function captureTurn(client, sessionID, startRequest, options = {}) { 0, Number(options.streamDropGraceMs ?? DEFAULT_STREAM_DROP_GRACE_MS) || DEFAULT_STREAM_DROP_GRACE_MS ); - const eventStreamDropPromise = eventStream.then(async () => { - if (state.completed || responseSettled) { - return state; - } - if (streamDropGraceMs > 0) { + const streamDropPollIntervalMs = Math.max( + 50, + Number(options.streamDropPollIntervalMs ?? DEFAULT_STREAM_DROP_POLL_INTERVAL_MS) || + DEFAULT_STREAM_DROP_POLL_INTERVAL_MS + ); + + // A dropped event stream is loss of one observation channel, NOT turn + // completion. The held-open /message response can still be running and + // succeed well after the drop, and the finished message is fetchable over + // HTTP. So once the stream closes: wait a short grace for the response and + // trailing events to land, then actively poll the server for the finished + // message until the turn completes another way (response fallback, + // recovery, session.idle over a reconnect) or the OUTER turn timeout fires. + // We never fail the turn merely because the stream ended (issue #30). + streamDropRecovery = eventStream.then(async () => { + if (streamDropGraceMs > 0 && !state.completed && !responseSettled) { await Promise.race([state.completion, responseSettledPromise, sleep(streamDropGraceMs)]); } - return state; + while (!state.completed && !timedOut) { + // Only fetch when we actually lack a result; skip the redundant GET if + // parts already streamed in before the drop (issue #12). + if (!state.error && !state.finalMessage && state.structuredOutput == null) { + await recoverFinalMessageFromServer(client, state, { recoveryTimeoutMs: options.recoveryTimeoutMs }); + } + if (state.completed || timedOut) { + break; + } + await Promise.race([state.completion, timeoutPromise, sleep(streamDropPollIntervalMs)]); + } }); - await Promise.race([state.completion, eventStreamDropPromise, timeoutPromise]); + await Promise.race([state.completion, streamDropRecovery, timeoutPromise]); // If we didn't capture a usable result, pull the finished message straight // from the server before giving up. Key this off whether we actually lack a @@ -1034,6 +1063,10 @@ async function captureTurn(client, sessionID, startRequest, options = {}) { } eventAbort.abort(); await eventStream.catch(() => {}); + // Drain the stream-drop recovery loop so it cannot poll past return. Its + // guard (state.completed || timedOut) is already satisfied once the outer + // race resolved, so this settles promptly. + await streamDropRecovery.catch(() => {}); if (state.fallbackTimer) { clearTimeout(state.fallbackTimer); } @@ -1557,6 +1590,7 @@ export function readOutputSchema(schemaPath) { export { DEFAULT_CONTINUE_PROMPT, TASK_SESSION_PREFIX, + captureTurn as captureTurnForTest, getAvailability as getOpencodeAvailability, getAuthStatus as getOpencodeAuthStatus }; diff --git a/tests/turn-capture.test.mjs b/tests/turn-capture.test.mjs new file mode 100644 index 0000000..34c072a --- /dev/null +++ b/tests/turn-capture.test.mjs @@ -0,0 +1,175 @@ +import http from "node:http"; +import net from "node:net"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { OpencodeServerClient } from "../plugins/opencode/scripts/lib/opencode-server.mjs"; +import { captureTurnForTest } from "../plugins/opencode/scripts/lib/opencode.mjs"; + +async function canListenLocalhost() { + return new Promise((resolve) => { + const server = net.createServer(); + let settled = false; + function finish(value) { + if (settled) { + return; + } + settled = true; + resolve(value); + } + server.once("error", () => finish(false)); + server.listen(0, "127.0.0.1", () => { + server.close(() => finish(true)); + }); + }); +} + +const LOCAL_LISTEN_AVAILABLE = await canListenLocalhost(); +const LOCAL_LISTEN_SKIP = LOCAL_LISTEN_AVAILABLE ? false : "local 127.0.0.1 listen is unavailable in this sandbox"; + +// A tiny OpenCode-shaped server whose event stream drops early while the +// held-open /message response resolves later. Behavior is tuned per test to +// reproduce the issue #30 timing precisely. +function startTurnServer({ eventDropMs, responseDelayMs, dropResponse = false }) { + const sessionID = "ses_capture"; + const messageID = "msg_capture"; + const finalText = "Recovered final answer."; + let stored = null; + + const server = http.createServer((req, res) => { + const url = new URL(req.url, "http://127.0.0.1"); + + if (req.method === "GET" && url.pathname === "/event") { + res.writeHead(200, { "content-type": "text/event-stream", "cache-control": "no-cache" }); + res.write(":ok\n\n"); + setTimeout(() => res.destroy(), eventDropMs); + return; + } + + if (req.method === "GET" && url.pathname === `/session/${sessionID}/message`) { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify(stored ? [stored] : [])); + return; + } + + if (req.method === "POST" && url.pathname === `/session/${sessionID}/message`) { + setTimeout(() => { + stored = { + info: { id: messageID, role: "assistant", sessionID }, + parts: [{ id: "prt_capture", sessionID, messageID, type: "text", text: finalText }] + }; + if (dropResponse) { + // The completed turn is only recoverable over HTTP GET now. + res.destroy(); + return; + } + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify(stored)); + }, responseDelayMs); + return; + } + + res.writeHead(404, { "content-type": "application/json" }); + res.end(JSON.stringify({ error: "not found" })); + }); + + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + resolve({ + server, + sessionID, + finalText, + baseUrl: `http://127.0.0.1:${server.address().port}`, + close: () => new Promise((done) => server.close(done)) + }); + }); + }); +} + +const FAST_CAPTURE_OPTIONS = { + streamDropGraceMs: 100, + streamDropPollIntervalMs: 80, + recoveryTimeoutMs: 500, + turnTimeoutMs: 4000, + eventOpenTimeoutMs: 2000 +}; + +test( + "captureTurn keeps waiting when the stream drops before a slow but successful response (issue #30)", + { skip: LOCAL_LISTEN_SKIP }, + async () => { + // Stream drops at 40ms; the /message response succeeds at 500ms — well past + // the 100ms stream-drop grace. Pre-#30 this failed at ~grace with an empty + // recovery instead of waiting for the response. + const fixture = await startTurnServer({ eventDropMs: 40, responseDelayMs: 500 }); + const client = new OpencodeServerClient(fixture.baseUrl); + + try { + const state = await captureTurnForTest( + client, + fixture.sessionID, + (signal) => client.sendMessage(fixture.sessionID, { parts: [{ type: "text", text: "go" }] }, { signal }), + FAST_CAPTURE_OPTIONS + ); + + assert.equal(state.error, null, state.error?.message); + assert.equal(state.finalMessage, fixture.finalText); + } finally { + await fixture.close(); + } + } +); + +test( + "captureTurn recovers over HTTP when the stream drops and the response POST is lost (issue #30)", + { skip: LOCAL_LISTEN_SKIP }, + async () => { + // Stream drops at 40ms and the held-open POST is destroyed after the turn + // completes server-side, so the answer is only reachable via GET. The poll + // loop must keep trying until the message is stored and recover it. + const fixture = await startTurnServer({ eventDropMs: 40, responseDelayMs: 400, dropResponse: true }); + const client = new OpencodeServerClient(fixture.baseUrl); + + try { + const state = await captureTurnForTest( + client, + fixture.sessionID, + (signal) => client.sendMessage(fixture.sessionID, { parts: [{ type: "text", text: "go" }] }, { signal }), + FAST_CAPTURE_OPTIONS + ); + + assert.equal(state.error, null, state.error?.message); + assert.equal(state.finalMessage, fixture.finalText); + assert.equal(state.recovered, true); + } finally { + await fixture.close(); + } + } +); + +test( + "captureTurn still times out when the stream drops and no result ever appears (issue #30)", + { skip: LOCAL_LISTEN_SKIP }, + async () => { + // Stream drops and the response never completes: the poll loop must be + // bounded by the outer turn timeout, not spin forever. + const fixture = await startTurnServer({ eventDropMs: 40, responseDelayMs: 60_000 }); + const client = new OpencodeServerClient(fixture.baseUrl); + + try { + const startedAt = Date.now(); + const state = await captureTurnForTest( + client, + fixture.sessionID, + (signal) => client.sendMessage(fixture.sessionID, { parts: [{ type: "text", text: "go" }] }, { signal }), + { ...FAST_CAPTURE_OPTIONS, turnTimeoutMs: 700 } + ); + + assert.ok(state.error, "a stalled turn must surface an error"); + assert.equal(state.finalMessage, ""); + assert.ok(Date.now() - startedAt < 3000, "the turn must end near its timeout, not spin"); + } finally { + await fixture.close(); + } + } +);