From e673ed2851b4dac16d9f45354cc79dc511903d3c Mon Sep 17 00:00:00 2001 From: flowluap Date: Sun, 19 Jul 2026 22:41:20 +0200 Subject: [PATCH] perf(opencode): coalesce code mode progress updates --- packages/opencode/src/tool/code-mode.ts | 43 ++++- .../test/tool/code-mode-integration.test.ts | 7 +- packages/opencode/test/tool/code-mode.test.ts | 181 +++++++++++++++++- 3 files changed, 212 insertions(+), 19 deletions(-) diff --git a/packages/opencode/src/tool/code-mode.ts b/packages/opencode/src/tool/code-mode.ts index 332d4b43f150..fe3638bde1b3 100644 --- a/packages/opencode/src/tool/code-mode.ts +++ b/packages/opencode/src/tool/code-mode.ts @@ -1,6 +1,6 @@ import * as Tool from "./tool" import { CallToolResultSchema, type CallToolResult } from "@modelcontextprotocol/sdk/types.js" -import { Cause, Effect, Schema } from "effect" +import { Cause, Effect, Fiber, Schema } from "effect" import { CodeMode, Tool as SandboxTool, toolError } from "@opencode-ai/codemode" import { MCP } from "@/mcp" import { McpCatalog } from "@/mcp/catalog" @@ -213,8 +213,24 @@ export const CodeModeTool = Tool.define( const calls: CallEntry[] = [] const attachments: Attachment[] = [] - const publish = () => - ctx.metadata({ title: CODE_MODE_TOOL, metadata: { toolCalls: calls.map((c) => ({ ...c })) } }) + let version = 0 + let published = 0 + let warned = false + const publish = Effect.fnUntraced(function* () { + const current = version + if (current === published) return + yield* ctx.metadata({ title: CODE_MODE_TOOL, metadata: { toolCalls: calls.map((call) => ({ ...call })) } }) + published = current + }) + const publishProgress = () => + publish().pipe( + Effect.timeout("1 second"), + Effect.catchCause((cause) => { + if (warned) return Effect.void + warned = true + return Effect.logWarning("Failed to publish Code Mode progress", { cause }) + }), + ) let childCalls = 0 const callTool = (entry: CatalogEntry) => (input: unknown) => @@ -239,7 +255,7 @@ export const CodeModeTool = Tool.define( const runtime = CodeMode.make({ tools: toolTree(catalog, callTool), onToolCallStart: ({ index, name, input }) => - Effect.suspend(() => { + Effect.sync(() => { const shown = (() => { if (input === null || input === undefined) return if (typeof input === "object" && !Array.isArray(input)) { @@ -249,13 +265,13 @@ export const CodeModeTool = Tool.define( return { input } })() calls[index] = { tool: name, status: "running", ...(shown ? { input: shown } : {}) } - return publish() + version++ }), onToolCallEnd: ({ index, outcome }) => - Effect.suspend(() => { + Effect.sync(() => { const current = calls[index] if (current) calls[index] = { ...current, status: outcome === "success" ? "completed" : "error" } - return publish() + version++ }), }) @@ -271,7 +287,18 @@ export const CodeModeTool = Tool.define( toolCalls: calls.map((call) => ({ name: call.tool })), }) - const result = yield* Effect.raceFirst(runtime.execute(params.code), abort.pipe(Effect.map(cancelled))) + const result = yield* Effect.gen(function* () { + const updates = yield* Effect.gen(function* () { + while (true) { + yield* Effect.sleep("100 millis") + yield* publishProgress() + } + }).pipe(Effect.forkScoped) + const result = yield* Effect.raceFirst(runtime.execute(params.code), abort.pipe(Effect.map(cancelled))) + yield* Fiber.interrupt(updates) + yield* ctx.abort.aborted ? publishProgress() : Effect.raceFirst(publish(), abort) + return result + }).pipe(Effect.scoped) const logs = result.logs ?? [] const withLogs = (text: string) => { if (logs.length === 0) return text diff --git a/packages/opencode/test/tool/code-mode-integration.test.ts b/packages/opencode/test/tool/code-mode-integration.test.ts index 671acd896222..e1ba7854b42c 100644 --- a/packages/opencode/test/tool/code-mode-integration.test.ts +++ b/packages/opencode/test/tool/code-mode-integration.test.ts @@ -312,7 +312,7 @@ describe("code mode integration (real MCP server)", () => { expect(asked).toEqual(["fixtures_add", "fixtures_get_text"]) }) - test("streams running/completed metadata for child calls over a real transport", async () => { + test("publishes completed metadata for a quick child call over a real transport", async () => { const snapshots: Array<{ toolCalls: { tool: string; status: string; input?: Record }[] }> = [] const recordingCtx: Tool.Context = { ...ctx, @@ -321,10 +321,7 @@ describe("code mode integration (real MCP server)", () => { await Effect.runPromise( tool.execute({ code: "await tools.fixtures.add({ a: 1, b: 2 }); return 'done'" }, recordingCtx), ) - expect(snapshots).toContainEqual({ - toolCalls: [{ tool: "fixtures.add", status: "running", input: { a: 1, b: 2 } }], - }) - expect(snapshots).toContainEqual({ + expect(snapshots.at(-1)).toEqual({ toolCalls: [{ tool: "fixtures.add", status: "completed", input: { a: 1, b: 2 } }], }) }) diff --git a/packages/opencode/test/tool/code-mode.test.ts b/packages/opencode/test/tool/code-mode.test.ts index 34b3faa610d7..d789719474c2 100644 --- a/packages/opencode/test/tool/code-mode.test.ts +++ b/packages/opencode/test/tool/code-mode.test.ts @@ -462,7 +462,7 @@ describe("code mode execute", () => { expect(called).toEqual(["b"]) }) - test("streams live per-call metadata as a call starts and finishes", async () => { + test("publishes completed metadata for a quick child call", async () => { const snapshots: Array<{ toolCalls: { tool: string; status: string; input?: Record }[] }> = [] const recordingCtx: Tool.Context = { ...ctx, @@ -474,12 +474,130 @@ describe("code mode execute", () => { tool.execute({ code: "await tools.greeter.hello({ name: 'Ada' }); return 'done'" }, recordingCtx), ) - expect(snapshots).toContainEqual({ - toolCalls: [{ tool: "greeter.hello", status: "running", input: { name: "Ada" } }], - }) - expect(snapshots).toContainEqual({ + expect(snapshots.at(-1)).toEqual({ toolCalls: [{ tool: "greeter.hello", status: "completed", input: { name: "Ada" } }], }) + const count = snapshots.length + await Bun.sleep(150) + expect(snapshots).toHaveLength(count) + }) + + test("publishes running and completed metadata for a long child call", async () => { + const release = Promise.withResolvers() + const running = Promise.withResolvers() + const snapshots: Array<{ toolCalls: { tool: string; status: string }[] }> = [] + const recordingCtx: Tool.Context = { + ...ctx, + metadata: (val: any) => + Effect.sync(() => { + snapshots.push(val.metadata) + if (val.metadata.toolCalls.some((call: { status: string }) => call.status === "running")) running.resolve() + }), + } + const tool = await build({ + greeter_hello: mcpTool("hello", async () => { + await release.promise + return { content: [{ type: "text", text: "hi" }] } + }), + }) + + const execution = Effect.runPromise( + tool.execute({ code: "await tools.greeter.hello({}); return 'done'" }, recordingCtx), + ) + await running.promise + release.resolve() + await execution + + expect(snapshots).toContainEqual({ toolCalls: [{ tool: "greeter.hello", status: "running" }] }) + expect(snapshots.at(-1)).toEqual({ toolCalls: [{ tool: "greeter.hello", status: "completed" }] }) + }) + + test("discards an in-flight progress publish before the final snapshot", async () => { + const childRelease = Promise.withResolvers() + const publishStarted = Promise.withResolvers() + const snapshots: Array<{ toolCalls: { tool: string; status: string }[] }> = [] + let attempts = 0 + const recordingCtx: Tool.Context = { + ...ctx, + metadata: (val: any) => { + attempts++ + if (attempts > 1) return Effect.sync(() => void snapshots.push(val.metadata)) + return Effect.sync(() => publishStarted.resolve()).pipe(Effect.andThen(Effect.never)) + }, + } + const tool = await build({ + greeter_hello: mcpTool("hello", async () => { + await childRelease.promise + return { content: [{ type: "text", text: "hi" }] } + }), + }) + + const execution = Effect.runPromise( + tool.execute({ code: "await tools.greeter.hello({}); return 'done'" }, recordingCtx), + ) + await publishStarted.promise + childRelease.resolve() + const result = await execution + + expect(snapshots).toEqual([{ toolCalls: [{ tool: "greeter.hello", status: "completed" }] }]) + expect(result.metadata.toolCalls).toEqual([{ tool: "greeter.hello", status: "completed" }]) + }) + + test("coalesces metadata for a burst of child calls", async () => { + const snapshots: Array<{ toolCalls: { tool: string; status: string }[] }> = [] + const recordingCtx: Tool.Context = { + ...ctx, + metadata: (val: any) => Effect.sync(() => void snapshots.push(val.metadata)), + } + const tool = await build({ + greeter_hello: mcpTool("hello", () => ({ content: [{ type: "text", text: "hi" }] })), + }) + const calls = Array.from({ length: 20 }, () => "tools.greeter.hello({})").join(",") + + const result = await Effect.runPromise( + tool.execute({ code: `await Promise.all([${calls}]); return 'done'` }, recordingCtx), + ) + + expect(snapshots.length).toBeLessThan(40) + expect(snapshots.at(-1)?.toolCalls).toHaveLength(20) + expect(snapshots.at(-1)?.toolCalls.every((call) => call.status === "completed")).toBe(true) + expect(result.metadata.toolCalls).toHaveLength(20) + expect(result.metadata.toolCalls.every((call) => call.status === "completed")).toBe(true) + }) + + test("continues publishing after a transient metadata defect", async () => { + const running = Promise.withResolvers() + const release = Promise.withResolvers() + const snapshots: Array<{ toolCalls: { tool: string; status: string }[] }> = [] + let attempts = 0 + const recordingCtx: Tool.Context = { + ...ctx, + metadata: (val: any) => { + attempts++ + if (attempts === 1) return Effect.die("transient metadata failure") + return Effect.sync(() => { + snapshots.push(val.metadata) + if (val.metadata.toolCalls.some((call: { status: string }) => call.status === "running")) running.resolve() + }) + }, + } + const tool = await build({ + greeter_hello: mcpTool("hello", async () => { + await release.promise + return { content: [{ type: "text", text: "hi" }] } + }), + }) + + const execution = Effect.runPromise( + tool.execute({ code: "await tools.greeter.hello({}); return 'done'" }, recordingCtx), + ) + await running.promise + release.resolve() + const result = await execution + + expect(attempts).toBeGreaterThanOrEqual(3) + expect(snapshots.at(-1)).toEqual({ toolCalls: [{ tool: "greeter.hello", status: "completed" }] }) + expect(result.metadata.toolCalls).toEqual([{ tool: "greeter.hello", status: "completed" }]) }) test("marks a failed child call as error in the live metadata", async () => { @@ -603,6 +721,7 @@ describe("code mode execute", () => { test("cancelling via ctx.abort interrupts the running program", async () => { const controller = new AbortController() + const snapshots: Array<{ toolCalls: { tool: string; status: string }[] }> = [] const tool = await build({ host_trigger: mcpTool("trigger", () => { controller.abort() @@ -612,12 +731,62 @@ describe("code mode execute", () => { const output = await Effect.runPromise( tool.execute( { code: "try { await tools.host.trigger({}) } catch {} while (true) {}" }, - { ...ctx, abort: controller.signal }, + { + ...ctx, + abort: controller.signal, + metadata: (val: any) => Effect.sync(() => void snapshots.push(val.metadata)), + }, ), ) expect(output.output).toBe("Execution cancelled.") expect(output.metadata.error).toBe(true) expect(output.metadata.toolCalls).toEqual([{ tool: "host.trigger", status: "running" }]) + expect(snapshots.at(-1)).toEqual({ toolCalls: [{ tool: "host.trigger", status: "running" }] }) + }) + + test("cancelling via ctx.abort does not wait indefinitely for final metadata", async () => { + const controller = new AbortController() + const tool = await build({ + host_trigger: mcpTool("trigger", () => { + controller.abort() + return new Promise(() => {}) + }), + }) + + const output = await Effect.runPromise( + tool.execute( + { code: "await tools.host.trigger({})" }, + { ...ctx, abort: controller.signal, metadata: () => Effect.never }, + ), + ) + + expect(output.output).toBe("Execution cancelled.") + expect(output.metadata).toMatchObject({ toolCalls: [{ tool: "host.trigger", status: "running" }], error: true }) + }) + + test("a late ctx.abort interrupts a stalled final metadata publish", async () => { + const controller = new AbortController() + const publishStarted = Promise.withResolvers() + const tool = await build({ + host_complete: mcpTool("complete", () => ({ content: [{ type: "text", text: "done" }] })), + }) + + const execution = Effect.runPromise( + tool.execute( + { code: "return await tools.host.complete({})" }, + { + ...ctx, + abort: controller.signal, + metadata: () => Effect.sync(() => publishStarted.resolve()).pipe(Effect.andThen(Effect.never)), + }, + ), + ) + await publishStarted.promise + controller.abort() + const output = await execution + + expect(output.output).toBe("done") + expect(output.metadata.toolCalls).toEqual([{ tool: "host.complete", status: "completed" }]) }) test("a pre-aborted signal cancels before the program runs", async () => {