From c8fa3143f3f454c1b5f9f835858e5dcbcfd76ea6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=AB=98=E9=93=81?= Date: Wed, 20 May 2026 13:15:09 +0800 Subject: [PATCH] fix(core,cli): close 10 review threads on partial-tool_use repair (PR #4176) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Five Critical + four Suggestion threads from the gpt-5.5 + qwen-latest-series-invite-beta-v34 passes on commit ce68749b5, plus one N/A documented in code/test comments where the requested test target is provably unreachable. [A] CORRECTNESS — `repairOrphanedToolUseTurns` now collects EVERY location for each `functionResponse` id, not just the first. Previously the `matched` Map stored only the first occurrence per callId, so when the same id was echoed back more than once across the consecutive user turns (shape `model[fc id=cid], user[text], user[fr cid], user[fr cid]` — possible when 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 left a duplicate behind. The wire payload then serialized `model[tool_use] -> user[tool_result] -> user[tool_result]`, the backend rejected the trailing block as an orphan, and the session stayed wedged for the same 400 class this whole repair pass exists to escape. Fix: track all locations per id; pick the first as the canonical survivor (hoisted into history[i+1] when non-adjacent, or left in place when adjacent), and drop EVERY duplicate. New tests: "drops duplicate functionResponse entries for the same callId across user turns" (canonical non-adjacent + duplicate further down → 3-entry result), "drops duplicate fr even when the canonical copy is already in the adjacent turn" (no hoist needed but duplicate cleanup must still fire). gpt-5.5 review thread on PR #4176. [B] CORRECTNESS — Max-tokens recovery catch (`recoveryError` block) now pops the partial `model` turn FIRST, then the OUTPUT_RECOVERY_MESSAGE user turn. Before the fix, when a recovery stream errored AFTER yielding a `functionCall` chunk, `processStreamResponse` pushed a partial `model` turn into history before re-throwing — so by the time the catch ran, the trailing entries were `[..., user(OUTPUT_RECOVERY_MESSAGE), model(partial fc)]` and the naive `if last is user, pop` check no-op'd, stranding the control prompt as a real user turn. Two consequences: the recovery prompt's instructions polluted later turns, and the inline repair on the next sendMessageStream synthesized an `error` `functionResponse` for the dangling `functionCall` — which the dedup then dropped when the React scheduler's REAL tool result arrived, so the model saw an "execution result was not recorded" error for a tool that actually succeeded. Fix: pop the partial model turn, clear the partial-push markers, then pop the recovery user turn. New regression test: "should pop both the partial model turn AND the recovery user message when recovery throws after a functionCall". qwen-latest-series-invite-beta-v34 review thread. [C] PERFORMANCE — added `getHistoryFunctionResponseIds(): Set` to `GeminiChat` and a wrapper on `GeminiClient`. The dedup pass in `useGeminiStream.handleCompletedTools` previously called `geminiClient.getHistory()`, which returns `structuredClone(this.history)` — a recursive deep clone of every part including large tool outputs. On long sessions (200+ entries) running on every tool-completion batch, this caused multi-millisecond stalls on the React UI thread during streaming. The new accessor walks `this.history` in place and only collects the id strings the dedup actually needs. `useGeminiStream` prefers it; falls back to the cloning getHistory() for older mocks that don't expose it (tests stay green). qwen-latest-series-invite-beta-v34 review thread. [D] ROBUSTNESS — wrapped the `finally`-block JSONL flush in try/catch. 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. On failure the partial is dropped from JSONL but stays durable in `this.history`, so live behavior is unaffected; eventual consistency is restored on the next successful flush. qwen-latest-series-invite-beta-v34 review thread. [E] TEST COVERAGE — added an InvalidStreamError transient-retry rollback test exercising `popPartialIfPushed()` on the NO_FINISH_REASON / NO_RESPONSE_TEXT path (separate budget from the already-tested rate-limit branch). Stream yields a functionCall then throws InvalidStreamError; retry succeeds; assert no orphan functionCall remains in history. qwen-latest-series-invite-beta-v34 review thread. [F] N/A WITH RATIONALE — verified that the InvalidStreamError content-retry branch (geminiChat.ts ~line 1399) is unreachable for that error class: `isTransientStreamError` and `isContentError` are the same predicate (`error instanceof InvalidStreamError`), so the transient branch above always either continues or breaks before control reaches the content branch. The `popPartialIfPushed()` call is preserved as defense-in-depth for a future error class that should diverge the predicates. Documented at the call site (code comment) and at the would-be test site (test-block comment) so future readers see the unreachability analysis instead of repeating the review-comment ask. [G] DRIFT GUARD — `recordAssistantTurn` and `this.history.push` gates now share the single `willPersistToHistory` binding instead of re-deriving the same `hasToolCall && (thoughtContentPart || ...)` expression. Drift risk: tightening one without the other would silently desync the JSONL transcript from in-memory history. qwen-latest-series-invite-beta-v34 review thread. [H] DOUBLE-DISPATCH — `clientTools` filter in `useGeminiStream.handleCompletedTools` now excludes callIds whose fr is already in history. Previously a deduped client-initiated tool would have `markToolsAsSubmitted` called twice (once in the dedup block, once in the clientTools block), triggering an extra React render cycle. qwen-latest-series-invite-beta-v34 review thread. [I] TEST COVERAGE — added six tests under "partial-push marker invariants on history mutation" pinning that each of the six mutation methods (clearHistory, addHistory, setHistory, truncateHistory, stripThoughtsFromHistory, stripOrphanedUserEntriesFromHistory) calls clearPendingPartialState() in lockstep. Markers planted directly via private-field assignment (the lifecycle would otherwise clear them in finally before the test could observe them). Future refactors that drop a clearPendingPartialState() call from one site are now caught at test time rather than as a silent stale-partial wedge in production. qwen-latest-series-invite-beta-v34 review thread. [J] TEST COVERAGE — added a mixed-batch dedup test in useGeminiStream.test.tsx: scheduler completes two tools in the same batch, one whose callId already has an fr in history (deduped), one fresh. Asserts (a) markToolsAsSubmitted called with both, (b) recordCompletedToolCall fires exactly once per tool (deduped via the dedup-loop, fresh via the geminiTools loop — no double-record on the deduped id), (c) sendMessageStream IS called for the fresh tool's real result. The `!historyCallIdsWithResponse.has(callId)` filter on `geminiTools` is the only guard against double-counting telemetry; existing tests supplied only deduped tools. qwen-latest-series-invite-beta-v34 review thread. Tests: 347/347 (118 in geminiChat.test.ts, 96 in useGeminiStream.test.tsx, 133 in client.test.ts; net +8 new tests). tsc + eslint + prettier clean on changed files. --- .../cli/src/ui/hooks/useGeminiStream.test.tsx | 190 +++++++ packages/cli/src/ui/hooks/useGeminiStream.ts | 43 +- packages/core/src/core/client.ts | 16 + packages/core/src/core/geminiChat.test.ts | 477 ++++++++++++++++++ packages/core/src/core/geminiChat.ts | 192 +++++-- 5 files changed, 880 insertions(+), 38 deletions(-) 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: [