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(); + } + } +);