diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 02271ebc583..a58ef9fb3cb 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -391,6 +391,12 @@ runner supplies the pinned executable, selected model, and assigned MCP tools. Provider inference and reasoning still contribute to time to first text. These startup settings do not disable model reasoning. +An interrupted OpenCode turn does not request a replacement final answer. +This also applies when an unanswered native question moves to a durable wait. +The driver reports an interrupted idle as a cancelled terminal so the wait can +settle. An abort acknowledgement alone does not prove that the provider stopped. +A later authorized turn can use normal answer recovery. + On macOS, the verified OpenCode executable keeps its private launch pathname until the provider process exits. Removing that pathname immediately after spawn can kill or stall the signed binary before its health check succeeds. diff --git a/packages/paperclip-runner/src/drivers/opencode/interrupt-empty-response.test.ts b/packages/paperclip-runner/src/drivers/opencode/interrupt-empty-response.test.ts new file mode 100644 index 00000000000..3462e9ea308 --- /dev/null +++ b/packages/paperclip-runner/src/drivers/opencode/interrupt-empty-response.test.ts @@ -0,0 +1,124 @@ +import { chmod, copyFile, mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import type { PrpEvent } from "../../protocol/replay-contract.js"; +import { OpenCodeServerDriver } from "./opencode-server-driver.js"; + +it.each(["idle", "status", "silent", "correcting-idle", "correcting-silent", "busy-idle", "handoff-idle", "handoff-status", "handoff-silent"] as const)("does not restart an interrupted empty OpenCode response: %s", async terminal => { + const root = await mkdtemp(join(tmpdir(), "opencode-interrupt-empty-")); + const command = join(root, "fake-opencode.mjs"); + await copyFile(resolve("test/fixtures/fake-opencode-server.mjs"), command); + await chmod(command, 0o755); + let stream: ReadableStreamDefaultController | undefined; + let streamOpened!: () => void; + const opened = new Promise(resolve => { streamOpened = resolve; }); + const prompts: unknown[] = []; + const observed: PrpEvent[] = []; + const correcting = terminal.startsWith("correcting-"); + const silent = terminal.endsWith("silent"); + const busy = terminal === "busy-idle"; + const handoff = terminal.startsWith("handoff-"); + let ownedSessionId = ""; + const emit = (value: unknown) => stream!.enqueue(new TextEncoder().encode(`data: ${JSON.stringify(value)}\n\n`)); + let resumed = false; + const driver = new OpenCodeServerDriver({ + model: "openrouter/controlled-model", + runtimeDirectory: join(root, "runtime"), command, + environment: { PATH: process.env.PATH, OPENROUTER_API_KEY: "controlled-fixture-key" }, + fetch: async (input, init) => { + const url = new URL(String(input)); + if (url.pathname === "/event") return new Response(new ReadableStream({ + start(controller) { stream = controller; streamOpened(); }, + }), { headers: { "Content-Type": "text/event-stream" } }); + const sessionID = decodeURIComponent(url.pathname.split("/")[2] ?? ""); + if (url.pathname.endsWith("/prompt_async")) { + ownedSessionId = sessionID; + prompts.push(JSON.parse(String(init?.body))); + if (resumed) queueMicrotask(() => { + emit({ type: "message.updated", properties: { info: { id: "resumed-answer", sessionID, role: "assistant", finish: "stop", time: { completed: 2 } } } }); + emit({ type: "message.part.updated", properties: { part: { id: "resumed-part", messageID: "resumed-answer", sessionID, type: "text", text: "The authorized next turn completed.", time: { start: 1, end: 2 } } } }); + emit({ type: "session.idle", properties: { sessionID } }); + }); + else if (correcting && prompts.length === 1) queueMicrotask(() => { + emit({ type: "session.idle", properties: { sessionID } }); + }); + else if (busy && prompts.length === 1) queueMicrotask(() => { + emit({ type: "message.updated", properties: { info: { id: "tool-message", sessionID, role: "assistant" } } }); + emit({ type: "message.part.updated", properties: { part: { id: "owned-tool", messageID: "tool-message", sessionID, + type: "tool", tool: "bash", callID: "owned-call", state: { status: "running", input: { command: "controlled work" } } } } }); + }); + else if (handoff && prompts.length === 1) queueMicrotask(() => { + emit({ type: "question.asked", properties: { id: "owned-question", sessionID, questions: [ + { question: "Which color should I use?", header: "Color", options: [{ label: "Blue", description: "Use blue." }] }, + ] } }); + }); + return new Response(null, { status: 204 }); + } + if (url.pathname.endsWith("/abort")) { + if (!silent) queueMicrotask(() => { + emit(!terminal.endsWith("status") + ? { type: "session.idle", properties: { sessionID } } + : { type: "session.status", properties: { sessionID, status: { type: "idle" } } }); + }); + return Response.json(true); + } + if (url.pathname === "/question/owned-question/reject") return Response.json(true); + return fetch(input, init); + }, + }); + const session = await driver.openSession({ runId: "owned-run", normalizedSessionId: "owned-session", workingDirectory: root }); + let collecting: Promise | undefined; + try { + await opened; + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + collecting = (async () => { for await (const event of session.events()) observed.push(event); })(); + const first = await session.startTurn({ message: { role: "user", text: "Save a plan and wait for approval." } }); + await vi.advanceTimersByTimeAsync(0); + const allowedPrompts = correcting ? 2 : 1; + expect(prompts).toHaveLength(allowedPrompts); + if (handoff) { + const pending = session.pendingRuntimeRequests?.(); + expect(pending).toHaveLength(1); + const result = session.handoffRuntimeRequest?.({ requestId: "owned-question", turnId: first.turnId, + reason: "durable_handoff", signal: new AbortController().signal }); + expect(result?.result).toBe("handed_off"); + await result?.cleanup; + expect(observed.filter(event => event.eventType === "runtime_request.expired")).toHaveLength(1); + } else await session.interrupt?.({ turnId: first.turnId, reason: "Durable approval is pending." }); + await vi.advanceTimersByTimeAsync(0); + // HTTP acknowledgement alone is not a provider terminal. An observed idle + // is a terminal, but must never authorize a new corrective model request. + expect(prompts).toHaveLength(allowedPrompts); + const terminalEvents = () => observed.filter(event => ["turn.completed", "turn.cancelled", "turn.failed"].includes(event.eventType)); + expect(terminalEvents()).toHaveLength(silent || busy ? 0 : 1); + if (!silent && !busy) expect(terminalEvents()[0]?.eventType).toBe("turn.cancelled"); + await vi.advanceTimersByTimeAsync(20_001); + expect(prompts).toHaveLength(allowedPrompts); + expect(terminalEvents()).toHaveLength(silent || busy ? 0 : 1); + if (busy) { + emit({ type: "message.part.updated", properties: { part: { id: "owned-tool", messageID: "tool-message", sessionID: ownedSessionId, + type: "tool", tool: "bash", callID: "owned-call", state: { status: "completed", input: { command: "controlled work" }, output: "Work stopped.", time: { start: 1, end: 2 } } } } }); + emit({ type: "session.idle", properties: { sessionID: ownedSessionId } }); + await vi.advanceTimersByTimeAsync(0); + expect(terminalEvents()).toHaveLength(1); + expect(terminalEvents()[0]?.eventType).toBe("turn.cancelled"); + expect(prompts).toHaveLength(allowedPrompts); + } + if (!silent) { + resumed = true; + const next = await session.startTurn({ message: { role: "user", text: "The plan is approved. Continue." } }); + await vi.advanceTimersByTimeAsync(0); + expect(prompts).toHaveLength(allowedPrompts + 1); + expect(terminalEvents().filter(event => event.turnId === next.turnId)).toHaveLength(1); + expect(terminalEvents().find(event => event.turnId === next.turnId)?.eventType).toBe("turn.completed"); + expect(observed.some(event => event.eventType === "item.completed" && event.payload.kind === "agentMessage" + && event.payload.channel === "final" && event.payload.text === "The authorized next turn completed.")).toBe(true); + } + } finally { + vi.useRealTimers(); + await session.close({ reason: "controlled test complete" }); + await collecting; + await rm(root, { recursive: true, force: true }); + } +}, 15_000); 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 e21e3af9ff9..01504933c1e 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts @@ -544,6 +544,7 @@ class OpenCodeHarnessSession implements HarnessSession { #turnReasoningMode: OpenCodeReasoningMode = "default"; #emptyResponseRecovery: { turnId: string; priorMessages: Set; responseStarted: boolean; responseFinished: boolean } | null = null; #emptyResponseAckTimer: ReturnType | undefined; + #interruptedTurnId: string | null = null; #pendingToolCalls = 0; readonly #runningNativeToolParts = new Set(); #abort = new AbortController(); @@ -695,6 +696,7 @@ class OpenCodeHarnessSession implements HarnessSession { this.#turnStartSourceSeq = this.#sourceSequence; this.#turnReasoningMode = reasoningMode; this.#emptyResponseRecovery = null; + this.#interruptedTurnId = null; this.#runningNativeToolParts.clear(); this.#emit("turn.submitted", { envelopeSchema: this.#taskEnvelope.schema, @@ -765,7 +767,7 @@ class OpenCodeHarnessSession implements HarnessSession { } #failEmptyResponse(turnId: string, message: string): void { - if (this.#closed || this.#activeTurnId !== turnId || this.#result !== null || this.#pendingToolCalls > 0 || this.#runningNativeToolParts.size > 0) return; + if (this.#closed || this.#activeTurnId !== turnId || this.#interruptedTurnId === turnId || this.#result !== null || this.#pendingToolCalls > 0 || this.#runningNativeToolParts.size > 0) return; this.#emit("turn.failed", { status: "failed", error: { code: "opencode_empty_response", message, retryable: false } }, { turnId }); this.#terminalTurns.set(turnId, canonicalJson({ status: "failed" })); this.#activeTurnId = null; @@ -774,6 +776,9 @@ class OpenCodeHarnessSession implements HarnessSession { #recoverEmptyResponse(turnId: string): boolean { if (this.#result !== null || this.#pendingRuntimeRequests.size > 0) return false; if (this.#pendingToolCalls > 0 || this.#runningNativeToolParts.size > 0) return true; + // An intentional pause needs the provider's idle fact, not another model + // request. Keep in-flight tool work gated by the checks above. + if (this.#interruptedTurnId === turnId) return false; const recovery = this.#emptyResponseRecovery; const boundary = Math.max(this.#turnStartSourceSeq, this.#lastNonTerminalToolSourceSeq); if (this.#completedTextParts.some(part => part.observedSourceSeq > boundary)) return false; @@ -800,13 +805,18 @@ class OpenCodeHarnessSession implements HarnessSession { } async interrupt(input: { turnId?: string; reason?: string }): Promise { - await this.#completionSettlement; + const turnId = this.#activeTurnId; if ( input.turnId && - this.#activeTurnId && - input.turnId !== this.#activeTurnId + turnId && + input.turnId !== turnId ) throw new Error("stale OpenCode turn"); + if (!turnId) return; + this.#interruptedTurnId = turnId; + this.#clearEmptyResponseAckTimer(); + await this.#completionSettlement; + if (this.#activeTurnId !== turnId) return; await api( this.#fetch, this.#runtime, @@ -982,12 +992,7 @@ class OpenCodeHarnessSession implements HarnessSession { `/question/${encodeURIComponent(input.requestId)}/reject?${workspace}`, { method: "POST" }, ), - api( - this.#fetch, - this.#runtime, - `/session/${encodeURIComponent(this.#providerSessionId)}/abort`, - { method: "POST" }, - ), + this.interrupt({ turnId: input.turnId, reason: "Durable user input handoff." }), ]).then(() => undefined); return { result: "handed_off", cleanup }; } @@ -1864,12 +1869,14 @@ class OpenCodeHarnessSession implements HarnessSession { turnId ) { if (this.#recoverEmptyResponse(turnId)) return; + const interrupted = this.#interruptedTurnId === turnId; + const status = interrupted ? "cancelled" : "completed"; const fingerprint = canonicalJson({ - status: "completed", + status, semanticResult: this.#resultFingerprint, }); this.#terminalTurns.set(turnId, fingerprint); - this.#emitSettledFinalAgentMessage(turnId); + if (!interrupted) this.#emitSettledFinalAgentMessage(turnId); const workspace = this.#workspaceChangesByTurn.get(turnId); if (workspace !== undefined) this.#emit( @@ -1884,7 +1891,7 @@ class OpenCodeHarnessSession implements HarnessSession { // 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. - this.#emit("turn.completed", { status: "completed" }, { turnId }); + this.#emit(interrupted ? "turn.cancelled" : "turn.completed", { status }, { turnId }); this.#activeTurnId = null; return; }