diff --git a/docs/architecture/native-tool-call-parser-scoping-model.md b/docs/architecture/native-tool-call-parser-scoping-model.md new file mode 100644 index 0000000000..ee665a88a2 --- /dev/null +++ b/docs/architecture/native-tool-call-parser-scoping-model.md @@ -0,0 +1,46 @@ +# Native tool-call parser request-scope model check + +Zoo Code checks native tool-call parser request isolation with a bounded, exhaustive replay model. It is a child submodel in the umbrella task lifecycle verification suite, which runs in CI and locally with: + +```sh +pnpm lifecycle:model-check +``` + +For focused debugging, run this submodel directly with: + +```sh +pnpm parser-scope:model-check +``` + +The command is composed into the same verification suite, but this remains a separate protocol and state space from the persisted task lifecycle model and shared-store concurrency model. It owns its parser-scoping invariants and adds no parser state to `HistoryItem` or `taskLifecycle.ts`; instead, it replays the public production `NativeToolCallParser` APIs using two independent scope objects. + +## Bounds and replay + +The source of truth is `scripts/check-native-tool-call-parser-scoping.ts`. The model has two request scopes, A and B. Both receive provider raw tool index zero, but each has a distinct tool-call ID and two distinct JSON argument fragments. Each scope follows this local order: + +1. open the request scope; +2. start raw call index zero and its streaming accumulator; +3. add two distinct argument fragments through both production accumulation APIs; +4. finalize the raw call and reject duplicate raw finalization; +5. finalize the streaming call, reject duplicate streaming finalization, and clear both kinds of state; and +6. deliver late raw and streaming fragments. + +The checker exhausts all 924 order-preserving interleavings of those two six-action sequences. Opening, raw start, fragment delivery, raw finalization, streaming finalization/cleanup, and late fragment delivery are independently schedulable protocol phases. Fragment delivery remains one bounded action per scope and replays both argument fragments through both production accumulation APIs; streaming cleanup remains attached to streaming finalization because late delivery is the only valid following local phase. This preserves each request's local order while keeping CI runtime bounded. The expected schedule count, maximum schedule budget, scope count, raw index, and actions per scope are explicit. It fails if schedule enumeration differs from the binomial bound or exceeds the budget, so truncated exploration cannot pass. + +Each schedule uses fresh production scope objects and calls `processRawChunk`, `startStreamingToolCall`, `processStreamingChunk`, `finalizeRawChunks`, `finalizeStreamingToolCall`, `clearRawChunkState`, `clearAllStreamingToolCalls`, and `hasActiveStreamingToolCalls`. It neither inspects private parser maps nor duplicates their transition logic. + +## Invariants and landmarks + +Every replay checks: + +1. emitted start, delta, and end events retain the owning scope's call ID; +2. finalized arguments contain only the owning scope's fragments; +3. cleanup in one scope cannot change the other scope's active streaming state; +4. each scope emits exactly one raw end and one streaming final result; +5. repeated finalization is empty/null rather than duplicate; +6. late raw and streaming fragments are ignored after cleanup; +7. every modeled action is reachable. + +Named landmarks require simultaneous active scopes, B opening while A has received its fragments, either scope raw-finalizing while the other remains active, either scope streaming-finalizing and cleaning up while the other remains active, and symmetric late-fragment schedules in which the other scope is still active. + +These are finite safety claims only. The model does not claim provider transport ordering, retry liveness, fairness, persistence, or arbitrary call counts. Provider suites separately test their public stream contracts with two overlapping streams, while focused parser and Task tests cover production integration. diff --git a/docs/architecture/task-lifecycle-model.md b/docs/architecture/task-lifecycle-model.md index 588ffd5204..25995cad6a 100644 --- a/docs/architecture/task-lifecycle-model.md +++ b/docs/architecture/task-lifecycle-model.md @@ -1,12 +1,22 @@ -# Task lifecycle model check +# Task lifecycle model-check suite -Zoo Code checks its persisted task delegation lifecycle with a bounded, exhaustive state explorer. Run it locally with: +Zoo Code checks task lifecycle protocols through one compositional verification suite. Run the complete suite locally with: ```sh pnpm lifecycle:model-check ``` -The check runs in the `compile` CI job after type checking. It fails if it finds an invariant violation, a modeled action becomes unreachable, or exploration exceeds its declared state budget. A violation includes the shortest breadth-first event trace, every intermediate state, and the active bounds so the sequence can be replayed as a focused regression test. +The command runs three independent bounded submodels in sequence: + +1. the persisted task delegation lifecycle; +2. shared-store concurrency across task-history hosts; and +3. request-stream parser scoping. + +This umbrella command is the single model-check entry point in the `compile` CI job after type checking. Command-level composition does not merge the submodels' state spaces: each checker retains its own bounds, transitions, invariant ownership, reachability requirements, and counterexample format. In particular, parser state is not part of the persisted lifecycle graph. The focused parser checker remains directly runnable with `pnpm parser-scope:model-check` for debugging. + +An individual checker fails if it finds an invariant violation, a modeled action becomes unreachable, or exploration exceeds its declared state budget. A lifecycle violation includes the shortest breadth-first event trace, every intermediate state, and the active bounds so the sequence can be replayed as a focused regression test. + +Executable cross-model composition should be added only when a correctness claim genuinely spans two or more submodels and there is an explicit, production-grounded boundary mapping between their events or state. That composition must state a bounded joint exploration strategy and own cross-model invariants that cannot be proved within either child model alone. Shared command orchestration or conceptual adjacency is not sufficient reason to multiply independent state spaces. ## Why an executable TypeScript model @@ -81,16 +91,16 @@ These are safety claims within the documented bounds. The check does not claim l The following map separates issue observations from the architectural interpretation encoded here. Open issues can change after this document is written; follow each link for current status. -| Issue and directly observed evidence | Derived protocol rule | Production transition and current check | -| ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| [#1469](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1469): the issue report states that a barrier-controlled two-host run reproduced an old child completion clearing a newer handoff 25/25 times. | Completion is conditional on the parent still awaiting that exact child; a live-linked child must remain owned by its parent. | `completeDelegatedChild` rejects stale authoritative input. The lifecycle explorer checks that reducer rule, while the shared-store explorer reproduces the cross-host stale-cache counterexample with an exact causal witness. | -| [#1021](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1021): an in-flight `saveClineMessages` can restore parent/root IDs after abandonment cleared them. | Detachment should be monotonic: later lifecycle work must not reattach an abandoned child. | `abandonDelegatedChild` clears both sides. The shared-store explorer proves the detach commit occurs, then reproduces a refreshed-cache delta that preserves interrupted status while restoring stale live-task lineage. | -| [#1453](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1453), under user report [#1279](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1279): CI observed `TaskCompleted` before restart-visible API history once; 120 local repetitions did not reproduce it, while an Alloy abstraction permits the ordering. | A completion/readiness contract must define whether completion implies restart visibility. This is a liveness/durability boundary, not only a `HistoryItem` safety transition. | Not claimed by this checker. Add a controlled persistence barrier test after the contract decision; move to temporal model checking if eventual readiness and failure handling become protocol guarantees. | -| [#921](https://github.com/Zoo-Code-Org/Zoo-Code/issues/921): delegation across parallel tabs lacks coverage for different view-local mode/profile state. | Delegation must bind an explicit immutable execution-context snapshot rather than read whichever view is focused later. | The persisted ownership transition is covered; mode/profile snapshot isolation is outside this state model and belongs in a production adapter/model-based test. | -| [#920](https://github.com/Zoo-Code-Org/Zoo-Code/issues/920): issue analysis identifies a missing cross-instance history-update test and potential lost writes. | Distinct task writes must not overwrite one another, and same-task conflicts need an explicit merge/ownership rule. | The shared-store explorer checks distinct-task writes and same-record independent deltas. Cross-instance store tests retain production API coverage, and the synchronized real-filesystem smoke test exercises the actual lock/write path without claiming exhaustive filesystem proof. | -| [#369](https://github.com/Zoo-Code-Org/Zoo-Code/issues/369) and [#372](https://github.com/Zoo-Code-Org/Zoo-Code/issues/372): planned fan-out keeps a parent live while a child runs and requires completion routing by explicit parent ID, single-writer result readiness, permit release, and orphan cleanup. | Persisted `delegated` status is ownership, not proof that the parent instance is suspended. Completion must route by IDs; scheduler resources and live-instance state need separate invariants. | Nested and sibling lifecycle ownership are covered. Scheduler permits, live/suspended parent selection, orphan cancellation, and single-writer message readiness must be added when fan-out lands; they should not be folded into `HistoryItem` fields prematurely. | -| [#1468](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1468): a late chunk from one request combined tool identity with arguments from another request; rerun passed. | Every stream accumulator needs a request/task generation key, and late events cannot mutate another scope. | Separate protocol. It warrants a parser-scope model or deterministic interleaving test, not an unrelated field in the delegation model. | -| [#612](https://github.com/Zoo-Code-Org/Zoo-Code/issues/612): the CLI copied a status union and omitted `interrupted`. | Lifecycle vocabulary should have one type owner. | `HistoryItemStatus` is derived from `HistoryItem`, and production/checker transitions share `taskLifecycle.ts`; consumers should import rather than copy the union. | +| Issue and directly observed evidence | Derived protocol rule | Production transition and current check | +| ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| [#1469](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1469): the issue report states that a barrier-controlled two-host run reproduced an old child completion clearing a newer handoff 25/25 times. | Completion is conditional on the parent still awaiting that exact child; a live-linked child must remain owned by its parent. | `completeDelegatedChild` rejects stale authoritative input. The lifecycle explorer checks that reducer rule, while the shared-store explorer reproduces the cross-host stale-cache counterexample with an exact causal witness. | +| [#1021](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1021): an in-flight `saveClineMessages` can restore parent/root IDs after abandonment cleared them. | Detachment should be monotonic: later lifecycle work must not reattach an abandoned child. | `abandonDelegatedChild` clears both sides. The shared-store explorer proves the detach commit occurs, then reproduces a refreshed-cache delta that preserves interrupted status while restoring stale live-task lineage. | +| [#1453](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1453), under user report [#1279](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1279): CI observed `TaskCompleted` before restart-visible API history once; 120 local repetitions did not reproduce it, while an Alloy abstraction permits the ordering. | A completion/readiness contract must define whether completion implies restart visibility. This is a liveness/durability boundary, not only a `HistoryItem` safety transition. | Not claimed by this checker. Add a controlled persistence barrier test after the contract decision; move to temporal model checking if eventual readiness and failure handling become protocol guarantees. | +| [#921](https://github.com/Zoo-Code-Org/Zoo-Code/issues/921): delegation across parallel tabs lacks coverage for different view-local mode/profile state. | Delegation must bind an explicit immutable execution-context snapshot rather than read whichever view is focused later. | The persisted ownership transition is covered; mode/profile snapshot isolation is outside this state model and belongs in a production adapter/model-based test. | +| [#920](https://github.com/Zoo-Code-Org/Zoo-Code/issues/920): issue analysis identifies a missing cross-instance history-update test and potential lost writes. | Distinct task writes must not overwrite one another, and same-task conflicts need an explicit merge/ownership rule. | The shared-store explorer checks distinct-task writes and same-record independent deltas. Cross-instance store tests retain production API coverage, and the synchronized real-filesystem smoke test exercises the actual lock/write path without claiming exhaustive filesystem proof. | +| [#369](https://github.com/Zoo-Code-Org/Zoo-Code/issues/369) and [#372](https://github.com/Zoo-Code-Org/Zoo-Code/issues/372): planned fan-out keeps a parent live while a child runs and requires completion routing by explicit parent ID, single-writer result readiness, permit release, and orphan cleanup. | Persisted `delegated` status is ownership, not proof that the parent instance is suspended. Completion must route by IDs; scheduler resources and live-instance state need separate invariants. | Nested and sibling lifecycle ownership are covered. Scheduler permits, live/suspended parent selection, orphan cancellation, and single-writer message readiness must be added when fan-out lands; they should not be folded into `HistoryItem` fields prematurely. | +| [#1468](https://github.com/Zoo-Code-Org/Zoo-Code/issues/1468): a late chunk from one request combined tool identity with arguments from another request; rerun passed. | Every stream accumulator needs a request/task generation key, and late events cannot mutate another scope. | Separate protocol. The [native tool-call parser request-scope model](./native-tool-call-parser-scoping-model.md), whose source of truth is `scripts/check-native-tool-call-parser-scoping.ts`, exhaustively replays bounded production-parser interleavings without adding fields to this lifecycle model. | +| [#612](https://github.com/Zoo-Code-Org/Zoo-Code/issues/612): the CLI copied a status union and omitted `interrupted`. | Lifecycle vocabulary should have one type owner. | `HistoryItemStatus` is derived from `HistoryItem`, and production/checker transitions share `taskLifecycle.ts`; consumers should import rather than copy the union. | The issue-derived cases intentionally map to bug classes rather than issue-specific flags. In particular, stale event ownership, monotonic terminal/detached state, explicit scope, and single-writer boundaries generalize to future concurrent task work. @@ -106,6 +116,8 @@ When production lifecycle behavior changes: Do not weaken bounds or remove an invariant merely to make CI pass. If state growth becomes difficult to control, split independent protocols or move the model to TLC/Quint with an implementation trace adapter rather than silently sampling the state space. +Parser request scoping is one such independent bounded submodel within the umbrella suite. Extend `scripts/check-native-tool-call-parser-scoping.ts` and its focused architecture document instead of adding parser state or transitions to `taskLifecycle.ts` or the persisted lifecycle state graph. + ## Test layering Keep reducer permutations in this model and focused Vitest suites. The real VS Code extension-host suite using a mocked provider in `apps/vscode-e2e/src/suite/subtasks.test.ts` already covers the boundaries the pure explorer cannot: task creation and rehydration, persisted parent-child state, cancellation during a delayed provider stream, interrupted-child resume, abandonment followed by a real resume/save/completion cycle, pending approvals across leave/return, and scheduler-driven resume. `restart-persistence.test.ts` separately verifies completion history through a fresh extension host. diff --git a/package.json b/package.json index 8431467918..9697f6f673 100644 --- a/package.json +++ b/package.json @@ -13,7 +13,8 @@ "check-types": "turbo check-types --log-order grouped --output-logs new-only", "test": "turbo test --log-order grouped --output-logs new-only", "test:mutation-ci": "node --test scripts/stryker-diff.test.mjs", - "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts", + "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts && pnpm parser-scope:model-check", + "parser-scope:model-check": "node scripts/run-native-tool-call-parser-scoping.mjs", "test:coverage": "turbo test:coverage --log-order grouped --output-logs new-only", "format": "turbo format --log-order grouped --output-logs new-only", "build": "turbo build --log-order grouped --output-logs new-only", diff --git a/scripts/check-native-tool-call-parser-scoping.ts b/scripts/check-native-tool-call-parser-scoping.ts new file mode 100644 index 0000000000..556b50f9b4 --- /dev/null +++ b/scripts/check-native-tool-call-parser-scoping.ts @@ -0,0 +1,315 @@ +import assert from "node:assert/strict" + +import { NativeToolCallParser, type ToolCallStreamEvent } from "../src/core/assistant-message/NativeToolCallParser" + +const scopeIds = ["A", "B"] as const +type ScopeId = (typeof scopeIds)[number] + +const localActions = [ + "open", + "start-raw-call", + "add-fragments", + "finalize-raw-call", + "finalize-streaming-call-and-cleanup", + "late-fragments", +] as const +type LocalAction = (typeof localActions)[number] + +interface ScheduledAction { + scopeId: ScopeId + action: LocalAction +} + +interface ScopeReplayState { + scope?: object + rawEndCount: number + streamFinalizationCount: number + lateFragmentsIgnored: boolean +} + +interface ReplayState { + scopes: Record + events: Array<{ owner: ScopeId; event: ToolCallStreamEvent }> +} + +const RAW_TOOL_INDEX = 0 +const MAX_ACTIONS_PER_SCOPE = localActions.length +const MAX_TOTAL_ACTIONS = MAX_ACTIONS_PER_SCOPE * scopeIds.length +const EXPECTED_SCHEDULES = binomial(MAX_TOTAL_ACTIONS, MAX_ACTIONS_PER_SCOPE) +const MAX_SCHEDULES = EXPECTED_SCHEDULES + +const callIds = { A: "call_scope_a", B: "call_scope_b" } satisfies Record +const paths = { A: "scope-a.ts", B: "scope-b.ts" } satisfies Record +const fragments = { + A: ['{"path":"scope-', 'a.ts"}'], + B: ['{"path":"scope-', 'b.ts"}'], +} satisfies Record + +const expectedActions = new Set(localActions) +const reachedActions = new Set() +const reachedLandmarks = new Set() + +const landmarkNames = [ + "simultaneous-active-scopes", + "B-opens-while-A-is-partial", + "A-raw-finalizes-while-B-is-active", + "B-raw-finalizes-while-A-is-active", + "A-stream-finalizes-while-B-is-active", + "B-stream-finalizes-while-A-is-active", + "A-late-fragment-while-B-is-active", + "B-late-fragment-while-A-is-active", +] as const + +function binomial(n: number, k: number): number { + let result = 1 + for (let index = 1; index <= k; index++) { + result = (result * (n - k + index)) / index + } + return result +} + +function initialReplayState(): ReplayState { + return { + scopes: { + A: { rawEndCount: 0, streamFinalizationCount: 0, lateFragmentsIgnored: false }, + B: { rawEndCount: 0, streamFinalizationCount: 0, lateFragmentsIgnored: false }, + }, + events: [], + } +} + +function activeAtProgress(progress: number): boolean { + return ( + progress >= localActions.indexOf("start-raw-call") + 1 && + progress < localActions.indexOf("finalize-streaming-call-and-cleanup") + 1 + ) +} + +function appendOwnedEvents(state: ReplayState, owner: ScopeId, events: ToolCallStreamEvent[]): void { + for (const event of events) { + state.events.push({ owner, event }) + assert.equal(event.id, callIds[owner], `${owner} emitted an event owned by the other request scope`) + } +} + +function requireScope(state: ReplayState, scopeId: ScopeId): object { + const scope = state.scopes[scopeId].scope + assert.ok(scope, `${scopeId} must be opened before ${scopeId}'s parser APIs are replayed`) + return scope +} + +function replayAction(state: ReplayState, scheduled: ScheduledAction): void { + const { scopeId, action } = scheduled + const scopeState = state.scopes[scopeId] + reachedActions.add(action) + + switch (action) { + case "open": + scopeState.scope = NativeToolCallParser.createScope() + break + case "start-raw-call": { + const scope = requireScope(state, scopeId) + const events = NativeToolCallParser.processRawChunk( + { index: RAW_TOOL_INDEX, id: callIds[scopeId], name: "read_file" }, + scope, + ) + assert.deepEqual(events, [{ type: "tool_call_start", id: callIds[scopeId], name: "read_file" }]) + appendOwnedEvents(state, scopeId, events) + NativeToolCallParser.startStreamingToolCall(callIds[scopeId], "read_file", scope) + break + } + case "add-fragments": { + const scope = requireScope(state, scopeId) + for (const fragment of fragments[scopeId]) { + const events = NativeToolCallParser.processRawChunk( + { index: RAW_TOOL_INDEX, arguments: fragment }, + scope, + ) + assert.deepEqual(events, [{ type: "tool_call_delta", id: callIds[scopeId], delta: fragment }]) + appendOwnedEvents(state, scopeId, events) + assert.notEqual( + NativeToolCallParser.processStreamingChunk(callIds[scopeId], fragment, scope), + null, + `${scopeId}'s fragment was not accepted by its streaming accumulator`, + ) + } + break + } + case "finalize-raw-call": { + const scope = requireScope(state, scopeId) + const events = NativeToolCallParser.finalizeRawChunks(scope) + assert.deepEqual(events, [{ type: "tool_call_end", id: callIds[scopeId] }]) + appendOwnedEvents(state, scopeId, events) + scopeState.rawEndCount += events.length + assert.deepEqual( + NativeToolCallParser.finalizeRawChunks(scope), + [], + `${scopeId} emitted a duplicate raw end`, + ) + break + } + case "finalize-streaming-call-and-cleanup": { + const scope = requireScope(state, scopeId) + const result = NativeToolCallParser.finalizeStreamingToolCall(callIds[scopeId], scope) + assert.equal(result?.type, "tool_use") + if (result?.type !== "tool_use" || result.name !== "read_file") { + throw new Error(`${scopeId}'s streaming result was not a read_file tool use`) + } + if (!result.nativeArgs || !("path" in result.nativeArgs)) { + throw new Error(`${scopeId}'s streaming result did not use current read_file arguments`) + } + assert.equal(result.nativeArgs?.path, paths[scopeId], `${scopeId}'s arguments crossed request scopes`) + scopeState.streamFinalizationCount += 1 + assert.equal( + NativeToolCallParser.finalizeStreamingToolCall(callIds[scopeId], scope), + null, + `${scopeId} finalized its streaming call twice`, + ) + NativeToolCallParser.clearRawChunkState(scope) + NativeToolCallParser.clearAllStreamingToolCalls(scope) + break + } + case "late-fragments": { + const scope = requireScope(state, scopeId) + const rawEvents = NativeToolCallParser.processRawChunk( + { index: RAW_TOOL_INDEX, arguments: `late-${scopeId}` }, + scope, + ) + const streamingResult = NativeToolCallParser.processStreamingChunk( + callIds[scopeId], + `late-${scopeId}`, + scope, + ) + assert.deepEqual(rawEvents, [], `${scopeId} accepted a late raw fragment`) + assert.equal(streamingResult, null, `${scopeId} accepted a late streaming fragment`) + scopeState.lateFragmentsIgnored = true + break + } + } +} + +function checkInvariants(state: ReplayState, progress: Record, trace: ScheduledAction[]): void { + for (const scopeId of scopeIds) { + const scopeState = state.scopes[scopeId] + const scope = scopeState.scope + const expectedActive = activeAtProgress(progress[scopeId]) + assert.equal( + scope ? NativeToolCallParser.hasActiveStreamingToolCalls(scope) : false, + expectedActive, + `${scopeId}'s active streaming state was changed by the other request scope`, + ) + assert.ok(scopeState.rawEndCount <= 1, `${scopeId} emitted duplicate raw finalization events`) + assert.ok(scopeState.streamFinalizationCount <= 1, `${scopeId} finalized its streaming call more than once`) + } + + for (const { owner, event } of state.events) { + assert.equal(event.id, callIds[owner], `${owner}'s event log contains another scope's call ID`) + } + + const last = trace.at(-1) + if (!last) return + if (activeAtProgress(progress.A) && activeAtProgress(progress.B)) reachedLandmarks.add("simultaneous-active-scopes") + if (last.scopeId === "B" && last.action === "open" && progress.A === 3) { + reachedLandmarks.add("B-opens-while-A-is-partial") + } + if (last.action === "finalize-raw-call" && activeAtProgress(progress[last.scopeId === "A" ? "B" : "A"])) { + reachedLandmarks.add(`${last.scopeId}-raw-finalizes-while-${last.scopeId === "A" ? "B" : "A"}-is-active`) + } + if ( + last.action === "finalize-streaming-call-and-cleanup" && + activeAtProgress(progress[last.scopeId === "A" ? "B" : "A"]) + ) { + reachedLandmarks.add(`${last.scopeId}-stream-finalizes-while-${last.scopeId === "A" ? "B" : "A"}-is-active`) + } + if (last.action === "late-fragments" && activeAtProgress(progress[last.scopeId === "A" ? "B" : "A"])) { + reachedLandmarks.add(`${last.scopeId}-late-fragment-while-${last.scopeId === "A" ? "B" : "A"}-is-active`) + } +} + +function cleanupReplay(state: ReplayState): void { + for (const scopeId of scopeIds) { + const scope = state.scopes[scopeId].scope + if (!scope) continue + NativeToolCallParser.clearRawChunkState(scope) + NativeToolCallParser.clearAllStreamingToolCalls(scope) + } +} + +function replaySchedule(trace: ScheduledAction[]): void { + const state = initialReplayState() + const progress: Record = { A: 0, B: 0 } + try { + for (const scheduled of trace) { + replayAction(state, scheduled) + progress[scheduled.scopeId] += 1 + checkInvariants(state, progress, trace.slice(0, progress.A + progress.B)) + } + for (const scopeId of scopeIds) { + assert.equal(state.scopes[scopeId].rawEndCount, 1, `${scopeId} did not emit exactly one raw end`) + assert.equal(state.scopes[scopeId].streamFinalizationCount, 1, `${scopeId} did not finalize exactly once`) + assert.equal( + state.scopes[scopeId].lateFragmentsIgnored, + true, + `${scopeId}'s late fragments were not checked`, + ) + } + } catch (error) { + const formattedTrace = trace + .map(({ scopeId, action }, index) => `${index + 1}. ${scopeId}.${action}`) + .join("\n") + throw new Error( + `Native tool-call parser scope invariant failed within bounds scopes=${scopeIds.length}, actions-per-scope=${MAX_ACTIONS_PER_SCOPE}, schedules=${MAX_SCHEDULES}\n${formattedTrace}`, + { cause: error }, + ) + } finally { + cleanupReplay(state) + } +} + +function enumerateSchedules(): number { + const trace: ScheduledAction[] = [] + const progress: Record = { A: 0, B: 0 } + let exploredSchedules = 0 + + function visit(): void { + if (trace.length === MAX_TOTAL_ACTIONS) { + exploredSchedules += 1 + if (exploredSchedules > MAX_SCHEDULES) { + throw new Error(`Parser-scope exploration exceeded its ${MAX_SCHEDULES}-schedule budget`) + } + replaySchedule(trace) + return + } + + for (const scopeId of scopeIds) { + const localProgress = progress[scopeId] + if (localProgress === MAX_ACTIONS_PER_SCOPE) continue + const action = localActions[localProgress] + if (!action) throw new Error(`${scopeId} has no modeled action at local progress ${localProgress}`) + trace.push({ scopeId, action }) + progress[scopeId] += 1 + visit() + progress[scopeId] -= 1 + trace.pop() + } + } + + visit() + return exploredSchedules +} + +const exploredSchedules = enumerateSchedules() +assert.equal( + exploredSchedules, + EXPECTED_SCHEDULES, + `Parser-scope exploration truncated: expected ${EXPECTED_SCHEDULES} schedules, explored ${exploredSchedules}`, +) + +const unreachableActions = [...expectedActions].filter((action) => !reachedActions.has(action)) +assert.deepEqual(unreachableActions, [], `Parser-scope model has unreachable actions: ${unreachableActions.join(", ")}`) +const missingLandmarks = landmarkNames.filter((name) => !reachedLandmarks.has(name)) +assert.deepEqual(missingLandmarks, [], `Parser-scope model has unreachable landmarks: ${missingLandmarks.join(", ")}`) + +console.log( + `Native tool-call parser scope model check passed: ${exploredSchedules}/${EXPECTED_SCHEDULES} valid local-order interleavings, ${localActions.length}/${localActions.length} actions reachable, ${landmarkNames.length}/${landmarkNames.length} landmarks reached, scopes=${scopeIds.length}, raw-index=${RAW_TOOL_INDEX}, actions-per-scope=${MAX_ACTIONS_PER_SCOPE}`, +) diff --git a/scripts/run-native-tool-call-parser-scoping.mjs b/scripts/run-native-tool-call-parser-scoping.mjs new file mode 100644 index 0000000000..cedbd9be1a --- /dev/null +++ b/scripts/run-native-tool-call-parser-scoping.mjs @@ -0,0 +1,25 @@ +import { mkdtemp, rm } from "node:fs/promises" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { fileURLToPath, pathToFileURL } from "node:url" + +import { build } from "esbuild" + +const entryPoint = fileURLToPath(new URL("./check-native-tool-call-parser-scoping.ts", import.meta.url)) +const temporaryDirectory = await mkdtemp(join(tmpdir(), "zoo-parser-scope-model-")) +const outfile = join(temporaryDirectory, "model.cjs") + +try { + await build({ + entryPoints: [entryPoint], + bundle: true, + platform: "node", + format: "cjs", + external: ["vscode"], + outfile, + logLevel: "info", + }) + await import(pathToFileURL(outfile).href) +} finally { + await rm(temporaryDirectory, { recursive: true, force: true }) +} diff --git a/scripts/stryker-diff.mjs b/scripts/stryker-diff.mjs index c0e8a6cd1a..297bacf0a3 100644 --- a/scripts/stryker-diff.mjs +++ b/scripts/stryker-diff.mjs @@ -292,18 +292,6 @@ export function parseVitestTestFiles(report, runRoot) { ] } -export function preferDirectTestFiles(testFiles, sourceFiles) { - const sourceNames = sourceFiles.map((sourceFile) => path.posix.basename(sourceFile, path.posix.extname(sourceFile))) - const direct = testFiles.filter((testFile) => { - const testName = path.posix.basename(testFile) - return sourceNames.some( - (sourceName) => - testName.startsWith(`${sourceName}.`) && /\.(?:test|spec)(?:\.[^.]+)?\.[cm]?[jt]sx?$/.test(testName), - ) - }) - return direct.length > 0 ? direct : testFiles -} - export function resolveVitestBinary(repoRoot, packageEntry) { const packageRoot = path.join(repoRoot, packageEntry.root) const runRoot = path.join(repoRoot, packageEntry.runRoot ?? packageEntry.root) @@ -343,10 +331,7 @@ export function discoverRelatedTestFiles(repoRoot, packageEntry, reportDirectory ) } - const testFiles = preferDirectTestFiles( - parseVitestTestFiles(JSON.parse(fs.readFileSync(outputFile, "utf8")), runRoot), - sourceFiles, - ) + const testFiles = parseVitestTestFiles(JSON.parse(fs.readFileSync(outputFile, "utf8")), runRoot) if (testFiles.length === 0) throw new Error(`${packageEntry.id} has no tests related to the changed executable lines`) return testFiles diff --git a/scripts/stryker-diff.test.mjs b/scripts/stryker-diff.test.mjs index 0f39dc507f..04c8b2d2e4 100644 --- a/scripts/stryker-diff.test.mjs +++ b/scripts/stryker-diff.test.mjs @@ -22,7 +22,6 @@ import { parseChangedLines, parseNameStatus, parseVitestTestFiles, - preferDirectTestFiles, resolveVitestBinary, packageForPath, runManifest, @@ -180,36 +179,26 @@ describe("packageForPath", () => { }) describe("parseVitestTestFiles", () => { - it("normalizes and deduplicates Vitest related-test results", () => { + it("normalizes and deduplicates all Vitest related-test results without filename filtering", () => { assert.deepEqual( parseVitestTestFiles( { testResults: [ { name: "/repo/webview-ui/src/utils/__tests__/value.test.ts" }, { name: "/repo/webview-ui/src/utils/__tests__/value.test.ts" }, + { name: "/repo/webview-ui/src/components/__tests__/consumer-named.spec.tsx" }, ], }, "/repo", ), - ["webview-ui/src/utils/__tests__/value.test.ts"], + [ + "webview-ui/src/utils/__tests__/value.test.ts", + "webview-ui/src/components/__tests__/consumer-named.spec.tsx", + ], ) }) }) -describe("preferDirectTestFiles", () => { - it("uses matching focused specs and falls back to all related tests", () => { - const related = [ - "webview-ui/src/__tests__/App.spec.tsx", - "webview-ui/src/utils/__tests__/path-mentions.test.ts", - "webview-ui/src/components/chat/__tests__/ChatView.spec.tsx", - ] - assert.deepEqual(preferDirectTestFiles(related, ["webview-ui/src/utils/path-mentions.ts"]), [ - "webview-ui/src/utils/__tests__/path-mentions.test.ts", - ]) - assert.deepEqual(preferDirectTestFiles(related, ["webview-ui/src/utils/unmatched.ts"]), related) - }) -}) - describe("related-test discovery", () => { it("resolves Vitest from each package before falling back to the repository", () => { const repo = fs.mkdtempSync(path.join(os.tmpdir(), "stryker-vitest-")) diff --git a/src/api/providers/__tests__/lmstudio-native-tools.spec.ts b/src/api/providers/__tests__/lmstudio-native-tools.spec.ts index c6a63902a1..7cc9551a4e 100644 --- a/src/api/providers/__tests__/lmstudio-native-tools.spec.ts +++ b/src/api/providers/__tests__/lmstudio-native-tools.spec.ts @@ -3,6 +3,7 @@ // Mock OpenAI client - must come before other imports const mockCreate = vi.fn() import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" +import { collectStreamAndParseToolCalls } from "../../../test-utils/native-tool-call-stream" import { clearAllMocks } from "../../../test-utils/reset" vi.mock("openai", () => { return { @@ -268,6 +269,123 @@ describe("LmStudioHandler Native Tools", () => { expect(endChunks[0].id).toBe("call_lmstudio_test") }) + it("emits completion only for identified calls and clears completed IDs", async () => { + const toolCall = (id?: string) => ({ + choices: [ + { + delta: { + tool_calls: [ + { index: 0, id, function: { name: "test_tool", arguments: '{"arg1":"value"}' } }, + ], + }, + }, + ], + }) + mockCreate + .mockImplementationOnce(() => + asyncStreamFrom([toolCall(), { choices: [{ delta: {}, finish_reason: "tool_calls" }] }]), + ) + .mockImplementationOnce(() => + asyncStreamFrom([ + toolCall("call_lmstudio_stop"), + { choices: [{ delta: {}, finish_reason: "stop" }] }, + ]), + ) + .mockImplementationOnce(() => + asyncStreamFrom([ + toolCall("call_lmstudio_once"), + { choices: [{ delta: {}, finish_reason: "tool_calls" }] }, + { choices: [{ delta: {}, finish_reason: "tool_calls" }] }, + ]), + ) + + const createMessage = () => handler.createMessage("test prompt", [], { taskId: "task", tools: testTools }) + const idlessChunks = await collectStream(createMessage()) + const stoppedChunks = await collectStream(createMessage()) + const completedChunks = await collectStream(createMessage()) + + expect(idlessChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([]) + expect(stoppedChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([]) + expect(completedChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_lmstudio_once" }, + ]) + }) + + it("isolates overlapping tool-call finalization between provider streams", async () => { + let releaseFirstStream: (() => void) | undefined + let markFirstStreamPaused: (() => void) | undefined + const firstStreamRelease = new Promise((resolve) => { + releaseFirstStream = resolve + }) + const firstStreamPaused = new Promise((resolve) => { + markFirstStreamPaused = resolve + }) + const firstStream = async function* () { + yield { + choices: [ + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_lmstudio_a", + function: { name: "test_tool", arguments: '{"arg1":"a' }, + }, + ], + }, + }, + ], + } + markFirstStreamPaused?.() + await firstStreamRelease + yield { choices: [{ delta: {}, finish_reason: "tool_calls" }] } + } + const secondStream = asyncStreamFrom([ + { + choices: [ + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_lmstudio_b", + function: { name: "test_tool", arguments: '{"arg1":"b' }, + }, + ], + }, + }, + ], + }, + { choices: [{ delta: {}, finish_reason: "tool_calls" }] }, + ]) + mockCreate.mockImplementationOnce(() => firstStream()).mockImplementationOnce(() => secondStream) + + const firstChunksPromise = collectStreamAndParseToolCalls( + handler.createMessage("first", [], { taskId: "task-a", tools: testTools }), + ) + await firstStreamPaused + const secondChunks = await collectStreamAndParseToolCalls( + handler.createMessage("second", [], { taskId: "task-b", tools: testTools }), + ) + releaseFirstStream?.() + const firstChunks = await firstChunksPromise + + expect(secondChunks.chunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_lmstudio_b" }, + ]) + expect(firstChunks.chunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_lmstudio_a" }, + ]) + expect(firstChunks.parserEvents).toEqual([ + { type: "tool_call_start", id: "call_lmstudio_a", name: "test_tool" }, + { type: "tool_call_delta", id: "call_lmstudio_a", delta: '{"arg1":"a' }, + ]) + expect(secondChunks.parserEvents).toEqual([ + { type: "tool_call_start", id: "call_lmstudio_b", name: "test_tool" }, + { type: "tool_call_delta", id: "call_lmstudio_b", delta: '{"arg1":"b' }, + ]) + }) + it("should work with parallel tool calls disabled (sends false)", async () => { mockCreate.mockImplementationOnce(() => asyncStreamFrom([{ choices: [{ delta: { content: "Response" } }] }]), diff --git a/src/api/providers/__tests__/openrouter.spec.ts b/src/api/providers/__tests__/openrouter.spec.ts index 18b6286d09..744bc66e99 100644 --- a/src/api/providers/__tests__/openrouter.spec.ts +++ b/src/api/providers/__tests__/openrouter.spec.ts @@ -23,6 +23,7 @@ import { OpenRouterHandler } from "../openrouter" import { Package } from "../../../shared/package" import { makeApiHandlerOptions } from "../../../test-utils/api" import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" +import { collectStreamAndParseToolCalls } from "../../../test-utils/native-tool-call-stream" import { clearAllMocks } from "../../../test-utils/reset" vitest.mock("openai") @@ -543,6 +544,144 @@ describe("OpenRouterHandler", () => { expect(endChunks).toHaveLength(1) expect(endChunks[0].id).toBe("call_openrouter_test") }) + + it("emits completion only for identified calls and clears completed IDs", async () => { + const toolCall = (id?: string) => ({ + id: "stream", + choices: [ + { + delta: { + tool_calls: [ + { index: 0, id, function: { name: "read_file", arguments: '{"path":"test.ts"}' } }, + ], + }, + index: 0, + }, + ], + }) + const mockCreate = vitest + .fn() + .mockResolvedValueOnce( + asyncStreamFrom([ + toolCall(), + { id: "stream", choices: [{ delta: {}, finish_reason: "tool_calls", index: 0 }] }, + ]), + ) + .mockResolvedValueOnce( + asyncStreamFrom([ + toolCall("call_openrouter_stop"), + { id: "stream", choices: [{ delta: {}, finish_reason: "stop", index: 0 }] }, + ]), + ) + .mockResolvedValueOnce( + asyncStreamFrom([ + toolCall("call_openrouter_once"), + { id: "stream", choices: [{ delta: {}, finish_reason: "tool_calls", index: 0 }] }, + { id: "stream", choices: [{ delta: {}, finish_reason: "tool_calls", index: 0 }] }, + ]), + ) + Object.defineProperty(OpenAI.prototype, "chat", { + configurable: true, + value: { completions: { create: mockCreate } }, + }) + const handler = new OpenRouterHandler(mockOptions) + + const idlessChunks = await collectStream(handler.createMessage("idless", [])) + const stoppedChunks = await collectStream(handler.createMessage("stopped", [])) + const completedChunks = await collectStream(handler.createMessage("completed", [])) + + expect(idlessChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([]) + expect(stoppedChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([]) + expect(completedChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_openrouter_once" }, + ]) + }) + + it("isolates overlapping tool-call finalization between provider streams", async () => { + const { NativeToolCallParser } = await import("../../../core/assistant-message/NativeToolCallParser") + NativeToolCallParser.clearRawChunkState() + + let releaseFirstStream: (() => void) | undefined + let markFirstStreamPaused: (() => void) | undefined + const firstStreamRelease = new Promise((resolve) => { + releaseFirstStream = resolve + }) + const firstStreamPaused = new Promise((resolve) => { + markFirstStreamPaused = resolve + }) + const firstStream = async function* () { + yield { + id: "stream-a", + choices: [ + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_openrouter_a", + function: { name: "read_file", arguments: '{"path":"a' }, + }, + ], + }, + index: 0, + }, + ], + } + markFirstStreamPaused?.() + await firstStreamRelease + yield { + id: "stream-a", + choices: [{ delta: {}, finish_reason: "tool_calls", index: 0 }], + } + } + const secondStream = asyncStreamFrom([ + { + id: "stream-b", + choices: [ + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_openrouter_b", + function: { name: "read_file", arguments: '{"path":"b' }, + }, + ], + }, + index: 0, + }, + ], + }, + { id: "stream-b", choices: [{ delta: {}, finish_reason: "tool_calls", index: 0 }] }, + ]) + const mockCreate = vitest.fn().mockResolvedValueOnce(firstStream()).mockResolvedValueOnce(secondStream) + Object.defineProperty(OpenAI.prototype, "chat", { + configurable: true, + value: { completions: { create: mockCreate } }, + }) + const handler = new OpenRouterHandler(mockOptions) + + const firstChunksPromise = collectStreamAndParseToolCalls(handler.createMessage("first", [])) + await firstStreamPaused + const secondChunks = await collectStreamAndParseToolCalls(handler.createMessage("second", [])) + releaseFirstStream?.() + const firstChunks = await firstChunksPromise + + expect(secondChunks.chunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_openrouter_b" }, + ]) + expect(firstChunks.chunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_openrouter_a" }, + ]) + expect(firstChunks.parserEvents).toEqual([ + { type: "tool_call_start", id: "call_openrouter_a", name: "read_file" }, + { type: "tool_call_delta", id: "call_openrouter_a", delta: '{"path":"a' }, + ]) + expect(secondChunks.parserEvents).toEqual([ + { type: "tool_call_start", id: "call_openrouter_b", name: "read_file" }, + { type: "tool_call_delta", id: "call_openrouter_b", delta: '{"path":"b' }, + ]) + }) }) describe("completePrompt", () => { diff --git a/src/api/providers/__tests__/qwen-code-native-tools.spec.ts b/src/api/providers/__tests__/qwen-code-native-tools.spec.ts index 54df551d4e..e80f328ebb 100644 --- a/src/api/providers/__tests__/qwen-code-native-tools.spec.ts +++ b/src/api/providers/__tests__/qwen-code-native-tools.spec.ts @@ -10,6 +10,7 @@ vi.mock("node:fs", () => ({ const mockCreate = vi.fn() import { asyncStreamFrom, collectStream } from "../../../test-utils/stream" +import { collectStreamAndParseToolCalls } from "../../../test-utils/native-tool-call-stream" import { clearAllMocks } from "../../../test-utils/reset" vi.mock("openai", () => { return { @@ -285,6 +286,120 @@ describe("QwenCodeHandler Native Tools", () => { expect(endChunks[0].id).toBe("call_qwen_test") }) + it("emits completion only for identified calls and clears completed IDs", async () => { + const toolCall = (id?: string) => ({ + choices: [ + { + delta: { + tool_calls: [ + { index: 0, id, function: { name: "test_tool", arguments: '{"arg1":"value"}' } }, + ], + }, + }, + ], + }) + mockCreate + .mockImplementationOnce(() => + asyncStreamFrom([toolCall(), { choices: [{ delta: {}, finish_reason: "tool_calls" }] }]), + ) + .mockImplementationOnce(() => + asyncStreamFrom([toolCall("call_qwen_stop"), { choices: [{ delta: {}, finish_reason: "stop" }] }]), + ) + .mockImplementationOnce(() => + asyncStreamFrom([ + toolCall("call_qwen_once"), + { choices: [{ delta: {}, finish_reason: "tool_calls" }] }, + { choices: [{ delta: {}, finish_reason: "tool_calls" }] }, + ]), + ) + + const createMessage = () => handler.createMessage("test prompt", [], { taskId: "task", tools: testTools }) + const idlessChunks = await collectStream(createMessage()) + const stoppedChunks = await collectStream(createMessage()) + const completedChunks = await collectStream(createMessage()) + + expect(idlessChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([]) + expect(stoppedChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([]) + expect(completedChunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_qwen_once" }, + ]) + }) + + it("isolates overlapping tool-call finalization between provider streams", async () => { + let releaseFirstStream: (() => void) | undefined + let markFirstStreamPaused: (() => void) | undefined + const firstStreamRelease = new Promise((resolve) => { + releaseFirstStream = resolve + }) + const firstStreamPaused = new Promise((resolve) => { + markFirstStreamPaused = resolve + }) + const firstStream = async function* () { + yield { + choices: [ + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_qwen_a", + function: { name: "test_tool", arguments: '{"arg1":"a' }, + }, + ], + }, + }, + ], + } + markFirstStreamPaused?.() + await firstStreamRelease + yield { choices: [{ delta: {}, finish_reason: "tool_calls" }] } + } + const secondStream = asyncStreamFrom([ + { + choices: [ + { + delta: { + tool_calls: [ + { + index: 0, + id: "call_qwen_b", + function: { name: "test_tool", arguments: '{"arg1":"b' }, + }, + ], + }, + }, + ], + }, + { choices: [{ delta: {}, finish_reason: "tool_calls" }] }, + ]) + mockCreate.mockImplementationOnce(() => firstStream()).mockImplementationOnce(() => secondStream) + + const firstChunksPromise = collectStreamAndParseToolCalls( + handler.createMessage("first", [], { taskId: "task-a", tools: testTools }), + ) + await firstStreamPaused + const secondChunks = await collectStreamAndParseToolCalls( + handler.createMessage("second", [], { taskId: "task-b", tools: testTools }), + ) + releaseFirstStream?.() + const firstChunks = await firstChunksPromise + + expect(secondChunks.chunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_qwen_b" }, + ]) + expect(firstChunks.chunks.filter((chunk) => chunk.type === "tool_call_end")).toEqual([ + { type: "tool_call_end", id: "call_qwen_a" }, + ]) + expect(firstChunks.parserEvents).toEqual([ + { type: "tool_call_start", id: "call_qwen_a", name: "test_tool" }, + { type: "tool_call_delta", id: "call_qwen_a", delta: '{"arg1":"a' }, + ]) + expect(secondChunks.parserEvents).toEqual([ + { type: "tool_call_start", id: "call_qwen_b", name: "test_tool" }, + { type: "tool_call_delta", id: "call_qwen_b", delta: '{"arg1":"b' }, + ]) + }) + it("streams reasoning chunks from delta.reasoning_content", async () => { mockCreate.mockImplementationOnce(() => asyncStreamFrom([ diff --git a/src/api/providers/lm-studio.ts b/src/api/providers/lm-studio.ts index 59f484829c..040a0827d3 100644 --- a/src/api/providers/lm-studio.ts +++ b/src/api/providers/lm-studio.ts @@ -11,7 +11,6 @@ import { import type { ApiHandlerOptions } from "../../shared/api" -import { NativeToolCallParser } from "../../core/assistant-message/NativeToolCallParser" import { TagMatcher } from "../../utils/tag-matcher" import { convertToOpenAiMessages } from "../transform/openai-format" @@ -118,6 +117,7 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan text: chunk.data, }) as const, ) + const activeToolCallIds = new Set() for await (const chunk of results) { const delta = chunk.choices[0]?.delta @@ -142,6 +142,9 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan // Handle tool calls in stream - emit partial chunks for NativeToolCallParser if (delta?.tool_calls) { for (const toolCall of delta.tool_calls) { + if (toolCall.id) { + activeToolCallIds.add(toolCall.id) + } yield { type: "tool_call_partial", index: toolCall.index, @@ -153,11 +156,11 @@ export class LmStudioHandler extends BaseProvider implements SingleCompletionHan } // Process finish_reason to emit tool_call_end events - if (finishReason) { - const endEvents = NativeToolCallParser.processFinishReason(finishReason) - for (const event of endEvents) { - yield event + if (finishReason === "tool_calls") { + for (const id of activeToolCallIds) { + yield { type: "tool_call_end", id } } + activeToolCallIds.clear() } } diff --git a/src/api/providers/openrouter.ts b/src/api/providers/openrouter.ts index f61e007214..ed53c111b5 100644 --- a/src/api/providers/openrouter.ts +++ b/src/api/providers/openrouter.ts @@ -14,8 +14,6 @@ import { } from "@roo-code/types" import { TelemetryService } from "@roo-code/telemetry" -import { NativeToolCallParser } from "../../core/assistant-message/NativeToolCallParser" - import type { ApiHandlerOptions } from "../../shared/api" import { @@ -401,6 +399,7 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH // When reasoning_details has displayable content (reasoning.text or reasoning.summary), // we skip yielding the top-level reasoning field to avoid duplicate display. let hasYieldedReasoningFromDetails = false + const activeToolCallIds = new Set() for await (const chunk of stream) { // OpenRouter returns an error object instead of the OpenAI SDK throwing an error. @@ -491,6 +490,9 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH // Emit raw tool call chunks - NativeToolCallParser handles state management if ("tool_calls" in delta && Array.isArray(delta.tool_calls)) { for (const toolCall of delta.tool_calls) { + if (toolCall.id) { + activeToolCallIds.add(toolCall.id) + } yield { type: "tool_call_partial", index: toolCall.index, @@ -508,11 +510,11 @@ export class OpenRouterHandler extends BaseProvider implements SingleCompletionH // Process finish_reason to emit tool_call_end events // This ensures tool calls are finalized even if the stream doesn't properly close - if (finishReason) { - const endEvents = NativeToolCallParser.processFinishReason(finishReason) - for (const event of endEvents) { - yield event + if (finishReason === "tool_calls") { + for (const id of activeToolCallIds) { + yield { type: "tool_call_end", id } } + activeToolCallIds.clear() } if (chunk.usage) { diff --git a/src/api/providers/qwen-code.ts b/src/api/providers/qwen-code.ts index 7d98bcb77d..686e8ef8fd 100644 --- a/src/api/providers/qwen-code.ts +++ b/src/api/providers/qwen-code.ts @@ -8,8 +8,6 @@ import { type ModelInfo, type QwenCodeModelId, qwenCodeModels, qwenCodeDefaultMo import type { ApiHandlerOptions } from "../../shared/api" -import { NativeToolCallParser } from "../../core/assistant-message/NativeToolCallParser" - import { convertToOpenAiMessages } from "../transform/openai-format" import { ApiStream } from "../transform/stream" @@ -243,6 +241,7 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan let fullContent = "" + const activeToolCallIds = new Set() for await (const apiChunk of stream) { const delta = apiChunk.choices[0]?.delta ?? {} const finishReason = apiChunk.choices[0]?.finish_reason @@ -293,6 +292,9 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan // Handle tool calls in stream - emit partial chunks for NativeToolCallParser if (delta.tool_calls) { for (const toolCall of delta.tool_calls) { + if (toolCall.id) { + activeToolCallIds.add(toolCall.id) + } yield { type: "tool_call_partial", index: toolCall.index, @@ -304,11 +306,11 @@ export class QwenCodeHandler extends BaseProvider implements SingleCompletionHan } // Process finish_reason to emit tool_call_end events - if (finishReason) { - const endEvents = NativeToolCallParser.processFinishReason(finishReason) - for (const event of endEvents) { - yield event + if (finishReason === "tool_calls") { + for (const id of activeToolCallIds) { + yield { type: "tool_call_end", id } } + activeToolCallIds.clear() } if (apiChunk.usage) { diff --git a/src/core/assistant-message/NativeToolCallParser.ts b/src/core/assistant-message/NativeToolCallParser.ts index 9639ae1baa..828e926504 100644 --- a/src/core/assistant-message/NativeToolCallParser.ts +++ b/src/core/assistant-message/NativeToolCallParser.ts @@ -51,28 +51,43 @@ export type ToolCallStreamEvent = ApiStreamToolCallStartChunk | ApiStreamToolCal * provider-level raw chunks into start/delta/end events. */ export class NativeToolCallParser { + private static readonly defaultScope = {} + // Streaming state management for argument accumulation (keyed by tool call id) // Note: name is string to accommodate dynamic MCP tools (mcp--serverName--toolName) - private static streamingToolCalls = new Map< - string, - { - id: string - name: string - argumentsAccumulator: string - } + private static streamingToolCallsByScope = new WeakMap< + object, + Map >() - // Raw chunk tracking state (keyed by index from API stream) - private static rawChunkTracker = new Map< - number, - { - id: string - name: string - hasStarted: boolean - deltaBuffer: string[] - } + // Raw chunk tracking state (keyed by index from one API stream) + private static rawChunkTrackersByScope = new WeakMap< + object, + Map >() + public static createScope(): object { + return {} + } + + private static getStreamingToolCalls(scope = this.defaultScope) { + let streamingToolCalls = this.streamingToolCallsByScope.get(scope) + if (!streamingToolCalls) { + streamingToolCalls = new Map() + this.streamingToolCallsByScope.set(scope, streamingToolCalls) + } + return streamingToolCalls + } + + private static getRawChunkTracker(scope = this.defaultScope) { + let rawChunkTracker = this.rawChunkTrackersByScope.get(scope) + if (!rawChunkTracker) { + rawChunkTracker = new Map() + this.rawChunkTrackersByScope.set(scope, rawChunkTracker) + } + return rawChunkTracker + } + private static coerceOptionalBoolean(value: unknown): boolean | undefined { if (typeof value === "boolean") { return value @@ -96,16 +111,20 @@ export class NativeToolCallParser { * This is the entry point for providers that emit tool_call_partial chunks. * Returns an array of events to be processed by the consumer. */ - public static processRawChunk(chunk: { - index: number - id?: string - name?: string - arguments?: string - }): ToolCallStreamEvent[] { + public static processRawChunk( + chunk: { + index: number + id?: string + name?: string + arguments?: string + }, + scope = this.defaultScope, + ): ToolCallStreamEvent[] { const events: ToolCallStreamEvent[] = [] const { index, id, name, arguments: args } = chunk + const rawChunkTracker = this.getRawChunkTracker(scope) - let tracked = this.rawChunkTracker.get(index) + let tracked = rawChunkTracker.get(index) // Initialize new tool call tracking when we receive an id if (id && !tracked) { @@ -115,7 +134,7 @@ export class NativeToolCallParser { hasStarted: false, deltaBuffer: [], } - this.rawChunkTracker.set(index, tracked) + rawChunkTracker.set(index, tracked) } if (!tracked) { @@ -167,11 +186,15 @@ export class NativeToolCallParser { * Process stream finish reason. * Emits end events when finish_reason is 'tool_calls'. */ - public static processFinishReason(finishReason: string | null | undefined): ToolCallStreamEvent[] { + public static processFinishReason( + finishReason: string | null | undefined, + scope = this.defaultScope, + ): ToolCallStreamEvent[] { const events: ToolCallStreamEvent[] = [] + const rawChunkTracker = this.rawChunkTrackersByScope.get(scope) - if (finishReason === "tool_calls" && this.rawChunkTracker.size > 0) { - for (const [, tracked] of this.rawChunkTracker.entries()) { + if (finishReason === "tool_calls" && rawChunkTracker) { + for (const [, tracked] of rawChunkTracker.entries()) { events.push({ type: "tool_call_end", id: tracked.id, @@ -186,11 +209,12 @@ export class NativeToolCallParser { * Finalize any remaining tool calls that weren't explicitly ended. * Should be called at the end of stream processing. */ - public static finalizeRawChunks(): ToolCallStreamEvent[] { + public static finalizeRawChunks(scope = this.defaultScope): ToolCallStreamEvent[] { const events: ToolCallStreamEvent[] = [] + const rawChunkTracker = this.rawChunkTrackersByScope.get(scope) - if (this.rawChunkTracker.size > 0) { - for (const [, tracked] of this.rawChunkTracker.entries()) { + if (rawChunkTracker) { + for (const [, tracked] of rawChunkTracker.entries()) { if (tracked.hasStarted) { events.push({ type: "tool_call_end", @@ -198,8 +222,8 @@ export class NativeToolCallParser { }) } } - this.rawChunkTracker.clear() } + this.rawChunkTrackersByScope.delete(scope) return events } @@ -208,8 +232,8 @@ export class NativeToolCallParser { * Clear all raw chunk tracking state. * Should be called when a new API request starts. */ - public static clearRawChunkState(): void { - this.rawChunkTracker.clear() + public static clearRawChunkState(scope = this.defaultScope): void { + this.rawChunkTrackersByScope.delete(scope) } /** @@ -217,8 +241,8 @@ export class NativeToolCallParser { * Initializes tracking for incremental argument parsing. * Accepts string to support both ToolName and dynamic MCP tools (mcp--serverName--toolName). */ - public static startStreamingToolCall(id: string, name: string): void { - this.streamingToolCalls.set(id, { + public static startStreamingToolCall(id: string, name: string, scope = this.defaultScope): void { + this.getStreamingToolCalls(scope).set(id, { id, name, argumentsAccumulator: "", @@ -230,16 +254,16 @@ export class NativeToolCallParser { * Should be called when a new API request starts to prevent memory leaks * from interrupted streams. */ - public static clearAllStreamingToolCalls(): void { - this.streamingToolCalls.clear() + public static clearAllStreamingToolCalls(scope = this.defaultScope): void { + this.streamingToolCallsByScope.delete(scope) } /** * Check if there are any active streaming tool calls. * Useful for debugging and testing. */ - public static hasActiveStreamingToolCalls(): boolean { - return this.streamingToolCalls.size > 0 + public static hasActiveStreamingToolCalls(scope = this.defaultScope): boolean { + return (this.streamingToolCallsByScope.get(scope)?.size ?? 0) > 0 } /** @@ -247,8 +271,8 @@ export class NativeToolCallParser { * Uses partial-json-parser to extract values from incomplete JSON immediately. * Returns a partial ToolUse with currently parsed parameters. */ - public static processStreamingChunk(id: string, chunk: string): ToolUse | null { - const toolCall = this.streamingToolCalls.get(id) + public static processStreamingChunk(id: string, chunk: string, scope = this.defaultScope): ToolUse | null { + const toolCall = this.streamingToolCallsByScope.get(scope)?.get(id) if (!toolCall) { return null } @@ -291,8 +315,12 @@ export class NativeToolCallParser { * Finalize a streaming tool call. * Parses the complete JSON and returns the final ToolUse or McpToolUse. */ - public static finalizeStreamingToolCall(id: string): ToolUse | McpToolUse | null { - const toolCall = this.streamingToolCalls.get(id) + public static finalizeStreamingToolCall(id: string, scope = this.defaultScope): ToolUse | McpToolUse | null { + const streamingToolCalls = this.streamingToolCallsByScope.get(scope) + if (!streamingToolCalls) { + return null + } + const toolCall = streamingToolCalls.get(id) if (!toolCall) { return null } @@ -306,7 +334,12 @@ export class NativeToolCallParser { }) // Clean up streaming state - this.streamingToolCalls.delete(id) + streamingToolCalls.delete(id) + // Stryker disable next-line ConditionalExpression: retaining an empty WeakMap value is only observable as GC eligibility. + if (streamingToolCalls.size === 0) { + // Stryker disable next-line CallExpression: deleting an empty WeakMap value is only observable as GC eligibility. + this.streamingToolCallsByScope.delete(scope) + } return finalToolUse } diff --git a/src/core/assistant-message/__tests__/NativeToolCallParser.spec.ts b/src/core/assistant-message/__tests__/NativeToolCallParser.spec.ts index 2c15e12069..37d63297af 100644 --- a/src/core/assistant-message/__tests__/NativeToolCallParser.spec.ts +++ b/src/core/assistant-message/__tests__/NativeToolCallParser.spec.ts @@ -293,7 +293,165 @@ describe("NativeToolCallParser", () => { }) }) + describe("processFinishReason", () => { + it("keeps finish-reason events scoped while preserving default-scope compatibility", () => { + const firstScope = NativeToolCallParser.createScope() + const secondScope = NativeToolCallParser.createScope() + + NativeToolCallParser.processRawChunk({ index: 0, id: "call_first_finish", name: "read_file" }, firstScope) + NativeToolCallParser.processRawChunk({ index: 0, id: "call_second_finish", name: "read_file" }, secondScope) + + expect(NativeToolCallParser.processFinishReason(null, firstScope)).toEqual([]) + expect(NativeToolCallParser.processFinishReason(undefined, firstScope)).toEqual([]) + expect(NativeToolCallParser.processFinishReason("stop", firstScope)).toEqual([]) + expect(NativeToolCallParser.processFinishReason("tool_calls", firstScope)).toEqual([ + { type: "tool_call_end", id: "call_first_finish" }, + ]) + expect(NativeToolCallParser.processFinishReason("tool_calls", secondScope)).toEqual([ + { type: "tool_call_end", id: "call_second_finish" }, + ]) + + NativeToolCallParser.processRawChunk({ + index: 0, + id: "call_default_finish", + name: "read_file", + }) + expect(NativeToolCallParser.processFinishReason("tool_calls")).toEqual([ + { type: "tool_call_end", id: "call_default_finish" }, + ]) + + NativeToolCallParser.clearRawChunkState(firstScope) + NativeToolCallParser.clearRawChunkState(secondScope) + NativeToolCallParser.clearRawChunkState() + }) + + it("returns no events for unused and argument-only scopes", () => { + const unusedScope = NativeToolCallParser.createScope() + const argumentOnlyScope = NativeToolCallParser.createScope() + + expect(NativeToolCallParser.processFinishReason("tool_calls", unusedScope)).toEqual([]) + expect( + NativeToolCallParser.processRawChunk( + { index: 0, arguments: '{"path":"buffered.ts"}' }, + argumentOnlyScope, + ), + ).toEqual([]) + expect(NativeToolCallParser.processFinishReason("tool_calls", argumentOnlyScope)).toEqual([]) + expect(NativeToolCallParser.finalizeRawChunks(argumentOnlyScope)).toEqual([]) + }) + }) + describe("processStreamingChunk", () => { + it("retains peer calls until each call in a scope is finalized", () => { + const scope = NativeToolCallParser.createScope() + NativeToolCallParser.startStreamingToolCall("call_first", "read_file", scope) + NativeToolCallParser.startStreamingToolCall("call_second", "read_file", scope) + NativeToolCallParser.processStreamingChunk("call_first", '{"path":"first.ts"}', scope) + NativeToolCallParser.processStreamingChunk("call_second", '{"path":"second.ts"}', scope) + + const firstResult = NativeToolCallParser.finalizeStreamingToolCall("call_first", scope) + expect(firstResult?.type).toBe("tool_use") + if (firstResult?.type === "tool_use") expect(firstResult.nativeArgs).toMatchObject({ path: "first.ts" }) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(scope)).toBe(true) + const secondResult = NativeToolCallParser.finalizeStreamingToolCall("call_second", scope) + expect(secondResult?.type).toBe("tool_use") + if (secondResult?.type === "tool_use") expect(secondResult.nativeArgs).toMatchObject({ path: "second.ts" }) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(scope)).toBe(false) + }) + + it("clears active raw and streaming state without affecting unused scopes", () => { + const activeScope = NativeToolCallParser.createScope() + const unusedScope = NativeToolCallParser.createScope() + NativeToolCallParser.processRawChunk({ index: 0, id: "call_active", name: "read_file" }, activeScope) + NativeToolCallParser.startStreamingToolCall("call_active", "read_file", activeScope) + + NativeToolCallParser.clearRawChunkState(activeScope) + NativeToolCallParser.clearAllStreamingToolCalls(activeScope) + + expect(NativeToolCallParser.finalizeRawChunks(activeScope)).toEqual([]) + expect(NativeToolCallParser.processFinishReason("tool_calls", activeScope)).toEqual([]) + expect(NativeToolCallParser.processStreamingChunk("call_active", "{}", activeScope)).toBeNull() + expect(NativeToolCallParser.processStreamingChunk("missing", "{}", unusedScope)).toBeNull() + expect(NativeToolCallParser.hasActiveStreamingToolCalls(activeScope)).toBe(false) + }) + + it("keeps interleaved task streams isolated", () => { + const firstScope = NativeToolCallParser.createScope() + const secondScope = NativeToolCallParser.createScope() + + const firstStart = NativeToolCallParser.processRawChunk( + { index: 0, id: "call_first", name: "read_file" }, + firstScope, + ) + NativeToolCallParser.startStreamingToolCall("call_first", "read_file", firstScope) + + NativeToolCallParser.clearRawChunkState(secondScope) + NativeToolCallParser.clearAllStreamingToolCalls(secondScope) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(firstScope)).toBe(true) + + const secondStart = NativeToolCallParser.processRawChunk( + { index: 0, id: "call_second", name: "read_file" }, + secondScope, + ) + + expect(firstStart).toEqual([{ type: "tool_call_start", id: "call_first", name: "read_file" }]) + expect(secondStart).toEqual([{ type: "tool_call_start", id: "call_second", name: "read_file" }]) + + NativeToolCallParser.startStreamingToolCall("call_second", "read_file", secondScope) + + const firstDelta = NativeToolCallParser.processRawChunk( + { index: 0, arguments: JSON.stringify({ path: "first.ts" }) }, + firstScope, + ) + const secondDelta = NativeToolCallParser.processRawChunk( + { index: 0, arguments: JSON.stringify({ path: "second.ts" }) }, + secondScope, + ) + + expect(firstDelta).toEqual([ + { type: "tool_call_delta", id: "call_first", delta: JSON.stringify({ path: "first.ts" }) }, + ]) + expect(secondDelta).toEqual([ + { type: "tool_call_delta", id: "call_second", delta: JSON.stringify({ path: "second.ts" }) }, + ]) + if (firstDelta[0]?.type !== "tool_call_delta" || secondDelta[0]?.type !== "tool_call_delta") { + throw new Error("Expected argument delta events") + } + + NativeToolCallParser.processStreamingChunk("call_first", firstDelta[0].delta, firstScope) + NativeToolCallParser.processStreamingChunk("call_second", secondDelta[0].delta, secondScope) + + const firstFinalizeEvents = NativeToolCallParser.finalizeRawChunks(firstScope) + expect(firstFinalizeEvents).toEqual([{ type: "tool_call_end", id: "call_first" }]) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(firstScope)).toBe(true) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(secondScope)).toBe(true) + + const firstResult = NativeToolCallParser.finalizeStreamingToolCall("call_first", firstScope) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(firstScope)).toBe(false) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(secondScope)).toBe(true) + + const secondFinalizeEvents = NativeToolCallParser.finalizeRawChunks(secondScope) + expect(secondFinalizeEvents).toEqual([{ type: "tool_call_end", id: "call_second" }]) + const secondResult = NativeToolCallParser.finalizeStreamingToolCall("call_second", secondScope) + expect(NativeToolCallParser.hasActiveStreamingToolCalls(secondScope)).toBe(false) + expect(firstResult?.type).toBe("tool_use") + expect(secondResult?.type).toBe("tool_use") + if (firstResult?.type !== "tool_use" || secondResult?.type !== "tool_use") { + throw new Error("Expected native tool uses") + } + expect(firstResult.nativeArgs).toEqual({ path: "first.ts" }) + expect(secondResult.nativeArgs).toEqual({ path: "second.ts" }) + + expect(NativeToolCallParser.finalizeRawChunks(firstScope)).toEqual([]) + expect(NativeToolCallParser.finalizeStreamingToolCall("call_first", firstScope)).toBeNull() + expect( + NativeToolCallParser.processRawChunk({ index: 0, arguments: "ignored-after-cleanup" }, firstScope), + ).toEqual([]) + expect( + NativeToolCallParser.processRawChunk({ index: 0, id: "call_reprobe", name: "read_file" }, firstScope), + ).toEqual([{ type: "tool_call_start", id: "call_reprobe", name: "read_file" }]) + }) + describe("read_file tool", () => { it("should emit a partial ToolUse with nativeArgs.path during streaming", () => { const id = "toolu_streaming_123" diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 37281a9010..977af7f6f4 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -2999,9 +2999,7 @@ export class Task extends EventEmitter implements TaskLike { this.presentAssistantMessageHasPendingUpdates = false // No legacy text-stream tool parser. this.streamingToolCallIndices.clear() - // Clear any leftover streaming tool call state from previous interrupted streams - NativeToolCallParser.clearAllStreamingToolCalls() - NativeToolCallParser.clearRawChunkState() + const nativeToolCallParserScope = NativeToolCallParser.createScope() await this.diffViewProvider.reset() @@ -3096,12 +3094,15 @@ export class Task extends EventEmitter implements TaskLike { case "tool_call_partial": { // Process raw tool call chunk through NativeToolCallParser // which handles tracking, buffering, and emits events - const events = NativeToolCallParser.processRawChunk({ - index: chunk.index, - id: chunk.id, - name: chunk.name, - arguments: chunk.arguments, - }) + const events = NativeToolCallParser.processRawChunk( + { + index: chunk.index, + id: chunk.id, + name: chunk.name, + arguments: chunk.arguments, + }, + nativeToolCallParserScope, + ) for (const event of events) { if (event.type === "tool_call_start") { @@ -3118,7 +3119,11 @@ export class Task extends EventEmitter implements TaskLike { } // Initialize streaming in NativeToolCallParser - NativeToolCallParser.startStreamingToolCall(event.id, event.name as ToolName) + NativeToolCallParser.startStreamingToolCall( + event.id, + event.name as ToolName, + nativeToolCallParserScope, + ) // Before adding a new tool, finalize any preceding text block // This prevents the text block from blocking tool presentation @@ -3153,6 +3158,7 @@ export class Task extends EventEmitter implements TaskLike { const partialToolUse = NativeToolCallParser.processStreamingChunk( event.id, event.delta, + nativeToolCallParserScope, ) if (partialToolUse) { @@ -3172,7 +3178,10 @@ export class Task extends EventEmitter implements TaskLike { } } else if (event.type === "tool_call_end") { // Finalize the streaming tool call - const finalToolUse = NativeToolCallParser.finalizeStreamingToolCall(event.id) + const finalToolUse = NativeToolCallParser.finalizeStreamingToolCall( + event.id, + nativeToolCallParserScope, + ) // Get the index for this tool call const toolUseIndex = this.streamingToolCallIndices.get(event.id) @@ -3570,11 +3579,14 @@ export class Task extends EventEmitter implements TaskLike { // Finalize any remaining streaming tool calls that weren't explicitly ended // This is critical for MCP tools which need tool_call_end events to be properly // converted from ToolUse to McpToolUse via finalizeStreamingToolCall() - const finalizeEvents = NativeToolCallParser.finalizeRawChunks() + const finalizeEvents = NativeToolCallParser.finalizeRawChunks(nativeToolCallParserScope) for (const event of finalizeEvents) { if (event.type === "tool_call_end") { // Finalize the streaming tool call - const finalToolUse = NativeToolCallParser.finalizeStreamingToolCall(event.id) + const finalToolUse = NativeToolCallParser.finalizeStreamingToolCall( + event.id, + nativeToolCallParserScope, + ) // Get the index for this tool call const toolUseIndex = this.streamingToolCallIndices.get(event.id) diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index 37e228f887..ef8c21f5f2 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -27,6 +27,7 @@ import { ContextProxy } from "../../config/ContextProxy" import { processUserContentMentions } from "../../mentions/processUserContentMentions" import { MultiSearchReplaceDiffStrategy } from "../../diff/strategies/multi-search-replace" import type { ApiMessage } from "../../task-persistence" +import { asyncStreamFrom } from "../../../test-utils/stream" type TaskTestAccess = { getSystemPrompt: () => Promise @@ -465,6 +466,93 @@ describe("Cline", () => { }) }) + describe("native tool-call request isolation", () => { + it("keeps overlapping Task parser state scoped to each request", async () => { + const firstTask = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "first task", + startTask: false, + }) + const secondTask = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "second task", + startTask: false, + }) + + let releaseFirstStream: (() => void) | undefined + let markFirstStreamPaused: (() => void) | undefined + const firstStreamRelease = new Promise((resolve) => { + releaseFirstStream = resolve + }) + const firstStreamPaused = new Promise((resolve) => { + markFirstStreamPaused = resolve + }) + const firstStream = async function* (): AsyncGenerator { + yield { + type: "tool_call_partial", + index: 0, + id: "call_first", + name: "read_file", + } + yield { type: "tool_call_partial", index: 0, arguments: '{"path":"first' } + yield { type: "usage", inputTokens: 0, outputTokens: 0 } + markFirstStreamPaused?.() + await firstStreamRelease + yield { type: "tool_call_partial", index: 0, arguments: 'Task.ts"}' } + } + + for (const task of [firstTask, secondTask]) { + vi.spyOn(task.diffViewProvider, "reset").mockResolvedValue(undefined) + vi.spyOn(getTaskTestAccess(task), "safeEnsureModelFetched").mockResolvedValue(undefined) + vi.spyOn(getTaskTestAccess(task), "presentAssistantMessageSafe").mockImplementation(() => {}) + } + vi.spyOn(firstTask, "attemptApiRequest").mockImplementation(() => firstStream()) + vi.spyOn(secondTask, "attemptApiRequest").mockImplementation(() => + asyncStreamFrom([ + { + type: "tool_call_partial", + index: 0, + id: "call_second", + name: "read_file", + }, + { type: "tool_call_partial", index: 0, arguments: '{"path":"secondTask.ts"}' }, + ]), + ) + + const firstRequest = firstTask.recursivelyMakeClineRequests([{ type: "text", text: "first request" }]) + await firstStreamPaused + await secondTask.recursivelyMakeClineRequests([{ type: "text", text: "second request" }]) + releaseFirstStream?.() + await firstRequest + + const firstAssistantMessage = firstTask.apiConversationHistory.find( + (message) => message.role === "assistant", + ) + const secondAssistantMessage = secondTask.apiConversationHistory.find( + (message) => message.role === "assistant", + ) + + expect(firstAssistantMessage?.content).toEqual([ + { + type: "tool_use", + id: "call_first", + name: "read_file", + input: { path: "firstTask.ts" }, + }, + ]) + expect(secondAssistantMessage?.content).toEqual([ + { + type: "tool_use", + id: "call_second", + name: "read_file", + input: { path: "secondTask.ts" }, + }, + ]) + }) + }) + describe("constructor", () => { it("should always have diff strategy defined", async () => { const cline = new Task({ diff --git a/src/test-utils/native-tool-call-stream.ts b/src/test-utils/native-tool-call-stream.ts new file mode 100644 index 0000000000..14cbac8577 --- /dev/null +++ b/src/test-utils/native-tool-call-stream.ts @@ -0,0 +1,33 @@ +import type { ApiStreamChunk } from "../api/transform/stream" +import { NativeToolCallParser, type ToolCallStreamEvent } from "../core/assistant-message/NativeToolCallParser" + +export async function collectStreamAndParseToolCalls(stream: AsyncIterable): Promise<{ + chunks: ApiStreamChunk[] + parserEvents: ToolCallStreamEvent[] +}> { + const chunks: ApiStreamChunk[] = [] + const parserEvents: ToolCallStreamEvent[] = [] + const parserScope = NativeToolCallParser.createScope() + + try { + for await (const chunk of stream) { + if (chunk.type === "tool_call_partial") { + parserEvents.push( + ...NativeToolCallParser.processRawChunk( + { + index: chunk.index, + id: chunk.id, + name: chunk.name, + arguments: chunk.arguments, + }, + parserScope, + ), + ) + } + chunks.push(chunk) + } + return { chunks, parserEvents } + } finally { + NativeToolCallParser.clearRawChunkState(parserScope) + } +}