vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-08-31 10:16:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 15

fix(core,cli): close tool_use↔tool_result invariant at failure points

Browse source 
Post-review audit of 3de3241a2 surfaced two real races the earlier fix
did not actually close:

  - Race A (the one yiliang114 originally flagged): handleCompletedTools
    is gated on `isResponding` (useGeminiStream:1971) BEFORE the dedup
    branch ran, so when the user Ctrl+Y'd mid-tool, the dedup never
    fired and the in-flight tool stayed permanently stuck in
    `completed-but-not-submitted` (the scheduler's
    `allToolCallsCompleteHandler` is single-shot). The dedup branch was
    structurally unreachable on the very race it was meant to defend
    against.

  - Race in the Retry path: the repair pass in
    `client.sendMessageStream` Retry branch ran BEFORE the chat-internal
    `push(userContent)`, so a Retry of a previous ToolResult submission
    (lastPrompt is a functionResponse part array) raced its own real
    `functionResponse` against the synthesized error one. The synthesis
    was winning and pre-empting the real result, producing two
    `functionResponse` entries with the same callId on the wire.

This commit relocates both pieces to where the contract actually holds:

1. Move the dedup branch in `handleCompletedTools` to BEFORE the
   `isResponding` early-return so `markToolsAsSubmitted` always runs
   for callIds already paired in history, unblocking the UI/scheduler
   even when a new stream is in flight. The submission-side `geminiTools
   = completedAndReadyToSubmitTools.filter(... && !inHistory ...)` then
   covers the no-double-submit case in one expression.

2. Move the repair call from `client.sendMessageStream` Retry branch
   into `chat.sendMessageStream` immediately AFTER the user-supplied
   turn is pushed. The user's own tool_result (when the Retry payload
   carries one) gets the first chance to close the pair before the
   synthesizer sees it as dangling. `client.startChat()` keeps a
   belt-and-suspenders pass at session-load time so any pre-send code
   reading `chat.history` sees a well-formed shape.

Adds three integration tests covering the new wiring:

  - geminiChat.test.ts: chat.sendMessageStream synthesizes a
    functionResponse when history carries a dangling tool_use AND the
    user-supplied content doesn't already close it (Race B/C plus
    Race A from the chat side).
  - geminiChat.test.ts: chat.sendMessageStream does NOT synthesize
    when the user-supplied content IS a matching functionResponse
    (the Retry-of-ToolResult race the previous wiring tripped on).
  - useGeminiStream.test.tsx: handleCompletedTools dedups a late real
    result whose callId already has a functionResponse in chat.history
    (Race A end-to-end: markToolsAsSubmitted fires but no
    submitQuery is dispatched).

Test summary: 90/90 geminiChat, 131/131 client, 92/92 useGeminiStream;
all 8186 core tests pass; tsc clean.

Known limitation logged for follow-up: a Retry of ToolResult whose
intervening stream itself partial-pushed a SECOND tool_use produces a
trailing user turn with both the stale re-pushed `fr_A` and the
synthesized `fr_B` for the second call — `fr_A` retry lands as an
orphan because `fc_A` was already paired earlier in history. Triggered
only when partial-push fires on the second stream AND user retries the
tool_result rather than the user prompt; extremely low frequency in
practice. Cleanest fix is in the retryLastPrompt layer (don't re-push
an already-paired tool_result), which is out of scope for this PR.
This commit is contained in:
wenshao 2026-05-15 22:45:49 +08:00
parent 3de3241a2c
commit b2fed61cdd
5 changed files with 337 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

122
packages/cli/src/ui/hooks/useGeminiStream.test.tsx
View file

@ -714,6 +714,128 @@ describe('useGeminiStream', () => {
});
});
it('drops a late tool result whose callId is already paired in chat.history (Race A dedup)', async () => {
// Race A repro: the chat-internal repair pass already synthesized a
// functionResponse for this callId on the Retry push (because the
// partial-tool_use turn was orphan when Ctrl+Y landed). The live
// scheduler's late real result must NOT also be submitted, otherwise
// the wire payload would carry two functionResponse parts for the
// same callId and the second one would land as an orphan tool_result.
// The dedup MUST run regardless of `isResponding`, because the
// scheduler's `onAllToolCallsComplete` is single-shot and would
// otherwise leave the tool stuck in `completed-but-not-submitted`.
const lateRealResult: TrackedCompletedToolCall = {
request: {
callId: 'call_race_A',
name: 'read_file',
args: { path: '/tmp/x.txt' },
isClientInitiated: false,
prompt_id: 'prompt-race-a',
},
status: 'success',
responseSubmittedToGemini: false,
response: {
callId: 'call_race_A',
responseParts: [
{
functionResponse: {
id: 'call_race_A',
name: 'read_file',
response: { output: 'real file contents' },
},
},
],
errorType: undefined,
},
tool: { displayName: 'ReadFile' },
invocation: {
getDescription: () => 'read /tmp/x.txt',
} as unknown as AnyToolInvocation,
} as TrackedCompletedToolCall;
const client = new MockedGeminiClientClass(mockConfig);
// Simulate the chat-internal repair pass having already planted a
// synthetic functionResponse for the same callId on the previous
// (Retry) push.
client.getHistory = vi.fn().mockReturnValue([
{ role: 'user', parts: [{ text: 'open /tmp/x.txt' }] },
{
role: 'model',
parts: [
{
functionCall: {
id: 'call_race_A',
name: 'read_file',
args: { path: '/tmp/x.txt' },
},
},
],
},
{
role: 'user',
parts: [
{ text: 'retry' },
{
functionResponse: {
id: 'call_race_A',
name: 'read_file',
response: {
error: 'Tool execution result was not recorded',
},
},
},
],
},
]);
let capturedOnComplete:
| ((completedTools: TrackedToolCall[]) => Promise<void>)
| null = null;
mockUseReactToolScheduler.mockImplementation((onComplete) => {
capturedOnComplete = onComplete;
return [[], mockScheduleToolCalls, mockMarkToolsAsSubmitted];
});
renderHook(() =>
useGeminiStream(
client,
[],
mockAddItem,
mockConfig,
mockLoadedSettings,
mockOnDebugMessage,
mockHandleSlashCommand,
false,
() => 'vscode' as EditorType,
() => {},
() => Promise.resolve(),
false,
() => {},
() => {},
() => {},
() => {},
80,
24,
),
);
await act(async () => {
if (capturedOnComplete) {
await capturedOnComplete([lateRealResult]);
}
});
await waitFor(() => {
// The dedup hit must `markToolsAsSubmitted` so the UI/scheduler is
// unblocked even though we drop the real result on the wire.
expect(mockMarkToolsAsSubmitted).toHaveBeenCalledWith(['call_race_A']);
});
// No follow-up submission: the synthetic in history already closes
// the tool_use ↔ tool_result pair.
expect(mockSendMessageStream).not.toHaveBeenCalled();
});
it('should not flicker streaming state to Idle between tool completion and submission', async () => {
const toolCallResponseParts: PartListUnion = [
{ text: 'tool 1 final response' },

100
packages/cli/src/ui/hooks/useGeminiStream.ts
View file

@ -1968,10 +1968,6 @@ export const useGeminiStream = (
const handleCompletedTools = useCallback(
async (completedToolCallsFromScheduler: TrackedToolCall[]) => {
if (isResponding) {
return;
}
const completedAndReadyToSubmitTools =
completedToolCallsFromScheduler.filter(
(
@ -1994,6 +1990,49 @@ export const useGeminiStream = (
},
);
// History-based dedup MUST run before the `isResponding` early-return.
// If a synthetic `functionResponse` for this callId is already in
// chat.history (planted on session-load by
// `client.repairOrphanedToolUseTurnsInHistory` or on every
// `chat.sendMessageStream` push by the inline repair pass), the
// in-flight scheduler result must be marked submitted NOW —
// `useReactToolScheduler.allToolCallsCompleteHandler` is single-shot
// per batch, so a later isResponding=true early-return would leave
// the tool stuck in `completed-but-not-submitted` forever (Race A
// surfaced in PR #4176 review). The real result is dropped on the
// wire — same trade-off upstream Claude Code makes when its
// `StreamingToolExecutor.discard()` follows a
// `yieldMissingToolResultBlocks` synthesis (`query.ts:733` + `:984`).
const historyCallIdsWithResponse = new Set<string>();
// Guard the call: some test harnesses build a partial GeminiClient
// mock without `getHistory`. Skipping dedup in that case is safe —
// it just means tests that never set up the repair pre-condition
// run with the original (pre-dedup) submission shape.
if (geminiClient && typeof geminiClient.getHistory === 'function') {
for (const entry of geminiClient.getHistory()) {
if (entry.role !== 'user') continue;
for (const part of entry.parts ?? []) {
const id = part.functionResponse?.id;
if (id) historyCallIdsWithResponse.add(id);
}
}
}
const dedupedCallIds = completedAndReadyToSubmitTools
.filter((tc) => historyCallIdsWithResponse.has(tc.request.callId))
.map((tc) => tc.request.callId);
if (dedupedCallIds.length > 0) {
debugLogger.warn(
`[REPAIR] Dropping ${dedupedCallIds.length} late tool result(s) ` +
`whose callId already has a functionResponse in history: ` +
`${dedupedCallIds.join(', ')}`,
);
markToolsAsSubmitted(dedupedCallIds);
}
if (isResponding) {
return;
}
// Finalize any client-initiated tools as soon as they are done.
const clientTools = completedAndReadyToSubmitTools.filter(
(t) => t.request.isClientInitiated,
@ -2019,62 +2058,19 @@ export const useGeminiStream = (
);
}
const geminiToolsRaw = completedAndReadyToSubmitTools.filter(
(t) => !t.request.isClientInitiated,
const geminiTools = completedAndReadyToSubmitTools.filter(
(t) =>
!t.request.isClientInitiated &&
!historyCallIdsWithResponse.has(t.request.callId),
);
for (const toolCall of geminiToolsRaw) {
for (const toolCall of geminiTools) {
geminiClient?.recordCompletedToolCall(
toolCall.request.name,
toolCall.request.args as Record<string, unknown>,
);
}
// History-based dedup: if a synthetic `functionResponse` for this
// callId is already in chat.history (planted by
// `client.repairOrphanedToolUseTurnsInHistory()` on session-load or
// Retry), the in-flight scheduler result would land as a duplicate
// `tool_result` and produce two consecutive user turns where the
// second is orphaned (no preceding tool_use — the synthetic ate it).
//
// For dedup hits: mark the tool as submitted so the UI advances and
// `useReactToolScheduler.allToolCallsCompleteHandler` (single-shot)
// doesn't leave the call permanently stuck in `completed-but-not-
// submitted`. The real result is dropped on the wire — same trade-off
// upstream Claude Code makes when its `StreamingToolExecutor.discard()`
// is followed by a `yieldMissingToolResultBlocks` synthesis
// (`query.ts:733` + `:984`). The model sees the synthetic error and
// can retry the tool if it still wants the result.
const historyCallIdsWithResponse = new Set<string>();
// Guard the call: some test harnesses build a partial GeminiClient
// mock without `getHistory`. Skipping dedup in that case is safe —
// it just means tests that never set up the repair pre-condition
// run with the original (pre-dedup) submission shape.
if (geminiClient && typeof geminiClient.getHistory === 'function') {
for (const entry of geminiClient.getHistory()) {
if (entry.role !== 'user') continue;
for (const part of entry.parts ?? []) {
const id = part.functionResponse?.id;
if (id) historyCallIdsWithResponse.add(id);
}
}
}
const dedupedCallIds = geminiToolsRaw
.filter((tc) => historyCallIdsWithResponse.has(tc.request.callId))
.map((tc) => tc.request.callId);
if (dedupedCallIds.length > 0) {
debugLogger.warn(
`[REPAIR] Dropping ${dedupedCallIds.length} late tool result(s) ` +
`whose callId already has a synthetic functionResponse in ` +
`history: ${dedupedCallIds.join(', ')}`,
);
markToolsAsSubmitted(dedupedCallIds);
}
const geminiTools = geminiToolsRaw.filter(
(tc) => !historyCallIdsWithResponse.has(tc.request.callId),
);
if (geminiTools.length === 0) {
return;
}

34
packages/core/src/core/client.ts
View file

@ -700,9 +700,14 @@ export class GeminiClient {
// partial-tool_use push (see `processStreamResponse`) and the React
// scheduler's tool_result submission. Without this pass, the first
// API call on a resumed session would 400 with the same
// `tool_use_id ... corresponding tool_use` error this whole subsystem
// is trying to escape.
this.chat.repairOrphanedToolUseTurns();
// `tool_use_id ... corresponding tool_use` error this whole
// subsystem is trying to escape. (Belt-and-suspenders: the same
// helper runs again inside `chat.sendMessageStream` after the user
// content is pushed, so a dangling left here by setHistory /
// compaction reordering is also caught — but doing it here keeps
// any pre-send code reading `chat.history` from seeing a malformed
// shape.)
this.repairOrphanedToolUseTurnsInHistory();
const sessionStartAdditionalContext =
await this.fireSessionStartHook(sessionStartSource);
@ -1091,22 +1096,13 @@ export class GeminiClient {
if (messageType === SendMessageType.Retry) {
this.stripOrphanedUserEntriesFromHistory();
// Close any dangling `model[functionCall]` whose tool_result never
// landed before composing the retry payload. Ctrl+Y race: the user
// retried while a tool was still running on a partial-tool_use turn
// pushed by `processStreamResponse`'s mid-stream error path. The
// scheduler's `onAllToolCallsComplete` is single-shot and gated on
// `isResponding` (`useGeminiStream:1971`), so the eventual
// `tool_result` would otherwise be silently swallowed and the next
// API call would 400 with "tool_use_id ... corresponding tool_use"
// anyway. The synthesized `error` `functionResponse` keeps the wire
// invariant intact; the live scheduler dedupes against history in
// `handleCompletedTools` before submitting its real result so the
// synthetic doesn't collide with a late real one.
//
// Restricted to the Retry branch to mirror `stripOrphanedUserEntries`
// scope. Crash-resume's path is covered separately in `startChat()`.
this.repairOrphanedToolUseTurnsInHistory();
// The matching dangling-`functionCall` repair runs inside
// `chat.sendMessageStream` AFTER the user content is pushed, so any
// tool_result the user is supplying (Retry of a ToolResult
// submission, lastPrompt === fr parts) closes the pair via the real
// `functionResponse` before we synthesize an error one. Doing the
// repair here would happen pre-push and race against the user
// content's own pairing — see PR #4176 review for the corner.
}
// Fire UserPromptSubmit hook through MessageBus (only if hooks are enabled)

132
packages/core/src/core/geminiChat.test.ts
View file

@ -597,6 +597,138 @@ describe('GeminiChat', async () => {
'This is the visible text that should not be lost.',
);
});
it('synthesizes a functionResponse for a dangling tool_use before sending', async () => {
// End-to-end: when sendMessageStream is invoked on a chat whose
// history carries a dangling `model[functionCall]` (typical state
// after a Ctrl+Y race or a crash-resume on a partial-tool_use
// turn), the inline repair pass closes the pair against the
// just-pushed user content so the wire payload doesn't 400 with
// "tool_use_id ... corresponding tool_use".
chat.setHistory([
{ role: 'user', parts: [{ text: 'first message' }] },
{
role: 'model',
parts: [
{
functionCall: {
id: 'call_dangling_for_send',
name: 'read_file',
args: { path: '/tmp/x' },
},
},
],
},
]);
const ackStream = (async function* () {
yield {
candidates: [
{
content: { role: 'model', parts: [{ text: 'ok' }] },
finishReason: 'STOP',
},
],
} as unknown as GenerateContentResponse;
})();
vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue(
ackStream,
);
const stream = await chat.sendMessageStream(
'test-model',
{ message: 'next user prompt after a stream-error-mid-tool_use' },
'prompt-send-repair',
);
for await (const _ of stream) {
/* drain */
}
const history = chat.getHistory();
// The dangling fc should now be followed by a user turn that
// carries both the user-supplied text AND the synthetic fr that
// closes the pair.
const userTurn = history[2]!;
expect(userTurn.role).toBe('user');
const fr = userTurn.parts!.find((p) => p.functionResponse);
expect(fr?.functionResponse?.id).toBe('call_dangling_for_send');
expect(fr?.functionResponse?.name).toBe('read_file');
expect(
(fr?.functionResponse?.response as { error?: string })?.error,
).toMatch(/interrupted/i);
// The user's own text part is still present.
expect(
userTurn.parts!.some(
(p) =>
p.text === 'next user prompt after a stream-error-mid-tool_use',
),
).toBe(true);
});
it('does NOT synthesize when the user supplies a matching tool_result', async () => {
// Retry-of-ToolResult case (lastPrompt is a functionResponse Part
// array): the user-supplied tool_result must close the pair before
// the inline repair pass sees it, so no synthetic error is
// injected. Otherwise the wire payload would carry two
// functionResponse parts for the same callId — the real one and a
// bogus synthetic.
chat.setHistory([
{ role: 'user', parts: [{ text: 'do the read' }] },
{
role: 'model',
parts: [
{
functionCall: {
id: 'call_retry_real_fr',
name: 'read_file',
args: { path: '/tmp/y' },
},
},
],
},
]);
const ackStream = (async function* () {
yield {
candidates: [
{
content: { role: 'model', parts: [{ text: 'ack' }] },
finishReason: 'STOP',
},
],
} as unknown as GenerateContentResponse;
})();
vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue(
ackStream,
);
const stream = await chat.sendMessageStream(
'test-model',
{
message: {
functionResponse: {
id: 'call_retry_real_fr',
name: 'read_file',
response: { output: 'real-tool-output' },
},
},
},
'prompt-retry-real-fr',
);
for await (const _ of stream) {
/* drain */
}
const userTurn = chat.getHistory()[2]!;
const frParts = userTurn.parts!.filter((p) => p.functionResponse);
// Exactly ONE functionResponse — the real one. No synthetic.
expect(frParts.length).toBe(1);
expect(frParts[0]!.functionResponse?.id).toBe('call_retry_real_fr');
expect(
(frParts[0]!.functionResponse?.response as { output?: string })?.output,
).toBe('real-tool-output');
});
it('should throw an error when a tool call is followed by an empty stream response', async () => {
vi.useFakeTimers();
try {

20
packages/core/src/core/geminiChat.ts
View file

@ -728,6 +728,26 @@ export class GeminiChat {
// Add user content to history ONCE before any attempts.
this.history.push(userContent);
userContentAdded = true;
// Close any dangling `model[functionCall]` whose `functionResponse`
// never landed by the time we compose the request. Runs AFTER the
// user-supplied turn lands so a tool_result the user is supplying
// gets the first chance to close the pair before we synthesize an
// `error` `functionResponse`. Covers:
// - Stream errored mid-tool_use (partial assistant push left a
// dangling functionCall), then the React scheduler's eventual
// tool_result lost the race against a Ctrl+Y retry whose
// onAllToolCallsComplete fired into `isResponding=true` and
// skipped submission.
// - The same shape from a process crash / OOM mid-flight (the
// transcript JSONL preserves the dangling model[fc] across
// `--resume`; `startChat()` calls this once on load, but a
// belt-and-suspenders pass here covers anything that slipped
// past — including dangling shapes the load-time repair didn't
// visit because compaction / setHistory ran after it).
// The React scheduler's late real result is then dedup'd against
// chat.history in `useGeminiStream.handleCompletedTools` so the
// synthetic doesn't collide with it on the wire.
repairOrphanedToolUseTurns(this.history);
requestContents = this.getHistory(true);
} catch (error) {
if (userContentAdded) {