import Anthropic from "@anthropic-ai/sdk"; import { ChatOpenAI } from "@langchain/openai"; import { HumanMessage, AIMessage, SystemMessage, ToolMessage, BaseMessage } from "@langchain/core/messages"; import { spawn } from "child_process"; import fs from "fs"; import path from "path"; import { RojoToolContext } from "./tools/RojoTool.js"; export interface AgentConfig { baseURL?: string; model: string; apiKey: string; provider?: "openai" | "claude" | "claude-code"; headers?: Record; systemPrompt?: string; tools?: object[]; } export type AgentEvent = | { type: "text"; text: string } | { type: "tool_call"; name: string; args: Record } | { type: "tool_result"; name: string; result: string }; type ToolExecutor = ( name: string, args: Record, ctx: RojoToolContext ) => Promise; type SessionData = { id: string; messages: BaseMessage[] }; type ClaudeSessionData = { id: string; messages: Anthropic.MessageParam[] }; const sessions = new Map(); const claudeSessions = new Map(); const claudeCodeSessions = new Set(); // ── Claude binary path ─────────────────────────────────────────────────────── function findClaudeBin(): string { const win = process.platform === "win32"; const direct = path.join( process.cwd(), "node_modules", "@anthropic-ai", "claude-code", "bin", win ? "claude.exe" : "claude" ); if ( fs.existsSync( direct ) ) return direct; const wrapper = path.join( process.cwd(), "node_modules", ".bin", win ? "claude.cmd" : "claude" ); return wrapper; } const CLAUDE_BIN = findClaudeBin(); // ── Main entry ─────────────────────────────────────────────────────────────── export async function* getAgentStream( id: string, userMessage: string, config: AgentConfig, toolContext?: RojoToolContext, toolExecutor?: ToolExecutor, ): AsyncGenerator { if ( config.provider === "claude" ) { yield* getClaudeAgentStream( id, userMessage, config, toolContext, toolExecutor ); return; } if ( config.provider === "claude-code" ) { yield* getClaudeCodeAgentStream( id, userMessage, config ); return; } // ── OpenAI-compatible path ───────────────────────────────────────────────── const model = new ChatOpenAI( { model: config.model, apiKey: config.apiKey, configuration: { baseURL: config.baseURL, defaultHeaders: config.headers ?? {}, }, } ); if ( !sessions.has( id ) ) { const messages: BaseMessage[] = []; if ( config.systemPrompt?.trim() ) { messages.push( new SystemMessage( config.systemPrompt ) ); } sessions.set( id, { id, messages } ); } const session = sessions.get( id )!; const tools = config.tools ?? []; const boundModel = tools.length > 0 ? model.bindTools( tools ) : model; session.messages.push( new HumanMessage( userMessage ) ); while ( true ) { const stream = await boundModel.stream( session.messages ); let accumulated: any = null; for await ( const chunk of stream ) { const text = extractText( chunk.content ); if ( text ) yield { type: "text", text }; accumulated = accumulated ? accumulated.concat( chunk ) : chunk; } const toolCalls: any[] = accumulated?.tool_calls ?? []; console.log( "[RojosAgent] loop end — tool_calls:", JSON.stringify( toolCalls ), "| text length:", extractText( accumulated?.content ).length, "| accumulated null:", accumulated === null || accumulated === undefined, "| content type:", typeof accumulated?.content, "| content:", JSON.stringify( accumulated?.content )?.slice( 0, 200 ) ); if ( toolCalls.length > 0 && toolContext && toolExecutor ) { session.messages.push( new AIMessage( { content: extractText( accumulated?.content ), tool_calls: toolCalls } ) ); for ( const tc of toolCalls ) { yield { type: "tool_call", name: tc.name, args: tc.args }; let result: unknown; try { result = await toolExecutor( tc.name, tc.args, toolContext ); } catch ( err ) { result = { error: String( err ) }; } const resultStr = typeof result === "string" ? result : JSON.stringify( result ); yield { type: "tool_result", name: tc.name, result: resultStr }; session.messages.push( new ToolMessage( { content: resultStr, tool_call_id: tc.id } ) ); } } else { const text = extractText( accumulated?.content ); if ( text ) session.messages.push( new AIMessage( text ) ); break; } } } // ── Claude API provider ────────────────────────────────────────────────────── function toClaudeTool( t: any ): Anthropic.Tool { const fn = t.function ?? t; return { name: fn.name, description: fn.description ?? "", input_schema: fn.parameters as Anthropic.Tool[ "input_schema" ], }; } async function* getClaudeAgentStream( id: string, userMessage: string, config: AgentConfig, toolContext?: RojoToolContext, toolExecutor?: ToolExecutor, ): AsyncGenerator { const client = new Anthropic( { apiKey: config.apiKey } ); if ( !claudeSessions.has( id ) ) { claudeSessions.set( id, { id, messages: [] } ); } const session = claudeSessions.get( id )!; const claudeTools = ( config.tools ?? [] ).map( toClaudeTool ); session.messages.push( { role: "user", content: userMessage } ); while ( true ) { const streamParams: Anthropic.MessageStreamParams = { model: config.model, max_tokens: 8192, messages: session.messages, ...( config.systemPrompt?.trim() ? { system: config.systemPrompt } : {} ), ...( claudeTools.length > 0 ? { tools: claudeTools } : {} ), }; const stream = client.messages.stream( streamParams ); for await ( const event of stream ) { if ( event.type === "content_block_delta" && event.delta.type === "text_delta" ) { yield { type: "text", text: event.delta.text }; } } const finalMsg = await stream.finalMessage(); const toolBlocks = finalMsg.content.filter( b => b.type === "tool_use" ) as Anthropic.ToolUseBlock[]; console.log( "[RojosAgent/claude] stop_reason:", finalMsg.stop_reason, "| tool_calls:", toolBlocks.length ); if ( toolBlocks.length > 0 && toolContext && toolExecutor ) { session.messages.push( { role: "assistant", content: finalMsg.content } ); const toolResults: Anthropic.ToolResultBlockParam[] = []; for ( const tc of toolBlocks ) { yield { type: "tool_call", name: tc.name, args: tc.input as Record }; let result: unknown; try { result = await toolExecutor( tc.name, tc.input as Record, toolContext ); } catch ( err ) { result = { error: String( err ) }; } const resultStr = typeof result === "string" ? result : JSON.stringify( result ); yield { type: "tool_result", name: tc.name, result: resultStr }; toolResults.push( { type: "tool_result", tool_use_id: tc.id, content: resultStr } ); } session.messages.push( { role: "user", content: toolResults } ); } else { const fullText = finalMsg.content .filter( b => b.type === "text" ) .map( b => ( b as Anthropic.TextBlock ).text ) .join( "" ); session.messages.push( { role: "assistant", content: fullText } ); break; } } } // ── Claude Code provider (subprocess) ─────────────────────────────────────── async function* getClaudeCodeAgentStream( id: string, userMessage: string, config: AgentConfig, ): AsyncGenerator { const isFirst = !claudeCodeSessions.has( id ); claudeCodeSessions.add( id ); const args: string[] = [ "--print", "--output-format", "stream-json", "--verbose", "--dangerously-skip-permissions", ]; if ( isFirst ) { args.push( "--session-id", id ); if ( config.systemPrompt?.trim() ) { args.push( "--append-system-prompt", config.systemPrompt ); } } else { args.push( "--resume", id ); } args.push( userMessage ); console.log( "[RojosAgent/claude-code] spawning:", CLAUDE_BIN, args.slice( 0, 6 ).join( " " ) ); yield* spawnJsonLines( CLAUDE_BIN, args ); } async function* spawnJsonLines( bin: string, args: string[] ): AsyncGenerator { const queue: AgentEvent[] = []; const waiters: Array<() => void> = []; let done = false; let spawnError: Error | null = null; function push( ev: AgentEvent ): void { queue.push( ev ); if ( waiters.length ) waiters.shift()!(); } function finish( err?: Error ): void { done = true; spawnError = err ?? null; while ( waiters.length ) waiters.shift()!(); } const proc = spawn( bin, args, { cwd: process.cwd(), shell: false, stdio: [ "ignore", "pipe", "pipe" ] } ); let buf = ""; proc.stdout.on( "data", ( chunk: Buffer ) => { buf += chunk.toString( "utf8" ); const lines = buf.split( "\n" ); buf = lines.pop() ?? ""; for ( const line of lines ) { const t = line.trim(); if ( !t ) continue; try { const msg = JSON.parse( t ) as any; const ev = claudeCodeEventToAgentEvent( msg ); if ( ev ) push( ev ); } catch {} } } ); proc.stderr.on( "data", ( chunk: Buffer ) => { console.error( "[RojosAgent/claude-code] stderr:", chunk.toString( "utf8" ).trim() ); } ); proc.on( "error", ( err ) => finish( err ) ); proc.on( "close", ( code ) => { if ( code !== 0 ) console.warn( "[RojosAgent/claude-code] exited with code", code ); finish(); } ); while ( true ) { if ( queue.length > 0 ) { yield queue.shift()!; } else if ( done ) { break; } else { await new Promise( resolve => waiters.push( resolve ) ); } } if ( spawnError ) throw spawnError; } function claudeCodeEventToAgentEvent( msg: any ): AgentEvent | null { if ( msg.type === "assistant" ) { const content = msg.message?.content; if ( !Array.isArray( content ) ) return null; const text = content .filter( ( b: any ) => b.type === "text" ) .map( ( b: any ) => b.text as string ) .join( "" ); return text ? { type: "text", text } : null; } return null; } // ── Utilities ──────────────────────────────────────────────────────────────── function extractText( content: unknown ): string { if ( typeof content === "string" ) return content; if ( Array.isArray( content ) ) { return content .map( c => typeof c === "string" ? c : ( ( c as any )?.text ?? "" ) ) .join( "" ); } return ""; }