openclaw/scripts/e2e/telegram-mantis-sut.ts

#!/usr/bin/env -S node --import tsx
// Telegram Mantis SUT script owns the isolated OpenClaw side of desktop proof.

import { spawn, spawnSync, type SpawnOptionsWithoutStdio } from "node:child_process";
import { randomUUID } from "node:crypto";
import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { sliceUtf16Safe } from "@openclaw/normalization-core/utf16-slice";
import { z } from "zod";
import { coerceErrorMessage } from "../lib/error-format.mts";
import { sleep } from "../lib/sleep.mjs";
import { createPnpmRunnerSpawnSpec } from "../pnpm-runner.mts";
import { readTextFileTail } from "./lib/text-file-utils.mjs";
import { telegramBotApi } from "./telegram-bot-api.ts";

type GatewaySpawnSpec = {
  args: string[];
  command: string;
  options: SpawnOptionsWithoutStdio;
};

type JsonObject = Record<string, unknown>;
type MantisSutLane = "baseline" | "candidate";
type SpawnedDaemon = { child: ReturnType<typeof spawn>; error?: Error };

type MantisSutRuntime = {
  configPath: string;
  containerName: string;
  drained: {
    drained: number;
    pendingAfter?: number;
    pendingBefore?: number;
    webhookUrlSet: boolean;
  };
  gatewayLog: string;
  gatewayPid: number;
  mockLog: string;
  mockResponseControl: string;
  requestLog: string;
  stateDir: string;
  sutAttestation: { lane: MantisSutLane; sha: string };
  tempRoot: string;
  workspace: string;
};

export type MantisSutRecovery = Pick<
  MantisSutRuntime,
  "containerName" | "gatewayLog" | "mockLog" | "mockResponseControl" | "requestLog" | "tempRoot"
>;

function childProcessBaseEnv(): NodeJS.ProcessEnv {
  const keys = [
    "CI",
    "COREPACK_HOME",
    "FORCE_COLOR",
    "HOME",
    "LANG",
    "LC_ALL",
    "NODE_OPTIONS",
    "OPENCLAW_BUILD_PRIVATE_QA",
    "OPENCLAW_ENABLE_PRIVATE_QA_CLI",
    "PATH",
    "PNPM_HOME",
    "SHELL",
    "TEMP",
    "TMP",
    "TMPDIR",
    "USER",
    "XDG_CACHE_HOME",
    "XDG_CONFIG_HOME",
  ];
  const env: NodeJS.ProcessEnv = {};
  for (const key of keys) {
    const value = process.env[key];
    if (value) {
      env[key] = value;
    }
  }
  return env;
}

export function createMantisMockServerEnv(params: {
  mockPort: number;
  mockResponseChunkDelayMs?: number;
  mockResponseText: string;
  requestLog: string;
}): NodeJS.ProcessEnv {
  return {
    ...childProcessBaseEnv(),
    MOCK_PORT: String(params.mockPort),
    MOCK_REQUEST_LOG: params.requestLog,
    SUCCESS_MARKER: params.mockResponseText,
    ...(params.mockResponseChunkDelayMs === undefined
      ? {}
      : { MOCK_RESPONSE_CHUNK_DELAY_MS: String(params.mockResponseChunkDelayMs) }),
  };
}

export function createMantisGatewayEnv(params: {
  configPath: string;
  gatewayPassword?: string;
  stateDir: string;
  sutToken: string;
  tailscaleProxyDir?: string;
}): NodeJS.ProcessEnv {
  return {
    ...childProcessBaseEnv(),
    OPENAI_API_KEY: "sk-openclaw-e2e-mock",
    OPENCLAW_CONFIG_PATH: params.configPath,
    ...(params.gatewayPassword ? { OPENCLAW_GATEWAY_PASSWORD: params.gatewayPassword } : {}),
    OPENCLAW_STATE_DIR: params.stateDir,
    ...(params.tailscaleProxyDir
      ? { PATH: `${params.tailscaleProxyDir}${path.delimiter}${process.env.PATH ?? ""}` }
      : {}),
    TELEGRAM_BOT_TOKEN: params.sutToken,
  };
}

export function createOpenClawGatewaySpawnSpec(params: {
  env: NodeJS.ProcessEnv;
  gatewayPort: number;
  repoRoot: string;
  comSpec?: string;
  nodeExecPath?: string;
  npmExecPath?: string;
  pnpmExecPath?: string;
  platform?: NodeJS.Platform;
}): GatewaySpawnSpec {
  if (params.pnpmExecPath) {
    return {
      args: ["openclaw", "gateway", "--port", String(params.gatewayPort)],
      command: params.pnpmExecPath,
      options: { cwd: params.repoRoot, env: params.env, shell: false },
    };
  }
  const spec = createPnpmRunnerSpawnSpec({
    comSpec: params.comSpec,
    cwd: params.repoRoot,
    env: params.env,
    nodeExecPath: params.nodeExecPath,
    npmExecPath: params.npmExecPath,
    platform: params.platform,
    pnpmArgs: ["openclaw", "gateway", "--port", String(params.gatewayPort)],
  });
  return {
    args: spec.args,
    command: spec.command,
    options: {
      cwd: spec.options.cwd,
      env: spec.options.env,
      shell: spec.options.shell,
      windowsVerbatimArguments: spec.options.windowsVerbatimArguments,
    },
  };
}

export function writeSutConfig(params: {
  gatewayPort: number;
  groupId: string;
  humanDelayFixedMs?: number;
  linkPreview?: boolean;
  mcpAppFixture?: boolean;
  mockPort: number;
  outputDir: string;
  repoRoot?: string;
  testerId: string;
}) {
  const tempRoot = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-tg-crabbox-sut-"));
  const stateDir = path.join(tempRoot, "state");
  const workspace = path.join(tempRoot, "workspace");
  fs.mkdirSync(stateDir, { recursive: true });
  fs.mkdirSync(workspace, { recursive: true });
  const configPath = path.join(tempRoot, "openclaw.json");
  const config = {
    agents: {
      defaults: {
        ...(params.humanDelayFixedMs === undefined
          ? {}
          : {
              humanDelay: {
                maxMs: params.humanDelayFixedMs,
                minMs: params.humanDelayFixedMs,
                mode: "custom",
              },
            }),
        model: { primary: "openai/gpt-5.6-luna" },
        models: {
          "openai/gpt-5.6-luna": { params: { openaiWsWarmup: false, transport: "sse" } },
        },
      },
      entries: {
        main: {
          default: true,
          model: { primary: "openai/gpt-5.6-luna" },
          name: "Main",
          workspace,
        },
      },
    },
    logging: { audit: { enabled: true, executionIdentity: true, messages: "direct" } },
    channels: {
      telegram: {
        allowFrom: [params.testerId],
        apiRoot: "http://telegram-api-proxy:8080",
        botToken: { id: "TELEGRAM_BOT_TOKEN", provider: "default", source: "env" },
        commands: { native: true, nativeSkills: false },
        dmPolicy: "allowlist",
        enabled: true,
        groupAllowFrom: [params.testerId],
        groupPolicy: "allowlist",
        groups: {
          [params.groupId]: {
            allowFrom: [params.testerId],
            groupPolicy: "allowlist",
            requireMention: false,
          },
        },
        ...(params.linkPreview === undefined ? {} : { linkPreview: params.linkPreview }),
      },
    },
    gateway: params.mcpAppFixture
      ? {
          auth: {
            mode: "password",
            password: {
              id: "OPENCLAW_GATEWAY_PASSWORD",
              provider: "default",
              source: "env",
            },
          },
          bind: "loopback",
          mode: "local",
          port: params.gatewayPort,
          tailscale: { mode: "funnel" },
        }
      : { auth: { mode: "none" }, bind: "loopback", mode: "local", port: params.gatewayPort },
    ...(params.mcpAppFixture
      ? {
          mcp: {
            servers: {
              fixture: {
                args: [
                  path.join(
                    params.repoRoot ?? process.cwd(),
                    "scripts/e2e/mcp-app-conformance-server.mjs",
                  ),
                ],
                command: process.execPath,
              },
            },
          },
        }
      : {}),
    messages: { groupChat: { visibleReplies: "automatic" } },
    models: {
      providers: {
        openai: {
          api: "openai-responses",
          apiKey: { id: "OPENAI_API_KEY", provider: "default", source: "env" },
          baseUrl: `http://127.0.0.1:${params.mockPort}/v1`,
          models: [
            {
              api: "openai-responses",
              contextWindow: 128000,
              id: "gpt-5.6-luna",
              name: "gpt-5.6-luna",
            },
          ],
          request: { allowPrivateNetwork: true },
        },
      },
    },
    plugins: {
      allow: ["telegram", "openai"],
      enabled: true,
      entries: { openai: { enabled: true }, telegram: { enabled: true } },
    },
  };
  fs.writeFileSync(configPath, `${JSON.stringify(config, null, 2)}\n`);
  return { configPath, stateDir, tempRoot, workspace };
}

function telegramResultObject(value: unknown, label: string): JsonObject {
  if (!value || typeof value !== "object" || Array.isArray(value)) {
    throw new Error(`${label} returned an invalid payload.`);
  }
  return value as JsonObject;
}

export async function drainSutUpdates(sutToken: string) {
  const before = telegramResultObject(
    await telegramBotApi(sutToken, "getWebhookInfo", {}),
    "getWebhookInfo",
  );
  const rawUpdates = await telegramBotApi(sutToken, "getUpdates", {
    allowed_updates: ["message", "edited_message"],
    timeout: 0,
  });
  if (!Array.isArray(rawUpdates)) {
    throw new Error("getUpdates returned an invalid payload.");
  }
  if (rawUpdates.length) {
    const last = rawUpdates.at(-1);
    if (
      last &&
      typeof last === "object" &&
      "update_id" in last &&
      typeof last.update_id === "number"
    ) {
      await telegramBotApi(sutToken, "getUpdates", { offset: last.update_id + 1, timeout: 0 });
    }
  }
  const after = telegramResultObject(
    await telegramBotApi(sutToken, "getWebhookInfo", {}),
    "getWebhookInfo",
  );
  return {
    drained: rawUpdates.length,
    pendingAfter:
      typeof after.pending_update_count === "number" ? after.pending_update_count : undefined,
    pendingBefore:
      typeof before.pending_update_count === "number" ? before.pending_update_count : undefined,
    webhookUrlSet: typeof before.url === "string" && before.url.length > 0,
  };
}

function spawnDaemon(params: {
  args: string[];
  command: string;
  cwd: string;
  env: NodeJS.ProcessEnv;
  logPath: string;
  shell?: boolean;
  windowsVerbatimArguments?: boolean;
}): SpawnedDaemon {
  const log = fs.openSync(params.logPath, "a");
  const child = spawn(params.command, params.args, {
    cwd: params.cwd,
    detached: true,
    env: params.env,
    shell: params.shell,
    stdio: ["ignore", log, log],
    windowsVerbatimArguments: params.windowsVerbatimArguments,
  });
  const daemon: SpawnedDaemon = { child };
  child.on("error", (error) => {
    daemon.error = error;
  });
  child.unref();
  fs.closeSync(log);
  return daemon;
}

function readLogTail(logPath: string, maxBytes = 256 * 1024): string {
  return readTextFileTail(logPath, Math.max(1, maxBytes));
}

function logTailDiagnostic(logPath: string): string {
  const tail = sliceUtf16Safe(readLogTail(logPath), -4000);
  return `${path.basename(logPath)}:${tail ? `\n${tail}` : " <empty>"}`;
}

function describeDaemonFailure(daemon: SpawnedDaemon): string | undefined {
  if (daemon.error) {
    return `failed to start: ${daemon.error.message}`;
  }
  if (daemon.child.signalCode) {
    return `was terminated by signal ${daemon.child.signalCode}`;
  }
  if (daemon.child.exitCode !== null) {
    return `exited with exit code ${daemon.child.exitCode}`;
  }
  return undefined;
}

export async function waitForLog(
  logPath: string,
  pattern: RegExp,
  label: string,
  timeoutMs: number,
  daemonContext?: { daemon: SpawnedDaemon; logPath: string },
): Promise<void> {
  const started = Date.now();
  while (true) {
    if (pattern.test(readLogTail(logPath))) {
      return;
    }
    if (daemonContext) {
      // The awaited log lives inside the daemon, so a daemon that already died can never
      // satisfy the pattern: report how it died instead of waiting out the full timeout.
      const failure = describeDaemonFailure(daemonContext.daemon);
      if (failure) {
        throw new Error(
          `Container-isolated SUT ${failure} before ${label} became ready.\n${logTailDiagnostic(daemonContext.logPath)}\n${logTailDiagnostic(logPath)}`,
        );
      }
    }
    const remainingMs = timeoutMs - (Date.now() - started);
    if (remainingMs <= 0) {
      break;
    }
    await sleep(Math.min(500, remainingMs));
  }
  const timeoutDetail = daemonContext
    ? `; container-isolated SUT is still running (pid ${daemonContext.daemon.child.pid ?? "unknown"}).\n${logTailDiagnostic(daemonContext.logPath)}\n${logTailDiagnostic(logPath)}`
    : `\n${sliceUtf16Safe(readLogTail(logPath), -4000)}`;
  throw new Error(`${label} did not become ready within ${timeoutMs}ms${timeoutDetail}`);
}

export function createContainerizedSutSpawnSpec(params: {
  containerName: string;
  gatewayPort: number;
  mockPort: number;
  mockResponseChunkDelayMs?: number;
  mockResponseText: string;
  repoRoot: string;
  runtimeRoot: string;
  sutLane: MantisSutLane;
  gatewayEnv: NodeJS.ProcessEnv;
}) {
  const containerHome = path.join(params.runtimeRoot, "container-home");
  fs.mkdirSync(containerHome, { recursive: true });
  const inputPath = path.join(params.runtimeRoot, "container-input.json");
  fs.writeFileSync(
    inputPath,
    `${JSON.stringify({
      gatewayPassword: params.gatewayEnv.OPENCLAW_GATEWAY_PASSWORD,
      mockResponseChunkDelayMs: params.mockResponseChunkDelayMs,
      mockResponseText: params.mockResponseText,
      telegramBotToken: params.gatewayEnv.TELEGRAM_BOT_TOKEN,
    })}\n`,
    { mode: 0o600 },
  );
  return {
    args: [
      "-n",
      "/usr/local/sbin/openclaw-mantis-sut-container",
      "run",
      params.containerName,
      params.sutLane,
      params.repoRoot,
      params.runtimeRoot,
      String(params.gatewayPort),
      String(params.mockPort),
    ],
    command: "sudo",
    inputPath,
    options: {
      cwd: process.cwd(),
      env: childProcessBaseEnv(),
      shell: false,
    } satisfies SpawnOptionsWithoutStdio,
  };
}

type SutContainerAction = "destroy" | "stop";
type SutContainerCommandRunner = (
  command: string,
  args: string[],
  options: { encoding: "utf8"; env: NodeJS.ProcessEnv; stdio: "pipe" },
) => {
  error?: Error;
  signal?: NodeJS.Signals | null;
  status: number | null;
  stderr?: string;
};

export function runSutContainerAction(
  action: SutContainerAction,
  containerName: string | undefined,
  runtimeRoot: string | undefined,
  run: SutContainerCommandRunner = spawnSync,
): void {
  if (!containerName || !runtimeRoot) {
    return;
  }
  const result = run(
    "sudo",
    ["-n", "/usr/local/sbin/openclaw-mantis-sut-container", action, containerName, runtimeRoot],
    { encoding: "utf8", env: childProcessBaseEnv(), stdio: "pipe" },
  );
  if (result.error) {
    throw new Error(`Failed to ${action} container-isolated SUT: ${result.error.message}`, {
      cause: result.error,
    });
  }
  const stderr = result.stderr?.toString().trim().slice(-4_000);
  if (result.signal) {
    throw new Error(
      `Container-isolated SUT ${action} was terminated by ${result.signal}.${stderr ? `\n${stderr}` : ""}`,
    );
  }
  if (result.status !== 0) {
    throw new Error(
      `Container-isolated SUT ${action} failed with exit code ${result.status ?? "unknown"}.${stderr ? `\n${stderr}` : ""}`,
    );
  }
}

export function preserveMantisSutRuntimeArtifacts(
  sut: Pick<MantisSutRuntime, "gatewayLog" | "mockLog" | "requestLog">,
  outputDir: string,
): void {
  for (const source of [sut.gatewayLog, sut.mockLog, sut.requestLog]) {
    const target = path.join(outputDir, path.basename(source));
    if (path.resolve(source) !== path.resolve(target) && fs.existsSync(source)) {
      fs.copyFileSync(source, target);
    }
  }
}

export function stopMantisSut(sut: Pick<MantisSutRuntime, "containerName" | "tempRoot">): void {
  runSutContainerAction("stop", sut.containerName, sut.tempRoot);
}

export function destroyMantisSut(sut: Pick<MantisSutRuntime, "containerName" | "tempRoot">): void {
  runSutContainerAction("destroy", sut.containerName, sut.tempRoot);
}

function cleanupFailureMessage(message: string, cleanupErrors: unknown[]): string {
  return [
    message,
    ...cleanupErrors.map((error) => `Cleanup failure: ${coerceErrorMessage(error)}`),
  ].join("\n");
}

export async function startMantisSut(params: {
  gatewayPort: number;
  groupId: string;
  humanDelayFixedMs?: number;
  linkPreview?: boolean;
  mockPort: number;
  mockResponseChunkDelayMs?: number;
  mockResponseText: string;
  outputDir: string;
  repoRoot: string;
  sutLane: MantisSutLane;
  sutToken: string;
  testerId: string;
  onRuntimeCreated?: (runtime: MantisSutRecovery) => void;
  onRuntimeDisposed?: () => void;
}): Promise<MantisSutRuntime> {
  const drained = await drainSutUpdates(params.sutToken);
  const config = writeSutConfig(params);
  // The root wrapper relocates tempRoot into its bounded filesystem, then restores this
  // exact path as a symlink before Docker starts. Keep controller and claim paths anchored
  // here so live log reads, mock updates, stop, and destroy all share one runtime identity.
  const requestLog = path.join(config.tempRoot, "mock-openai-requests.ndjson");
  const mockLog = path.join(config.tempRoot, "mock-openai.log");
  const mockResponseControlDir = path.join(config.tempRoot, "mock-control");
  fs.mkdirSync(mockResponseControlDir, { mode: 0o700 });
  const mockResponseControl = path.join(mockResponseControlDir, "response.json");
  fs.writeFileSync(
    mockResponseControl,
    `${JSON.stringify({
      chunkDelayMs: params.mockResponseChunkDelayMs ?? 0,
      text: params.mockResponseText,
    })}\n`,
    { mode: 0o600 },
  );
  const gatewayLog = path.join(config.tempRoot, "gateway.log");
  const gatewayEnv = createMantisGatewayEnv({ ...config, sutToken: params.sutToken });
  const containerName = `openclaw-telegram-sut-${randomUUID()}`;
  const spec = createContainerizedSutSpawnSpec({
    containerName,
    gatewayEnv,
    gatewayPort: params.gatewayPort,
    mockPort: params.mockPort,
    mockResponseChunkDelayMs: params.mockResponseChunkDelayMs,
    mockResponseText: params.mockResponseText,
    repoRoot: params.repoRoot,
    runtimeRoot: config.tempRoot,
    sutLane: params.sutLane,
  });
  params.onRuntimeCreated?.({
    containerName,
    gatewayLog,
    mockLog,
    mockResponseControl,
    requestLog,
    tempRoot: config.tempRoot,
  });
  try {
    const daemonLogPath = path.join(params.outputDir, "sut-container.log");
    const daemon = spawnDaemon({
      args: spec.args,
      command: spec.command,
      cwd: typeof spec.options.cwd === "string" ? spec.options.cwd : process.cwd(),
      env: spec.options.env ?? {},
      logPath: daemonLogPath,
      shell: false,
    });
    const daemonContext = { daemon, logPath: daemonLogPath };
    await waitForLog(mockLog, /mock-openai listening/u, "mock-openai", 30_000, daemonContext);
    await waitForLog(gatewayLog, /\[gateway\] ready/u, "gateway", 60_000, daemonContext);
    const gatewayPid = daemon.child.pid;
    if (!gatewayPid) {
      throw new Error("Container-isolated SUT became ready without a daemon process id.");
    }
    const sutAttestation = z
      .object({ lane: z.enum(["baseline", "candidate"]), sha: z.string().regex(/^[0-9a-f]{40}$/u) })
      .parse(
        JSON.parse(fs.readFileSync(path.join(config.tempRoot, "sut-attestation.json"), "utf8")),
      );
    if (sutAttestation.lane !== params.sutLane) {
      throw new Error("Container-isolated SUT attestation mismatch.");
    }
    return {
      ...config,
      containerName,
      drained,
      gatewayLog,
      gatewayPid,
      mockLog,
      mockResponseControl,
      requestLog,
      sutAttestation,
    };
  } catch (error) {
    const cleanupErrors: unknown[] = [];
    let stopped = false;
    try {
      runSutContainerAction("stop", containerName, config.tempRoot);
      stopped = true;
    } catch (cleanupError) {
      cleanupErrors.push(cleanupError);
    }
    if (stopped) {
      try {
        preserveMantisSutRuntimeArtifacts({ gatewayLog, mockLog, requestLog }, params.outputDir);
      } catch (cleanupError) {
        cleanupErrors.push(cleanupError);
      }
    }
    try {
      runSutContainerAction("destroy", containerName, config.tempRoot);
    } catch (cleanupError) {
      cleanupErrors.push(cleanupError);
    }
    fs.rmSync(spec.inputPath, { force: true });
    if (cleanupErrors.length > 0) {
      throw new Error(
        cleanupFailureMessage(
          "Local SUT startup failed and cleanup was incomplete.",
          cleanupErrors,
        ),
        { cause: error },
      );
    }
    params.onRuntimeDisposed?.();
    throw error;
  }
}