github/claude-code-best
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: backfill upstream OpenAI fixes (c82f5994, 901628b4) + regression tests

Browse Source 
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
This commit is contained in:
James Feng 2026-06-04 20:20:21 +08:00
parent 05cb6b0a7a
commit 12301713b9
9 changed files with 2093 additions and 262 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/services/api/claude.ts
View File

@ -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,
)

674
src/services/api/openai/__tests__/queryModelOpenAI.isolated.ts Normal file
View File

@ -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<string, any> = {},
): 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<string, any> = {},
): 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<string, string | undefined> = {},
) {
// Wire events into the mocked stream adapter
_nextEvents = events
// Save + apply env overrides
const saved: Record<string, string | undefined> = {}
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<string, any> | 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<string, any>) => {
_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(
'<available-deferred-tools>\\nmcp__wechat__send_message\\n</available-deferred-tools>',
)
} finally {
_toolSearchEnabled = false
}
})
})

540
src/services/api/openai/__tests__/queryModelOpenAI.runner.ts Normal file
View File

@ -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<string, any> = {},
): 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<string, any> = {},
): 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<string, string | undefined> = {},
) {
// Wire events into the mocked stream adapter
_nextEvents = events
// Save + apply env overrides
const saved: Record<string, string | undefined> = {}
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<string, any> | null = null
mock.module('../client.js', () => ({
getOpenAIClient: () => ({
chat: {
completions: {
create: async (args: Record<string, any>) => {
_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)
})
})

26
src/services/api/openai/__tests__/queryModelOpenAI.test.ts Normal file
View File

@ -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',
)
})
})

651
src/services/api/openai/__tests__/streamAdapter.test.ts
View File

@ -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<ChatCompletionChunk> {
function mockStream(
chunks: ChatCompletionChunk[],
): AsyncIterable<ChatCompletionChunk> {
return {
[Symbol.asyncIterator]() {
let i = 0
@ -18,7 +20,9 @@ function mockStream(chunks: ChatCompletionChunk[]): AsyncIterable<ChatCompletion
}
/** Create a minimal ChatCompletionChunk */
function makeChunk(overrides: Partial<ChatCompletionChunk> & any = {}): ChatCompletionChunk {
function makeChunk(
overrides: Partial<ChatCompletionChunk> & any = {},
): ChatCompletionChunk {
return {
id: 'chatcmpl-test',
object: 'chat.completion.chunk',
@ -29,9 +33,13 @@ function makeChunk(overrides: Partial<ChatCompletionChunk> & 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)
})
})

258
src/services/api/openai/index.ts
View File

@ -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<string>,
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: `<available-deferred-tools>\n${deferredToolList}\n</available-deferred-tools>`,
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<number, any>
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<number, any> = {}
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<string, unknown> = {
@ -355,7 +509,7 @@ export function buildOpenAIRequestBody(params: {
}
if (maxTokens !== undefined) {
body.max_completion_tokens = maxTokens
body.max_tokens = maxTokens
}
return body

70
src/services/api/openai/streamAdapter.ts
View File

@ -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<number, { contentIndex: number; id: string; name: string; arguments: string }>()
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<number>()
// 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
}
}
/**

130
src/utils/__tests__/formatBriefTimestamp.test.ts
View File

@ -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(':')
})
})

1
src/utils/formatBriefTimestamp.ts
View File

@ -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()
}