diff --git a/packages/core/src/agents/runtime/agent-core.ts b/packages/core/src/agents/runtime/agent-core.ts index 4630c79c5..e12446e1b 100644 --- a/packages/core/src/agents/runtime/agent-core.ts +++ b/packages/core/src/agents/runtime/agent-core.ts @@ -31,6 +31,7 @@ import type { ToolCallConfirmationDetails, } from '../../tools/tools.js'; import { getInitialChatHistory } from '../../utils/environmentContext.js'; +import { FinishReason } from '@google/genai'; import type { Content, Part, @@ -533,6 +534,7 @@ export class AgentCore { let lastUsage: GenerateContentResponseUsageMetadata | undefined = undefined; let currentResponseId: string | undefined = undefined; + let wasOutputTruncated = false; for await (const streamEvent of responseStream) { if (roundAbortController.signal.aborted) { @@ -557,6 +559,9 @@ export class AgentCore { currentResponseId = resp.responseId; } if (resp.functionCalls) functionCalls.push(...resp.functionCalls); + if (resp.candidates?.[0]?.finishReason === FinishReason.MAX_TOKENS) { + wasOutputTruncated = true; + } const content = resp.candidates?.[0]?.content; const parts = content?.parts || []; for (const p of parts) { @@ -610,6 +615,7 @@ export class AgentCore { turnCounter, toolsList, currentResponseId, + wasOutputTruncated, ); // ── P0: Doom loop detection ─────────────────────────── @@ -820,6 +826,7 @@ export class AgentCore { currentRound: number, toolsList: FunctionDeclaration[], responseId?: string, + wasOutputTruncated = false, ): Promise { const toolResponseParts: Part[] = []; @@ -1082,6 +1089,7 @@ export class AgentCore { isClientInitiated: true, prompt_id: promptId, response_id: responseId, + wasOutputTruncated, }; const description = this.getToolDescription(toolName, args); diff --git a/packages/core/src/agents/runtime/agent-headless.test.ts b/packages/core/src/agents/runtime/agent-headless.test.ts index 01ff1b040..e04b9cbfd 100644 --- a/packages/core/src/agents/runtime/agent-headless.test.ts +++ b/packages/core/src/agents/runtime/agent-headless.test.ts @@ -48,6 +48,7 @@ import type { ToolConfig, } from './agent-types.js'; import { AgentTerminateMode } from './agent-types.js'; +import { WriteFileTool } from '../../tools/write-file.js'; vi.mock('../../core/geminiChat.js'); vi.mock('../../core/contentGenerator.js', async (importOriginal) => { @@ -1192,6 +1193,100 @@ describe('subagent.ts', () => { expect(readResult).toBeDefined(); expect(readResult!.success).toBe(true); }); + + it('should mark truncated subagent write_file calls as output-truncated errors', async () => { + const writeFileToolDef: FunctionDeclaration = { + name: WriteFileTool.Name, + description: 'Writes a file', + parameters: { type: Type.OBJECT, properties: {} }, + }; + + const { config } = await createMockConfig({ + getFunctionDeclarationsFiltered: vi + .fn() + .mockReturnValue([writeFileToolDef]), + getTool: vi.fn().mockImplementation((name: string) => { + if (name === WriteFileTool.Name) { + return new WriteFileTool(config); + } + return undefined; + }), + }); + + const toolConfig: ToolConfig = { tools: [WriteFileTool.Name] }; + const toolResultEvents: AgentToolResultEvent[] = []; + const eventEmitter = new AgentEventEmitter(); + eventEmitter.on(AgentEventType.TOOL_RESULT, (event: unknown) => { + toolResultEvents.push(event as AgentToolResultEvent); + }); + + mockSendMessageStream.mockImplementation(async () => + (async function* () { + yield { + type: 'chunk', + value: { + functionCalls: [ + { + id: 'call_write', + name: WriteFileTool.Name, + args: { file_path: '/tmp/truncated.txt' }, + }, + ], + }, + }; + yield { + type: 'chunk', + value: { + candidates: [ + { + finishReason: 'MAX_TOKENS', + content: { parts: [] }, + }, + ], + }, + }; + yield { + type: 'chunk', + value: { + candidates: [ + { + content: { + parts: [{ text: 'done' }], + }, + }, + ], + }, + }; + })(), + ); + + const scope = await AgentHeadless.create( + 'test-agent', + config, + promptConfig, + defaultModelConfig, + defaultRunConfig, + toolConfig, + eventEmitter, + ); + + await scope.execute(new ContextState()); + + const writeResult = toolResultEvents.find( + (event) => event.name === WriteFileTool.Name, + ); + expect(writeResult).toBeDefined(); + expect(writeResult!.success).toBe(false); + expect(writeResult!.error).toContain( + 'truncated due to max_tokens limit', + ); + expect(writeResult!.error).toContain( + 'rejected to prevent writing truncated content', + ); + expect(writeResult!.error).not.toContain( + "params must have required property 'content'", + ); + }); }); }); }); diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index 1ff0f559b..7a5eaacda 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -1314,6 +1314,12 @@ export class Config { return; } + // Strip thinking blocks from conversation history on model switch. + // reasoning_content is a non-standard field that causes strict + // OpenAI-compatible providers to reject requests with 422 errors + // when thought parts from a previous model leak into the payload (#3304). + this.geminiClient.stripThoughtsFromHistory(); + // Full refresh path await this.refreshAuth(authType); } diff --git a/packages/core/src/core/client.ts b/packages/core/src/core/client.ts index f449c28b7..b69afaff3 100644 --- a/packages/core/src/core/client.ts +++ b/packages/core/src/core/client.ts @@ -656,6 +656,10 @@ export class GeminiClient { this.config.getChatRecordingService()?.recordUserMessage(request); // strip thoughts from history before sending the message + // NOTE: backport of upstream #3590 changed sessionService default to + // KEEP thoughts (preserves reasoning_content for DeepSeek/reasoning + // models on resume). The mid-stream stripThoughtsFromHistory() here + // remains for active turns to avoid stale thoughts polluting cache. this.stripThoughtsFromHistory(); // Capture history length for rewind support. diff --git a/packages/core/src/core/coreToolScheduler.test.ts b/packages/core/src/core/coreToolScheduler.test.ts index 1fab80dbc..15438cbaf 100644 --- a/packages/core/src/core/coreToolScheduler.test.ts +++ b/packages/core/src/core/coreToolScheduler.test.ts @@ -7,6 +7,7 @@ import { describe, it, expect, vi, beforeEach } from 'vitest'; import type { Mock } from 'vitest'; import type { + AnyDeclarativeTool, Config, ToolCallConfirmationDetails, ToolConfirmationPayload, @@ -42,6 +43,7 @@ import { MessageBusType } from '../confirmation-bus/types.js'; import type { HookExecutionResponse } from '../confirmation-bus/types.js'; import { type NotificationType } from '../hooks/types.js'; import type { MessageBus } from '../confirmation-bus/message-bus.js'; +import { WriteFileTool } from '../tools/write-file.js'; vi.mock('fs/promises', () => ({ writeFile: vi.fn(), @@ -1801,7 +1803,7 @@ describe('CoreToolScheduler request queueing', () => { describe('CoreToolScheduler truncated output protection', () => { function createTruncationTestScheduler( - tool: TestApprovalTool | MockTool, + tool: AnyDeclarativeTool, toolNames: string[], ) { const onAllToolCallsComplete = vi.fn(); @@ -1969,6 +1971,59 @@ describe('CoreToolScheduler truncated output protection', () => { // Non-Edit tools should still execute even when output was truncated expect(completedCalls[0].status).toBe('success'); }); + + it('should prefer truncation rejection over validation errors for truncated write_file calls', async () => { + const writeFileConfig = { + getProjectRoot: () => '/tmp', + getTargetDir: () => '/tmp', + getFileSystemService: () => ({ + readTextFile: vi.fn(), + writeTextFile: vi.fn(), + }), + getDefaultFileEncoding: () => undefined, + setApprovalMode: vi.fn(), + } as unknown as Config; + const writeFileTool = new WriteFileTool(writeFileConfig); + const { scheduler, onAllToolCallsComplete } = createTruncationTestScheduler( + writeFileTool, + [WriteFileTool.Name], + ); + + await scheduler.schedule( + [ + { + callId: '1', + name: WriteFileTool.Name, + args: { file_path: '/tmp/test.txt' }, + isClientInitiated: false, + prompt_id: 'prompt-id-write-file-truncated', + wasOutputTruncated: true, + }, + ], + new AbortController().signal, + ); + + await vi.waitFor(() => { + expect(onAllToolCallsComplete).toHaveBeenCalled(); + }); + + const completedCalls = onAllToolCallsComplete.mock + .calls[0][0] as ToolCall[]; + expect(completedCalls).toHaveLength(1); + const completedCall = completedCalls[0]; + expect(completedCall.status).toBe('error'); + + if (completedCall.status === 'error') { + const errorMessage = completedCall.response.error?.message; + expect(errorMessage).toContain('truncated due to max_tokens limit'); + expect(errorMessage).toContain( + 'rejected to prevent writing truncated content', + ); + expect(errorMessage).not.toContain( + "params must have required property 'content'", + ); + } + }); }); describe('CoreToolScheduler Sequential Execution', () => { diff --git a/packages/core/src/core/coreToolScheduler.ts b/packages/core/src/core/coreToolScheduler.ts index 53903033d..9ee38a4e3 100644 --- a/packages/core/src/core/coreToolScheduler.ts +++ b/packages/core/src/core/coreToolScheduler.ts @@ -778,6 +778,24 @@ export class CoreToolScheduler { continue; } + // Reject file-modifying calls when truncated to prevent + // writing incomplete content, even if params failed schema validation. + if (reqInfo.wasOutputTruncated && toolInstance.kind === Kind.Edit) { + const truncationError = new Error(TRUNCATION_EDIT_REJECTION); + newToolCalls.push({ + status: 'error', + request: reqInfo, + tool: toolInstance, + response: createErrorResponse( + reqInfo, + truncationError, + ToolErrorType.OUTPUT_TRUNCATED, + ), + durationMs: 0, + }); + continue; + } + const invocationOrError = this.buildInvocation( toolInstance, reqInfo.args, diff --git a/packages/core/src/core/openaiContentGenerator/converter.test.ts b/packages/core/src/core/openaiContentGenerator/converter.test.ts index 7dc0d390f..00818e495 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.test.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.test.ts @@ -6,7 +6,7 @@ import { describe, it, expect, beforeEach } from 'vitest'; import { OpenAIContentConverter } from './converter.js'; -import type { StreamingToolCallParser } from './streamingToolCallParser.js'; +import { StreamingToolCallParser } from './streamingToolCallParser.js'; import { Type, FinishReason, @@ -31,57 +31,33 @@ describe('OpenAIContentConverter', () => { }); }); - describe('resetStreamingToolCalls', () => { - it('should clear streaming tool calls accumulator', () => { - // Access private field for testing - const parser = ( - converter as unknown as { - streamingToolCallParser: StreamingToolCallParser; - } - ).streamingToolCallParser; - - // Add some test data to the parser - parser.addChunk(0, '{"arg": "value"}', 'test-id', 'test-function'); - parser.addChunk(1, '{"arg2": "value2"}', 'test-id-2', 'test-function-2'); - - // Verify data is present - expect(parser.getBuffer(0)).toBe('{"arg": "value"}'); - expect(parser.getBuffer(1)).toBe('{"arg2": "value2"}'); - - // Call reset method - converter.resetStreamingToolCalls(); - - // Verify data is cleared - expect(parser.getBuffer(0)).toBe(''); - expect(parser.getBuffer(1)).toBe(''); - }); + describe('createStreamContext', () => { + it('returns a fresh context with its own StreamingToolCallParser', () => { + const ctx1 = converter.createStreamContext(); + const ctx2 = converter.createStreamContext(); - it('should be safe to call multiple times', () => { - // Call reset multiple times - converter.resetStreamingToolCalls(); - converter.resetStreamingToolCalls(); - converter.resetStreamingToolCalls(); - - // Should not throw any errors - const parser = ( - converter as unknown as { - streamingToolCallParser: StreamingToolCallParser; - } - ).streamingToolCallParser; - expect(parser.getBuffer(0)).toBe(''); + expect(ctx1.toolCallParser).toBeInstanceOf(StreamingToolCallParser); + expect(ctx2.toolCallParser).toBeInstanceOf(StreamingToolCallParser); + expect(ctx1.toolCallParser).not.toBe(ctx2.toolCallParser); + expect(ctx1.thinkBuffer).toBe(''); + expect(ctx1.inThinkTag).toBe(false); }); - it('should be safe to call on empty accumulator', () => { - // Call reset on empty accumulator - converter.resetStreamingToolCalls(); - - // Should not throw any errors - const parser = ( - converter as unknown as { - streamingToolCallParser: StreamingToolCallParser; - } - ).streamingToolCallParser; - expect(parser.getBuffer(0)).toBe(''); + it('isolates two contexts so writes to one do not leak into the other', () => { + // Regression for issue #3516: previously the parser lived on the + // Converter as an instance field, so two concurrent streams sharing + // the same Config.contentGenerator would overwrite each other's + // tool-call buffers. Per-stream contexts eliminate that contention. + const ctx1 = converter.createStreamContext(); + const ctx2 = converter.createStreamContext(); + + ctx1.toolCallParser.addChunk(0, '{"a":1}', 'call_A', 'fn_A'); + ctx2.toolCallParser.addChunk(0, '{"b":2}', 'call_B', 'fn_B'); + + expect(ctx1.toolCallParser.getBuffer(0)).toBe('{"a":1}'); + expect(ctx2.toolCallParser.getBuffer(0)).toBe('{"b":2}'); + expect(ctx1.toolCallParser.getToolCallMeta(0).id).toBe('call_A'); + expect(ctx2.toolCallParser.getToolCallMeta(0).id).toBe('call_B'); }); }); @@ -846,6 +822,104 @@ describe('OpenAIContentConverter', () => { content: '', }); }); + + describe('assistant message with reasoning-only content (issue #3421)', () => { + /** + * When a model (e.g. Ollama qwen3.5:9b) returns a response that contains + * reasoning content but an empty text body, the converted assistant message + * must use content: "" instead of content: null. Some OpenAI-compatible + * providers reject content: null with HTTP 400 when reasoning_content is + * present. + */ + it('should use empty string instead of null for content when assistant has only reasoning parts', () => { + const request: GenerateContentParameters = { + model: 'models/test', + contents: [ + { role: 'user', parts: [{ text: 'Think about this.' }] }, + { + role: 'model', + parts: [{ text: 'I reasoned about it.', thought: true }], + }, + { role: 'user', parts: [{ text: 'What did you conclude?' }] }, + ], + }; + + const messages = converter.convertGeminiRequestToOpenAI(request); + + const assistantMsg = messages.find((m) => m.role === 'assistant'); + expect(assistantMsg).toBeDefined(); + expect((assistantMsg as { content: unknown }).content).toBe(''); + expect( + (assistantMsg as { reasoning_content?: string }).reasoning_content, + ).toBe('I reasoned about it.'); + }); + + it('should keep content null when assistant has only tool_calls and no reasoning', () => { + const request: GenerateContentParameters = { + model: 'models/test', + contents: [ + { role: 'user', parts: [{ text: 'Call the tool.' }] }, + { + role: 'model', + parts: [ + { + functionCall: { + id: 'call_1', + name: 'some_tool', + args: {}, + }, + }, + ], + }, + { + role: 'user', + parts: [ + { + functionResponse: { + id: 'call_1', + name: 'some_tool', + response: { output: 'done' }, + }, + }, + ], + }, + ], + }; + + const messages = converter.convertGeminiRequestToOpenAI(request); + + const assistantMsg = messages.find((m) => m.role === 'assistant'); + expect(assistantMsg).toBeDefined(); + expect((assistantMsg as { content: unknown }).content).toBeNull(); + }); + + it('should use actual text content when assistant has both reasoning and text', () => { + const request: GenerateContentParameters = { + model: 'models/test', + contents: [ + { role: 'user', parts: [{ text: 'Explain.' }] }, + { + role: 'model', + parts: [ + { text: 'My hidden reasoning.', thought: true }, + { text: 'Here is my answer.' }, + ], + }, + ], + }; + + const messages = converter.convertGeminiRequestToOpenAI(request); + + const assistantMsg = messages.find((m) => m.role === 'assistant'); + expect(assistantMsg).toBeDefined(); + expect((assistantMsg as { content: unknown }).content).toBe( + 'Here is my answer.', + ); + expect( + (assistantMsg as { reasoning_content?: string }).reasoning_content, + ).toBe('My hidden reasoning.'); + }); + }); }); describe('MCP multi-part tool results (issue #1520)', () => { @@ -1063,23 +1137,26 @@ describe('OpenAIContentConverter', () => { }); it('should convert streaming reasoning_content delta to a thought part', () => { - const chunk = converter.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'chunk-1', - created: 456, - choices: [ - { - index: 0, - delta: { - content: 'visible text', - reasoning_content: 'thinking...', + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-1', + created: 456, + choices: [ + { + index: 0, + delta: { + content: 'visible text', + reasoning_content: 'thinking...', + }, + finish_reason: 'stop', + logprobs: null, }, - finish_reason: 'stop', - logprobs: null, - }, - ], - model: 'gpt-test', - } as unknown as OpenAI.Chat.ChatCompletionChunk); + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + converter.createStreamContext(), + ); const parts = chunk.candidates?.[0]?.content?.parts; expect(parts?.[0]).toEqual( @@ -1091,23 +1168,26 @@ describe('OpenAIContentConverter', () => { }); it('should convert streaming reasoning delta to a thought part', () => { - const chunk = converter.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'chunk-1b', - created: 456, - choices: [ - { - index: 0, - delta: { - content: 'visible text', - reasoning: 'thinking...', + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-1b', + created: 456, + choices: [ + { + index: 0, + delta: { + content: 'visible text', + reasoning: 'thinking...', + }, + finish_reason: 'stop', + logprobs: null, }, - finish_reason: 'stop', - logprobs: null, - }, - ], - model: 'gpt-test', - } as unknown as OpenAI.Chat.ChatCompletionChunk); + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + converter.createStreamContext(), + ); const parts = chunk.candidates?.[0]?.content?.parts; expect(parts?.[0]).toEqual( @@ -1119,21 +1199,24 @@ describe('OpenAIContentConverter', () => { }); it('should not throw when streaming chunk has no delta', () => { - const chunk = converter.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'chunk-2', - created: 456, - choices: [ - { - index: 0, - // Some OpenAI-compatible providers may omit delta entirely. - delta: undefined, - finish_reason: null, - logprobs: null, - }, - ], - model: 'gpt-test', - } as unknown as OpenAI.Chat.ChatCompletionChunk); + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-2', + created: 456, + choices: [ + { + index: 0, + // Some OpenAI-compatible providers may omit delta entirely. + delta: undefined, + finish_reason: null, + logprobs: null, + }, + ], + model: 'gpt-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + converter.createStreamContext(), + ); const parts = chunk.candidates?.[0]?.content?.parts; expect(parts).toEqual([]); @@ -1145,33 +1228,36 @@ describe('OpenAIContentConverter', () => { // with partial JSON. Emitting such a tool call would corrupt the // conversation history (downstream provider Pydantic validator fails) // and the tool would be invoked with empty args. - const chunk = converter.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'chunk-malformed', - created: 789, - choices: [ - { - index: 0, - delta: { - tool_calls: [ - { - index: 0, - id: 'call_bad', - type: 'function', - function: { - name: 'Read', - arguments: - 'Current state — research in progress{"file_path": and more prose without closure', + const chunk = converter.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-malformed', + created: 789, + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_bad', + type: 'function', + function: { + name: 'Read', + arguments: + 'Current state — research in progress{"file_path": and more prose without closure', + }, }, - }, - ], + ], + }, + finish_reason: 'tool_calls', + logprobs: null, }, - finish_reason: 'tool_calls', - logprobs: null, - }, - ], - model: 'qwen-test', - } as unknown as OpenAI.Chat.ChatCompletionChunk); + ], + model: 'qwen-test', + } as unknown as OpenAI.Chat.ChatCompletionChunk, + converter.createStreamContext(), + ); const parts = chunk.candidates?.[0]?.content?.parts ?? []; expect(parts.some((p) => 'functionCall' in p)).toBe(false); @@ -1291,13 +1377,15 @@ describe('OpenAIContentConverter', () => { }); describe('streaming (convertOpenAIChunkToGemini)', () => { + let ctx: ReturnType; beforeEach(() => { - converter.resetStreamingToolCalls(); + ctx = converter.createStreamContext(); }); it('extracts content arriving in a single chunk', () => { const chunk = converter.convertOpenAIChunkToGemini( makeChunk('reasoninganswer'), + ctx, ); const parts = chunk.candidates?.[0]?.content?.parts; expect( @@ -1312,10 +1400,12 @@ describe('OpenAIContentConverter', () => { // Chunk 1: opening tag only const c1 = converter.convertOpenAIChunkToGemini( makeChunk('rea'), + ctx, ); // Chunk 2: rest of thought + closing tag + answer const c2 = converter.convertOpenAIChunkToGemini( makeChunk('soninganswer'), + ctx, ); const thoughtParts1 = c1.candidates?.[0]?.content?.parts?.filter( @@ -1336,10 +1426,11 @@ describe('OpenAIContentConverter', () => { }); it('handles opening tag split across chunks', () => { - converter.convertOpenAIChunkToGemini(makeChunk('reasoning')); + converter.convertOpenAIChunkToGemini(makeChunk('reasoning'), ctx); const c3 = converter.convertOpenAIChunkToGemini( makeChunk('done'), + ctx, ); const textPart = c3.candidates?.[0]?.content?.parts?.find( @@ -1348,10 +1439,15 @@ describe('OpenAIContentConverter', () => { expect(textPart?.text).toBe('done'); }); - it('resets state between streams via resetStreamingToolCalls', () => { - converter.convertOpenAIChunkToGemini(makeChunk('orphaned')); - converter.resetStreamingToolCalls(); - const c = converter.convertOpenAIChunkToGemini(makeChunk('clean text')); + it('isolates state between streams via separate contexts', () => { + // Replaces the old reset-via-method test: per-stream contexts + // give us natural isolation without any explicit reset call. + converter.convertOpenAIChunkToGemini(makeChunk('orphaned'), ctx); + const ctx2 = converter.createStreamContext(); + const c = converter.convertOpenAIChunkToGemini( + makeChunk('clean text'), + ctx2, + ); const parts = c.candidates?.[0]?.content?.parts; expect(parts).toHaveLength(1); expect(parts?.[0]?.text).toBe('clean text'); @@ -2191,51 +2287,58 @@ describe('Truncated tool call detection in streaming', () => { }>, finishReason: string, ) { + const ctx = conv.createStreamContext(); // Feed argument chunks (no finish_reason yet) for (const tc of toolCallChunks) { - conv.convertOpenAIChunkToGemini({ + conv.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'chunk-stream', + created: 100, + model: 'test-model', + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: tc.index, + id: tc.id, + type: 'function' as const, + function: { + name: tc.name, + arguments: tc.arguments, + }, + }, + ], + }, + finish_reason: null, + logprobs: null, + }, + ], + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); + } + + // Final chunk with finish_reason + return conv.convertOpenAIChunkToGemini( + { object: 'chat.completion.chunk', - id: 'chunk-stream', - created: 100, + id: 'chunk-final', + created: 101, model: 'test-model', choices: [ { index: 0, - delta: { - tool_calls: [ - { - index: tc.index, - id: tc.id, - type: 'function' as const, - function: { - name: tc.name, - arguments: tc.arguments, - }, - }, - ], - }, - finish_reason: null, + delta: {}, + finish_reason: finishReason, logprobs: null, }, ], - } as unknown as OpenAI.Chat.ChatCompletionChunk); - } - - // Final chunk with finish_reason - return conv.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'chunk-final', - created: 101, - model: 'test-model', - choices: [ - { - index: 0, - delta: {}, - finish_reason: finishReason, - logprobs: null, - }, - ], - } as unknown as OpenAI.Chat.ChatCompletionChunk); + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); } it('should override finishReason to MAX_TOKENS when tool call JSON is truncated and provider reports "stop"', () => { @@ -2336,70 +2439,80 @@ describe('Truncated tool call detection in streaming', () => { it('should detect truncation with multi-chunk streaming arguments', () => { // Feed arguments in multiple small chunks like real streaming const conv = new OpenAIContentConverter('test-model'); + const ctx = conv.createStreamContext(); // Chunk 1: start of JSON with tool metadata - conv.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'c1', - created: 100, - model: 'test-model', - choices: [ - { - index: 0, - delta: { - tool_calls: [ - { - index: 0, - id: 'call_1', - type: 'function' as const, - function: { name: 'write_file', arguments: '{"file_' }, - }, - ], + conv.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'c1', + created: 100, + model: 'test-model', + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_1', + type: 'function' as const, + function: { name: 'write_file', arguments: '{"file_' }, + }, + ], + }, + finish_reason: null, + logprobs: null, }, - finish_reason: null, - logprobs: null, - }, - ], - } as unknown as OpenAI.Chat.ChatCompletionChunk); + ], + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); // Chunk 2: more arguments - conv.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'c2', - created: 100, - model: 'test-model', - choices: [ - { - index: 0, - delta: { - tool_calls: [ - { - index: 0, - function: { arguments: 'path": "/tmp/f.txt", "conten' }, - }, - ], + conv.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'c2', + created: 100, + model: 'test-model', + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + function: { arguments: 'path": "/tmp/f.txt", "conten' }, + }, + ], + }, + finish_reason: null, + logprobs: null, }, - finish_reason: null, - logprobs: null, - }, - ], - } as unknown as OpenAI.Chat.ChatCompletionChunk); + ], + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); // Final chunk: finish_reason "stop" but JSON is still incomplete - const result = conv.convertOpenAIChunkToGemini({ - object: 'chat.completion.chunk', - id: 'c3', - created: 101, - model: 'test-model', - choices: [ - { - index: 0, - delta: {}, - finish_reason: 'stop', - logprobs: null, - }, - ], - } as unknown as OpenAI.Chat.ChatCompletionChunk); + const result = conv.convertOpenAIChunkToGemini( + { + object: 'chat.completion.chunk', + id: 'c3', + created: 101, + model: 'test-model', + choices: [ + { + index: 0, + delta: {}, + finish_reason: 'stop', + logprobs: null, + }, + ], + } as unknown as OpenAI.Chat.ChatCompletionChunk, + ctx, + ); expect(result.candidates?.[0]?.finishReason).toBe(FinishReason.MAX_TOKENS); }); diff --git a/packages/core/src/core/openaiContentGenerator/converter.ts b/packages/core/src/core/openaiContentGenerator/converter.ts index 13cf1ccb7..330934bf4 100644 --- a/packages/core/src/core/openaiContentGenerator/converter.ts +++ b/packages/core/src/core/openaiContentGenerator/converter.ts @@ -90,6 +90,23 @@ type OpenAIContentPart = | OpenAIContentPartVideoUrl | OpenAIContentPartFile; +/** + * Per-stream state for tool-call parsing and `` tag accumulation. + * Created by `OpenAIContentConverter.createStreamContext()` at the start + * of each streaming response and passed into every + * `convertOpenAIChunkToGemini` call on that stream, so concurrent streams + * (parallel subagents, fork children, ACP concurrent Agent calls) never + * share parser state. + * + * Backport of upstream #3525 + extended to also scope `` tag + * parsing state per-stream (Minimax/QwQ inline-XML reasoning). + */ +export interface ConverterStreamContext { + toolCallParser: StreamingToolCallParser; + thinkBuffer: string; + inThinkTag: boolean; +} + /** * Converter class for transforming data between Gemini and OpenAI formats */ @@ -97,22 +114,6 @@ export class OpenAIContentConverter { private model: string; private schemaCompliance: SchemaComplianceMode; private modalities: InputModalities; - private streamingToolCallParser: StreamingToolCallParser = - new StreamingToolCallParser(); - - /** - * Stateful buffer for streaming `` tag parsing. - * - * Some models (e.g. Minimax, QwQ) emit chain-of-thought as raw XML in the - * `content` field rather than via a dedicated `reasoning_content` field: - * - * reasoning hereactual answer - * - * Because tags may be split across multiple stream chunks we accumulate - * state here across calls to `convertOpenAIChunkToGemini`. - */ - private _thinkBuffer = ''; - private _inThinkTag = false; constructor( model: string, @@ -140,14 +141,26 @@ export class OpenAIContentConverter { } /** - * Reset streaming tool calls parser for new stream processing - * This should be called at the beginning of each stream to prevent - * data pollution from previous incomplete streams + * Create fresh per-stream state for processing one OpenAI streaming + * response. The returned context is passed into every + * `convertOpenAIChunkToGemini` call for that stream, then discarded. + * + * Previously the tool-call parser and `` parse state lived on + * the Converter instance and were shared by every caller of the + * singleton `Config.contentGenerator`. Concurrent streams (parallel + * subagents, fork children, ACP concurrent Agent calls per #3463) + * raced on that shared state: each stream's stream-start `reset()` + * wiped the other's partial tool-call buffers, chunks from different + * streams landed at the same `index=0` bucket, and + * `getCompletedToolCalls()` returned interleaved corrupt JSON that + * surfaced upstream as `NO_RESPONSE_TEXT` (issue #3516). */ - resetStreamingToolCalls(): void { - this.streamingToolCallParser.reset(); - this._thinkBuffer = ''; - this._inThinkTag = false; + createStreamContext(): ConverterStreamContext { + return { + toolCallParser: new StreamingToolCallParser(), + thinkBuffer: '', + inThinkTag: false, + }; } /** @@ -176,19 +189,23 @@ export class OpenAIContentConverter { * Process a streaming text chunk that may contain partial or complete * `` XML tags. * - * Mutates `this._thinkBuffer` / `this._inThinkTag` to track cross-chunk state. + * Mutates `ctx.thinkBuffer` / `ctx.inThinkTag` to track cross-chunk state + * for this specific stream. * * Returns `{ thoughtDelta, textDelta }` where either (or both) may be empty * strings when there is nothing to emit yet. */ - private processThinkChunk(chunk: string): { + private processThinkChunk( + chunk: string, + ctx: ConverterStreamContext, + ): { thoughtDelta: string; textDelta: string; } { // Prepend any buffered partial opening tag from the previous chunk. - if (this._thinkBuffer) { - chunk = this._thinkBuffer + chunk; - this._thinkBuffer = ''; + if (ctx.thinkBuffer) { + chunk = ctx.thinkBuffer + chunk; + ctx.thinkBuffer = ''; } let thoughtDelta = ''; @@ -196,7 +213,7 @@ export class OpenAIContentConverter { let i = 0; while (i < chunk.length) { - if (this._inThinkTag) { + if (ctx.inThinkTag) { // We are inside a block — scan for the closing tag. const closeIdx = chunk.indexOf('', i); if (closeIdx === -1) { @@ -206,7 +223,7 @@ export class OpenAIContentConverter { } else { // Found the closing tag. thoughtDelta += chunk.slice(i, closeIdx); - this._inThinkTag = false; + ctx.inThinkTag = false; i = closeIdx + ''.length; } } else { @@ -221,19 +238,19 @@ export class OpenAIContentConverter { if (partialOpen > 0) { // Buffer the possible partial tag; emit everything before it as text. textDelta += tail.slice(0, tail.length - partialOpen); - this._thinkBuffer = tail.slice(tail.length - partialOpen); + ctx.thinkBuffer = tail.slice(tail.length - partialOpen); } else { // Flush any buffered partial tag as text now that we know it wasn't // a real . - textDelta += this._thinkBuffer + tail; - this._thinkBuffer = ''; + textDelta += ctx.thinkBuffer + tail; + ctx.thinkBuffer = ''; } i = chunk.length; } else { // Flush buffered partial + text before the tag. - textDelta += this._thinkBuffer + chunk.slice(i, openIdx); - this._thinkBuffer = ''; - this._inThinkTag = true; + textDelta += ctx.thinkBuffer + chunk.slice(i, openIdx); + ctx.thinkBuffer = ''; + ctx.inThinkTag = true; i = openIdx + ''.length; } } @@ -624,7 +641,12 @@ export class OpenAIContentConverter { .join(''); const assistantMessage: ExtendedChatCompletionAssistantMessageParam = { role: 'assistant', - content: assistantTextContent || null, + // When there is reasoning content but no text, use "" instead of null. + // Some OpenAI-compatible providers (e.g. Ollama) reject content: null + // when reasoning_content is present, returning HTTP 400. + // For tool-call-only messages we keep null to stay spec-compliant. + content: + assistantTextContent || (reasoningParts.length > 0 ? '' : null), }; if (toolCalls.length > 0) { @@ -1076,10 +1098,17 @@ export class OpenAIContentConverter { } /** - * Convert OpenAI stream chunk to Gemini format + * Convert OpenAI stream chunk to Gemini format. + * + * `ctx` carries the tool-call parser and ``-tag state for this + * stream. Callers MUST obtain it from `createStreamContext()` at the + * start of the stream and pass the same instance for every chunk of + * that stream. Concurrent streams MUST use distinct contexts or their + * tool-call buffers will interleave (issue #3516). */ convertOpenAIChunkToGemini( chunk: OpenAI.Chat.ChatCompletionChunk, + ctx: ConverterStreamContext, ): GenerateContentResponse { const choice = chunk.choices?.[0]; const response = new GenerateContentResponse(); @@ -1101,6 +1130,7 @@ export class OpenAIContentConverter { if (typeof choice.delta.content === 'string') { const { thoughtDelta, textDelta } = this.processThinkChunk( choice.delta.content, + ctx, ); // Only emit a thought part when there is no dedicated reasoning field // (avoid double-counting when a model provides both). @@ -1113,14 +1143,14 @@ export class OpenAIContentConverter { } } - // Handle tool calls using the streaming parser + // Handle tool calls using the stream-local parser if (choice.delta?.tool_calls) { for (const toolCall of choice.delta.tool_calls) { const index = toolCall.index ?? 0; // Process the tool call chunk through the streaming parser if (toolCall.function?.arguments) { - this.streamingToolCallParser.addChunk( + ctx.toolCallParser.addChunk( index, toolCall.function.arguments, toolCall.id, @@ -1128,7 +1158,7 @@ export class OpenAIContentConverter { ); } else { // Handle metadata-only chunks (id and/or name without arguments) - this.streamingToolCallParser.addChunk( + ctx.toolCallParser.addChunk( index, '', // Empty chunk for metadata-only updates toolCall.id, @@ -1144,11 +1174,9 @@ export class OpenAIContentConverter { // Detect truncation the provider may not report correctly. // Some providers (e.g. DashScope/Qwen) send "stop" or "tool_calls" // even when output was cut off mid-JSON due to max_tokens. - toolCallsTruncated = - this.streamingToolCallParser.hasIncompleteToolCalls(); + toolCallsTruncated = ctx.toolCallParser.hasIncompleteToolCalls(); - const completedToolCalls = - this.streamingToolCallParser.getCompletedToolCalls(); + const completedToolCalls = ctx.toolCallParser.getCompletedToolCalls(); for (const toolCall of completedToolCalls) { if (!toolCall.name) continue; @@ -1178,8 +1206,9 @@ export class OpenAIContentConverter { }); } - // Clear the parser for the next stream - this.streamingToolCallParser.reset(); + // Parser is stream-local; it will be discarded with the + // ConverterStreamContext when the stream finishes. No manual + // reset needed. } // If tool call JSON was truncated, override to "length" so downstream diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.concurrent.test.ts b/packages/core/src/core/openaiContentGenerator/pipeline.concurrent.test.ts new file mode 100644 index 000000000..0c83ab47c --- /dev/null +++ b/packages/core/src/core/openaiContentGenerator/pipeline.concurrent.test.ts @@ -0,0 +1,350 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Integration test — deliberately does NOT mock `./converter.js`. Unlike + * `pipeline.test.ts` which stubs the converter, this suite drives the real + * `ContentGenerationPipeline` + real `OpenAIContentConverter` through two + * streams that interleave on the event loop, and asserts that tool-call + * arguments from one stream never bleed into the other's output. + * + * This is the regression test for issue #3516: before the per-stream + * parser scoping fix, the Converter singleton held a single + * `StreamingToolCallParser` instance. Two concurrent streams would share + * it; each stream's entry-time reset wiped the other's partial buffers, + * and chunks routed by `index: 0` interleaved into corrupt JSON. + * + * With the fix, `processStreamWithLogging` creates a fresh + * `ConverterStreamContext` at stream entry, so each concurrent generator + * has its own parser. This test would fail deterministically on pre-fix + * code because stream B's entry would wipe stream A's accumulator + * mid-flight, and A's finish chunk would emit zero function calls + * (`wasOutputTruncated`-style behavior). + */ + +import { describe, it, expect, vi } from 'vitest'; +import type OpenAI from 'openai'; +import type { GenerateContentParameters } from '@google/genai'; +import type { Part } from '@google/genai'; +import type { PipelineConfig } from './pipeline.js'; +import { ContentGenerationPipeline } from './pipeline.js'; +import type { Config } from '../../config/config.js'; +import type { ContentGeneratorConfig, AuthType } from '../contentGenerator.js'; +import type { OpenAICompatibleProvider } from './provider/index.js'; +import type { ErrorHandler } from './errorHandler.js'; + +type ChunkFactory = () => OpenAI.Chat.ChatCompletionChunk; + +/** + * Build a slow stream that yields to the event loop between chunks. + * Without the `setImmediate` await, a `for await` loop on one stream + * drains synchronously and `Promise.all` degenerates to serial execution, + * which hides the cross-stream bug. + */ +async function* interleavingStream( + chunks: ChunkFactory[], +): AsyncGenerator { + for (const make of chunks) { + // Yield control so the sibling stream can advance one step before we do. + await new Promise((r) => setImmediate(r)); + yield make(); + } +} + +function openerChunk( + id: string, + name: string, + firstArgs: string, +): OpenAI.Chat.ChatCompletionChunk { + return { + id: `${id}-opener`, + object: 'chat.completion.chunk', + created: 1, + model: 'test', + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id, + type: 'function', + function: { name, arguments: firstArgs }, + }, + ], + }, + finish_reason: null, + logprobs: null, + }, + ], + } as unknown as OpenAI.Chat.ChatCompletionChunk; +} + +function continuationChunk( + argsFragment: string, +): OpenAI.Chat.ChatCompletionChunk { + return { + id: 'cont', + object: 'chat.completion.chunk', + created: 1, + model: 'test', + choices: [ + { + index: 0, + delta: { + tool_calls: [{ index: 0, function: { arguments: argsFragment } }], + }, + finish_reason: null, + logprobs: null, + }, + ], + } as unknown as OpenAI.Chat.ChatCompletionChunk; +} + +function finisherChunk(): OpenAI.Chat.ChatCompletionChunk { + return { + id: 'finish', + object: 'chat.completion.chunk', + created: 1, + model: 'test', + choices: [ + { + index: 0, + delta: {}, + finish_reason: 'tool_calls', + logprobs: null, + }, + ], + usage: { prompt_tokens: 10, completion_tokens: 20, total_tokens: 30 }, + } as unknown as OpenAI.Chat.ChatCompletionChunk; +} + +describe('ContentGenerationPipeline — concurrent streams (issue #3516)', () => { + function buildPipeline( + createStreamImpl: () => AsyncIterable, + ) { + const mockClient = { + chat: { + completions: { + // Each call returns a fresh stream. The real Pipeline will + // invoke this twice — once per concurrent executeStream call. + create: vi.fn().mockImplementation(() => createStreamImpl()), + }, + }, + } as unknown as OpenAI; + + const mockProvider: OpenAICompatibleProvider = { + buildClient: vi.fn().mockReturnValue(mockClient), + buildRequest: vi.fn().mockImplementation((req) => req), + buildHeaders: vi.fn().mockReturnValue({}), + getDefaultGenerationConfig: vi.fn().mockReturnValue({}), + } as unknown as OpenAICompatibleProvider; + + const mockErrorHandler: ErrorHandler = { + handle: vi.fn().mockImplementation((error: unknown) => { + throw error; + }), + shouldSuppressErrorLogging: vi.fn().mockReturnValue(false), + } as unknown as ErrorHandler; + + const contentGeneratorConfig: ContentGeneratorConfig = { + model: 'test-model', + authType: 'openai' as AuthType, + } as ContentGeneratorConfig; + + const config: PipelineConfig = { + cliConfig: {} as Config, + provider: mockProvider, + contentGeneratorConfig, + errorHandler: mockErrorHandler, + }; + + return { pipeline: new ContentGenerationPipeline(config), mockClient }; + } + + it('two concurrent streams keep their tool-call buffers isolated', async () => { + // Queue of pending stream factories — each call to the mocked + // chat.completions.create consumes one. + const streamQueue: Array< + () => AsyncIterable + > = []; + + streamQueue.push(() => + interleavingStream([ + () => openerChunk('call_A', 'read_file', '{"file_path":"/a'), + () => continuationChunk('/one.ts"}'), + () => finisherChunk(), + ]), + ); + streamQueue.push(() => + interleavingStream([ + () => openerChunk('call_B', 'read_file', '{"file_path":"/b'), + () => continuationChunk('/two.ts"}'), + () => finisherChunk(), + ]), + ); + + const { pipeline } = buildPipeline(() => { + const next = streamQueue.shift(); + if (!next) throw new Error('unexpected extra stream request'); + return next(); + }); + + const request: GenerateContentParameters = { + model: 'test-model', + contents: [{ role: 'user', parts: [{ text: 'read the files' }] }], + }; + + // Kick off both streams *before* consuming either, so the two generators + // are actually alive on the event loop at the same time. + const [streamA, streamB] = await Promise.all([ + pipeline.executeStream(request, 'prompt-a'), + pipeline.executeStream(request, 'prompt-b'), + ]); + + // Interleaved consumption: alternate one chunk from each to maximize + // parser state overlap. + const collectedA: unknown[] = []; + const collectedB: unknown[] = []; + + const aIter = streamA[Symbol.asyncIterator](); + const bIter = streamB[Symbol.asyncIterator](); + + while (true) { + const [aNext, bNext] = await Promise.all([aIter.next(), bIter.next()]); + if (!aNext.done) collectedA.push(aNext.value); + if (!bNext.done) collectedB.push(bNext.value); + if (aNext.done && bNext.done) break; + } + + const extractFunctionCall = (responses: unknown[]) => { + for (const resp of responses) { + const candidates = ( + resp as { candidates?: Array<{ content?: { parts?: Part[] } }> } + ).candidates; + const parts = candidates?.[0]?.content?.parts ?? []; + const fc = parts.find((p) => p.functionCall)?.functionCall; + if (fc) return fc; + } + return undefined; + }; + + const fnA = extractFunctionCall(collectedA); + const fnB = extractFunctionCall(collectedB); + + // Pre-fix behaviour: at least one of these would either be undefined + // (buffer wiped by the other stream's reset) or carry the wrong args + // (other stream's chunks merged into this bucket). + expect(fnA?.name).toBe('read_file'); + expect(fnA?.id).toBe('call_A'); + expect(fnA?.args).toEqual({ file_path: '/a/one.ts' }); + + expect(fnB?.name).toBe('read_file'); + expect(fnB?.id).toBe('call_B'); + expect(fnB?.args).toEqual({ file_path: '/b/two.ts' }); + }); + + it('an error in one stream does not poison a concurrent stream (no shared reset on error)', async () => { + // Stream A: normal tool call. + // Stream B: yields an `error_finish` chunk mid-flight, which the + // Pipeline wraps as StreamContentError. + // Pre-fix: the error path ran `resetStreamingToolCalls()` on the shared + // converter, wiping A's partial buffers. Post-fix: streamCtx is local + // to each generator, so A is untouched. + const streamQueue: Array< + () => AsyncIterable + > = []; + + streamQueue.push(() => + interleavingStream([ + () => openerChunk('call_A', 'read_file', '{"file_path":"/x'), + () => continuationChunk('.ts"}'), + () => finisherChunk(), + ]), + ); + + streamQueue.push(() => + interleavingStream([ + () => openerChunk('call_B', 'read_file', '{"file_path":"/y'), + // Inject an error_finish chunk — this triggers StreamContentError + // inside processStreamWithLogging's catch block. + () => + ({ + id: 'err', + object: 'chat.completion.chunk', + created: 1, + model: 'test', + choices: [ + { + index: 0, + delta: { content: 'rate limit' }, + finish_reason: 'error_finish', + logprobs: null, + }, + ], + }) as unknown as OpenAI.Chat.ChatCompletionChunk, + ]), + ); + + const { pipeline } = buildPipeline(() => { + const next = streamQueue.shift(); + if (!next) throw new Error('unexpected extra stream request'); + return next(); + }); + + const request: GenerateContentParameters = { + model: 'test-model', + contents: [{ role: 'user', parts: [{ text: 'read the files' }] }], + }; + + const [streamA, streamB] = await Promise.all([ + pipeline.executeStream(request, 'prompt-a'), + pipeline.executeStream(request, 'prompt-b'), + ]); + + const consumeA = (async () => { + const out: unknown[] = []; + for await (const r of streamA) out.push(r); + return out; + })(); + const consumeB = (async () => { + try { + for await (const _ of streamB) { + /* drain */ + } + return 'completed'; + } catch (e) { + return e instanceof Error ? e.message : String(e); + } + })(); + + const [aResults, bOutcome] = await Promise.all([consumeA, consumeB]); + + // Stream B blew up as expected. + expect(typeof bOutcome).toBe('string'); + expect(bOutcome).toContain('rate limit'); + + // Stream A still emitted its function call cleanly, despite B's error + // path running concurrently. On pre-fix code the error path would have + // called converter.resetStreamingToolCalls(), wiping A's in-flight + // buffer and causing A to emit zero function calls. + const fnA = (() => { + for (const resp of aResults) { + const parts = + (resp as { candidates?: Array<{ content?: { parts?: Part[] } }> }) + .candidates?.[0]?.content?.parts ?? []; + const fc = parts.find((p) => p.functionCall)?.functionCall; + if (fc) return fc; + } + return undefined; + })(); + + expect(fnA?.name).toBe('read_file'); + expect(fnA?.id).toBe('call_A'); + expect(fnA?.args).toEqual({ file_path: '/x.ts' }); + }); +}); diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts index 6969a51ef..6ecf2460e 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts @@ -52,7 +52,11 @@ describe('ContentGenerationPipeline', () => { convertOpenAIResponseToGemini: vi.fn(), convertOpenAIChunkToGemini: vi.fn(), convertGeminiToolsToOpenAI: vi.fn(), - resetStreamingToolCalls: vi.fn(), + createStreamContext: vi.fn().mockReturnValue({ + toolCallParser: { addChunk: vi.fn(), getCompletedToolCalls: vi.fn() }, + thinkBuffer: '', + inThinkTag: false, + }), } as unknown as OpenAIContentConverter; // Mock provider @@ -392,7 +396,7 @@ describe('ContentGenerationPipeline', () => { expect(results).toHaveLength(2); expect(results[0]).toBe(mockGeminiResponse1); expect(results[1]).toBe(mockGeminiResponse2); - expect(mockConverter.resetStreamingToolCalls).toHaveBeenCalled(); + expect(mockConverter.createStreamContext).toHaveBeenCalled(); expect(mockClient.chat.completions.create).toHaveBeenCalledWith( expect.objectContaining({ stream: true, @@ -504,7 +508,9 @@ describe('ContentGenerationPipeline', () => { } expect(results).toHaveLength(0); // No results due to error - expect(mockConverter.resetStreamingToolCalls).toHaveBeenCalledTimes(2); // Once at start, once on error + // Per-stream context is created once per stream entry; no manual reset + // on error path anymore (parser is GC'd with the context). + expect(mockConverter.createStreamContext).toHaveBeenCalledTimes(1); expect(mockErrorHandler.handle).toHaveBeenCalledWith( testError, expect.any(Object), diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts index 78acdcf87..e3ce33247 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.ts @@ -131,8 +131,12 @@ export class ContentGenerationPipeline { // Accumulate streamed response text for telemetry prompt logging const completionParts: string[] = []; - // Reset streaming tool calls to prevent data pollution from previous streams - this.converter.resetStreamingToolCalls(); + // Stream-local parser state. Previously the tool-call parser lived on + // the Converter singleton and was reset at stream start — but that + // wiped concurrent streams' in-flight buffers (e.g. parallel subagents + // sharing the same Config.contentGenerator). Scoping it per-stream + // fixes issue #3516. + const streamCtx = this.converter.createStreamContext(); // State for handling chunk merging. // pendingFinishResponse holds a finish chunk waiting to be merged with @@ -179,7 +183,10 @@ export class ContentGenerationPipeline { throw new StreamContentError(errorContent); } - const response = this.converter.convertOpenAIChunkToGemini(chunk); + const response = this.converter.convertOpenAIChunkToGemini( + chunk, + streamCtx, + ); // Stage 2b: Filter empty responses to avoid downstream issues if ( @@ -299,8 +306,8 @@ export class ContentGenerationPipeline { context.span.end(); } } catch (error) { - // Clear streaming tool calls on error to prevent data pollution - this.converter.resetStreamingToolCalls(); + // No manual parser cleanup needed — streamCtx is stream-local and + // will be garbage collected when this generator unwinds. // End the OTel span on error if it was deferred for streaming if (context.span) { diff --git a/packages/core/src/permissions/permission-manager.test.ts b/packages/core/src/permissions/permission-manager.test.ts index 5b1048d5c..01d9b2f64 100644 --- a/packages/core/src/permissions/permission-manager.test.ts +++ b/packages/core/src/permissions/permission-manager.test.ts @@ -186,7 +186,32 @@ describe('parseRule', () => { it('handles malformed pattern (no closing paren)', async () => { const r = parseRule('Bash(git status'); + expect(r.invalid).toBe(true); + expect(r.toolName).toBe('run_shell_command'); expect(r.specifier).toBeUndefined(); + // Must not match any command + expect(matchesRule(r, 'run_shell_command', 'git status')).toBe(false); + expect(matchesRule(r, 'run_shell_command', 'rm -rf /')).toBe(false); + }); + + it('handles malformed pattern with trailing junk after paren', async () => { + const r = parseRule('Bash(rm -rf /)*'); + expect(r.invalid).toBe(true); + expect(matchesRule(r, 'run_shell_command', 'git status')).toBe(false); + expect(matchesRule(r, 'run_shell_command', 'rm -rf /')).toBe(false); + }); + + it('handles malformed pattern with only open paren', async () => { + const r = parseRule('Bash('); + expect(r.invalid).toBe(true); + expect(matchesRule(r, 'run_shell_command', 'ls')).toBe(false); + }); + + it('still parses well-formed rules correctly', async () => { + const r = parseRule('Bash(rm -rf /)'); + expect(r.invalid).toBeUndefined(); + expect(matchesRule(r, 'run_shell_command', 'rm -rf /')).toBe(true); + expect(matchesRule(r, 'run_shell_command', 'git status')).toBe(false); }); }); @@ -1284,6 +1309,29 @@ describe('PermissionManager', () => { pm.addSessionDenyRule('run_shell_command'); expect(await pm.evaluate({ toolName: 'run_shell_command' })).toBe('deny'); }); + + it('malformed session allow rule is silently ignored', async () => { + pm.addSessionAllowRule('Bash(git commit'); + // 'git commit' is not readonly, so default is 'ask'. + // The malformed rule must not act as catch-all allow. + expect( + await pm.evaluate({ + toolName: 'run_shell_command', + command: 'git commit', + }), + ).toBe('ask'); + }); + + it('malformed session deny rule is silently ignored', async () => { + pm.addSessionDenyRule('Bash(rm -rf /)*'); + // Should NOT deny — the malformed rule must not act as catch-all + expect( + await pm.evaluate({ + toolName: 'run_shell_command', + command: 'git status', + }), + ).not.toBe('deny'); + }); }); describe('allowedTools via permissionsAllow', () => { @@ -1320,6 +1368,21 @@ describe('PermissionManager', () => { ); expect(sessionAllow?.rule.toolName).toBe('run_shell_command'); }); + + it('excludes malformed rules from listing', async () => { + pm = new PermissionManager( + makeConfig({ + permissionsAllow: ['ReadFileTool'], + permissionsDeny: ['Bash(rm -rf /)*'], + }), + ); + pm.initialize(); + + const rules = pm.listRules(); + // The malformed deny rule should be filtered out + expect(rules.length).toBe(1); + expect(rules[0]!.rule.toolName).toBe('read_file'); + }); }); describe('hasMatchingAskRule', () => { diff --git a/packages/core/src/permissions/permission-manager.ts b/packages/core/src/permissions/permission-manager.ts index 35e92b838..810ceab1f 100644 --- a/packages/core/src/permissions/permission-manager.ts +++ b/packages/core/src/permissions/permission-manager.ts @@ -16,6 +16,7 @@ import { extractShellOperations } from './shell-semantics.js'; import type { ShellOperation } from './shell-semantics.js'; import { isShellCommandReadOnlyAST } from '../utils/shellAstParser.js'; import { detectCommandSubstitution } from '../utils/shell-utils.js'; +import { createDebugLogger } from '../utils/debugLogger.js'; import type { PermissionCheckContext, PermissionDecision, @@ -27,6 +28,8 @@ import type { } from './types.js'; import type { AutoApproveClassifier } from './auto-approve-classifier.js'; +const debugLogger = createDebugLogger('PERMISSIONS'); + /** * Numeric priority for each PermissionDecision. * Higher number = more restrictive. Used to combine decisions by taking @@ -710,7 +713,14 @@ export class PermissionManager { */ addSessionAllowRule(raw: string): void { if (raw && raw.trim()) { - this.sessionRules.allow.push(parseRule(raw)); + const rule = parseRule(raw); + if (rule.invalid) { + debugLogger.warn( + `Ignoring malformed allow rule (unbalanced parentheses): ${rule.raw}`, + ); + return; + } + this.sessionRules.allow.push(rule); } } @@ -719,7 +729,14 @@ export class PermissionManager { */ addSessionDenyRule(raw: string): void { if (raw && raw.trim()) { - this.sessionRules.deny.push(parseRule(raw)); + const rule = parseRule(raw); + if (rule.invalid) { + debugLogger.warn( + `Ignoring malformed deny rule (unbalanced parentheses): ${rule.raw}`, + ); + return; + } + this.sessionRules.deny.push(rule); } } @@ -728,7 +745,14 @@ export class PermissionManager { */ addSessionAskRule(raw: string): void { if (raw && raw.trim()) { - this.sessionRules.ask.push(parseRule(raw)); + const rule = parseRule(raw); + if (rule.invalid) { + debugLogger.warn( + `Ignoring malformed ask rule (unbalanced parentheses): ${rule.raw}`, + ); + return; + } + this.sessionRules.ask.push(rule); } } @@ -747,6 +771,12 @@ export class PermissionManager { */ addPersistentRule(raw: string, type: RuleType): PermissionRule { const rule = parseRule(raw); + if (rule.invalid) { + debugLogger.warn( + `Ignoring malformed ${type} rule (unbalanced parentheses): ${rule.raw}`, + ); + return rule; + } // Deduplicate: skip if a rule with the same raw string already exists const exists = this.persistentRules[type].some((r) => r.raw === rule.raw); if (!exists) { @@ -817,7 +847,9 @@ export class PermissionManager { scope: RuleScope, ) => { for (const rule of rules) { - result.push({ rule, type, scope }); + if (!rule.invalid) { + result.push({ rule, type, scope }); + } } }; diff --git a/packages/core/src/permissions/rule-parser.ts b/packages/core/src/permissions/rule-parser.ts index 992f75a33..f8f4d4748 100644 --- a/packages/core/src/permissions/rule-parser.ts +++ b/packages/core/src/permissions/rule-parser.ts @@ -7,6 +7,9 @@ import path from 'node:path'; import os from 'node:os'; import picomatch from 'picomatch'; +import { createDebugLogger } from '../utils/debugLogger.js'; + +const debugLogger = createDebugLogger('PERMISSIONS'); /** * Normalize a filesystem path to use POSIX-style forward slashes. @@ -271,10 +274,13 @@ export function parseRule(raw: string): PermissionRule { } const toolPart = normalized.substring(0, openParen).trim(); - const specifier = normalized.endsWith(')') - ? normalized.substring(openParen + 1, normalized.length - 1) - : undefined; + if (!normalized.endsWith(')')) { + // Malformed: unbalanced parentheses — mark as invalid so it never matches. + return { raw: trimmed, toolName: resolveToolName(toolPart), invalid: true }; + } + + const specifier = normalized.substring(openParen + 1, normalized.length - 1); const canonicalName = resolveToolName(toolPart); const specifierKind = specifier ? getSpecifierKind(canonicalName) : undefined; @@ -291,7 +297,17 @@ export function parseRule(raw: string): PermissionRule { * silently skipping any empty entries. */ export function parseRules(raws: string[]): PermissionRule[] { - return raws.filter((r) => r && r.trim()).map(parseRule); + return raws + .filter((r) => r && r.trim()) + .map(parseRule) + .map((r) => { + if (r.invalid) { + debugLogger.warn( + `Ignoring malformed rule (unbalanced parentheses): ${r.raw}`, + ); + } + return r; + }); } // ───────────────────────────────────────────────────────────────────────────── @@ -931,6 +947,11 @@ export function matchesRule( ): boolean { const canonicalCtxToolName = resolveToolName(toolName); + // ── Invalid (malformed) rules never match anything ────────────────── + if (rule.invalid) { + return false; + } + // ── MCP tool matching ──────────────────────────────────────────────── if ( rule.toolName.startsWith('mcp__') || diff --git a/packages/core/src/permissions/types.ts b/packages/core/src/permissions/types.ts index 01d919cba..d7c2ff296 100644 --- a/packages/core/src/permissions/types.ts +++ b/packages/core/src/permissions/types.ts @@ -59,6 +59,8 @@ export interface PermissionRule { * Set automatically during parsing based on the tool name/category. */ specifierKind?: SpecifierKind; + /** True if the raw rule was malformed (e.g. unbalanced parens) and should never match. */ + invalid?: boolean; } /** A complete set of permission rules organized by type. */ diff --git a/packages/core/src/services/sessionService.test.ts b/packages/core/src/services/sessionService.test.ts index 58ff1f235..3cfa2ee56 100644 --- a/packages/core/src/services/sessionService.test.ts +++ b/packages/core/src/services/sessionService.test.ts @@ -717,5 +717,126 @@ describe('SessionService', () => { postCompressionRecord.message, ]); }); + + it('should preserve thought parts by default (stripThoughtsFromHistory=false)', () => { + const modelWithThought: ChatRecord = { + uuid: 't1', + parentUuid: 'a1', + sessionId: sessionIdA, + timestamp: '2024-01-01T01:00:00Z', + type: 'assistant', + message: { + role: 'model', + parts: [ + { text: 'reasoning step', thought: true }, + { text: 'final answer' }, + ], + }, + cwd: '/test/project/root', + version: '1.0.0', + }; + + const conversation: ConversationRecord = { + sessionId: sessionIdA, + projectHash: 'test-project-hash', + startTime: '2024-01-01T00:00:00Z', + lastUpdated: '2024-01-01T01:00:00Z', + messages: [recordA1, modelWithThought], + }; + + const history = buildApiHistoryFromConversation(conversation); + + // Thought parts should be preserved by default + expect(history).toHaveLength(2); + expect(history[1].parts).toEqual([ + { text: 'reasoning step', thought: true }, + { text: 'final answer' }, + ]); + }); + + it('should strip thought parts when stripThoughtsFromHistory=true', () => { + const modelWithThought: ChatRecord = { + uuid: 't1', + parentUuid: 'a1', + sessionId: sessionIdA, + timestamp: '2024-01-01T01:00:00Z', + type: 'assistant', + message: { + role: 'model', + parts: [ + { text: 'reasoning step', thought: true }, + { text: 'final answer' }, + ], + }, + cwd: '/test/project/root', + version: '1.0.0', + }; + + const conversation: ConversationRecord = { + sessionId: sessionIdA, + projectHash: 'test-project-hash', + startTime: '2024-01-01T00:00:00Z', + lastUpdated: '2024-01-01T01:00:00Z', + messages: [recordA1, modelWithThought], + }; + + const history = buildApiHistoryFromConversation(conversation, { + stripThoughtsFromHistory: true, + }); + + // Thought parts should be stripped + expect(history).toHaveLength(2); + expect(history[1].parts).toEqual([{ text: 'final answer' }]); + }); + + it('should preserve thought parts in compressed history by default', () => { + const compressionRecord: ChatRecord = { + uuid: 'c1', + parentUuid: 'b2', + sessionId: sessionIdA, + timestamp: '2024-01-02T03:00:00Z', + type: 'system', + subtype: 'chat_compression', + cwd: '/test/project/root', + version: '1.0.0', + gitBranch: 'main', + systemPayload: { + info: { + originalTokenCount: 100, + newTokenCount: 50, + compressionStatus: CompressionStatus.COMPRESSED, + }, + compressedHistory: [ + { role: 'user', parts: [{ text: 'summary' }] }, + { + role: 'model', + parts: [ + { text: 'deep thinking', thought: true }, + { text: 'final answer' }, + ], + }, + ], + }, + }; + + const conversation: ConversationRecord = { + sessionId: sessionIdA, + projectHash: 'test-project-hash', + startTime: '2024-01-01T00:00:00Z', + lastUpdated: '2024-01-02T03:00:00Z', + messages: [recordA1, recordB2, compressionRecord], + }; + + const history = buildApiHistoryFromConversation(conversation); + + // Thought parts should be preserved in compressed history by default. + // The compressedHistory has 2 entries (user, model), and no messages + // exist after the compression record, so the result is 2 items. + expect(history).toHaveLength(2); + expect(history[1].parts).toEqual([ + { text: 'deep thinking', thought: true }, + { text: 'final answer' }, + ]); + }); }); }); diff --git a/packages/core/src/services/sessionService.ts b/packages/core/src/services/sessionService.ts index e5d88be87..0f78b6131 100644 --- a/packages/core/src/services/sessionService.ts +++ b/packages/core/src/services/sessionService.ts @@ -552,7 +552,9 @@ export interface BuildApiHistoryOptions { /** * Whether to strip thought parts from the history. * Thought parts are content parts that have `thought: true`. - * @default true + * Keeping thoughts ensures `reasoning_content` from reasoning models + * (e.g. DeepSeek) is properly passed back in subsequent API calls. + * @default false */ stripThoughtsFromHistory?: boolean; } @@ -593,7 +595,7 @@ export function buildApiHistoryFromConversation( conversation: ConversationRecord, options: BuildApiHistoryOptions = {}, ): Content[] { - const { stripThoughtsFromHistory = true } = options; + const { stripThoughtsFromHistory = false } = options; const { messages } = conversation; let lastCompressionIndex = -1; diff --git a/packages/core/src/skills/skill-manager.test.ts b/packages/core/src/skills/skill-manager.test.ts index d17308024..163f311d7 100644 --- a/packages/core/src/skills/skill-manager.test.ts +++ b/packages/core/src/skills/skill-manager.test.ts @@ -6,9 +6,14 @@ import { vi, describe, it, expect, beforeEach, afterEach } from 'vitest'; import * as fs from 'fs/promises'; +import * as fsSync from 'fs'; import * as path from 'path'; import * as os from 'os'; -import { SkillManager } from './skill-manager.js'; +import { + SkillManager, + WATCHER_MAX_DEPTH, + watcherIgnored, +} from './skill-manager.js'; import { type SkillConfig, SkillError } from './types.js'; import type { Config } from '../config/config.js'; import { makeFakeConfig } from '../test-utils/config.js'; @@ -17,6 +22,24 @@ import { makeFakeConfig } from '../test-utils/config.js'; vi.mock('fs/promises'); vi.mock('os'); +const { mockWatch, mockWatcher } = vi.hoisted(() => { + const mockWatcher = { + on: vi.fn().mockReturnThis(), + close: vi.fn().mockResolvedValue(undefined), + }; + const mockWatch = vi.fn().mockReturnValue(mockWatcher); + return { mockWatch, mockWatcher }; +}); + +vi.mock('chokidar', () => ({ + watch: mockWatch, +})); + +vi.mock('fs', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, existsSync: vi.fn(actual.existsSync) }; +}); + // Mock yaml parser - use vi.hoisted for proper hoisting const mockParseYaml = vi.hoisted(() => vi.fn()); @@ -902,4 +925,56 @@ Symlinked skill content`); ]); }); }); + + describe('file watchers', () => { + it('should pass ignored function and shallow depth to chokidar', async () => { + const projectSkillsDir = path.join('/test/project', '.qwen', 'skills'); + vi.mocked(fsSync.existsSync).mockImplementation( + (p) => String(p) === projectSkillsDir, + ); + + vi.mocked(fs.readdir).mockResolvedValue( + [] as unknown as Awaited>, + ); + + mockWatch.mockClear(); + mockWatcher.on.mockClear(); + + await manager.startWatching(); + + expect(mockWatch).toHaveBeenCalledWith(projectSkillsDir, { + ignoreInitial: true, + ignored: watcherIgnored, + depth: WATCHER_MAX_DEPTH, + }); + expect(WATCHER_MAX_DEPTH).toBe(2); + }); + + it('watcherIgnored should reject .git directories', () => { + expect(watcherIgnored(path.join('/skills', '.git', 'config'))).toBe(true); + expect(watcherIgnored(path.join('/skills', '.git'))).toBe(true); + expect(watcherIgnored(path.join('/skills', 'my-skill', 'SKILL.md'))).toBe( + false, + ); + }); + + it('watcherIgnored should reject special file types', () => { + const socketStats = { + isFile: () => false, + isDirectory: () => false, + } as fsSync.Stats; + const fileStats = { + isFile: () => true, + isDirectory: () => false, + } as fsSync.Stats; + const dirStats = { + isFile: () => false, + isDirectory: () => true, + } as fsSync.Stats; + + expect(watcherIgnored('/skills/some.sock', socketStats)).toBe(true); + expect(watcherIgnored('/skills/SKILL.md', fileStats)).toBe(false); + expect(watcherIgnored('/skills/my-skill', dirStats)).toBe(false); + }); + }); }); diff --git a/packages/core/src/skills/skill-manager.ts b/packages/core/src/skills/skill-manager.ts index 4b838889c..3d3c3e035 100644 --- a/packages/core/src/skills/skill-manager.ts +++ b/packages/core/src/skills/skill-manager.ts @@ -30,6 +30,21 @@ const QWEN_CONFIG_DIR = '.qwen'; const SKILLS_CONFIG_DIR = 'skills'; const SKILL_MANIFEST_FILE = 'SKILL.md'; +// Skills have a fixed layout (/SKILL.md), so depth 2 is enough to +// detect any change. This keeps chokidar out of heavy subtrees like node_modules +// that would otherwise exhaust file descriptors. +export const WATCHER_MAX_DEPTH = 2; + +// Reject special file types (sockets, FIFOs, devices) that cannot be watched +// and would error with EOPNOTSUPP, plus .git directories. +export function watcherIgnored( + filePath: string, + stats?: fsSync.Stats, +): boolean { + if (stats && !stats.isFile() && !stats.isDirectory()) return true; + return filePath.split(path.sep).includes('.git'); +} + /** * Manages skill configurations stored as directories containing SKILL.md files. * Provides discovery, parsing, validation, and caching for skills. @@ -674,6 +689,8 @@ export class SkillManager { try { const watcher = watchFs(watchPath, { ignoreInitial: true, + ignored: watcherIgnored, + depth: WATCHER_MAX_DEPTH, }) .on('all', () => { this.scheduleRefresh(); diff --git a/packages/core/src/telemetry/file-exporters.test.ts b/packages/core/src/telemetry/file-exporters.test.ts new file mode 100644 index 000000000..9c68a7c91 --- /dev/null +++ b/packages/core/src/telemetry/file-exporters.test.ts @@ -0,0 +1,50 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { FileSpanExporter } from './file-exporters.js'; + +type SerializeAccess = { serialize: (data: unknown) => string }; + +describe('FileExporter.serialize', () => { + let tmpDir: string; + let exporter: FileSpanExporter; + let serialize: (data: unknown) => string; + + beforeEach(() => { + tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'file-exporters-test-')); + exporter = new FileSpanExporter(path.join(tmpDir, 'out.jsonl')); + serialize = (exporter as unknown as SerializeAccess).serialize.bind( + exporter, + ); + }); + + afterEach(async () => { + await exporter.shutdown(); + fs.rmSync(tmpDir, { recursive: true, force: true }); + }); + + // A raw JSON.stringify on a ReadableSpan crashed because + // BatchSpanProcessor._shutdownOnce -> BindOnceFuture._that forms a cycle. + // The exporter must delegate to safeJsonStringify so cycles become + // "[Circular]" instead of throwing. + it('does not throw on BatchSpanProcessor-shaped cycle', () => { + const proc: Record = { kind: 'BatchSpanProcessor' }; + const future: Record = { kind: 'BindOnceFuture' }; + proc['_shutdownOnce'] = future; + future['_that'] = proc; + const span = { name: 'span-1', _spanProcessor: proc }; + + expect(() => serialize(span)).not.toThrow(); + const out = serialize(span); + expect(out).toContain('"name": "span-1"'); + expect(out).toContain('"[Circular]"'); + expect(out.endsWith('\n')).toBe(true); + }); +}); diff --git a/packages/core/src/telemetry/file-exporters.ts b/packages/core/src/telemetry/file-exporters.ts index 55fe64103..def6e91f4 100644 --- a/packages/core/src/telemetry/file-exporters.ts +++ b/packages/core/src/telemetry/file-exporters.ts @@ -17,6 +17,7 @@ import type { PushMetricExporter, } from '@opentelemetry/sdk-metrics'; import { AggregationTemporality } from '@opentelemetry/sdk-metrics'; +import { safeJsonStringify } from '../utils/safeJsonStringify.js'; class FileExporter { protected writeStream: fs.WriteStream; @@ -26,7 +27,7 @@ class FileExporter { } protected serialize(data: unknown): string { - return JSON.stringify(data, null, 2) + '\n'; + return safeJsonStringify(data, 2) + '\n'; } shutdown(): Promise {