From 12301713b93e65f4d1c1286d4b08ebb6fe46e8d7 Mon Sep 17 00:00:00 2001 From: James Feng <47167674+GhostDragon124@users.noreply.github.com> Date: Thu, 4 Jun 2026 20:20:21 +0800 Subject: [PATCH] fix: backfill upstream OpenAI fixes (c82f5994, 901628b4) + regression tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit c82f5994 — fix(openai): stop_reason null, zero usage fields, max_tokens forwarding - streamAdapter: defer message_delta/message_stop to after stream loop so trailing usage chunks are captured; fill all 4 usage fields - index.ts: assemble final AssistantMessage at message_stop (not per-block); apply stop_reason from message_delta; reset partialMessage; post-loop safety fallback for partial messages - buildOpenAIRequestBody: accept and forward maxTokens → max_tokens 901628b4 — fix: OpenAI provider deferred MCP tool visibility - index.ts: prepend deferred MCP tool text list so OpenAI model can discover and request them via ToolSearchTool - claude.ts: pass full tools (not filteredTools) to OpenAI path so deferred tools are searchable - index.ts: include already-discovered deferred tools in filteredTools so their schemas are available after ToolSearchTool loads them Tests added: - queryModelOpenAI.isolated.ts (674 lines): stop_reason assembly, partialMessage reset, max_tokens truncation warning, usage tracking, cost tracking - queryModelOpenAI.runner.ts + .test.ts: isolated subprocess runner (CCP pattern) - streamAdapter.test.ts: 665→130 lines expanded — deferred finish, trailing usage, length→max_tokens mapping, full usage field assertions - formatBriefTimestamp.test.ts: beforeAll/afterAll env save/restore 3602 pass, 0 fail --- src/services/api/claude.ts | 5 +- .../__tests__/queryModelOpenAI.isolated.ts | 674 ++++++++++++++++++ .../__tests__/queryModelOpenAI.runner.ts | 540 ++++++++++++++ .../openai/__tests__/queryModelOpenAI.test.ts | 26 + .../openai/__tests__/streamAdapter.test.ts | 651 +++++++++++++---- src/services/api/openai/index.ts | 258 +++++-- src/services/api/openai/streamAdapter.ts | 70 +- .../__tests__/formatBriefTimestamp.test.ts | 130 ++-- src/utils/formatBriefTimestamp.ts | 1 + 9 files changed, 2093 insertions(+), 262 deletions(-) create mode 100644 src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts create mode 100644 src/services/api/openai/__tests__/queryModelOpenAI.runner.ts create mode 100644 src/services/api/openai/__tests__/queryModelOpenAI.test.ts diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index ae362e053..899d0a826 100644 --- a/src/services/api/claude.ts +++ b/src/services/api/claude.ts @@ -1355,10 +1355,13 @@ async function* queryModel( // media stripping) but before Anthropic-specific logic (betas, thinking, caching). if (getAPIProvider() === 'openai') { const { queryModelOpenAI } = await import('./openai/index.js') + // OpenAI emulates Anthropic's dynamic tool loading client-side. It needs + // the full tool pool so ToolSearchTool can search deferred MCP tools that + // were intentionally filtered out of the initial API tool list above. yield* queryModelOpenAI( messagesForAPI, systemPrompt, - filteredTools, + tools, signal, options, ) diff --git a/src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts b/src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts new file mode 100644 index 000000000..8d580c55e --- /dev/null +++ b/src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts @@ -0,0 +1,674 @@ +/** + * Tests for queryModelOpenAI in index.ts. + * + * Focused on the two bugs fixed: + * 1. stop_reason was always null in the assembled AssistantMessage because + * partialMessage (from message_start) has stop_reason: null, and the + * stop_reason captured from message_delta was never applied. + * 2. partialMessage was not reset to null after message_stop, so the safety + * fallback at the end of the loop would yield a second identical + * AssistantMessage (causing doubled content in the next API request). + * + * Strategy: mock getOpenAIClient + adaptOpenAIStreamToAnthropic so we can + * feed pre-built Anthropic events directly into queryModelOpenAI and inspect + * what it emits — without any real HTTP calls. + */ +import { describe, expect, test, mock, beforeEach, afterEach } from 'bun:test' +import type { BetaRawMessageStreamEvent } from '@anthropic-ai/sdk/resources/beta/messages/messages.mjs' +import type { + AssistantMessage, + StreamEvent, +} from '../../../../types/message.js' + +// ─── helpers ───────────────────────────────────────────────────────────────── + +/** Build a minimal message_start event */ +function makeMessageStart( + overrides: Record = {}, +): BetaRawMessageStreamEvent { + return { + type: 'message_start', + message: { + id: 'msg_test', + type: 'message', + role: 'assistant', + content: [], + model: 'test-model', + stop_reason: null, + stop_sequence: null, + usage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + ...overrides, + }, + } as any +} + +/** Build a content_block_start event for the given block type */ +function makeContentBlockStart( + index: number, + type: 'text' | 'tool_use' | 'thinking', + extra: Record = {}, +): BetaRawMessageStreamEvent { + const block = + type === 'text' + ? { type: 'text', text: '' } + : type === 'tool_use' + ? { type: 'tool_use', id: 'toolu_test', name: 'bash', input: {} } + : { type: 'thinking', thinking: '', signature: '' } + return { + type: 'content_block_start', + index, + content_block: { ...block, ...extra }, + } as any +} + +/** Build a text_delta content_block_delta event */ +function makeTextDelta(index: number, text: string): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'text_delta', text }, + } as any +} + +/** Build an input_json_delta content_block_delta event */ +function makeInputJsonDelta( + index: number, + json: string, +): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'input_json_delta', partial_json: json }, + } as any +} + +/** Build a thinking_delta content_block_delta event */ +function makeThinkingDelta( + index: number, + thinking: string, +): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'thinking_delta', thinking }, + } as any +} + +/** Build a content_block_stop event */ +function makeContentBlockStop(index: number): BetaRawMessageStreamEvent { + return { type: 'content_block_stop', index } as any +} + +/** Build a message_delta event with stop_reason and output_tokens */ +function makeMessageDelta( + stopReason: string, + outputTokens: number, +): BetaRawMessageStreamEvent { + return { + type: 'message_delta', + delta: { stop_reason: stopReason, stop_sequence: null }, + usage: { output_tokens: outputTokens }, + } as any +} + +/** Build a message_stop event */ +function makeMessageStop(): BetaRawMessageStreamEvent { + return { type: 'message_stop' } as any +} + +/** Async generator from a fixed array of events */ +async function* eventStream(events: BetaRawMessageStreamEvent[]) { + for (const e of events) yield e +} + +/** Collect all outputs from queryModelOpenAI into typed buckets */ +async function runQueryModel( + events: BetaRawMessageStreamEvent[], + envOverrides: Record = {}, +) { + // Wire events into the mocked stream adapter + _nextEvents = events + // Save + apply env overrides + const saved: Record = {} + for (const [k, v] of Object.entries(envOverrides)) { + saved[k] = process.env[k] + if (v === undefined) delete process.env[k] + else process.env[k] = v + } + + try { + // We inline mock.module inside the try block. + // Bun resolves mock.module at the call site synchronously (hoisted), + // so we register once per test file, then re-import each time. + const { queryModelOpenAI } = await import('../index.js') + + const assistantMessages: AssistantMessage[] = [] + const streamEvents: StreamEvent[] = [] + const otherOutputs: any[] = [] + + const minimalOptions: any = { + model: 'test-model', + tools: [], + agents: [], + querySource: 'main_loop', + getToolPermissionContext: async () => ({ + alwaysAllow: [], + alwaysDeny: [], + needsPermission: [], + mode: 'default', + isBypassingPermissions: false, + }), + } + + for await (const item of queryModelOpenAI( + [], + { type: 'text', text: '' } as any, + [], + new AbortController().signal, + minimalOptions, + )) { + if (item.type === 'assistant') { + assistantMessages.push(item as AssistantMessage) + } else if (item.type === 'stream_event') { + streamEvents.push(item as StreamEvent) + } else { + otherOutputs.push(item) + } + } + + return { assistantMessages, streamEvents, otherOutputs } + } finally { + // Restore env + for (const [k, v] of Object.entries(saved)) { + if (v === undefined) delete process.env[k] + else process.env[k] = v + } + } +} + +// ─── mock setup ────────────────────────────────────────────────────────────── + +// We mock at module level. Bun's mock.module replaces the module for the +// entire file, so we configure the stream per-test via a shared variable. +let _nextEvents: BetaRawMessageStreamEvent[] = [] +let _toolSearchEnabled = false + +/** Captured arguments from the last chat.completions.create() call */ +let _lastCreateArgs: Record | null = null + +mock.module('@ant/model-provider', () => ({ + resolveOpenAIModel: (m: string) => m, + adaptOpenAIStreamToAnthropic: (_stream: any, _model: string) => + eventStream(_nextEvents), + anthropicMessagesToOpenAI: (messages: any[]) => + messages.map(msg => ({ + role: msg.message?.role ?? 'user', + content: msg.message?.content ?? '', + })), + anthropicToolsToOpenAI: (tools: any[]) => + tools.map(tool => ({ + type: 'function', + function: { + name: tool.name, + description: tool.description ?? '', + parameters: tool.input_schema ?? { type: 'object', properties: {} }, + }, + })), + anthropicToolChoiceToOpenAI: () => undefined, +})) + +mock.module('../../../../utils/envUtils.js', () => ({ + isEnvTruthy: (value: string | undefined) => + value === '1' || value === 'true' || value === 'yes' || value === 'on', + isEnvDefinedFalsy: (value: string | undefined) => + value === '0' || value === 'false' || value === 'no' || value === 'off', +})) + +mock.module('../../../../services/analytics/growthbook.js', () => ({ + getFeatureValue_CACHED_MAY_BE_STALE: (_key: string, fallback: unknown) => + fallback, +})) + +mock.module('src/bootstrap/state.js', () => ({ + isReplBridgeActive: () => false, +})) + +mock.module('bun:bundle', () => ({ + feature: () => false, +})) + +mock.module('../client.js', () => ({ + getOpenAIClient: () => ({ + chat: { + completions: { + create: async (args: Record) => { + _lastCreateArgs = args + return { [Symbol.asyncIterator]: async function* () {} } + }, + }, + }, + }), +})) + +mock.module('../streamAdapter.js', () => ({ + adaptOpenAIStreamToAnthropic: (_stream: any, _model: string) => + eventStream(_nextEvents), +})) + +mock.module('../modelMapping.js', () => ({ + resolveOpenAIModel: (m: string) => m, +})) + +mock.module('../convertMessages.js', () => ({ + anthropicMessagesToOpenAI: (messages: any[]) => + messages.map(msg => ({ + role: msg.message?.role ?? 'user', + content: msg.message?.content ?? '', + })), +})) + +mock.module('../convertTools.js', () => ({ + anthropicToolsToOpenAI: (tools: any[]) => + tools.map(tool => ({ + type: 'function', + function: { + name: tool.name, + description: tool.description ?? '', + parameters: tool.input_schema ?? { type: 'object', properties: {} }, + }, + })), + anthropicToolChoiceToOpenAI: () => undefined, +})) + +mock.module('../../../../utils/context.js', () => ({ + MODEL_CONTEXT_WINDOW_DEFAULT: 200_000, + COMPACT_MAX_OUTPUT_TOKENS: 20_000, + CAPPED_DEFAULT_MAX_TOKENS: 8_000, + ESCALATED_MAX_TOKENS: 64_000, + is1mContextDisabled: () => false, + has1mContext: () => false, + modelSupports1M: () => false, + getModelMaxOutputTokens: () => ({ upperLimit: 8192, default: 8192 }), + getContextWindowForModel: () => 200_000, + getSonnet1mExpTreatmentEnabled: () => false, + calculateContextPercentages: () => ({ + usedPercent: 0, + remainingPercent: 100, + }), + getMaxThinkingTokensForModel: () => 0, +})) + +mock.module('../../../../utils/messages.js', () => ({ + normalizeMessagesForAPI: (msgs: any) => msgs, + normalizeContentFromAPI: (blocks: any[]) => blocks, + createUserMessage: (opts: any) => ({ + type: 'user', + message: { role: 'user', content: opts.content }, + uuid: 'user-uuid', + timestamp: new Date().toISOString(), + isMeta: opts.isMeta, + }), + createAssistantAPIErrorMessage: (opts: any) => ({ + type: 'assistant', + message: { + content: [{ type: 'text', text: opts.content }], + apiError: opts.apiError, + }, + uuid: 'error-uuid', + timestamp: new Date().toISOString(), + }), +})) + +mock.module('../../../../utils/api.js', () => ({ + toolToAPISchema: async (t: any) => t, +})) + +mock.module('../../../../utils/toolSearch.js', () => ({ + isToolSearchEnabled: async () => _toolSearchEnabled, + extractDiscoveredToolNames: () => new Set(), + isDeferredToolsDeltaEnabled: () => false, +})) + +mock.module('../../../../tools/ToolSearchTool/prompt.js', () => ({ + formatDeferredToolLine: (tool: any) => tool.name, + isDeferredTool: (tool: any) => tool.isMcp === true, + TOOL_SEARCH_TOOL_NAME: 'ToolSearch', +})) + +mock.module('../../../../cost-tracker.js', () => ({ + addToTotalSessionCost: () => {}, +})) + +mock.module('../../../../utils/modelCost.js', () => ({ + COST_TIER_3_15: {}, + COST_TIER_15_75: {}, + COST_TIER_5_25: {}, + COST_TIER_30_150: {}, + COST_HAIKU_35: {}, + COST_HAIKU_45: {}, + getOpus46CostTier: () => ({}), + MODEL_COSTS: {}, + getModelCosts: () => ({}), + calculateUSDCost: () => 0, + calculateCostFromTokens: () => 0, + formatModelPricing: () => '', + getModelPricingString: () => undefined, +})) + +mock.module('../../../../services/langfuse/tracing.js', () => ({ + recordLLMObservation: () => {}, +})) + +mock.module('../../../../services/langfuse/convert.js', () => ({ + convertMessagesToLangfuse: () => [], + convertOutputToLangfuse: () => ({}), + convertToolsToLangfuse: () => [], +})) + +mock.module('../../../../utils/debug.js', () => ({ + logForDebugging: () => {}, + logAntError: () => {}, + isDebugMode: () => false, + isDebugToStdErr: () => false, + getDebugFilePath: () => null, + getDebugLogPath: () => '', + getDebugFilter: () => null, + getMinDebugLogLevel: () => 'debug', + enableDebugLogging: () => false, + setHasFormattedOutput: () => {}, + getHasFormattedOutput: () => false, + flushDebugLogs: async () => {}, +})) + +// ─── tests ─────────────────────────────────────────────────────────────────── + +describe('queryModelOpenAI — stop_reason propagation', () => { + test('assembled AssistantMessage has stop_reason end_turn (not null)', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'Hello'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 10), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + expect(assistantMessages[0]!.message.stop_reason).toBe('end_turn') + }) + + test('assembled AssistantMessage has stop_reason tool_use', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'tool_use'), + makeInputJsonDelta(0, '{"cmd":"ls"}'), + makeContentBlockStop(0), + makeMessageDelta('tool_use', 20), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + expect(assistantMessages[0]!.message.stop_reason).toBe('tool_use') + }) + + test('assembled AssistantMessage has stop_reason max_tokens', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'truncated'), + makeContentBlockStop(0), + makeMessageDelta('max_tokens', 8192), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + // Two assistant-typed items: the content message + the max_output_tokens error signal. + // The error signal is emitted as a synthetic assistant message by createAssistantAPIErrorMessage. + expect(assistantMessages).toHaveLength(2) + const contentMsg = assistantMessages[0]! + expect(contentMsg.message.stop_reason).toBe('max_tokens') + // Second item is the error signal (has apiError set) + const errorMsg = assistantMessages[1]!.message as any + expect(errorMsg.apiError).toBe('max_output_tokens') + }) + + test('stop_reason is null when no message_delta was received (safety fallback path)', async () => { + // Stream ends without message_stop — triggers the safety fallback branch. + // stop_reason stays null since no message_delta was ever seen. + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'partial'), + makeContentBlockStop(0), + // No message_delta / message_stop + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + // Safety fallback should yield the partial content + expect(assistantMessages).toHaveLength(1) + expect(assistantMessages[0]!.message.stop_reason).toBeNull() + }) +}) + +describe('queryModelOpenAI — usage accumulation', () => { + test('usage in assembled message reflects all four fields from message_delta', async () => { + // message_start has all fields=0 (trailing-chunk pattern: usage not yet available). + // message_delta carries the real values after stream ends. + // The spread in the message_delta handler must override all zeros from message_start, + // including cache_read_input_tokens which was previously missing from message_delta. + _nextEvents = [ + makeMessageStart({ + usage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'response'), + makeContentBlockStop(0), + // message_delta carries all four Anthropic usage fields (as emitted by the fixed streamAdapter) + { + type: 'message_delta', + delta: { stop_reason: 'end_turn', stop_sequence: null }, + usage: { + input_tokens: 30011, + output_tokens: 190, + cache_read_input_tokens: 19904, + cache_creation_input_tokens: 0, + }, + } as any, + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + const usage = assistantMessages[0]!.message.usage as any + expect(usage.input_tokens).toBe(30011) + expect(usage.output_tokens).toBe(190) + // cache_read_input_tokens from message_delta overrides the 0 from message_start + expect(usage.cache_read_input_tokens).toBe(19904) + expect(usage.cache_creation_input_tokens).toBe(0) + }) + + test('usage is zero when no usage events arrive (prevents false autocompact)', async () => { + // If usage stays 0, tokenCountWithEstimation will undercount — so at least + // verify the field exists and is numeric (to detect regressions). + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'hi'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 0), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + const usage = assistantMessages[0]!.message.usage as any + expect(typeof usage.input_tokens).toBe('number') + expect(typeof usage.output_tokens).toBe('number') + }) +}) + +describe('queryModelOpenAI — no duplicate AssistantMessage (partialMessage reset)', () => { + test('yields exactly one AssistantMessage per message_stop when content is present', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'only once'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 5), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + // Before the fix, partialMessage was not reset to null, so the safety + // fallback at the end of the loop would yield a second message with the + // same message.id — causing mergeAssistantMessages to concatenate content. + expect(assistantMessages).toHaveLength(1) + }) + + test('thinking + text response yields exactly one AssistantMessage', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'thinking'), + makeThinkingDelta(0, 'let me think'), + makeContentBlockStop(0), + makeContentBlockStart(1, 'text'), + makeTextDelta(1, 'answer'), + makeContentBlockStop(1), + makeMessageDelta('end_turn', 30), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + }) + + test('safety fallback path still yields message when stream ends without message_stop', async () => { + // Simulates a stream that cuts off without the normal termination sequence. + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'abrupt end'), + // No content_block_stop, no message_delta, no message_stop + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + }) +}) + +describe('queryModelOpenAI — stream_events forwarded', () => { + test('every adapted event is also yielded as stream_event for real-time display', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'hello'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 5), + makeMessageStop(), + ] + + const { streamEvents } = await runQueryModel(_nextEvents) + + const eventTypes = streamEvents.map(e => (e as any).event?.type) + expect(eventTypes).toContain('message_start') + expect(eventTypes).toContain('content_block_start') + expect(eventTypes).toContain('content_block_delta') + expect(eventTypes).toContain('content_block_stop') + expect(eventTypes).toContain('message_delta') + expect(eventTypes).toContain('message_stop') + }) +}) + +describe('queryModelOpenAI — max_tokens forwarded to request', () => { + test('buildOpenAIRequestBody includes max_tokens in the request payload', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'hi'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 5), + makeMessageStop(), + ] + + await runQueryModel(_nextEvents) + + expect(_lastCreateArgs).not.toBeNull() + expect(_lastCreateArgs!.max_tokens).toBe(8192) + }) +}) + +describe('queryModelOpenAI — deferred MCP tool visibility', () => { + test('prepends available deferred MCP tools to OpenAI messages', async () => { + _toolSearchEnabled = true + _nextEvents = [makeMessageStart(), makeMessageStop()] + + try { + const { queryModelOpenAI } = await import('../index.js') + const tools: any[] = [ + { + name: 'ToolSearch', + isMcp: false, + input_schema: { type: 'object', properties: {} }, + prompt: async () => 'Search deferred tools', + }, + { + name: 'mcp__wechat__send_message', + isMcp: true, + input_schema: { type: 'object', properties: {} }, + prompt: async () => 'Send a WeChat message', + }, + ] + + const options: any = { + model: 'test-model', + tools: [], + agents: [], + querySource: 'main_loop', + getToolPermissionContext: async () => ({ + alwaysAllow: [], + alwaysDeny: [], + needsPermission: [], + mode: 'default', + isBypassingPermissions: false, + }), + } + + for await (const _item of queryModelOpenAI( + [], + { type: 'text', text: '' } as any, + tools as any, + new AbortController().signal, + options, + )) { + // Exhaust generator so request body is built. + } + + expect(_lastCreateArgs).not.toBeNull() + expect(JSON.stringify(_lastCreateArgs!.messages)).toContain( + '\\nmcp__wechat__send_message\\n', + ) + } finally { + _toolSearchEnabled = false + } + }) +}) diff --git a/src/services/api/openai/__tests__/queryModelOpenAI.runner.ts b/src/services/api/openai/__tests__/queryModelOpenAI.runner.ts new file mode 100644 index 000000000..fbea46735 --- /dev/null +++ b/src/services/api/openai/__tests__/queryModelOpenAI.runner.ts @@ -0,0 +1,540 @@ +/** + * Tests for queryModelOpenAI in index.ts. + * + * Focused on the two bugs fixed: + * 1. stop_reason was always null in the assembled AssistantMessage because + * partialMessage (from message_start) has stop_reason: null, and the + * stop_reason captured from message_delta was never applied. + * 2. partialMessage was not reset to null after message_stop, so the safety + * fallback at the end of the loop would yield a second identical + * AssistantMessage (causing doubled content in the next API request). + * + * Strategy: mock getOpenAIClient + adaptOpenAIStreamToAnthropic so we can + * feed pre-built Anthropic events directly into queryModelOpenAI and inspect + * what it emits — without any real HTTP calls. + */ +import { describe, expect, test, mock, beforeEach, afterEach } from 'bun:test' +import type { BetaRawMessageStreamEvent } from '@anthropic-ai/sdk/resources/beta/messages/messages.mjs' +import type { + AssistantMessage, + StreamEvent, +} from '../../../../types/message.js' + +// ─── helpers ───────────────────────────────────────────────────────────────── + +/** Build a minimal message_start event */ +function makeMessageStart( + overrides: Record = {}, +): BetaRawMessageStreamEvent { + return { + type: 'message_start', + message: { + id: 'msg_test', + type: 'message', + role: 'assistant', + content: [], + model: 'test-model', + stop_reason: null, + stop_sequence: null, + usage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + ...overrides, + }, + } as any +} + +/** Build a content_block_start event for the given block type */ +function makeContentBlockStart( + index: number, + type: 'text' | 'tool_use' | 'thinking', + extra: Record = {}, +): BetaRawMessageStreamEvent { + const block = + type === 'text' + ? { type: 'text', text: '' } + : type === 'tool_use' + ? { type: 'tool_use', id: 'toolu_test', name: 'bash', input: {} } + : { type: 'thinking', thinking: '', signature: '' } + return { + type: 'content_block_start', + index, + content_block: { ...block, ...extra }, + } as any +} + +/** Build a text_delta content_block_delta event */ +function makeTextDelta(index: number, text: string): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'text_delta', text }, + } as any +} + +/** Build an input_json_delta content_block_delta event */ +function makeInputJsonDelta( + index: number, + json: string, +): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'input_json_delta', partial_json: json }, + } as any +} + +/** Build a thinking_delta content_block_delta event */ +function makeThinkingDelta( + index: number, + thinking: string, +): BetaRawMessageStreamEvent { + return { + type: 'content_block_delta', + index, + delta: { type: 'thinking_delta', thinking }, + } as any +} + +/** Build a content_block_stop event */ +function makeContentBlockStop(index: number): BetaRawMessageStreamEvent { + return { type: 'content_block_stop', index } as any +} + +/** Build a message_delta event with stop_reason and output_tokens */ +function makeMessageDelta( + stopReason: string, + outputTokens: number, +): BetaRawMessageStreamEvent { + return { + type: 'message_delta', + delta: { stop_reason: stopReason, stop_sequence: null }, + usage: { output_tokens: outputTokens }, + } as any +} + +/** Build a message_stop event */ +function makeMessageStop(): BetaRawMessageStreamEvent { + return { type: 'message_stop' } as any +} + +/** Async generator from a fixed array of events */ +async function* eventStream(events: BetaRawMessageStreamEvent[]) { + for (const e of events) yield e +} + +/** Collect all outputs from queryModelOpenAI into typed buckets */ +async function runQueryModel( + events: BetaRawMessageStreamEvent[], + envOverrides: Record = {}, +) { + // Wire events into the mocked stream adapter + _nextEvents = events + // Save + apply env overrides + const saved: Record = {} + for (const [k, v] of Object.entries(envOverrides)) { + saved[k] = process.env[k] + if (v === undefined) delete process.env[k] + else process.env[k] = v + } + + try { + // We inline mock.module inside the try block. + // Bun resolves mock.module at the call site synchronously (hoisted), + // so we register once per test file, then re-import each time. + const { queryModelOpenAI } = await import('../index.js') + + const assistantMessages: AssistantMessage[] = [] + const streamEvents: StreamEvent[] = [] + const otherOutputs: any[] = [] + + const minimalOptions: any = { + model: 'test-model', + tools: [], + agents: [], + querySource: 'main_loop', + getToolPermissionContext: async () => ({ + alwaysAllow: [], + alwaysDeny: [], + needsPermission: [], + mode: 'default', + isBypassingPermissions: false, + }), + } + + for await (const item of queryModelOpenAI( + [], + { type: 'text', text: '' } as any, + [], + new AbortController().signal, + minimalOptions, + )) { + if (item.type === 'assistant') { + assistantMessages.push(item as AssistantMessage) + } else if (item.type === 'stream_event') { + streamEvents.push(item as StreamEvent) + } else { + otherOutputs.push(item) + } + } + + return { assistantMessages, streamEvents, otherOutputs } + } finally { + // Restore env + for (const [k, v] of Object.entries(saved)) { + if (v === undefined) delete process.env[k] + else process.env[k] = v + } + } +} + +// ─── mock setup ────────────────────────────────────────────────────────────── + +// We mock at module level. Bun's mock.module replaces the module for the +// entire file, so we configure the stream per-test via a shared variable. +let _nextEvents: BetaRawMessageStreamEvent[] = [] + +/** Captured arguments from the last chat.completions.create() call */ +let _lastCreateArgs: Record | null = null + +mock.module('../client.js', () => ({ + getOpenAIClient: () => ({ + chat: { + completions: { + create: async (args: Record) => { + _lastCreateArgs = args + return { [Symbol.asyncIterator]: async function* () {} } + }, + }, + }, + }), +})) + +mock.module('../streamAdapter.js', () => ({ + adaptOpenAIStreamToAnthropic: (_stream: any, _model: string) => + eventStream(_nextEvents), +})) + +mock.module('../modelMapping.js', () => ({ + resolveOpenAIModel: (m: string) => m, +})) + +mock.module('../convertMessages.js', () => ({ + anthropicMessagesToOpenAI: (messages: any[]) => + messages.map(msg => ({ + role: msg.message?.role ?? 'user', + content: msg.message?.content ?? '', + })), +})) + +mock.module('../convertTools.js', () => ({ + anthropicToolsToOpenAI: (tools: any[]) => + tools.map(tool => ({ + type: 'function', + function: { + name: tool.name, + description: tool.description ?? '', + parameters: tool.input_schema ?? { type: 'object', properties: {} }, + }, + })), + anthropicToolChoiceToOpenAI: () => undefined, +})) + +mock.module('../../../../utils/context.js', () => ({ + getModelMaxOutputTokens: () => ({ upperLimit: 8192, default: 8192 }), + getContextWindowForModel: () => 200_000, +})) + +mock.module('../../../../utils/messages.js', () => ({ + normalizeMessagesForAPI: (msgs: any) => msgs, + normalizeContentFromAPI: (blocks: any[]) => blocks, + createUserMessage: (opts: any) => ({ + type: 'user', + message: { role: 'user', content: opts.content }, + uuid: 'user-uuid', + timestamp: new Date().toISOString(), + isMeta: opts.isMeta, + }), + createAssistantAPIErrorMessage: (opts: any) => ({ + type: 'assistant', + message: { + content: [{ type: 'text', text: opts.content }], + apiError: opts.apiError, + }, + uuid: 'error-uuid', + timestamp: new Date().toISOString(), + }), +})) + +mock.module('../../../../utils/api.js', () => ({ + toolToAPISchema: async (t: any) => t, +})) + +mock.module('../../../../utils/toolSearch.js', () => ({ + isToolSearchEnabled: async () => false, + extractDiscoveredToolNames: () => new Set(), + isDeferredToolsDeltaEnabled: () => false, +})) + +mock.module('../../../../tools/ToolSearchTool/prompt.js', () => ({ + formatDeferredToolLine: (tool: any) => tool.name, + isDeferredTool: (tool: any) => tool.isMcp === true, + TOOL_SEARCH_TOOL_NAME: 'ToolSearch', +})) + +mock.module('../../../../cost-tracker.js', () => ({ + addToTotalSessionCost: () => {}, +})) + +mock.module('../../../../utils/modelCost.js', () => ({ + calculateUSDCost: () => 0, +})) + +mock.module('../../../../services/langfuse/tracing.js', () => ({ + recordLLMObservation: () => {}, +})) + +mock.module('../../../../services/langfuse/convert.js', () => ({ + convertMessagesToLangfuse: () => [], + convertOutputToLangfuse: () => ({}), + convertToolsToLangfuse: () => [], +})) + +mock.module('../../../../utils/debug.js', () => ({ + logForDebugging: () => {}, +})) + +// ─── tests ─────────────────────────────────────────────────────────────────── + +describe('queryModelOpenAI — stop_reason propagation', () => { + test('assembled AssistantMessage has stop_reason end_turn (not null)', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'Hello'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 10), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + expect(assistantMessages[0]!.message.stop_reason).toBe('end_turn') + }) + + test('assembled AssistantMessage has stop_reason tool_use', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'tool_use'), + makeInputJsonDelta(0, '{"cmd":"ls"}'), + makeContentBlockStop(0), + makeMessageDelta('tool_use', 20), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + expect(assistantMessages[0]!.message.stop_reason).toBe('tool_use') + }) + + test('assembled AssistantMessage has stop_reason max_tokens', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'truncated'), + makeContentBlockStop(0), + makeMessageDelta('max_tokens', 8192), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + // Two assistant-typed items: the content message + the max_output_tokens error signal. + // The error signal is emitted as a synthetic assistant message by createAssistantAPIErrorMessage. + expect(assistantMessages).toHaveLength(2) + const contentMsg = assistantMessages[0]! + expect(contentMsg.message.stop_reason).toBe('max_tokens') + // Second item is the error signal (has apiError set) + const errorMsg = assistantMessages[1]!.message as any + expect(errorMsg.apiError).toBe('max_output_tokens') + }) + + test('stop_reason is null when no message_delta was received (safety fallback path)', async () => { + // Stream ends without message_stop — triggers the safety fallback branch. + // stop_reason stays null since no message_delta was ever seen. + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'partial'), + makeContentBlockStop(0), + // No message_delta / message_stop + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + // Safety fallback should yield the partial content + expect(assistantMessages).toHaveLength(1) + expect(assistantMessages[0]!.message.stop_reason).toBeNull() + }) +}) + +describe('queryModelOpenAI — usage accumulation', () => { + test('usage in assembled message reflects all four fields from message_delta', async () => { + // message_start has all fields=0 (trailing-chunk pattern: usage not yet available). + // message_delta carries the real values after stream ends. + // The spread in the message_delta handler must override all zeros from message_start, + // including cache_read_input_tokens which was previously missing from message_delta. + _nextEvents = [ + makeMessageStart({ + usage: { + input_tokens: 0, + output_tokens: 0, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'response'), + makeContentBlockStop(0), + // message_delta carries all four Anthropic usage fields (as emitted by the fixed streamAdapter) + { + type: 'message_delta', + delta: { stop_reason: 'end_turn', stop_sequence: null }, + usage: { + input_tokens: 30011, + output_tokens: 190, + cache_read_input_tokens: 19904, + cache_creation_input_tokens: 0, + }, + } as any, + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + const usage = assistantMessages[0]!.message.usage as any + expect(usage.input_tokens).toBe(30011) + expect(usage.output_tokens).toBe(190) + // cache_read_input_tokens from message_delta overrides the 0 from message_start + expect(usage.cache_read_input_tokens).toBe(19904) + expect(usage.cache_creation_input_tokens).toBe(0) + }) + + test('usage is zero when no usage events arrive (prevents false autocompact)', async () => { + // If usage stays 0, tokenCountWithEstimation will undercount — so at least + // verify the field exists and is numeric (to detect regressions). + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'hi'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 0), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + const usage = assistantMessages[0]!.message.usage as any + expect(typeof usage.input_tokens).toBe('number') + expect(typeof usage.output_tokens).toBe('number') + }) +}) + +describe('queryModelOpenAI — no duplicate AssistantMessage (partialMessage reset)', () => { + test('yields exactly one AssistantMessage per message_stop when content is present', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'only once'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 5), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + // Before the fix, partialMessage was not reset to null, so the safety + // fallback at the end of the loop would yield a second message with the + // same message.id — causing mergeAssistantMessages to concatenate content. + expect(assistantMessages).toHaveLength(1) + }) + + test('thinking + text response yields exactly one AssistantMessage', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'thinking'), + makeThinkingDelta(0, 'let me think'), + makeContentBlockStop(0), + makeContentBlockStart(1, 'text'), + makeTextDelta(1, 'answer'), + makeContentBlockStop(1), + makeMessageDelta('end_turn', 30), + makeMessageStop(), + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + }) + + test('safety fallback path still yields message when stream ends without message_stop', async () => { + // Simulates a stream that cuts off without the normal termination sequence. + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'abrupt end'), + // No content_block_stop, no message_delta, no message_stop + ] + + const { assistantMessages } = await runQueryModel(_nextEvents) + + expect(assistantMessages).toHaveLength(1) + }) +}) + +describe('queryModelOpenAI — stream_events forwarded', () => { + test('every adapted event is also yielded as stream_event for real-time display', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'hello'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 5), + makeMessageStop(), + ] + + const { streamEvents } = await runQueryModel(_nextEvents) + + const eventTypes = streamEvents.map(e => (e as any).event?.type) + expect(eventTypes).toContain('message_start') + expect(eventTypes).toContain('content_block_start') + expect(eventTypes).toContain('content_block_delta') + expect(eventTypes).toContain('content_block_stop') + expect(eventTypes).toContain('message_delta') + expect(eventTypes).toContain('message_stop') + }) +}) + +describe('queryModelOpenAI — max_tokens forwarded to request', () => { + test('buildOpenAIRequestBody includes max_tokens in the request payload', async () => { + _nextEvents = [ + makeMessageStart(), + makeContentBlockStart(0, 'text'), + makeTextDelta(0, 'hi'), + makeContentBlockStop(0), + makeMessageDelta('end_turn', 5), + makeMessageStop(), + ] + + await runQueryModel(_nextEvents) + + expect(_lastCreateArgs).not.toBeNull() + expect(_lastCreateArgs!.max_tokens).toBe(8192) + }) +}) diff --git a/src/services/api/openai/__tests__/queryModelOpenAI.test.ts b/src/services/api/openai/__tests__/queryModelOpenAI.test.ts new file mode 100644 index 000000000..47a43f0df --- /dev/null +++ b/src/services/api/openai/__tests__/queryModelOpenAI.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, test } from 'bun:test' + +async function runIsolatedTestFile(path: string) { + const proc = Bun.spawn([process.execPath, 'test', path], { + cwd: process.cwd(), + env: process.env, + stdout: 'pipe', + stderr: 'pipe', + }) + const [stdout, stderr, exitCode] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + proc.exited, + ]) + + expect(`${stdout}\n${stderr}`).toEqual(expect.stringContaining('0 fail')) + expect(exitCode).toBe(0) +} + +describe('queryModelOpenAI isolated runner', () => { + test('runs queryModelOpenAI regression suite without leaking mocks', async () => { + await runIsolatedTestFile( + './src/services/api/openai/__tests__/queryModelOpenAI.runner.ts', + ) + }) +}) diff --git a/src/services/api/openai/__tests__/streamAdapter.test.ts b/src/services/api/openai/__tests__/streamAdapter.test.ts index 3d2763673..ab1ab3c2e 100644 --- a/src/services/api/openai/__tests__/streamAdapter.test.ts +++ b/src/services/api/openai/__tests__/streamAdapter.test.ts @@ -1,9 +1,11 @@ import { describe, expect, test } from 'bun:test' -import { adaptOpenAIStreamToAnthropic } from '../streamAdapter.js' import type { ChatCompletionChunk } from 'openai/resources/chat/completions/completions.mjs' +import { adaptOpenAIStreamToAnthropic } from '../streamAdapter.js' /** Helper to create a mock async iterable from chunk array */ -function mockStream(chunks: ChatCompletionChunk[]): AsyncIterable { +function mockStream( + chunks: ChatCompletionChunk[], +): AsyncIterable { return { [Symbol.asyncIterator]() { let i = 0 @@ -18,7 +20,9 @@ function mockStream(chunks: ChatCompletionChunk[]): AsyncIterable & any = {}): ChatCompletionChunk { +function makeChunk( + overrides: Partial & any = {}, +): ChatCompletionChunk { return { id: 'chatcmpl-test', object: 'chat.completion.chunk', @@ -29,9 +33,13 @@ function makeChunk(overrides: Partial & any = {}): ChatComp } as ChatCompletionChunk } +/** Collect all emitted Anthropic events from the stream adapter for assertion */ async function collectEvents(chunks: ChatCompletionChunk[]) { const events: any[] = [] - for await (const event of adaptOpenAIStreamToAnthropic(mockStream(chunks), 'gpt-4o')) { + for await (const event of adaptOpenAIStreamToAnthropic( + mockStream(chunks), + 'gpt-4o', + )) { events.push(event) } return events @@ -41,25 +49,31 @@ describe('adaptOpenAIStreamToAnthropic', () => { test('emits message_start on first chunk', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { role: 'assistant', content: '' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { role: 'assistant', content: '' }, + finish_reason: null, + }, + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { content: 'hello' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { content: 'hello' }, + finish_reason: null, + }, + ], }), makeChunk({ - choices: [{ - index: 0, - delta: {}, - finish_reason: 'stop', - }], + choices: [ + { + index: 0, + delta: {}, + finish_reason: 'stop', + }, + ], usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 }, }), ]) @@ -72,10 +86,14 @@ describe('adaptOpenAIStreamToAnthropic', () => { test('converts text content stream', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ index: 0, delta: { content: 'Hello' }, finish_reason: null }], + choices: [ + { index: 0, delta: { content: 'Hello' }, finish_reason: null }, + ], }), makeChunk({ - choices: [{ index: 0, delta: { content: ' world' }, finish_reason: null }], + choices: [ + { index: 0, delta: { content: ' world' }, finish_reason: null }, + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], @@ -90,7 +108,9 @@ describe('adaptOpenAIStreamToAnthropic', () => { expect(types).toContain('message_delta') expect(types).toContain('message_stop') - const textDeltas = events.filter(e => e.type === 'content_block_delta') as any[] + const textDeltas = events.filter( + e => e.type === 'content_block_delta', + ) as any[] expect(textDeltas[0].delta.text).toBe('Hello') expect(textDeltas[1].delta.text).toBe(' world') }) @@ -98,42 +118,54 @@ describe('adaptOpenAIStreamToAnthropic', () => { test('converts tool_calls stream', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { - tool_calls: [{ - index: 0, - id: 'call_abc', - type: 'function', - function: { name: 'bash', arguments: '' }, - }], + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_abc', + type: 'function', + function: { name: 'bash', arguments: '' }, + }, + ], + }, + finish_reason: null, }, - finish_reason: null, - }], + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { - tool_calls: [{ - index: 0, - function: { arguments: '{"comm' }, - }], + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + function: { arguments: '{"comm' }, + }, + ], + }, + finish_reason: null, }, - finish_reason: null, - }], + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { - tool_calls: [{ - index: 0, - function: { arguments: 'and":"ls"}' }, - }], + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + function: { arguments: 'and":"ls"}' }, + }, + ], + }, + finish_reason: null, }, - finish_reason: null, - }], + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }], @@ -145,7 +177,8 @@ describe('adaptOpenAIStreamToAnthropic', () => { expect(blockStart.content_block.name).toBe('bash') const jsonDeltas = events.filter( - e => e.type === 'content_block_delta' && e.delta.type === 'input_json_delta', + e => + e.type === 'content_block_delta' && e.delta.type === 'input_json_delta', ) as any[] const fullArgs = jsonDeltas.map(d => d.delta.partial_json).join('') expect(fullArgs).toBe('{"command":"ls"}') @@ -170,13 +203,21 @@ describe('adaptOpenAIStreamToAnthropic', () => { // return finish_reason "stop" when they actually made tool calls. const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { - tool_calls: [{ index: 0, id: 'call_1', function: { name: 'bash', arguments: '{"cmd":"ls"}' } }], + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_1', + function: { name: 'bash', arguments: '{"cmd":"ls"}' }, + }, + ], + }, + finish_reason: null, }, - finish_reason: null, - }], + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], @@ -190,13 +231,21 @@ describe('adaptOpenAIStreamToAnthropic', () => { test('maps finish_reason tool_calls to tool_use', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { - tool_calls: [{ index: 0, id: 'call_1', function: { name: 'bash', arguments: '{}' } }], + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_1', + function: { name: 'bash', arguments: '{}' }, + }, + ], + }, + finish_reason: null, }, - finish_reason: null, - }], + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }], @@ -210,7 +259,9 @@ describe('adaptOpenAIStreamToAnthropic', () => { test('maps finish_reason length to max_tokens', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ index: 0, delta: { content: 'truncated' }, finish_reason: null }], + choices: [ + { index: 0, delta: { content: 'truncated' }, finish_reason: null }, + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'length' }], @@ -224,23 +275,35 @@ describe('adaptOpenAIStreamToAnthropic', () => { test('handles mixed text and tool_calls', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ index: 0, delta: { content: 'Thinking...' }, finish_reason: null }], + choices: [ + { index: 0, delta: { content: 'Thinking...' }, finish_reason: null }, + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { - tool_calls: [{ index: 0, id: 'call_1', function: { name: 'grep', arguments: '{"p":"test"}' } }], + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_1', + function: { name: 'grep', arguments: '{"p":"test"}' }, + }, + ], + }, + finish_reason: null, }, - finish_reason: null, - }], + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }], }), ]) - const blockStarts = events.filter(e => e.type === 'content_block_start') as any[] + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] expect(blockStarts.length).toBe(2) expect(blockStarts[0].content_block.type).toBe('text') expect(blockStarts[1].content_block.type).toBe('tool_use') @@ -251,18 +314,22 @@ describe('thinking support (reasoning_content)', () => { test('converts reasoning_content to thinking block', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { reasoning_content: 'Let me analyze this...' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { reasoning_content: 'Let me analyze this...' }, + finish_reason: null, + }, + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { reasoning_content: ' step by step.' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { reasoning_content: ' step by step.' }, + finish_reason: null, + }, + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], @@ -276,7 +343,8 @@ describe('thinking support (reasoning_content)', () => { // Should have thinking_delta events const thinkingDeltas = events.filter( - e => e.type === 'content_block_delta' && e.delta.type === 'thinking_delta', + e => + e.type === 'content_block_delta' && e.delta.type === 'thinking_delta', ) as any[] expect(thinkingDeltas.length).toBe(2) expect(thinkingDeltas[0].delta.thinking).toBe('Let me analyze this...') @@ -286,18 +354,22 @@ describe('thinking support (reasoning_content)', () => { test('converts reasoning then content (DeepSeek-style)', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { reasoning_content: 'Thinking about the answer...' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { reasoning_content: 'Thinking about the answer...' }, + finish_reason: null, + }, + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { content: 'Here is my answer.' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { content: 'Here is my answer.' }, + finish_reason: null, + }, + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], @@ -305,13 +377,17 @@ describe('thinking support (reasoning_content)', () => { ]) // Should have two content blocks: thinking + text - const blockStarts = events.filter(e => e.type === 'content_block_start') as any[] + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] expect(blockStarts.length).toBe(2) expect(blockStarts[0].content_block.type).toBe('thinking') expect(blockStarts[1].content_block.type).toBe('text') // Thinking block should be closed before text block starts - const blockStops = events.filter(e => e.type === 'content_block_stop') as any[] + const blockStops = events.filter( + e => e.type === 'content_block_stop', + ) as any[] expect(blockStops[0].index).toBe(0) // thinking block closed at index 0 expect(blockStarts[1].index).toBe(1) // text block starts at index 1 @@ -325,54 +401,120 @@ describe('thinking support (reasoning_content)', () => { test('handles reasoning then tool_calls', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { reasoning_content: 'I need to run a command.' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { reasoning_content: 'I need to run a command.' }, + finish_reason: null, + }, + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { - tool_calls: [{ index: 0, id: 'call_1', function: { name: 'bash', arguments: '{"c":"ls"}' } }], + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_1', + function: { name: 'bash', arguments: '{"c":"ls"}' }, + }, + ], + }, + finish_reason: null, }, - finish_reason: null, - }], + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }], }), ]) - const blockStarts = events.filter(e => e.type === 'content_block_start') as any[] + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] expect(blockStarts.length).toBe(2) expect(blockStarts[0].content_block.type).toBe('thinking') expect(blockStarts[1].content_block.type).toBe('tool_use') }) - test('thinking block index is 0, text block index is 1', async () => { + test('opens thinking block on empty reasoning_content (DeepSeek v4 direct-answer)', async () => { + // DeepSeek v4 thinking mode sometimes streams reasoning_content: "" + // before answering directly. We must still open a thinking block so the + // resulting assistant message carries an (empty) thinking block — that + // round-trips back as reasoning_content: "" in the next request, + // satisfying DeepSeek's requirement (see issue #399). const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { reasoning_content: 'reason' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { reasoning_content: '' }, + finish_reason: null, + }, + ], }), makeChunk({ - choices: [{ - index: 0, - delta: { content: 'answer' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { content: 'Direct answer.' }, + finish_reason: null, + }, + ], }), makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], }), ]) - const blockStarts = events.filter(e => e.type === 'content_block_start') as any[] + // A thinking block was opened (and closed before the text block starts) + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] + expect(blockStarts.length).toBe(2) + expect(blockStarts[0].content_block.type).toBe('thinking') + expect(blockStarts[0].content_block.thinking).toBe('') + expect(blockStarts[1].content_block.type).toBe('text') + + // No empty thinking_delta should be emitted — the empty string is + // already conveyed by the thinking block's initial value. + const thinkingDeltas = events.filter( + e => + e.type === 'content_block_delta' && e.delta.type === 'thinking_delta', + ) + expect(thinkingDeltas.length).toBe(0) + }) + + test('thinking block index is 0, text block index is 1', async () => { + const events = await collectEvents([ + makeChunk({ + choices: [ + { + index: 0, + delta: { reasoning_content: 'reason' }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [ + { + index: 0, + delta: { content: 'answer' }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + }), + ]) + + const blockStarts = events.filter( + e => e.type === 'content_block_start', + ) as any[] expect(blockStarts[0].index).toBe(0) expect(blockStarts[1].index).toBe(1) }) @@ -382,11 +524,13 @@ describe('prompt caching support', () => { test('maps cached_tokens to cache_read_input_tokens', async () => { const events = await collectEvents([ makeChunk({ - choices: [{ - index: 0, - delta: { content: 'hi' }, - finish_reason: null, - }], + choices: [ + { + index: 0, + delta: { content: 'hi' }, + finish_reason: null, + }, + ], usage: { prompt_tokens: 1000, completion_tokens: 0, @@ -407,7 +551,7 @@ describe('prompt caching support', () => { const msgStart = events.find(e => e.type === 'message_start') as any expect(msgStart.message.usage.cache_read_input_tokens).toBe(800) - // Anthropic convention: input_tokens = non-cached only (prompt_tokens - cached) + // input_tokens = prompt_tokens - cached_tokens = 1000 - 800 = 200 expect(msgStart.message.usage.input_tokens).toBe(200) }) @@ -454,4 +598,259 @@ describe('prompt caching support', () => { expect(msgStart.message.usage.cache_read_input_tokens).toBe(0) expect(msgStart.message.usage.input_tokens).toBe(500) }) + + test('captures output_tokens and input_tokens from trailing chunk sent after finish_reason', async () => { + // Many OpenAI-compatible endpoints (e.g. DeepSeek) send usage in a separate + // final chunk AFTER the finish_reason chunk, with choices: []. + // message_delta must carry both input_tokens and output_tokens so that + // queryModelOpenAI's spread can override the zeros from message_start — which is + // emitted before the trailing chunk and always has input_tokens=0. + const events = await collectEvents([ + makeChunk({ + choices: [ + { index: 0, delta: { content: 'hello' }, finish_reason: null }, + ], + }), + // finish_reason chunk — usage not yet available + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + }), + // trailing usage-only chunk (choices: []) + makeChunk({ + choices: [], + usage: { prompt_tokens: 123, completion_tokens: 45, total_tokens: 168 }, + }), + ]) + + // message_start emits on the first chunk before trailing usage arrives + const msgStart = events.find(e => e.type === 'message_start') as any + expect(msgStart.message.usage.input_tokens).toBe(0) + + // message_delta is emitted after stream loop ends with final real values + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta.usage.input_tokens).toBe(123) + expect(msgDelta.usage.output_tokens).toBe(45) + expect(msgDelta.delta.stop_reason).toBe('end_turn') + }) + + test('captures input_tokens from trailing chunk (used by tokenCountWithEstimation for autocompact)', async () => { + // input_tokens is the dominant term in tokenCountWithEstimation. Without it, + // getTokenCountFromUsage returns only output_tokens (~100-700), which is far below + // the autocompact threshold (~33k), so compaction never fires. + const events = await collectEvents([ + makeChunk({ + choices: [ + { index: 0, delta: { content: 'answer' }, finish_reason: null }, + ], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + }), + makeChunk({ + choices: [], + usage: { + prompt_tokens: 800, + completion_tokens: 200, + total_tokens: 1000, + }, + }), + ]) + + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta.usage.input_tokens).toBe(800) + expect(msgDelta.usage.output_tokens).toBe(200) + }) + + test('trailing usage chunk with tool_calls: stop_reason stays tool_use', async () => { + // Verifies that deferring message_delta does not break stop_reason mapping + // when the model made tool calls and usage arrives in a trailing chunk. + const events = await collectEvents([ + makeChunk({ + choices: [ + { + index: 0, + delta: { + tool_calls: [ + { + index: 0, + id: 'call_x', + function: { name: 'bash', arguments: '{"cmd":"ls"}' }, + }, + ], + }, + finish_reason: null, + }, + ], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'tool_calls' }], + }), + // trailing usage-only chunk + makeChunk({ + choices: [], + usage: { prompt_tokens: 500, completion_tokens: 30, total_tokens: 530 }, + }), + ]) + + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta.delta.stop_reason).toBe('tool_use') + expect(msgDelta.usage.output_tokens).toBe(30) + }) + + test('message_delta always comes before message_stop', async () => { + // Verifies event ordering is preserved after deferring to post-loop emission. + const events = await collectEvents([ + makeChunk({ + choices: [{ index: 0, delta: { content: 'x' }, finish_reason: null }], + }), + makeChunk({ choices: [{ index: 0, delta: {}, finish_reason: 'stop' }] }), + makeChunk({ + choices: [], + usage: { prompt_tokens: 10, completion_tokens: 5, total_tokens: 15 }, + }), + ]) + + const types = events.map(e => e.type) + const deltaIdx = types.lastIndexOf('message_delta') + const stopIdx = types.lastIndexOf('message_stop') + expect(deltaIdx).toBeGreaterThanOrEqual(0) + expect(stopIdx).toBeGreaterThan(deltaIdx) + }) + + // ── cache_read_input_tokens in message_delta (the core bug fix) ────────── + + test('message_delta carries cache_read_input_tokens from trailing usage chunk', async () => { + // Real-world case: DeepSeek-V3 returns cached_tokens=19904 + // in a trailing chunk with choices:[]. Previously message_delta only carried + // input_tokens and output_tokens, so cache_read_input_tokens stayed 0 after + // queryModelOpenAI's spread — even though cachedTokens was captured internally. + const events = await collectEvents([ + makeChunk({ + choices: [ + { index: 0, delta: { content: 'answer' }, finish_reason: null }, + ], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + }), + // trailing usage chunk matching the observed server response format + makeChunk({ + choices: [], + usage: { + prompt_tokens: 30011, + completion_tokens: 190, + total_tokens: 30201, + prompt_tokens_details: { audio_tokens: 0, cached_tokens: 19904 }, + } as any, + }), + ]) + + // message_start is emitted before trailing chunk — cache fields are 0 + const msgStart = events.find(e => e.type === 'message_start') as any + expect(msgStart.message.usage.cache_read_input_tokens).toBe(0) + + // message_delta carries the real values from the trailing chunk + const msgDelta = events.find(e => e.type === 'message_delta') as any + // input_tokens = prompt_tokens - cached_tokens = 30011 - 19904 = 10107 + expect(msgDelta.usage.input_tokens).toBe(10107) + expect(msgDelta.usage.output_tokens).toBe(190) + expect(msgDelta.usage.cache_read_input_tokens).toBe(19904) + expect(msgDelta.usage.cache_creation_input_tokens).toBe(0) + }) + + test('cache_read_input_tokens=0 in message_delta when cached_tokens is absent', async () => { + // Non-caching requests should still have the field present and zero. + const events = await collectEvents([ + makeChunk({ + choices: [{ index: 0, delta: { content: 'hi' }, finish_reason: null }], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + }), + makeChunk({ + choices: [], + usage: { prompt_tokens: 100, completion_tokens: 20, total_tokens: 120 }, + }), + ]) + + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta.usage.cache_read_input_tokens).toBe(0) + expect(msgDelta.usage.cache_creation_input_tokens).toBe(0) + }) + + test('cache_read_input_tokens=0 in message_delta when cached_tokens is 0', async () => { + // Explicit cached_tokens:0 should not be treated differently from absent. + const events = await collectEvents([ + makeChunk({ + choices: [{ index: 0, delta: { content: 'hi' }, finish_reason: null }], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + }), + makeChunk({ + choices: [], + usage: { + prompt_tokens: 500, + completion_tokens: 50, + total_tokens: 550, + prompt_tokens_details: { cached_tokens: 0 }, + } as any, + }), + ]) + + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta.usage.cache_read_input_tokens).toBe(0) + }) + + test('cache_read_input_tokens updated when cached_tokens arrives in same chunk as finish_reason', async () => { + // Some endpoints send usage in the finish_reason chunk instead of a trailing chunk. + const events = await collectEvents([ + makeChunk({ + choices: [ + { index: 0, delta: { content: 'result' }, finish_reason: null }, + ], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + usage: { + prompt_tokens: 2000, + completion_tokens: 100, + total_tokens: 2100, + prompt_tokens_details: { cached_tokens: 1500 }, + } as any, + }), + ]) + + const msgDelta = events.find(e => e.type === 'message_delta') as any + expect(msgDelta.usage.cache_read_input_tokens).toBe(1500) + // input_tokens = prompt_tokens - cached_tokens = 2000 - 1500 = 500 + expect(msgDelta.usage.input_tokens).toBe(500) + expect(msgDelta.usage.output_tokens).toBe(100) + }) + + test('subtracts cached_tokens from input_tokens to match Anthropic semantic', async () => { + // Anthropic's input_tokens = non-cached tokens only. + // OpenAI's prompt_tokens = total input including cached. + // The adapter must subtract: input_tokens = prompt_tokens - cached_tokens. + const events = await collectEvents([ + makeChunk({ + choices: [{ index: 0, delta: { content: 'hi' }, finish_reason: null }], + }), + makeChunk({ + choices: [{ index: 0, delta: {}, finish_reason: 'stop' }], + usage: { + prompt_tokens: 34097, + completion_tokens: 30, + total_tokens: 34127, + prompt_tokens_details: { cached_tokens: 34048 }, + } as any, + }), + ]) + + const msgDelta = events.find(e => e.type === 'message_delta') as any + // input_tokens = 34097 - 34048 = 49 (non-cached input only) + expect(msgDelta.usage.input_tokens).toBe(49) + expect(msgDelta.usage.cache_read_input_tokens).toBe(34048) + expect(msgDelta.usage.output_tokens).toBe(30) + }) }) diff --git a/src/services/api/openai/index.ts b/src/services/api/openai/index.ts index 007170538..7b04870bf 100644 --- a/src/services/api/openai/index.ts +++ b/src/services/api/openai/index.ts @@ -5,6 +5,7 @@ import type { StreamEvent, SystemAPIErrorMessage, AssistantMessage, + UserMessage, } from '../../../types/message.js' import type { Tools } from '../../../Tool.js' import { getOpenAIClient } from './client.js' @@ -24,18 +25,131 @@ import { import { logForDebugging } from '../../../utils/debug.js' import { addToTotalSessionCost } from '../../../cost-tracker.js' import { calculateUSDCost } from '../../../utils/modelCost.js' +import { getModelMaxOutputTokens } from '../../../utils/context.js' +import { recordLLMObservation } from '../../../services/langfuse/tracing.js' +import { + convertMessagesToLangfuse, + convertOutputToLangfuse, + convertToolsToLangfuse, +} from '../../../services/langfuse/convert.js' import type { Options } from '../claude.js' import { randomUUID } from 'crypto' import { createAssistantAPIErrorMessage, + createUserMessage, normalizeContentFromAPI, } from '../../../utils/messages.js' -import { isToolSearchEnabled } from '../../../utils/toolSearch.js' import { + isToolSearchEnabled, + extractDiscoveredToolNames, + isDeferredToolsDeltaEnabled, +} from '../../../utils/toolSearch.js' +import { + formatDeferredToolLine, isDeferredTool, TOOL_SEARCH_TOOL_NAME, } from '../../../tools/ToolSearchTool/prompt.js' +/** + * Mirrors the Anthropic request path's deferred-tool announcement for OpenAI. + * + * OpenAI-compatible endpoints cannot consume Anthropic's `defer_loading` or + * `tool_reference` beta payloads directly, so the model needs the same textual + * list of deferred MCP tool names before it can ask ToolSearchTool to load + * their full schemas. + */ +function prependDeferredToolListIfNeeded( + messages: (AssistantMessage | UserMessage)[], + tools: Tools, + deferredToolNames: Set, + useToolSearch: boolean, +): (AssistantMessage | UserMessage)[] { + if (!useToolSearch || isDeferredToolsDeltaEnabled()) return messages + + const deferredToolList = tools + .filter(tool => deferredToolNames.has(tool.name)) + .map(formatDeferredToolLine) + .sort() + .join('\n') + + if (!deferredToolList) return messages + + return [ + createUserMessage({ + content: `\n${deferredToolList}\n`, + isMeta: true, + }), + ...messages, + ] +} + +function isOpenAIConvertibleMessage( + msg: Message, +): msg is AssistantMessage | UserMessage { + return msg.type === 'assistant' || msg.type === 'user' +} + +function assembleFinalAssistantOutputs(params: { + partialMessage: any + contentBlocks: Record + tools: Tools + agentId: string | undefined + usage: { + input_tokens: number + output_tokens: number + cache_creation_input_tokens: number + cache_read_input_tokens: number + } + stopReason: string | null + maxTokens: number +}): (AssistantMessage | SystemAPIErrorMessage)[] { + const { + partialMessage, + contentBlocks, + tools, + agentId, + usage, + stopReason, + maxTokens, + } = params + const outputs: (AssistantMessage | SystemAPIErrorMessage)[] = [] + + const allBlocks = Object.keys(contentBlocks) + .sort((a, b) => Number(a) - Number(b)) + .map(k => contentBlocks[Number(k)]) + .filter(Boolean) + + if (allBlocks.length > 0) { + outputs.push({ + message: { + ...partialMessage, + content: normalizeContentFromAPI(allBlocks, tools, agentId), + usage, + stop_reason: stopReason, + stop_sequence: null, + }, + requestId: undefined, + type: 'assistant', + uuid: randomUUID(), + timestamp: new Date().toISOString(), + } as AssistantMessage) + } + + if (stopReason === 'max_tokens') { + outputs.push( + createAssistantAPIErrorMessage({ + content: + `Output truncated: response exceeded the ${maxTokens} token limit. ` + + `Set CLAUDE_CODE_MAX_OUTPUT_TOKENS to override.`, + apiError: 'max_output_tokens', + error: 'max_output_tokens', + }), + ) + } + + return outputs +} + /** * OpenAI-compatible query path. Converts Anthropic-format messages/tools to * OpenAI format, calls the OpenAI-compatible endpoint, and converts the @@ -83,13 +197,15 @@ export async function* queryModelOpenAI( // at runtime. Keeping the tools array stable preserves the prompt cache. let filteredTools = tools if (useToolSearch && deferredToolNames.size > 0) { + const discoveredToolNames = extractDiscoveredToolNames(messages) + filteredTools = tools.filter(tool => { // Always include non-deferred tools if (!deferredToolNames.has(tool.name)) return true // Always include ToolSearchTool (so it can discover more tools) if (toolMatchesName(tool, TOOL_SEARCH_TOOL_NAME)) return true - // All other deferred tools are excluded — use ExecuteExtraTool instead - return false + // Only include deferred tools whose schemas have already been discovered + return discoveredToolNames.has(tool.name) }) } @@ -117,13 +233,27 @@ export async function* queryModelOpenAI( ) // 7. Convert messages and tools to OpenAI format + const enableThinking = isOpenAIThinkingEnabled(openaiModel) + const openAIConvertibleMessages = messagesForAPI.filter( + isOpenAIConvertibleMessage, + ) + const messagesWithDeferredToolList = prependDeferredToolListIfNeeded( + openAIConvertibleMessages, + tools, + deferredToolNames, + useToolSearch, + ) const openaiMessages = anthropicMessagesToOpenAI( - messagesForAPI, + messagesWithDeferredToolList, systemPrompt, + { enableThinking }, ) const openaiTools = anthropicToolsToOpenAI(standardTools) const openaiToolChoice = anthropicToolChoiceToOpenAI(options.toolChoice) + const { upperLimit } = getModelMaxOutputTokens(openaiModel) + const maxTokens = options.maxOutputTokensOverride ?? upperLimit + // 8. Get client and make streaming request const client = getOpenAIClient({ maxRetries: 0, @@ -132,28 +262,22 @@ export async function* queryModelOpenAI( }) logForDebugging( - `[OpenAI] Calling model=${openaiModel}, messages=${openaiMessages.length}, tools=${openaiTools.length}`, + `[OpenAI] Calling model=${openaiModel}, messages=${openaiMessages.length}, tools=${openaiTools.length}, thinking=${enableThinking}`, ) // 9. Call OpenAI API with streaming - const stream = await client.chat.completions.create( - { - model: openaiModel, - messages: openaiMessages, - ...(openaiTools.length > 0 && { - tools: openaiTools, - ...(openaiToolChoice && { tool_choice: openaiToolChoice }), - }), - stream: true, - stream_options: { include_usage: true }, - ...(options.temperatureOverride !== undefined && { - temperature: options.temperatureOverride, - }), - }, - { - signal, - }, - ) + const requestBody = buildOpenAIRequestBody({ + model: openaiModel, + messages: openaiMessages, + tools: openaiTools, + toolChoice: openaiToolChoice, + enableThinking, + maxTokens, + temperatureOverride: options.temperatureOverride, + }) + const stream = await client.chat.completions.create(requestBody, { + signal, + }) // 10. Convert OpenAI stream to Anthropic events, then process into // AssistantMessage + StreamEvent (matching the Anthropic path behavior) @@ -161,7 +285,9 @@ export async function* queryModelOpenAI( // Accumulate content blocks and usage, same as the Anthropic path in claude.ts const contentBlocks: Record = {} + const collectedMessages: AssistantMessage[] = [] let partialMessage: any + let stopReason: string | null = null let usage = { input_tokens: 0, output_tokens: 0, @@ -215,21 +341,7 @@ export async function* queryModelOpenAI( break } case 'content_block_stop': { - const idx = (event as any).index - const block = contentBlocks[idx] - if (!block || !partialMessage) break - - const m: AssistantMessage = { - message: { - ...partialMessage, - content: normalizeContentFromAPI([block], tools, options.agentId), - }, - requestId: undefined, - type: 'assistant', - uuid: randomUUID(), - timestamp: new Date().toISOString(), - } - yield m + // Block accumulation is complete; assembly happens at message_stop. break } case 'message_delta': { @@ -237,21 +349,33 @@ export async function* queryModelOpenAI( if (deltaUsage) { usage = { ...usage, ...deltaUsage } } - // Update the stop_reason on the last yielded message - // (we don't have a reference here, but the consumer handles this) + if ((event as any).delta?.stop_reason != null) { + stopReason = (event as any).delta.stop_reason + } break } - case 'message_stop': + case 'message_stop': { + if (partialMessage) { + for (const output of assembleFinalAssistantOutputs({ + partialMessage, + contentBlocks, + tools, + agentId: options.agentId, + usage, + stopReason, + maxTokens, + })) { + if (output.type === 'assistant') collectedMessages.push(output) + yield output + } + partialMessage = null + } + if (usage.input_tokens + usage.output_tokens > 0) { + const costUSD = calculateUSDCost(openaiModel, usage as any) + addToTotalSessionCost(costUSD, usage as any, options.model) + } break - } - - // Track cost and token usage (matching the Anthropic path in claude.ts) - if ( - event.type === 'message_stop' && - usage.input_tokens + usage.output_tokens > 0 - ) { - const costUSD = calculateUSDCost(openaiModel, usage as any) - addToTotalSessionCost(costUSD, usage as any, options.model) + } } // Also yield as StreamEvent for real-time display (matching Anthropic path) @@ -261,6 +385,37 @@ export async function* queryModelOpenAI( ...(event.type === 'message_start' ? { ttftMs } : undefined), } as StreamEvent } + + recordLLMObservation(options.langfuseTrace ?? null, { + model: openaiModel, + provider: 'openai', + input: convertMessagesToLangfuse(openaiMessages), + output: convertOutputToLangfuse(collectedMessages), + usage: { + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + cache_creation_input_tokens: usage.cache_creation_input_tokens, + cache_read_input_tokens: usage.cache_read_input_tokens, + }, + startTime: new Date(start), + endTime: new Date(), + completionStartTime: ttftMs > 0 ? new Date(start + ttftMs) : undefined, + tools: convertToolsToLangfuse(toolSchemas as unknown[]), + }) + + if (partialMessage) { + for (const output of assembleFinalAssistantOutputs({ + partialMessage, + contentBlocks, + tools, + agentId: options.agentId, + usage, + stopReason, + maxTokens, + })) { + yield output + } + } } catch (error) { const errorMessage = error instanceof Error ? error.message : String(error) logForDebugging(`[OpenAI] Error: ${errorMessage}`, { level: 'error' }) @@ -329,7 +484,6 @@ export function buildOpenAIRequestBody(params: { enableThinking = false, temperatureOverride, maxTokens, - systemPrompt, } = params const body: Record = { @@ -355,7 +509,7 @@ export function buildOpenAIRequestBody(params: { } if (maxTokens !== undefined) { - body.max_completion_tokens = maxTokens + body.max_tokens = maxTokens } return body diff --git a/src/services/api/openai/streamAdapter.ts b/src/services/api/openai/streamAdapter.ts index b510e731e..028136b52 100644 --- a/src/services/api/openai/streamAdapter.ts +++ b/src/services/api/openai/streamAdapter.ts @@ -18,6 +18,10 @@ import { randomUUID } from 'crypto' * prompt_tokens_details.cached_tokens → cache_read_input_tokens * (no OpenAI equivalent) → cache_creation_input_tokens (always 0) * + * All four fields are emitted in the post-loop message_delta (not message_start) + * so that trailing usage chunks sent after finish_reason are fully captured + * before the final counts are reported. + * * Thinking support: * DeepSeek and compatible providers send `delta.reasoning_content` for chain-of-thought. * This is mapped to Anthropic's `thinking` content blocks: @@ -38,7 +42,10 @@ export async function* adaptOpenAIStreamToAnthropic( let currentContentIndex = -1 // Track tool_use blocks: tool_calls index → { contentIndex, id, name, arguments } - const toolBlocks = new Map() + const toolBlocks = new Map< + number, + { contentIndex: number; id: string; name: string; arguments: string } + >() // Track thinking block state let thinkingBlockOpen = false @@ -57,6 +64,11 @@ export async function* adaptOpenAIStreamToAnthropic( // Track all open content block indices (for cleanup) const openBlockIndices = new Set() + // Deferred finish state: emit message_delta/message_stop after the stream loop + // so trailing usage-only chunks can update token counts first. + let pendingFinishReason: string | null = null + let pendingHasToolCalls = false + for await (const chunk of stream) { const choice = chunk.choices?.[0] const delta = choice?.delta @@ -203,7 +215,8 @@ export async function* adaptOpenAIStreamToAnthropic( // Start new tool_use block currentContentIndex++ - const toolId = tc.id || `toolu_${randomUUID().replace(/-/g, '').slice(0, 24)}` + const toolId = + tc.id || `toolu_${randomUUID().replace(/-/g, '').slice(0, 24)}` const toolName = tc.function?.name || '' toolBlocks.set(tcIndex, { @@ -242,7 +255,8 @@ export async function* adaptOpenAIStreamToAnthropic( } } - // Handle finish + // Handle finish: close open blocks now, but defer final message events until + // after the stream loop so trailing usage chunks are included. if (choice?.finish_reason) { // Close thinking block if still open if (thinkingBlockOpen) { @@ -275,27 +289,8 @@ export async function* adaptOpenAIStreamToAnthropic( } } - // Map finish_reason to Anthropic stop_reason. - // Some backends return "stop" even when tool_calls are present — - // force "tool_use" when we saw any tool blocks to ensure the query - // loop actually executes the tools. - const hasToolCalls = toolBlocks.size > 0 - const stopReason = hasToolCalls ? 'tool_use' : mapFinishReason(choice.finish_reason) - - yield { - type: 'message_delta', - delta: { - stop_reason: stopReason, - stop_sequence: null, - }, - usage: { - output_tokens: outputTokens, - }, - } as BetaRawMessageStreamEvent - - yield { - type: 'message_stop', - } as BetaRawMessageStreamEvent + pendingFinishReason = choice.finish_reason + pendingHasToolCalls = toolBlocks.size > 0 } } @@ -306,6 +301,33 @@ export async function* adaptOpenAIStreamToAnthropic( index: idx, } as BetaRawMessageStreamEvent } + + if (pendingFinishReason !== null) { + const stopReason = + pendingFinishReason === 'length' + ? 'max_tokens' + : pendingHasToolCalls + ? 'tool_use' + : mapFinishReason(pendingFinishReason) + + yield { + type: 'message_delta', + delta: { + stop_reason: stopReason, + stop_sequence: null, + }, + usage: { + input_tokens: inputTokens, + output_tokens: outputTokens, + cache_read_input_tokens: cachedTokens, + cache_creation_input_tokens: 0, + }, + } as BetaRawMessageStreamEvent + + yield { + type: 'message_stop', + } as BetaRawMessageStreamEvent + } } /** diff --git a/src/utils/__tests__/formatBriefTimestamp.test.ts b/src/utils/__tests__/formatBriefTimestamp.test.ts index 9a89b16fc..3f18d740b 100644 --- a/src/utils/__tests__/formatBriefTimestamp.test.ts +++ b/src/utils/__tests__/formatBriefTimestamp.test.ts @@ -1,81 +1,93 @@ -import { describe, expect, test } from "bun:test"; -import { formatBriefTimestamp } from "../formatBriefTimestamp"; +import { afterAll, beforeAll, describe, expect, test } from 'bun:test' +import { formatBriefTimestamp } from '../formatBriefTimestamp' -// Force en-US locale for deterministic test output regardless of system LANG -process.env.LC_ALL = 'en-US'; -// Force UTC timezone — system is CST (UTC+8) which shifts weekdays -process.env.TZ = 'UTC'; +let savedLcAll: string | undefined +let savedTz: string | undefined -describe("formatBriefTimestamp", () => { +beforeAll(() => { + savedLcAll = process.env.LC_ALL + savedTz = process.env.TZ + process.env.LC_ALL = 'en_US.UTF-8' + process.env.TZ = 'UTC' +}) + +afterAll(() => { + if (savedLcAll === undefined) delete process.env.LC_ALL + else process.env.LC_ALL = savedLcAll + if (savedTz === undefined) delete process.env.TZ + else process.env.TZ = savedTz +}) + +describe('formatBriefTimestamp', () => { // Fixed "now" for deterministic tests: 2026-04-02T14:00:00Z (Thursday) - const now = new Date("2026-04-02T14:00:00Z"); + const now = new Date('2026-04-02T14:00:00Z') - test("same day timestamp returns time only (contains colon)", () => { - const result = formatBriefTimestamp("2026-04-02T10:30:00Z", now); - expect(result).toContain(":"); + test('same day timestamp returns time only (contains colon)', () => { + const result = formatBriefTimestamp('2026-04-02T10:30:00Z', now) + expect(result).toContain(':') // Should NOT contain a weekday name since it's the same day expect(result).not.toMatch( - /Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday/ - ); - }); + /Monday|Tuesday|Wednesday|Thursday|Friday|Saturday|Sunday/, + ) + }) - test("yesterday returns weekday and time", () => { + test('yesterday returns weekday and time', () => { // 2026-04-01 is Wednesday - const result = formatBriefTimestamp("2026-04-01T16:15:00Z", now); - expect(result).toContain("Wednesday"); - expect(result).toContain(":"); - }); + const result = formatBriefTimestamp('2026-04-01T16:15:00Z', now) + expect(result).toContain('Wednesday') + expect(result).toContain(':') + }) - test("3 days ago returns weekday and time", () => { + test('3 days ago returns weekday and time', () => { // 2026-03-30 is Monday - const result = formatBriefTimestamp("2026-03-30T09:00:00Z", now); - expect(result).toContain("Monday"); - expect(result).toContain(":"); - }); + const result = formatBriefTimestamp('2026-03-30T09:00:00Z', now) + expect(result).toContain('Monday') + expect(result).toContain(':') + }) - test("6 days ago returns weekday and time (still within 6-day window)", () => { + test('6 days ago returns weekday and time (still within 6-day window)', () => { // 2026-03-27 is Friday - const result = formatBriefTimestamp("2026-03-27T12:00:00Z", now); - expect(result).toContain("Friday"); - expect(result).toContain(":"); - }); + const result = formatBriefTimestamp('2026-03-27T12:00:00Z', now) + expect(result).toContain('Friday') + expect(result).toContain(':') + }) - test("7+ days ago returns weekday, month, day, and time", () => { + test('7+ days ago returns weekday, month, day, and time', () => { // 2026-03-20 is Friday, 13 days ago - const result = formatBriefTimestamp("2026-03-20T14:30:00Z", now); - expect(result).toContain("Friday"); - expect(result).toContain(":"); + const result = formatBriefTimestamp('2026-03-20T14:30:00Z', now) + expect(result).toContain('Friday') + expect(result).toContain(':') // Should contain month abbreviation (Mar) - expect(result).toMatch(/Mar/); - }); + expect(result).toMatch(/Mar/) + }) - test("much older date returns full format with month", () => { - const result = formatBriefTimestamp("2025-12-25T08:00:00Z", now); - expect(result).toContain(":"); - expect(result).toMatch(/Dec/); - }); + test('much older date returns full format with month', () => { + const result = formatBriefTimestamp('2025-12-25T08:00:00Z', now) + expect(result).toContain(':') + expect(result).toMatch(/Dec/) + }) - test("invalid ISO string returns empty string", () => { - expect(formatBriefTimestamp("not-a-date", now)).toBe(""); - }); + test('invalid ISO string returns empty string', () => { + expect(formatBriefTimestamp('not-a-date', now)).toBe('') + }) - test("empty string returns empty string", () => { - expect(formatBriefTimestamp("", now)).toBe(""); - }); + test('empty string returns empty string', () => { + expect(formatBriefTimestamp('', now)).toBe('') + }) - test("same day early morning returns time format", () => { - const result = formatBriefTimestamp("2026-04-02T01:05:00Z", now); - expect(result).toContain(":"); + test('same day early morning returns time format', () => { + const result = formatBriefTimestamp('2026-04-02T01:05:00Z', now) + expect(result).toContain(':') // Should be time-only format - expect(result.length).toBeLessThan(20); - }); + expect(result.length).toBeLessThan(20) + }) - test("uses current time as default when now is not provided", () => { + test('uses current time as default when now is not provided', () => { // Just verify it returns a non-empty string for a recent timestamp - const recent = new Date(); - recent.setMinutes(recent.getMinutes() - 5); - const result = formatBriefTimestamp(recent.toISOString()); - expect(result).not.toBe(""); - expect(result).toContain(":"); - }); -}); + const recent = new Date() + recent.setMinutes(recent.getMinutes() - 5) + const result = formatBriefTimestamp(recent.toISOString()) + expect(result).not.toBe('') + expect(result).toContain(':') + }) +}) diff --git a/src/utils/formatBriefTimestamp.ts b/src/utils/formatBriefTimestamp.ts index fb9ca65c9..f457d557f 100644 --- a/src/utils/formatBriefTimestamp.ts +++ b/src/utils/formatBriefTimestamp.ts @@ -76,6 +76,7 @@ function getLocale(): string | undefined { } } +/** Return the epoch-ms of the start of the local calendar day for `d`. */ function startOfDay(d: Date): number { return new Date(d.getFullYear(), d.getMonth(), d.getDate()).getTime() }