From 7a2715ef407ce82ce73780924ec484cceaa1ca85 Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Thu, 1 Oct 2026 15:35:17 -0500 Subject: [PATCH 01/10] fix(task): recover completed tool turn continuation --- ...resentAssistantMessage-custom-tool.spec.ts | 18 ++++ .../presentAssistantMessage.ts | 13 ++- src/core/task/Task.ts | 46 +++++++++- src/core/task/__tests__/Task.spec.ts | 88 +++++++++++++++++++ 4 files changed, 162 insertions(+), 3 deletions(-) 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..fbe2ac2bd6 100644 --- a/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts +++ b/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts @@ -209,6 +209,24 @@ describe("presentAssistantMessage - Custom Tool Recording", () => { }) 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..3ca51677ae 100644 --- a/src/core/assistant-message/presentAssistantMessage.ts +++ b/src/core/assistant-message/presentAssistantMessage.ts @@ -98,7 +98,16 @@ export async function presentAssistantMessage(cline: Task) { cline.presentAssistantMessageLocked = true cline.presentAssistantMessageHasPendingUpdates = false + try { + await presentAssistantMessageBlock(cline) + } 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.currentStreamingContentIndex >= cline.assistantMessageContent.length) { // This may happen if the last content block was completed before // streaming could finish. If streaming is finished, and we're out of @@ -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 presentAssistantMessage(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 @@ -1127,7 +1136,7 @@ export async function presentAssistantMessage(cline: Task) { // Block is partial, but the read stream may have finished. if (cline.presentAssistantMessageHasPendingUpdates) { - return presentAssistantMessage(cline) + return await presentAssistantMessage(cline) } } diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 4de2b84590..fb9da7fe8c 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -562,6 +562,44 @@ 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 { + 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 +4690,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..004ec361fe 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -43,6 +43,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 +611,93 @@ 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" }] + + expect(readiness()).toBe(false) + task.didCompleteReadingStream = true + task.presentAssistantMessageLocked = true + expect(readiness()).toBe(false) + task.presentAssistantMessageLocked = false + 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("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 is no longer running, but the one-shot readiness latch + // never flips, so the old loop waits forever before the next request. + task.userMessageContentReady = false + }) + + 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") + }) + + 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) + } finally { + vi.mocked(pWaitFor).mockImplementation(async () => {}) + } + }) + it("keeps overlapping Task parser state scoped to each request", async () => { const firstTask = new Task({ provider: mockProvider, From 7ab55fa0d6b1b3967873fe53c1f488c6fa127bc8 Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Thu, 1 Oct 2026 20:50:30 -0500 Subject: [PATCH 02/10] fix(task): require completed turn presentation --- src/core/task/Task.ts | 6 ++- src/core/task/__tests__/Task.spec.ts | 80 +++++++++++++++++++++++++++- 2 files changed, 83 insertions(+), 3 deletions(-) diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index fb9da7fe8c..431267892c 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -573,7 +573,11 @@ export class Task extends EventEmitter implements TaskLike { * final, and every tool call has its matching result. */ private hasCompleteToolResultsForCurrentTurn(): boolean { - if (!this.didCompleteReadingStream || this.presentAssistantMessageLocked) { + if ( + !this.didCompleteReadingStream || + this.presentAssistantMessageLocked || + this.currentStreamingContentIndex < this.assistantMessageContent.length + ) { return false } diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index 004ec361fe..b987ad11f7 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -636,6 +636,8 @@ describe("Cline", () => { task.presentAssistantMessageLocked = true expect(readiness()).toBe(false) task.presentAssistantMessageLocked = false + expect(readiness()).toBe(false) + task.currentStreamingContentIndex = task.assistantMessageContent.length task.assistantMessageContent[0].partial = true expect(readiness()).toBe(false) task.assistantMessageContent[0].partial = false @@ -645,6 +647,64 @@ describe("Cline", () => { expect(readiness()).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() + const finishPresentation = () => { + task.currentStreamingContentIndex = task.assistantMessageContent.length + } + task.didCompleteReadingStream = true + + task.assistantMessageContent = [{ type: "text", content: "finished", partial: false }] + finishPresentation() + expect(readiness()).toBe(false) + + task.assistantMessageContent = [ + { + type: "tool_use", + name: "read_file", + params: {}, + partial: false, + }, + ] + finishPresentation() + expect(readiness()).toBe(false) + + task.assistantMessageContent = [ + { + type: "mcp_tool_use", + id: "", + name: "mcp_server_tool", + serverName: "server", + toolName: "tool", + arguments: {}, + partial: false, + }, + ] + finishPresentation() + 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, @@ -667,11 +727,13 @@ describe("Cline", () => { content: "File: README.md\nfinished", }) // Reproduce the persisted field symptom: the tool result exists and - // presentation is no longer running, but the one-shot readiness latch - // never flips, so the old loop waits forever before the next request. + // 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(() => @@ -681,6 +743,11 @@ describe("Cline", () => { ]), ) .mockImplementationOnce(() => { + const continuationMessage = task.apiConversationHistory.at(-1) + continuationUserContent = + continuationMessage?.role === "user" && Array.isArray(continuationMessage.content) + ? continuationMessage.content + : undefined throw new Error("continuation request reached") }) @@ -693,6 +760,15 @@ describe("Cline", () => { try { await task.recursivelyMakeClineRequests([{ type: "text", text: "read a file, then continue" }]) expect(attemptApiRequestSpy).toHaveBeenCalledTimes(2) + expect(continuationUserContent).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "tool_result", + tool_use_id: "call_read", + content: "File: README.md\nfinished", + }), + ]), + ) } finally { vi.mocked(pWaitFor).mockImplementation(async () => {}) } From 9599e0821f542e898735d4e5ebe63f8eae002107 Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Thu, 1 Oct 2026 21:10:21 -0500 Subject: [PATCH 03/10] fix(task): avoid stranding presenter failures --- src/core/task/Task.ts | 10 ++++----- src/core/task/__tests__/Task.spec.ts | 33 +++++++++++++++++++++------- 2 files changed, 30 insertions(+), 13 deletions(-) diff --git a/src/core/task/Task.ts b/src/core/task/Task.ts index 431267892c..d8f84e3608 100644 --- a/src/core/task/Task.ts +++ b/src/core/task/Task.ts @@ -573,11 +573,11 @@ export class Task extends EventEmitter implements TaskLike { * final, and every tool call has its matching result. */ private hasCompleteToolResultsForCurrentTurn(): boolean { - if ( - !this.didCompleteReadingStream || - this.presentAssistantMessageLocked || - this.currentStreamingContentIndex < this.assistantMessageContent.length - ) { + // 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 } diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index b987ad11f7..1b2ac383f6 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -636,8 +636,6 @@ describe("Cline", () => { task.presentAssistantMessageLocked = true expect(readiness()).toBe(false) task.presentAssistantMessageLocked = false - expect(readiness()).toBe(false) - task.currentStreamingContentIndex = task.assistantMessageContent.length task.assistantMessageContent[0].partial = true expect(readiness()).toBe(false) task.assistantMessageContent[0].partial = false @@ -647,6 +645,31 @@ describe("Cline", () => { 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, @@ -655,13 +678,9 @@ describe("Cline", () => { startTask: false, }) const readiness = () => getTaskTestAccess(task).hasCompleteToolResultsForCurrentTurn() - const finishPresentation = () => { - task.currentStreamingContentIndex = task.assistantMessageContent.length - } task.didCompleteReadingStream = true task.assistantMessageContent = [{ type: "text", content: "finished", partial: false }] - finishPresentation() expect(readiness()).toBe(false) task.assistantMessageContent = [ @@ -672,7 +691,6 @@ describe("Cline", () => { partial: false, }, ] - finishPresentation() expect(readiness()).toBe(false) task.assistantMessageContent = [ @@ -686,7 +704,6 @@ describe("Cline", () => { partial: false, }, ] - finishPresentation() expect(readiness()).toBe(false) const mcpToolUse = task.assistantMessageContent[0] From fbcb70ba772b6270c02d0c1d040ed1cc862f3d02 Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Thu, 1 Oct 2026 21:54:54 -0500 Subject: [PATCH 04/10] test(task): isolate stream completion readiness --- src/core/task/__tests__/Task.spec.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index 1b2ac383f6..92feb001d9 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -630,12 +630,15 @@ describe("Cline", () => { }, ] 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 From 7e699b593ab64d92e94884a560551409daa02366 Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Sun, 4 Oct 2026 16:25:25 -0500 Subject: [PATCH 05/10] test(task): cover real presenter continuation --- src/core/task/__tests__/Task.spec.ts | 141 +++++++++++++++++++++++++++ 1 file changed, 141 insertions(+) diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index 92feb001d9..cc782a806f 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 = { @@ -794,6 +795,146 @@ describe("Cline", () => { } }) + 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: 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) { + 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", name: "read_file" }, + { type: "tool_call_partial", index: 0, arguments: '{"path":"README.md"}' }, + { 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).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + type: "tool_result", + tool_use_id: "call_read", + content: "File: README.md\nfinished", + }), + ]), + ) + expect(task.userMessageContentReady).toBe(false) + expect(task.presentAssistantMessageLocked).toBe(false) + expect(sawCompleteToolTurn).toBe(true) + } finally { + vi.mocked(pWaitFor).mockImplementation(async () => {}) + readinessSpy.mockRestore() + readFileHandleSpy.mockRestore() + saySpy.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, From e4d65d7f457340f0eb5a3ddf8b9f898f25a8a636 Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Sun, 4 Oct 2026 17:21:29 -0500 Subject: [PATCH 06/10] test(task): cover multi-tool continuation --- src/core/task/__tests__/Task.spec.ts | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index cc782a806f..fa3aa09fcb 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -808,7 +808,7 @@ describe("Cline", () => { const readFileHandleSpy = vi .spyOn(readFileTool, "handle") .mockImplementation(async (_task, block, callbacks) => { - if (!block.partial) callbacks.pushToolResult("File: README.md\nfinished") + 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) => { @@ -833,8 +833,10 @@ describe("Cline", () => { .spyOn(task, "attemptApiRequest") .mockImplementationOnce(() => asyncStreamFrom([ - { type: "tool_call_partial", index: 0, id: "call_read", name: "read_file" }, + { 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" }, ]), ) @@ -855,15 +857,18 @@ describe("Cline", () => { await task.recursivelyMakeClineRequests([{ type: "text", text: "read a file, then continue" }]) expect(attemptApiRequestSpy).toHaveBeenCalledTimes(2) - expect(continuationUserContent).toEqual( - expect.arrayContaining([ - expect.objectContaining({ - type: "tool_result", - tool_use_id: "call_read", - content: "File: README.md\nfinished", - }), - ]), - ) + 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.userMessageContentReady).toBe(false) expect(task.presentAssistantMessageLocked).toBe(false) expect(sawCompleteToolTurn).toBe(true) From 2453214073309c4a1aa8c5cdc63ecfdcd60f226e Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Tue, 6 Oct 2026 02:08:34 -0500 Subject: [PATCH 07/10] fix(task): keep presenter lock ownership in wrapper --- ...resentAssistantMessage-custom-tool.spec.ts | 92 +++++++++++++++++++ .../presentAssistantMessage.ts | 26 ++---- 2 files changed, 102 insertions(+), 16 deletions(-) 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 fbe2ac2bd6..e37bf2cd48 100644 --- a/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts +++ b/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts @@ -208,6 +208,98 @@ describe("presentAssistantMessage - Custom Tool Recording", () => { }) }) + describe("Presentation lock ownership", () => { + 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 = [ diff --git a/src/core/assistant-message/presentAssistantMessage.ts b/src/core/assistant-message/presentAssistantMessage.ts index 3ca51677ae..dc0a8e18f8 100644 --- a/src/core/assistant-message/presentAssistantMessage.ts +++ b/src/core/assistant-message/presentAssistantMessage.ts @@ -97,7 +97,6 @@ export async function presentAssistantMessage(cline: Task) { } cline.presentAssistantMessageLocked = true - cline.presentAssistantMessageHasPendingUpdates = false try { await presentAssistantMessageBlock(cline) } finally { @@ -108,6 +107,14 @@ export async function presentAssistantMessage(cline: Task) { } 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) { // This may happen if the last content block was completed before // streaming could finish. If streaming is finished, and we're out of @@ -117,7 +124,6 @@ async function presentAssistantMessageBlock(cline: Task): Promise { cline.userMessageContentReady = true } - cline.presentAssistantMessageLocked = false return } @@ -134,7 +140,6 @@ async function presentAssistantMessageBlock(cline: Task): Promise { `Block content:`, JSON.stringify(cline.assistantMessageContent[cline.currentStreamingContentIndex], null, 2), ) - cline.presentAssistantMessageLocked = false return } @@ -1086,17 +1091,6 @@ async function presentAssistantMessageBlock(cline: Task): Promise { } } - // 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 @@ -1124,7 +1118,7 @@ async function presentAssistantMessageBlock(cline: Task): Promise { if (cline.currentStreamingContentIndex < cline.assistantMessageContent.length) { // There are already more content blocks to stream, so we'll call // this function ourselves. - return await 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 @@ -1136,7 +1130,7 @@ async function presentAssistantMessageBlock(cline: Task): Promise { // Block is partial, but the read stream may have finished. if (cline.presentAssistantMessageHasPendingUpdates) { - return await presentAssistantMessage(cline) + return await presentAssistantMessageBlock(cline) } } From 2e90d6c32f576eabc340d5acf3ac1ddee7b5c285 Mon Sep 17 00:00:00 2001 From: Jason Hicks Date: Tue, 6 Oct 2026 04:14:20 -0500 Subject: [PATCH 08/10] fix(task): drain pending presenter updates before unlock --- ...resentAssistantMessage-custom-tool.spec.ts | 69 +++++++++++++++++++ .../presentAssistantMessage.ts | 14 ++-- 2 files changed, 78 insertions(+), 5 deletions(-) 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 e37bf2cd48..e9ff1127ff 100644 --- a/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts +++ b/src/core/assistant-message/__tests__/presentAssistantMessage-custom-tool.spec.ts @@ -209,6 +209,75 @@ 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 }, diff --git a/src/core/assistant-message/presentAssistantMessage.ts b/src/core/assistant-message/presentAssistantMessage.ts index dc0a8e18f8..417dcf7a45 100644 --- a/src/core/assistant-message/presentAssistantMessage.ts +++ b/src/core/assistant-message/presentAssistantMessage.ts @@ -98,7 +98,13 @@ export async function presentAssistantMessage(cline: Task) { cline.presentAssistantMessageLocked = true try { - await presentAssistantMessageBlock(cline) + // 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. @@ -1128,10 +1134,8 @@ async function presentAssistantMessageBlock(cline: Task): Promise { } } - // Block is partial, but the read stream may have finished. - if (cline.presentAssistantMessageHasPendingUpdates) { - return await presentAssistantMessageBlock(cline) - } + // Pending updates are drained by presentAssistantMessage, which owns the + // lock and checks for them immediately before releasing it. } /** From e35972fc74ae4e6dab84cfb1b2a2d8effe7feb54 Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Fri, 9 Oct 2026 03:11:43 +0000 Subject: [PATCH 09/10] test(task): cover tool-turn readiness gaps --- src/core/task/__tests__/Task.spec.ts | 108 ++++++++++++++++++++++++--- 1 file changed, 98 insertions(+), 10 deletions(-) diff --git a/src/core/task/__tests__/Task.spec.ts b/src/core/task/__tests__/Task.spec.ts index fa3aa09fcb..aa3e75d640 100644 --- a/src/core/task/__tests__/Task.spec.ts +++ b/src/core/task/__tests__/Task.spec.ts @@ -649,6 +649,33 @@ describe("Cline", () => { 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, @@ -687,6 +714,8 @@ describe("Cline", () => { 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", @@ -781,15 +810,13 @@ describe("Cline", () => { try { await task.recursivelyMakeClineRequests([{ type: "text", text: "read a file, then continue" }]) expect(attemptApiRequestSpy).toHaveBeenCalledTimes(2) - expect(continuationUserContent).toEqual( - expect.arrayContaining([ - expect.objectContaining({ - type: "tool_result", - tool_use_id: "call_read", - content: "File: README.md\nfinished", - }), - ]), - ) + 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 () => {}) } @@ -869,7 +896,6 @@ describe("Cline", () => { content: "File: package.json\nfinished", }), ]) - expect(task.userMessageContentReady).toBe(false) expect(task.presentAssistantMessageLocked).toBe(false) expect(sawCompleteToolTurn).toBe(true) } finally { @@ -880,6 +906,68 @@ describe("Cline", () => { } }) + 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, From 4d33eeb9683b5fac92ff9795b8e01f5f68a6fbab Mon Sep 17 00:00:00 2001 From: Elliott de Launay Date: Fri, 9 Oct 2026 03:13:45 +0000 Subject: [PATCH 10/10] test(task): model tool-turn continuation lock and latch --- docs/architecture/task-lifecycle-model.md | 41 +- package.json | 3 +- scripts/check-tool-turn-continuation.ts | 475 ++++++++++++++++++++++ 3 files changed, 515 insertions(+), 4 deletions(-) create mode 100644 scripts/check-tool-turn-continuation.ts 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)}`, +)