diff --git a/packages/cli/src/ui/hooks/useGeminiStream.test.tsx b/packages/cli/src/ui/hooks/useGeminiStream.test.tsx index 6230b9009b..92571cdaa8 100644 --- a/packages/cli/src/ui/hooks/useGeminiStream.test.tsx +++ b/packages/cli/src/ui/hooks/useGeminiStream.test.tsx @@ -714,6 +714,128 @@ describe('useGeminiStream', () => { }); }); + it('drops a late tool result whose callId is already paired in chat.history (Race A dedup)', async () => { + // Race A repro: the chat-internal repair pass already synthesized a + // functionResponse for this callId on the Retry push (because the + // partial-tool_use turn was orphan when Ctrl+Y landed). The live + // scheduler's late real result must NOT also be submitted, otherwise + // the wire payload would carry two functionResponse parts for the + // same callId and the second one would land as an orphan tool_result. + // The dedup MUST run regardless of `isResponding`, because the + // scheduler's `onAllToolCallsComplete` is single-shot and would + // otherwise leave the tool stuck in `completed-but-not-submitted`. + const lateRealResult: TrackedCompletedToolCall = { + request: { + callId: 'call_race_A', + name: 'read_file', + args: { path: '/tmp/x.txt' }, + isClientInitiated: false, + prompt_id: 'prompt-race-a', + }, + status: 'success', + responseSubmittedToGemini: false, + response: { + callId: 'call_race_A', + responseParts: [ + { + functionResponse: { + id: 'call_race_A', + name: 'read_file', + response: { output: 'real file contents' }, + }, + }, + ], + errorType: undefined, + }, + tool: { displayName: 'ReadFile' }, + invocation: { + getDescription: () => 'read /tmp/x.txt', + } as unknown as AnyToolInvocation, + } as TrackedCompletedToolCall; + + const client = new MockedGeminiClientClass(mockConfig); + // Simulate the chat-internal repair pass having already planted a + // synthetic functionResponse for the same callId on the previous + // (Retry) push. + client.getHistory = vi.fn().mockReturnValue([ + { role: 'user', parts: [{ text: 'open /tmp/x.txt' }] }, + { + role: 'model', + parts: [ + { + functionCall: { + id: 'call_race_A', + name: 'read_file', + args: { path: '/tmp/x.txt' }, + }, + }, + ], + }, + { + role: 'user', + parts: [ + { text: 'retry' }, + { + functionResponse: { + id: 'call_race_A', + name: 'read_file', + response: { + error: 'Tool execution result was not recorded', + }, + }, + }, + ], + }, + ]); + + let capturedOnComplete: + | ((completedTools: TrackedToolCall[]) => Promise) + | null = null; + mockUseReactToolScheduler.mockImplementation((onComplete) => { + capturedOnComplete = onComplete; + return [[], mockScheduleToolCalls, mockMarkToolsAsSubmitted]; + }); + + renderHook(() => + useGeminiStream( + client, + [], + mockAddItem, + mockConfig, + mockLoadedSettings, + mockOnDebugMessage, + mockHandleSlashCommand, + false, + () => 'vscode' as EditorType, + () => {}, + () => Promise.resolve(), + false, + () => {}, + () => {}, + () => {}, + () => {}, + 80, + 24, + ), + ); + + await act(async () => { + if (capturedOnComplete) { + await capturedOnComplete([lateRealResult]); + } + }); + + await waitFor(() => { + // The dedup hit must `markToolsAsSubmitted` so the UI/scheduler is + // unblocked even though we drop the real result on the wire. + expect(mockMarkToolsAsSubmitted).toHaveBeenCalledWith(['call_race_A']); + }); + + // No follow-up submission: the synthetic in history already closes + // the tool_use ↔ tool_result pair. + expect(mockSendMessageStream).not.toHaveBeenCalled(); + }); + it('should not flicker streaming state to Idle between tool completion and submission', async () => { const toolCallResponseParts: PartListUnion = [ { text: 'tool 1 final response' }, diff --git a/packages/cli/src/ui/hooks/useGeminiStream.ts b/packages/cli/src/ui/hooks/useGeminiStream.ts index e74636fc11..7f4ecdd2e8 100644 --- a/packages/cli/src/ui/hooks/useGeminiStream.ts +++ b/packages/cli/src/ui/hooks/useGeminiStream.ts @@ -1968,10 +1968,6 @@ export const useGeminiStream = ( const handleCompletedTools = useCallback( async (completedToolCallsFromScheduler: TrackedToolCall[]) => { - if (isResponding) { - return; - } - const completedAndReadyToSubmitTools = completedToolCallsFromScheduler.filter( ( @@ -1994,6 +1990,49 @@ export const useGeminiStream = ( }, ); + // History-based dedup MUST run before the `isResponding` early-return. + // If a synthetic `functionResponse` for this callId is already in + // chat.history (planted on session-load by + // `client.repairOrphanedToolUseTurnsInHistory` or on every + // `chat.sendMessageStream` push by the inline repair pass), the + // in-flight scheduler result must be marked submitted NOW — + // `useReactToolScheduler.allToolCallsCompleteHandler` is single-shot + // per batch, so a later isResponding=true early-return would leave + // the tool stuck in `completed-but-not-submitted` forever (Race A + // surfaced in PR #4176 review). The real result is dropped on the + // wire — same trade-off upstream Claude Code makes when its + // `StreamingToolExecutor.discard()` follows a + // `yieldMissingToolResultBlocks` synthesis (`query.ts:733` + `:984`). + const historyCallIdsWithResponse = new Set(); + // Guard the call: some test harnesses build a partial GeminiClient + // mock without `getHistory`. Skipping dedup in that case is safe — + // it just means tests that never set up the repair pre-condition + // run with the original (pre-dedup) submission shape. + if (geminiClient && typeof geminiClient.getHistory === 'function') { + for (const entry of geminiClient.getHistory()) { + if (entry.role !== 'user') continue; + for (const part of entry.parts ?? []) { + const id = part.functionResponse?.id; + if (id) historyCallIdsWithResponse.add(id); + } + } + } + const dedupedCallIds = completedAndReadyToSubmitTools + .filter((tc) => historyCallIdsWithResponse.has(tc.request.callId)) + .map((tc) => tc.request.callId); + if (dedupedCallIds.length > 0) { + debugLogger.warn( + `[REPAIR] Dropping ${dedupedCallIds.length} late tool result(s) ` + + `whose callId already has a functionResponse in history: ` + + `${dedupedCallIds.join(', ')}`, + ); + markToolsAsSubmitted(dedupedCallIds); + } + + if (isResponding) { + return; + } + // Finalize any client-initiated tools as soon as they are done. const clientTools = completedAndReadyToSubmitTools.filter( (t) => t.request.isClientInitiated, @@ -2019,62 +2058,19 @@ export const useGeminiStream = ( ); } - const geminiToolsRaw = completedAndReadyToSubmitTools.filter( - (t) => !t.request.isClientInitiated, + const geminiTools = completedAndReadyToSubmitTools.filter( + (t) => + !t.request.isClientInitiated && + !historyCallIdsWithResponse.has(t.request.callId), ); - for (const toolCall of geminiToolsRaw) { + for (const toolCall of geminiTools) { geminiClient?.recordCompletedToolCall( toolCall.request.name, toolCall.request.args as Record, ); } - // History-based dedup: if a synthetic `functionResponse` for this - // callId is already in chat.history (planted by - // `client.repairOrphanedToolUseTurnsInHistory()` on session-load or - // Retry), the in-flight scheduler result would land as a duplicate - // `tool_result` and produce two consecutive user turns where the - // second is orphaned (no preceding tool_use — the synthetic ate it). - // - // For dedup hits: mark the tool as submitted so the UI advances and - // `useReactToolScheduler.allToolCallsCompleteHandler` (single-shot) - // doesn't leave the call permanently stuck in `completed-but-not- - // submitted`. The real result is dropped on the wire — same trade-off - // upstream Claude Code makes when its `StreamingToolExecutor.discard()` - // is followed by a `yieldMissingToolResultBlocks` synthesis - // (`query.ts:733` + `:984`). The model sees the synthetic error and - // can retry the tool if it still wants the result. - const historyCallIdsWithResponse = new Set(); - // Guard the call: some test harnesses build a partial GeminiClient - // mock without `getHistory`. Skipping dedup in that case is safe — - // it just means tests that never set up the repair pre-condition - // run with the original (pre-dedup) submission shape. - if (geminiClient && typeof geminiClient.getHistory === 'function') { - for (const entry of geminiClient.getHistory()) { - if (entry.role !== 'user') continue; - for (const part of entry.parts ?? []) { - const id = part.functionResponse?.id; - if (id) historyCallIdsWithResponse.add(id); - } - } - } - - const dedupedCallIds = geminiToolsRaw - .filter((tc) => historyCallIdsWithResponse.has(tc.request.callId)) - .map((tc) => tc.request.callId); - if (dedupedCallIds.length > 0) { - debugLogger.warn( - `[REPAIR] Dropping ${dedupedCallIds.length} late tool result(s) ` + - `whose callId already has a synthetic functionResponse in ` + - `history: ${dedupedCallIds.join(', ')}`, - ); - markToolsAsSubmitted(dedupedCallIds); - } - const geminiTools = geminiToolsRaw.filter( - (tc) => !historyCallIdsWithResponse.has(tc.request.callId), - ); - if (geminiTools.length === 0) { return; } diff --git a/packages/core/src/core/client.ts b/packages/core/src/core/client.ts index 867fa2146e..f0df569630 100644 --- a/packages/core/src/core/client.ts +++ b/packages/core/src/core/client.ts @@ -700,9 +700,14 @@ export class GeminiClient { // partial-tool_use push (see `processStreamResponse`) and the React // scheduler's tool_result submission. Without this pass, the first // API call on a resumed session would 400 with the same - // `tool_use_id ... corresponding tool_use` error this whole subsystem - // is trying to escape. - this.chat.repairOrphanedToolUseTurns(); + // `tool_use_id ... corresponding tool_use` error this whole + // subsystem is trying to escape. (Belt-and-suspenders: the same + // helper runs again inside `chat.sendMessageStream` after the user + // content is pushed, so a dangling left here by setHistory / + // compaction reordering is also caught — but doing it here keeps + // any pre-send code reading `chat.history` from seeing a malformed + // shape.) + this.repairOrphanedToolUseTurnsInHistory(); const sessionStartAdditionalContext = await this.fireSessionStartHook(sessionStartSource); @@ -1091,22 +1096,13 @@ export class GeminiClient { if (messageType === SendMessageType.Retry) { this.stripOrphanedUserEntriesFromHistory(); - // Close any dangling `model[functionCall]` whose tool_result never - // landed before composing the retry payload. Ctrl+Y race: the user - // retried while a tool was still running on a partial-tool_use turn - // pushed by `processStreamResponse`'s mid-stream error path. The - // scheduler's `onAllToolCallsComplete` is single-shot and gated on - // `isResponding` (`useGeminiStream:1971`), so the eventual - // `tool_result` would otherwise be silently swallowed and the next - // API call would 400 with "tool_use_id ... corresponding tool_use" - // anyway. The synthesized `error` `functionResponse` keeps the wire - // invariant intact; the live scheduler dedupes against history in - // `handleCompletedTools` before submitting its real result so the - // synthetic doesn't collide with a late real one. - // - // Restricted to the Retry branch to mirror `stripOrphanedUserEntries` - // scope. Crash-resume's path is covered separately in `startChat()`. - this.repairOrphanedToolUseTurnsInHistory(); + // The matching dangling-`functionCall` repair runs inside + // `chat.sendMessageStream` AFTER the user content is pushed, so any + // tool_result the user is supplying (Retry of a ToolResult + // submission, lastPrompt === fr parts) closes the pair via the real + // `functionResponse` before we synthesize an error one. Doing the + // repair here would happen pre-push and race against the user + // content's own pairing — see PR #4176 review for the corner. } // Fire UserPromptSubmit hook through MessageBus (only if hooks are enabled) diff --git a/packages/core/src/core/geminiChat.test.ts b/packages/core/src/core/geminiChat.test.ts index bd281d537a..cacd85af8a 100644 --- a/packages/core/src/core/geminiChat.test.ts +++ b/packages/core/src/core/geminiChat.test.ts @@ -597,6 +597,138 @@ describe('GeminiChat', async () => { 'This is the visible text that should not be lost.', ); }); + + it('synthesizes a functionResponse for a dangling tool_use before sending', async () => { + // End-to-end: when sendMessageStream is invoked on a chat whose + // history carries a dangling `model[functionCall]` (typical state + // after a Ctrl+Y race or a crash-resume on a partial-tool_use + // turn), the inline repair pass closes the pair against the + // just-pushed user content so the wire payload doesn't 400 with + // "tool_use_id ... corresponding tool_use". + chat.setHistory([ + { role: 'user', parts: [{ text: 'first message' }] }, + { + role: 'model', + parts: [ + { + functionCall: { + id: 'call_dangling_for_send', + name: 'read_file', + args: { path: '/tmp/x' }, + }, + }, + ], + }, + ]); + + const ackStream = (async function* () { + yield { + candidates: [ + { + content: { role: 'model', parts: [{ text: 'ok' }] }, + finishReason: 'STOP', + }, + ], + } as unknown as GenerateContentResponse; + })(); + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + ackStream, + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'next user prompt after a stream-error-mid-tool_use' }, + 'prompt-send-repair', + ); + for await (const _ of stream) { + /* drain */ + } + + const history = chat.getHistory(); + // The dangling fc should now be followed by a user turn that + // carries both the user-supplied text AND the synthetic fr that + // closes the pair. + const userTurn = history[2]!; + expect(userTurn.role).toBe('user'); + const fr = userTurn.parts!.find((p) => p.functionResponse); + expect(fr?.functionResponse?.id).toBe('call_dangling_for_send'); + expect(fr?.functionResponse?.name).toBe('read_file'); + expect( + (fr?.functionResponse?.response as { error?: string })?.error, + ).toMatch(/interrupted/i); + // The user's own text part is still present. + expect( + userTurn.parts!.some( + (p) => + p.text === 'next user prompt after a stream-error-mid-tool_use', + ), + ).toBe(true); + }); + + it('does NOT synthesize when the user supplies a matching tool_result', async () => { + // Retry-of-ToolResult case (lastPrompt is a functionResponse Part + // array): the user-supplied tool_result must close the pair before + // the inline repair pass sees it, so no synthetic error is + // injected. Otherwise the wire payload would carry two + // functionResponse parts for the same callId — the real one and a + // bogus synthetic. + chat.setHistory([ + { role: 'user', parts: [{ text: 'do the read' }] }, + { + role: 'model', + parts: [ + { + functionCall: { + id: 'call_retry_real_fr', + name: 'read_file', + args: { path: '/tmp/y' }, + }, + }, + ], + }, + ]); + + const ackStream = (async function* () { + yield { + candidates: [ + { + content: { role: 'model', parts: [{ text: 'ack' }] }, + finishReason: 'STOP', + }, + ], + } as unknown as GenerateContentResponse; + })(); + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + ackStream, + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { + message: { + functionResponse: { + id: 'call_retry_real_fr', + name: 'read_file', + response: { output: 'real-tool-output' }, + }, + }, + }, + 'prompt-retry-real-fr', + ); + for await (const _ of stream) { + /* drain */ + } + + const userTurn = chat.getHistory()[2]!; + const frParts = userTurn.parts!.filter((p) => p.functionResponse); + // Exactly ONE functionResponse — the real one. No synthetic. + expect(frParts.length).toBe(1); + expect(frParts[0]!.functionResponse?.id).toBe('call_retry_real_fr'); + expect( + (frParts[0]!.functionResponse?.response as { output?: string })?.output, + ).toBe('real-tool-output'); + }); + it('should throw an error when a tool call is followed by an empty stream response', async () => { vi.useFakeTimers(); try { diff --git a/packages/core/src/core/geminiChat.ts b/packages/core/src/core/geminiChat.ts index ee4e6d90fc..44d50e64d2 100644 --- a/packages/core/src/core/geminiChat.ts +++ b/packages/core/src/core/geminiChat.ts @@ -728,6 +728,26 @@ export class GeminiChat { // Add user content to history ONCE before any attempts. this.history.push(userContent); userContentAdded = true; + // Close any dangling `model[functionCall]` whose `functionResponse` + // never landed by the time we compose the request. Runs AFTER the + // user-supplied turn lands so a tool_result the user is supplying + // gets the first chance to close the pair before we synthesize an + // `error` `functionResponse`. Covers: + // - Stream errored mid-tool_use (partial assistant push left a + // dangling functionCall), then the React scheduler's eventual + // tool_result lost the race against a Ctrl+Y retry whose + // onAllToolCallsComplete fired into `isResponding=true` and + // skipped submission. + // - The same shape from a process crash / OOM mid-flight (the + // transcript JSONL preserves the dangling model[fc] across + // `--resume`; `startChat()` calls this once on load, but a + // belt-and-suspenders pass here covers anything that slipped + // past — including dangling shapes the load-time repair didn't + // visit because compaction / setHistory ran after it). + // The React scheduler's late real result is then dedup'd against + // chat.history in `useGeminiStream.handleCompletedTools` so the + // synthetic doesn't collide with it on the wire. + repairOrphanedToolUseTurns(this.history); requestContents = this.getHistory(true); } catch (error) { if (userContentAdded) {