From b41da4e3210f5ea1c034d46d84ecc7328f41815d Mon Sep 17 00:00:00 2001 From: Fabian Csy Date: Thu, 8 Oct 2026 18:46:34 +0000 Subject: [PATCH] fix(opencode): do not read a tracked tool part for each streamed input delta Every tool-input-delta and tool-input-end went through ensureToolCall, which reads the stored part back: one database round trip per streamed chunk of a tool call's arguments, though neither event writes anything. A 10k-token document edit in production produced about 10,200 reads in 93 seconds. A call already in ctx.toolcalls has its part from tool-input-start, so those events now only create one when it is missing. The test streams 2,000 deltas (2,004 reads before, under 10 after) and also pins that a tool's error text reaches the model unchanged, as an error-text tool result. Co-Authored-By: Claude Opus 5.5 (1M context) --- documentor/FORK.md | 30 +-- packages/opencode/src/session/processor.ts | 7 +- .../test/pg/processor-tool-call.test.ts | 229 ++++++++++++++++++ 3 files changed, 249 insertions(+), 17 deletions(-) create mode 100644 packages/opencode/test/pg/processor-tool-call.test.ts diff --git a/documentor/FORK.md b/documentor/FORK.md index 316e1dd36f6e..b491b1851e2d 100644 --- a/documentor/FORK.md +++ b/documentor/FORK.md @@ -11,21 +11,21 @@ files it changes, because those are what a rebase meets. ## Upstream files changed -| File | What differs | Why | -| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| The 11 table files: `packages/core/src/{account,credential,event,permission,project,session,share}/sql.ts`, `control-plane/workspace.sql.ts`, `data-migration.sql.ts`, `database/schema.sql.ts`, `database/path.ts` | Postgres table builder; `count`, `jsonb` and `text` columns from `effect-drizzle-pg`; time columns are `double precision` | Written by `documentor/scripts/convert-schema.py`, never by hand | -| `packages/core/src/database/schema.gen.ts`, `packages/core/schema.json`, `packages/core/drizzle.config.ts` | The Postgres baseline, as one statement batch | Generated; see "Regenerating the baseline" | -| `packages/core/src/database/database.ts` | Connects with `@effect/sql-pg`; verifies collation and migrations and changes nothing; a throwaway schema per layer when `OPENCODE_DATABASE_EPHEMERAL=1` | The service itself | -| `packages/core/src/database/migration.ts` | `migrate`, `migrateAll`, `verify`, `pending`, `BASELINE_IDS`, `ported` | Schema work is a command, not a start-up side effect | -| `packages/core/src/project/directories.ts` | `Transaction` is `Database.Transaction` | It was typed against the SQLite client | -| `packages/core/src/flag/flag.ts` | No `OPENCODE_DB` | Nothing reads it | -| `packages/core/package.json` | Depends on `@effect/sql-pg`, `pg`, `@opencode-ai/effect-drizzle-pg`; tests run with `--timeout 30000` | Each test database is a real schema, about 110 ms to build | -| `packages/opencode/src/cli/cmd/db.ts` | `db `, `db migrate [--all]`, `db provision`, `db grant-read`, all outside the application runtime | The runtime's database service refuses the schemas these commands exist to create and migrate | -| `packages/opencode/src/session/processor.ts` | One lock per processor around each read-then-write of a tool part | The processor and the running tool both update a tool part. SQLite made each update atomic; a network database lets them interleave, and the tool's metadata was lost | -| `packages/core/src/integration.ts` | An OAuth attempt's credential is stored before the attempt reads as complete, inside one update of the attempts map | The attempt was marked complete first and the credential written after. On SQLite nothing could look in between; on a network database a caller that saw "complete" could list no credential | -| `patches/effect@4.0.0-beta.83.patch` | Adds Effect's own fix for re-entrant interruption, first released in `4.0.0-beta.100` | A fiber that woke another fiber which interrupted it synchronously had its continuation stack corrupted once its cleanup awaited anything. With Postgres that leaked a pooled connection. Drop these hunks when upstream moves to `beta.100` or later | -| `turbo.json` | `OPENCODE_TEST_DATABASE_URL` passes through to every task; `@opencode-ai/effect-drizzle-pg#test` is a task | turbo strips variables it is not told about | -| Test helpers: `packages/core/test/preload.ts`, `packages/opencode/test/preload.ts`, `packages/opencode/test/fixture/db.ts`, `packages/opencode/test/server/httpapi-exercise/environment.ts`, `packages/sdk-next/test/embedded.test.ts` | Ask for a throwaway schema; no database file to delete | | +| File | What differs | Why | +| -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| The 11 table files: `packages/core/src/{account,credential,event,permission,project,session,share}/sql.ts`, `control-plane/workspace.sql.ts`, `data-migration.sql.ts`, `database/schema.sql.ts`, `database/path.ts` | Postgres table builder; `count`, `jsonb` and `text` columns from `effect-drizzle-pg`; time columns are `double precision` | Written by `documentor/scripts/convert-schema.py`, never by hand | +| `packages/core/src/database/schema.gen.ts`, `packages/core/schema.json`, `packages/core/drizzle.config.ts` | The Postgres baseline, as one statement batch | Generated; see "Regenerating the baseline" | +| `packages/core/src/database/database.ts` | Connects with `@effect/sql-pg`; verifies collation and migrations and changes nothing; a throwaway schema per layer when `OPENCODE_DATABASE_EPHEMERAL=1` | The service itself | +| `packages/core/src/database/migration.ts` | `migrate`, `migrateAll`, `verify`, `pending`, `BASELINE_IDS`, `ported` | Schema work is a command, not a start-up side effect | +| `packages/core/src/project/directories.ts` | `Transaction` is `Database.Transaction` | It was typed against the SQLite client | +| `packages/core/src/flag/flag.ts` | No `OPENCODE_DB` | Nothing reads it | +| `packages/core/package.json` | Depends on `@effect/sql-pg`, `pg`, `@opencode-ai/effect-drizzle-pg`; tests run with `--timeout 30000` | Each test database is a real schema, about 110 ms to build | +| `packages/opencode/src/cli/cmd/db.ts` | `db `, `db migrate [--all]`, `db provision`, `db grant-read`, all outside the application runtime | The runtime's database service refuses the schemas these commands exist to create and migrate | +| `packages/opencode/src/session/processor.ts` | One lock per processor around each read-then-write of a tool part; `tool-input-delta` and `tool-input-end` do not read a part the processor already tracks | The processor and the running tool both update a tool part. SQLite made each update atomic; a network database lets them interleave, and the tool's metadata was lost. Each read is a round trip, and a large edit streams thousands of deltas | +| `packages/core/src/integration.ts` | An OAuth attempt's credential is stored before the attempt reads as complete, inside one update of the attempts map | The attempt was marked complete first and the credential written after. On SQLite nothing could look in between; on a network database a caller that saw "complete" could list no credential | +| `patches/effect@4.0.0-beta.83.patch` | Adds Effect's own fix for re-entrant interruption, first released in `4.0.0-beta.100` | A fiber that woke another fiber which interrupted it synchronously had its continuation stack corrupted once its cleanup awaited anything. With Postgres that leaked a pooled connection. Drop these hunks when upstream moves to `beta.100` or later | +| `turbo.json` | `OPENCODE_TEST_DATABASE_URL` passes through to every task; `@opencode-ai/effect-drizzle-pg#test` is a task | turbo strips variables it is not told about | +| Test helpers: `packages/core/test/preload.ts`, `packages/opencode/test/preload.ts`, `packages/opencode/test/fixture/db.ts`, `packages/opencode/test/server/httpapi-exercise/environment.ts`, `packages/sdk-next/test/embedded.test.ts` | Ask for a throwaway schema; no database file to delete | | Upstream tests that are changed or deleted are listed, with reasons, in [`upstream-test-exceptions.md`](upstream-test-exceptions.md). diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 9cfdbd9c1b1b..6c0a302b7d7d 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -333,12 +333,15 @@ const layer = Layer.effect( yield* ensureToolCall(value) return + // A call already in ctx.toolcalls has its part from tool-input-start, and + // neither event writes to it. Reading it back for each delta costs one + // database round trip per streamed chunk: thousands for a large edit. case "tool-input-delta": - yield* ensureToolCall(value) + if (!(value.id in ctx.toolcalls)) yield* ensureToolCall(value) return case "tool-input-end": { - yield* ensureToolCall(value) + if (!(value.id in ctx.toolcalls)) yield* ensureToolCall(value) return } diff --git a/packages/opencode/test/pg/processor-tool-call.test.ts b/packages/opencode/test/pg/processor-tool-call.test.ts new file mode 100644 index 000000000000..b639d5cd4b90 --- /dev/null +++ b/packages/opencode/test/pg/processor-tool-call.test.ts @@ -0,0 +1,229 @@ +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { Database } from "@opencode-ai/core/database/database" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { ModelV2 } from "@opencode-ai/core/model" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { LLMEvent } from "@opencode-ai/llm" +import { expect } from "bun:test" +import { Effect, Layer, Stream } from "effect" +import path from "path" +import type { Agent } from "../../src/agent/agent" +import { EventV2Bridge } from "@/event-v2-bridge" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { Provider } from "@/provider/provider" +import { Session } from "@/session/session" +import { LLM } from "../../src/session/llm" +import { MessageV2 } from "../../src/session/message-v2" +import { SessionProcessor } from "../../src/session/processor" +import { MessageID, PartID, SessionID } from "../../src/session/schema" +import { SessionStatus } from "../../src/session/status" +import { SessionSummary } from "../../src/session/summary" +import { provideTmpdirInstance } from "../fixture/fixture" +import { testEffect } from "../lib/effect" + +// A tool call's arguments stream as one delta per chunk the provider sends: +// thousands for a large edit. The processor keeps the call in memory from +// tool-input-start on, so a delta must not read the stored part — on Postgres +// each read is a round trip. + +const ref = { providerID: ProviderV2.ID.make("test"), modelID: ModelV2.ID.make("test-model") } + +const cfg = { + provider: { + test: { + name: "Test", + id: "test", + env: [], + npm: "@ai-sdk/openai-compatible", + models: { + "test-model": { + id: "test-model", + name: "Test Model", + attachment: false, + reasoning: false, + temperature: false, + tool_call: true, + release_date: "2025-01-01", + limit: { context: 100000, output: 10000 }, + cost: { input: 0, output: 0 }, + options: {}, + }, + }, + options: { apiKey: "test-key", baseURL: "http://localhost:1/v1" }, + }, + }, +} + +const DELTAS = 2000 + +// What kpg-agent-tools returns when it refuses a batch of proposals; the MCP +// client turns an isError result into an Error carrying this text. +const REFUSAL = JSON.stringify({ + success: false, + error: "Invalid document proposals. Revise the listed changes and call this tool again; do not apply them directly.", + errors: Array.from( + { length: 9 }, + (_, index) => + `workspace:notes.md: files[0].changes[${index * 2 + 1}] and files[0].changes[${index * 2 + 2}] overlap. Use separate non-overlapping original passages, or combine dependent edits into one suggestion.`, + ), + checkedFiles: ["workspace:notes.md"], + uncheckedFiles: [], +}) + +const scripted = Layer.succeed( + LLM.Service, + LLM.Service.of({ + stream: () => + Stream.make( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolInputStart({ id: "call-1", name: "propose_document_changes" }), + ...Array.from({ length: DELTAS }, () => + LLMEvent.toolInputDelta({ id: "call-1", name: "propose_document_changes", text: "x" }), + ), + LLMEvent.toolInputEnd({ id: "call-1", name: "propose_document_changes" }), + LLMEvent.toolCall({ id: "call-1", name: "propose_document_changes", input: { files: [] } }), + LLMEvent.toolError({ + id: "call-1", + name: "propose_document_changes", + message: REFUSAL, + error: new Error(REFUSAL), + }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + LLMEvent.finish({ reason: "tool-calls" }), + ), + }), +) + +const summary = Layer.succeed( + SessionSummary.Service, + SessionSummary.Service.of({ + summarize: () => Effect.void, + diff: () => Effect.succeed([]), + computeDiff: () => Effect.succeed([]), + }), +) + +const it = testEffect( + LayerNode.compile( + LayerNode.group([ + SessionProcessor.node, + Session.node, + SessionProjector.node, + Provider.node, + Database.node, + EventV2Bridge.node, + SessionStatus.node, + CrossSpawnSpawner.node, + ]), + [ + [SessionSummary.node, summary], + [RuntimeFlags.node, RuntimeFlags.layer({ experimentalEventSystem: true })], + [LLM.node, scripted], + ], + ), +) + +const agent: Agent.Info = { + name: "build", + mode: "primary", + options: {}, + permission: [{ permission: "*", pattern: "*", action: "allow" }], +} + +it.live("a streamed tool call reads its part a bounded number of times, and its error reaches the model whole", () => + provideTmpdirInstance( + (dir) => + Effect.gen(function* () { + const processors = yield* SessionProcessor.Service + const session = yield* Session.Service + const provider = yield* Provider.Service + + const chat = yield* session.create({}) + const parent = yield* session.updateMessage({ + id: MessageID.ascending(), + role: "user", + sessionID: chat.id, + agent: "build", + model: ref, + time: { created: Date.now() }, + }) + yield* session.updatePart({ + id: PartID.ascending(), + messageID: parent.id, + sessionID: chat.id, + type: "text", + text: "revise the notes", + }) + const msg = yield* session.updateMessage({ + id: MessageID.ascending(), + role: "assistant", + sessionID: chat.id, + mode: "build", + agent: "build", + path: { cwd: path.resolve(dir), root: path.resolve(dir) }, + cost: 0, + tokens: { total: 0, input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + modelID: ref.modelID, + providerID: ref.providerID, + parentID: parent.id, + time: { created: Date.now() }, + finish: "end_turn", + } satisfies SessionV1.Assistant) + const model = yield* provider.getModel(ref.providerID, ref.modelID) + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model }) + + // The processor holds this same service object, so wrapping the method counts its reads. + const getPart = session.getPart + let reads = 0 + Object.assign(session, { + getPart: (input: Parameters[0]) => { + reads++ + return getPart(input) + }, + }) + yield* handle.process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: ref, + } satisfies SessionV1.User, + sessionID: chat.id, + model, + agent, + system: [], + messages: [{ role: "user", content: "revise the notes" }], + tools: {}, + }) + Object.assign(session, { getPart }) + + expect(reads).toBeLessThan(10) + + const parts = yield* MessageV2.parts(msg.id) + const call = parts.find((part): part is SessionV1.ToolPart => part.type === "tool") + expect(call?.state.status).toBe("error") + if (call?.state.status === "error") expect(call.state.error).toBe(REFUSAL) + + const userParts = yield* MessageV2.parts(parent.id) + const messages = yield* Effect.promise(() => + MessageV2.toModelMessages( + [ + { info: parent, parts: userParts }, + { info: msg, parts }, + ], + model, + ), + ) + const results = messages.flatMap((message) => + message.role === "tool" ? message.content.filter((content) => content.type === "tool-result") : [], + ) + expect(results).toHaveLength(1) + expect(results[0].output).toEqual({ type: "error-text", value: REFUSAL }) + }), + { config: cfg }, + ), +)