diff --git a/packages/cli/src/ui/hooks/useGeminiStream.test.tsx b/packages/cli/src/ui/hooks/useGeminiStream.test.tsx index 797f875899..e951912cc6 100644 --- a/packages/cli/src/ui/hooks/useGeminiStream.test.tsx +++ b/packages/cli/src/ui/hooks/useGeminiStream.test.tsx @@ -1137,6 +1137,196 @@ describe('useGeminiStream', () => { releaseStream(); }); + it('handles a mixed batch (one deduped + one non-deduped) without double-counting telemetry (qwen-latest-series-invite-beta-v34 thread on PR #4176)', async () => { + // The dedup filter on `geminiTools` (`!historyCallIdsWithResponse.has(callId)`) + // is the only thing preventing double `recordCompletedToolCall` + // for tools whose late real result lands AFTER the orphan-tool_use + // repair already planted a synthetic. Existing dedup tests supply + // ONLY deduped tools, so a regression that removed that filter + // would silently inflate `toolCallCount` (and flip + // `skillsModifiedInSession` for the SAME skill-write callId twice) + // without breaking any current test. + // + // Mixed-batch repro: scheduler completes two tools in the same + // batch — one whose callId already has a fr in history (deduped), + // one whose callId is fresh (must reach sendMessageStream). Pin: + // (a) markToolsAsSubmitted called with BOTH callIds, + // (b) recordCompletedToolCall fires once per non-deduped tool, + // NOT twice for the deduped one, + // (c) sendMessageStream IS called (the non-deduped tool's real + // result must reach the wire). + const dedupedTool = { + request: { + callId: 'call_mixed_deduped', + name: 'read_file', + args: { path: '/tmp/d.txt' }, + isClientInitiated: false, + prompt_id: 'prompt-mixed', + }, + status: 'success', + responseSubmittedToGemini: false, + response: { + callId: 'call_mixed_deduped', + responseParts: [ + { + functionResponse: { + id: 'call_mixed_deduped', + name: 'read_file', + response: { output: 'late real for deduped' }, + }, + }, + ], + resultDisplay: undefined, + error: undefined, + errorType: undefined, + }, + tool: { + name: 'read_file', + displayName: 'ReadFile', + description: 'Read a file', + build: vi.fn(), + } as any, + invocation: { + getDescription: () => 'read /tmp/d.txt', + } as unknown as AnyToolInvocation, + } as unknown as TrackedCompletedToolCall; + + const freshTool = { + request: { + callId: 'call_mixed_fresh', + name: 'read_file', + args: { path: '/tmp/f.txt' }, + isClientInitiated: false, + prompt_id: 'prompt-mixed', + }, + status: 'success', + responseSubmittedToGemini: false, + response: { + callId: 'call_mixed_fresh', + responseParts: [ + { + functionResponse: { + id: 'call_mixed_fresh', + name: 'read_file', + response: { output: 'real for fresh' }, + }, + }, + ], + resultDisplay: undefined, + error: undefined, + errorType: undefined, + }, + tool: { + name: 'read_file', + displayName: 'ReadFile', + description: 'Read a file', + build: vi.fn(), + } as any, + invocation: { + getDescription: () => 'read /tmp/f.txt', + } as unknown as AnyToolInvocation, + } as unknown as TrackedCompletedToolCall; + + const client = new MockedGeminiClientClass(mockConfig); + // History has fr for ONLY the deduped callId — `call_mixed_fresh` + // is not paired and must flow through to sendMessageStream. + client.getHistory = vi.fn().mockReturnValue([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [ + { + functionCall: { + id: 'call_mixed_deduped', + name: 'read_file', + args: { path: '/tmp/d.txt' }, + }, + }, + { + functionCall: { + id: 'call_mixed_fresh', + name: 'read_file', + args: { path: '/tmp/f.txt' }, + }, + }, + ], + }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'call_mixed_deduped', + 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([dedupedTool, freshTool]); + } + }); + + await waitFor(() => { + // (a) Both callIds were marked submitted somewhere across the + // dedup pass (deduped) and the post-isResponding flow (fresh). + const allMarked = mockMarkToolsAsSubmitted.mock.calls.flatMap( + (call) => call[0] as string[], + ); + expect(allMarked).toContain('call_mixed_deduped'); + expect(allMarked).toContain('call_mixed_fresh'); + }); + + // (b) recordCompletedToolCall fires EXACTLY once per tool (deduped + // gets one call from the dedup-loop; fresh gets one from the + // geminiTools loop). The filter is what prevents the double + // record on the deduped callId. + const recordedCallIds = ( + client.recordCompletedToolCall as unknown as ReturnType + ).mock.calls.map((call) => (call[1] as { path: string }).path); + expect(recordedCallIds.filter((p) => p === '/tmp/d.txt').length).toBe(1); + expect(recordedCallIds.filter((p) => p === '/tmp/f.txt').length).toBe(1); + + // (c) The fresh tool's real result reaches sendMessageStream — + // dedup didn't accidentally suppress it. + expect(mockSendMessageStream).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 c414223b90..bc8f15c548 100644 --- a/packages/cli/src/ui/hooks/useGeminiStream.ts +++ b/packages/cli/src/ui/hooks/useGeminiStream.ts @@ -2044,12 +2044,32 @@ export const useGeminiStream = ( // 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') { + // Walk raw history WITHOUT cloning — `geminiClient.getHistory()` + // returns `structuredClone(this.history)`, which on long sessions + // (200+ entries with sizable tool outputs) costs several ms on + // the React UI thread and visibly stalls streaming when the + // dedup pass runs on every tool-completion batch. The + // `getHistoryFunctionResponseIds` accessor walks history in + // place and only collects the id strings we actually need. + // Guard the call: some test harnesses build a partial + // GeminiClient mock without it. Skipping dedup in that case is + // safe — tests that never set up the repair pre-condition run + // with the original (pre-dedup) submission shape. We fall back + // to the cloning getHistory() path for older mocks that only + // expose that method, so legacy tests stay green. + // qwen-latest-series-invite-beta-v34 thread on PR #4176. + let historyCallIdsWithResponse: Set; + if ( + geminiClient && + typeof geminiClient.getHistoryFunctionResponseIds === 'function' + ) { + historyCallIdsWithResponse = + geminiClient.getHistoryFunctionResponseIds(); + } else if ( + geminiClient && + typeof geminiClient.getHistory === 'function' + ) { + historyCallIdsWithResponse = new Set(); for (const entry of geminiClient.getHistory()) { if (entry.role !== 'user') continue; for (const part of entry.parts ?? []) { @@ -2057,6 +2077,8 @@ export const useGeminiStream = ( if (id) historyCallIdsWithResponse.add(id); } } + } else { + historyCallIdsWithResponse = new Set(); } const dedupedTools = completedAndReadyToSubmitTools.filter((tc) => historyCallIdsWithResponse.has(tc.request.callId), @@ -2104,8 +2126,15 @@ export const useGeminiStream = ( } // Finalize any client-initiated tools as soon as they are done. + // Skip ones whose callId already lives in chat history with a + // matching `functionResponse` — the dedup block above already + // called `markToolsAsSubmitted` for those, and re-dispatching + // the same callIds here would queue an extra React render. + // qwen-latest-series-invite-beta-v34 thread on PR #4176. const clientTools = completedAndReadyToSubmitTools.filter( - (t) => t.request.isClientInitiated, + (t) => + t.request.isClientInitiated && + !historyCallIdsWithResponse.has(t.request.callId), ); if (clientTools.length > 0) { markToolsAsSubmitted(clientTools.map((t) => t.request.callId)); diff --git a/packages/core/src/core/client.ts b/packages/core/src/core/client.ts index a69071d0a9..cef37e3e67 100644 --- a/packages/core/src/core/client.ts +++ b/packages/core/src/core/client.ts @@ -314,6 +314,22 @@ export class GeminiClient { return this.getChat().getHistoryTail(count, curated); } + /** + * Walk-only accessor for the set of `functionResponse.id` strings in + * raw history. Callers that only need the dedup id set (notably + * `useGeminiStream.handleCompletedTools`) MUST prefer this over + * {@link getHistory}, which deep-clones the entire conversation via + * `structuredClone` on every call. On long sessions with sizable + * tool outputs the clone is a multi-millisecond hit on the React UI + * thread; running it on every tool-completion batch caused visible + * frame drops during streaming. See + * `GeminiChat.getHistoryFunctionResponseIds` for the implementation + * and the qwen-latest-series-invite-beta-v34 thread on PR #4176. + */ + getHistoryFunctionResponseIds(): Set { + return this.getChat().getHistoryFunctionResponseIds(); + } + /** * Pop orphaned trailing user entries from the in-memory chat history. * Used by: diff --git a/packages/core/src/core/geminiChat.test.ts b/packages/core/src/core/geminiChat.test.ts index c498ef58e2..cf383ab203 100644 --- a/packages/core/src/core/geminiChat.test.ts +++ b/packages/core/src/core/geminiChat.test.ts @@ -2524,6 +2524,98 @@ describe('GeminiChat', async () => { } }); + it('rolls back the partial assistant turn when an InvalidStreamError fires after a tool_use chunk on the transient-stream retry budget (qwen-latest-series-invite-beta-v34 thread on PR #4176)', async () => { + // Counterpart to the rate-limit rollback above. The + // transient-stream retry budget (NO_FINISH_REASON / + // NO_RESPONSE_TEXT) has its own popPartialIfPushed call site — + // separate from the rate-limit branch the existing test + // covers. Without a regression test, that call could be + // accidentally removed and the rate-limit test would still + // pass while a stale partial silently rode the retry. + vi.useFakeTimers(); + try { + const failingStream = (async function* () { + yield { + candidates: [ + { + content: { + parts: [ + { + functionCall: { + id: 'call_transient_retry_partial', + name: 'read_file', + args: { path: '/tmp/t.txt' }, + }, + }, + ], + }, + }, + ], + } as unknown as GenerateContentResponse; + // Mid-tool_use cut without a finish reason — the transient- + // stream retry budget catches this and retries with delay. + throw new InvalidStreamError( + 'Model stream ended without a finish reason.', + 'NO_FINISH_REASON', + ); + })(); + const successStream = (async function* () { + yield { + candidates: [ + { + content: { parts: [{ text: 'Recovered on retry' }] }, + finishReason: 'STOP', + }, + ], + } as unknown as GenerateContentResponse; + })(); + vi.mocked(mockContentGenerator.generateContentStream) + .mockResolvedValueOnce(failingStream) + .mockResolvedValueOnce(successStream); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + 'prompt-rollback-transient', + ); + const iterator = stream[Symbol.asyncIterator](); + // Advance through the transient-retry delay (initial 2000 ms). + for (;;) { + const next = iterator.next(); + await vi.advanceTimersByTimeAsync(5_000); + const r = await next; + if (r.done) break; + } + + const history = chat.getHistory(); + // Final shape must be clean: [user, model(success text)]. + // The failed attempt's partial functionCall must NOT survive. + expect(history.length).toBe(2); + expect(history[0]!.role).toBe('user'); + expect(history[1]!.role).toBe('model'); + expect(history[1]!.parts!.find((p) => p.text)?.text).toBe( + 'Recovered on retry', + ); + expect(history.some((h) => h.parts?.some((p) => p.functionCall))).toBe( + false, + ); + } finally { + vi.useRealTimers(); + } + }); + + // NOTE: no test for the InvalidStreamError content-retry branch + // (geminiChat.ts ~line 1399). Verified unreachable for that error + // class: `isTransientStreamError` and `isContentError` are the + // same predicate (`error instanceof InvalidStreamError`), so the + // transient branch above always either `continue`s or `break`s + // before control reaches the content branch. The + // `popPartialIfPushed()` call there is preserved as + // defense-in-depth for a future error class that should diverge + // the predicates; see the comment block at that call site for + // the full analysis. qwen-latest-series-invite-beta-v34 thread + // on PR #4176. + it('rolls back the chat-recording entry too when the retry succeeds (yiliang114 PR #4176 follow-up)', async () => { // The in-memory rollback test above asserts `this.history` ends // clean after a retry-success. This test asserts the same about @@ -3775,6 +3867,161 @@ describe('GeminiChat', async () => { }); }); + describe('partial-push marker invariants on history mutation (qwen-latest-series-invite-beta-v34 thread on PR #4176)', () => { + // The whole partial-push lifecycle relies on the invariant + // "every history-mutation method clears the partial-push markers" + // — six sites enforce it (clearHistory, addHistory, setHistory, + // truncateHistory, stripThoughtsFromHistory, + // stripOrphanedUserEntriesFromHistory). If any site forgets, a + // stale `pendingPartialAssistantTurnIndex` could line up with an + // unrelated model turn in the post-mutation history and cause + // `popPartialIfPushed` to splice the WRONG entry — silently losing + // a real assistant response. + // + // The markers are ephemeral within a single sendMessageStream + // call: the `finally` block flushes the deferred JSONL record + // and calls `clearPendingPartialState()` before the generator + // unwinds. So we can't observe non-null markers after a real + // mid-stream error completes — by that point the lifecycle has + // already cleared them. Instead, we plant the markers directly + // via the same private-field assignment the production code uses, + // then call each mutation method and verify both fields are reset + // in lockstep. This pins the invariant against future refactors + // that drop a `clearPendingPartialState()` call from one site + // while the other five still pass. + type PrivateFields = { + pendingPartialAssistantTurnIndex: number | null; + pendingPartialAssistantRecord: unknown; + }; + function plantMarkers(c: GeminiChat): void { + const internal = c as unknown as PrivateFields; + internal.pendingPartialAssistantTurnIndex = 0; + internal.pendingPartialAssistantRecord = { + model: 'test-model', + message: [{ functionCall: { id: 'call_test', name: 't', args: {} } }], + }; + } + function markers(c: GeminiChat): { + idx: number | null; + record: unknown; + } { + const internal = c as unknown as PrivateFields; + return { + idx: internal.pendingPartialAssistantTurnIndex, + record: internal.pendingPartialAssistantRecord, + }; + } + + it('clearHistory() clears the partial-push markers', () => { + chat.setHistory([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [{ functionCall: { id: 'x', name: 't', args: {} } }], + }, + ]); + plantMarkers(chat); + expect(markers(chat).idx).toBe(0); + + chat.clearHistory(); + + expect(markers(chat).idx).toBeNull(); + expect(markers(chat).record).toBeNull(); + }); + + it('addHistory() clears the partial-push markers (violation path)', () => { + // addHistory is documented to be called between sends, NOT + // mid-send. Calling it with markers active is a violation — + // the implementation logs a warn so the offending caller is + // visible in diagnostics, then clears the markers. + chat.setHistory([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [{ functionCall: { id: 'x', name: 't', args: {} } }], + }, + ]); + plantMarkers(chat); + expect(markers(chat).idx).toBe(0); + + chat.addHistory({ role: 'user', parts: [{ text: 'between sends' }] }); + + expect(markers(chat).idx).toBeNull(); + expect(markers(chat).record).toBeNull(); + }); + + it('setHistory() clears the partial-push markers', () => { + chat.setHistory([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [{ functionCall: { id: 'x', name: 't', args: {} } }], + }, + ]); + plantMarkers(chat); + expect(markers(chat).idx).toBe(0); + + chat.setHistory([{ role: 'user', parts: [{ text: 'replacement' }] }]); + + expect(markers(chat).idx).toBeNull(); + expect(markers(chat).record).toBeNull(); + }); + + it('truncateHistory() clears the partial-push markers', () => { + chat.setHistory([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [{ functionCall: { id: 'x', name: 't', args: {} } }], + }, + ]); + plantMarkers(chat); + expect(markers(chat).idx).toBe(0); + + chat.truncateHistory(1); + + expect(markers(chat).idx).toBeNull(); + expect(markers(chat).record).toBeNull(); + }); + + it('stripThoughtsFromHistory() clears the partial-push markers', () => { + chat.setHistory([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [{ functionCall: { id: 'x', name: 't', args: {} } }], + }, + ]); + plantMarkers(chat); + expect(markers(chat).idx).toBe(0); + + chat.stripThoughtsFromHistory(); + + expect(markers(chat).idx).toBeNull(); + expect(markers(chat).record).toBeNull(); + }); + + it('stripOrphanedUserEntriesFromHistory() clears the partial-push markers', () => { + // History tail is a model turn — strip is a no-op on history, + // but the marker reset must still fire so all six mutation + // sites stay uniform. + chat.setHistory([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [{ functionCall: { id: 'x', name: 't', args: {} } }], + }, + ]); + plantMarkers(chat); + expect(markers(chat).idx).toBe(0); + + chat.stripOrphanedUserEntriesFromHistory(); + + expect(markers(chat).idx).toBeNull(); + expect(markers(chat).record).toBeNull(); + }); + }); + describe('repairOrphanedToolUseTurns', () => { // Verifies the inverse-of-strip pass: every `model[functionCall]` // without a matching `user[functionResponse]` in the next turn gets @@ -4265,6 +4512,138 @@ describe('GeminiChat', async () => { expect(history[2]!.parts![1]).toEqual({ text: 'never mind' }); expect(history[3]!.parts).toEqual([{ text: 'thanks anyway' }]); }); + + it('drops duplicate functionResponse entries for the same callId across user turns (gpt-5.5 thread on PR #4176)', () => { + // Critical regression: when the same callId is echoed back more + // than once (e.g. the React scheduler retries the late submitQuery + // after the orphan repair already planted one, or two parallel + // late-submit paths land), hoisting only the first leaves the + // duplicate behind. The wire payload then serializes + // `model[tool_use] -> user[tool_result] -> user[tool_result]` + // and Anthropic-compatible backends reject the trailing block as + // an orphan, re-wedging the session. The repair MUST hoist one + // canonical fr into the adjacent turn AND delete every duplicate. + chat.setHistory([ + { role: 'user', parts: [{ text: 'open file' }] }, + { + role: 'model', + parts: [ + { + functionCall: { id: 'cid_dup', name: 'read_file', args: {} }, + }, + ], + }, + { role: 'user', parts: [{ text: 'never mind' }] }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'cid_dup', + name: 'read_file', + response: { output: 'data' }, + }, + }, + ], + }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'cid_dup', + name: 'read_file', + response: { output: 'data' }, + }, + }, + ], + }, + ]); + + const result = chat.repairOrphanedToolUseTurns(); + + expect(result.injected).toEqual([]); + const history = chat.getHistory(); + // 5 → 3: both source turns held only the duplicate fr, so both + // are removed; the canonical fr is hoisted into history[2] and + // sits at the head before the text part. + expect(history.length).toBe(3); + expect(history[2]!.parts![0]!.functionResponse?.id).toBe('cid_dup'); + expect(history[2]!.parts![1]).toEqual({ text: 'never mind' }); + // No fr for cid_dup remains anywhere AFTER the adjacent turn. + const trailingHasDup = history + .slice(3) + .some((entry) => + (entry.parts ?? []).some( + (part) => part.functionResponse?.id === 'cid_dup', + ), + ); + expect(trailingHasDup).toBe(false); + }); + + it('drops duplicate fr even when the canonical copy is already in the adjacent turn', () => { + // Variant of the duplicate case where the FIRST fr lands in the + // immediate next user turn (no hoist needed) but a second + // duplicate copy is in a later user turn. The hoist branch is + // skipped, but duplicate cleanup must still fire — otherwise the + // wire payload still has two `tool_result` blocks for the same id. + chat.setHistory([ + { role: 'user', parts: [{ text: 'kick off' }] }, + { + role: 'model', + parts: [ + { + functionCall: { id: 'cid_adj_dup', name: 'read_file', args: {} }, + }, + ], + }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'cid_adj_dup', + name: 'read_file', + response: { output: 'real' }, + }, + }, + ], + }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'cid_adj_dup', + name: 'read_file', + response: { output: 'real' }, + }, + }, + { text: 'follow up' }, + ], + }, + ]); + + const result = chat.repairOrphanedToolUseTurns(); + + expect(result.injected).toEqual([]); + const history = chat.getHistory(); + // The source duplicate turn loses its fr but keeps its text part + // → 4 entries preserved, but the duplicate fr is gone. + expect(history.length).toBe(4); + expect(history[2]!.parts![0]!.functionResponse?.id).toBe('cid_adj_dup'); + expect(history[2]!.parts!.length).toBe(1); + expect(history[3]!.parts).toEqual([{ text: 'follow up' }]); + // The model[fc] is followed by exactly one fr for that id across + // all subsequent user turns. + const allFrIds = history + .slice(2) + .flatMap((entry) => + (entry.parts ?? []).map((p) => p.functionResponse?.id), + ) + .filter((id): id is string => Boolean(id)); + expect(allFrIds).toEqual(['cid_adj_dup']); + }); }); describe('output token recovery', () => { @@ -4470,6 +4849,104 @@ describe('GeminiChat', async () => { expect(lastEntry.parts!.length).toBeGreaterThan(0); }); + it('should pop both the partial model turn AND the recovery user message when recovery throws after a functionCall (qwen-latest-series-invite-beta-v34 thread on PR #4176)', async () => { + // Critical regression for the recovery catch's pop ordering. + // When the recovery stream yields a `functionCall` chunk and + // then throws, `processStreamResponse` pushes a partial `model` + // turn into history BEFORE re-throwing — so by the time the + // recovery catch runs, the trailing entries are + // [..., user(OUTPUT_RECOVERY_MESSAGE), model(partial fc)] + // The naive "if last is user, pop" check would no-op here (last + // is now `model`), leaving the OUTPUT_RECOVERY_MESSAGE control + // prompt stranded as a real user turn. The catch must pop the + // partial model turn FIRST, then the recovery user turn, and + // clear the partial-push markers so the outer `finally` JSONL + // flush doesn't resurrect the partial we just deleted. + const streams = [ + // Initial: text + MAX_TOKENS → triggers escalation. + makeStream([makeChunk([{ text: 'initial' }], 'MAX_TOKENS')]), + // Escalated: text + MAX_TOKENS → triggers recovery iteration 1. + makeStream([makeChunk([{ text: 'escalated' }], 'MAX_TOKENS')]), + // Recovery iter 1: yields functionCall chunk, then throws. + // processStreamResponse pushes a partial model turn before + // re-throwing the synthetic error. + (async function* () { + yield { + candidates: [ + { + content: { + parts: [ + { + functionCall: { + id: 'call_recovery_throw', + name: 'read_file', + args: { path: '/tmp/r.txt' }, + }, + }, + ], + }, + }, + ], + } as unknown as GenerateContentResponse; + throw new Error('synthetic recovery mid-tool_use cut'); + })(), + ]; + let callIndex = 0; + vi.mocked(mockContentGenerator.generateContentStream).mockImplementation( + async () => streams[callIndex++]!, + ); + + const stream = await chat.sendMessageStream( + 'gemini-3-pro', + { message: 'recovery throws after functionCall' }, + 'prompt-recovery-fc-throw', + ); + + // Consume; the catch swallows the error and emits a synthetic + // STOP chunk so the consumer sees a clean termination. + for await (const _ of stream) { + /* consume */ + } + + const history = chat.getHistory(); + + // OUTPUT_RECOVERY_MESSAGE must NOT appear anywhere in history. + // The pop-ordering bug strands it as a real user turn that then + // pollutes durable history and biases later turns. + const flattened = JSON.stringify(history); + expect(flattened).not.toContain('Output token limit hit'); + expect(flattened).not.toContain('Resume directly'); + + // The partial model[functionCall] from the recovery throw must + // also be popped — leaving it would create a dangling tool_use + // that the inline repair on the next sendMessageStream would + // synthesize an `error` functionResponse for, and the React + // scheduler's late real result would be dropped by the + // history-based dedup. Symptom: model sees an "execution result + // was not recorded" error for a tool that actually succeeded. + const stillHasPartialFc = history.some((entry) => + (entry.parts ?? []).some( + (part) => part.functionCall?.id === 'call_recovery_throw', + ), + ); + expect(stillHasPartialFc).toBe(false); + + // Roles must strictly alternate (no consecutive same-role) so + // providers don't reject the next turn. + for (let i = 1; i < history.length; i++) { + expect(history[i]!.role).not.toBe(history[i - 1]!.role); + } + + // History tail should be the escalated model response (text: + // 'escalated'), preserved as the user-visible answer. + const lastEntry = history[history.length - 1]!; + expect(lastEntry.role).toBe('model'); + const lastModelText = (lastEntry.parts ?? []) + .map((p) => ('text' in p ? ((p as { text?: string }).text ?? '') : '')) + .join(''); + expect(lastModelText).toContain('escalated'); + }); + it('should stop recovery mid-loop when a later iteration emits functionCall', async () => { // Covers the cross-iteration guard: iter 1 returns plain text (recovery // proceeds), iter 2 returns a functionCall (recovery must break before diff --git a/packages/core/src/core/geminiChat.ts b/packages/core/src/core/geminiChat.ts index 736c07bef6..6dcc6df874 100644 --- a/packages/core/src/core/geminiChat.ts +++ b/packages/core/src/core/geminiChat.ts @@ -486,9 +486,25 @@ export function repairOrphanedToolUseTurns( // they reject with "tool_use_id ... must have a corresponding // tool_use block in the previous message". gpt-5.5 review thread // on PR #4176. + // Collect EVERY (turnIdx, partIdx, part) for every functionResponse + // we encounter, keyed by id. Storing all locations (not just the + // first) is load-bearing for the duplicate case: if the same callId + // is echoed back more than once across the consecutive user turns + // (e.g. `model[fc id=cid], user[text], user[fr cid], user[fr cid]` + // — possible when the React scheduler retries the late submitQuery + // and a duplicate fr lands), hoisting only the first leaves the + // duplicate(s) behind in a non-adjacent later user turn. The wire + // payload then serializes + // `model[tool_use] -> user[tool_result] -> user[tool_result]` + // and the backend rejects the trailing block as an orphan + // ("tool_use_id ... must have a corresponding tool_use block in the + // previous message"), so the session stays wedged for the same 400 + // class this repair pass exists to escape. We MUST drop every + // duplicate; the survivor at history[i+1] is whichever copy we + // hoist, and the rest are erased. (gpt-5.5 thread on PR #4176.) const matched = new Map< string, - { turnIdx: number; partIdx: number; part: Part } + Array<{ turnIdx: number; partIdx: number; part: Part }> >(); let scanIdx = i + 1; while (scanIdx < history.length && history[scanIdx]?.role === 'user') { @@ -496,29 +512,55 @@ export function repairOrphanedToolUseTurns( for (let pIdx = 0; pIdx < parts.length; pIdx++) { const part = parts[pIdx]; const id = part.functionResponse?.id; - if (id && !matched.has(id)) { - matched.set(id, { turnIdx: scanIdx, partIdx: pIdx, part }); + if (id) { + const list = matched.get(id); + if (list) list.push({ turnIdx: scanIdx, partIdx: pIdx, part }); + else matched.set(id, [{ turnIdx: scanIdx, partIdx: pIdx, part }]); } } scanIdx++; } const synthesizeIds: Array<[string, string]> = []; - const hoistLocations: Array<{ - turnIdx: number; - partIdx: number; - part: Part; - }> = []; + const hoistedParts: Part[] = []; + // Removal targets cover BOTH: + // - the survivor's original location for ids being hoisted (we + // move it into history[i+1]) + // - every duplicate copy of an id (hoisted or already-adjacent) + // For an id whose canonical fr is already in history[i+1] but has + // duplicates further down, we don't hoist (no relocation needed) + // but we still must drop the duplicates so the wire format only + // contains one fr per call. + const allRemovalTargets: Array<{ turnIdx: number; partIdx: number }> = []; for (const [id, name] of expected) { - const loc = matched.get(id); - if (!loc) { + const locations = matched.get(id); + if (!locations || locations.length === 0) { synthesizeIds.push([id, name]); - } else if (loc.turnIdx !== i + 1) { - hoistLocations.push(loc); + continue; + } + // Pick the first location's part as the canonical survivor (any + // copy works — the response payload should be identical for the + // same callId; if they differ the wire is already corrupt and the + // backend will reject regardless). + const survivor = locations[0]!; + if (survivor.turnIdx !== i + 1) { + // Hoist: the canonical part needs to move into history[i+1]. + hoistedParts.push(survivor.part); + allRemovalTargets.push({ + turnIdx: survivor.turnIdx, + partIdx: survivor.partIdx, + }); + } + // else: canonical already adjacent — no relocation. + // Either way, drop EVERY duplicate beyond the first. + for (let k = 1; k < locations.length; k++) { + allRemovalTargets.push({ + turnIdx: locations[k]!.turnIdx, + partIdx: locations[k]!.partIdx, + }); } - // else: already in the immediate next user turn — wire format ok. } - if (synthesizeIds.length === 0 && hoistLocations.length === 0) continue; + if (synthesizeIds.length === 0 && allRemovalTargets.length === 0) continue; const syntheticParts: Part[] = synthesizeIds.map(([callId, name]) => ({ functionResponse: { @@ -527,23 +569,19 @@ export function repairOrphanedToolUseTurns( response: { error: reason }, }, })); - // Hoisted parts come from REAL tool_results elsewhere in history. - // We capture references first, then splice them out below in - // descending order so earlier removals don't shift later indices. - const hoistedParts: Part[] = hoistLocations.map((loc) => loc.part); // Synthetics first, then hoisted. Order between synthetic and // hoisted is internal — backends just need ALL tool_results at the // head of the user turn, regardless of order among themselves. const partsToInject: Part[] = [...syntheticParts, ...hoistedParts]; - // Remove hoisted parts from their original turns. Sort by - // (turnIdx desc, partIdx desc) so each splice operates on stable - // indices for everything still to be removed. - const removalOrder = [...hoistLocations].sort((a, b) => { + // Remove every removal target (hoist survivors + all duplicates). + // Sort by (turnIdx desc, partIdx desc) so each splice operates on + // stable indices for everything still to be removed. + allRemovalTargets.sort((a, b) => { if (a.turnIdx !== b.turnIdx) return b.turnIdx - a.turnIdx; return b.partIdx - a.partIdx; }); - for (const loc of removalOrder) { + for (const loc of allRemovalTargets) { const turnParts = history[loc.turnIdx].parts; if (turnParts) turnParts.splice(loc.partIdx, 1); } @@ -1357,7 +1395,18 @@ export class GeminiChat { break; } - // Other content validation errors (e.g. NO_FINISH_REASON). + // Currently unreachable for `InvalidStreamError`. The + // `isContentError` predicate is identical to + // `isTransientStreamError` (`error instanceof InvalidStreamError`), + // and the transient branch above already either continued or + // broke for that class. The branch is preserved as + // defense-in-depth: a future error class that should consume + // its own content-retry budget but NOT the transient one + // could be threaded through here without re-deriving the + // popPartialIfPushed sequence. Reviewer thread on PR #4176 + // (qwen-latest-series-invite-beta-v34) flagged the absence + // of a test — there is no reachable test path until the + // predicates diverge. const isContentError = error instanceof InvalidStreamError; if (isContentError) { if (attempt < INVALID_CONTENT_RETRY_OPTIONS.maxAttempts - 1) { @@ -1499,6 +1548,37 @@ export class GeminiChat { // If a recovery attempt fails (e.g., empty response, network // error), stop recovering and let the partial output stand. // Pop the dangling recovery message to keep history valid. + // + // Order matters: when the recovery stream errors AFTER + // yielding a `functionCall` chunk, `processStreamResponse` + // pushes a partial `model` turn into history before + // re-throwing. The naive "if last is user, pop" check + // would then no-op (last is now the partial `model`), + // leaving `user(OUTPUT_RECOVERY_MESSAGE)` stranded as a + // real user turn the user never sent. Two consequences: + // - the control-prompt text (which carries instructions + // meant only for the model's own continuation context) + // pollutes durable history and biases later turns, + // - the inline repair on the next sendMessageStream + // synthesizes an `error` `functionResponse` for the + // dangling `functionCall`, which the + // `handleCompletedTools` history-based dedup then drops + // when the React scheduler's REAL tool result arrives, + // so the model sees an "execution result was not + // recorded" error for a tool that actually succeeded. + // Pop the partial model turn FIRST, then the recovery + // user turn. The partial-push markers are also cleared + // in lockstep so the outer `finally` JSONL flush can't + // resurrect a partial we just deleted from live history. + // qwen-latest-series-invite-beta-v34 thread on PR #4176. + if ( + self.pendingPartialAssistantTurnIndex !== null && + self.history.length > 0 && + self.history[self.history.length - 1].role === 'model' + ) { + self.history.pop(); + self.clearPendingPartialState(); + } if ( self.history.length > 0 && self.history[self.history.length - 1].role === 'user' @@ -1585,9 +1665,28 @@ export class GeminiChat { // and stash are dropped together to preserve the // "marker non-null ⇔ stash non-null" invariant. if (self.pendingPartialAssistantRecord) { - self.chatRecordingService?.recordAssistantTurn( - self.pendingPartialAssistantRecord, - ); + // Recording-service errors (disk full, write permission, + // serialization failure) MUST NOT propagate out of the + // generator's `finally` — that would mask the real send + // outcome (success or original throw) with a JSONL-write + // error the caller can't usefully act on. Instead, log and + // drop the record: the partial is already durable in + // `this.history`, so live behavior is unaffected; only the + // disk transcript loses this turn (eventual consistency + // restored on the next successful flush of any other turn). + // qwen-latest-series-invite-beta-v34 thread on PR #4176. + try { + self.chatRecordingService?.recordAssistantTurn( + self.pendingPartialAssistantRecord, + ); + } catch (recordErr) { + debugLogger.warn( + '[PARTIAL_FLUSH] Failed to persist deferred JSONL record: ' + + (recordErr instanceof Error + ? recordErr.message + : String(recordErr)), + ); + } self.clearPendingPartialState(); } } @@ -1698,6 +1797,32 @@ export class GeminiChat { return this.history.length; } + /** + * Returns the set of `functionResponse.id` values present anywhere + * in user turns of the raw chat history. Walks `this.history` in + * place — no `structuredClone`, no per-part copy — so it is safe to + * call on hot paths (e.g. on every tool-completion batch in + * `useGeminiStream.handleCompletedTools`). + * + * The dedup pass only needs id strings; routing it through + * {@link getHistory} would deep-clone the entire conversation + * (recursive `structuredClone` over every part, including large tool + * outputs) on every batch and visibly stall the React UI thread on + * long sessions (200+ entries with sizable tool results). + * qwen-latest-series-invite-beta-v34 thread on PR #4176. + */ + getHistoryFunctionResponseIds(): Set { + const ids = new Set(); + for (const entry of this.history) { + if (entry.role !== 'user') continue; + for (const part of entry.parts ?? []) { + const id = part.functionResponse?.id; + if (id) ids.add(id); + } + } + return ids; + } + /** * Clears the chat history. */ @@ -2056,10 +2181,15 @@ export class GeminiChat { // re-issues it; a stale partial-text model turn between them would // either bias the retry or surface as a duplicate. if (streamError !== null) { - if ( - hasToolCall && - (thoughtContentPart || consolidatedHistoryParts.length > 0) - ) { + // Reuse the `willPersistToHistory` gate from the recordAssistantTurn + // block above instead of re-deriving it. When `streamError !== null`, + // `willPersistToHistory` reduces to exactly the original expression + // `hasToolCall && (thoughtContentPart || consolidatedHistoryParts.length > 0)`; + // sharing the single binding eliminates drift risk if one gate is + // tightened without the other and the JSONL recording silently + // desyncs from in-memory history. + // qwen-latest-series-invite-beta-v34 thread on PR #4176. + if (willPersistToHistory) { this.history.push({ role: 'model', parts: [