fix(core): close 3 review threads on partial-tool_use repair (PR #4176)

Three follow-ups from the qwen-latest-series-invite-beta-v28 pass on
commit fd12639c9:

[E] CORRECTNESS — `repairOrphanedToolUseTurns` now scans EVERY
consecutive `user` turn after a `model[functionCall]` (not just
`history[i+1]`) when building the matched-id set. The shape
`model[fc], user[text], user[fr_real]` arises when a user aborts a
long-running tool, types a follow-up text turn, and the React
scheduler's late `submitQuery` appends the real `tool_result` as a
SEPARATE user entry. Previously the repair saw only the immediate
follow-up text turn, found no matching fr, and synthesized a duplicate
`error` `functionResponse` for a callId that already had a real result
two turns down. The downstream dedup in
`useGeminiStream.handleCompletedTools` then dropped the REAL result on
the next submission (its callId was now "already in history" thanks to
the synthetic), and the model only ever saw the error placeholder —
silently swapping a real tool success for an error.

Two regression tests in `geminiChat.test.ts`:
  - "does NOT synthesize when the real functionResponse lives in a
    non-adjacent later user turn" — proves the forward-scan finds the
    real fr at history[i+2]+ and skips synthesis.
  - "still synthesizes for a partial mismatch when forward scan misses
    one callId" — counterpart that confirms parallel tool_use shapes
    where only some callIds have real fr's still get synthetic ones
    hoisted onto the immediate next user turn.

Verified via regression-injection (revert to scanning only
`history[i+1]`): the non-adjacent test fails with `injected =
[{callId:'call_nonadjacent_real',...}]` — proves the test catches the
bug, not just verifies a no-op.

[F] OBSERVABILITY — added a `[PARTIAL_PUSH]` `debugLogger.warn` at the
partial-turn push site in `processStreamResponse`. The repair lifecycle
already had push/pop/repair logging at the dedup and synthesis ends
(`[REPAIR] Dropping ...`, `[REPAIR] Synthesized ...`), but the original
PUSH event — root cause of every downstream recovery — was unlogged.
At 3 AM investigating a stale-partial wedge, the trace now anchors at
the exact moment the partial was created with pendingIndex, callIds,
and the originating error message.

[G] INVARIANT — `stripOrphanedUserEntriesFromHistory` now calls
`clearPendingPartialState()` after popping. Today this is safe even
without the reset (only trailing `user` entries are popped, which
can't shift the index of an earlier `model` partial), but every other
history-mutation method in the class — `clearHistory`, `addHistory`,
`setHistory`, `truncateHistory`, `stripThoughtsFromHistory` — now
clears the partial-push state in lockstep. Omitting it here would be
a silent exception to the uniform invariant the helper extraction
established. A future caller invoking this between the deferred JSONL
flush and the next `sendMessageStream` would otherwise leave a stale
marker that happens to line up with whatever model entry is at that
index in the meanwhile.

Tests: 239/239 core (+2 new), 95/95 useGeminiStream. tsc + eslint +
prettier clean. Latest CI run: all three platforms (Linux/macOS/
Windows) green.

packages/core/src/core/geminiChat.test.ts

@@ -4092,6 +4092,118 @@ describe('GeminiChat', async () => {
       const fr = chat.getHistory()[2]!.parts![0]!.functionResponse;
       expect((fr?.response as { error?: string })?.error).toBe('custom reason');
     });
+
+    it('does NOT synthesize when the real functionResponse lives in a non-adjacent later user turn', () => {
+      // Regression for the qwen-latest-series-invite-beta-v28 thread on
+      // PR #4176: shape `[user, model[fc], user[text], user[fr_real]]`
+      // arises when the user aborts a long-running tool, types a
+      // follow-up text turn, and then the React scheduler's late
+      // submitQuery appends the real tool_result as a SEPARATE user
+      // entry. The repair pass previously checked only history[i+1]
+      // (the immediate next turn) — the real fr at history[i+2] was
+      // invisible, so the repair synthesized a duplicate `error` fr for
+      // a callId that already had a real result. The downstream dedup
+      // in `useGeminiStream.handleCompletedTools` then dropped the REAL
+      // tool_result on the next submission (its callId was now
+      // "already in history" because of the synthetic), and the model
+      // only ever saw the error placeholder — silently swapping a real
+      // success for an error.
+      chat.setHistory([
+        { role: 'user', parts: [{ text: 'open /tmp/long.txt' }] },
+        {
+          role: 'model',
+          parts: [
+            {
+              functionCall: {
+                id: 'call_nonadjacent_real',
+                name: 'read_file',
+                args: { path: '/tmp/long.txt' },
+              },
+            },
+          ],
+        },
+        { role: 'user', parts: [{ text: 'never mind, do something else' }] },
+        {
+          role: 'user',
+          parts: [
+            {
+              functionResponse: {
+                id: 'call_nonadjacent_real',
+                name: 'read_file',
+                response: { output: 'real file contents' },
+              },
+            },
+          ],
+        },
+      ]);
+
+      const result = chat.repairOrphanedToolUseTurns();
+
+      // No synthesis: the real fr at history[3] satisfies the model[fc]
+      // at history[1].
+      expect(result.injected).toEqual([]);
+      const history = chat.getHistory();
+      // History shape is unchanged: 4 entries, no inserted synthetic.
+      expect(history.length).toBe(4);
+      expect(history[3]!.parts![0]!.functionResponse?.response).toEqual({
+        output: 'real file contents',
+      });
+      // The follow-up text turn is untouched (no synthetic fr hoisted
+      // into it — confirms the forward-scan fix, not just a no-op
+      // synthesis).
+      expect(history[2]!.parts).toEqual([
+        { text: 'never mind, do something else' },
+      ]);
+    });
+
+    it('still synthesizes for a partial mismatch when forward scan misses one callId', () => {
+      // Counterpart to the above: when the real fr only covers SOME
+      // of the callIds in a parallel tool_use, the missing ones must
+      // still get synthetic error fr's hoisted onto the immediate next
+      // user turn (preserving the existing parallel-tool_use behavior).
+      // Confirms the forward-scan didn't accidentally over-collect.
+      chat.setHistory([
+        { role: 'user', parts: [{ text: 'fan out two reads' }] },
+        {
+          role: 'model',
+          parts: [
+            {
+              functionCall: { id: 'cid_a', name: 'read_file', args: {} },
+            },
+            {
+              functionCall: { id: 'cid_b', name: 'read_file', args: {} },
+            },
+          ],
+        },
+        { role: 'user', parts: [{ text: 'follow up' }] },
+        {
+          role: 'user',
+          parts: [
+            {
+              functionResponse: {
+                id: 'cid_a',
+                name: 'read_file',
+                response: { output: 'real for a' },
+              },
+            },
+          ],
+        },
+      ]);
+
+      const result = chat.repairOrphanedToolUseTurns();
+
+      // cid_a satisfied by real fr at history[3]; cid_b missing →
+      // synthetic for cid_b only.
+      expect(result.injected).toEqual([{ callId: 'cid_b', name: 'read_file' }]);
+      const history = chat.getHistory();
+      expect(history.length).toBe(4);
+      // Synthetic for cid_b hoisted into history[2] (immediate next
+      // user turn) before the text part.
+      expect(history[2]!.parts![0]!.functionResponse?.id).toBe('cid_b');
+      expect(history[2]!.parts![1]).toEqual({ text: 'follow up' });
+      // Real fr for cid_a still at history[3], untouched.
+      expect(history[3]!.parts![0]!.functionResponse?.id).toBe('cid_a');
+    });
   });
 
   describe('output token recovery', () => {

packages/core/src/core/geminiChat.ts

@@ -441,18 +441,35 @@ export function repairOrphanedToolUseTurns(
     }
     if (expected.size === 0) continue;
 
-    const next = history[i + 1];
+    // Scan forward across EVERY consecutive user turn until the next
+    // non-user (model) entry, gathering all functionResponse ids. The
+    // real tool_result for this model[fc] may live in a non-adjacent
+    // user turn — common shape: user aborts a long-running tool, types
+    // a follow-up text turn, then the React scheduler's late
+    // submitQuery appends the real `user[fr]` as a SEPARATE turn,
+    // producing `model[fc], user[text], user[fr_real]`. Without
+    // forward scanning, `matched` would be empty (only `user[text]`
+    // visited), the repair would synthesize an `error` `functionResponse`
+    // for a callId that already has a real result downstream, and the
+    // `handleCompletedTools` dedup would then drop the REAL result on
+    // the next submission (callId already in history → submit skipped),
+    // so the model only ever sees the synthetic error placeholder.
+    // (qwen-latest-series-invite-beta-v28 thread on PR #4176.)
     const matched = new Set<string>();
-    if (next?.role === 'user') {
-      for (const part of next.parts ?? []) {
+    let scanIdx = i + 1;
+    while (scanIdx < history.length && history[scanIdx]?.role === 'user') {
+      for (const part of history[scanIdx].parts ?? []) {
         const id = part.functionResponse?.id;
         if (id) matched.add(id);
       }
+      scanIdx++;
     }
 
     const missing = [...expected.entries()].filter(([id]) => !matched.has(id));
     if (missing.length === 0) continue;
 
+    const next = history[i + 1];
+
     const syntheticParts: Part[] = missing.map(([callId, name]) => ({
       functionResponse: {
         id: callId,
@@ -1622,6 +1639,18 @@ export class GeminiChat {
     ) {
       this.history.pop();
     }
+    // Today this is safe even without the reset — only trailing user
+    // entries are popped, which can't shift the index of an earlier
+    // `model` partial. But every other history-mutation method now
+    // clears the partial-push state in lockstep
+    // (clearHistory/addHistory/setHistory/truncateHistory/
+    // stripThoughtsFromHistory), so omitting it here would be a silent
+    // exception to the uniform invariant: a future caller invoking
+    // this method between the deferred JSONL flush and the next
+    // `sendMessageStream` would otherwise leave a stale marker that
+    // happens to line up with whatever model entry is at that index
+    // in the meanwhile.
+    this.clearPendingPartialState();
   }
 
   /**
@@ -1886,6 +1915,28 @@ export class GeminiChat {
         // partial `model[functionCall]` as a stale leading model turn in
         // front of the retry's real response.
         this.pendingPartialAssistantTurnIndex = this.history.length - 1;
+        // Trace the push event so the lifecycle is observable end-to-end:
+        // dedup in `useGeminiStream.handleCompletedTools` already logs
+        // `[REPAIR] Dropping ...`, and `repairOrphanedToolUseTurnsInHistory`
+        // logs `[REPAIR] Synthesized ...`. Without a corresponding
+        // `[PARTIAL_PUSH]` line here, an investigator looking at a
+        // stale-partial wedge sees the downstream symptom but has no
+        // anchor for when/why the partial originated.
+        debugLogger.warn(
+          '[PARTIAL_PUSH] Persisting partial assistant turn for ' +
+            'mid-stream error recovery (will be rolled back if retry ' +
+            'succeeds, kept if break is unretryable). ' +
+            `pendingIndex=${this.pendingPartialAssistantTurnIndex} ` +
+            `callIds=${consolidatedHistoryParts
+              .map((p) => p.functionCall?.id)
+              .filter((id): id is string => Boolean(id))
+              .join(',')} ` +
+            `error=${
+              streamError instanceof Error
+                ? streamError.message
+                : String(streamError)
+            }`,
+        );
       }
       throw streamError;
     }