Skip to content

Commit 7b5e2f7

Browse files
edelaunaedelauna
committed
refactor(parser): cut PR #700 to parser-only streaming scope
Restore Task.ts persistence coordination and stream-time tool_call_end finalization to main; retain only the #695 parser fixes with their focused tests.
1 parent 4ffe890 commit 7b5e2f7

5 files changed

Lines changed: 165 additions & 318 deletions

File tree

‎src/core/task/Task.ts‎

Lines changed: 70 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -437,7 +437,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
437437
* Reset to `false` at the start of each API request.
438438
* Set to `true` only after the assistant message is durably saved.
439439
*/
440-
assistantMessageSavedToHistory = true
440+
assistantMessageSavedToHistory = false
441441
private assistantMessagePersistencePromise!: Promise<AssistantMessagePersistenceResult>
442442
private resolveAssistantMessagePersistence!: (result: AssistantMessagePersistenceResult) => void
443443
private assistantMessagePersistenceCancellation?: AssistantMessagePersistenceCancellation
@@ -500,46 +500,6 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
500500
// Native tool call streaming state (track which index each tool is at)
501501
private streamingToolCallIndices: Map<string, number> = new Map()
502502

503-
/**
504-
* Finalize a streaming native tool call by id and present it.
505-
*
506-
* Shared by every site that observes a tool_call_end: the per-chunk event
507-
* loop, the stream-level tool_call_end case, and the end-of-stream
508-
* finalizeRawChunks() pass. Calling it again for an already-finalized id is a
509-
* safe no-op because finalizeStreamingToolCall() and the index map entry are
510-
* both cleared on first finalize.
511-
*/
512-
private finalizeStreamingToolCallById(id: string, nativeToolCallParserScope: object): void {
513-
const finalToolUse = NativeToolCallParser.finalizeStreamingToolCall(id, nativeToolCallParserScope)
514-
const toolUseIndex = this.streamingToolCallIndices.get(id)
515-
516-
if (finalToolUse) {
517-
;(finalToolUse as any).id = id
518-
if (toolUseIndex !== undefined) {
519-
this.assistantMessageContent[toolUseIndex] = finalToolUse
520-
}
521-
this.streamingToolCallIndices.delete(id)
522-
this.userMessageContentReady = false
523-
/* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */
524-
this.presentAssistantMessageSafe()
525-
} else if (toolUseIndex !== undefined) {
526-
// finalizeStreamingToolCall returned null (malformed JSON or missing args).
527-
// Clear partial arguments before presentation so truncated values cannot run
528-
// or enter conversation history.
529-
const existingToolUse = this.assistantMessageContent[toolUseIndex]
530-
if (existingToolUse && existingToolUse.type === "tool_use") {
531-
existingToolUse.partial = false
532-
existingToolUse.nativeArgs = undefined
533-
existingToolUse.params = {}
534-
;(existingToolUse as any).id = id
535-
}
536-
this.streamingToolCallIndices.delete(id)
537-
this.userMessageContentReady = false
538-
/* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */
539-
this.presentAssistantMessageSafe()
540-
}
541-
}
542-
543503
// Cached model info for current streaming session (set at start of each API request)
544504
// This prevents excessive getModel() calls during tool execution
545505
cachedStreamingModel?: { id: string; info: ModelInfo }
@@ -1074,9 +1034,6 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
10741034
* If the message resolves a pending action, retries the save on initial failure before clearing the action.
10751035
*/
10761036
private async addToApiConversationHistory(message: Anthropic.MessageParam, reasoning?: string): Promise<void> {
1077-
if (message.role === "assistant") {
1078-
this.assistantMessageSavedToHistory = false
1079-
}
10801037
const resolvesPendingAction =
10811038
this.pendingAction &&
10821039
message.role === "user" &&
@@ -1194,18 +1151,17 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
11941151
* So we usually only need to flush the pending user message with tool_results.
11951152
*/
11961153
public async flushPendingToolResultsToHistory(): Promise<boolean> {
1154+
// Only flush if there's actually pending content to save
1155+
if (this.userMessageContent.length === 0) {
1156+
return true
1157+
}
11971158
if (this.abort) {
11981159
return false
11991160
}
12001161

1201-
// CRITICAL: Wait for the assistant message to be saved to API history before
1202-
// any early return, including the empty-content case below. Delegation relies
1203-
// on this barrier: an auto-approved new_task can execute while the parent
1204-
// assistant turn still awaits persistence, and disposing the parent at that
1205-
// point would leave its tool_use turn out of the durable history.
1206-
//
1207-
// Without the wait, tool_result blocks would also appear BEFORE tool_use blocks
1208-
// in the conversation history, causing API errors like:
1162+
// CRITICAL: Wait for the assistant message to be saved to API history first.
1163+
// Without this, tool_result blocks would appear BEFORE tool_use blocks in the
1164+
// conversation history, causing API errors like:
12091165
// "unexpected `tool_use_id` found in `tool_result` blocks"
12101166
//
12111167
// This can happen when parallel tools are called (e.g., update_todo_list + new_task).
@@ -1230,11 +1186,6 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
12301186
}
12311187
}
12321188

1233-
// Only flush if there's actually pending content to save
1234-
if (this.userMessageContent.length === 0) {
1235-
return true
1236-
}
1237-
12381189
// If task was aborted while waiting, don't flush
12391190
if (this.abort) {
12401191
return false
@@ -3285,6 +3236,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
32853236
this.didRejectTool = false
32863237
this.didAlreadyUseTool = false
32873238
this.assistantMessageSavedToHistory = false
3239+
this.didFinishAbortingStream = false
32883240
this.resetAssistantMessagePersistence()
32893241
// Reset tool failure flag for each new assistant turn - this ensures that tool failures
32903242
// only prevent attempt_completion within the same assistant message, not across turns
@@ -3474,22 +3426,11 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
34743426
this.presentAssistantMessageSafe()
34753427
}
34763428
}
3477-
} else if (event.type === "tool_call_end") {
3478-
this.finalizeStreamingToolCallById(event.id, nativeToolCallParserScope)
34793429
}
34803430
}
34813431
break
34823432
}
34833433

3484-
case "tool_call_end": {
3485-
// Providers emit a tool_call_end chunk when finish_reason is
3486-
// "tool_calls" (either directly or via processFinishReason).
3487-
// Finalize the streaming tool call now so it is presented during
3488-
// streaming rather than waiting for finalizeRawChunks() at stream end.
3489-
this.finalizeStreamingToolCallById(chunk.id, nativeToolCallParserScope)
3490-
break
3491-
}
3492-
34933434
case "tool_call": {
34943435
// Legacy: Handle complete tool calls (for backward compatibility)
34953436
// Convert native tool call to ToolUse format
@@ -3775,8 +3716,8 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
37753716
await abortStream(cancelReason, streamingFailedMessage)
37763717

37773718
if (this.abort) {
3778-
// User cancelled - abort the entire task
3779-
this.abortReason = cancelReason
3719+
// ??= keeps the first reason; a cancel can land during abortStream after cancelReason was already computed.
3720+
this.abortReason ??= "user_cancelled"
37803721
await this.abortTask()
37813722
} else if (error instanceof OutputTokenLimitError) {
37823723
// Truncation repeats on an identical request, so never auto-retry it
@@ -3811,8 +3752,7 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
38113752
console.log(
38123753
`[Task#${this.taskId}.${this.instanceId}] Task aborted during mid-stream retry backoff`,
38133754
)
3814-
// Abort the entire task
3815-
this.abortReason = "user_cancelled"
3755+
this.abortReason ??= "user_cancelled"
38163756
await this.abortTask()
38173757
break
38183758
}
@@ -3858,7 +3798,64 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
38583798
const finalizeEvents = NativeToolCallParser.finalizeRawChunks(nativeToolCallParserScope)
38593799
for (const event of finalizeEvents) {
38603800
if (event.type === "tool_call_end") {
3861-
this.finalizeStreamingToolCallById(event.id, nativeToolCallParserScope)
3801+
// Finalize the streaming tool call
3802+
const finalToolUse = NativeToolCallParser.finalizeStreamingToolCall(
3803+
event.id,
3804+
nativeToolCallParserScope,
3805+
)
3806+
3807+
// Get the index for this tool call
3808+
const toolUseIndex = this.streamingToolCallIndices.get(event.id)
3809+
3810+
if (finalToolUse) {
3811+
// Store the tool call ID
3812+
;(finalToolUse as any).id = event.id
3813+
3814+
// Get the index and replace partial with final
3815+
if (toolUseIndex !== undefined) {
3816+
this.assistantMessageContent[toolUseIndex] = finalToolUse
3817+
}
3818+
3819+
// Clean up tracking
3820+
this.streamingToolCallIndices.delete(event.id)
3821+
3822+
// Mark that we have new content to process
3823+
this.userMessageContentReady = false
3824+
3825+
// Present the finalized tool call
3826+
/* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */
3827+
this.presentAssistantMessageSafe()
3828+
} else if (toolUseIndex !== undefined) {
3829+
// finalizeStreamingToolCall returned null (malformed JSON or missing args).
3830+
// existingToolUse is the same object the streaming phase was mutating in
3831+
// place, so it still carries nativeArgs AND params built from the incomplete
3832+
// partial parse (e.g. a truncated write_to_file `content` string) - both were
3833+
// only ever meant for live progress display, never for execution or for
3834+
// ending up in conversation history. Mark the tool as non-partial so it's
3835+
// presented as complete, and clear both so presentAssistantMessage's
3836+
// `!block.nativeArgs` guard short-circuits with a structured tool_result
3837+
// instead of executing the truncated value, and so the toolUse.nativeArgs ||
3838+
// toolUse.params fallback used when recording history doesn't fall through to
3839+
// the same truncated data under a different name.
3840+
const existingToolUse = this.assistantMessageContent[toolUseIndex]
3841+
if (existingToolUse && existingToolUse.type === "tool_use") {
3842+
existingToolUse.partial = false
3843+
existingToolUse.nativeArgs = undefined
3844+
existingToolUse.params = {}
3845+
// Ensure it has the ID for native protocol
3846+
;(existingToolUse as any).id = event.id
3847+
}
3848+
3849+
// Clean up tracking
3850+
this.streamingToolCallIndices.delete(event.id)
3851+
3852+
// Mark that we have new content to process
3853+
this.userMessageContentReady = false
3854+
3855+
// Present the tool call - validation will handle missing params
3856+
/* v8 ignore next -- streaming presenter; .catch lives in presentAssistantMessageSafe (covered) */
3857+
this.presentAssistantMessageSafe()
3858+
}
38623859
}
38633860
}
38643861

‎src/core/task/__tests__/Task.persistence.spec.ts‎

Lines changed: 0 additions & 76 deletions
Original file line numberDiff line numberDiff line change
@@ -923,7 +923,6 @@ describe("Task persistence", () => {
923923
task: "test task",
924924
startTask: false,
925925
})
926-
task.assistantMessageSavedToHistory = false
927926
task.userMessageContent = [{ type: "tool_result", tool_use_id: "tool-1", content: "done" }]
928927
vi.spyOn(task, "waitForCurrentAssistantMessagePersistence").mockResolvedValue(false)
929928

@@ -966,81 +965,6 @@ describe("Task persistence", () => {
966965
vi.useRealTimers()
967966
}
968967
})
969-
970-
it("waits for current assistant persistence before an empty-result flush", async () => {
971-
vi.useFakeTimers()
972-
const assistantSave = createDeferred<void>()
973-
mockSaveApiMessages.mockReturnValueOnce(assistantSave.promise)
974-
const task = new Task({
975-
provider: mockProvider,
976-
apiConfiguration: mockApiConfig,
977-
task: "test task",
978-
startTask: false,
979-
})
980-
981-
try {
982-
const saving = getTaskPersistenceAccess(task).addToApiConversationHistory({
983-
role: "assistant",
984-
content: [{ type: "text", text: "turn still being written" }],
985-
})
986-
expect(task.assistantMessageSavedToHistory).toBe(false)
987-
988-
let settled = false
989-
const flushing = task.flushPendingToolResultsToHistory().then((value) => {
990-
settled = true
991-
return value
992-
})
993-
994-
await vi.advanceTimersByTimeAsync(0)
995-
expect(settled).toBe(false)
996-
997-
assistantSave.resolve(undefined)
998-
await vi.runAllTimersAsync()
999-
await expect(flushing).resolves.toBe(true)
1000-
await saving
1001-
1002-
expect(task.assistantMessageSavedToHistory).toBe(true)
1003-
expect(task.userMessageContent).toEqual([])
1004-
expect(mockSaveApiMessages).toHaveBeenCalledTimes(1)
1005-
} finally {
1006-
assistantSave.resolve(undefined)
1007-
mockSaveApiMessages.mockResolvedValue(undefined)
1008-
vi.useRealTimers()
1009-
}
1010-
})
1011-
1012-
it("does not report success for an empty-result flush when assistant persistence fails", async () => {
1013-
vi.useFakeTimers()
1014-
mockSaveApiMessages.mockRejectedValue(new Error("assistant write failed"))
1015-
const consoleWarn = vi.spyOn(console, "warn").mockImplementation(() => undefined)
1016-
const task = new Task({
1017-
provider: mockProvider,
1018-
apiConfiguration: mockApiConfig,
1019-
task: "test task",
1020-
startTask: false,
1021-
})
1022-
1023-
try {
1024-
await getTaskPersistenceAccess(task).addToApiConversationHistory({
1025-
role: "assistant",
1026-
content: [{ type: "text", text: "turn still being written" }],
1027-
})
1028-
1029-
const flushing = task.flushPendingToolResultsToHistory()
1030-
await vi.runAllTimersAsync()
1031-
1032-
await expect(flushing).resolves.toBe(false)
1033-
expect(task.assistantMessageSavedToHistory).toBe(false)
1034-
expect(consoleWarn).toHaveBeenCalledWith(
1035-
expect.stringContaining("failed to persist assistant message"),
1036-
expect.any(Error),
1037-
)
1038-
} finally {
1039-
consoleWarn.mockRestore()
1040-
mockSaveApiMessages.mockResolvedValue(undefined)
1041-
vi.useRealTimers()
1042-
}
1043-
})
1044968
})
1045969

1046970
// ── saveClineMessages ────────────────────────────────────────────────

0 commit comments

Comments
 (0)