From 8c01754cf17c40ce8bfa93a2cec42b4385e82f81 Mon Sep 17 00:00:00 2001 From: mdwsk88 <924038395@qq.com> Date: Sat, 26 Sep 2026 21:42:07 +0900 Subject: [PATCH 1/5] fix(codebuddy): serialize parallel tool-use blocks without corrupting calls (#5929) Carried from #5929 as one squashed commit. Co-authored-by: mdwsk88 <924038395@qq.com> --- src/adapters/coding-agent/protocol.ts | 97 ++++++++++++++-- src/adapters/coding-agent/turn.ts | 14 +-- structure/providers-and-adapters.md | 5 + tests/providers/codebuddy-protocol.test.ts | 73 +++++++++++- .../codebuddy-tool-bridge-turn.test.ts | 105 ++++++++++++++++++ 5 files changed, 270 insertions(+), 24 deletions(-) diff --git a/src/adapters/coding-agent/protocol.ts b/src/adapters/coding-agent/protocol.ts index cccbb05db58..09a8e38ea5c 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; with several open blocks attribution is unknowable, + * so the frame is dropped and the turn fails closed at the terminal accounting check. + */ +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,16 @@ 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. 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 +434,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..f66dfb43836 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, }; @@ -527,8 +527,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 +552,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 +573,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", diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index 9871c973ec4..1d6d5911cd0 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -1,5 +1,10 @@ # 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. + 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/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..6791e4aeeef 100644 --- a/tests/providers/codebuddy-tool-bridge-turn.test.ts +++ b/tests/providers/codebuddy-tool-bridge-turn.test.ts @@ -333,6 +333,111 @@ 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("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); From 38d1668b43acbd4deafc540e8d4a9890d4240cf4 Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 26 Sep 2026 21:45:52 +0900 Subject: [PATCH 2/5] fix(codebuddy): validate bridge init at tool-use start arrival --- src/adapters/coding-agent/turn.ts | 45 ++++++++++--------- structure/providers-and-adapters.md | 2 + .../codebuddy-tool-bridge-turn.test.ts | 22 +++++++++ 3 files changed, 49 insertions(+), 20 deletions(-) diff --git a/src/adapters/coding-agent/turn.ts b/src/adapters/coding-agent/turn.ts index f66dfb43836..af9c6de946e 100644 --- a/src/adapters/coding-agent/turn.ts +++ b/src/adapters/coding-agent/turn.ts @@ -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({ @@ -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 1d6d5911cd0..92a83626765 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -4,6 +4,8 @@ The coding-agent stream parser buffers each tool-use block by its content-block 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. +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. diff --git a/tests/providers/codebuddy-tool-bridge-turn.test.ts b/tests/providers/codebuddy-tool-bridge-turn.test.ts index 6791e4aeeef..5c374af2c79 100644 --- a/tests/providers/codebuddy-tool-bridge-turn.test.ts +++ b/tests/providers/codebuddy-tool-bridge-turn.test.ts @@ -464,6 +464,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); From c6b9a73b652fc3ec8426f2519322f82beb7ef90b Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 26 Sep 2026 22:26:17 +0900 Subject: [PATCH 3/5] fix(codebuddy): reject ambiguous indexless tool argument deltas --- src/adapters/coding-agent/protocol.ts | 10 +++++-- structure/providers-and-adapters.md | 2 ++ .../codebuddy-tool-bridge-turn.test.ts | 27 +++++++++++++++++++ 3 files changed, 37 insertions(+), 2 deletions(-) diff --git a/src/adapters/coding-agent/protocol.ts b/src/adapters/coding-agent/protocol.ts index 09a8e38ea5c..71085e30f47 100644 --- a/src/adapters/coding-agent/protocol.ts +++ b/src/adapters/coding-agent/protocol.ts @@ -366,8 +366,8 @@ function toolBlockKey(state: StreamParseState, event: StreamMessage): number { * 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; with several open blocks attribution is unknowable, - * so the frame is dropped and the turn fails closed at the terminal accounting check. + * 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; @@ -418,6 +418,12 @@ function mapRawStreamEvent(event: StreamMessage, state: StreamParseState): Adapt // 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) { const key = resolveToolBlockKey(state, event); diff --git a/structure/providers-and-adapters.md b/structure/providers-and-adapters.md index 92a83626765..8f6309a4a2c 100644 --- a/structure/providers-and-adapters.md +++ b/structure/providers-and-adapters.md @@ -4,6 +4,8 @@ The coding-agent stream parser buffers each tool-use block by its content-block 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. diff --git a/tests/providers/codebuddy-tool-bridge-turn.test.ts b/tests/providers/codebuddy-tool-bridge-turn.test.ts index 5c374af2c79..98598f8f790 100644 --- a/tests/providers/codebuddy-tool-bridge-turn.test.ts +++ b/tests/providers/codebuddy-tool-bridge-turn.test.ts @@ -390,6 +390,33 @@ describe("CodeBuddy capture-only tool bridge turn", () => { 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); From 314f6eddb183a1950305180197459e5e7f22d68a Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 26 Sep 2026 22:27:03 +0900 Subject: [PATCH 4/5] docs(devlog): batch 8 plan --- devlog/_plan/260926_bug_train_6/040_batch8.md | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) create mode 100644 devlog/_plan/260926_bug_train_6/040_batch8.md 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. From 7308442e75928cb25b501bd3428b5e53bef7abfc Mon Sep 17 00:00:00 2001 From: JUN Date: Sat, 26 Sep 2026 22:26:44 +0900 Subject: [PATCH 5/5] fix(test): bound privacy gate aggregate child The macOS control batch stopped at the first aggregate-gate shell invocation and hit the 300s batch watchdog even though spawnSync specified 5s. Replace the synchronous call with an async spawn, ignored stdin, and an independent SIGKILL deadline that reports the child by name. Cover event-loop progress and the deadline path. --- tests/ci-workflows/ci-privacy-gate.test.ts | 93 +++++++++++++++------- 1 file changed, 63 insertions(+), 30 deletions(-) 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"); });