import { extractUsage } from "./usageTracking.ts"; import { parseSSEDataPayload } from "./streamHelpers.ts"; import { backfillResponsesCompletedOutput, normalizeResponsesSseIds, pushUniqueResponsesOutputItems, stringifyIdValue, stripResponsesLifecycleEcho, } from "./responsesStreamHelpers.ts"; import { getAnyReasoningValue } from "./reasoningFields.ts"; type JsonRecord = Record; type EventPrefixBuffer = { eventType: () => string; flush: () => string; prefixData: (output: string, line: string) => string; remember: (line: string) => void; }; export type PassthroughTailProcessorContext = { getSkipPassthroughEvent: () => boolean; setSkipPassthroughEvent: (value: boolean) => void; clearPendingPassthroughEvent: () => void; shouldAbortOnClaudeLifecycle: (payload: unknown) => boolean; emitClaudeEmptyStreamErrorAndAbort: () => void; isClaudeEventPayload: (payload: unknown) => boolean; updateClaudeEmptyResponseLifecycle: (payload: unknown) => void; passthroughEventPrefix: EventPrefixBuffer; emitConvertedOutput: (output: string) => void; pushProviderPayload: (payload: unknown) => void; pushClientPayload: (payload: unknown) => void; setPassthroughResponsesId: (value: string) => void; setUsage: (value: unknown) => void; addTotalContentLength: (value: number) => void; appendPassthroughContent: (value: string) => void; appendPassthroughReasoning: (value: string) => void; getResponsesReasoningKey: (payload: Record) => string | null; markResponsesReasoningSummarySeen: (key: string) => void; ensureVisibleResponsesReasoningSummary: (payload: Record) => boolean; emitSyntheticResponsesReasoningSummary: (payload: Record) => void; passthroughResponsesOutputItems: unknown[]; passthroughResponsesPendingFunctionCalls: Map; getPassthroughResponsesCurrentFunctionCallKey: () => string | null; setPassthroughResponsesCurrentFunctionCallKey: (value: string | null) => void; hasPassthroughToolCalls: () => boolean; toResponsesCompletedWithToolCalls: (parsed: JsonRecord) => JsonRecord; }; function asRecord(value: unknown): JsonRecord { return value && typeof value === "object" && !Array.isArray(value) ? (value as JsonRecord) : {}; } function getFunctionCallPendingKey(item: JsonRecord | null): string | null { if (!item) return null; if (typeof item.id === "string") return item.id; if (typeof item.call_id === "string") return item.call_id; return null; } function handleResponsesTailPayload( parsed: JsonRecord, output: string, context: PassthroughTailProcessorContext ): string { const responsesIdsNormalized = normalizeResponsesSseIds(parsed); const parsedResponse = asRecord(parsed.response); const responseId = (parsedResponse ? stringifyIdValue(parsedResponse.id) : null) || stringifyIdValue(parsed.response_id); if (responseId) { context.setPassthroughResponsesId(responseId); } const extracted = extractUsage(parsed); if (extracted) { context.setUsage(extracted); } if (typeof parsed.delta === "string") { context.addTotalContentLength(parsed.delta.length); } if (parsed.type === "response.output_text.delta" && typeof parsed.delta === "string") { context.appendPassthroughContent(parsed.delta); } if ( parsed.type === "response.reasoning_summary_text.delta" || parsed.type === "response.reasoning_summary_text.done" || parsed.type === "response.reasoning_summary_part.done" ) { const reasoningKey = context.getResponsesReasoningKey(parsed); if (reasoningKey) { context.markResponsesReasoningSummarySeen(reasoningKey); } } if ( parsed.type === "response.output_item.added" && asRecord(parsed.item).type === "function_call" ) { const item = { ...(parsed.item as JsonRecord) }; const pendingKey = getFunctionCallPendingKey(item); if (pendingKey) { if (typeof item.arguments !== "string") { item.arguments = ""; } context.passthroughResponsesPendingFunctionCalls.set(pendingKey, item); context.setPassthroughResponsesCurrentFunctionCallKey(pendingKey); } } if (parsed.type === "response.function_call_arguments.delta") { const pendingKey = typeof parsed.item_id === "string" ? parsed.item_id : context.getPassthroughResponsesCurrentFunctionCallKey(); const pending = pendingKey ? context.passthroughResponsesPendingFunctionCalls.get(pendingKey) : undefined; if (pending && typeof parsed.delta === "string") { const previousArgs = typeof pending.arguments === "string" ? pending.arguments : ""; pending.arguments = previousArgs + parsed.delta; } } if (parsed.type === "response.function_call_arguments.done") { const pendingKey = typeof parsed.item_id === "string" ? parsed.item_id : context.getPassthroughResponsesCurrentFunctionCallKey(); const pending = pendingKey ? context.passthroughResponsesPendingFunctionCalls.get(pendingKey) : undefined; if (pending) { if (typeof parsed.arguments === "string") { pending.arguments = parsed.arguments; } pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [pending]); } } if (parsed.type === "response.output_item.done" && parsed.item) { const reasoningSummaryInjected = context.ensureVisibleResponsesReasoningSummary(parsed); context.emitSyntheticResponsesReasoningSummary(parsed); pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [parsed.item]); if (reasoningSummaryInjected) { output = `data: ${JSON.stringify(parsed)}\n\n`; } const item = asRecord(parsed.item); if (item.type === "function_call") { const pendingKey = getFunctionCallPendingKey(item); if (pendingKey) { context.passthroughResponsesPendingFunctionCalls.delete(pendingKey); if (context.getPassthroughResponsesCurrentFunctionCallKey() === pendingKey) { context.setPassthroughResponsesCurrentFunctionCallKey(null); } } } } if ( parsed.type === "response.completed" && Array.isArray(asRecord(parsed.response).output) && (asRecord(parsed.response).output as unknown[]).length > 0 ) { pushUniqueResponsesOutputItems( context.passthroughResponsesOutputItems, asRecord(parsed.response).output as unknown[] ); } if ( parsed.type === "response.completed" && context.passthroughResponsesPendingFunctionCalls.size > 0 ) { pushUniqueResponsesOutputItems(context.passthroughResponsesOutputItems, [ ...context.passthroughResponsesPendingFunctionCalls.values(), ]); context.passthroughResponsesPendingFunctionCalls.clear(); context.setPassthroughResponsesCurrentFunctionCallKey(null); } const textualToolCallBackfilled = parsed.type === "response.completed" && context.hasPassthroughToolCalls(); const outputPayload = textualToolCallBackfilled ? context.toResponsesCompletedWithToolCalls(parsed) : parsed; const stripped = stripResponsesLifecycleEcho(outputPayload); const backfilled = backfillResponsesCompletedOutput( outputPayload, context.passthroughResponsesOutputItems ); if (stripped || backfilled || textualToolCallBackfilled || responsesIdsNormalized) { output = `data: ${JSON.stringify(outputPayload)}\n\n`; } return output; } function handleOpenAiTailPayload(parsed: JsonRecord, context: PassthroughTailProcessorContext) { const firstChoice = Array.isArray(parsed.choices) ? (parsed.choices[0] as JsonRecord | undefined) : undefined; const delta = asRecord(firstChoice?.delta); if (typeof delta.content === "string") { context.appendPassthroughContent(delta.content); context.addTotalContentLength(delta.content.length); } const reasoningDelta = getAnyReasoningValue(delta); if (reasoningDelta) { context.appendPassthroughReasoning(reasoningDelta); } } export function processBufferedPassthroughLine( line: string, context: PassthroughTailProcessorContext ): boolean { const trimmed = line.trim(); if (context.getSkipPassthroughEvent()) { if (!trimmed) { context.setSkipPassthroughEvent(false); context.clearPendingPassthroughEvent(); } return false; } if (/^event:\s*keepalive\b/i.test(trimmed)) { context.setSkipPassthroughEvent(true); context.clearPendingPassthroughEvent(); return false; } if (/^event:/i.test(trimmed)) { const eventType = trimmed.replace(/^event:\s*/i, ""); if (context.shouldAbortOnClaudeLifecycle({ type: eventType })) { context.emitClaudeEmptyStreamErrorAndAbort(); return true; } context.passthroughEventPrefix.remember(line); return false; } if (/^(?::|id:|retry:)/i.test(trimmed)) { context.passthroughEventPrefix.remember(line); return false; } if (!trimmed) { const pendingOutput = context.passthroughEventPrefix.flush(); if (pendingOutput) { context.emitConvertedOutput(pendingOutput); } context.clearPendingPassthroughEvent(); return false; } if (!trimmed.startsWith("data:")) { context.passthroughEventPrefix.remember(line); return false; } const parsedPassthroughData = parseSSEDataPayload(trimmed.slice(5), { eventType: context.passthroughEventPrefix.eventType(), }); if (parsedPassthroughData?.done === true) { return false; } let output = line.startsWith("data:") && !line.startsWith("data: ") ? `data: ${line.slice(5)}\n\n` : `${line}\n\n`; if (parsedPassthroughData) { context.pushProviderPayload(parsedPassthroughData); if (context.shouldAbortOnClaudeLifecycle(parsedPassthroughData)) { context.emitClaudeEmptyStreamErrorAndAbort(); return true; } if (context.isClaudeEventPayload(parsedPassthroughData)) { context.updateClaudeEmptyResponseLifecycle(parsedPassthroughData); } const parsed = parsedPassthroughData as JsonRecord; const parsedType = typeof parsed.type === "string" ? parsed.type : ""; const isResponses = parsedType.startsWith("response."); const isClaude = context.isClaudeEventPayload(parsed); if (isResponses) { output = handleResponsesTailPayload(parsed, output, context); } else if (!isClaude) { handleOpenAiTailPayload(parsed, context); } context.pushClientPayload(parsed); } output = context.passthroughEventPrefix.prefixData(output, line); context.emitConvertedOutput(output); return false; }