rojects/source/server/rojos/RojosAgent.ts

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<string, string>;
  systemPrompt?: string;
  tools?:        object[];
}

export type AgentEvent =
  | { type: "text";        text: string }
  | { type: "tool_call";  name: string; args: Record<string, unknown> }
  | { type: "tool_result"; name: string; result: string };

type ToolExecutor = ( name: string, args: Record<string, unknown>, ctx: RojoToolContext ) => Promise<unknown>;

type SessionData       = { id: string; messages: BaseMessage[] };
type ClaudeSessionData = { id: string; messages: Anthropic.MessageParam[] };

const sessions            = new Map<string, SessionData>();
const claudeSessions      = new Map<string, ClaudeSessionData>();
const claudeCodeSessions  = new Set<string>();

// ── 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<AgentEvent>
{
  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<AgentEvent>
{
  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<string, unknown> };

        let result: unknown;

        try
        {
          result = await toolExecutor( tc.name, tc.input as Record<string, unknown>, 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<AgentEvent>
{
  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<AgentEvent>
{
  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<void>( 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 "";
}