diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts index d98ffbda2ab..bb893beab84 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts @@ -1022,6 +1022,87 @@ describe("OpenCodeServerDriver", () => { await session.close({ reason: "test" }); }); + it("delivers the settlement event for a runtime request whose turn fails through the same single-pass consumer that reads the turn", async () => { + await chmod(fixture, 0o755); + const root = await mkdtemp( + join(tmpdir(), "paperclip-opencode-pending-then-fail-"), + ); + const workspace = await mkdtemp( + join(tmpdir(), "paperclip-opencode-pending-then-fail-workspace-"), + ); + roots.push(root, workspace); + const driver = new OpenCodeServerDriver({ + model: "openrouter/deepseek/deepseek-v4-flash-0731", + runtimeDirectory: root, + command: fixture, + environment: { + PATH: process.env.PATH, + OPENROUTER_API_KEY: "fixture-key", + }, + }); + const session = await driver.openSession({ + runId: "run-pending-then-fail", + normalizedSessionId: "pending-then-fail", + workingDirectory: workspace, + }); + + const { turnId } = await session.startTurn({ + message: { role: "user", text: "pending-request-then-turn-fails" }, + }); + // The fixture asks a native question, which leaves a runtime request + // pending, then fails the same turn through `session.error` without + // ever resolving that request. `collectTurnEvents` reads exactly one + // pass and stops at the turn's terminal event, the same rule the + // production consumer applies (see its doc comment above). The + // settlement event must arrive inside that same pass: a production + // consumer that stops at `turn.failed` never opens a second read + // afterward, so a settlement event that only `close()` produced later + // would never reach it. + const turnEvents = await collectTurnEvents(session.events()); + expect( + turnEvents.some( + (event) => event.eventType === "runtime_request.created", + ), + ).toBe(true); + const settlementIndex = turnEvents.findIndex( + (event) => event.eventType === "runtime_request.expired", + ); + const terminalIndex = turnEvents.findIndex( + (event) => event.eventType === "turn.failed", + ); + expect(settlementIndex).toBeGreaterThanOrEqual(0); + expect(terminalIndex).toBeGreaterThanOrEqual(0); + // The settlement fact must precede the turn's terminal event, or a + // consumer that stops reading at that terminal event misses it. + expect(settlementIndex).toBeLessThan(terminalIndex); + expect(turnEvents[settlementIndex]).toMatchObject({ + turnId, + itemId: "question-native-1", + }); + expect(turnEvents[terminalIndex]).toMatchObject({ turnId }); + expect( + turnEvents.some( + (event) => + event.eventType === "harness.diagnostic" && + event.payload.code === "opencode_late_terminal_turn_event_dropped" && + event.payload.turnId === turnId, + ), + ).toBe(false); + // The request already settled with the turn; nothing is left pending + // for `close()` to settle a second time. + expect(session.pendingRuntimeRequests?.()).toHaveLength(0); + + await session.close({ reason: "test" }); + const closeEvents = await collectTurnEvents(session.events()); + expect( + closeEvents.some((event) => + ["runtime_request.expired", "runtime_request.cancelled"].includes( + event.eventType, + ), + ), + ).toBe(false); + }); + it("keeps the session usable after a cancelled turn so the next turn on the same session still completes", async () => { await chmod(fixture, 0o755); const root = await mkdtemp( diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts index eff4b210449..3726050255e 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts @@ -864,22 +864,48 @@ class OpenCodeHarnessSession implements HarnessSession { if (this.#closed) return; this.#closed = true; this.#abort.abort(); - for (const { request } of this.#pendingRuntimeRequests.values()) { - this.#emit( - request.input === undefined - ? "runtime_request.cancelled" - : "runtime_request.expired", - request.input === undefined - ? harnessRuntimeRequestOutcome(request, { reason: "session_closed" }) - : harnessRuntimeInputExpiredOutcome(request, "provider_process_lost"), - { turnId: request.turnId, itemId: request.itemId }, - ); - } - this.#pendingRuntimeRequests.clear(); + // Settle whatever is still pending, including a request whose turn + // already went terminal without the driver observing it (see + // `#settlePendingRuntimeRequestsForTurn`). Bypass the terminal-turn + // gate so each settlement event still reaches the consumer instead of + // getting dropped as a late frame. + for (const requestId of [...this.#pendingRuntimeRequests.keys()]) + this.#settlePendingRuntimeRequest(requestId); this.#events.close(); await this.#runtime.close(); } + // A request can outlive its own turn: the turn can fail or get cancelled + // while the request is still pending, which clears `#activeTurnId` + // without settling the request. Settle any such request the moment its + // owning turn goes terminal, before emitting the terminal turn event + // itself, so the settlement event reaches the consumer within the same + // turn it belongs to instead of waiting for a later, separate close. + #settlePendingRuntimeRequestsForTurn(turnId: string): void { + const requestIds = [...this.#pendingRuntimeRequests] + .filter(([, pending]) => pending.request.turnId === turnId) + .map(([requestId]) => requestId); + for (const requestId of requestIds) + this.#settlePendingRuntimeRequest(requestId); + } + + #settlePendingRuntimeRequest(requestId: string): void { + const pending = this.#pendingRuntimeRequests.get(requestId); + if (!pending) return; + this.#pendingRuntimeRequests.delete(requestId); + const { request } = pending; + this.#emit( + request.input === undefined + ? "runtime_request.cancelled" + : "runtime_request.expired", + request.input === undefined + ? harnessRuntimeRequestOutcome(request, { reason: "session_closed" }) + : harnessRuntimeInputExpiredOutcome(request, "provider_process_lost"), + { turnId: request.turnId, itemId: request.itemId }, + { bypassTerminalTurnGate: true }, + ); + } + async dispatchTool(call: { tool: string; callId: string; @@ -1508,6 +1534,10 @@ class OpenCodeHarnessSession implements HarnessSession { { ...workspace, complete: true }, { turnId, itemId: `${turnId}:workspace` }, ); + // Settle any request this turn never answered before the terminal + // event, so a consumer that stops reading at that terminal event still + // observes the settlement. + this.#settlePendingRuntimeRequestsForTurn(turnId); // Emit while this turn is still `#activeTurnId`; the gate in `#emit` // drops any frame whose turnId is not the active turn, so nulling it // first would make `#emit` drop this very event. @@ -1530,6 +1560,10 @@ class OpenCodeHarnessSession implements HarnessSession { // card. Preserve the provider fact as a cancelled terminal event; the // native session loop independently commits the authoritative yielded // result when this abort followed a governed wait. + // Settle any request this turn never answered before the terminal + // event, so a consumer that stops reading at that terminal event + // still observes the settlement. + this.#settlePendingRuntimeRequestsForTurn(turnId); this.#emit( "turn.cancelled", { @@ -1559,6 +1593,10 @@ class OpenCodeHarnessSession implements HarnessSession { }, { turnId, itemId: `${turnId}:session-error` }, ); + // Settle any request this turn never answered before the terminal + // event, so a consumer that stops reading at that terminal event still + // observes the settlement. + this.#settlePendingRuntimeRequestsForTurn(turnId); this.#emit( "turn.failed", { status: "failed", error: bounded(properties.error ?? properties) }, @@ -1829,8 +1867,10 @@ class OpenCodeHarnessSession implements HarnessSession { eventType: PrpEvent["eventType"], payload: Record, refs: { turnId?: string; itemId?: string } = {}, + options?: { bypassTerminalTurnGate?: boolean }, ): void { if ( + !options?.bypassTerminalTurnGate && eventType !== "harness.diagnostic" && refs.turnId !== undefined && refs.turnId !== this.#activeTurnId diff --git a/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs b/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs index 5fc57106eb5..dd0173702c4 100644 --- a/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs +++ b/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs @@ -483,6 +483,32 @@ const server = createServer(async (request, response) => { }); return; } + if ( + String(parsedPrompt.message ?? "").includes( + "pending-request-then-turn-fails", + ) + ) { + pendingQuestion = nativeQuestion(); + emit({ + type: "question.asked", + id: "event-question-pending-then-fail", + properties: pendingQuestion, + }); + setTimeout(() => { + emit({ + type: "session.error", + id: "event-session-failed-with-pending-request", + properties: { + sessionID: session.id, + error: { + name: "ProviderError", + message: "The fake provider failed with a request still pending.", + }, + }, + }); + }, 10); + return; + } if ( String(parsedPrompt.message ?? "").includes("late-straggler-source") ) {