diff --git a/docs/architecture/task-lifecycle-model.md b/docs/architecture/task-lifecycle-model.md index 7b2eabdd63..eaeb61d987 100644 --- a/docs/architecture/task-lifecycle-model.md +++ b/docs/architecture/task-lifecycle-model.md @@ -6,15 +6,16 @@ Zoo Code checks task lifecycle protocols through one umbrella command for indepe pnpm lifecycle:model-check ``` -The baseline command runs seven independent bounded checks in sequence: +The baseline command runs eight independent bounded checks in sequence: 1. the persisted task delegation lifecycle; 2. shared-store concurrency across task-history hosts; 3. production-backed handoff reducers with an abstract provider/scheduler protocol; 4. the task cleanup protocol; 5. request-stream parser scoping; -6. completion persistence; and -7. delegated-mode reader refinement. +6. completion persistence; +7. delegated-mode reader refinement; and +8. tool-turn continuation. The planned two-sibling fan-out protocol is intentionally outside the baseline and CI umbrella. Run it explicitly with `pnpm fanout-protocol:model-check`; it describes optional future functionality, not current production coverage. @@ -91,6 +92,40 @@ The known-unsafe witnesses currently compare exact shortest action sequences. Th The umbrella command also runs a separate bounded child model for in-memory abort, disposal, and provider-shutdown ordering. It models cleanup settlement and rejection as environment transitions and makes no filesystem, editor Promise, fairness, or timing-liveness claim. See [Task cleanup protocol model check](./task-cleanup-protocol-model.md). +## Tool-turn continuation model + +`scripts/check-tool-turn-continuation.ts` is a separate bounded child model for the in-memory handoff between the presenter and the request loop. It also runs directly with `pnpm tool-turn:model-check`. It models one assistant turn with two tool blocks and one trailing text block. It makes no claim about tool handlers, approvals, or timing liveness. + +The model maps its state to production fields as follows: + +| Model state | Production field | +| -------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | +| `phase`, `index` | `Task.assistantMessageContent` partial flags, matching `tool_result` blocks in `userMessageContent`, and `currentStreamingContentIndex` | +| `stream` | `didCompleteReadingStream` | +| `lock`, `owner`, `pending` | `presentAssistantMessageLocked` and `presentAssistantMessageHasPendingUpdates`, owned by the `presentAssistantMessage` wrapper | +| `latch` | `userMessageContentReady` | +| `epoch` | the per-request reset in the request loop (`Task.ts`) | +| `poll` and `post-wait` | the `pWaitFor` condition and the abort check that follows it | + +The checker enforces these invariants in every reachable state: + +1. The next request never starts while a tool block has no result. +2. The next request never starts while a presenter pass is live. +3. The presenter lock is never held without a live pass, including after a presenter failure. +4. A released lock never hides a pending update, unless the task aborted or a pass failed. +5. The next request never starts after abort. +6. Every reachable state either can continue without abort or has started the request. +7. A live pass always holds the lock. + +Five injected policies must each violate one named invariant: a latch-only wait (the #1883 hang), a readiness check with no lock gate, a check on the first tool result only, a presenter that keeps the lock after a failure, and a wait with no abort check. Six landmarks keep the intended paths reachable, including recovery from a lost latch, continuation after a presenter failure, a two-tool turn, abort on a ready turn, and a stale pass that unwinds after a retry. + +The model has two documented gaps. The checker asserts that both still reproduce, so a fix must update the model and this section: + +- A presenter failure on a tool block before it pushes a result releases the lock but leaves no result. The derived check stays false and the wait has no timeout. This hang existed before the derived check, when the lock stayed held. The change deliberately does not synthesize a tool result. +- The wrapper `finally` clears the lock without an ownership check. After a mid-stream retry resets the turn, a stale pass that unwinds late can clear the lock of a newer pass and break invariant 7. A per-pass owner token removes this gap in the model. Invariants 1 to 7 hold for that policy. + +This model is not production-backed. The flags are fields on `Task`, not pure reducers. The real-presenter tests in `Task.spec.ts` and `presentAssistantMessage-custom-tool.spec.ts` tie it to production. A later change can extract the readiness check as a pure function so the checker and `Task` share it. + ## Provider handoff and scheduler model `scripts/check-provider-handoff-scheduler.ts` is a separate bounded adapter model for the runtime boundary that the persisted lifecycle graph does not represent. Its breadth-first explorer normalizes provider-keyed records and owner arrays before deduplicating canonical states, then exhaustively explores enabled action orderings through depth 15 with a 20,000-state budget. It imports `selectHandoffExecutionContext` and the existing `delegateTaskToChild` and `completeDelegatedChild` reducers. A direct saved, unsaved, and locked-profile matrix checks task-local configuration selection within those cases. Stale provider lookup is caught before this pure selector, so focused provider tests check the failed lookup, contextual log, and fallback. The protocol state then models two provider instances, their claims and parent snapshots, authoritative parent/child records, current task publication, commit/start ownership, the child scheduler permit, queued and resumed parent state, and one bounded redelegation generation. diff --git a/package.json b/package.json index 1bf0b4e8dd..d8d84be173 100644 --- a/package.json +++ b/package.json @@ -14,9 +14,10 @@ "test": "turbo test --log-order grouped --output-logs new-only", "test:code-qa-ci": "node --test scripts/code-qa-workflow.test.mjs", "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 && tsx scripts/check-provider-handoff-scheduler.ts && pnpm cleanup-protocol:model-check && pnpm parser-scope:model-check && tsx scripts/check-completion-persistence.ts && tsx scripts/check-delegated-mode-readers.ts", + "lifecycle:model-check": "tsx scripts/check-task-lifecycle.ts && tsx scripts/check-task-store-concurrency.ts && tsx scripts/check-provider-handoff-scheduler.ts && pnpm cleanup-protocol:model-check && pnpm parser-scope:model-check && tsx scripts/check-completion-persistence.ts && tsx scripts/check-delegated-mode-readers.ts && pnpm tool-turn:model-check", "fanout-protocol:model-check": "tsx scripts/check-task-fanout-protocol.ts", "cleanup-protocol:model-check": "tsx scripts/check-task-cleanup-protocol.ts", + "tool-turn:model-check": "tsx scripts/check-tool-turn-continuation.ts", "parser-scope:model-check": "node scripts/run-native-tool-call-parser-scoping.mjs", "mcp:integration-check": "tsx scripts/check-mcp-oauth-integration.ts", "test:coverage": "turbo test:coverage --log-order grouped --output-logs new-only", diff --git a/scripts/check-tool-turn-continuation.ts b/scripts/check-tool-turn-continuation.ts new file mode 100644 index 0000000000..d4ac8c0fdf --- /dev/null +++ b/scripts/check-tool-turn-continuation.ts @@ -0,0 +1,475 @@ +import assert from "node:assert/strict" + +// Bounded model of tool-turn continuation: the presenter dispatch lock, the one-shot +// `userMessageContentReady` latch, and the derived readiness check that +// `Task.hasCompleteToolResultsForCurrentTurn()` adds to the request wait (#1883). +// It models one assistant turn with up to two tool blocks and one trailing text block. +// Tool handlers, approvals, and timing liveness are outside this model. + +type Phase = "absent" | "streaming" | "final" | "running" | "resulted" +type PassStep = "block" | "running" | "settling" | "check" | "done" +type Request = "waiting" | "released" | "started" | "aborted" + +interface Pass { + id: number + epoch: number + step: PassStep +} + +interface ModelState { + phase: Phase[] // Task.assistantMessageContent partial flag plus tool_result presence + stream: "open" | "complete" // Task.didCompleteReadingStream + index: number // Task.currentStreamingContentIndex + lock: boolean // Task.presentAssistantMessageLocked + owner: number + pending: boolean // Task.presentAssistantMessageHasPendingUpdates + latch: boolean // Task.userMessageContentReady + abort: boolean + threw: boolean + epoch: number // Task.ts request-loop reset of the turn state + nextId: number + passes: Pass[] + request: Request // the request wait and the next API request +} + +interface Policy { + name: string + derived: "every" | "first" | "off" + lockGate: boolean + release: "unconditional" | "owner" + releaseOnThrow: boolean + abortCheck: boolean + retry: boolean + toolThrow: boolean // also allow a failure on a tool block before its result is pushed +} + +interface Step { + action: string + state: ModelState + tag?: string +} + +interface Violation { + invariant: string + message: string + trace: Step[] +} + +const kinds = ["tool", "tool", "text"] as const +const MAX_DEPTH = 60 +const MAX_STATES = 500_000 +const expectedActions = [ + "begin", + "finalize", + "complete", + "start-block", + "push", + "finish", + "check", + "throw", + "abort", + "retry-reset", + "stale-unwind", + "poll", + "post-wait", +] as const + +const production: Policy = { + name: "production", + derived: "every", + lockGate: true, + release: "unconditional", + releaseOnThrow: true, + abortCheck: true, + retry: false, + toolThrow: false, +} +const ownerToken: Policy = { ...production, name: "owner-token", release: "owner", retry: true } +const unconditionalWithRetry: Policy = { ...production, name: "unconditional-release-with-retry", retry: true } +const toolThrowGap: Policy = { ...production, name: "tool-throw-before-result", toolThrow: true } +const injected: Array<{ policy: Policy; expected: string; detectStuck?: false }> = [ + { policy: { ...production, name: "latch-only", derived: "off" }, expected: "stuck" }, + { policy: { ...production, name: "no-lock-gate", lockGate: false }, expected: "continue-under-pass" }, + { + policy: { ...production, name: "first-tool-only", derived: "first", toolThrow: true }, + expected: "premature-continue", + detectStuck: false, // the shortest violation is the hang, which hides the premature continue + }, + { policy: { ...production, name: "no-finally-on-throw", releaseOnThrow: false }, expected: "stranded-lock" }, + { policy: { ...production, name: "no-abort-check", abortCheck: false }, expected: "continue-after-abort" }, +] + +function initialState(): ModelState { + return { + phase: ["absent", "absent", "absent"], + stream: "open", + index: 0, + lock: false, + owner: 0, + pending: false, + latch: false, + abort: false, + threw: false, + epoch: 0, + nextId: 1, + passes: [], + request: "waiting", + } +} + +function clone(state: ModelState): ModelState { + return structuredClone(state) +} + +function blockCount(state: ModelState): number { + return state.phase.filter((phase) => phase !== "absent").length +} + +function currentPasses(state: ModelState): Pass[] { + return state.passes.filter((pass) => pass.epoch === state.epoch && pass.step !== "done") +} + +// Mirrors presentAssistantMessage(): queue when locked, otherwise take the lock. +function enter(state: ModelState): ModelState { + const next = clone(state) + if (next.abort) return next + if (next.lock) { + next.pending = true + return next + } + next.lock = true + next.owner = next.nextId + next.passes.push({ id: next.nextId, epoch: next.epoch, step: "block" }) + next.nextId += 1 + return next +} + +// Mirrors the wrapper's `finally`. The owner policy is the proposed fix for a stale pass. +function release(state: ModelState, pass: Pass, policy: Policy): void { + pass.step = "done" + if (policy.release === "owner" && state.owner !== pass.id) return + state.lock = false + state.owner = 0 +} + +function afterBlock(state: ModelState, pass: Pass): void { + if (state.index >= blockCount(state)) { + if (state.stream === "complete") state.latch = true + pass.step = "check" + } else { + pass.step = "block" + } +} + +// Mirrors Task.hasCompleteToolResultsForCurrentTurn(). +function derivedReady(state: ModelState, policy: Policy): boolean { + if (policy.derived === "off") return false + if (state.stream !== "complete" || (policy.lockGate && state.lock)) return false + if (state.phase.includes("streaming")) return false + const tools = [0, 1].filter((index) => state.phase[index] !== "absent") + if (tools.length === 0) return false + return policy.derived === "every" + ? tools.every((index) => state.phase[index] === "resulted") + : state.phase[tools[0]!] === "resulted" +} + +function transitions(state: ModelState, policy: Policy): Step[] { + const result: Step[] = [] + if (state.request === "started" || state.request === "aborted") return result + + if (state.request === "released") { + const next = clone(state) + next.request = policy.abortCheck && next.abort ? "aborted" : "started" + return [{ action: "post-wait()", state: next }] + } + + for (const pass of currentPasses(state)) { + const index = state.index + if (pass.step === "block") { + const next = clone(state) + const current = next.passes.find((candidate) => candidate.id === pass.id)! + if (next.abort) { + current.step = "check" + } else { + next.pending = false + if (index >= blockCount(next)) { + if (next.stream === "complete") next.latch = true + current.step = "check" + } else if (next.phase[index] === "streaming") { + current.step = "check" + } else if (kinds[index] === "tool") { + next.phase[index] = "running" + current.step = "running" + } else { + next.phase[index] = "resulted" + next.index += 1 + afterBlock(next, current) + } + } + result.push({ action: `start-block(${pass.id})`, state: next }) + + // A non-abort presenter failure. By default it hits the trailing text block, as in the PR's + // real-presenter test. The toolThrow variant hits a tool block before it pushes a result. + const failsHere = kinds[index] === "text" || policy.toolThrow + if (!state.abort && index < blockCount(state) && failsHere && state.phase[index] === "final") { + const failed = clone(state) + const failedPass = failed.passes.find((candidate) => candidate.id === pass.id)! + failed.threw = true + if (policy.releaseOnThrow) release(failed, failedPass, policy) + else failedPass.step = "done" + result.push({ action: `throw(${pass.id})`, state: failed }) + } + } else if (pass.step === "running") { + // The handler pushes its result, then keeps running (for example a checkpoint save). + const next = clone(state) + const current = next.passes.find((candidate) => candidate.id === pass.id)! + next.phase[index] = "resulted" + current.step = "settling" + result.push({ action: `push(${pass.id})`, state: next }) + } else if (pass.step === "settling") { + const next = clone(state) + const current = next.passes.find((candidate) => candidate.id === pass.id)! + next.index += 1 + afterBlock(next, current) + result.push({ action: `finish(${pass.id})`, state: next }) + } else if (pass.step === "check") { + const next = clone(state) + const current = next.passes.find((candidate) => candidate.id === pass.id)! + const drains = next.pending && !next.abort + if (drains) current.step = "block" + else release(next, current, policy) + result.push({ action: `check(${pass.id})`, state: next, tag: drains ? "drain-loop" : undefined }) + } + } + + for (const pass of state.passes) { + if (pass.epoch < state.epoch && pass.step !== "done") { + const next = clone(state) + release(next, next.passes.find((candidate) => candidate.id === pass.id)!, policy) + result.push({ action: `stale-unwind(${pass.id})`, state: next }) + } + } + + // Block and check steps run synchronously after an await resolves, so timers and the retry catch + // cannot interleave with them. Only running and settling passes are suspended on an await. + const microtaskPending = currentPasses(state).some((pass) => pass.step === "block" || pass.step === "check") + if (state.stream === "open") { + const first = state.phase.indexOf("absent") + if (first !== -1) { + const next = clone(state) + next.phase[first] = "streaming" + result.push({ action: `begin(${first})`, state: next }) + } + for (const [index, phase] of state.phase.entries()) { + if (phase === "streaming" && kinds[index] === "tool") { + const next = clone(state) + next.phase[index] = "final" + result.push({ action: `finalize(${index})`, state: enter(next) }) + } + } + if (blockCount(state) > 0) { + const next = clone(state) + const hadPartial = next.phase.includes("streaming") + next.phase = next.phase.map((phase) => (phase === "streaming" ? "final" : phase)) + next.stream = "complete" + result.push({ action: "complete()", state: hadPartial ? enter(next) : next }) + } + if (policy.retry && state.epoch === 0 && !microtaskPending) { + const next = clone(state) + next.phase = ["absent", "absent", "absent"] + next.index = 0 + next.lock = false + next.owner = 0 + next.pending = false + next.latch = false + next.threw = false + next.epoch = 1 + result.push({ action: "retry-reset()", state: next }) + } + } + + if (!state.abort) { + const next = clone(state) + next.abort = true + result.push({ action: "abort()", state: next }) + } + + if (!microtaskPending && (state.latch || derivedReady(state, policy) || state.abort)) { + const next = clone(state) + next.request = "released" + result.push({ action: "poll()", state: next }) + } + return result +} + +function invariantViolations(state: ModelState): Array<{ invariant: string; message: string }> { + const violations: Array<{ invariant: string; message: string }> = [] + const livePasses = currentPasses(state) + if (state.request === "started") { + const unresolved = [0, 1].some((index) => state.phase[index] !== "absent" && state.phase[index] !== "resulted") + if (unresolved) { + violations.push({ + invariant: "premature-continue", + message: "next request started while a tool block has no result", + }) + } + if (livePasses.length > 0) { + violations.push({ + invariant: "continue-under-pass", + message: "next request started while a presenter pass was live", + }) + } + if (state.abort) { + violations.push({ invariant: "continue-after-abort", message: "next request started after abort" }) + } + } + if (livePasses.length === 0 && state.lock) { + violations.push({ invariant: "stranded-lock", message: "presenter lock held with no live pass" }) + } + if (!state.lock && state.pending && !state.abort && !state.threw) { + violations.push({ invariant: "stranded-update", message: "pending update left behind a released lock" }) + } + if (livePasses.length > 0 && !state.lock) { + violations.push({ invariant: "ownership", message: "a live pass runs while the lock is released" }) + } + return violations +} + +function canonical(state: ModelState): string { + return JSON.stringify(state) +} + +function formatTrace(trace: Step[]): string { + return trace.map((step, index) => `${index + 1}. ${step.action}`).join("\n") +} + +interface Exploration { + violation?: Violation + states: number + actions: Set + landmarks: Set + drainLoopSeen: boolean +} + +const landmarkPredicates: Record boolean> = { + "lost-latch-recovered-by-derived-path": (state) => state.request === "started" && !state.latch, + "latch-path-continues": (state) => state.request === "started" && state.latch, + "two-tool-turn-continues": (state) => + state.request === "started" && state.phase[0] === "resulted" && state.phase[1] === "resulted", + "presenter-throw-then-continue": (state) => state.request === "started" && state.threw, + "abort-stops-ready-turn": (state) => + state.request === "aborted" && state.stream === "complete" && state.phase[0] === "resulted", + "stale-pass-unwinds-after-retry": (state) => + state.passes.some((pass) => pass.epoch < state.epoch && pass.step === "done"), +} + +function explore(policy: Policy, detectStuck = true): Exploration { + const start = initialState() + const queue: Array<{ state: ModelState; trace: Step[] }> = [{ state: start, trace: [] }] + const visited = new Set([canonical(start)]) + const actions = new Set() + const landmarks = new Set() + const frontier: ModelState[] = [] + let drainLoopSeen = false + + for (let index = 0; index < queue.length; index++) { + const node = queue[index]! + for (const [name, predicate] of Object.entries(landmarkPredicates)) { + if (predicate(node.state)) landmarks.add(name) + } + const found = invariantViolations(node.state)[0] + if (found) { + return { + violation: { ...found, trace: node.trace }, + states: visited.size, + actions, + landmarks, + drainLoopSeen, + } + } + const next = transitions(node.state, policy) + // Abort is always available, so a state whose only enabled action is abort has no way to continue. + if (detectStuck && node.state.request === "waiting" && next.every((step) => step.action === "abort()")) { + return { + violation: { + invariant: "stuck", + message: "only abort is enabled and the next request has not started", + trace: node.trace, + }, + states: visited.size, + actions, + landmarks, + drainLoopSeen, + } + } + if (node.trace.length === MAX_DEPTH) { + frontier.push(node.state) + continue + } + for (const step of next) { + actions.add(step.action.slice(0, step.action.indexOf("("))) + if (step.tag === "drain-loop") drainLoopSeen = true + const key = canonical(step.state) + if (visited.has(key)) continue + visited.add(key) + queue.push({ state: step.state, trace: [...node.trace, step] }) + if (visited.size > MAX_STATES) { + throw new Error(`Tool-turn model exceeded its ${MAX_STATES}-state budget for ${policy.name}`) + } + } + } + const unseen = frontier.flatMap((state) => transitions(state, policy)).find((s) => !visited.has(canonical(s.state))) + if (unseen) throw new Error(`Tool-turn model truncated before unseen action ${unseen.action}`) + return { states: visited.size, actions, landmarks, drainLoopSeen } +} + +function requireClean(policy: Policy): Exploration { + const result = explore(policy) + if (result.violation) { + throw new Error( + `Tool-turn invariant failed under ${policy.name}: ${result.violation.invariant}: ${result.violation.message}\n${formatTrace(result.violation.trace)}`, + ) + } + return result +} + +const fixed = [requireClean(production), requireClean(ownerToken)] +const full = fixed[1]! + +const missingActions = expectedActions.filter((action) => !full.actions.has(action)) +assert.deepEqual(missingActions, [], `Tool-turn model has unreachable actions: ${missingActions.join(", ")}`) +const missingLandmarks = Object.keys(landmarkPredicates).filter((name) => !full.landmarks.has(name)) +assert.deepEqual(missingLandmarks, [], `Tool-turn model has unreachable landmarks: ${missingLandmarks.join(", ")}`) +assert.ok(full.drainLoopSeen, "Tool-turn model never reaches the pending-update drain loop") + +const witnesses: string[] = [] +for (const { policy, expected, detectStuck } of injected) { + const result = explore(policy, detectStuck) + assert.equal( + result.violation?.invariant, + expected, + `Injected policy ${policy.name} must violate ${expected}, got ${result.violation?.invariant ?? "no violation"}`, + ) + witnesses.push(`${policy.name}=${expected}(${result.violation!.trace.length} steps)`) +} + +// Known gap (#1884 review finding 3): the production `finally` clears the lock unconditionally. +// After a request-loop reset, a stale pass can release the lock of a newer pass. The owner-token +// policy above is the proposed fix. This witness keeps the gap visible until production changes. +const gap = explore(unconditionalWithRetry) +assert.ok(gap.violation, "Known stale-pass gap no longer reproduces: update the model and the document") + +// Known gap (#1884 review finding 2): a failure on a tool block before it pushes a result releases the +// lock, but no result exists, so the derived check stays false and the wait has no timeout. The hang +// existed before the PR (the lock stayed held). The PR deliberately does not synthesize a result. +const throwGap = explore(toolThrowGap) +assert.equal(throwGap.violation?.invariant, "stuck", "Known tool-throw hang no longer reproduces: update the model") + +console.log( + `Tool-turn continuation model check passed: ${fixed.map((r) => r.states).join("/")} reachable states (production/owner-token), ${full.actions.size}/${expectedActions.length} actions, ${full.landmarks.size}/${Object.keys(landmarkPredicates).length} landmarks, depth <= ${MAX_DEPTH}`, +) +console.log(`Known-unsafe witnesses: ${witnesses.join(", ")}`) +console.log(`Known gap (tool failure before result): stuck after ${throwGap.violation!.trace.length} steps`) +console.log( + `Known gap (stale unconditional release): ${gap.violation!.invariant} after ${gap.violation!.trace.length} steps\n${formatTrace(gap.violation!.trace)}`, +) diff --git a/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts b/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts index aa278e077d..e9ff1127ff 100644 --- a/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts +++ b/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts @@ -208,7 +208,186 @@ describe("presentAssistantMessage - Custom Tool Recording", () => { }) }) + describe("Presentation lock ownership", () => { + it("drains an update queued during the helper-to-wrapper handoff", async () => { + mockTask.assistantMessageContent = [{ type: "text", content: "initial", partial: true }] + let queuedCall: Promise | undefined + let lockedDuringHandoff = false + mockTask.say.mockImplementationOnce(async () => { + // The first microtask precedes the helper's await continuation. + // The second follows that continuation but precedes the wrapper's. + queueMicrotask(() => { + queueMicrotask(() => { + lockedDuringHandoff = mockTask.presentAssistantMessageLocked + mockTask.assistantMessageContent[0] = { type: "text", content: "final", partial: false } + mockTask.didCompleteReadingStream = true + queuedCall = presentAssistantMessage(mockTask) + }) + }) + }) + + await presentAssistantMessage(mockTask) + await queuedCall + + expect(lockedDuringHandoff).toBe(true) + expect(mockTask.say.mock.calls).toEqual([ + ["text", "initial", undefined, true], + ["text", "final", undefined, false], + ]) + expect(mockTask.currentStreamingContentIndex).toBe(1) + expect(mockTask.userMessageContentReady).toBe(true) + expect(mockTask.presentAssistantMessageHasPendingUpdates).toBe(false) + expect(mockTask.presentAssistantMessageLocked).toBe(false) + }) + + it("stops draining an update queued during the handoff once the task aborts", async () => { + mockTask.assistantMessageContent = [{ type: "text", content: "initial", partial: true }] + let queuedCall: Promise | undefined + let pendingDuringHandoff = false + let aborted = false + let abortChecksAfterAbort = 0 + Object.defineProperty(mockTask, "abort", { + configurable: true, + get: () => { + // Fail fast instead of hanging if a drain keeps retrying an + // aborted pass whose pending update can never be consumed. + if (aborted && ++abortChecksAfterAbort > 10) { + throw new Error("presenter kept draining after abort") + } + return aborted + }, + }) + mockTask.say.mockImplementationOnce(async () => { + queueMicrotask(() => { + queueMicrotask(() => { + mockTask.assistantMessageContent[0] = { type: "text", content: "final", partial: false } + mockTask.didCompleteReadingStream = true + queuedCall = presentAssistantMessage(mockTask) + pendingDuringHandoff = mockTask.presentAssistantMessageHasPendingUpdates + aborted = true + }) + }) + }) + + await presentAssistantMessage(mockTask) + await queuedCall + + expect(pendingDuringHandoff).toBe(true) + expect(mockTask.say).toHaveBeenCalledExactlyOnceWith("text", "initial", undefined, true) + expect(mockTask.currentStreamingContentIndex).toBe(0) + expect(mockTask.presentAssistantMessageLocked).toBe(false) + }) + + it("holds one lock across consecutive blocks and queues overlapping calls", async () => { + mockTask.assistantMessageContent = [ + { type: "text", content: "first", partial: false }, + { type: "text", content: "second", partial: false }, + ] + mockTask.didCompleteReadingStream = true + const lockChanges: boolean[] = [] + let locked = false + Object.defineProperty(mockTask, "presentAssistantMessageLocked", { + configurable: true, + get: () => locked, + set: (value: boolean) => { + locked = value + lockChanges.push(value) + }, + }) + mockTask.say.mockImplementation(async (_type: string, content: string) => { + expect(mockTask.presentAssistantMessageLocked).toBe(true) + if (content === "second") { + await presentAssistantMessage(mockTask) + expect(mockTask.presentAssistantMessageHasPendingUpdates).toBe(true) + } + }) + + await presentAssistantMessage(mockTask) + + expect(mockTask.say.mock.calls).toEqual([ + ["text", "first", undefined, false], + ["text", "second", undefined, false], + ]) + expect(mockTask.currentStreamingContentIndex).toBe(2) + expect(mockTask.userMessageContentReady).toBe(true) + expect(mockTask.presentAssistantMessageHasPendingUpdates).toBe(false) + expect(lockChanges).toEqual([true, false]) + }) + + it("consumes a pending update for a partial block without repeating it indefinitely", async () => { + mockTask.assistantMessageContent = [{ type: "text", content: "initial", partial: true }] + mockTask.say + .mockImplementationOnce(async () => { + mockTask.assistantMessageContent[0] = { type: "text", content: "updated", partial: true } + await presentAssistantMessage(mockTask) + expect(mockTask.presentAssistantMessageHasPendingUpdates).toBe(true) + }) + .mockImplementationOnce(async () => { + expect(mockTask.presentAssistantMessageLocked).toBe(true) + expect(mockTask.presentAssistantMessageHasPendingUpdates).toBe(false) + }) + + await presentAssistantMessage(mockTask) + + expect(mockTask.say.mock.calls).toEqual([ + ["text", "initial", undefined, true], + ["text", "updated", undefined, true], + ]) + expect(mockTask.currentStreamingContentIndex).toBe(0) + expect(mockTask.presentAssistantMessageHasPendingUpdates).toBe(false) + expect(mockTask.presentAssistantMessageLocked).toBe(false) + }) + + it("stops internal continuation when the task aborts between blocks", async () => { + mockTask.assistantMessageContent = [ + { type: "text", content: "first", partial: false }, + { type: "text", content: "second", partial: false }, + ] + mockTask.say.mockImplementationOnce(async () => { + mockTask.abort = true + }) + + await presentAssistantMessage(mockTask) + + expect(mockTask.say).toHaveBeenCalledExactlyOnceWith("text", "first", undefined, false) + expect(mockTask.currentStreamingContentIndex).toBe(1) + expect(mockTask.presentAssistantMessageLocked).toBe(false) + }) + + it("releases the presentation lock when a later block throws", async () => { + mockTask.assistantMessageContent = [ + { type: "text", content: "first", partial: false }, + { type: "text", content: "second", partial: false }, + ] + mockTask.say.mockResolvedValueOnce(undefined).mockRejectedValueOnce(new Error("second block failed")) + + await expect(presentAssistantMessage(mockTask)).rejects.toThrow("second block failed") + + expect(mockTask.say).toHaveBeenCalledTimes(2) + expect(mockTask.currentStreamingContentIndex).toBe(1) + expect(mockTask.presentAssistantMessageLocked).toBe(false) + }) + }) + describe("Custom tool error recording", () => { + it("releases the presentation lock when dispatch throws", async () => { + mockTask.assistantMessageContent = [ + { + type: "tool_use", + id: "tool_call_dispatch_failure", + name: "read_file", + params: {}, + partial: false, + }, + ] + mockTask.providerRef.deref = () => ({ + getState: vi.fn().mockRejectedValue(new Error("provider state failed")), + }) + + await expect(presentAssistantMessage(mockTask)).rejects.toThrow("provider state failed") + expect(mockTask.presentAssistantMessageLocked).toBe(false) + }) + it("should record custom tool error as 'custom_tool'", async () => { const toolCallId = "tool_call_custom_error_123" mockTask.assistantMessageContent = [ diff --git a/src/core/assistant-message/presentAssistantMessage.ts b/src/core/assistant-message/presentAssistantMessage.ts index b11538a2c6..417dcf7a45 100644 --- a/src/core/assistant-message/presentAssistantMessage.ts +++ b/src/core/assistant-message/presentAssistantMessage.ts @@ -97,6 +97,28 @@ export async function presentAssistantMessage(cline: Task) { } cline.presentAssistantMessageLocked = true + try { + // Drain updates queued while the lock was held, including one that + // arrives after the helper's final check but before this continuation. + // The last pending check and the release below run in the same + // synchronous step, so no update can be stranded behind the lock. + do { + await presentAssistantMessageBlock(cline) + } while (!cline.abort && cline.presentAssistantMessageHasPendingUpdates) + } finally { + // Tool handlers and provider-state reads can reject. Never strand the + // task behind a dispatch lock after the presenter has unwound. + cline.presentAssistantMessageLocked = false + } +} + +async function presentAssistantMessageBlock(cline: Task): Promise { + if (cline.abort) { + return + } + + // Each internal pass consumes the pending update. New updates arriving + // during an awaited operation can request another pass. cline.presentAssistantMessageHasPendingUpdates = false if (cline.currentStreamingContentIndex >= cline.assistantMessageContent.length) { @@ -108,7 +130,6 @@ export async function presentAssistantMessage(cline: Task) { cline.userMessageContentReady = true } - cline.presentAssistantMessageLocked = false return } @@ -125,7 +146,6 @@ export async function presentAssistantMessage(cline: Task) { `Block content:`, JSON.stringify(cline.assistantMessageContent[cline.currentStreamingContentIndex], null, 2), ) - cline.presentAssistantMessageLocked = false return } @@ -1077,17 +1097,6 @@ export async function presentAssistantMessage(cline: Task) { } } - // Seeing out of bounds is fine, it means that the next too call is being - // built up and ready to add to assistantMessageContent to present. - // When you see the UI inactive during this, it means that a tool is - // breaking without presenting any UI. For example the write_to_file tool - // was breaking when relpath was undefined, and for invalid relpath it never - // presented UI. - // This needs to be placed here, if not then calling - // cline.presentAssistantMessage below would fail (sometimes) since it's - // locked. - cline.presentAssistantMessageLocked = false - // NOTE: When tool is rejected, iterator stream is interrupted and it waits // for `userMessageContentReady` to be true. Future calls to present will // skip execution since `didRejectTool` and iterate until `contentIndex` is @@ -1115,7 +1124,7 @@ export async function presentAssistantMessage(cline: Task) { if (cline.currentStreamingContentIndex < cline.assistantMessageContent.length) { // There are already more content blocks to stream, so we'll call // this function ourselves. - return presentAssistantMessage(cline) + return await presentAssistantMessageBlock(cline) } else { // CRITICAL FIX: If we're out of bounds and the stream is complete, set userMessageContentReady // This handles the case where assistantMessageContent is empty or becomes empty after processing @@ -1125,10 +1134,8 @@ export async function presentAssistantMessage(cline: Task) { } } - // Block is partial, but the read stream may have finished. - if (cline.presentAssistantMessageHasPendingUpdates) { - return presentAssistantMessage(cline) - } + // Pending updates are drained by presentAssistantMessage, which owns the + // lock and checks for them immediately before releasing it. } /** diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 4de2b84590..d8f84e3608 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -562,6 +562,48 @@ export class Task extends EventEmitter implements TaskLike { this.userMessageContent.push(toolResult) return true } + + /** + * Derives terminal tool-turn readiness from the protocol state instead of + * relying exclusively on the presenter's one-shot boolean latch. + * + * A re-entrant presenter can lose the latch update after every tool has + * already completed. At that point the assistant turn is safe to continue + * when the stream is closed, presentation is idle, every content block is + * final, and every tool call has its matching result. + */ + private hasCompleteToolResultsForCurrentTurn(): boolean { + // Do not require currentStreamingContentIndex to reach the end here. + // A non-abort presenter rejection releases the lock and is logged, but + // nothing schedules another presentation pass. Gating on the index would + // strand this wait after exactly that failure. + if (!this.didCompleteReadingStream || this.presentAssistantMessageLocked) { + return false + } + + const toolResultIds = new Set( + this.userMessageContent + .filter((block): block is Anthropic.ToolResultBlockParam => block.type === "tool_result") + .map((block) => block.tool_use_id), + ) + let toolUseCount = 0 + + for (const block of this.assistantMessageContent) { + if (block.partial) { + return false + } + if (block.type !== "tool_use" && block.type !== "mcp_tool_use") { + continue + } + + toolUseCount++ + if (!block.id || !toolResultIds.has(sanitizeToolUseId(block.id))) { + return false + } + } + + return toolUseCount > 0 + } didRejectTool = false didAlreadyUseTool = false didToolFailInCurrentTurn = false @@ -4652,7 +4694,13 @@ export class Task extends EventEmitter implements TaskLike { // this.userMessageContentReady = true // } - await pWaitFor(() => this.userMessageContentReady || this.abort || this.abandoned) + await pWaitFor( + () => + this.userMessageContentReady || + this.hasCompleteToolResultsForCurrentTurn() || + this.abort || + this.abandoned, + ) if (this.abort || this.abandoned) { throw new Error( diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index 06f07d2e02..aa3e75d640 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -33,6 +33,7 @@ import type { ApiMessage } from "../../task-persistence" import { asyncStreamFrom } from "../../../test-utils/stream" import { McpHub } from "../../../services/mcp/McpHub" import { McpServerManager } from "../../../services/mcp/McpServerManager" +import { readFileTool } from "../../tools/ReadFileTool" import { writeToFileTool } from "../../tools/WriteToFileTool" type TaskTestAccess = { @@ -43,6 +44,7 @@ type TaskTestAccess = { startTask: (task?: string, images?: string[]) => Promise resumeTaskFromHistory: () => Promise presentAssistantMessageSafe: () => void + hasCompleteToolResultsForCurrentTurn: () => boolean addToClineMessages: (message: import("@roo-code/types").ClineMessage) => Promise updateClineMessage: (message: import("@roo-code/types").ClineMessage) => Promise saveClineMessages: () => Promise @@ -610,6 +612,422 @@ describe("Cline", () => { }) describe("native tool-call request isolation", () => { + it("derives readiness only from a closed, idle, fully paired tool turn", () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "tool turn readiness test", + startTask: false, + }) + const readiness = () => getTaskTestAccess(task).hasCompleteToolResultsForCurrentTurn() + task.assistantMessageContent = [ + { + type: "tool_use", + id: "call_ready", + name: "read_file", + params: {}, + nativeArgs: { path: "README.md" }, + partial: false, + }, + ] + task.userMessageContent = [{ type: "tool_result", tool_use_id: "call_ready", content: "finished" }] + task.currentStreamingContentIndex = task.assistantMessageContent.length + + expect(readiness()).toBe(false) + task.didCompleteReadingStream = true + expect(readiness()).toBe(true) + task.presentAssistantMessageLocked = true + expect(readiness()).toBe(false) + task.presentAssistantMessageLocked = false + expect(readiness()).toBe(true) + task.assistantMessageContent[0].partial = true + expect(readiness()).toBe(false) + task.assistantMessageContent[0].partial = false + task.userMessageContent = [{ type: "tool_result", tool_use_id: "different_call", content: "finished" }] + expect(readiness()).toBe(false) + task.userMessageContent = [{ type: "tool_result", tool_use_id: "call_ready", content: "finished" }] + expect(readiness()).toBe(true) + }) + + it("requires a result for every tool call in the turn", () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "multi tool readiness test", + startTask: false, + }) + const readiness = () => getTaskTestAccess(task).hasCompleteToolResultsForCurrentTurn() + task.didCompleteReadingStream = true + task.assistantMessageContent = [ + { type: "tool_use", id: "call_first", name: "read_file", params: {}, partial: false }, + { type: "tool_use", id: "call_middle", name: "read_file", params: {}, partial: false }, + { type: "tool_use", id: "call_last", name: "read_file", params: {}, partial: false }, + ] + const resultsFor = (...ids: string[]): Anthropic.ToolResultBlockParam[] => + ids.map((id) => ({ type: "tool_result", tool_use_id: id, content: "finished" })) + + task.userMessageContent = resultsFor("call_first") + expect(readiness()).toBe(false) + task.userMessageContent = resultsFor("call_last") + expect(readiness()).toBe(false) + task.userMessageContent = resultsFor("call_first", "call_last") + expect(readiness()).toBe(false) + task.userMessageContent = resultsFor("call_first", "call_middle", "call_last") + expect(readiness()).toBe(true) + }) + + it("does not strand a paired tool turn after a presenter failure", () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "presenter failure readiness test", + startTask: false, + }) + task.didCompleteReadingStream = true + task.presentAssistantMessageLocked = false + task.assistantMessageContent = [ + { + type: "tool_use", + id: "call_ready", + name: "read_file", + params: {}, + partial: false, + }, + { type: "text", content: "presenter failed here", partial: false }, + ] + task.currentStreamingContentIndex = 1 + task.userMessageContent = [{ type: "tool_result", tool_use_id: "call_ready", content: "finished" }] + + expect(getTaskTestAccess(task).hasCompleteToolResultsForCurrentTurn()).toBe(true) + }) + + it("requires an identifiable tool call and matches sanitized result IDs", () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "tool turn identity test", + startTask: false, + }) + const readiness = () => getTaskTestAccess(task).hasCompleteToolResultsForCurrentTurn() + task.didCompleteReadingStream = true + + task.assistantMessageContent = [{ type: "text", content: "finished", partial: false }] + expect(readiness()).toBe(false) + + // An empty result ID must not satisfy a tool call that has no ID. + task.userMessageContent = [{ type: "tool_result", tool_use_id: "", content: "finished" }] + task.assistantMessageContent = [ + { + type: "tool_use", + name: "read_file", + params: {}, + partial: false, + }, + ] + expect(readiness()).toBe(false) + + task.assistantMessageContent = [ + { + type: "mcp_tool_use", + id: "", + name: "mcp_server_tool", + serverName: "server", + toolName: "tool", + arguments: {}, + partial: false, + }, + ] + expect(readiness()).toBe(false) + + const mcpToolUse = task.assistantMessageContent[0] + if (mcpToolUse.type !== "mcp_tool_use") { + throw new Error("Expected MCP tool use fixture") + } + mcpToolUse.id = "functions.read_file:0" + task.userMessageContent = [ + { type: "tool_result", tool_use_id: "functions.read_file:0", content: "finished" }, + ] + expect(readiness()).toBe(false) + + task.userMessageContent = [ + { type: "tool_result", tool_use_id: "functions_read_file_0", content: "finished" }, + ] + expect(readiness()).toBe(true) + }) + + it("continues after a complete read_file result when the readiness flag update is lost", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "read file continuation test", + startTask: false, + }) + + vi.spyOn(task.diffViewProvider, "reset").mockResolvedValue(undefined) + vi.spyOn(getTaskTestAccess(task), "safeEnsureModelFetched").mockResolvedValue(stubModelInfo) + vi.spyOn(getTaskTestAccess(task), "presentAssistantMessageSafe").mockImplementation(() => { + const completedRead = task.assistantMessageContent.find( + (block) => block.type === "tool_use" && block.name === "read_file" && !block.partial, + ) + if (!completedRead || completedRead.type !== "tool_use" || !completedRead.id) return + + task.pushToolResultToUserContent({ + type: "tool_result", + tool_use_id: completedRead.id, + content: "File: README.md\nfinished", + }) + // Reproduce the persisted field symptom: the tool result exists and + // presentation consumed the completed block, but the one-shot readiness + // latch never flips, so the old loop waits forever before the next request. + task.currentStreamingContentIndex = task.assistantMessageContent.length + task.userMessageContentReady = false + }) + + let continuationUserContent: Anthropic.Messages.ContentBlockParam[] | undefined + const attemptApiRequestSpy = vi + .spyOn(task, "attemptApiRequest") + .mockImplementationOnce(() => + asyncStreamFrom([ + { type: "tool_call_partial", index: 0, id: "call_read", name: "read_file" }, + { type: "tool_call_partial", index: 0, arguments: '{"path":"README.md"}' }, + ]), + ) + .mockImplementationOnce(() => { + const continuationMessage = task.apiConversationHistory.at(-1) + continuationUserContent = + continuationMessage?.role === "user" && Array.isArray(continuationMessage.content) + ? continuationMessage.content + : undefined + throw new Error("continuation request reached") + }) + + vi.mocked(pWaitFor).mockImplementation(async (condition) => { + if (!(await condition())) { + throw new Error("read_file continuation would stall") + } + }) + + try { + await task.recursivelyMakeClineRequests([{ type: "text", text: "read a file, then continue" }]) + expect(attemptApiRequestSpy).toHaveBeenCalledTimes(2) + expect(continuationUserContent?.filter((block) => block.type === "tool_result")).toEqual([ + expect.objectContaining({ + type: "tool_result", + tool_use_id: "call_read", + content: "File: README.md\nfinished", + }), + ]) + } finally { + vi.mocked(pWaitFor).mockImplementation(async () => {}) + } + }) + + it("continues after the real presenter leaves a complete tool result behind a non-abort failure", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "real presenter continuation test", + startTask: false, + }) + + vi.spyOn(task.diffViewProvider, "reset").mockResolvedValue(undefined) + vi.spyOn(getTaskTestAccess(task), "safeEnsureModelFetched").mockResolvedValue(stubModelInfo) + const readFileHandleSpy = vi + .spyOn(readFileTool, "handle") + .mockImplementation(async (_task, block, callbacks) => { + if (!block.partial) callbacks.pushToolResult(`File: ${block.params.path}\nfinished`) + }) + const say = task.say.bind(task) + const saySpy = vi.spyOn(task, "say").mockImplementation(async (...args) => { + const [type, _text, _images, partial] = args + if (type === "text" && partial === false) { + throw new Error("presenter text failed") + } + return say(...args) + }) + const readiness = getTaskTestAccess(task).hasCompleteToolResultsForCurrentTurn.bind(task) + let sawCompleteToolTurn = false + const readinessSpy = vi + .spyOn(getTaskTestAccess(task), "hasCompleteToolResultsForCurrentTurn") + .mockImplementation(() => { + const ready = readiness() + if (ready) sawCompleteToolTurn = true + return ready + }) + + let continuationUserContent: Anthropic.Messages.ContentBlockParam[] | undefined + const attemptApiRequestSpy = vi + .spyOn(task, "attemptApiRequest") + .mockImplementationOnce(() => + asyncStreamFrom([ + { type: "tool_call_partial", index: 0, id: "call_read_first", name: "read_file" }, + { type: "tool_call_partial", index: 0, arguments: '{"path":"README.md"}' }, + { type: "tool_call_partial", index: 1, id: "call_read_second", name: "read_file" }, + { type: "tool_call_partial", index: 1, arguments: '{"path":"package.json"}' }, + { type: "text", text: "present this after the tool" }, + ]), + ) + .mockImplementationOnce(() => { + const continuationMessage = task.apiConversationHistory.at(-1) + continuationUserContent = + continuationMessage?.role === "user" && Array.isArray(continuationMessage.content) + ? continuationMessage.content + : undefined + throw new Error("continuation request reached") + }) + + const { default: realPWaitFor } = await vi.importActual("p-wait-for") + vi.mocked(pWaitFor).mockImplementation((condition) => + realPWaitFor(condition, { interval: 1, timeout: 1_000 }), + ) + try { + await task.recursivelyMakeClineRequests([{ type: "text", text: "read a file, then continue" }]) + + expect(attemptApiRequestSpy).toHaveBeenCalledTimes(2) + expect(continuationUserContent?.filter((block) => block.type === "tool_result")).toEqual([ + expect.objectContaining({ + type: "tool_result", + tool_use_id: "call_read_first", + content: "File: README.md\nfinished", + }), + expect.objectContaining({ + type: "tool_result", + tool_use_id: "call_read_second", + content: "File: package.json\nfinished", + }), + ]) + expect(task.presentAssistantMessageLocked).toBe(false) + expect(sawCompleteToolTurn).toBe(true) + } finally { + vi.mocked(pWaitFor).mockImplementation(async () => {}) + readinessSpy.mockRestore() + readFileHandleSpy.mockRestore() + saySpy.mockRestore() + } + }) + + it("waits for a running tool handler even when its result is already pushed", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "running handler continuation test", + startTask: false, + }) + + vi.spyOn(task.diffViewProvider, "reset").mockResolvedValue(undefined) + vi.spyOn(getTaskTestAccess(task), "safeEnsureModelFetched").mockResolvedValue(stubModelInfo) + let releaseHandler: (() => void) | undefined + const handlerGate = new Promise((resolve) => { + releaseHandler = resolve + }) + let handlerPushedResult: (() => void) | undefined + const resultPushed = new Promise((resolve) => { + handlerPushedResult = resolve + }) + const readFileHandleSpy = vi + .spyOn(readFileTool, "handle") + .mockImplementation(async (_task, block, callbacks) => { + if (block.partial) return + callbacks.pushToolResult("File: README.md\nfinished") + handlerPushedResult?.() + // The handler keeps running after it pushes the result, for example while it waits for approval. + await handlerGate + }) + const attemptApiRequestSpy = vi + .spyOn(task, "attemptApiRequest") + .mockImplementationOnce(() => + asyncStreamFrom([ + { type: "tool_call_partial", index: 0, id: "call_read", name: "read_file" }, + { type: "tool_call_partial", index: 0, arguments: '{"path":"README.md"}' }, + ]), + ) + .mockImplementationOnce(() => { + throw new Error("continuation request reached") + }) + + const { default: realPWaitFor } = await vi.importActual("p-wait-for") + vi.mocked(pWaitFor).mockImplementation((condition) => + realPWaitFor(condition, { interval: 1, timeout: 1_000 }), + ) + try { + const run = task.recursivelyMakeClineRequests([{ type: "text", text: "read a file, then wait" }]) + await resultPushed + // Give the request loop several poll intervals to continue early. + await new Promise((resolve) => setTimeout(resolve, 50)) + expect(task.presentAssistantMessageLocked).toBe(true) + expect(attemptApiRequestSpy).toHaveBeenCalledTimes(1) + + releaseHandler?.() + await run + expect(attemptApiRequestSpy).toHaveBeenCalledTimes(2) + expect(task.presentAssistantMessageLocked).toBe(false) + } finally { + releaseHandler?.() + vi.mocked(pWaitFor).mockImplementation(async () => {}) + readFileHandleSpy.mockRestore() + } + }) + + it("lets abort win when the real presenter leaves a complete tool result", async () => { + const task = new Task({ + provider: mockProvider, + apiConfiguration: mockApiConfig, + task: "real presenter abort test", + startTask: false, + }) + + vi.spyOn(task.diffViewProvider, "reset").mockResolvedValue(undefined) + vi.spyOn(getTaskTestAccess(task), "safeEnsureModelFetched").mockResolvedValue(stubModelInfo) + const readFileHandleSpy = vi + .spyOn(readFileTool, "handle") + .mockImplementation(async (_task, block, callbacks) => { + if (!block.partial) callbacks.pushToolResult("File: README.md\nfinished") + }) + const say = task.say.bind(task) + const saySpy = vi.spyOn(task, "say").mockImplementation(async (...args) => { + const [type, _text, _images, partial] = args + if (type === "text" && partial === false) { + task.abort = true + throw new Error("presenter text failed while aborting") + } + return say(...args) + }) + + const readiness = getTaskTestAccess(task).hasCompleteToolResultsForCurrentTurn.bind(task) + let sawReadyWhileAborted = false + const readinessSpy = vi + .spyOn(getTaskTestAccess(task), "hasCompleteToolResultsForCurrentTurn") + .mockImplementation(() => { + const ready = readiness() + if (ready && task.abort) sawReadyWhileAborted = true + return ready + }) + const attemptApiRequestSpy = vi.spyOn(task, "attemptApiRequest").mockImplementation(() => + asyncStreamFrom([ + { type: "tool_call_partial", index: 0, id: "call_read", name: "read_file" }, + { type: "tool_call_partial", index: 0, arguments: '{"path":"README.md"}' }, + { type: "text", text: "present this after the tool" }, + ]), + ) + + const { default: realPWaitFor } = await vi.importActual("p-wait-for") + vi.mocked(pWaitFor).mockImplementation((condition) => + realPWaitFor(condition, { interval: 1, timeout: 1_000 }), + ) + try { + await task.recursivelyMakeClineRequests([{ type: "text", text: "read a file, then abort" }]) + + expect(attemptApiRequestSpy).toHaveBeenCalledOnce() + expect(sawReadyWhileAborted).toBe(true) + expect(task.presentAssistantMessageLocked).toBe(false) + } finally { + vi.mocked(pWaitFor).mockImplementation(async () => {}) + readinessSpy.mockRestore() + readFileHandleSpy.mockRestore() + saySpy.mockRestore() + } + }) + it("keeps overlapping Task parser state scoped to each request", async () => { const firstTask = new Task({ provider: mockProvider,