From 294f200437d2692969c54da83b18388c0a973ed3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 9 Nov 2026 02:02:22 -0600 Subject: [PATCH] fix(runner): keep discarded OpenCode token deltas from delaying tool calls The OpenCode app-server proxy forwarded every reasoning and tool-output token delta to runnerd as its own item/delta frame. runnerd reads these frames in order, only after the controller acknowledges earlier events, and rewrites its whole durable provider state with fsync for each frame, including every tool result retained in the turn. Its Codex facade then discards item/delta. A tool call request therefore waited behind hundreds of discarded frames, and the wait grew with each retained result. The proxy now forwards only the assistant text deltas runnerd consumes. Completed reasoning and tool items still arrive as item/completed. Co-Authored-By: Claude Opus 5.5 --- .../src/cli/opencode-app-server-proxy.ts | 6 ++- .../src/cli/opencode-proxy-events.test.ts | 53 +++++++++++++++++-- .../src/cli/opencode-proxy-events.ts | 16 +++++- 3 files changed, 67 insertions(+), 8 deletions(-) diff --git a/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts b/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts index e677c5ae001..00bac836112 100644 --- a/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts +++ b/packages/paperclip-runner/src/cli/opencode-app-server-proxy.ts @@ -223,11 +223,13 @@ async function pumpEvents(opened: HarnessSession): Promise { // The inner OpenCode driver publishes session-scoped model metadata as // an item before any turn exists. The outer Codex protocol facade emits // its own model item, so forwarding this unbound duplicate would quite - // correctly trip the facade's strict turn-binding validator. + // correctly trip the facade's strict turn-binding validator. Token + // deltas runnerd does not consume stay here; see the predicate. if ( !shouldForwardOpenCodeProxyItem({ + eventType: event.eventType, turnId: event.turnId, - kind: payload.kind, + payload, }) ) continue; diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-events.test.ts b/packages/paperclip-runner/src/cli/opencode-proxy-events.test.ts index 9d484ba314f..afac016bd10 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-events.test.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-events.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { boundedOpenCodeTerminalError, @@ -8,17 +8,62 @@ import { shouldAnnounceOpenCodeProxyTurn, shouldForwardOpenCodeProxyItem, } from "./opencode-proxy-events.js"; +import { createOpenCodeProxyOutput } from "./opencode-proxy-output.js"; describe("OpenCode runnerd proxy event boundary", () => { it("drops only the duplicate session-scoped model item", () => { - expect(shouldForwardOpenCodeProxyItem({ kind: "model" })).toBe(false); + expect(shouldForwardOpenCodeProxyItem({ payload: { kind: "model" } })).toBe(false); expect( shouldForwardOpenCodeProxyItem({ turnId: "turn-1", - kind: "model", + payload: { kind: "model" }, }), ).toBe(true); - expect(shouldForwardOpenCodeProxyItem({ kind: "agentMessage" })).toBe(true); + expect(shouldForwardOpenCodeProxyItem({ payload: { kind: "agentMessage" } })).toBe(true); + }); + + it("keeps per-token deltas runnerd discards from queueing ahead of a tool call", () => { + // One provider step: streamed reasoning and tool output, a completed + // reasoning item, then the tool call request. runnerd pays a durable state + // save for every frame it reads, in order, before it reaches the request. + const framesBeforeToolCall = (tokens: number) => { + vi.useFakeTimers(); + try { + const written: Array<{ method?: string }> = []; + const output = createOpenCodeProxyOutput(frame => written.push(frame as { method?: string })); + const forward = (eventType: "item.started" | "item.delta" | "item.completed", itemId: string, payload: Record) => { + if (shouldForwardOpenCodeProxyItem({ eventType, turnId: "turn-1", payload })) + output.send(openCodeProxyItemNotification({ eventType, threadId: "thread-1", turnId: "turn-1", itemId, payload })); + }; + for (let index = 0; index < tokens; index += 1) { + forward("item.delta", "reasoning-1", { kind: "reasoning", channel: "detail", providerPhase: "reasoning", text: `token ${index} `, + item: { id: "reasoning-1", type: "reasoning", text: `token ${index} ` } }); + forward("item.delta", "tool-1", { kind: "tool", text: `line ${index}\n`, item: { id: "tool-1", type: "tool", state: { status: "running" } } }); + } + forward("item.completed", "reasoning-1", { kind: "reasoning", text: "Reasoning summary.", + item: { id: "reasoning-1", type: "reasoning", text: "Reasoning summary." } }); + output.send({ id: "opencode-1", method: "item/tool/call", params: { tool: "search_api" } }); + const position = written.findIndex(frame => frame.method === "item/tool/call"); + expect(position).toBeGreaterThanOrEqual(0); + expect(written.some(frame => frame.method === "item/delta")).toBe(false); + expect(written.at(position - 1)).toMatchObject({ method: "item/completed", params: { kind: "reasoning" } }); + return position; + } finally { + vi.useRealTimers(); + } + }; + expect(framesBeforeToolCall(1_000)).toBe(framesBeforeToolCall(1)); + }); + + it("still forwards streamed assistant text for runnerd's live reply", () => { + const payload = { kind: "agentMessage", channel: "progress", text: "Checking the plugin.", item: { type: "agentMessage" } }; + expect(shouldForwardOpenCodeProxyItem({ eventType: "item.delta", turnId: "turn-1", payload })).toBe(true); + expect(shouldForwardOpenCodeProxyItem({ eventType: "item.delta", turnId: "turn-1", + payload: { kind: "text", text: "Legacy text part." } })).toBe(true); + expect(shouldForwardOpenCodeProxyItem({ eventType: "item.delta", turnId: "turn-1", + payload: { kind: "reasoning", text: "Private reasoning" } })).toBe(false); + expect(shouldForwardOpenCodeProxyItem({ eventType: "item.completed", turnId: "turn-1", + payload: { kind: "reasoning", text: "Private reasoning" } })).toBe(true); }); it("announces one outer turn when SSE wins the response race", () => { diff --git a/packages/paperclip-runner/src/cli/opencode-proxy-events.ts b/packages/paperclip-runner/src/cli/opencode-proxy-events.ts index 1dfe90af890..ad80b81769e 100644 --- a/packages/paperclip-runner/src/cli/opencode-proxy-events.ts +++ b/packages/paperclip-runner/src/cli/opencode-proxy-events.ts @@ -19,11 +19,23 @@ export function openCodeProxyToolNotification(input: { } }; } +/** + * runnerd reads this proxy's frames in order, only after its controller + * acknowledges earlier events, and saves its whole durable provider state + * (including every tool result retained in the turn) for each frame it reads. + * Its Codex facade turns only assistant text deltas into events; it discards + * every other `item/delta`. OpenCode reports reasoning and tool output one + * token at a time, so forwarding those deltas queued each tool call behind + * hundreds of discarded frames, and the wait grew with every result the turn + * retained. Completed reasoning and tool items still arrive as item/completed. + */ export function shouldForwardOpenCodeProxyItem(input: { + eventType?: string; turnId?: string; - kind?: unknown; + payload: Record; }): boolean { - return !(input.turnId === undefined && input.kind === "model"); + if (input.eventType === "item.delta") return openCodeProxyAssistantText(input.payload) !== null; + return !(input.turnId === undefined && input.payload.kind === "model"); } export function openCodeProxyAssistantText(