EJClaw/src/agent-runner-process.ts

import type { ChildProcess } from 'child_process';
import fs from 'fs';
import path from 'path';

import {
  AGENT_MAX_OUTPUT_SIZE,
  AGENT_TIMEOUT,
  IDLE_TIMEOUT,
  LOG_LEVEL,
} from './config.js';
import { logger } from './logger.js';
import { OUTPUT_END_MARKER, OUTPUT_START_MARKER } from './agent-protocol.js';
import type { AgentInput, AgentOutput } from './agent-runner.js';
import type { RegisteredGroup } from './types.js';
import { getErrorMessage } from './utils.js';

interface RunSpawnedAgentProcessArgs {
  proc: ChildProcess;
  group: RegisteredGroup;
  input: AgentInput;
  processName: string;
  logsDir: string;
  startTime: number;
  onOutput?: (output: AgentOutput) => Promise<void>;
  onTerminalStreamedOutputFlushed?: (output: AgentOutput) => void;
}

function isTerminalStreamedOutput(output: AgentOutput): boolean {
  return (output.phase ?? 'final') !== 'progress';
}

function parseLegacyAgentOutput(stdout: string): AgentOutput {
  const startIdx = stdout.indexOf(OUTPUT_START_MARKER);
  const endIdx = stdout.indexOf(OUTPUT_END_MARKER);
  let jsonLine: string;
  if (startIdx !== -1 && endIdx !== -1 && endIdx > startIdx) {
    jsonLine = stdout
      .slice(startIdx + OUTPUT_START_MARKER.length, endIdx)
      .trim();
  } else {
    const lines = stdout.trim().split('\n');
    jsonLine = lines[lines.length - 1];
  }
  return JSON.parse(jsonLine) as AgentOutput;
}

function logStreamedOutputDeliveryError(
  err: unknown,
  group: RegisteredGroup,
  input: AgentInput,
): void {
  logger.warn(
    {
      group: group.name,
      chatJid: input.chatJid,
      runId: input.runId,
      error: getErrorMessage(err),
    },
    'Streamed agent output delivery failed',
  );
}

function logStreamedAgentErrorOutput(
  output: AgentOutput,
  group: RegisteredGroup,
  input: AgentInput,
): void {
  if (output.status !== 'error') return;
  logger.warn(
    {
      group: group.name,
      chatJid: input.chatJid,
      runId: input.runId,
      error: output.error,
      newSessionId: output.newSessionId,
    },
    'Streamed agent error output',
  );
}

function chainStreamedOutputDelivery(args: {
  outputChain: Promise<void>;
  parsed: AgentOutput;
  onOutput: (output: AgentOutput) => Promise<void>;
  onTerminalStreamedOutputFlushed?: (output: AgentOutput) => void;
  group: RegisteredGroup;
  input: AgentInput;
}): Promise<void> {
  return args.outputChain.then(async () => {
    try {
      await args.onOutput(args.parsed);
      if (isTerminalStreamedOutput(args.parsed)) {
        args.onTerminalStreamedOutputFlushed?.(args.parsed);
      }
    } catch (err) {
      logStreamedOutputDeliveryError(err, args.group, args.input);
    }
  });
}

function writeTimeoutLog(args: {
  logsDir: string;
  input: AgentInput;
  group: RegisteredGroup;
  processName: string;
  duration: number;
  code: number | null;
  signal: NodeJS.Signals | null;
  hadStreamingOutput: boolean;
}): void {
  const ts = new Date().toISOString().replace(/[:.]/g, '-');
  fs.writeFileSync(
    path.join(args.logsDir, `agent-${args.input.runId || 'adhoc'}-${ts}.log`),
    [
      `=== Agent Run Log (TIMEOUT) ===`,
      `Timestamp: ${new Date().toISOString()}`,
      `Group: ${args.group.name}`,
      `ChatJid: ${args.input.chatJid}`,
      `RunId: ${args.input.runId || 'n/a'}`,
      `Process: ${args.processName}`,
      `Duration: ${args.duration}ms`,
      `Exit Code: ${args.code}`,
      `Signal: ${args.signal}`,
      `Had Streaming Output: ${args.hadStreamingOutput}`,
    ].join('\n'),
  );
}

export function runSpawnedAgentProcess(
  args: RunSpawnedAgentProcessArgs,
): Promise<AgentOutput> {
  const { proc, group, input, processName, logsDir, startTime, onOutput } =
    args;
  return new Promise((resolve) => {
    const stdoutStream = proc.stdout;
    const stderrStream = proc.stderr;
    if (!stdoutStream || !stderrStream) {
      resolve({
        status: 'error',
        result: null,
        error: 'Agent process stdio pipes are unavailable',
      });
      return;
    }

    let stdout = '';
    let stderr = '';
    let stdoutTruncated = false;
    let stderrTruncated = false;

    // Use UTF-8 string decoding on streams to avoid splitting multi-byte
    // characters (e.g. Korean) at chunk boundaries, which produces U+FFFD.
    stdoutStream.setEncoding('utf8');
    stderrStream.setEncoding('utf8');

    // Streaming output: parse OUTPUT_START/END marker pairs as they arrive.
    let parseBuffer = '';
    let newSessionId: string | undefined;
    let outputChain = Promise.resolve();

    let timedOut = false;
    let hadStreamingOutput = false;
    const configTimeout = group.agentConfig?.timeout || AGENT_TIMEOUT;
    const timeoutMs = Math.max(configTimeout, IDLE_TIMEOUT + 30_000);

    const killOnTimeout = () => {
      timedOut = true;
      logger.error(
        {
          group: group.name,
          chatJid: input.chatJid,
          runId: input.runId,
          processName,
        },
        'Agent timeout, sending SIGTERM',
      );
      proc.kill('SIGTERM');
      setTimeout(() => {
        if (!proc.killed) proc.kill('SIGKILL');
      }, 15000);
    };

    let timeout = setTimeout(killOnTimeout, timeoutMs);

    const resetTimeout = () => {
      clearTimeout(timeout);
      timeout = setTimeout(killOnTimeout, timeoutMs);
    };

    stdoutStream.on('data', (data) => {
      const chunk = data.toString();

      if (!stdoutTruncated) {
        const remaining = AGENT_MAX_OUTPUT_SIZE - stdout.length;
        if (chunk.length > remaining) {
          stdout += chunk.slice(0, remaining);
          stdoutTruncated = true;
          logger.warn(
            {
              group: group.name,
              chatJid: input.chatJid,
              runId: input.runId,
              size: stdout.length,
            },
            'Agent stdout truncated due to size limit',
          );
        } else {
          stdout += chunk;
        }
      }

      if (!onOutput) {
        return;
      }

      parseBuffer += chunk;
      let startIdx: number;
      while ((startIdx = parseBuffer.indexOf(OUTPUT_START_MARKER)) !== -1) {
        const endIdx = parseBuffer.indexOf(OUTPUT_END_MARKER, startIdx);
        if (endIdx === -1) break;

        const jsonStr = parseBuffer
          .slice(startIdx + OUTPUT_START_MARKER.length, endIdx)
          .trim();
        parseBuffer = parseBuffer.slice(endIdx + OUTPUT_END_MARKER.length);

        try {
          const parsed: AgentOutput = JSON.parse(jsonStr);
          if (parsed.newSessionId) {
            newSessionId = parsed.newSessionId;
          }
          hadStreamingOutput = true;
          resetTimeout();
          logStreamedAgentErrorOutput(parsed, group, input);
          outputChain = chainStreamedOutputDelivery({
            outputChain,
            parsed,
            onOutput,
            onTerminalStreamedOutputFlushed:
              args.onTerminalStreamedOutputFlushed,
            group,
            input,
          });
        } catch (err) {
          logger.warn(
            {
              group: group.name,
              chatJid: input.chatJid,
              runId: input.runId,
              error: err,
            },
            'Failed to parse streamed output chunk',
          );
        }
      }
    });

    stderrStream.on('data', (data) => {
      const chunk = data.toString();
      const lines = chunk.trim().split('\n');
      for (const line of lines) {
        if (!line) continue;
        if (
          line.includes('Turn in progress') ||
          line.includes('Subagent') ||
          line.includes('Intermediate assistant') ||
          line.includes('Promoting') ||
          line.includes('Flushing') ||
          line.includes('Result #') ||
          line.includes('Query done') ||
          line.includes('Terminal') ||
          line.includes('Assistant: stop=') ||
          line.includes('Close sentinel')
        ) {
          logger.info(
            { group: group.name, chatJid: input.chatJid, runId: input.runId },
            line.replace(/^\[.*?\]\s*/, ''),
          );
        } else {
          logger.debug(
            { agent: group.folder, chatJid: input.chatJid, runId: input.runId },
            line,
          );
        }
      }
      resetTimeout();
      if (stderrTruncated) return;
      const remaining = AGENT_MAX_OUTPUT_SIZE - stderr.length;
      if (chunk.length > remaining) {
        stderr += chunk.slice(0, remaining);
        stderrTruncated = true;
      } else {
        stderr += chunk;
      }
    });

    proc.on('close', (code, signal) => {
      clearTimeout(timeout);
      const duration = Date.now() - startTime;

      if (timedOut) {
        writeTimeoutLog({
          logsDir,
          input,
          group,
          processName,
          duration,
          code,
          signal,
          hadStreamingOutput,
        });

        if (hadStreamingOutput) {
          logger.info(
            {
              group: group.name,
              chatJid: input.chatJid,
              runId: input.runId,
              processName,
              duration,
              code,
              signal,
            },
            'Agent timed out after output (idle cleanup)',
          );
          outputChain.then(() => {
            resolve({ status: 'success', result: null, newSessionId });
          });
          return;
        }

        resolve({
          status: 'error',
          result: null,
          error: `Agent timed out after ${configTimeout}ms`,
        });
        return;
      }

      if (code === null && signal) {
        if (hadStreamingOutput) {
          logger.info(
            {
              group: group.name,
              chatJid: input.chatJid,
              runId: input.runId,
              processName,
              duration,
              signal,
            },
            'Agent terminated by signal after output delivery (normal cleanup)',
          );
          outputChain.then(() => {
            resolve({ status: 'success', result: null, newSessionId });
          });
          return;
        }

        logger.error(
          {
            group: group.name,
            chatJid: input.chatJid,
            runId: input.runId,
            processName,
            duration,
            signal,
          },
          'Agent killed by signal before producing output',
        );
        outputChain.then(() => {
          resolve({
            status: 'error',
            result: null,
            error: `Agent killed by ${signal} before producing output`,
          });
        });
        return;
      }

      const timestamp = new Date().toISOString().replace(/[:.]/g, '-');
      const logFile = path.join(
        logsDir,
        `agent-${input.runId || 'adhoc'}-${timestamp}.log`,
      );
      const isVerbose = LOG_LEVEL === 'debug' || LOG_LEVEL === 'trace';
      const logLines = [
        `=== Agent Run Log ===`,
        `Timestamp: ${new Date().toISOString()}`,
        `Group: ${group.name}`,
        `ChatJid: ${input.chatJid}`,
        `GroupFolder: ${input.groupFolder}`,
        `RunId: ${input.runId || 'n/a'}`,
        `IsMain: ${input.isMain}`,
        `AgentType: ${group.agentType || 'claude-code'}`,
        `Duration: ${duration}ms`,
        `Exit Code: ${code}`,
        `Signal: ${signal}`,
        ``,
      ];

      const isError = code !== 0;
      if (isVerbose || isError) {
        logLines.push(
          `=== Input ===`,
          JSON.stringify(input, null, 2),
          ``,
          `=== Stderr ===`,
          stderr,
          ``,
          `=== Stdout ===`,
          stdout,
        );
      } else {
        logLines.push(
          `Prompt length: ${input.prompt.length} chars`,
          `Session ID: ${input.sessionId || 'new'}`,
        );
      }

      fs.writeFileSync(logFile, logLines.join('\n'));

      if (code !== 0) {
        logger.error(
          {
            group: group.name,
            chatJid: input.chatJid,
            runId: input.runId,
            code,
            duration,
            logFile,
          },
          'Agent exited with error',
        );
        outputChain.then(() => {
          resolve({
            status: 'error',
            result: null,
            error: `Agent exited with code ${code}: ${stderr.slice(-200)}`,
          });
        });
        return;
      }

      if (onOutput) {
        outputChain.then(() => {
          logger.info(
            {
              group: group.name,
              chatJid: input.chatJid,
              runId: input.runId,
              duration,
              newSessionId,
            },
            'Agent completed (streaming mode)',
          );
          resolve({ status: 'success', result: null, newSessionId });
        });
        return;
      }

      try {
        const output = parseLegacyAgentOutput(stdout);
        logger.info(
          {
            group: group.name,
            chatJid: input.chatJid,
            runId: input.runId,
            duration,
            status: output.status,
          },
          'Agent completed',
        );
        resolve(output);
      } catch (err) {
        logger.error(
          {
            group: group.name,
            chatJid: input.chatJid,
            runId: input.runId,
            error: err,
          },
          'Failed to parse agent output',
        );
        resolve({
          status: 'error',
          result: null,
          error: `Failed to parse agent output: ${getErrorMessage(err)}`,
        });
      }
    });

    proc.on('error', (err) => {
      clearTimeout(timeout);
      logger.error(
        {
          group: group.name,
          chatJid: input.chatJid,
          runId: input.runId,
          processName,
          error: err,
        },
        'Agent spawn error',
      );
      resolve({
        status: 'error',
        result: null,
        error: `Agent spawn error: ${err.message}`,
      });
    });
  });
}