diff --git a/src/glmt/glmt-transformer.ts b/src/glmt/glmt-transformer.ts index dd038f6f..cc6aae19 100644 --- a/src/glmt/glmt-transformer.ts +++ b/src/glmt/glmt-transformer.ts @@ -1,452 +1,136 @@ /** - * GlmtTransformer - Convert between Anthropic and OpenAI formats with thinking and tool support + * GlmtTransformer - Orchestrator for Anthropic ↔ OpenAI format transformation * - * Features: - * - Request: Anthropic → OpenAI (inject reasoning params, transform tools) - * - Response: OpenAI reasoning_content → Anthropic thinking blocks - * - Tool Support: Anthropic tools ↔ OpenAI function calling (bidirectional) - * - Streaming: Real-time tool calls with input_json deltas - * - Debug mode: Log raw data to ~/.ccs/logs/ (CCS_DEBUG=1) - * - Verbose mode: Console logging with timestamps - * - Validation: Self-test transformation results - * - * Control Tags (in user prompt): - * - Enable/disable reasoning - * - Control reasoning depth + * Pipeline Architecture: + * - RequestTransformer: Anthropic → OpenAI request conversion + * - StreamParser: Delta processing for streaming responses + * - ResponseBuilder: SSE event generation + * - ToolCallHandler: Tool call processing */ -import * as crypto from 'crypto'; import * as fs from 'fs'; import * as path from 'path'; import * as os from 'os'; import { DeltaAccumulator } from './delta-accumulator'; -import { LocaleEnforcer } from './locale-enforcer'; -import { ReasoningEnforcer } from './reasoning-enforcer'; - -// Types -interface ContentBlock { - type: string; - text?: string; - thinking?: string; - signature?: ThinkingSignature; - id?: string; - name?: string; - input?: Record; - tool_use_id?: string; - content?: string | unknown; -} - -interface Message { - role: string; - content: string | ContentBlock[]; -} - -interface AnthropicTool { - name: string; - description: string; - input_schema: Record; -} - -interface AnthropicRequest { - model: string; - messages: Message[]; - max_tokens?: number; - temperature?: number; - top_p?: number; - tools?: AnthropicTool[]; - stream?: boolean; - thinking?: { - type: 'enabled' | 'disabled'; - budget_tokens?: number; - }; -} - -interface OpenAITool { - type: 'function'; - function: { - name: string; - description: string; - parameters: Record; - }; -} - -interface OpenAIRequest { - model: string; - messages: Message[]; - max_tokens?: number; - temperature?: number; - top_p?: number; - tools?: OpenAITool[]; - tool_choice?: string; - stream?: boolean; - do_sample?: boolean; - reasoning?: boolean; - reasoning_effort?: string; -} - -interface ThinkingConfig { - thinking: boolean; - effort: string; -} - -interface TransformResult { - openaiRequest: OpenAIRequest | AnthropicRequest; - thinkingConfig: ThinkingConfig; - error?: string; -} - -interface ThinkingSignature { - type: string; - hash: string; - length: number; - timestamp: number; -} - -interface OpenAIToolCall { - id: string; - type: string; - function: { - name: string; - arguments: string; - }; -} - -interface OpenAIChoice { - message: { - role?: string; - content?: string | null; - reasoning_content?: string; - tool_calls?: OpenAIToolCall[]; - }; - delta?: { - role?: string; - content?: string; - reasoning_content?: string; - tool_calls?: OpenAIToolCallDelta[]; - }; - finish_reason?: string; -} - -interface OpenAIToolCallDelta { - index: number; - id?: string; - type?: string; - function?: { - name?: string; - arguments?: string; - }; -} - -interface OpenAIResponse { - id?: string; - model?: string; - choices?: OpenAIChoice[]; - usage?: { - prompt_tokens?: number; - completion_tokens?: number; - }; -} - -interface AnthropicResponse { - id: string; - type: string; - role: string; - content: ContentBlock[]; - model: string; - stop_reason: string; - usage: { - input_tokens: number; - output_tokens: number; - }; -} - -interface SSEEvent { - event?: string; - data?: OpenAIResponse & { - usage?: { - prompt_tokens?: number; - completion_tokens?: number; - }; - }; -} - -interface AnthropicSSEEvent { - event: string; - data: Record; -} - -interface ValidationResult { - checks: Record; - passed: number; - total: number; - valid: boolean; -} - -interface GlmtTransformerConfig { - defaultThinking?: boolean; - verbose?: boolean; - debugLog?: boolean; - debugMode?: boolean; - debugLogDir?: string; - explicitReasoning?: boolean; -} - -interface AccumulatorBlock { - index: number; - type: string; - content: string; - stopped: boolean; -} +import { + RequestTransformer, + StreamParser, + ResponseBuilder, + ToolCallHandler, + ContentTransformer, + type AnthropicRequest, + type OpenAIResponse, + type AnthropicResponse, + type ThinkingConfig, + type TransformResult, + type SSEEvent, + type AnthropicSSEEvent, + type GlmtTransformerConfig, + type ContentBlock, + type Message, + type ThinkingSignature, + type ValidationResult, +} from './pipeline'; export class GlmtTransformer { - private defaultThinking: boolean; private verbose: boolean; private debugLog: boolean; - private debugMode: boolean; - debugLogDir: string; // public for external access - private modelMaxTokens: Record; - private localeEnforcer: LocaleEnforcer; - private reasoningEnforcer: ReasoningEnforcer; + debugLogDir: string; + + private requestTransformer: RequestTransformer; + private streamParser: StreamParser; + private responseBuilder: ResponseBuilder; + private toolCallHandler: ToolCallHandler; + private contentTransformer: ContentTransformer; constructor(config: GlmtTransformerConfig = {}) { - this.defaultThinking = config.defaultThinking ?? true; this.verbose = config.verbose || false; - // CCS_DEBUG controls all debug logging (file + console) const debugEnabled = process.env.CCS_DEBUG === '1'; this.debugLog = config.debugLog ?? debugEnabled; - this.debugMode = config.debugMode ?? debugEnabled; - this.debugLogDir = config.debugLogDir || path.join(os.homedir(), '.ccs', 'logs'); - this.modelMaxTokens = { - 'GLM-4.6': 128000, - 'GLM-4.5': 96000, - 'GLM-4.5-air': 16000, - }; - // Initialize locale enforcer (always enforce English) - this.localeEnforcer = new LocaleEnforcer(); + // Initialize pipeline components + this.requestTransformer = new RequestTransformer({ + defaultThinking: config.defaultThinking ?? true, + verbose: this.verbose, + explicitReasoning: config.explicitReasoning ?? true, + log: (msg) => this.log(msg), + }); - // Initialize reasoning enforcer (enabled by default for all GLMT usage) - this.reasoningEnforcer = new ReasoningEnforcer({ - enabled: config.explicitReasoning ?? true, + this.responseBuilder = new ResponseBuilder(this.verbose); + this.toolCallHandler = new ToolCallHandler(); + this.contentTransformer = new ContentTransformer(config.defaultThinking ?? true); + this.streamParser = new StreamParser({ + verbose: this.verbose, + debugMode: config.debugMode ?? debugEnabled, + debugLog: this.debugLog, + writeDebugLog: (type, data) => this.writeDebugLog(type, data), }); } - /** - * Transform Anthropic request to OpenAI format - */ + /** Transform Anthropic request to OpenAI format */ transformRequest(anthropicRequest: AnthropicRequest): TransformResult { - // Log original request this.writeDebugLog('request-anthropic', anthropicRequest); - - try { - // 1. Extract thinking control from messages (tags like ) - const thinkingConfig = this.extractThinkingControl(anthropicRequest.messages || []); - const hasControlTags = this.hasThinkingTags(anthropicRequest.messages || []); - - // 2. Detect "think" keywords in user prompts (Anthropic-style) - const keywordConfig = this.detectThinkKeywords(anthropicRequest.messages || []); - if (keywordConfig && !anthropicRequest.thinking && !hasControlTags) { - thinkingConfig.thinking = keywordConfig.thinking; - thinkingConfig.effort = keywordConfig.effort; - this.log( - `Detected think keyword: ${keywordConfig.keyword}, effort=${keywordConfig.effort}` - ); - } - - // 3. Check anthropicRequest.thinking parameter (takes precedence) - if (anthropicRequest.thinking) { - if (anthropicRequest.thinking.type === 'enabled') { - thinkingConfig.thinking = true; - this.log('Claude CLI explicitly enabled thinking (overrides budget)'); - } else if (anthropicRequest.thinking.type === 'disabled') { - thinkingConfig.thinking = false; - this.log('Claude CLI explicitly disabled thinking (overrides budget)'); - } else { - this.log(`Warning: Unknown thinking type: ${anthropicRequest.thinking.type}`); - } - } - - this.log(`Final thinking control: ${JSON.stringify(thinkingConfig)}`); - - // 3. Map model - const glmModel = this.mapModel(anthropicRequest.model); - - // 4. Inject locale instruction before sanitization - const messagesWithLocale = this.localeEnforcer.injectInstruction( - (anthropicRequest.messages || []) as Parameters< - typeof this.localeEnforcer.injectInstruction - >[0] - ) as unknown as Message[]; - - // 4.5. Inject reasoning instruction (if enabled or thinking requested) - const messagesWithReasoning = this.reasoningEnforcer.injectInstruction( - messagesWithLocale as unknown as Parameters< - typeof this.reasoningEnforcer.injectInstruction - >[0], - thinkingConfig - ) as unknown as Message[]; - - // 5. Convert to OpenAI format - const openaiRequest: OpenAIRequest = { - model: glmModel, - messages: this.sanitizeMessages(messagesWithReasoning), - max_tokens: this.getMaxTokens(glmModel), - stream: anthropicRequest.stream ?? false, - }; - - // 5.5. Transform tools parameter if present - if (anthropicRequest.tools && anthropicRequest.tools.length > 0) { - openaiRequest.tools = this.transformTools(anthropicRequest.tools); - // Always use "auto" as Z.AI doesn't support other modes - openaiRequest.tool_choice = 'auto'; - this.log(`Transformed ${anthropicRequest.tools.length} tools for OpenAI format`); - } - - // 6. Preserve optional parameters - if (anthropicRequest.temperature !== undefined) { - openaiRequest.temperature = anthropicRequest.temperature; - } - if (anthropicRequest.top_p !== undefined) { - openaiRequest.top_p = anthropicRequest.top_p; - } - - // 7. Handle streaming - if (anthropicRequest.stream !== undefined) { - openaiRequest.stream = anthropicRequest.stream; - } - - // 8. Inject reasoning parameters - this.injectReasoningParams(openaiRequest, thinkingConfig); - - // Log transformed request - this.writeDebugLog('request-openai', openaiRequest); - - return { openaiRequest, thinkingConfig }; - } catch (error) { - const err = error as Error; - console.error('[glmt-transformer] Request transformation error:', err); - // Return original request with warning - return { - openaiRequest: anthropicRequest, - thinkingConfig: { thinking: false, effort: 'medium' }, - error: err.message, - }; - } + const result = this.requestTransformer.transform(anthropicRequest); + this.writeDebugLog('request-openai', result.openaiRequest); + return result; } - /** - * Transform OpenAI response to Anthropic format - */ + /** Transform OpenAI response to Anthropic format */ transformResponse( openaiResponse: OpenAIResponse, _thinkingConfig: ThinkingConfig = { thinking: false, effort: 'medium' } ): AnthropicResponse { - // Log original response this.writeDebugLog('response-openai', openaiResponse); try { const choice = openaiResponse.choices?.[0]; - if (!choice) { - throw new Error('No choices in OpenAI response'); - } + if (!choice) throw new Error('No choices in OpenAI response'); const message = choice.message; const content: ContentBlock[] = []; - // Add thinking block if reasoning_content exists if (message.reasoning_content) { - const length = message.reasoning_content.length; - const lineCount = message.reasoning_content.split('\n').length; - const preview = message.reasoning_content.substring(0, 100).replace(/\n/g, ' ').trim(); - - this.log(`Detected reasoning_content:`); - this.log(` Length: ${length} characters`); - this.log(` Lines: ${lineCount}`); - this.log(` Preview: ${preview}...`); - + this.log(`Detected reasoning_content: ${message.reasoning_content.length} chars`); content.push({ type: 'thinking', thinking: message.reasoning_content, - signature: this.generateThinkingSignature(message.reasoning_content), + signature: this.responseBuilder.generateThinkingSignature(message.reasoning_content), }); - } else { - this.log('No reasoning_content in OpenAI response'); - this.log('Note: This is expected if thinking not requested or model cannot reason'); } - // Add text content if (message.content) { - content.push({ - type: 'text', - text: message.content, - }); + content.push({ type: 'text', text: message.content }); } - // Handle tool_calls if present - if (message.tool_calls && message.tool_calls.length > 0) { - message.tool_calls.forEach((toolCall) => { - let parsedInput: Record; - try { - parsedInput = JSON.parse(toolCall.function.arguments || '{}'); - } catch (parseError) { - const err = parseError as Error; - this.log(`Warning: Invalid JSON in tool arguments: ${err.message}`); - parsedInput = { _error: 'Invalid JSON', _raw: toolCall.function.arguments }; - } - - content.push({ - type: 'tool_use', - id: toolCall.id, - name: toolCall.function.name, - input: parsedInput, - }); - }); + if (message.tool_calls?.length) { + content.push(...this.toolCallHandler.processToolCalls(message.tool_calls)); } const anthropicResponse: AnthropicResponse = { id: openaiResponse.id || 'msg_' + Date.now(), type: 'message', role: 'assistant', - content: content, + content, model: openaiResponse.model || 'glm-4.6', - stop_reason: this.mapStopReason(choice.finish_reason || 'stop'), + stop_reason: this.responseBuilder.mapStopReason(choice.finish_reason || 'stop'), usage: { input_tokens: openaiResponse.usage?.prompt_tokens || 0, output_tokens: openaiResponse.usage?.completion_tokens || 0, }, }; - // Validate transformation in verbose mode - if (this.verbose) { - const validation = this.validateTransformation(anthropicResponse); - this.log( - `Transformation validation: ${validation.passed}/${validation.total} checks passed` - ); - if (!validation.valid) { - this.log(`Failed checks: ${JSON.stringify(validation.checks, null, 2)}`); - } - } - - // Log transformed response this.writeDebugLog('response-anthropic', anthropicResponse); - return anthropicResponse; } catch (error) { const err = error as Error; console.error('[glmt-transformer] Response transformation error:', err); - // Return minimal valid response return { id: 'msg_error_' + Date.now(), type: 'message', role: 'assistant', - content: [ - { - type: 'text', - text: '[Transformation Error] ' + err.message, - }, - ], + content: [{ type: 'text', text: '[Transformation Error] ' + err.message }], model: 'glm-4.6', stop_reason: 'end_turn', usage: { input_tokens: 0, output_tokens: 0 }, @@ -454,261 +138,55 @@ export class GlmtTransformer { } } - /** - * Sanitize messages for OpenAI API compatibility - */ - private sanitizeMessages(messages: Message[]): Message[] { - const result: Message[] = []; - - for (const msg of messages) { - // If content is a string, add as-is - if (typeof msg.content === 'string') { - result.push(msg); - continue; - } - - // If content is an array, process blocks - if (Array.isArray(msg.content)) { - // Separate tool_result blocks from other content - const toolResults = msg.content.filter((block) => block.type === 'tool_result'); - const textBlocks = msg.content.filter((block) => block.type === 'text'); - // const toolUseBlocks = msg.content.filter(block => block.type === 'tool_use'); - - // CRITICAL: Tool messages must come BEFORE user text in OpenAI API - for (const toolResult of toolResults) { - result.push({ - role: 'tool', - content: - typeof toolResult.content === 'string' - ? toolResult.content - : JSON.stringify(toolResult.content), - } as Message & { tool_call_id: string }); - } - - // Add text content as user/assistant message AFTER tool messages - if (textBlocks.length > 0) { - const textContent = - textBlocks.length === 1 - ? textBlocks[0].text || '' - : textBlocks.map((b) => b.text || '').join('\n'); - - result.push({ - role: msg.role, - content: textContent, - }); - } - - // If no content at all, add empty message - if (textBlocks.length === 0 && toolResults.length === 0) { - result.push({ - role: msg.role, - content: '', - }); - } - - continue; - } - - // Fallback: return message as-is - result.push(msg); - } - - return result; + /** Transform streaming delta (delegates to StreamParser) */ + transformDelta(openaiEvent: SSEEvent, accumulator: DeltaAccumulator): AnthropicSSEEvent[] { + return this.streamParser.transformDelta(openaiEvent, accumulator); } - /** - * Transform Anthropic tools to OpenAI tools format - */ - private transformTools(anthropicTools: AnthropicTool[]): OpenAITool[] { - return anthropicTools.map((tool) => ({ - type: 'function' as const, - function: { - name: tool.name, - description: tool.description, - parameters: tool.input_schema || {}, - }, - })); + /** Finalize streaming (delegates to StreamParser) */ + finalizeDelta(accumulator: DeltaAccumulator): AnthropicSSEEvent[] { + return this.streamParser.finalizeDelta(accumulator); } - /** - * Check if messages contain thinking control tags - */ - private hasThinkingTags(messages: Message[]): boolean { - for (const msg of messages) { - if (msg.role !== 'user') continue; - const content = msg.content; - if (typeof content !== 'string') continue; - - // Check for control tags - if (//i.test(content) || //i.test(content)) { - return true; - } - } - return false; - } - - /** - * Extract thinking control tags from user messages - */ - private extractThinkingControl(messages: Message[]): ThinkingConfig { - const config: ThinkingConfig = { - thinking: this.defaultThinking, - effort: 'medium', - }; - - // Scan user messages for control tags - for (const msg of messages) { - if (msg.role !== 'user') continue; - - const content = msg.content; - if (typeof content !== 'string') continue; - - // Check for - const thinkingMatch = content.match(//i); - if (thinkingMatch) { - config.thinking = thinkingMatch[1].toLowerCase() === 'on'; - } - - // Check for - const effortMatch = content.match(//i); - if (effortMatch) { - config.effort = effortMatch[1].toLowerCase(); - } - } - - return config; - } - - /** - * Generate thinking signature for Claude Code UI - */ - private generateThinkingSignature(thinking: string): ThinkingSignature { - // Generate signature hash - const hash = crypto.createHash('sha256').update(thinking).digest('hex').substring(0, 16); - - return { - type: 'thinking_signature', - hash: hash, - length: thinking.length, - timestamp: Date.now(), - }; - } - - /** - * Detect Anthropic-style "think" keywords in user prompts - */ - private detectThinkKeywords( - messages: Message[] - ): { thinking: boolean; effort: string; keyword: string } | null { - if (!messages || messages.length === 0) return null; - - // Extract text from user messages - const text = messages - .filter((m) => m.role === 'user') - .map((m) => { - if (typeof m.content === 'string') return m.content; - if (Array.isArray(m.content)) { - return m.content - .filter((block) => block.type === 'text') - .map((block) => block.text || '') - .join(' '); - } - return ''; - }) - .join(' '); - - // Priority: ultrathink > think harder > think hard > think - if (/\bultrathink\b/i.test(text)) { - return { thinking: true, effort: 'max', keyword: 'ultrathink' }; - } - if (/\bthink\s+harder\b/i.test(text)) { - return { thinking: true, effort: 'high', keyword: 'think harder' }; - } - if (/\bthink\s+hard\b/i.test(text)) { - return { thinking: true, effort: 'medium', keyword: 'think hard' }; - } - if (/\bthink\b/i.test(text)) { - return { thinking: true, effort: 'low', keyword: 'think' }; - } - - return null; // No keywords detected - } - - /** - * Inject reasoning parameters into OpenAI request - */ - private injectReasoningParams( - openaiRequest: OpenAIRequest, - thinkingConfig: ThinkingConfig - ): void { - // Always enable sampling for temperature/top_p to work - openaiRequest.do_sample = true; - - // Add thinking-specific parameters if enabled - if (thinkingConfig.thinking) { - openaiRequest.reasoning = true; - openaiRequest.reasoning_effort = thinkingConfig.effort; - } - } - - /** - * Map Anthropic model to GLM model - */ - private mapModel(_anthropicModel: string): string { - // Default to GLM-4.6 (latest and most capable) - return 'GLM-4.6'; - } - - /** - * Get max tokens for model - */ - private getMaxTokens(model: string): number { - return this.modelMaxTokens[model] || 128000; - } - - /** - * Map OpenAI stop reason to Anthropic stop reason - */ - private mapStopReason(openaiReason: string): string { - const mapping: Record = { - stop: 'end_turn', - length: 'max_tokens', - tool_calls: 'tool_use', - content_filter: 'stop_sequence', - }; - return mapping[openaiReason] || 'end_turn'; - } - - /** - * Write debug log to file - */ private writeDebugLog(type: string, data: unknown): void { if (!this.debugLog) return; - try { const timestamp = new Date().toISOString().replace(/[:.]/g, '-').split('.')[0]; - const filename = `${timestamp}-${type}.json`; - const filepath = path.join(this.debugLogDir, filename); - - // Ensure directory exists + const filepath = path.join(this.debugLogDir, `${timestamp}-${type}.json`); fs.mkdirSync(this.debugLogDir, { recursive: true }); - - // Write file (pretty-printed) fs.writeFileSync(filepath, JSON.stringify(data, null, 2) + '\n', 'utf8'); - - if (this.verbose) { - this.log(`Debug log written: ${filepath}`); - } } catch (error) { - const err = error as Error; - console.error(`[glmt-transformer] Failed to write debug log: ${err.message}`); + console.error(`[glmt-transformer] Debug log error: ${(error as Error).message}`); } } - /** - * Validate transformed Anthropic response - */ - private validateTransformation(anthropicResponse: AnthropicResponse): ValidationResult { + private log(message: string): void { + if (this.verbose) { + console.error(`[glmt-transformer] [${new Date().toTimeString().split(' ')[0]}] ${message}`); + } + } + + // ========== Backwards-compatible public methods ========== + + /** Generate thinking signature (delegates to ResponseBuilder) */ + generateThinkingSignature(thinking: string): ThinkingSignature { + return this.responseBuilder.generateThinkingSignature(thinking); + } + + /** Map stop reason (delegates to ResponseBuilder) */ + mapStopReason(openaiReason: string): string { + return this.responseBuilder.mapStopReason(openaiReason); + } + + /** Detect think keywords (delegates to ContentTransformer) */ + detectThinkKeywords( + messages: Message[] + ): { thinking: boolean; effort: string; keyword: string } | null { + return this.contentTransformer.detectThinkKeywords(messages); + } + + /** Validate transformation result */ + validateTransformation(anthropicResponse: AnthropicResponse): ValidationResult { const checks: Record = { hasContent: Boolean(anthropicResponse.content && anthropicResponse.content.length > 0), hasThinking: anthropicResponse.content?.some((block) => block.type === 'thinking') || false, @@ -723,400 +201,6 @@ export class GlmtTransformer { return { checks, passed, total, valid: passed === total }; } - - /** - * Transform OpenAI streaming delta to Anthropic events - */ - transformDelta(openaiEvent: SSEEvent, accumulator: DeltaAccumulator): AnthropicSSEEvent[] { - const events: AnthropicSSEEvent[] = []; - - // Debug logging for streaming deltas - if (this.debugLog && openaiEvent.data) { - this.writeDebugLog('delta-openai', openaiEvent.data); - } - - // Handle [DONE] marker - if (openaiEvent.event === 'done') { - if (!accumulator.isFinalized()) { - return this.finalizeDelta(accumulator); - } - return []; // Already finalized - } - - // Usage update (appears in final chunk, may be before choice data) - if (openaiEvent.data?.usage) { - accumulator.updateUsage(openaiEvent.data.usage); - - // If we have both usage AND finish_reason, finalize immediately - if (accumulator.getFinishReason()) { - events.push(...this.finalizeDelta(accumulator)); - return events; - } - } - - const choice = openaiEvent.data?.choices?.[0]; - if (!choice) return events; - - const delta = choice.delta; - if (!delta) return events; - - // Message start - if (!accumulator.isMessageStarted()) { - if (openaiEvent.data?.model) { - accumulator.setModel(openaiEvent.data.model); - } - events.push(this.createMessageStartEvent(accumulator)); - accumulator.setMessageStarted(true); - } - - // Role - if (delta.role) { - accumulator.setRole(delta.role); - } - - // Reasoning content delta - if (delta.reasoning_content) { - const currentBlock = accumulator.getCurrentBlock(); - - if (this.debugMode) { - console.error(`[GLMT-DEBUG] Reasoning delta: ${delta.reasoning_content.length} chars`); - console.error( - `[GLMT-DEBUG] Current block: ${currentBlock?.type || 'none'}, index: ${currentBlock?.index ?? 'N/A'}` - ); - } - - if (!currentBlock || currentBlock.type !== 'thinking') { - // Start thinking block - const block = accumulator.startBlock('thinking'); - events.push(this.createContentBlockStartEvent(block)); - - if (this.debugMode) { - console.error(`[GLMT-DEBUG] Started new thinking block ${block.index}`); - } - } - - accumulator.addDelta(delta.reasoning_content); - const currentThinkingBlock = accumulator.getCurrentBlock(); - if (currentThinkingBlock) { - events.push(this.createThinkingDeltaEvent(currentThinkingBlock, delta.reasoning_content)); - } - } - - // Text content delta - if (delta.content) { - const currentBlock = accumulator.getCurrentBlock(); - - // Close thinking block if transitioning from thinking to text - if (currentBlock && currentBlock.type === 'thinking' && !currentBlock.stopped) { - const signatureEvent = this.createSignatureDeltaEvent(currentBlock); - if (signatureEvent) { - events.push(signatureEvent); - } - events.push(this.createContentBlockStopEvent(currentBlock)); - accumulator.stopCurrentBlock(); - } - - if (!accumulator.getCurrentBlock() || accumulator.getCurrentBlock()?.type !== 'text') { - // Start text block - const block = accumulator.startBlock('text'); - events.push(this.createContentBlockStartEvent(block)); - } - - accumulator.addDelta(delta.content); - const currentTextBlock = accumulator.getCurrentBlock(); - if (currentTextBlock) { - events.push(this.createTextDeltaEvent(currentTextBlock, delta.content)); - } - } - - // Check for planning loop - if (accumulator.checkForLoop()) { - this.log( - 'WARNING: Planning loop detected - 3 consecutive thinking blocks with no tool calls' - ); - this.log('Forcing early finalization to prevent unbounded planning'); - - // Close current block if any - const currentBlock = accumulator.getCurrentBlock(); - if (currentBlock && !currentBlock.stopped) { - if (currentBlock.type === 'thinking') { - const signatureEvent = this.createSignatureDeltaEvent(currentBlock); - if (signatureEvent) { - events.push(signatureEvent); - } - } - events.push(this.createContentBlockStopEvent(currentBlock)); - accumulator.stopCurrentBlock(); - } - - // Force finalization - events.push(...this.finalizeDelta(accumulator)); - return events; - } - - // Tool calls deltas - if (delta.tool_calls && delta.tool_calls.length > 0) { - // Close current content block ONCE before processing any tool calls - const currentBlock = accumulator.getCurrentBlock(); - if (currentBlock && !currentBlock.stopped) { - if (currentBlock.type === 'thinking') { - const signatureEvent = this.createSignatureDeltaEvent(currentBlock); - if (signatureEvent) { - events.push(signatureEvent); - } - } - events.push(this.createContentBlockStopEvent(currentBlock)); - accumulator.stopCurrentBlock(); - } - - // Process each tool call delta - for (const toolCallDelta of delta.tool_calls) { - // Track tool call state - const isNewToolCall = !accumulator.hasToolCall(toolCallDelta.index); - accumulator.addToolCallDelta(toolCallDelta); - - // Emit tool use events (start + input_json deltas) - if (isNewToolCall) { - // Start new tool_use block in accumulator - const block = accumulator.startBlock('tool_use'); - const toolCall = accumulator.getToolCall(toolCallDelta.index); - - events.push({ - event: 'content_block_start', - data: { - type: 'content_block_start', - index: block.index, - content_block: { - type: 'tool_use', - id: toolCall?.id || `tool_${toolCallDelta.index}`, - name: toolCall?.function?.name || '', - }, - }, - }); - } - - // Emit input_json delta if arguments present - if (toolCallDelta.function?.arguments) { - const currentToolBlock = accumulator.getCurrentBlock(); - if (currentToolBlock && currentToolBlock.type === 'tool_use') { - events.push({ - event: 'content_block_delta', - data: { - type: 'content_block_delta', - index: currentToolBlock.index, - delta: { - type: 'input_json_delta', - partial_json: toolCallDelta.function.arguments, - }, - }, - }); - } - } - } - } - - // Finish reason - if (choice.finish_reason) { - accumulator.setFinishReason(choice.finish_reason); - - // If we have both finish_reason AND usage, finalize immediately - if (accumulator.hasUsageReceived()) { - events.push(...this.finalizeDelta(accumulator)); - } - } - - // Debug logging for generated events - if (this.debugLog && events.length > 0) { - this.writeDebugLog('delta-anthropic-events', { - events, - accumulator: accumulator.getSummary(), - }); - } - - return events; - } - - /** - * Finalize streaming and generate closing events - */ - finalizeDelta(accumulator: DeltaAccumulator): AnthropicSSEEvent[] { - if (accumulator.isFinalized()) { - return []; // Already finalized - } - - const events: AnthropicSSEEvent[] = []; - - // Close current content block if any - const currentBlock = accumulator.getCurrentBlock(); - if (currentBlock && !currentBlock.stopped) { - if (currentBlock.type === 'thinking') { - const signatureEvent = this.createSignatureDeltaEvent(currentBlock); - if (signatureEvent) { - events.push(signatureEvent); - } - } - events.push(this.createContentBlockStopEvent(currentBlock)); - accumulator.stopCurrentBlock(); - } - - // Message delta (stop reason + usage) - events.push({ - event: 'message_delta', - data: { - type: 'message_delta', - delta: { - stop_reason: this.mapStopReason(accumulator.getFinishReason() || 'stop'), - }, - usage: { - input_tokens: accumulator.getInputTokens(), - output_tokens: accumulator.getOutputTokens(), - }, - }, - }); - - // Message stop - events.push({ - event: 'message_stop', - data: { - type: 'message_stop', - }, - }); - - accumulator.setFinalized(true); - return events; - } - - /** - * Create message_start event - */ - private createMessageStartEvent(accumulator: DeltaAccumulator): AnthropicSSEEvent { - return { - event: 'message_start', - data: { - type: 'message_start', - message: { - id: accumulator.getMessageId(), - type: 'message', - role: accumulator.getRole(), - content: [], - model: accumulator.getModel() || 'glm-4.6', - stop_reason: null, - usage: { - input_tokens: accumulator.getInputTokens(), - output_tokens: 0, - }, - }, - }, - }; - } - - /** - * Create content_block_start event - */ - private createContentBlockStartEvent(block: AccumulatorBlock): AnthropicSSEEvent { - return { - event: 'content_block_start', - data: { - type: 'content_block_start', - index: block.index, - content_block: { - type: block.type, - [block.type]: '', - }, - }, - }; - } - - /** - * Create thinking_delta event - */ - private createThinkingDeltaEvent(block: AccumulatorBlock, delta: string): AnthropicSSEEvent { - return { - event: 'content_block_delta', - data: { - type: 'content_block_delta', - index: block.index, - delta: { - type: 'thinking_delta', - thinking: delta, - }, - }, - }; - } - - /** - * Create text_delta event - */ - private createTextDeltaEvent(block: AccumulatorBlock, delta: string): AnthropicSSEEvent { - return { - event: 'content_block_delta', - data: { - type: 'content_block_delta', - index: block.index, - delta: { - type: 'text_delta', - text: delta, - }, - }, - }; - } - - /** - * Create thinking signature delta event - */ - private createSignatureDeltaEvent(block: AccumulatorBlock): AnthropicSSEEvent | null { - // FIX: Guard against empty content (signature timing race) - if (!block.content || block.content.length === 0) { - if (this.verbose) { - this.log(`WARNING: Skipping signature for empty thinking block ${block.index}`); - this.log( - `This indicates a race condition - signature requested before content accumulated` - ); - } - return null; - } - - const signature = this.generateThinkingSignature(block.content); - - if (this.verbose) { - this.log(`Generating signature for block ${block.index}: ${block.content.length} chars`); - } - - return { - event: 'content_block_delta', - data: { - type: 'content_block_delta', - index: block.index, - delta: { - type: 'thinking_signature_delta', - signature: signature, - }, - }, - }; - } - - /** - * Create content_block_stop event - */ - private createContentBlockStopEvent(block: AccumulatorBlock): AnthropicSSEEvent { - return { - event: 'content_block_stop', - data: { - type: 'content_block_stop', - index: block.index, - }, - }; - } - - /** - * Log message if verbose - */ - private log(message: string): void { - if (this.verbose) { - const timestamp = new Date().toTimeString().split(' ')[0]; // HH:MM:SS - console.error(`[glmt-transformer] [${timestamp}] ${message}`); - } - } } export default GlmtTransformer; diff --git a/src/glmt/pipeline/content-transformer.ts b/src/glmt/pipeline/content-transformer.ts new file mode 100644 index 00000000..21b67533 --- /dev/null +++ b/src/glmt/pipeline/content-transformer.ts @@ -0,0 +1,183 @@ +/** + * ContentTransformer - Handle message sanitization and content transformations + * + * Responsibilities: + * - Sanitize messages for OpenAI API compatibility + * - Extract thinking control tags from messages + * - Detect think keywords in user prompts + * - Transform tools between Anthropic and OpenAI formats + */ + +import type { Message, ContentBlock, AnthropicTool, OpenAITool, ThinkingConfig } from './types'; + +export class ContentTransformer { + private defaultThinking: boolean; + + constructor(defaultThinking = true) { + this.defaultThinking = defaultThinking; + } + + /** + * Sanitize messages for OpenAI API compatibility + */ + sanitizeMessages(messages: Message[]): Message[] { + const result: Message[] = []; + + for (const msg of messages) { + // If content is a string, add as-is + if (typeof msg.content === 'string') { + result.push(msg); + continue; + } + + // If content is an array, process blocks + if (Array.isArray(msg.content)) { + // Separate tool_result blocks from other content + const toolResults = msg.content.filter((block) => block.type === 'tool_result'); + const textBlocks = msg.content.filter((block) => block.type === 'text'); + + // CRITICAL: Tool messages must come BEFORE user text in OpenAI API + for (const toolResult of toolResults) { + result.push({ + role: 'tool', + content: + typeof toolResult.content === 'string' + ? toolResult.content + : JSON.stringify(toolResult.content), + } as Message & { tool_call_id: string }); + } + + // Add text content as user/assistant message AFTER tool messages + if (textBlocks.length > 0) { + const textContent = + textBlocks.length === 1 + ? textBlocks[0].text || '' + : textBlocks.map((b) => b.text || '').join('\n'); + + result.push({ + role: msg.role, + content: textContent, + }); + } + + // If no content at all, add empty message + if (textBlocks.length === 0 && toolResults.length === 0) { + result.push({ + role: msg.role, + content: '', + }); + } + + continue; + } + + // Fallback: return message as-is + result.push(msg); + } + + return result; + } + + /** + * Transform Anthropic tools to OpenAI tools format + */ + transformTools(anthropicTools: AnthropicTool[]): OpenAITool[] { + return anthropicTools.map((tool) => ({ + type: 'function' as const, + function: { + name: tool.name, + description: tool.description, + parameters: tool.input_schema || {}, + }, + })); + } + + /** + * Check if messages contain thinking control tags + */ + hasThinkingTags(messages: Message[]): boolean { + for (const msg of messages) { + if (msg.role !== 'user') continue; + const content = msg.content; + if (typeof content !== 'string') continue; + + // Check for control tags + if (//i.test(content) || //i.test(content)) { + return true; + } + } + return false; + } + + /** + * Extract thinking control tags from user messages + */ + extractThinkingControl(messages: Message[]): ThinkingConfig { + const config: ThinkingConfig = { + thinking: this.defaultThinking, + effort: 'medium', + }; + + // Scan user messages for control tags + for (const msg of messages) { + if (msg.role !== 'user') continue; + + const content = msg.content; + if (typeof content !== 'string') continue; + + // Check for + const thinkingMatch = content.match(//i); + if (thinkingMatch) { + config.thinking = thinkingMatch[1].toLowerCase() === 'on'; + } + + // Check for + const effortMatch = content.match(//i); + if (effortMatch) { + config.effort = effortMatch[1].toLowerCase(); + } + } + + return config; + } + + /** + * Detect Anthropic-style "think" keywords in user prompts + */ + detectThinkKeywords( + messages: Message[] + ): { thinking: boolean; effort: string; keyword: string } | null { + if (!messages || messages.length === 0) return null; + + // Extract text from user messages + const text = messages + .filter((m) => m.role === 'user') + .map((m) => { + if (typeof m.content === 'string') return m.content; + if (Array.isArray(m.content)) { + return (m.content as ContentBlock[]) + .filter((block) => block.type === 'text') + .map((block) => block.text || '') + .join(' '); + } + return ''; + }) + .join(' '); + + // Priority: ultrathink > think harder > think hard > think + if (/\bultrathink\b/i.test(text)) { + return { thinking: true, effort: 'max', keyword: 'ultrathink' }; + } + if (/\bthink\s+harder\b/i.test(text)) { + return { thinking: true, effort: 'high', keyword: 'think harder' }; + } + if (/\bthink\s+hard\b/i.test(text)) { + return { thinking: true, effort: 'medium', keyword: 'think hard' }; + } + if (/\bthink\b/i.test(text)) { + return { thinking: true, effort: 'low', keyword: 'think' }; + } + + return null; // No keywords detected + } +} diff --git a/src/glmt/pipeline/index.ts b/src/glmt/pipeline/index.ts new file mode 100644 index 00000000..7b157513 --- /dev/null +++ b/src/glmt/pipeline/index.ts @@ -0,0 +1,35 @@ +/** + * Pipeline Module Exports + * + * Barrel file for the GLMT transformation pipeline + */ + +// Types +export type { + ContentBlock, + Message, + AnthropicTool, + OpenAITool, + OpenAIToolCall, + OpenAIToolCallDelta, + AnthropicRequest, + OpenAIRequest, + ThinkingConfig, + TransformResult, + ThinkingSignature, + OpenAIChoice, + OpenAIResponse, + AnthropicResponse, + SSEEvent, + AnthropicSSEEvent, + AccumulatorBlock, + ValidationResult, + GlmtTransformerConfig, +} from './types'; + +// Pipeline components +export { ContentTransformer } from './content-transformer'; +export { ToolCallHandler } from './tool-call-handler'; +export { ResponseBuilder } from './response-builder'; +export { StreamParser, type StreamParserConfig } from './stream-parser'; +export { RequestTransformer, type RequestTransformerConfig } from './request-transformer'; diff --git a/src/glmt/pipeline/request-transformer.ts b/src/glmt/pipeline/request-transformer.ts new file mode 100644 index 00000000..98a08ea9 --- /dev/null +++ b/src/glmt/pipeline/request-transformer.ts @@ -0,0 +1,161 @@ +/** + * RequestTransformer - Handle Anthropic → OpenAI request transformation + * + * Responsibilities: + * - Transform Anthropic request format to OpenAI format + * - Inject locale and reasoning instructions + * - Map models and configure parameters + */ + +import { LocaleEnforcer } from '../locale-enforcer'; +import { ReasoningEnforcer } from '../reasoning-enforcer'; +import { ContentTransformer } from './content-transformer'; +import type { + Message, + AnthropicRequest, + OpenAIRequest, + ThinkingConfig, + TransformResult, +} from './types'; + +export interface RequestTransformerConfig { + defaultThinking?: boolean; + verbose?: boolean; + explicitReasoning?: boolean; + log?: (message: string) => void; +} + +export class RequestTransformer { + private localeEnforcer: LocaleEnforcer; + private reasoningEnforcer: ReasoningEnforcer; + private contentTransformer: ContentTransformer; + private log: (message: string) => void; + private modelMaxTokens: Record; + + constructor(config: RequestTransformerConfig = {}) { + const defaultThinking = config.defaultThinking ?? true; + this.log = config.log || (() => {}); + + this.localeEnforcer = new LocaleEnforcer(); + this.reasoningEnforcer = new ReasoningEnforcer({ + enabled: config.explicitReasoning ?? true, + }); + this.contentTransformer = new ContentTransformer(defaultThinking); + + this.modelMaxTokens = { + 'GLM-4.6': 128000, + 'GLM-4.5': 96000, + 'GLM-4.5-air': 16000, + }; + } + + /** + * Transform Anthropic request to OpenAI format + */ + transform(anthropicRequest: AnthropicRequest): TransformResult { + try { + const messages = anthropicRequest.messages || []; + + // 1. Extract thinking control from messages + const thinkingConfig = this.contentTransformer.extractThinkingControl(messages); + const hasControlTags = this.contentTransformer.hasThinkingTags(messages); + + // 2. Detect "think" keywords in user prompts + const keywordConfig = this.contentTransformer.detectThinkKeywords(messages); + if (keywordConfig && !anthropicRequest.thinking && !hasControlTags) { + thinkingConfig.thinking = keywordConfig.thinking; + thinkingConfig.effort = keywordConfig.effort; + this.log( + `Detected think keyword: ${keywordConfig.keyword}, effort=${keywordConfig.effort}` + ); + } + + // 3. Check anthropicRequest.thinking parameter (takes precedence) + if (anthropicRequest.thinking) { + if (anthropicRequest.thinking.type === 'enabled') { + thinkingConfig.thinking = true; + this.log('Claude CLI explicitly enabled thinking'); + } else if (anthropicRequest.thinking.type === 'disabled') { + thinkingConfig.thinking = false; + this.log('Claude CLI explicitly disabled thinking'); + } + } + + this.log(`Final thinking control: ${JSON.stringify(thinkingConfig)}`); + + // 4. Map model + const glmModel = this.mapModel(anthropicRequest.model); + + // 5. Inject locale instruction + const messagesWithLocale = this.localeEnforcer.injectInstruction( + messages as Parameters[0] + ) as unknown as Message[]; + + // 6. Inject reasoning instruction + const messagesWithReasoning = this.reasoningEnforcer.injectInstruction( + messagesWithLocale as unknown as Parameters< + typeof this.reasoningEnforcer.injectInstruction + >[0], + thinkingConfig + ) as unknown as Message[]; + + // 7. Build OpenAI request + const openaiRequest: OpenAIRequest = { + model: glmModel, + messages: this.contentTransformer.sanitizeMessages(messagesWithReasoning), + max_tokens: this.getMaxTokens(glmModel), + stream: anthropicRequest.stream ?? false, + }; + + // 8. Transform tools if present + if (anthropicRequest.tools && anthropicRequest.tools.length > 0) { + openaiRequest.tools = this.contentTransformer.transformTools(anthropicRequest.tools); + openaiRequest.tool_choice = 'auto'; + this.log(`Transformed ${anthropicRequest.tools.length} tools for OpenAI format`); + } + + // 9. Preserve optional parameters + if (anthropicRequest.temperature !== undefined) { + openaiRequest.temperature = anthropicRequest.temperature; + } + if (anthropicRequest.top_p !== undefined) { + openaiRequest.top_p = anthropicRequest.top_p; + } + if (anthropicRequest.stream !== undefined) { + openaiRequest.stream = anthropicRequest.stream; + } + + // 10. Inject reasoning parameters + this.injectReasoningParams(openaiRequest, thinkingConfig); + + return { openaiRequest, thinkingConfig }; + } catch (error) { + const err = error as Error; + console.error('[RequestTransformer] Transformation error:', err); + return { + openaiRequest: anthropicRequest, + thinkingConfig: { thinking: false, effort: 'medium' }, + error: err.message, + }; + } + } + + private injectReasoningParams( + openaiRequest: OpenAIRequest, + thinkingConfig: ThinkingConfig + ): void { + openaiRequest.do_sample = true; + if (thinkingConfig.thinking) { + openaiRequest.reasoning = true; + openaiRequest.reasoning_effort = thinkingConfig.effort; + } + } + + private mapModel(_anthropicModel: string): string { + return 'GLM-4.6'; + } + + private getMaxTokens(model: string): number { + return this.modelMaxTokens[model] || 128000; + } +} diff --git a/src/glmt/pipeline/response-builder.ts b/src/glmt/pipeline/response-builder.ts new file mode 100644 index 00000000..416a034f --- /dev/null +++ b/src/glmt/pipeline/response-builder.ts @@ -0,0 +1,202 @@ +/** + * ResponseBuilder - Create SSE events for Anthropic streaming format + * + * Responsibilities: + * - Create message_start, message_delta, message_stop events + * - Create content_block_start, content_block_delta, content_block_stop events + * - Generate thinking signatures + * - Map stop reasons between formats + */ + +import * as crypto from 'crypto'; +import type { DeltaAccumulator } from '../delta-accumulator'; +import type { AccumulatorBlock, AnthropicSSEEvent, ThinkingSignature } from './types'; + +export class ResponseBuilder { + private verbose: boolean; + + constructor(verbose = false) { + this.verbose = verbose; + } + + /** + * Create message_start event + */ + createMessageStartEvent(accumulator: DeltaAccumulator): AnthropicSSEEvent { + return { + event: 'message_start', + data: { + type: 'message_start', + message: { + id: accumulator.getMessageId(), + type: 'message', + role: accumulator.getRole(), + content: [], + model: accumulator.getModel() || 'glm-4.6', + stop_reason: null, + usage: { + input_tokens: accumulator.getInputTokens(), + output_tokens: 0, + }, + }, + }, + }; + } + + /** + * Create content_block_start event + */ + createContentBlockStartEvent(block: AccumulatorBlock): AnthropicSSEEvent { + return { + event: 'content_block_start', + data: { + type: 'content_block_start', + index: block.index, + content_block: { + type: block.type, + [block.type]: '', + }, + }, + }; + } + + /** + * Create thinking_delta event + */ + createThinkingDeltaEvent(block: AccumulatorBlock, delta: string): AnthropicSSEEvent { + return { + event: 'content_block_delta', + data: { + type: 'content_block_delta', + index: block.index, + delta: { + type: 'thinking_delta', + thinking: delta, + }, + }, + }; + } + + /** + * Create text_delta event + */ + createTextDeltaEvent(block: AccumulatorBlock, delta: string): AnthropicSSEEvent { + return { + event: 'content_block_delta', + data: { + type: 'content_block_delta', + index: block.index, + delta: { + type: 'text_delta', + text: delta, + }, + }, + }; + } + + /** + * Create thinking signature delta event + */ + createSignatureDeltaEvent(block: AccumulatorBlock): AnthropicSSEEvent | null { + // FIX: Guard against empty content (signature timing race) + if (!block.content || block.content.length === 0) { + if (this.verbose) { + console.error( + `[ResponseBuilder] WARNING: Skipping signature for empty thinking block ${block.index}` + ); + console.error( + `This indicates a race condition - signature requested before content accumulated` + ); + } + return null; + } + + const signature = this.generateThinkingSignature(block.content); + + if (this.verbose) { + console.error( + `[ResponseBuilder] Generating signature for block ${block.index}: ${block.content.length} chars` + ); + } + + return { + event: 'content_block_delta', + data: { + type: 'content_block_delta', + index: block.index, + delta: { + type: 'thinking_signature_delta', + signature: signature, + }, + }, + }; + } + + /** + * Create content_block_stop event + */ + createContentBlockStopEvent(block: AccumulatorBlock): AnthropicSSEEvent { + return { + event: 'content_block_stop', + data: { + type: 'content_block_stop', + index: block.index, + }, + }; + } + + /** + * Generate finalization events (message_delta + message_stop) + */ + createFinalizationEvents(accumulator: DeltaAccumulator, stopReason: string): AnthropicSSEEvent[] { + return [ + { + event: 'message_delta', + data: { + type: 'message_delta', + delta: { + stop_reason: stopReason, + }, + usage: { + input_tokens: accumulator.getInputTokens(), + output_tokens: accumulator.getOutputTokens(), + }, + }, + }, + { + event: 'message_stop', + data: { + type: 'message_stop', + }, + }, + ]; + } + + /** + * Generate thinking signature for Claude Code UI + */ + generateThinkingSignature(thinking: string): ThinkingSignature { + // Generate signature hash + const hash = crypto.createHash('sha256').update(thinking).digest('hex').substring(0, 16); + + return { + type: 'thinking_signature', + hash: hash, + length: thinking.length, + timestamp: Date.now(), + }; + } + + /** + * Map OpenAI stop reason to Anthropic stop reason + */ + mapStopReason(openaiReason: string): string { + const mapping: Record = { + stop: 'end_turn', + length: 'max_tokens', + tool_calls: 'tool_use', + content_filter: 'stop_sequence', + }; + return mapping[openaiReason] || 'end_turn'; + } +} diff --git a/src/glmt/pipeline/stream-parser.ts b/src/glmt/pipeline/stream-parser.ts new file mode 100644 index 00000000..c0b6fc71 --- /dev/null +++ b/src/glmt/pipeline/stream-parser.ts @@ -0,0 +1,308 @@ +/** + * StreamParser - Transform OpenAI streaming deltas to Anthropic SSE events + * + * Responsibilities: + * - Process streaming deltas (reasoning_content, content, tool_calls) + * - Coordinate with accumulator for state tracking + * - Detect and handle planning loops + * - Generate appropriate Anthropic SSE events + */ + +import type { DeltaAccumulator } from '../delta-accumulator'; +import type { SSEEvent, AnthropicSSEEvent, AccumulatorBlock } from './types'; +import { ResponseBuilder } from './response-builder'; +import { ToolCallHandler } from './tool-call-handler'; + +export interface StreamParserConfig { + verbose?: boolean; + debugMode?: boolean; + debugLog?: boolean; + writeDebugLog?: (type: string, data: unknown) => void; +} + +export class StreamParser { + private verbose: boolean; + private debugMode: boolean; + private debugLog: boolean; + private responseBuilder: ResponseBuilder; + private toolCallHandler: ToolCallHandler; + private writeDebugLog: (type: string, data: unknown) => void; + + constructor(config: StreamParserConfig = {}) { + this.verbose = config.verbose || false; + this.debugMode = config.debugMode || false; + this.debugLog = config.debugLog || false; + this.responseBuilder = new ResponseBuilder(this.verbose); + this.toolCallHandler = new ToolCallHandler(); + this.writeDebugLog = config.writeDebugLog || (() => {}); + } + + /** + * Transform OpenAI streaming delta to Anthropic events + */ + transformDelta(openaiEvent: SSEEvent, accumulator: DeltaAccumulator): AnthropicSSEEvent[] { + const events: AnthropicSSEEvent[] = []; + + // Debug logging for streaming deltas + if (this.debugLog && openaiEvent.data) { + this.writeDebugLog('delta-openai', openaiEvent.data); + } + + // Handle [DONE] marker + if (openaiEvent.event === 'done') { + if (!accumulator.isFinalized()) { + return this.finalizeDelta(accumulator); + } + return []; // Already finalized + } + + // Usage update (appears in final chunk, may be before choice data) + if (openaiEvent.data?.usage) { + accumulator.updateUsage(openaiEvent.data.usage); + + // If we have both usage AND finish_reason, finalize immediately + if (accumulator.getFinishReason()) { + events.push(...this.finalizeDelta(accumulator)); + return events; + } + } + + const choice = openaiEvent.data?.choices?.[0]; + if (!choice) return events; + + const delta = choice.delta; + if (!delta) return events; + + // Message start + if (!accumulator.isMessageStarted()) { + if (openaiEvent.data?.model) { + accumulator.setModel(openaiEvent.data.model); + } + events.push(this.responseBuilder.createMessageStartEvent(accumulator)); + accumulator.setMessageStarted(true); + } + + // Role + if (delta.role) { + accumulator.setRole(delta.role); + } + + // Reasoning content delta + if (delta.reasoning_content) { + events.push(...this.handleReasoningDelta(delta.reasoning_content, accumulator)); + } + + // Text content delta + if (delta.content) { + events.push(...this.handleContentDelta(delta.content, accumulator)); + } + + // Check for planning loop + if (accumulator.checkForLoop()) { + this.log( + 'WARNING: Planning loop detected - 3 consecutive thinking blocks with no tool calls' + ); + this.log('Forcing early finalization to prevent unbounded planning'); + events.push(...this.forceFinalization(accumulator)); + return events; + } + + // Tool calls deltas + if (delta.tool_calls && delta.tool_calls.length > 0) { + events.push(...this.handleToolCallDeltas(delta.tool_calls, accumulator)); + } + + // Finish reason + if (choice.finish_reason) { + accumulator.setFinishReason(choice.finish_reason); + + // If we have both finish_reason AND usage, finalize immediately + if (accumulator.hasUsageReceived()) { + events.push(...this.finalizeDelta(accumulator)); + } + } + + // Debug logging for generated events + if (this.debugLog && events.length > 0) { + this.writeDebugLog('delta-anthropic-events', { + events, + accumulator: accumulator.getSummary(), + }); + } + + return events; + } + + /** + * Handle reasoning content delta + */ + private handleReasoningDelta( + reasoningContent: string, + accumulator: DeltaAccumulator + ): AnthropicSSEEvent[] { + const events: AnthropicSSEEvent[] = []; + const currentBlock = accumulator.getCurrentBlock(); + + if (this.debugMode) { + console.error(`[StreamParser] Reasoning delta: ${reasoningContent.length} chars`); + console.error( + `[StreamParser] Current block: ${currentBlock?.type || 'none'}, index: ${currentBlock?.index ?? 'N/A'}` + ); + } + + if (!currentBlock || currentBlock.type !== 'thinking') { + // Start thinking block + const block = accumulator.startBlock('thinking'); + events.push(this.responseBuilder.createContentBlockStartEvent(block)); + + if (this.debugMode) { + console.error(`[StreamParser] Started new thinking block ${block.index}`); + } + } + + accumulator.addDelta(reasoningContent); + const currentThinkingBlock = accumulator.getCurrentBlock(); + if (currentThinkingBlock) { + events.push( + this.responseBuilder.createThinkingDeltaEvent(currentThinkingBlock, reasoningContent) + ); + } + + return events; + } + + /** + * Handle content delta + */ + private handleContentDelta(content: string, accumulator: DeltaAccumulator): AnthropicSSEEvent[] { + const events: AnthropicSSEEvent[] = []; + const currentBlock = accumulator.getCurrentBlock(); + + // Close thinking block if transitioning from thinking to text + if (currentBlock && currentBlock.type === 'thinking' && !currentBlock.stopped) { + events.push(...this.closeThinkingBlock(currentBlock, accumulator)); + } + + if (!accumulator.getCurrentBlock() || accumulator.getCurrentBlock()?.type !== 'text') { + // Start text block + const block = accumulator.startBlock('text'); + events.push(this.responseBuilder.createContentBlockStartEvent(block)); + } + + accumulator.addDelta(content); + const currentTextBlock = accumulator.getCurrentBlock(); + if (currentTextBlock) { + events.push(this.responseBuilder.createTextDeltaEvent(currentTextBlock, content)); + } + + return events; + } + + /** + * Handle tool call deltas + */ + private handleToolCallDeltas( + toolCallDeltas: import('./types').OpenAIToolCallDelta[], + accumulator: DeltaAccumulator + ): AnthropicSSEEvent[] { + const events: AnthropicSSEEvent[] = []; + + if (!toolCallDeltas) return events; + + // Close current content block ONCE before processing any tool calls + const currentBlock = accumulator.getCurrentBlock(); + if (currentBlock && !currentBlock.stopped) { + if (currentBlock.type === 'thinking') { + events.push(...this.closeThinkingBlock(currentBlock, accumulator)); + } else { + events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock)); + accumulator.stopCurrentBlock(); + } + } + + // Process tool call deltas + events.push(...this.toolCallHandler.processToolCallDeltas(toolCallDeltas, accumulator)); + + return events; + } + + /** + * Close thinking block with signature + */ + private closeThinkingBlock( + block: AccumulatorBlock, + accumulator: DeltaAccumulator + ): AnthropicSSEEvent[] { + const events: AnthropicSSEEvent[] = []; + + const signatureEvent = this.responseBuilder.createSignatureDeltaEvent(block); + if (signatureEvent) { + events.push(signatureEvent); + } + events.push(this.responseBuilder.createContentBlockStopEvent(block)); + accumulator.stopCurrentBlock(); + + return events; + } + + /** + * Force finalization due to loop detection + */ + private forceFinalization(accumulator: DeltaAccumulator): AnthropicSSEEvent[] { + const events: AnthropicSSEEvent[] = []; + + // Close current block if any + const currentBlock = accumulator.getCurrentBlock(); + if (currentBlock && !currentBlock.stopped) { + if (currentBlock.type === 'thinking') { + events.push(...this.closeThinkingBlock(currentBlock, accumulator)); + } else { + events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock)); + accumulator.stopCurrentBlock(); + } + } + + // Force finalization + events.push(...this.finalizeDelta(accumulator)); + return events; + } + + /** + * Finalize streaming and generate closing events + */ + finalizeDelta(accumulator: DeltaAccumulator): AnthropicSSEEvent[] { + if (accumulator.isFinalized()) { + return []; // Already finalized + } + + const events: AnthropicSSEEvent[] = []; + + // Close current content block if any + const currentBlock = accumulator.getCurrentBlock(); + if (currentBlock && !currentBlock.stopped) { + if (currentBlock.type === 'thinking') { + events.push(...this.closeThinkingBlock(currentBlock, accumulator)); + } else { + events.push(this.responseBuilder.createContentBlockStopEvent(currentBlock)); + accumulator.stopCurrentBlock(); + } + } + + // Message delta (stop reason + usage) and message stop + const stopReason = this.responseBuilder.mapStopReason(accumulator.getFinishReason() || 'stop'); + events.push(...this.responseBuilder.createFinalizationEvents(accumulator, stopReason)); + + accumulator.setFinalized(true); + return events; + } + + /** + * Log message if verbose + */ + private log(message: string): void { + if (this.verbose) { + const timestamp = new Date().toTimeString().split(' ')[0]; // HH:MM:SS + console.error(`[StreamParser] [${timestamp}] ${message}`); + } + } +} diff --git a/src/glmt/pipeline/tool-call-handler.ts b/src/glmt/pipeline/tool-call-handler.ts new file mode 100644 index 00000000..ffe0d670 --- /dev/null +++ b/src/glmt/pipeline/tool-call-handler.ts @@ -0,0 +1,97 @@ +/** + * ToolCallHandler - Handle tool call processing for streaming responses + * + * Responsibilities: + * - Process tool call deltas from OpenAI + * - Generate tool_use content blocks for Anthropic format + * - Handle input_json_delta events + */ + +import type { DeltaAccumulator } from '../delta-accumulator'; +import type { OpenAIToolCallDelta, OpenAIToolCall, ContentBlock, AnthropicSSEEvent } from './types'; + +export class ToolCallHandler { + /** + * Process tool calls from non-streaming response + */ + processToolCalls(toolCalls: OpenAIToolCall[]): ContentBlock[] { + const content: ContentBlock[] = []; + + for (const toolCall of toolCalls) { + let parsedInput: Record; + try { + parsedInput = JSON.parse(toolCall.function.arguments || '{}'); + } catch (parseError) { + const err = parseError as Error; + console.error(`[ToolCallHandler] Invalid JSON in tool arguments: ${err.message}`); + parsedInput = { _error: 'Invalid JSON', _raw: toolCall.function.arguments }; + } + + content.push({ + type: 'tool_use', + id: toolCall.id, + name: toolCall.function.name, + input: parsedInput, + }); + } + + return content; + } + + /** + * Process tool call deltas during streaming + * Returns events for tool use blocks and input_json deltas + */ + processToolCallDeltas( + toolCallDeltas: OpenAIToolCallDelta[], + accumulator: DeltaAccumulator + ): AnthropicSSEEvent[] { + const events: AnthropicSSEEvent[] = []; + + for (const toolCallDelta of toolCallDeltas) { + // Track tool call state + const isNewToolCall = !accumulator.hasToolCall(toolCallDelta.index); + accumulator.addToolCallDelta(toolCallDelta); + + // Emit tool use events (start + input_json deltas) + if (isNewToolCall) { + // Start new tool_use block in accumulator + const block = accumulator.startBlock('tool_use'); + const toolCall = accumulator.getToolCall(toolCallDelta.index); + + events.push({ + event: 'content_block_start', + data: { + type: 'content_block_start', + index: block.index, + content_block: { + type: 'tool_use', + id: toolCall?.id || `tool_${toolCallDelta.index}`, + name: toolCall?.function?.name || '', + }, + }, + }); + } + + // Emit input_json delta if arguments present + if (toolCallDelta.function?.arguments) { + const currentToolBlock = accumulator.getCurrentBlock(); + if (currentToolBlock && currentToolBlock.type === 'tool_use') { + events.push({ + event: 'content_block_delta', + data: { + type: 'content_block_delta', + index: currentToolBlock.index, + delta: { + type: 'input_json_delta', + partial_json: toolCallDelta.function.arguments, + }, + }, + }); + } + } + } + + return events; + } +} diff --git a/src/glmt/pipeline/types.ts b/src/glmt/pipeline/types.ts new file mode 100644 index 00000000..27f68749 --- /dev/null +++ b/src/glmt/pipeline/types.ts @@ -0,0 +1,185 @@ +/** + * Pipeline Types - Shared types for GLMT transformation pipeline + */ + +// Content block types +export interface ContentBlock { + type: string; + text?: string; + thinking?: string; + signature?: ThinkingSignature; + id?: string; + name?: string; + input?: Record; + tool_use_id?: string; + content?: string | unknown; +} + +export interface Message { + role: string; + content: string | ContentBlock[]; +} + +// Tool types +export interface AnthropicTool { + name: string; + description: string; + input_schema: Record; +} + +export interface OpenAITool { + type: 'function'; + function: { + name: string; + description: string; + parameters: Record; + }; +} + +export interface OpenAIToolCall { + id: string; + type: string; + function: { + name: string; + arguments: string; + }; +} + +export interface OpenAIToolCallDelta { + index: number; + id?: string; + type?: string; + function?: { + name?: string; + arguments?: string; + }; +} + +// Request/Response types +export interface AnthropicRequest { + model: string; + messages: Message[]; + max_tokens?: number; + temperature?: number; + top_p?: number; + tools?: AnthropicTool[]; + stream?: boolean; + thinking?: { + type: 'enabled' | 'disabled'; + budget_tokens?: number; + }; +} + +export interface OpenAIRequest { + model: string; + messages: Message[]; + max_tokens?: number; + temperature?: number; + top_p?: number; + tools?: OpenAITool[]; + tool_choice?: string; + stream?: boolean; + do_sample?: boolean; + reasoning?: boolean; + reasoning_effort?: string; +} + +export interface ThinkingConfig { + thinking: boolean; + effort: string; +} + +export interface TransformResult { + openaiRequest: OpenAIRequest | AnthropicRequest; + thinkingConfig: ThinkingConfig; + error?: string; +} + +export interface ThinkingSignature { + type: string; + hash: string; + length: number; + timestamp: number; +} + +// OpenAI response types +export interface OpenAIChoice { + message: { + role?: string; + content?: string | null; + reasoning_content?: string; + tool_calls?: OpenAIToolCall[]; + }; + delta?: { + role?: string; + content?: string; + reasoning_content?: string; + tool_calls?: OpenAIToolCallDelta[]; + }; + finish_reason?: string; +} + +export interface OpenAIResponse { + id?: string; + model?: string; + choices?: OpenAIChoice[]; + usage?: { + prompt_tokens?: number; + completion_tokens?: number; + }; +} + +export interface AnthropicResponse { + id: string; + type: string; + role: string; + content: ContentBlock[]; + model: string; + stop_reason: string; + usage: { + input_tokens: number; + output_tokens: number; + }; +} + +// SSE event types +export interface SSEEvent { + event?: string; + data?: OpenAIResponse & { + usage?: { + prompt_tokens?: number; + completion_tokens?: number; + }; + }; +} + +export interface AnthropicSSEEvent { + event: string; + data: Record; +} + +// Accumulator block type +export interface AccumulatorBlock { + index: number; + type: string; + content: string; + stopped: boolean; +} + +// Validation result +export interface ValidationResult { + checks: Record; + passed: number; + total: number; + valid: boolean; +} + +// Config types +export interface GlmtTransformerConfig { + defaultThinking?: boolean; + verbose?: boolean; + debugLog?: boolean; + debugMode?: boolean; + debugLogDir?: string; + explicitReasoning?: boolean; +}