memind/services/orchestrator/executor-adapters.mjs

import { spawn as nodeSpawn } from 'node:child_process';
import fs from 'node:fs/promises';
import path from 'node:path';
import { EventEmitter } from 'node:events';
import { Agent, fetch as undiciFetch } from 'undici';

const DEFAULT_OUTPUT_LIMIT = 64 * 1024;
const DEFAULT_ALLOWED_ENV = Object.freeze(['PATH', 'LANG', 'LC_ALL', 'TMPDIR']);

function adapterError(code, message, { retryable = false, cause = null } = {}) {
  const error = new Error(message, cause ? { cause } : undefined);
  error.code = code;
  error.retryable = retryable;
  return error;
}

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

function isLoopbackHostname(hostname) {
  const normalized = String(hostname ?? '').trim().toLowerCase();
  return normalized === 'localhost'
    || normalized === '127.0.0.1'
    || normalized === '::1'
    || normalized === '[::1]'
    || normalized.endsWith('.localhost');
}

function isLocalExecutorHostname(hostname) {
  const normalized = String(hostname ?? '').trim().toLowerCase();
  return isLoopbackHostname(normalized)
    || normalized === 'host.docker.internal'
    || normalized === 'gateway.docker.internal';
}

function normalizeBaseUrl(value, { allowRemote = false } = {}) {
  const url = new URL(String(value ?? '').trim());
  if (!['http:', 'https:'].includes(url.protocol)) {
    throw new Error('Executor adapter URL must use HTTP or HTTPS');
  }
  if (!allowRemote && !isLoopbackHostname(url.hostname)) {
    throw new Error('Remote executor targets require explicit allowRemote');
  }
  url.username = '';
  url.password = '';
  url.hash = '';
  return url.toString().replace(/\/$/, '');
}

async function responseError(response, code, prefix) {
  const text = await response.text().catch(() => '');
  return adapterError(
    code,
    `${prefix} (${response.status})${text ? `: ${text.slice(0, 512)}` : ''}`,
    { retryable: response.status >= 500 || response.status === 429 },
  );
}

async function* parseSse(body) {
  const decoder = new TextDecoder();
  let buffer = '';
  for await (const chunk of body) {
    buffer += decoder.decode(chunk, { stream: true });
    while (true) {
      const boundary = buffer.search(/\r?\n\r?\n/);
      if (boundary < 0) break;
      const raw = buffer.slice(0, boundary);
      const separatorLength = buffer.startsWith('\r\n\r\n', boundary) ? 4 : 2;
      buffer = buffer.slice(boundary + separatorLength);
      let event = 'message';
      const data = [];
      for (const line of raw.split(/\r?\n/)) {
        if (line.startsWith('event:')) event = line.slice(6).trim() || 'message';
        if (line.startsWith('data:')) data.push(line.slice(5).trimStart());
      }
      if (data.length) {
        const text = data.join('\n');
        let parsed = text;
        try {
          parsed = JSON.parse(text);
        } catch {
          // Preserve non-JSON Goosed events as bounded text.
        }
        yield { event, data: parsed };
      }
    }
  }
}

function terminalGoosedEvent(event) {
  const type = String(
    event?.event
      ?? event?.data?.type
      ?? event?.data?.event_type
      ?? '',
  ).trim().toLowerCase();
  return ['finish', 'finished', 'completed', 'error', 'failed'].includes(type);
}

export function createGoosedExecutorAdapter({
  baseUrl,
  secret = '',
  fetchImpl = null,
  allowRemote = false,
  tlsInsecure = false,
  workspaceResolver = async (reference) => reference?.id ?? null,
} = {}) {
  const target = normalizeBaseUrl(baseUrl, { allowRemote });
  const targetUrl = new URL(target);
  if (tlsInsecure && !isLocalExecutorHostname(targetUrl.hostname)) {
    throw new Error('Insecure executor TLS is restricted to local targets');
  }
  const transport = fetchImpl ?? undiciFetch;
  if (typeof transport !== 'function') throw new Error('Goosed adapter requires fetch');
  const dispatcher = tlsInsecure
    ? new Agent({ connect: { rejectUnauthorized: false } })
    : null;
  const active = new Map();

  function headers(extra = {}) {
    return {
      accept: 'application/json',
      'content-type': 'application/json',
      ...(secret ? { 'X-Secret-Key': secret } : {}),
      ...extra,
    };
  }

  async function submit(request, {
    jobId,
    signal = null,
    emit = async () => {},
  } = {}) {
    const controller = new AbortController();
    const abort = () => controller.abort(signal?.reason);
    if (signal?.aborted) abort();
    else signal?.addEventListener?.('abort', abort, { once: true });
    active.set(jobId, controller);
    try {
      const workingDir = await workspaceResolver(request.task.workspaceRef);
      if (!workingDir) {
        throw adapterError(
          'GOOSED_WORKSPACE_UNRESOLVED',
          'Goosed workspace reference could not be resolved',
        );
      }
      const start = await transport(`${target}/agent/start`, {
        method: 'POST',
        headers: headers(),
        body: JSON.stringify({
          working_dir: workingDir,
          enable_context_memory: false,
        }),
        signal: controller.signal,
        ...(dispatcher ? { dispatcher } : {}),
      });
      if (!start.ok) throw await responseError(start, 'GOOSED_START_FAILED', 'Goosed start failed');
      const session = await start.json();
      const sessionId = String(session?.id ?? '').trim();
      if (!sessionId) {
        throw adapterError('GOOSED_SESSION_ID_MISSING', 'Goosed start response missing session id');
      }
      await emit('executor_job_session_created', {
        sessionRef: { kind: 'goosed-session', id: sessionId },
      });

      const eventsResponse = await transport(
        `${target}/sessions/${encodeURIComponent(sessionId)}/events`,
        {
          method: 'GET',
          headers: headers({ accept: 'text/event-stream' }),
          signal: controller.signal,
          ...(dispatcher ? { dispatcher } : {}),
        },
      );
      if (!eventsResponse.ok || !eventsResponse.body) {
        throw await responseError(
          eventsResponse,
          'GOOSED_EVENTS_FAILED',
          'Goosed event stream failed',
        );
      }

      const finishPromise = (async () => {
        let eventCount = 0;
        let finalEvent = null;
        for await (const event of parseSse(eventsResponse.body)) {
          eventCount += 1;
          const eventType = String(
            event?.event ?? event?.data?.type ?? event?.data?.event_type ?? 'message',
          ).slice(0, 96);
          await emit('executor_job_adapter_event', {
            adapter: 'goosed',
            eventType,
            eventCount,
          });
          if (terminalGoosedEvent(event)) {
            finalEvent = event;
            break;
          }
        }
        return { eventCount, finalEvent };
      })();

      const reply = await transport(
        `${target}/sessions/${encodeURIComponent(sessionId)}/reply`,
        {
          method: 'POST',
          headers: headers(),
          body: JSON.stringify({
            request_id: jobId,
            user_message: {
              role: 'user',
              content: [{ type: 'text', text: request.task.instruction }],
            },
          }),
          signal: controller.signal,
          ...(dispatcher ? { dispatcher } : {}),
        },
      );
      if (!reply.ok) throw await responseError(reply, 'GOOSED_REPLY_FAILED', 'Goosed reply failed');
      reply.body?.cancel?.().catch?.(() => {});

      const finished = await finishPromise;
      const finalType = String(
        finished.finalEvent?.event
          ?? finished.finalEvent?.data?.type
          ?? finished.finalEvent?.data?.event_type
          ?? '',
      ).toLowerCase();
      if (['error', 'failed'].includes(finalType)) {
        throw adapterError('GOOSED_RUN_FAILED', 'Goosed reported a failed terminal event');
      }
      return {
        outcome: 'completed',
        summary: `Goosed session completed after ${finished.eventCount} events`,
        artifactRefs: [{ kind: 'goosed-session', id: sessionId }],
        metrics: { eventCount: finished.eventCount },
      };
    } catch (error) {
      if (controller.signal.aborted || error?.name === 'AbortError') {
        throw adapterError('GOOSED_CANCELLED', 'Goosed execution was cancelled');
      }
      throw error;
    } finally {
      signal?.removeEventListener?.('abort', abort);
      active.delete(jobId);
    }
  }

  return {
    id: 'goosed',
    label: 'Goosed',
    kind: 'remote',
    enabled: true,
    dispatchImplemented: true,
    capabilities: ['session-execution', 'streaming', 'cancel'],
    submit,
    async cancel(jobId) {
      const controller = active.get(jobId);
      if (!controller) return false;
      controller.abort();
      return true;
    },
    async getState(jobId) {
      return { active: active.has(jobId) };
    },
    async health() {
      try {
        const response = await transport(`${target}/status`, {
          method: 'GET',
          headers: headers(),
          signal: AbortSignal.timeout(2_000),
          ...(dispatcher ? { dispatcher } : {}),
        });
        return { ok: response.ok, status: response.status };
      } catch (error) {
        return {
          ok: false,
          status: null,
          code: String(error?.code ?? error?.name ?? 'GOOSED_UNREACHABLE').slice(0, 128),
        };
      }
    },
    async *streamEvents() {
      throw adapterError(
        'GOOSED_STREAM_PUSH_ONLY',
        'Goosed events are delivered through submit emit callbacks',
      );
    },
  };
}

async function resolveWorkspace(reference, {
  allowedRoot,
  aliases = {},
} = {}) {
  const alias = String(reference?.id ?? '').trim();
  const configured = aliases[alias];
  if (!configured) {
    throw adapterError(
      'EXECUTOR_WORKSPACE_ALIAS_UNKNOWN',
      `Workspace alias is not configured: ${alias || '<empty>'}`,
    );
  }
  const root = await fs.realpath(allowedRoot);
  const resolved = await fs.realpath(path.resolve(configured));
  if (resolved !== root && !resolved.startsWith(`${root}${path.sep}`)) {
    throw adapterError(
      'EXECUTOR_WORKSPACE_OUTSIDE_ROOT',
      'Workspace resolves outside the configured worker root',
    );
  }
  return resolved;
}

function safeCommand(value) {
  const command = String(value ?? '').trim();
  if (!command || /[\r\n\0]/.test(command)) {
    throw new Error('Process executor requires a safe command');
  }
  return command;
}

export function createProcessExecutorAdapter({
  id,
  label = id,
  command,
  buildArgs,
  allowedRoot,
  workspaceAliases = {},
  spawnImpl = nodeSpawn,
  outputLimit = DEFAULT_OUTPUT_LIMIT,
  allowedEnv = DEFAULT_ALLOWED_ENV,
  extraEnv = {},
  artifactCollector = async ({ request }) => [request.task.workspaceRef],
} = {}) {
  const executorId = String(id ?? '').trim().toLowerCase();
  if (!['aider', 'openhands'].includes(executorId)) {
    throw new Error('Process executor id must be aider or openhands');
  }
  const executable = safeCommand(command);
  if (!allowedRoot) throw new Error('Process executor requires allowedRoot');
  if (typeof buildArgs !== 'function') throw new Error('Process executor requires buildArgs');
  const active = new Map();

  async function submit(request, {
    jobId,
    signal = null,
    emit = async () => {},
  } = {}) {
    const cwd = await resolveWorkspace(request.task.workspaceRef, {
      allowedRoot,
      aliases: workspaceAliases,
    });
    const args = buildArgs(request.task.instruction, request);
    if (!Array.isArray(args) || args.some((arg) => typeof arg !== 'string')) {
      throw new Error(`${executorId} buildArgs must return a string array`);
    }
    const env = {};
    for (const key of allowedEnv) {
      if (process.env[key] != null) env[key] = process.env[key];
    }
    Object.assign(env, extraEnv);

    return new Promise((resolve, reject) => {
      let stdout = '';
      let stderr = '';
      let settled = false;
      const child = spawnImpl(executable, args, {
        cwd,
        env,
        shell: false,
        detached: false,
        stdio: ['ignore', 'pipe', 'pipe'],
      });
      active.set(jobId, child);
      const cleanup = () => {
        active.delete(jobId);
        signal?.removeEventListener?.('abort', onAbort);
      };
      const fail = (error) => {
        if (settled) return;
        settled = true;
        cleanup();
        reject(error);
      };
      const onAbort = () => {
        child.kill('SIGTERM');
        fail(adapterError('EXECUTOR_PROCESS_CANCELLED', `${label} execution was cancelled`));
      };
      if (signal?.aborted) onAbort();
      else signal?.addEventListener?.('abort', onAbort, { once: true });
      child.stdout?.on?.('data', (chunk) => {
        stdout = appendBounded(stdout, chunk, outputLimit);
        void emit('executor_job_output_chunk', {
          adapter: executorId,
          stream: 'stdout',
          bytes: Buffer.byteLength(chunk),
        });
      });
      child.stderr?.on?.('data', (chunk) => {
        stderr = appendBounded(stderr, chunk, outputLimit);
        void emit('executor_job_output_chunk', {
          adapter: executorId,
          stream: 'stderr',
          bytes: Buffer.byteLength(chunk),
        });
      });
      child.once?.('error', (error) => {
        fail(adapterError(
          'EXECUTOR_PROCESS_START_FAILED',
          `${label} failed to start: ${error.message}`,
          { retryable: true, cause: error },
        ));
      });
      child.once?.('exit', async (code, exitSignal) => {
        if (settled) return;
        if (code !== 0) {
          fail(adapterError(
            'EXECUTOR_PROCESS_FAILED',
            `${label} exited with ${exitSignal ?? code}: ${stderr.slice(-1024)}`,
          ));
          return;
        }
        try {
          const artifactRefs = await artifactCollector({
            executor: executorId,
            request,
            cwd,
            stdout,
            stderr,
          });
          settled = true;
          cleanup();
          resolve({
            outcome: 'completed',
            summary: `${label} completed successfully`,
            artifactRefs,
            metrics: {
              stdoutBytes: Buffer.byteLength(stdout),
              stderrBytes: Buffer.byteLength(stderr),
            },
          });
        } catch (error) {
          fail(error);
        }
      });
    });
  }

  return {
    id: executorId,
    label,
    kind: 'isolated-process',
    enabled: true,
    dispatchImplemented: true,
    capabilities: ['code-edit', 'cancel', 'artifact-refs', 'workspace-alias'],
    submit,
    async cancel(jobId) {
      const child = active.get(jobId);
      if (!child) return false;
      child.kill('SIGTERM');
      return true;
    },
    async getState(jobId) {
      return { active: active.has(jobId) };
    },
    async health() {
      try {
        await fs.access(executable);
        await fs.access(allowedRoot);
        return { ok: true };
      } catch (error) {
        return {
          ok: false,
          code: String(error?.code ?? 'EXECUTOR_PROCESS_UNAVAILABLE').slice(0, 128),
        };
      }
    },
    async *streamEvents() {
      throw adapterError(
        'EXECUTOR_PROCESS_STREAM_PUSH_ONLY',
        'Process events are delivered through submit emit callbacks',
      );
    },
  };
}

export function createAiderExecutorAdapter(options = {}) {
  return createProcessExecutorAdapter({
    id: 'aider',
    label: 'Aider',
    buildArgs: (instruction) => ['--yes-always', '--message', instruction],
    ...options,
  });
}

export function createOpenHandsExecutorAdapter(options = {}) {
  return createProcessExecutorAdapter({
    id: 'openhands',
    label: 'OpenHands',
    buildArgs: (instruction) => ['--headless', '--task', instruction],
    ...options,
  });
}

export const executorAdapterInternals = {
  DEFAULT_ALLOWED_ENV,
  DEFAULT_OUTPUT_LIMIT,
  EventEmitter,
  adapterError,
  appendBounded,
  isLoopbackHostname,
  isLocalExecutorHostname,
  normalizeBaseUrl,
  parseSse,
  resolveWorkspace,
  terminalGoosedEvent,
};