pikasTech-HWLAB/scripts/harnessrl-native-l1-smoke.mjs

#!/usr/bin/env node

import { spawn } from "node:child_process";
import { createWriteStream } from "node:fs";
import { mkdir, readFile, rm } from "node:fs/promises";
import net from "node:net";
import path from "node:path";

const root = process.cwd();
const temporalAddress = requiredEnv("HARNESSRL_TEMPORAL_ADDRESS");
const temporalNamespace = process.env.HARNESSRL_TEMPORAL_NAMESPACE || "unidesk";
const startedAt = Date.now();
const tag = `${new Date().toISOString().replace(/[-:.]/gu, "").slice(0, 15)}-${process.pid}`;
const experimentRoot = path.join(process.env.TMPDIR || "/tmp", `hwlab-r45-caserun-${tag}`);
const logDir = path.join(experimentRoot, "logs");
const stateDir = path.join(experimentRoot, "state");
const databaseName = `hwlab_r45_${process.pid}_${Date.now() % 1000000}`;
const databaseUrl = `postgresql://root@/${databaseName}?host=/var/run/postgresql`;
const taskQueue = `hwlab-r45-${tag}`;
const defaultHttpTimeoutMs = 5000;
const defaultProcessTimeoutMs = 30000;
const processTerminationGraceMs = 2000;
const pollTimeoutMs = 30000;
const children = new Map();
const runIds = {};
let databaseCreated = false;
let result;
let failure;

await mkdir(logDir, { recursive: true });
await mkdir(stateDir, { recursive: true });

try {
  await run("postgres-probe", "psql", ["-d", "postgres", "-Atqc", "select current_user"]);
  await run("database-create", "createdb", [databaseName]);
  databaseCreated = true;

  const apiPort = await availablePort();
  const workerPort = await availablePort();
  const nativeFixturePort = await availablePort();
  const vitePort = await availablePort();
  const reloadPort = await availablePort();
  const baseUrl = `http://127.0.0.1:${apiPort}`;
  const nativeFixtureUrl = `http://127.0.0.1:${nativeFixturePort}`;
  const viteUrl = `http://127.0.0.1:${vitePort}`;
  const runtimeEnv = {
    ...process.env,
    HARNESSRL_DATABASE_URL: databaseUrl,
    HARNESSRL_TEMPORAL_ADDRESS: temporalAddress,
    HARNESSRL_TEMPORAL_NAMESPACE: temporalNamespace,
    HARNESSRL_TEMPORAL_TASK_QUEUE: taskQueue,
    HARNESSRL_STATE_DIR: stateDir,
    HWLAB_CASERUN_REPO_ROOT: root,
    HWLAB_CASERUN_CASE_REPO: root,
    HARNESSRL_API_HOST: "127.0.0.1",
    HARNESSRL_API_PORT: String(apiPort),
    HARNESSRL_WORKER_HEALTH_HOST: "127.0.0.1",
    HARNESSRL_WORKER_HEALTH_PORT: String(workerPort)
  };

  stage("infra-ready", { databaseName, temporalAddress, temporalNamespace, taskQueue });
  await startApi(runtimeEnv, baseUrl);

  runIds.apiRestart = `r45-api-restart-${tag}`;
  await submitHttp(baseUrl, runIds.apiRestart);
  await stopChild("api");
  await startApi(runtimeEnv, baseUrl);
  await startWorker(runtimeEnv, workerPort);
  const apiRestart = await pollHttp(baseUrl, runIds.apiRestart, "completed");
  stage("api-restart-recovery", identity(apiRestart));

  await stopChild("worker");
  runIds.workerRestart = `r45-worker-restart-${tag}`;
  await submitHttp(baseUrl, runIds.workerRestart);
  runIds.cancel = `r45-cancel-${tag}`;
  await submitHttp(baseUrl, runIds.cancel);
  await requestJson("cancel-run", `${baseUrl}/v1/caserun/runs/${encodeURIComponent(runIds.cancel)}/cancel`, { method: "POST" }, 202);
  await startWorker(runtimeEnv, workerPort);
  const workerRestart = await pollHttp(baseUrl, runIds.workerRestart, "completed");
  stage("worker-restart-recovery", identity(workerRestart));
  const canceled = await pollHttp(baseUrl, runIds.cancel, "canceled");
  stage("cancel", identity(canceled));

  runIds.local = `r45-local-${tag}`;
  runIds.overApi = `r45-over-api-${tag}`;
  await cli(runtimeEnv, ["case", "run", "start", "software-smoke", "--case-repo", root, "--run-id", runIds.local]);
  await cli(runtimeEnv, ["case", "run", "start", "software-smoke", "--over-api", "--base-url", baseUrl, "--run-id", runIds.overApi]);
  const localFinal = await pollCli(runtimeEnv, "local", runIds.local, "completed", baseUrl);
  const apiFinal = await pollCli(runtimeEnv, "api", runIds.overApi, "completed", baseUrl);
  const localEvents = await cli(runtimeEnv, ["case", "run", "events", runIds.local]);
  const apiEvents = await cli(runtimeEnv, ["case", "run", "events", runIds.overApi, "--over-api", "--base-url", baseUrl]);
  const localAggregate = await cli(runtimeEnv, ["case", "run", "aggregate", runIds.local]);
  const apiAggregate = await cli(runtimeEnv, ["case", "run", "aggregate", runIds.overApi, "--over-api", "--base-url", baseUrl]);
  requireCondition(runIds.local !== runIds.overApi, "local and over-api runId must be independent");
  requireCondition(localFinal.workflowRunId !== apiFinal.workflowRunId, "local and over-api workflow execution must be independent");
  requireCondition(localFinal.status === apiFinal.status, "local and over-api status differ");
  requireCondition(localEvents.events.length === apiEvents.events.length, "local and over-api event counts differ");
  const localArtifact = softwareSmokeArtifact(localAggregate);
  const apiArtifact = softwareSmokeArtifact(apiAggregate);
  requireCondition(validSha(localArtifact.sha256) && validSha(apiArtifact.sha256), "software-smoke artifact SHA is missing");
  stage("transport-equivalence", {
    localRunId: runIds.local,
    overApiRunId: runIds.overApi,
    status: localFinal.status,
    eventCount: localEvents.events.length,
    artifactCount: 1,
    artifactShaPresent: true
  });

  const missingRunId = `r45-missing-${tag}`;
  const localError = await cli(runtimeEnv, ["case", "run", "status", missingRunId], { allowFailure: true });
  const apiError = await cli(runtimeEnv, ["case", "run", "status", missingRunId, "--over-api", "--base-url", baseUrl], { allowFailure: true });
  requireCondition(localError.exitCode !== 0 && apiError.exitCode !== 0, "typed not-found must fail for both transports");
  requireCondition(localError.error?.code === "caserun_run_not_found", `unexpected local error ${localError.error?.code}`);
  requireCondition(apiError.error?.code === localError.error.code, `typed errors differ ${localError.error?.code}/${apiError.error?.code}`);
  stage("typed-error-equivalence", { code: localError.error.code });

  const webEnv = {
    ...process.env,
    HWLAB_CASERUN_NATIVE_TEST: "1",
    HWLAB_CASERUN_NATIVE_PORT: String(nativeFixturePort),
    HWLAB_CASERUN_NATIVE_URL: nativeFixtureUrl,
    HWLAB_CASERUN_RELOAD_SMOKE_PORT: String(reloadPort)
  };
  await startChild("native-fixture", ["bun", "scripts/caserun-native-supervisor.ts"], {
    cwd: path.join(root, "web/hwlab-cloud-web"), env: webEnv
  });
  await waitForHttp("wait-native-fixture", `${nativeFixtureUrl}/health`);
  await run("native-fixture-smoke", "bun", ["scripts/caserun-native-smoke.ts"], {
    cwd: path.join(root, "web/hwlab-cloud-web"), env: webEnv
  });
  await startChild("vite", ["bunx", "vite", "--host", "127.0.0.1", "--port", String(vitePort), "--strictPort"], {
    cwd: path.join(root, "web/hwlab-cloud-web"), env: webEnv
  });
  const viteHtml = await waitForHttp("wait-vite-workbench", `${viteUrl}/workbench`, { returnText: true });
  requireCondition(viteHtml.includes("/@vite/client"), "Vite HMR client was not injected");
  const proxiedCases = await requestJson("vite-caserun-cases", `${viteUrl}/v1/caserun/cases`);
  requireCondition(proxiedCases.mode === "native-test" && Array.isArray(proxiedCases.cases), "Vite native CaseRun proxy failed");
  await run("native-reload-smoke", "bun", ["scripts/caserun-native-reload-smoke.ts"], {
    cwd: path.join(root, "web/hwlab-cloud-web"), env: webEnv
  });
  stage("web-native-http-hmr", { nativeFixtureUrl, viteUrl, caseCount: proxiedCases.cases.length, hmrClient: true });

  const databaseCounts = await queryDatabase(databaseUrl, "select (select count(*) from harnessrl_runs)::int as runs,(select count(*) from harnessrl_run_events)::int as events");
  result = {
    ok: true,
    contractVersion: "hwlab-caserun-r45-native-smoke-v1",
    temporal: { address: temporalAddress, namespace: temporalNamespace, taskQueue },
    runs: runIds,
    registry: databaseCounts,
    web: { nativeFixtureUrl, viteUrl, hmrClient: true },
    nativeElapsedSeconds: Math.ceil((Date.now() - startedAt) / 1000)
  };
} catch (error) {
  const logs = await diagnosticLogs();
  failure = {
    ok: false,
    stage: "harnessrl-native-l1-smoke",
    error: error instanceof Error ? { name: error.name, message: error.message } : { message: String(error) },
    logs
  };
  process.exitCode = 1;
} finally {
  const cleanup = await cleanupResources();
  if (result) console.log(JSON.stringify({ ...result, cleanup }));
  else if (failure) console.error(JSON.stringify({ ...failure, cleanup }));
}

function requiredEnv(name) {
  const value = String(process.env[name] || "").trim();
  if (!value) throw new Error(`${name} is required; resolve the current Temporal frontend Service address before running the smoke`);
  return value;
}

function requireCondition(condition, message) {
  if (!condition) throw new Error(message);
}

function validSha(value) {
  return typeof value === "string" && /^[a-f0-9]{64}$/u.test(value);
}

function identity(record) {
  return { runId: record.runId, workflowId: record.workflowId, workflowRunId: record.workflowRunId, status: record.status };
}

function stage(name, details = {}) {
  console.error(JSON.stringify({ event: "r45-stage", name, at: new Date().toISOString(), ...details }));
}

async function availablePort() {
  return await new Promise((resolve, reject) => {
    const server = net.createServer();
    server.once("error", reject);
    server.listen(0, "127.0.0.1", () => {
      const address = server.address();
      server.close((error) => error ? reject(error) : resolve(address.port));
    });
  });
}

async function startApi(env, baseUrl) {
  await startChild("api", ["bun", "cmd/hwlab-harnessrl-api/main.ts"], { cwd: root, env });
  await waitForHttp("wait-api-ready", `${baseUrl}/readyz`);
  stage("api-started", { pid: children.get("api").pid });
}

async function startWorker(env, port) {
  await startChild("worker", ["bun", "cmd/hwlab-harnessrl-worker/main.ts"], { cwd: root, env });
  await waitForHttp("wait-worker-ready", `http://127.0.0.1:${port}/readyz`, { timeoutMs: 30000 });
  stage("worker-started", { pid: children.get("worker").pid });
}

async function startChild(name, argv, options) {
  requireCondition(!children.has(name), `${name} is already running`);
  const logPath = path.join(logDir, `${name}.log`);
  const log = createWriteStream(logPath, { flags: "a" });
  const child = spawn(argv[0], argv.slice(1), { cwd: options.cwd, env: options.env, stdio: ["ignore", "pipe", "pipe"] });
  child.stdout.pipe(log);
  child.stderr.pipe(log);
  child.once("exit", () => log.end());
  children.set(name, child);
}

async function stopChild(name) {
  const child = children.get(name);
  if (!child) return;
  children.delete(name);
  if (child.exitCode === null && child.signalCode === null) {
    child.kill("SIGTERM");
    const exit = await Promise.race([
      new Promise((resolve) => child.once("exit", (code, signal) => resolve({ code, signal }))),
      sleep(8000).then(() => null)
    ]);
    if (!exit && child.exitCode === null) {
      child.kill("SIGKILL");
      await new Promise((resolve) => child.once("exit", resolve));
    }
  }
  stage(`${name}-stopped`, { pid: child.pid, exitCode: child.exitCode, signal: child.signalCode });
}

async function run(label, command, args, options = {}) {
  const timeoutMs = positiveTimeout(options.timeoutMs, defaultProcessTimeoutMs, "process timeout");
  const completed = await new Promise((resolve, reject) => {
    const child = spawn(command, args, { cwd: options.cwd || root, env: options.env || process.env, stdio: ["ignore", "pipe", "pipe"] });
    let stdout = "";
    let stderr = "";
    let timedOut = false;
    let terminationTimer;
    const timeoutTimer = setTimeout(() => {
      timedOut = true;
      child.kill("SIGTERM");
      terminationTimer = setTimeout(() => {
        if (child.exitCode === null && child.signalCode === null) child.kill("SIGKILL");
      }, processTerminationGraceMs);
    }, timeoutMs);
    child.stdout.on("data", (chunk) => { stdout += chunk; });
    child.stderr.on("data", (chunk) => { stderr += chunk; });
    child.once("error", (error) => {
      clearTimeout(timeoutTimer);
      if (terminationTimer) clearTimeout(terminationTimer);
      reject(error);
    });
    child.once("exit", (exitCode, signal) => {
      clearTimeout(timeoutTimer);
      if (terminationTimer) clearTimeout(terminationTimer);
      resolve({ exitCode: exitCode ?? 1, signal, stdout, stderr, timedOut, timeoutMs });
    });
  });
  if (completed.timedOut) throw new Error(`process operation timed out operation=${clip(label, 160)} timeoutMs=${completed.timeoutMs}`);
  if (completed.exitCode !== 0 && !options.allowFailure) {
    throw new Error(`${label} failed exitCode=${completed.exitCode}: ${clip(completed.stderr || completed.stdout)}`);
  }
  return completed;
}

async function cli(env, args, options = {}) {
  const completed = await run(`hwlab-cli ${args.join(" ")}`, "bun", ["tools/hwlab-cli/bin/hwlab-cli.ts", ...args], {
    env, allowFailure: options.allowFailure, timeoutMs: options.timeoutMs
  });
  const payload = parseJson(completed.stdout, "hwlab-cli output");
  return { ...payload, exitCode: completed.exitCode };
}

async function submitHttp(baseUrl, runId) {
  const response = await requestJson("submit-run", `${baseUrl}/v1/caserun/runs`, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ caseId: "software-smoke", runId })
  }, 202);
  requireCondition(response.runId === runId, `submitted runId differs for ${runId}`);
  return response;
}

async function pollHttp(baseUrl, runId, expected) {
  const deadline = Date.now() + pollTimeoutMs;
  let previousStatus;
  for (let attempt = 1; Date.now() < deadline; attempt += 1) {
    const record = await requestJson("poll-run", `${baseUrl}/v1/caserun/runs/${encodeURIComponent(runId)}`, undefined, 200, {
      timeoutMs: Math.min(defaultHttpTimeoutMs, Math.max(1, deadline - Date.now()))
    });
    if (record.status !== previousStatus || record.status === expected) stage("poll", { transport: "api", runId, attempt, status: record.status });
    previousStatus = record.status;
    if (record.status === expected) return record;
    if (record.status === "failed") throw new Error(`run ${runId} failed: ${JSON.stringify(record.error)}`);
    await sleep(250);
  }
  throw new Error(`timeout waiting for ${runId}=${expected}`);
}

async function pollCli(env, mode, runId, expected, baseUrl) {
  const overApi = mode === "api" ? ["--over-api", "--base-url", baseUrl] : [];
  const deadline = Date.now() + pollTimeoutMs;
  let previousStatus;
  for (let attempt = 1; Date.now() < deadline; attempt += 1) {
    const record = await cli(env, ["case", "run", "status", runId, ...overApi], {
      timeoutMs: Math.min(defaultProcessTimeoutMs, Math.max(1, deadline - Date.now()))
    });
    if (record.status !== previousStatus || record.status === expected) stage("poll", { transport: mode, runId, attempt, status: record.status });
    previousStatus = record.status;
    if (record.status === expected) return record;
    if (record.status === "failed") throw new Error(`run ${runId} failed: ${JSON.stringify(record.error)}`);
    await sleep(250);
  }
  throw new Error(`timeout waiting for ${mode}/${runId}=${expected}`);
}

async function waitForHttp(operation, url, options = {}) {
  const timeoutMs = positiveTimeout(options.timeoutMs, 10000, "HTTP wait timeout");
  const deadline = Date.now() + timeoutMs;
  let lastError;
  while (Date.now() < deadline) {
    try {
      const requestTimeoutMs = Math.min(defaultHttpTimeoutMs, Math.max(1, deadline - Date.now()));
      const { response, text } = await requestText(operation, url, undefined, { timeoutMs: requestTimeoutMs });
      if (response.ok) return options.returnText ? text : response;
      lastError = new Error(`HTTP status operation=${boundedOperation(operation)} url=${boundedUrl(url)} status=${response.status}`);
    } catch (error) {
      lastError = error;
    }
    await sleep(100);
  }
  throw new Error(`HTTP wait timed out operation=${boundedOperation(operation)} url=${boundedUrl(url)} timeoutMs=${timeoutMs} lastError=${clip(lastError instanceof Error ? lastError.message : String(lastError), 400)}`);
}

async function requestJson(operation, url, init, expectedStatus = 200, options = {}) {
  const { response, text, timeoutMs } = await requestText(operation, url, init, options);
  let payload;
  try {
    payload = JSON.parse(text);
  } catch {
    throw new Error(`HTTP invalid JSON operation=${boundedOperation(operation)} url=${boundedUrl(url)} timeoutMs=${timeoutMs} body=${clip(text)}`);
  }
  if (response.status !== expectedStatus) {
    throw new Error(`HTTP status operation=${boundedOperation(operation)} url=${boundedUrl(url)} timeoutMs=${timeoutMs} status=${response.status} body=${clip(text)}`);
  }
  return payload;
}

async function requestText(operation, url, init, options = {}) {
  const timeoutMs = positiveTimeout(options.timeoutMs, defaultHttpTimeoutMs, "HTTP request timeout");
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), timeoutMs);
  try {
    const response = await fetch(url, { ...init, signal: controller.signal });
    const text = await response.text();
    return { response, text, timeoutMs };
  } catch (error) {
    if (controller.signal.aborted) {
      throw new Error(`HTTP request timed out operation=${boundedOperation(operation)} url=${boundedUrl(url)} timeoutMs=${timeoutMs}`);
    }
    throw new Error(`HTTP request failed operation=${boundedOperation(operation)} url=${boundedUrl(url)} timeoutMs=${timeoutMs} error=${clip(error instanceof Error ? error.message : String(error), 300)}`);
  } finally {
    clearTimeout(timer);
  }
}

async function queryDatabase(url, sql) {
  const completed = await run("database-query", "psql", [url, "-At", "-F", "|", "-c", sql]);
  const [runs, events] = completed.stdout.trim().split("|").map(Number);
  return { runs, events };
}

async function diagnosticLogs() {
  const logs = {};
  for (const name of ["api", "worker", "native-fixture", "vite"]) {
    const logPath = path.join(logDir, `${name}.log`);
    const text = await readFile(logPath, "utf8").catch(() => "");
    if (text) logs[name] = clip(text, 3000);
  }
  return logs;
}

async function cleanupResources() {
  const childNames = [...children.keys()];
  for (const name of childNames.reverse()) await stopChild(name);
  if (databaseCreated) {
    await run("database-drop", "dropdb", ["--if-exists", databaseName], { allowFailure: true });
    databaseCreated = false;
  }
  const databaseCheck = await run("database-cleanup-check", "psql", ["-d", "postgres", "-Atqc", `select count(*) from pg_database where datname='${databaseName}'`], { allowFailure: true });
  const remainingProcesses = [...children.values()].filter((child) => child.exitCode === null).length;
  await rm(experimentRoot, { recursive: true, force: true });
  const cleanup = {
    processesStopped: remainingProcesses === 0,
    databaseDropped: databaseCheck.stdout.trim() === "0",
    experimentDirectoryRemoved: true
  };
  if (!Object.values(cleanup).every(Boolean)) process.exitCode = 1;
  return cleanup;
}

function parseJson(text, label) {
  try {
    return JSON.parse(text);
  } catch {
    throw new Error(`invalid JSON from ${label}: ${clip(text)}`);
  }
}

function softwareSmokeArtifact(aggregate) {
  const artifacts = aggregate?.result?.summary?.artifacts;
  requireCondition(aggregate?.result?.summary?.artifactCount === 1, "software-smoke aggregate artifactCount must be 1");
  requireCondition(Array.isArray(artifacts) && artifacts.length === 1, "software-smoke aggregate artifact is missing");
  return artifacts[0];
}

function clip(value, limit = 1200) {
  const text = String(value || "").trim();
  return text.length <= limit ? text : `${text.slice(0, limit)}…`;
}

function positiveTimeout(value, fallback, label) {
  const timeoutMs = value === undefined ? fallback : Number(value);
  if (!Number.isSafeInteger(timeoutMs) || timeoutMs <= 0) throw new Error(`${label} must be a positive integer`);
  return timeoutMs;
}

function boundedOperation(value) {
  return clip(String(value || "unknown-operation").replace(/[\r\n\t]/gu, " "), 120);
}

function boundedUrl(value) {
  try {
    const url = new URL(String(value));
    url.username = "";
    url.password = "";
    url.search = "";
    url.hash = "";
    return clip(url.toString(), 300);
  } catch {
    return clip(String(value || "invalid-url").split(/[?#]/u, 1)[0], 300);
  }
}

function sleep(ms) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}