diff --git a/packages/core/src/core/geminiChat.test.ts b/packages/core/src/core/geminiChat.test.ts index 248736dc9ba..944d0eeb7f3 100644 --- a/packages/core/src/core/geminiChat.test.ts +++ b/packages/core/src/core/geminiChat.test.ts @@ -2715,6 +2715,47 @@ describe('GeminiChat', async () => { ); }); + it('excludes exact degraded placeholders without dropping legitimate mentions or tool calls', () => { + const toolCall = { + functionCall: { id: 'call-1', name: 'read_file', args: {} }, + }; + const functionResponse = { + functionResponse: { + id: 'call-1', + name: 'read_file', + response: { output: 'ok' }, + }, + }; + const history: Content[] = [ + { role: 'user', parts: [{ text: 'what happened?' }] }, + { + role: 'model', + parts: [{ text: 'The endpoint returned (request timeout) once.' }], + }, + { role: 'user', parts: [{ text: 'continue' }] }, + { role: 'model', parts: [{ text: ' (request timeout) ' }] }, + { role: 'user', parts: [{ text: 'try again' }] }, + { + role: 'model', + parts: [{ text: '(request timeout)' }, toolCall], + }, + { role: 'user', parts: [functionResponse] }, + ]; + chat.setHistory(history); + + expect(chat.getHistory(true)).toEqual([ + history[0], + history[1], + { + role: 'user', + parts: [{ text: 'continue' }, { text: 'try again' }], + }, + history[5], + history[6], + ]); + expect(chat.getHistory()).toEqual(history); + }); + it('should not update global telemetry when no telemetryService is provided (subagent isolation)', async () => { // Simulate a subagent GeminiChat: created without a telemetryService const subagentChat = new GeminiChat(mockConfig, config, []); @@ -5927,6 +5968,120 @@ describe('GeminiChat', async () => { } }); + it('retries a split degraded placeholder without yielding or persisting it', async () => { + vi.useFakeTimers(); + try { + vi.mocked(mockContentGenerator.generateContentStream) + .mockResolvedValueOnce( + streamResponse( + { + candidates: [{ content: { parts: [{ text: '(request ' }] } }], + } as unknown as GenerateContentResponse, + stopResponse([{ text: 'timeout)' }]), + ), + ) + .mockResolvedValueOnce( + streamResponse(stopResponse([{ text: 'Recovered response' }])), + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + 'prompt-id-degraded-placeholder', + ); + const events = await collectStreamWithFakeTimers(stream); + const emitted = events + .filter((event) => event.type === StreamEventType.CHUNK) + .flatMap( + (event) => + event.value.candidates?.[0]?.content?.parts?.map( + (part) => part.text, + ) ?? [], + ); + + expect(emitted).toEqual(['Recovered response']); + expect( + mockContentGenerator.generateContentStream, + ).toHaveBeenCalledTimes(2); + expect(mockLogContentRetry).toHaveBeenCalledWith( + mockConfig, + expect.objectContaining({ + error_type: 'UPSTREAM_DEGRADED_RESPONSE', + }), + ); + expect(chat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'test' }] }, + { role: 'model', parts: [{ text: 'Recovered response' }] }, + ]); + } finally { + vi.useRealTimers(); + } + }); + + it('passes through longer text that mentions the placeholder', async () => { + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + streamResponse( + stopResponse([ + { text: 'The endpoint returned (request timeout) once.' }, + ]), + ), + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + 'prompt-id-placeholder-mention', + ); + const events: StreamEvent[] = []; + for await (const event of stream) events.push(event); + + expect( + events.some( + (event) => + event.type === StreamEventType.CHUNK && + event.value.candidates?.[0]?.content?.parts?.[0]?.text === + 'The endpoint returned (request timeout) once.', + ), + ).toBe(true); + expect(mockContentGenerator.generateContentStream).toHaveBeenCalledTimes( + 1, + ); + }); + + it('does not reject a placeholder turn that contains a function call', async () => { + vi.mocked(mockContentGenerator.generateContentStream).mockResolvedValue( + streamResponse( + stopResponse([ + { text: '(request timeout)' }, + { + functionCall: { id: 'call-1', name: 'read_file', args: {} }, + }, + ]), + ), + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + 'prompt-id-placeholder-tool-call', + ); + const events: StreamEvent[] = []; + for await (const event of stream) events.push(event); + + expect( + events.some( + (event) => + event.type === StreamEventType.CHUNK && + event.value.candidates?.[0]?.content?.parts?.some( + (part) => part.functionCall?.id === 'call-1', + ), + ), + ).toBe(true); + expect(mockContentGenerator.generateContentStream).toHaveBeenCalledTimes( + 1, + ); + }); + it('should fail after all retries on persistent invalid content and report metrics', async () => { vi.useFakeTimers(); try { @@ -7515,6 +7670,55 @@ describe('GeminiChat', async () => { } }); + it('replays after a transport cut without leaking a placeholder prefix', async () => { + vi.useFakeTimers(); + try { + const transportError = Object.assign(new TypeError('terminated'), { + cause: Object.assign(new Error('other side closed'), { + code: 'UND_ERR_SOCKET', + }), + }); + vi.mocked(mockContentGenerator.generateContentStream) + .mockResolvedValueOnce( + (async function* () { + yield { + candidates: [{ content: { parts: [{ text: '(request ' }] } }], + } as unknown as GenerateContentResponse; + throw transportError; + })(), + ) + .mockResolvedValueOnce( + streamResponse(stopResponse([{ text: 'Recovered response' }])), + ); + + const stream = await chat.sendMessageStream( + 'test-model', + { message: 'test' }, + 'prompt-placeholder-prefix-transport-cut', + ); + const events = await collectStreamWithFakeTimers(stream, 5_000); + const emittedText = events + .filter((event) => event.type === StreamEventType.CHUNK) + .flatMap( + (event) => + event.value.candidates?.[0]?.content?.parts?.map( + (part) => part.text, + ) ?? [], + ); + + expect(emittedText).toEqual(['Recovered response']); + expect( + events.filter((event) => event.type === StreamEventType.RETRY), + ).toHaveLength(1); + expect(chat.getHistory()).toEqual([ + { role: 'user', parts: [{ text: 'test' }] }, + { role: 'model', parts: [{ text: 'Recovered response' }] }, + ]); + } finally { + vi.useRealTimers(); + } + }); + it('stops retrying retryable transport stream errors after the retry budget is exhausted', async () => { vi.useFakeTimers(); try { @@ -11999,6 +12203,66 @@ describe('GeminiChat', async () => { // (`--resume` of a crashed session, Ctrl+Y before in-flight tool // finishes, scheduler abort before submitQuery, manual JSONL edits). + it('keeps a tool result adjacent across a removable degraded placeholder', () => { + const history: Content[] = [ + { role: 'user', parts: [{ text: 'open /tmp/a.txt' }] }, + { + role: 'model', + parts: [ + { + functionCall: { + id: 'call-1', + name: 'read_file', + args: { path: '/tmp/a.txt' }, + }, + }, + ], + }, + { role: 'model', parts: [{ text: '(request timeout)' }] }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'call-1', + name: 'read_file', + response: { output: 'ok' }, + }, + }, + ], + }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'call-1', + name: 'read_file', + response: { output: 'ok' }, + }, + }, + ], + }, + ]; + const expectedHistory = structuredClone(history.slice(0, 4)); + chat.setHistory(history); + + expect(chat.repairOrphanedToolUseTurns()).toEqual({ + injected: [], + droppedDuplicates: [{ callId: 'call-1', name: 'read_file' }], + }); + expect(chat.repairOrphanedToolUseTurns()).toEqual({ + injected: [], + droppedDuplicates: [], + }); + expect(chat.getHistory()).toEqual(expectedHistory); + expect(chat.getHistory(true)).toEqual([ + expectedHistory[0], + expectedHistory[1], + expectedHistory[3], + ]); + }); + it('injects a synthetic functionResponse for a trailing tool_use (Race B/C)', () => { // --resume of a session that crashed after the partial-tool_use push // in `processStreamResponse` but before the scheduler submitted the diff --git a/packages/core/src/core/geminiChat.ts b/packages/core/src/core/geminiChat.ts index 4f791d121b7..728a9d2377a 100644 --- a/packages/core/src/core/geminiChat.ts +++ b/packages/core/src/core/geminiChat.ts @@ -1161,6 +1161,86 @@ function isValidContentPart(part: Part): boolean { return !isInvalid; } +const UPSTREAM_DEGRADED_PLACEHOLDER = '(request timeout)'; + +function degradedPlaceholderError(): InvalidStreamError { + return new InvalidStreamError( + 'Model response is an upstream fail-fast placeholder.', + 'UPSTREAM_DEGRADED_RESPONSE', + ); +} + +function isDegradedPlaceholderTurn(content: Content): boolean { + const parts = content.parts ?? []; + return ( + parts.length > 0 && + parts.every( + (part) => + part.functionCall === undefined && + (part.thought || part.text !== undefined), + ) && + parts + .filter((part) => !part.thought) + .map((part) => part.text ?? '') + .join('') + .trim() === UPSTREAM_DEGRADED_PLACEHOLDER + ); +} + +async function* rejectDegradedPlaceholderResponse( + stream: AsyncGenerator, +): AsyncGenerator { + const pending: GenerateContentResponse[] = []; + let text = ''; + let passthrough = false; + + for await (const chunk of stream) { + if (passthrough) { + yield chunk; + continue; + } + + const parts = chunk.candidates?.[0]?.content?.parts ?? []; + if ( + parts.some( + (part) => + part.functionCall !== undefined || + (!part.thought && part.text === undefined), + ) + ) { + yield* pending; + pending.length = 0; + yield chunk; + passthrough = true; + continue; + } + + const chunkText = parts + .filter((part) => !part.thought) + .map((part) => part.text ?? '') + .join(''); + if (pending.length === 0 && chunkText === '') { + yield chunk; + continue; + } + + pending.push(chunk); + text += chunkText; + const trimmed = text.trim(); + if (trimmed && !UPSTREAM_DEGRADED_PLACEHOLDER.startsWith(trimmed)) { + yield* pending; + pending.length = 0; + passthrough = true; + } + } + + if (passthrough) return; + if (text.trim() === UPSTREAM_DEGRADED_PLACEHOLDER) { + throw degradedPlaceholderError(); + } + yield* pending; +} + /** * Validates the history contains the correct roles. * @@ -1205,7 +1285,9 @@ function extractCuratedHistory(comprehensiveHistory: Content[]): Content[] { i++; } if (isValid) { - curatedHistory.push(...modelOutput); + curatedHistory.push( + ...modelOutput.filter((turn) => !isDegradedPlaceholderTurn(turn)), + ); } } } @@ -1475,12 +1557,14 @@ interface ScanResult { expected: Map; matched: Map; scanEnd: number; + adjacentIdx: number; } /** Decision-phase output: exact mutations the next phase will apply. */ interface RepairPlan { modelIdx: number; scanEnd: number; + adjacentIdx: number; synthesizeIds: Array<[string, string]>; hoistedParts: Part[]; removalTargets: Array<{ turnIdx: number; partIdx: number }>; @@ -1502,6 +1586,14 @@ function scanModelTurn(history: Content[], modelIdx: number): ScanResult { const matched = new Map(); let scanIdx = modelIdx + 1; + while ( + scanIdx < history.length && + history[scanIdx]?.role === 'model' && + isDegradedPlaceholderTurn(history[scanIdx]) + ) { + scanIdx++; + } + const adjacentIdx = scanIdx; while (scanIdx < history.length && history[scanIdx]?.role === 'user') { const parts = history[scanIdx].parts ?? []; for (let pIdx = 0; pIdx < parts.length; pIdx++) { @@ -1516,7 +1608,7 @@ function scanModelTurn(history: Content[], modelIdx: number): ScanResult { scanIdx++; } - return { modelIdx, expected, matched, scanEnd: scanIdx }; + return { modelIdx, expected, matched, scanEnd: scanIdx, adjacentIdx }; } /** @@ -1530,7 +1622,7 @@ function planRepair(scan: ScanResult): RepairPlan { const removalTargets: Array<{ turnIdx: number; partIdx: number }> = []; const droppedDuplicates: Array<{ callId: string; name: string }> = []; - const adjacentIdx = scan.modelIdx + 1; + const adjacentIdx = scan.adjacentIdx; for (const [id, name] of scan.expected) { const locations = scan.matched.get(id); if (!locations || locations.length === 0) { @@ -1560,6 +1652,7 @@ function planRepair(scan: ScanResult): RepairPlan { return { modelIdx: scan.modelIdx, scanEnd: scan.scanEnd, + adjacentIdx: scan.adjacentIdx, synthesizeIds, hoistedParts, removalTargets, @@ -1569,12 +1662,12 @@ function planRepair(scan: ScanResult): RepairPlan { /** * MUTATION — apply the plan to `history` in place. Returns the count - * of new user turns inserted ahead of `modelIdx + 1` (0 or 1) so the - * outer loop can advance its cursor. + * of new user turns inserted (0 or 1) so the outer loop can advance its + * cursor. * * Order: (1) splice removal targets desc-by-desc, (2) drop empty user - * turns in `[modelIdx + 2, scanEnd)`, (3) HEAD-insert at the adjacent - * user turn OR splice a new user turn between. The HEAD insert is + * turns after the resolved adjacent turn, (3) HEAD-insert at that user + * turn OR splice a new user turn there. The HEAD insert is * load-bearing (mirrors upstream `hoistToolResults`) — see the * canonical note for why tail-append re-triggers the wedge. */ @@ -1602,19 +1695,20 @@ function applyRepair( if (turnParts) turnParts.splice(loc.partIdx, 1); } - // (2) Drop now-empty user turns within [modelIdx + 2, scanEnd). + // (2) Drop now-empty user turns after the resolved adjacent turn. // Preserve the adjacent turn even if empty — we'll rewrite it // below. - const adjacentIdx = plan.modelIdx + 1; + const adjacentIdx = plan.adjacentIdx; for (let j = plan.scanEnd - 1; j > adjacentIdx; j--) { if (history[j]?.role === 'user' && (history[j].parts?.length ?? 0) === 0) { history.splice(j, 1); } } + if (partsToInject.length === 0) return { insertedBefore: 0 }; + // (3) Place new parts at the head of the adjacent user turn, OR - // insert a fresh user turn between this model turn and whatever - // follows. + // insert a fresh user turn at the resolved adjacency. const next = history[adjacentIdx]; if (next?.role === 'user') { const existing = next.parts ?? []; @@ -3972,7 +4066,7 @@ export class GeminiChat { return this.processStreamResponse( model, - streamResponse, + rejectDegradedPlaceholderResponse(streamResponse), goalContext, transportContinuationPrefix, ); diff --git a/packages/core/src/core/invalid-stream-error.ts b/packages/core/src/core/invalid-stream-error.ts index bb8de589fe7..e2873c959be 100644 --- a/packages/core/src/core/invalid-stream-error.ts +++ b/packages/core/src/core/invalid-stream-error.ts @@ -14,7 +14,8 @@ export class InvalidStreamError extends Error { | 'NO_TOOL_RESULT_PROGRESS' | 'NO_TOOL_RESULT_PROGRESS_MAX_TOKENS' | 'PROTOCOL_TAG_LEAK' - | 'MALFORMED_TOOL_CALL'; + | 'MALFORMED_TOOL_CALL' + | 'UPSTREAM_DEGRADED_RESPONSE'; constructor(message: string, type: InvalidStreamError['type']) { super(message);