23591589681

Committed 26 Mar 2026 11:20AM UTC coverage: 90.638% (-0.005%) from 90.643%

Build # 23591589681

Build Type

Pull #392

github

Committed by

web-flow

Commit Message

Merge 45759fefd into ac8a1088d

Pull Request Pull Request #392: refactor: reduce cognitive complexity - bot batch 3

Coverage Stats

6553 of 7667 branches covered (85.47%)

Branch coverage included in aggregate %.

78 of 84 new or added lines in 4 files covered. (92.86%)

6 existing lines in 3 files now uncovered.

11155 of 11870 relevant lines covered (93.98%)

224.11 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.37

/src/modules/cli-process.js

/**
 * CLIProcess — Claude CLI subprocess manager with dual-mode support.
 *
 * Spawns the `claude` binary directly in headless
 * mode.  Supports two lifecycle modes controlled by the `streaming` option:
 *
 * - **Short-lived** (default, `streaming: false`):  Each `send()` spawns a
 *   fresh `claude -p <prompt>` process that exits after returning its result.
 *   No token accumulation, clean abort via process kill.
 *
 * - **Long-lived** (`streaming: true`):  A single subprocess is kept alive
 *   across multiple `send()` calls using NDJSON stream-json I/O.  Tokens are
 *   tracked and the process is transparently recycled when a configurable
 *   threshold is exceeded.
 */

import { spawn } from 'node:child_process';
import { existsSync } from 'node:fs';
import { dirname, resolve } from 'node:path';
import { createInterface } from 'node:readline';
import { fileURLToPath } from 'node:url';
import { debug, info, error as logError, warn } from '../logger.js';
import { CLIProcessError } from '../utils/errors.js';

// Resolve the `claude` binary path from node_modules/.bin (may not be in PATH in Docker).
const __dirname = dirname(fileURLToPath(import.meta.url));
const LOCAL_BIN = resolve(__dirname, '..', '..', 'node_modules', '.bin', 'claude');
const CLAUDE_BIN = existsSync(LOCAL_BIN) ? LOCAL_BIN : 'claude';

export { CLIProcessError };

// ── AsyncQueue ───────────────────────────────────────────────────────────────

/**
 * Push-based async iterable for buffering stdin writes in long-lived mode.
 */
export class AsyncQueue {
  /** @type {Array<*>} */
  #queue = [];
  /** @type {Array<Function>} */
  #waiters = [];
  #closed = false;

  push(value) {
    if (this.#closed) return;
    if (this.#waiters.length > 0) {
      const resolve = this.#waiters.shift();
      resolve({ value, done: false });
    } else {
      this.#queue.push(value);
    }
  }

  close() {
    this.#closed = true;
    for (const resolve of this.#waiters) {
      resolve({ value: undefined, done: true });
    }
    this.#waiters.length = 0;
  }

  [Symbol.asyncIterator]() {
    return {
      next: () => {
        if (this.#queue.length > 0) {
          return Promise.resolve({ value: this.#queue.shift(), done: false });
        }
        if (this.#closed) {
          return Promise.resolve({ value: undefined, done: true });
        }
        return new Promise((resolve) => {
          this.#waiters.push(resolve);
        });
      },
    };
  }
}

// ── Helpers ──────────────────────────────────────────────────────────────────

const MAX_STDERR_LINES = 20;

/**
 * Build CLI argument array from a flags object.
 * @param {Object} flags
 * @param {boolean} longLived  Whether to include stream-json input flags.
 * @returns {string[]}
 */
function buildArgs(flags, longLived) {
  const args = ['-p'];

  // Always output NDJSON and enable verbose diagnostics
  args.push('--output-format', 'stream-json');
  args.push('--verbose');

  if (longLived) {
    args.push('--input-format', 'stream-json');
  }

  if (flags.model) {
    args.push('--model', flags.model);
  }

  if (flags.systemPromptFile) {
    args.push('--system-prompt-file', flags.systemPromptFile);
  }

  if (flags.systemPrompt) {
    args.push('--system-prompt', flags.systemPrompt);
  }

  if (flags.appendSystemPrompt) {
    args.push('--append-system-prompt', flags.appendSystemPrompt);
  }

  if (flags.tools !== undefined) {
    args.push('--tools', flags.tools);
  }

  if (flags.allowedTools) {
    const toolList = Array.isArray(flags.allowedTools) ? flags.allowedTools : [flags.allowedTools];
    for (const tool of toolList) {
      args.push('--allowedTools', tool);
    }
  }

  if (flags.permissionMode) {
    args.push('--permission-mode', flags.permissionMode);
  } else {
    args.push('--permission-mode', 'bypassPermissions');
  }

  // SAFETY: --dangerously-skip-permissions is required for non-interactive
  // (headless) use. Without it, the CLI blocks waiting for a TTY-based
  // permission prompt that can never be answered in a subprocess context.
  // This is safe here because the bot controls what prompts and tools are
  // passed — user input is never forwarded raw to the CLI. The bot's own
  // permission model (Discord permissions + config.json) gates access.
  args.push('--dangerously-skip-permissions');

  args.push('--no-session-persistence');

  if (flags.maxBudgetUsd != null) {
    args.push('--max-budget-usd', String(flags.maxBudgetUsd));
  }

  return args;
}

/**
 * Build the subprocess environment with thinking token configuration.
 * @param {Object} flags
 * @param {string} [flags.baseUrl]  Override ANTHROPIC_BASE_URL (e.g. for claude-code-router proxy)
 * @param {string} [flags.apiKey]   Override ANTHROPIC_API_KEY (e.g. for provider-specific key)
 * @returns {Object}
 */
function buildEnv(flags) {
  // Security: pass only what the Claude CLI subprocess actually needs.
  // Never spread process.env — that would leak DISCORD_TOKEN, DATABASE_URL,
  // BOT_API_SECRET, SESSION_SECRET, REDIS_URL, etc. to the child process.
  // See: https://github.com/VolvoxLLC/volvox-bot/issues/155
  const env = {
    PATH: process.env.PATH,
    HOME: process.env.HOME,
    ...(process.env.NODE_ENV && { NODE_ENV: process.env.NODE_ENV }),
    ...(process.env.DISABLE_PROMPT_CACHING && {
      DISABLE_PROMPT_CACHING: process.env.DISABLE_PROMPT_CACHING,
    }),
    MAX_THINKING_TOKENS: String(flags.thinkingTokens ?? 4096),
  };

  // Auth priority: explicit apiKey flag > ANTHROPIC_API_KEY env > CLAUDE_CODE_OAUTH_TOKEN env.
  // When flags.apiKey is provided we intentionally omit CLAUDE_CODE_OAUTH_TOKEN
  // to avoid conflicting auth headers in the subprocess.
  if (flags.apiKey) {
    env.ANTHROPIC_API_KEY = flags.apiKey;
  } else if (process.env.ANTHROPIC_API_KEY) {
    env.ANTHROPIC_API_KEY = process.env.ANTHROPIC_API_KEY;
  } else if (process.env.CLAUDE_CODE_OAUTH_TOKEN) {
    env.CLAUDE_CODE_OAUTH_TOKEN = process.env.CLAUDE_CODE_OAUTH_TOKEN;
  }

  // baseUrl is set from admin config (triage.classifyBaseUrl / respondBaseUrl),
  // never from user input. Validate URL format as defense-in-depth.
  if (flags.baseUrl) {
    try {
      const parsed = new URL(flags.baseUrl);
      if (parsed.protocol === 'https:' || parsed.protocol === 'http:') {
        env.ANTHROPIC_BASE_URL = flags.baseUrl;
      }
    } catch {
      warn('Ignoring malformed baseUrl — falling back to default Anthropic endpoint', {
        baseUrl: flags.baseUrl,
      });
    }
  }

  return env;
}

// ── CLIProcess ───────────────────────────────────────────────────────────────

export class CLIProcess {
  #name;
  #flags;
  #streaming;
  #tokenLimit;
  #timeout;

  // Long-lived state
  #proc = null;
  #sessionId = null;
  #alive = false;
  #accumulatedTokens = 0;
  #stderrBuffer = [];

  // Long-lived consume-loop bookkeeping
  #pendingResolve = null;
  #pendingReject = null;
  /** @type {string[]} Accumulated text blocks for the current long-lived turn */
  #longLivedTextParts = [];

  // Short-lived: reference to the in-flight process for abort
  #inflightProc = null;

  // Mutex state — serialises concurrent send() calls.
  #mutexPromise = Promise.resolve();

  /**
   * @param {string} name  Human-readable label ('classifier' | 'responder' | 'ai-chat')
   * @param {Object} flags  CLI flag configuration
   * @param {string} [flags.model]  Model name (e.g. 'claude-sonnet-4-6')
   * @param {string} [flags.systemPromptFile]  Path to system prompt .md file
   * @param {string} [flags.systemPrompt]  System prompt as a string
   * @param {string} [flags.appendSystemPrompt]  Text appended to system prompt
   * @param {string} [flags.tools]  Tools flag ('' to disable all)
   * @param {string|string[]} [flags.allowedTools]  Allowed tool names
   * @param {string} [flags.permissionMode]  Permission mode (default: 'bypassPermissions')
   * @param {number} [flags.maxBudgetUsd]  Budget cap per process lifetime
   * @param {number} [flags.thinkingTokens]  MAX_THINKING_TOKENS env (default: 4096)
   * @param {string} [flags.baseUrl]  Override ANTHROPIC_BASE_URL (e.g. 'http://router:3456' for CCR proxy)
   * @param {string} [flags.apiKey]  Override ANTHROPIC_API_KEY (e.g. provider-specific key for routed requests)
   * @param {Object} [meta]
   * @param {number} [meta.tokenLimit=20000]  Token threshold before auto-recycle (long-lived only)
   * @param {boolean} [meta.streaming=false]  true for long-lived mode
   * @param {number} [meta.timeout=120000]  Per-send timeout in milliseconds
   */
  constructor(name, flags = {}, { tokenLimit = 20000, streaming = false, timeout = 120_000 } = {}) {
    this.#name = name;
    this.#flags = flags;
    this.#streaming = streaming;
    this.#tokenLimit = tokenLimit;
    this.#timeout = timeout;
  }

  // ── Lifecycle ────────────────────────────────────────────────────────────

  async start() {
    if (this.#streaming) {
      await this.#startLongLived();
    } else {
      this.#alive = true;
      this.#accumulatedTokens = 0;
    }
  }

  async #startLongLived() {
    this.#accumulatedTokens = 0;
    this.#stderrBuffer = [];
    this.#sessionId = null;

    const args = buildArgs(this.#flags, true);
    const env = buildEnv(this.#flags);

    this.#proc = spawn(CLAUDE_BIN, args, {
      stdio: ['pipe', 'pipe', 'pipe'],
      env,
    });

    // EPIPE protection: if the child dies between the alive check and stdin.write,
    // catch the error instead of crashing the host process.
    this.#proc.stdin.on('error', (err) => {
      warn(`${this.#name}: stdin error (child may have exited)`, { error: err.message });
      this.#alive = false;
    });

    // Capture stderr for diagnostics
    this.#proc.stderr.on('data', (chunk) => {
      const lines = chunk.toString().split('\n').filter(Boolean);
      this.#stderrBuffer.push(...lines);
      if (this.#stderrBuffer.length > MAX_STDERR_LINES) {
        this.#stderrBuffer = this.#stderrBuffer.slice(-MAX_STDERR_LINES);
      }
    });

    // Handle unexpected exit
    this.#proc.on('exit', (code, signal) => {
      if (this.#alive) {
        warn(`${this.#name}: long-lived process exited`, { code, signal });
        this.#alive = false;
        if (this.#pendingReject) {
          this.#pendingReject(
            new CLIProcessError(
              `${this.#name}: process exited unexpectedly (code=${code}, signal=${signal})`,
              'exit',
              { code, signal },
            ),
          );
          this.#pendingReject = null;
          this.#pendingResolve = null;
        }
      }
    });

    // Start the background consume loop
    this.#runConsumeLoop();
    this.#alive = true;
    info(`${this.#name}: long-lived process started`, { pid: this.#proc.pid });
  }

  #runConsumeLoop() {
    const rl = createInterface({ input: this.#proc.stdout, crlfDelay: Infinity });

    rl.on('line', (line) => {
      if (!line.trim()) return;
      let msg;
      try {
        msg = JSON.parse(line);
      } catch {
        warn(`${this.#name}: non-JSON stdout line`, { line: line.slice(0, 200) });
        return;
      }

      // Capture session_id from init message
      if (msg.type === 'system' && msg.subtype === 'init') {
        this.#sessionId = msg.session_id;
        return;
      }

      // Accumulate text from assistant messages (long-lived mode)
      if (msg.type === 'assistant' && msg.message?.content) {
        for (const block of msg.message.content) {
          if (block.type === 'text' && block.text) {
            this.#longLivedTextParts.push(block.text);
          }
        }
      }

      if (msg.type === 'result') {
        // Reconstruct result text from accumulated assistant blocks
        if (msg.result === undefined && this.#longLivedTextParts.length > 0) {
          msg.result = this.#longLivedTextParts.join('');
        }
        this.#longLivedTextParts = [];
        this.#trackTokens(msg);
        this.#pendingResolve?.(msg);
        this.#pendingResolve = null;
        this.#pendingReject = null;
      }
    });

    rl.on('close', () => {
      if (this.#alive) {
        this.#alive = false;
        this.#pendingReject?.(
          new CLIProcessError(`${this.#name}: stdout closed unexpectedly`, 'exit'),
        );
        this.#pendingReject = null;
        this.#pendingResolve = null;
      }
    });
  }

  // ── send() ───────────────────────────────────────────────────────────────

  /**
   * Send a prompt and await the result.
   * Concurrent calls are serialised via an internal mutex.
   *
   * @param {string} prompt  The user-turn prompt text.
   * @param {Object} [overrides]  Per-call flag overrides (short-lived mode only).
   * @param {string} [overrides.systemPrompt]  Override system prompt string.
   * @param {string} [overrides.appendSystemPrompt]  Override append-system-prompt.
   * @param {string} [overrides.systemPromptFile]  Override system prompt file path.
   * @param {Object} [options]  Additional options.
   * @param {Function} [options.onEvent]  Callback for intermediate NDJSON messages (short-lived only).
   * @returns {Promise<Object>} The result message from the CLI.
   */
  async send(prompt, overrides = {}, { onEvent } = {}) {
    const release = await this.#acquireMutex();
    try {
      const result = this.#streaming
        ? await this.#sendLongLived(prompt)
        : await this.#sendShortLived(prompt, overrides, onEvent);

      // Token recycling — non-blocking so the caller gets the result now.
      if (this.#streaming && this.#accumulatedTokens >= this.#tokenLimit) {
        info(`Recycling ${this.#name} process`, {
          accumulatedTokens: this.#accumulatedTokens,
          tokenLimit: this.#tokenLimit,
        });
        this.recycle().catch((err) =>
          logError(`Failed to recycle ${this.#name}`, { error: err.message }),
        );
      }

      return result;
    } finally {
      release();
    }
  }

  /**
   * Parse a single stdout line from the CLI process, accumulating text parts
   * and detecting the result message.
   */
  #parseStdoutLine(line, textParts, onEvent) {
    if (!line.trim()) return null;
    let msg;
    try {
      msg = JSON.parse(line);
    } catch {
      debug(`${this.#name}: non-JSON stdout line (short-lived)`, { line: line.slice(0, 200) });
      return null;
    }
    if (msg.type === 'result') {
      if (msg.result === undefined && textParts.length > 0) {
        msg.result = textParts.join('');
      }
      return msg;
    }
    // Accumulate text from assistant messages (claude-code >=2.1.77)
    if (msg.type === 'assistant' && msg.message?.content) {
      for (const block of msg.message.content) {
        if (block.type === 'text' && block.text) {
          textParts.push(block.text);
        }
      }
    }
    if (onEvent) {
      onEvent(msg);
    }
    return null;
  }

  async #sendShortLived(prompt, overrides = {}, onEvent = null) {
    const mergedFlags = { ...this.#flags, ...overrides };
    const args = buildArgs(mergedFlags, false);

    // In short-lived mode, the prompt is a positional argument after -p
    args.push(prompt);

    const env = buildEnv(mergedFlags);
    const stderrLines = [];

    return new Promise((resolve, reject) => {
      const proc = spawn(CLAUDE_BIN, args, {
        stdio: ['ignore', 'pipe', 'pipe'],
        env,
      });

      this.#inflightProc = proc;

      // Timeout handling
      const timer = setTimeout(() => {
        proc.kill('SIGKILL');
        reject(
          new CLIProcessError(
            `${this.#name}: send() timed out after ${this.#timeout}ms`,
            'timeout',
          ),
        );
      }, this.#timeout);

      let result = null;
      const textParts = [];

      // Capture stderr
      proc.stderr.on('data', (chunk) => {
        const lines = chunk.toString().split('\n').filter(Boolean);
        stderrLines.push(...lines);
        if (stderrLines.length > MAX_STDERR_LINES) {
          stderrLines.splice(0, stderrLines.length - MAX_STDERR_LINES);
        }
      });

      const rl = createInterface({ input: proc.stdout, crlfDelay: Infinity });

      rl.on('line', (line) => {
        const parsed = this.#parseStdoutLine(line, textParts, onEvent);
        if (parsed) result = parsed;
      });

      proc.on('exit', (code, signal) => {
        clearTimeout(timer);
        this.#inflightProc = null;

        if (result) {
          try {
            resolve(this.#extractResult(result));
          } catch (err) {
            reject(err);
          }
        } else {
          const stderr = stderrLines.join('\n');
          reject(
            new CLIProcessError(
              `${this.#name}: process exited without result (code=${code}, signal=${signal})${stderr ? `\nstderr: ${stderr}` : ''}`,
              'exit',
              { code, signal },
            ),
          );
        }
      });

      proc.on('error', (err) => {
        clearTimeout(timer);
        this.#inflightProc = null;
        reject(
          new CLIProcessError(`${this.#name}: failed to spawn process — ${err.message}`, 'exit'),
        );
      });
    });
  }

  async #sendLongLived(prompt) {
    if (!this.#alive) {
      throw new CLIProcessError(`${this.#name}: process is not alive`, 'exit');
    }

    // Reset text accumulator for new turn
    this.#longLivedTextParts = [];

    return new Promise((resolve, reject) => {
      this.#pendingResolve = (msg) => {
        clearTimeout(timer);
        try {
          resolve(this.#extractResult(msg));
        } catch (err) {
          reject(err);
        }
      };
      this.#pendingReject = (err) => {
        clearTimeout(timer);
        reject(err);
      };

      // Timeout handling
      const timer = setTimeout(() => {
        this.#pendingResolve = null;
        this.#pendingReject = null;
        // Kill and restart the long-lived process
        this.#proc?.kill('SIGKILL');
        reject(
          new CLIProcessError(
            `${this.#name}: send() timed out after ${this.#timeout}ms`,
            'timeout',
          ),
        );
      }, this.#timeout);

      // Write NDJSON user-turn message to stdin
      const message = JSON.stringify({
        type: 'user',
        message: { role: 'user', content: prompt },
        session_id: this.#sessionId ?? '',
        parent_tool_use_id: null,
      });

      this.#proc.stdin.write(`${message}\n`);
    });
  }

  // ── Result extraction ────────────────────────────────────────────────────

  #extractResult(message) {
    if (message.is_error) {
      const errMsg = message.errors?.map((e) => e.message || e).join('; ') || 'Unknown CLI error';
      logError(`${this.#name}: CLI error`, {
        error: errMsg,
        errorCount: message.errors?.length ?? 0,
        resultSnippet: JSON.stringify(message).slice(0, 500),
      });
      throw new CLIProcessError(`${this.#name}: CLI error — ${errMsg}`, 'exit');
    }
    return message;
  }

  #trackTokens(message) {
    const usage = message.usage;
    if (usage) {
      const inp = usage.inputTokens ?? usage.input_tokens ?? 0;
      const out = usage.outputTokens ?? usage.output_tokens ?? 0;
      this.#accumulatedTokens += inp + out;
    }
  }

  // ── Recycle / restart ────────────────────────────────────────────────────

  async recycle() {
    this.close();
    await this.start();
  }

  async restart(attempt = 0) {
    const baseDelay = Math.min(1000 * 2 ** attempt, 30_000);
    const jitter = Math.floor(Math.random() * 1000);
    const delay = baseDelay + jitter;
    warn(`Restarting ${this.#name} process`, { attempt, delayMs: delay });
    await new Promise((r) => setTimeout(r, delay));
    try {
      await this.recycle();
    } catch (err) {
      logError(`${this.#name} restart failed`, { error: err.message, attempt });
      if (attempt < 5) {
        await this.restart(attempt + 1);
      } else {
        throw err;
      }
    }
  }

  close() {
    this.#killProc(this.#proc);
    this.#proc = null;

    this.#killProc(this.#inflightProc);
    this.#inflightProc = null;

    this.#alive = false;
    this.#sessionId = null;

    if (this.#pendingReject) {
      this.#pendingReject(new CLIProcessError(`${this.#name}: process closed`, 'killed'));
      this.#pendingReject = null;
      this.#pendingResolve = null;
    }
  }

  /**
   * Send SIGTERM to a child process, then escalate to SIGKILL after 2 seconds
   * if it hasn't exited. Prevents zombie processes from stuck CLI subprocesses.
   * @param {import('node:child_process').ChildProcess|null} proc
   */
  #killProc(proc) {
    if (!proc) return;
    try {
      proc.kill('SIGTERM');
    } catch {
      return; // Already exited
    }
    const sigkillTimer = setTimeout(() => {
      try {
        proc.kill('SIGKILL');
      } catch {
        // Already exited between SIGTERM and SIGKILL — expected
      }
    }, 2000);
    // Don't keep the event loop alive just for the SIGKILL escalation
    sigkillTimer.unref();
  }

  // ── Mutex ────────────────────────────────────────────────────────────────

  #acquireMutex() {
    let release;
    const next = new Promise((resolve) => {
      release = resolve;
    });
    const prev = this.#mutexPromise;
    this.#mutexPromise = prev.then(() => next);
    return prev.then(() => release);
  }

  // ── Accessors ────────────────────────────────────────────────────────────

  get alive() {
    return this.#alive;
  }

  get tokenCount() {
    return this.#accumulatedTokens;
  }

  get name() {
    return this.#name;
  }

  get stderrDiagnostics() {
    return this.#stderrBuffer.join('\n');
  }
}

VolvoxLLC / volvox-bot / 23591589681

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous