diff --git a/devlog/_plan/260926_bug_train_6/040_batch8.md b/devlog/_plan/260926_bug_train_6/040_batch8.md new file mode 100644 index 00000000000..9e2195d9773 --- /dev/null +++ b/devlog/_plan/260926_bug_train_6/040_batch8.md @@ -0,0 +1,28 @@ +# Batch 8 — plan + +Previous D: #5936 (`bb3f3c2d0d`) fixed the title bar layout defects and the `tests/cli` batch hang. The next step is to +carry the new bug PRs that the post-merge triage (Sol, read-only) classified as carry-with-fix. + +## Carry + +- #5929 (@mdwsk88): CodeBuddy parallel tool-use blocks are serialized without corrupting calls. Prepared on + `codex/bug-train-8-prep`: `5edec14b73` (squashed carry with author and `Co-authored-by`) and `df7ca1df3d` (P1 fix). + The P1 was a bridge-init check that ran when a buffered tool call closed. It now runs on the raw start frame, before + buffering. The start → init → stop regression is red before the fix and green after it. +- #5929 audit follow-up: with more than one tool block open, an indexless tool-argument delta fails the turn instead + of being dropped (the parser's fail-closed contract). +- CI: the `macos control` lane hung in `tests/ci-workflows/ci-privacy-gate.test.ts` (`spawnSync` child) on `dev` at + `bb3f3c2d0d`; same bounded treatment as #5936 if the logs confirm the same class. + +## Left out + +- #5927: the independent security review failed, so it is not carried. The finding is kept in scratch space for the + maintainers, not in this tracked plan. +- #5925 duplicates #5929's four CodeBuddy files, and its MCP-only half needs a rebase and its own security review for + undeclared-tool admission. +- #5926 and #5928 are owner PRs. + +## Check + +Focused adapter and image suites, layout, file-size and structure guards, tsc, privacy. The #5927 security review is +recorded on the batch PR. Exact-head CI, then `--admin --match-head-commit`. Close the carried PRs with the batch note. diff --git a/src/adapters/coding-agent/protocol.ts b/src/adapters/coding-agent/protocol.ts index cccbb05db58..71085e30f47 100644 --- a/src/adapters/coding-agent/protocol.ts +++ b/src/adapters/coding-agent/protocol.ts @@ -222,9 +222,23 @@ export interface StreamParseState { sawPartialText: boolean; sawPartialThinking: boolean; sawTerminalResult: boolean; - openToolCallId?: string; /** A `message_stop` stream event arrived: the assistant message is complete. */ sawMessageStop?: boolean; + /** + * Open tool_use blocks keyed by content-block index. CodeBuddy parallel tool calls arrive + * as several tool_use blocks on ONE shared content-block index — intermediate blocks never + * receive a stop and only the final block does — while deltas for different indices (for + * example a long thinking block) interleave freely (observed 2026-09-25/26: four parallel + * calls, one counted stop, "incomplete tool call" 502 at message_stop). A single open-call + * slot both mis-attributes argument fragments and miscounts completions. Downstream + * assembly keeps a single open call, so each block is buffered and emitted atomically: + * when its own stop arrives, or when a new tool_use start reuses its index. + */ + openToolBlocks?: Map; + /** Synthetic decreasing keys for tool_use start frames that omit the block index. */ + nextSyntheticToolBlockKey?: number; + /** Tool_use blocks opened in this stream, whether or not they have closed yet. */ + toolBlockStarts?: number; /** Completed tool_use content blocks observed in this stream. */ completedToolCalls?: number; /** Tool IDs already captured through partial events, for complete-assistant deduplication. */ @@ -332,6 +346,53 @@ export function mapStreamMessageToEvents(message: StreamMessage, state: StreamPa return events; } +/** One in-flight tool_use block: identity plus its buffered argument fragments. */ +export interface OpenToolBlock { + id: string; + name: string; + argParts: string[]; +} + +/** Key a tool_use start frame by content-block index, falling back to a synthetic key. */ +function toolBlockKey(state: StreamParseState, event: StreamMessage): number { + const index = event.index; + if (typeof index === "number" && Number.isInteger(index)) return index; + const key = state.nextSyntheticToolBlockKey ?? -1; + state.nextSyntheticToolBlockKey = key - 1; + return key; +} + +/** + * Resolve a delta/stop frame to an open tool block. An indexed frame only matches a block + * opened under the same index — CodeBuddy skips stop frames for thinking blocks, and such a + * stop must not close a tool block that happens to be open. An index-less frame resolves + * only when exactly one block is open. Ambiguous argument deltas are rejected before + * resolution; an unmatched stop cannot close a tool block. + */ +function resolveToolBlockKey(state: StreamParseState, event: StreamMessage): number | undefined { + const index = event.index; + if (typeof index === "number" && Number.isInteger(index)) { + return state.openToolBlocks?.has(index) ? index : undefined; + } + const blocks = state.openToolBlocks; + if (!blocks || blocks.size !== 1) return undefined; + return blocks.keys().next().value; +} + +/** + * Emit a closed block atomically — start, the buffered fragments in arrival order, end — + * so the strictly sequential downstream bridge never sees two calls open at once. + */ +function closeToolBlock(state: StreamParseState, key: number, events: AdapterEvent[]): void { + const block = state.openToolBlocks?.get(key); + if (!block || !state.openToolBlocks) return; + state.openToolBlocks.delete(key); + events.push({ type: "tool_call_start", id: block.id, name: block.name }); + for (const part of block.argParts) events.push({ type: "tool_call_delta", arguments: part }); + events.push({ type: "tool_call_end" }); + state.completedToolCalls = (state.completedToolCalls ?? 0) + 1; +} + /** Map a raw Anthropic SSE event (carried inside a `stream_event` frame) to AdapterEvents. */ function mapRawStreamEvent(event: StreamMessage, state: StreamParseState): AdapterEvent[] { const events: AdapterEvent[] = []; @@ -353,11 +414,22 @@ function mapRawStreamEvent(event: StreamMessage, state: StreamParseState): Adapt events.push({ type: "thinking_delta", thinking }); } } else if (deltaType === "input_json_delta") { - // Tool-input streaming. Live for capture-only bridge turns, where the advertised MCP - // catalog makes the CLI emit real tool_use blocks; parsed unconditionally so a stray - // frame on a tools-disabled turn is ignored rather than crashing. + // Tool-input streaming. Fragments are buffered under their own block index because + // CodeBuddy alternates deltas across interleaved parallel blocks; parsed + // unconditionally so a stray frame on a tools-disabled turn is ignored rather than + // crashing. + if ( + (state.openToolBlocks?.size ?? 0) > 1 + && (typeof event.index !== "number" || !Number.isInteger(event.index)) + ) { + throw new CodingAgentProtocolError("Coding-agent CLI sent an unindexed tool argument delta with multiple tool blocks open."); + } const partial = asString(delta?.partial_json); - if (partial && state.openToolCallId) events.push({ type: "tool_call_delta", arguments: partial }); + if (partial) { + const key = resolveToolBlockKey(state, event); + const block = key === undefined ? undefined : state.openToolBlocks?.get(key); + if (block) block.argParts.push(partial); + } } return events; } @@ -368,20 +440,27 @@ function mapRawStreamEvent(event: StreamMessage, state: StreamParseState): Adapt const id = asString(block?.id) ?? ""; const name = asString(block?.name) ?? "tool"; if (id) { - state.openToolCallId = id; + const key = toolBlockKey(state, event); + if (state.openToolBlocks?.has(key)) { + // CodeBuddy reuses one content-block index for a parallel batch: every call in the + // batch starts on the same index, intermediate blocks never receive a stop, and only + // the final block does (observed 2026-09-26: START 2 alpha, A's complete args, START 2 + // beta, B's complete args, one STOP 2). Parallel calls stream their arguments + // sequentially — never interleaved — so the block already open on this index is + // complete, and the new start implicitly closes it. + closeToolBlock(state, key, events); + } + (state.openToolBlocks ??= new Map()).set(key, { id, name, argParts: [] }); + state.toolBlockStarts = (state.toolBlockStarts ?? 0) + 1; state.partialToolCallIds?.add(id); - events.push({ type: "tool_call_start", id, name }); } } return events; } if (eventType === "content_block_stop") { - if (state.openToolCallId) { - state.openToolCallId = undefined; - state.completedToolCalls = (state.completedToolCalls ?? 0) + 1; - events.push({ type: "tool_call_end" }); - } + const key = resolveToolBlockKey(state, event); + if (key !== undefined) closeToolBlock(state, key, events); return events; } diff --git a/src/adapters/coding-agent/turn.ts b/src/adapters/coding-agent/turn.ts index 67014960621..af9c6de946e 100644 --- a/src/adapters/coding-agent/turn.ts +++ b/src/adapters/coding-agent/turn.ts @@ -380,7 +380,7 @@ export async function runCodingAgentTurn(input: CodingAgentTurnInput): Promise() : undefined, }; @@ -422,6 +422,29 @@ export async function runCodingAgentTurn(input: CodingAgentTurnInput): Promise + : undefined; + const rawBlock = rawEvent?.content_block; + if ( + !initValidated + && rawEvent?.type === "content_block_start" + && rawBlock !== null + && typeof rawBlock === "object" + && !Array.isArray(rawBlock) + && (rawBlock as Record).type === "tool_use" + ) { + emitOnce({ + type: "error", + message: "Coding-agent CLI called a tool before the tool bridge init handshake completed.", + status: 502, + errorType: "upstream_error", + code: "tool_bridge_init_missing", + retryable: false, + }); + kill(); + break; + } } const mappedEvents = mapStreamMessageToEvents(message, state); if (toolBridge && state.uncapturedToolUse) { @@ -451,24 +474,6 @@ export async function runCodingAgentTurn(input: CodingAgentTurnInput): Promise toolBridge.maxTurnToolCalls) { emitOnce({ @@ -527,8 +532,8 @@ export async function runCodingAgentTurn(input: CodingAgentTurnInput): Promise 0 - && (state.completedToolCalls ?? 0) !== toolCallStarts + && (state.toolBlockStarts ?? 0) > 0 + && (state.completedToolCalls ?? 0) !== (state.toolBlockStarts ?? 0) ) { // A terminal result that arrives while a captured tool call is still open must not // become a successful completion the client can accept. The message_stop check after @@ -552,8 +557,8 @@ export async function runCodingAgentTurn(input: CodingAgentTurnInput): Promise 0 - && (state.completedToolCalls ?? 0) === toolCallStarts + && (state.toolBlockStarts ?? 0) > 0 + && (state.completedToolCalls ?? 0) === (state.toolBlockStarts ?? 0) ) { // Every captured call completed and the CLI settled with a successful result before // message_stop (instead of parking on the never-answering capture server). Emitting @@ -573,8 +578,8 @@ export async function runCodingAgentTurn(input: CodingAgentTurnInput): Promise 0 - && (state.completedToolCalls ?? 0) !== toolCallStarts + && (state.toolBlockStarts ?? 0) > 0 + && (state.completedToolCalls ?? 0) !== (state.toolBlockStarts ?? 0) ) { emitOnce({ type: "error", @@ -588,8 +593,8 @@ export async function runCodingAgentTurn(input: CodingAgentTurnInput): Promise 0) { - // No init re-check here: a completed call implies a tool_call_start was mapped, and the - // arrival-time gate above already refuses any start that lands before the handshake. + // Raw tool_use starts are gated before buffering, so every completed call was admitted + // after the bridge init handshake. // The capture-only MCP handler never answers, so the CLI parks after message_stop. // The completed tool_use blocks are this turn's structured output: end the leg here // and terminate the tree; the client executes, and the next request continues. diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index 9871c973ec4..8f6309a4a2c 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -1,5 +1,14 @@ # Providers And Adapters +The coding-agent stream parser buffers each tool-use block by its content-block index +and emits a complete start/delta/end sequence on closure. A new start on an occupied +index closes the previous block; distinct indices can interleave. Turn completion +requires every opened block to close, preserving the downstream single-open-call contract. +An indexless argument delta belongs to the sole open block; with multiple blocks open, +the parser fails the turn before releasing their buffered calls. +The capture-only bridge checks each raw tool-use start against the init handshake before +buffering; a later init cannot authorize a call that started earlier. + RunTurn hosted search uses `src/web-search/run-turn-loop.ts`: synthetic calls remain private, progress reaches the bridge during collection, and a validated terminal precedes search execution. Complete search calls remain actionable at a truncated `done`; cancellation prevents subsequent queries and calls. OAuth preflight replay in `src/server/responses/run-turn-execution.ts` retains the synthetic tool while refreshing credential-scoped route state. In `src/server/responses/sidecar-execution.ts`, a search plan takes priority over image/video bridge execution for both transports; only fetch-capable adapters enter the fetch search loop. Combo preflight allows the private search tool only while a search plan is active; client tool declaration checks and replay-unsafe heartbeat protection remain enforced. diff --git a/tests/ci-workflows/ci-privacy-gate.test.ts b/tests/ci-workflows/ci-privacy-gate.test.ts index 365062a9119..fbe2e019d17 100644 --- a/tests/ci-workflows/ci-privacy-gate.test.ts +++ b/tests/ci-workflows/ci-privacy-gate.test.ts @@ -1,5 +1,5 @@ -import { describe, expect, test } from "bun:test"; -import { spawnSync } from "node:child_process"; +import { describe, expect, setDefaultTimeout, test } from "bun:test"; +import { spawn } from "node:child_process"; import { readFileSync } from "node:fs"; import { repoPath } from "../helpers/repo-root"; @@ -45,6 +45,9 @@ const producers: string[] = Array.isArray(aggregateNeeds) ? aggregateNeeds : []; // runner has no /bin/bash, so executing them there would test the host. const cannotRunShell = process.platform === "win32"; const cannotRunAggregate = cannotRunShell || !Bun.which("jq"); +const AGGREGATE_CHILD_DEADLINE_MS = 5_000; + +setDefaultTimeout(AGGREGATE_CHILD_DEADLINE_MS + 5_000); const scanners = Object.entries(jobs) .filter(([, job]) => (job.steps ?? []).some(step => step.run?.includes("bun run privacy:scan"))) @@ -100,37 +103,67 @@ describe("the privacy scan selection", () => { }); describe.skipIf(cannotRunAggregate)("the aggregate ci gate, executed", () => { - const runAggregate = (scope: Record, results: Record) => spawnSync("bash", ["-c", aggregateStep!.run!], { - encoding: "utf8", - env: { - PATH: process.env.PATH ?? "/usr/bin:/bin", - EVENT_NAME: "pull_request", - LANE: "", - CHANGES_CI: "false", - CHANGES_NATIVE: "false", - CHANGES_DESKTOP: "false", - CHANGES_PACKAGING: "false", - CHANGES_DOCS: "false", - CHANGES_STRUCTURE: "false", - CHANGES_SETUP_ACTION: "false", - CHANGES_REMOTE_HELPER: "false", - ...scope, - // needs serializes as an object per job; the gate reads .value.result. - RESULTS: JSON.stringify(Object.fromEntries(Object.entries(results).map(([job, result]) => [job, { result }]))), - }, - timeout: 5_000, + const runAggregate = ( + scope: Record, + results: Record, + script = aggregateStep!.run!, + deadlineMs = AGGREGATE_CHILD_DEADLINE_MS, + ): Promise<{ status: number | null; stdout: string; stderr: string }> => new Promise((resolve, reject) => { + const child = spawn("bash", ["-c", script], { + stdio: ["ignore", "pipe", "pipe"], + env: { + PATH: process.env.PATH ?? "/usr/bin:/bin", + EVENT_NAME: "pull_request", + LANE: "", + CHANGES_CI: "false", + CHANGES_NATIVE: "false", + CHANGES_DESKTOP: "false", + CHANGES_PACKAGING: "false", + CHANGES_DOCS: "false", + CHANGES_STRUCTURE: "false", + CHANGES_SETUP_ACTION: "false", + CHANGES_REMOTE_HELPER: "false", + ...scope, + // needs serializes as an object per job; the gate reads .value.result. + RESULTS: JSON.stringify(Object.fromEntries(Object.entries(results).map(([job, result]) => [job, { result }]))), + }, + }); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8").on("data", chunk => { stdout += chunk; }); + child.stderr.setEncoding("utf8").on("data", chunk => { stderr += chunk; }); + const deadline = setTimeout(() => { + child.kill("SIGKILL"); + reject(new Error(`aggregate ci gate child exceeded ${deadlineMs}ms`)); + }, deadlineMs); + child.once("error", error => { + clearTimeout(deadline); + reject(error); + }); + child.once("close", status => { + clearTimeout(deadline); + resolve({ status, stdout, stderr }); + }); }); const resultsWith = (succeeded: string[]): Record => Object.fromEntries(producers.map(job => [job, succeeded.includes(job) ? "success" : "skipped"])); - test("is green on a pull request that matches no filter only when the privacy gate ran", () => { + test("a hung aggregate child reports its own deadline", async () => { + await expect(runAggregate({}, {}, "while :; do :; done", 100)) + .rejects.toThrow("aggregate ci gate child exceeded 100ms"); + }); + + test("is green on a pull request that matches no filter only when the privacy gate ran", async () => { expect(producers).toContain("privacy-gate"); // The two unconditional producers plus the privacy gate, and nothing else. - const ran = runAggregate({}, resultsWith(["changes", "select-windows-runner", "privacy-gate"])); + let eventLoopAdvanced = false; + setTimeout(() => { eventLoopAdvanced = true; }, 0); + const ran = await runAggregate({}, resultsWith(["changes", "select-windows-runner", "privacy-gate"])); + expect(eventLoopAdvanced).toBe(true); expect(`status:${ran.status}`, ran.stdout + ran.stderr).toBe("status:0"); for (const result of ["skipped", "failure", "cancelled"]) { - const run = runAggregate({}, { + const run = await runAggregate({}, { ...resultsWith(["changes", "select-windows-runner"]), "privacy-gate": result, }); @@ -139,15 +172,15 @@ describe.skipIf(cannotRunAggregate)("the aggregate ci gate, executed", () => { } }); - test("requires the privacy gate alongside a narrow job on a docs-only pull request", () => { + test("requires the privacy gate alongside a narrow job on a docs-only pull request", async () => { // docs-site-build is pull-request scope like privacy-gate: the complement // must still request the scan, and a gate that did not succeed fails by name. const docsOnly = { CHANGES_DOCS: "true" }; - const ran = runAggregate(docsOnly, resultsWith(["changes", "select-windows-runner", "docs-site-build", "privacy-gate"])); + const ran = await runAggregate(docsOnly, resultsWith(["changes", "select-windows-runner", "docs-site-build", "privacy-gate"])); expect(`status:${ran.status}`, ran.stdout + ran.stderr).toBe("status:0"); for (const result of ["skipped", "failure", "cancelled"]) { - const run = runAggregate(docsOnly, { + const run = await runAggregate(docsOnly, { ...resultsWith(["changes", "select-windows-runner", "docs-site-build"]), "privacy-gate": result, }); @@ -156,14 +189,14 @@ describe.skipIf(cannotRunAggregate)("the aggregate ci gate, executed", () => { } }); - test("rejects a second scan on a pull request that gates already scans", () => { + test("rejects a second scan on a pull request that gates already scans", async () => { // Where ci is true, gates scans; a privacy gate that also ran means its // condition and this table have drifted apart. const both = { CHANGES_CI: "true" }; - const doubled = runAggregate(both, { ...resultsWith([]), "privacy-gate": "success" }); + const doubled = await runAggregate(both, { ...resultsWith([]), "privacy-gate": "success" }); expect(doubled.status).toBe(1); expect(doubled.stdout).toContain("privacy-gate was not requested by pull_request but reported 'success'"); - const single = runAggregate(both, resultsWith([])); + const single = await runAggregate(both, resultsWith([])); // The step prints RESULTS first, so look for the verdict line, not the job name. expect(single.stdout).not.toContain("privacy-gate was"); }); diff --git a/tests/providers/codebuddy-protocol.test.ts b/tests/providers/codebuddy-protocol.test.ts index 3cf7690156b..ffafe2966c4 100644 --- a/tests/providers/codebuddy-protocol.test.ts +++ b/tests/providers/codebuddy-protocol.test.ts @@ -232,20 +232,83 @@ describe("codebuddy stream-json event mapping", () => { }); test("parses tool_use blocks defensively even though v1 disables tools", () => { - const state = { sawPartialText: false, sawPartialThinking: false, sawTerminalResult: false, openToolCallId: undefined as string | undefined }; + const state = { sawPartialText: false, sawPartialThinking: false, sawTerminalResult: false }; const start = mapStreamMessageToEvents( { type: "stream_event", event: { type: "content_block_start", content_block: { type: "tool_use", id: "t1", name: "exec" } } }, state, ); - expect(start).toEqual([{ type: "tool_call_start", id: "t1", name: "exec" }]); + // Tool blocks are buffered and emitted atomically at their own stop: downstream keeps a + // single open call, and CodeBuddy interleaves parallel blocks. + expect(start).toEqual([]); const delta = mapStreamMessageToEvents( { type: "stream_event", event: { type: "content_block_delta", delta: { type: "input_json_delta", partial_json: "{\"a\":1}" } } }, state, ); - expect(delta).toEqual([{ type: "tool_call_delta", arguments: "{\"a\":1}" }]); + expect(delta).toEqual([]); const stop = mapStreamMessageToEvents({ type: "stream_event", event: { type: "content_block_stop" } }, state); - expect(stop).toEqual([{ type: "tool_call_end" }]); - expect(state.openToolCallId).toBeUndefined(); + expect(stop).toEqual([ + { type: "tool_call_start", id: "t1", name: "exec" }, + { type: "tool_call_delta", arguments: "{\"a\":1}" }, + { type: "tool_call_end" }, + ]); + expect(state.completedToolCalls).toBe(1); + expect(state.openToolBlocks?.size ?? 0).toBe(0); + }); + + test("interleaved parallel tool_use blocks are serialized per block index", () => { + const state = { sawPartialText: false, sawPartialThinking: false, sawTerminalResult: false }; + const feed = (event: unknown) => mapStreamMessageToEvents({ type: "stream_event", event: event as Record }, state); + const startAt = (index: number, id: string) => + feed({ type: "content_block_start", index, content_block: { type: "tool_use", id, name: "exec" } }); + const deltaAt = (index: number, part: string) => + feed({ type: "content_block_delta", index, delta: { type: "input_json_delta", partial_json: part } }); + + expect(startAt(1, "tu_a")).toEqual([]); + expect(startAt(2, "tu_b")).toEqual([]); + expect(deltaAt(1, "{\"cmd\":\"a")).toEqual([]); + expect(deltaAt(2, "{\"cmd\":\"b")).toEqual([]); + expect(deltaAt(1, "\"}")).toEqual([]); + // A stop for a non-tool block must not close an open tool block. + expect(feed({ type: "content_block_stop", index: 0 })).toEqual([]); + expect(feed({ type: "content_block_stop", index: 2 })).toEqual([ + { type: "tool_call_start", id: "tu_b", name: "exec" }, + { type: "tool_call_delta", arguments: "{\"cmd\":\"b" }, + { type: "tool_call_end" }, + ]); + expect(feed({ type: "content_block_stop", index: 1 })).toEqual([ + { type: "tool_call_start", id: "tu_a", name: "exec" }, + { type: "tool_call_delta", arguments: "{\"cmd\":\"a" }, + { type: "tool_call_delta", arguments: "\"}" }, + { type: "tool_call_end" }, + ]); + expect(state.toolBlockStarts).toBe(2); + expect(state.completedToolCalls).toBe(2); + }); + + test("a parallel batch reuses one block index; a new start implicitly closes the open block", () => { + // Live capture 2026-09-26 (CodeBuddy 2.158.0, kimi-k3-1, two parallel calls): START idx=2 + // alpha, alpha's complete args, START idx=2 beta (alpha never stopped), beta's complete + // args, one STOP idx=2, message_stop. A start that reuses an open block's index closes + // that block — parallel argument streams are sequential, so the open block is complete. + const state = { sawPartialText: false, sawPartialThinking: false, sawTerminalResult: false }; + const feed = (event: unknown) => mapStreamMessageToEvents({ type: "stream_event", event: event as Record }, state); + + expect(feed({ type: "content_block_start", index: 2, content_block: { type: "tool_use", id: "tu_a", name: "alpha" } })).toEqual([]); + expect(feed({ type: "content_block_delta", index: 2, delta: { type: "input_json_delta", partial_json: "{\"value\":\"A\"}" } })).toEqual([]); + expect(feed({ type: "content_block_start", index: 2, content_block: { type: "tool_use", id: "tu_b", name: "beta" } })).toEqual([ + { type: "tool_call_start", id: "tu_a", name: "alpha" }, + { type: "tool_call_delta", arguments: "{\"value\":\"A\"}" }, + { type: "tool_call_end" }, + ]); + expect(feed({ type: "content_block_delta", index: 2, delta: { type: "input_json_delta", partial_json: "{\"value\":\"B\"}" } })).toEqual([]); + expect(feed({ type: "content_block_stop", index: 2 })).toEqual([ + { type: "tool_call_start", id: "tu_b", name: "beta" }, + { type: "tool_call_delta", arguments: "{\"value\":\"B\"}" }, + { type: "tool_call_end" }, + ]); + expect(state.toolBlockStarts).toBe(2); + expect(state.completedToolCalls).toBe(2); + expect(state.openToolBlocks?.size ?? 0).toBe(0); }); test("usageFromResult returns undefined when no usage is present", () => { diff --git a/tests/providers/codebuddy-tool-bridge-turn.test.ts b/tests/providers/codebuddy-tool-bridge-turn.test.ts index 99e2a241c45..98598f8f790 100644 --- a/tests/providers/codebuddy-tool-bridge-turn.test.ts +++ b/tests/providers/codebuddy-tool-bridge-turn.test.ts @@ -333,6 +333,138 @@ describe("CodeBuddy capture-only tool bridge turn", () => { expect(events.some(e => e.type === "done")).toBe(false); }); + test("interleaved parallel tool calls are serialized per block and complete the leg", async () => { + const p = parsed([tool("exec")]); + const bridge = buildCodeBuddyToolBridge(p); + const cliName = [...bridge.emittedNameMap.keys()][0]!; + const wireName = bridge.emittedNameMap.get(cliName)!; + const startAt = (index: number, id: string) => ({ + type: "stream_event", + event: { type: "content_block_start", index, content_block: { type: "tool_use", id, name: cliName } }, + }); + const deltaAt = (index: number, part: string) => ({ + type: "stream_event", + event: { type: "content_block_delta", index, delta: { type: "input_json_delta", partial_json: part } }, + }); + const stopAt = (index: number) => ({ type: "stream_event", event: { type: "content_block_stop", index } }); + let child: FakeChild | undefined; + const spawn: SpawnFn = (_cmd, _args) => { + child = fakeChild(frameLines([ + INIT_OK, + // CodeBuddy emits genuinely interleaved blocks for parallel calls: starts arrive + // before earlier blocks stop and argument deltas alternate across indices (captured + // from the live CLI on 2026-09-25). Single-slot accounting led to a spurious + // "incomplete tool call" 502 at message_stop. + startAt(1, "tu_a"), + startAt(2, "tu_b"), + deltaAt(1, "{\"command\":[\"ec"), + deltaAt(2, "{\"command\":[\"ls"), + deltaAt(1, "ho\"]}"), + deltaAt(2, "\"]}"), + stopAt(2), + stopAt(1), + MESSAGE_STOP, + ])); + return child as unknown as ChildProcess; + }; + const adapter = createCodeBuddyAdapter(provider(), { spawn, which: () => "/usr/bin/codebuddy" }); + const events = await run(adapter, p); + expect(events.map(e => e.type)).toEqual([ + "tool_call_start", + "tool_call_delta", + "tool_call_delta", + "tool_call_end", + "tool_call_start", + "tool_call_delta", + "tool_call_delta", + "tool_call_end", + "done", + ]); + // Blocks are emitted atomically in stop order: tu_b closed first. + expect(events[0]).toMatchObject({ type: "tool_call_start", id: "tu_b", name: wireName }); + expect(events[1]).toMatchObject({ arguments: "{\"command\":[\"ls" }); + expect(events[4]).toMatchObject({ type: "tool_call_start", id: "tu_a", name: wireName }); + expect(events[5]).toMatchObject({ arguments: "{\"command\":[\"ec" }); + expect(events[6]).toMatchObject({ arguments: "ho\"]}" }); + expect(events[8]).toMatchObject({ type: "done", stopReason: "tool_use", endTurn: false }); + expect(child?.killed).toBe(true); + }); + + test("an indexless argument delta with two open tool blocks fails before emitting a tool call", async () => { + const p = parsed([tool("exec")]); + const cliName = [...buildCodeBuddyToolBridge(p).emittedNameMap.keys()][0]!; + const frame = (event: Record) => ({ type: "stream_event", event }); + let child: FakeChild | undefined; + const spawn: SpawnFn = () => { + child = fakeChild(frameLines([ + INIT_OK, + frame({ type: "content_block_start", index: 1, content_block: { type: "tool_use", id: "tu_a", name: cliName } }), + frame({ type: "content_block_start", index: 2, content_block: { type: "tool_use", id: "tu_b", name: cliName } }), + frame({ type: "content_block_delta", index: 1, delta: { type: "input_json_delta", partial_json: '{"a":' } }), + inputJsonDelta("2"), + frame({ type: "content_block_delta", index: 1, delta: { type: "input_json_delta", partial_json: "1}" } }), + frame({ type: "content_block_delta", index: 2, delta: { type: "input_json_delta", partial_json: '{"a":3}' } }), + frame({ type: "content_block_stop", index: 1 }), + frame({ type: "content_block_stop", index: 2 }), + MESSAGE_STOP, + ])); + return child as unknown as ChildProcess; + }; + const adapter = createCodeBuddyAdapter(provider(), { spawn, which: () => "/usr/bin/codebuddy" }); + const events = await run(adapter, p); + // Dropping the ambiguous "2" would still leave valid JSON ({"a":1}) and a successful turn. + expect(events).toEqual([expect.objectContaining({ type: "error", code: "protocol_error", status: 502, retryable: false })]); + expect(child?.killed).toBe(true); + }); + + test("a parallel batch on one shared block index completes every call in the leg", async () => { + const p = parsed([tool("exec")]); + const bridge = buildCodeBuddyToolBridge(p); + const cliName = [...bridge.emittedNameMap.keys()][0]!; + const wireName = bridge.emittedNameMap.get(cliName)!; + const start = (id: string) => ({ + type: "stream_event", + event: { type: "content_block_start", index: 2, content_block: { type: "tool_use", id, name: cliName } }, + }); + const delta = (part: string) => ({ + type: "stream_event", + event: { type: "content_block_delta", index: 2, delta: { type: "input_json_delta", partial_json: part } }, + }); + let child: FakeChild | undefined; + const spawn: SpawnFn = (_cmd, _args) => { + child = fakeChild(frameLines([ + INIT_OK, + // Live capture 2026-09-26 (CodeBuddy 2.158.0, kimi-k3-1): a parallel batch reuses one + // content-block index — alpha starts, streams its complete arguments, then beta starts + // on the same index with no stop for alpha; only the final block receives a stop. + start("tu_a"), + delta("{\"command\":[\"echo\"]}"), + start("tu_b"), + delta("{\"command\":[\"ls\"]}"), + { type: "stream_event", event: { type: "content_block_stop", index: 2 } }, + MESSAGE_STOP, + ])); + return child as unknown as ChildProcess; + }; + const adapter = createCodeBuddyAdapter(provider(), { spawn, which: () => "/usr/bin/codebuddy" }); + const events = await run(adapter, p); + expect(events.map(e => e.type)).toEqual([ + "tool_call_start", + "tool_call_delta", + "tool_call_end", + "tool_call_start", + "tool_call_delta", + "tool_call_end", + "done", + ]); + expect(events[0]).toMatchObject({ type: "tool_call_start", id: "tu_a", name: wireName }); + expect(events[1]).toMatchObject({ arguments: "{\"command\":[\"echo\"]}" }); + expect(events[3]).toMatchObject({ type: "tool_call_start", id: "tu_b", name: wireName }); + expect(events[4]).toMatchObject({ arguments: "{\"command\":[\"ls\"]}" }); + expect(events[6]).toMatchObject({ type: "done", stopReason: "tool_use", endTurn: false }); + expect(child?.killed).toBe(true); + }); + test("a tool call before the init frame fails closed with tool_bridge_init_missing", async () => { const p = parsed([tool("exec")]); const bridge = buildCodeBuddyToolBridge(p); @@ -359,6 +491,28 @@ describe("CodeBuddy capture-only tool bridge turn", () => { expect(events.some(e => e.type === "done")).toBe(false); }); + test("a tool start before init stays invalid when the block stops after init", async () => { + const p = parsed([tool("exec")]); + const cliName = [...buildCodeBuddyToolBridge(p).emittedNameMap.keys()][0]!; + const adapter = createCodeBuddyAdapter(provider(), { + spawn: () => fakeChild(frameLines([ + toolUseStart(cliName), + INIT_OK, + BLOCK_STOP, + MESSAGE_STOP, + ])) as unknown as ChildProcess, + which: () => "/usr/bin/codebuddy", + }); + const events = await run(adapter, p); + expect(events.at(-1)).toMatchObject({ + type: "error", + code: "tool_bridge_init_missing", + status: 502, + retryable: false, + }); + expect(events.some(e => e.type === "tool_call_start" || e.type === "done")).toBe(false); + }); + test("a result frame before message_stop defers to the synthesized tool_use done", async () => { const p = parsed([tool("exec")]); const bridge = buildCodeBuddyToolBridge(p);