memind/tool-gateway.mjs

import { spawn as nodeSpawn } from 'node:child_process';
import { EventEmitter } from 'node:events';

const CODE_EXECUTORS = new Set(['aider', 'openhands']);
const DEFAULT_STDIO_LIMIT = 64 * 1024;

function envFlag(value, fallback = false) {
  const raw = String(value ?? '').trim().toLowerCase();
  if (!raw) return fallback;
  return ['1', 'true', 'yes', 'on'].includes(raw);
}

function positiveInteger(value, fallback) {
  const n = Number(value);
  if (!Number.isFinite(n) || n <= 0) return fallback;
  return Math.floor(n);
}

function normalizeExecutor(value, fallback = 'aider') {
  const normalized = String(value ?? fallback).trim().toLowerCase();
  if (CODE_EXECUTORS.has(normalized)) return normalized;
  return fallback;
}

function csvSet(value) {
  return new Set(
    String(value ?? '')
      .split(',')
      .map((item) => item.trim().toLowerCase())
      .filter(Boolean),
  );
}

function appendLimited(current, chunk, limit) {
  const next = `${current}${Buffer.isBuffer(chunk) ? chunk.toString('utf8') : String(chunk ?? '')}`;
  if (next.length <= limit) return next;
  return next.slice(next.length - limit);
}

export function extractToolInstruction(userMessage) {
  const content = userMessage?.content;
  if (typeof content === 'string') return content.trim();
  if (!Array.isArray(content)) {
    return String(userMessage?.text ?? userMessage?.value ?? '').trim();
  }
  return content
    .map((item) => {
      if (typeof item === 'string') return item;
      if (item?.type === 'text') return item.text ?? '';
      return '';
    })
    .map((item) => String(item ?? '').trim())
    .filter(Boolean)
    .join('\n')
    .trim();
}

export function createToolGateway({
  llmProviderService,
  env = process.env,
  spawnImpl = nodeSpawn,
} = {}) {
  const enabled = envFlag(env.MEMIND_TOOL_GATEWAY_ENABLED, false);
  const dryRun = envFlag(env.MEMIND_TOOL_GATEWAY_DRY_RUN, false);
  const defaultExecutor = normalizeExecutor(env.MEMIND_TOOL_GATEWAY_DEFAULT_EXECUTOR, 'aider');
  const openhandsTaskTypes = csvSet(
    env.MEMIND_TOOL_GATEWAY_OPENHANDS_TASK_TYPES ?? 'repo_refactor,multi_file,complex_repo',
  );
  const stdioLimit = positiveInteger(env.MEMIND_TOOL_GATEWAY_STDIO_LIMIT, DEFAULT_STDIO_LIMIT);

  function getStatus() {
    return {
      enabled,
      dryRun,
      protocol: 'agent-run-v1',
      executors: ['aider', 'openhands'],
      defaultExecutor,
      openhandsTaskTypes: [...openhandsTaskTypes],
    };
  }

  function selectExecutor({ userMessage, taskType } = {}) {
    const metadata = userMessage?.metadata ?? {};
    const runMetadata = metadata.memindRun ?? metadata.agentRun ?? {};
    const requested = normalizeExecutor(runMetadata.executor, '');
    if (requested) return requested;
    const normalizedTaskType = String(taskType ?? runMetadata.taskType ?? '').trim().toLowerCase();
    if (openhandsTaskTypes.has(normalizedTaskType)) return 'openhands';
    return defaultExecutor;
  }

  async function executeJob({
    runId,
    requestId,
    userId,
    userMessage,
    taskType,
    cwd,
    timeoutMs = 15 * 60 * 1000,
  } = {}) {
    if (!enabled) {
      throw new Error('Tool Gateway is disabled');
    }
    if (!llmProviderService?.getExecutorLaunchPlan) {
      throw new Error('Tool Gateway missing llm provider service');
    }
    const instruction = extractToolInstruction(userMessage);
    if (!instruction) {
      throw new Error('Tool Gateway job missing instruction');
    }
    const executor = selectExecutor({ userMessage, taskType });
    const plan = await llmProviderService.getExecutorLaunchPlan(executor, {
      cwd,
      mode: 'headless',
      instruction,
      purpose: 'default',
      includeSecret: true,
    });
    if (!plan?.ok) {
      throw new Error(plan?.message ?? `Tool Gateway launch plan unavailable for ${executor}`);
    }
    if (dryRun) {
      return {
        ok: true,
        dryRun: true,
        executor,
        protocol: 'agent-run-v1',
        cwd: plan.cwd,
        command: plan.command,
        args: plan.args ?? [],
        runId,
        requestId,
        userId,
      };
    }

    return await new Promise((resolve, reject) => {
      let stdout = '';
      let stderr = '';
      const child = spawnImpl(plan.command, plan.args ?? [], {
        cwd: plan.cwd,
        env: { ...process.env, ...(plan.env ?? {}) },
        stdio: ['ignore', 'pipe', 'pipe'],
      });
      const cleanup = () => {
        if (timer) clearTimeout(timer);
      };
      const timer = setTimeout(() => {
        child.kill('SIGTERM');
        const err = new Error(`Tool Gateway job timed out after ${timeoutMs}ms`);
        err.code = 'TOOL_GATEWAY_TIMEOUT';
        reject(err);
      }, timeoutMs);

      if (child.stdout instanceof EventEmitter) {
        child.stdout.on('data', (chunk) => {
          stdout = appendLimited(stdout, chunk, stdioLimit);
        });
      }
      if (child.stderr instanceof EventEmitter) {
        child.stderr.on('data', (chunk) => {
          stderr = appendLimited(stderr, chunk, stdioLimit);
        });
      }
      child.on('error', (err) => {
        cleanup();
        reject(err);
      });
      child.on('exit', (code, signal) => {
        cleanup();
        if (code === 0) {
          resolve({
            ok: true,
            executor,
            protocol: 'agent-run-v1',
            cwd: plan.cwd,
            command: plan.command,
            args: plan.args ?? [],
            exitCode: code,
            signal: signal ?? null,
            stdout,
            stderr,
          });
          return;
        }
        const err = new Error(`Tool Gateway executor ${executor} exited with ${signal ?? code}`);
        err.code = 'TOOL_GATEWAY_EXECUTOR_FAILED';
        err.executor = executor;
        err.exitCode = code;
        err.signal = signal ?? null;
        err.stdout = stdout;
        err.stderr = stderr;
        reject(err);
      });
    });
  }

  return {
    getStatus,
    selectExecutor,
    executeJob,
  };
}