pikasTech-HWLAB/cmd/hwlab-gateway/main.ts

#!/usr/bin/env bun
import os from "node:os";
import { spawn } from "node:child_process";

import { createAuditEvent, createEvidenceRecord, operationContext } from "../../internal/sim/l2-runtime.mjs";
import { createGatewayState, createHealth } from "../../internal/sim/model.mjs";
import { createJsonServer, fetchJson, listen, parsePort } from "../../internal/sim/http.mjs";
import { ENVIRONMENT_DEV, ERROR_CODES, JSON_RPC_VERSION, createRpcErrorBody, isProtocolEnvironment } from "../../internal/protocol/index.mjs";

const gatewayId = process.env.HWLAB_GATEWAY_ID ?? "gateway_dev_stub";
const gatewaySessionId = process.env.HWLAB_GATEWAY_SESSION_ID ?? `gws_${gatewayId}`;
const port = parsePort(process.env.PORT, 7001);
const cloudUrl = trimUrl(process.env.HWLAB_GATEWAY_CLOUD_URL);
const pollIntervalMs = parsePositiveInteger(process.env.HWLAB_GATEWAY_POLL_INTERVAL_MS, 500);
const commandTimeoutMs = parsePositiveInteger(process.env.HWLAB_GATEWAY_CMD_TIMEOUT_MS, 120000);
const outputLimitBytes = parsePositiveInteger(process.env.HWLAB_GATEWAY_CMD_OUTPUT_LIMIT_BYTES, 65536);
const maxInflightRequests = parsePositiveInteger(process.env.HWLAB_GATEWAY_MAX_INFLIGHT, 3);
// CI probe: touches only the hwlab-gateway code identity for env reuse validation round 2.
const gatewayEnvironment = normalizeEnvironment(process.env.HWLAB_ENVIRONMENT || process.env.HWLAB_GITOPS_PROFILE || ENVIRONMENT_DEV);
const resourceId = process.env.HWLAB_GATEWAY_RESOURCE_ID ?? "res_windows_host";
const boxId = process.env.HWLAB_GATEWAY_BOX_ID ?? "box_windows_host";
const commandExecutionEnabled = truthy(process.env.HWLAB_GATEWAY_CMD_EXEC_ENABLED) || truthy(process.env.HWLAB_GATEWAY_DEMO_OPEN);
const inflightRequests = new Map();
const state = createGatewayState({
  gatewayId,
  serviceId: "hwlab-gateway",
  endpoint: process.env.HWLAB_GATEWAY_ENDPOINT ?? `http://127.0.0.1:${port}`,
  boxes: [resourceId]
});
state.session.gatewaySessionId = gatewaySessionId;
state.session.status = "connected";
state.registry = {
  gatewayId,
  gatewaySessionId,
  resourceId,
  boxId,
  capabilities: shellCapabilities()
};
state.outbound = {
  enabled: Boolean(cloudUrl),
  cloudUrl,
  pollIntervalMs,
  commandExecutionEnabled,
  maxInflightRequests,
  inflightCount: 0,
  inflightRequests: [],
  lastPollAt: null,
  lastPollError: null,
  lastResultAt: null,
  lastResultError: null,
  lastBusyAt: null
};

const routes = new Map();

routes.set("GET /health/live", () => ({
  ...createHealth({ serviceId: "hwlab-gateway", name: gatewayId }),
  gatewayId,
  gatewaySessionId,
  outboundConnected: Boolean(state.outbound.lastPollAt && !state.outbound.lastPollError)
}));
routes.set("GET /status", () => ({
  ...state,
  note: "Outbound poll demo gateway; cloud dispatches through a gateway-initiated connection."
}));
routes.set("GET /capabilities", () => ({
  serviceId: "hwlab-gateway",
  gatewayId,
  gatewaySessionId,
  resourceId,
  boxId,
  capabilities: shellCapabilities()
}));

listen(createJsonServer({ routes }), port);

if (cloudUrl) {
  ensureNoProxyForLocalCloud(cloudUrl);
  schedulePoll(0);
}

function schedulePoll(delayMs = pollIntervalMs) {
  setTimeout(async () => {
    await pollCloudOnce();
    schedulePoll();
  }, delayMs).unref?.();
}

async function pollCloudOnce() {
  try {
    const response = await fetchJson(new URL("/v1/gateway/poll", ensureTrailingSlash(cloudUrl)), {
      method: "POST",
      body: registrationPayload()
    });
    if (!response.ok) {
      throw new Error(`cloud poll failed: HTTP ${response.status}`);
    }
    state.outbound.lastPollAt = new Date().toISOString();
    state.outbound.lastPollError = null;
    state.session.lastSeenAt = state.outbound.lastPollAt;
    if (response.body?.request) {
      startCloudRequest(response.body.request);
    }
  } catch (error) {
    state.outbound.lastPollError = error instanceof Error ? error.message : String(error);
  }
}

function startCloudRequest(request) {
  const requestId = String(request?.id ?? `req_gateway_${Date.now()}`);
  if (inflightRequests.size >= maxInflightRequests) {
    state.outbound.lastBusyAt = new Date().toISOString();
    postGatewayBusyResult({
      ...request,
      id: requestId
    }).catch((error) => {
      state.outbound.lastResultError = error instanceof Error ? error.message : String(error);
    });
    return;
  }

  const startedAt = new Date().toISOString();
  inflightRequests.set(requestId, {
    requestId,
    method: request?.method ?? null,
    operationId: request?.params?.operationId ?? null,
    traceId: request?.meta?.traceId ?? null,
    startedAt
  });
  refreshInflightState();

  Promise.resolve()
    .then(() => handleCloudRequest({
      ...request,
      id: requestId
    }))
    .catch((error) => {
      state.outbound.lastResultError = error instanceof Error ? error.message : String(error);
    })
    .finally(() => {
      inflightRequests.delete(requestId);
      refreshInflightState();
    });
}

async function handleCloudRequest(request) {
  const environment = requestEnvironment(request?.meta);
  const context = operationContext({
    operationId: request.params?.operationId,
    traceId: request.meta?.traceId,
    requestId: request.id,
    actorId: request.meta?.actorId ?? "svc_hwlab-cloud-api"
  }, {
    serviceId: "hwlab-gateway",
    actorId: "svc_hwlab-gateway"
  });

  let response;
  try {
    if (request.method !== "hardware.invoke.shell") {
      throw rpcError(ERROR_CODES.methodNotFound, `Gateway method ${request.method ?? "unknown"} is not implemented`, {
        method: request.method ?? null
      });
    }
    const result = await invokeShell(request.params ?? {}, context, environment);
    response = {
      jsonrpc: JSON_RPC_VERSION,
      id: request.id,
      result,
      meta: responseMeta(context.traceId, environment)
    };
  } catch (error) {
    const body = error?.body ?? createRpcErrorBody({
      code: error?.code ?? ERROR_CODES.hwlabUnknown,
      message: error instanceof Error ? error.message : String(error),
      data: error?.data ?? {}
    });
    response = {
      jsonrpc: JSON_RPC_VERSION,
      id: request.id,
      error: body.error,
      meta: responseMeta(context.traceId, environment)
    };
  }

  await postGatewayResult(response);
}

async function postGatewayBusyResult(request) {
  const traceId = request?.meta?.traceId ?? `trc_gateway_busy_${Date.now()}`;
  const environment = requestEnvironment(request?.meta);
  const body = createRpcErrorBody({
    code: ERROR_CODES.operationRejected,
    message: "Gateway is busy; retry after an in-flight command finishes or increase HWLAB_GATEWAY_MAX_INFLIGHT.",
    data: {
      reason: "gateway_busy",
      operationId: request?.params?.operationId ?? null,
      inflightCount: inflightRequests.size,
      maxInflightRequests
    }
  });
  await postGatewayResult({
    jsonrpc: JSON_RPC_VERSION,
    id: request?.id ?? null,
    error: body.error,
    meta: responseMeta(traceId, environment)
  });
}

async function postGatewayResult(response) {
  const result = await fetchJson(new URL("/v1/gateway/result", ensureTrailingSlash(cloudUrl)), {
    method: "POST",
    body: {
      gatewayId,
      gatewaySessionId,
      response
    }
  });
  if (!result.ok) {
    throw new Error(`cloud result failed: HTTP ${result.status}`);
  }
  state.outbound.lastResultAt = new Date().toISOString();
  state.outbound.lastResultError = null;
}

async function invokeShell(params, context, environment = gatewayEnvironment) {
  if (!commandExecutionEnabled) {
    throw rpcError(ERROR_CODES.operationRejected, "Gateway command execution is disabled", {
      reason: "cmd_exec_disabled",
      requiredEnv: "HWLAB_GATEWAY_CMD_EXEC_ENABLED=1"
    });
  }

  const shellInput = normalizeShellInput(params);
  const startedAt = Date.now();
  const execution = await executeCommand(shellInput, {
    timeoutMs: shellInput.timeoutMs ?? commandTimeoutMs,
    outputLimitBytes
  });
  const durationMs = Date.now() - startedAt;
  const now = new Date().toISOString();
  const status = execution.timedOut ? "timed_out" : execution.exitCode === 0 ? "succeeded" : "failed";
  const audit = createAuditEvent({
    serviceId: "hwlab-gateway",
    action: "hardware.invoke.shell",
    targetType: "box_resource",
    targetId: params.resourceId ?? resourceId,
    projectId: params.projectId,
    gatewaySessionId,
    operationId: context.operationId,
    traceId: context.traceId,
    actorType: "service",
    actorId: "svc_hwlab-gateway",
    outcome: status === "succeeded" ? "succeeded" : "failed",
    environment,
    metadata: {
      gatewayId,
      resourceId: params.resourceId ?? resourceId,
      capabilityId: params.capabilityId ?? "cap_windows_cmd_exec",
      command: redactCommand(shellInput.command),
      cwd: execution.cwd,
      exitCode: execution.exitCode,
      timedOut: execution.timedOut,
      durationMs
    },
    occurredAt: now
  });
  const evidence = createEvidenceRecord({
    serviceId: "hwlab-gateway",
    projectId: params.projectId ?? "prj_mvp_topology",
    operationId: context.operationId,
    traceId: context.traceId,
    kind: "trace",
    environment,
    payload: {
      gatewayId,
      gatewaySessionId,
      resourceId: params.resourceId ?? resourceId,
      capabilityId: params.capabilityId ?? "cap_windows_cmd_exec",
      shell: execution
    },
    metadata: {
      gatewaySessionId,
      producerServiceId: "hwlab-gateway"
    },
    createdAt: now
  });

  state.lastDispatch = {
    operationId: context.operationId,
    traceId: context.traceId,
    resourceId: params.resourceId ?? resourceId,
    observedAt: now,
    accepted: true,
    status
  };
  state.updatedAt = now;

  return {
    accepted: true,
    status,
    serviceId: "hwlab-gateway",
    environment,
    gatewayId,
    gatewaySessionId,
    operationId: context.operationId,
    traceId: context.traceId,
    shellExecuted: true,
    dispatchStatus: status,
    ...execution,
    durationMs,
    auditId: audit.auditId,
    evidenceId: evidence.evidenceId,
    audit,
    evidence,
    gateway: {
      gatewayId,
      gatewaySessionId
    }
  };
}

function executeCommand(input, { timeoutMs, outputLimitBytes: limitBytes }) {
  return new Promise((resolve) => {
    const startedAt = new Date().toISOString();
    const shell = shellCommand(input.command);
    const child = spawn(shell.command, shell.args, {
      cwd: input.cwd,
      windowsHide: true,
      env: {
        ...process.env,
        PYTHONUTF8: process.env.PYTHONUTF8 ?? "1",
        PYTHONIOENCODING: process.env.PYTHONIOENCODING ?? "utf-8",
        NO_PROXY: process.env.NO_PROXY,
        no_proxy: process.env.no_proxy
      },
      stdio: ["ignore", "pipe", "pipe"]
    });
    const stdout = createLimitedCollector(limitBytes);
    const stderr = createLimitedCollector(limitBytes);
    let timedOut = false;
    let settled = false;
    const timer = setTimeout(() => {
      timedOut = true;
      killProcessTree(child);
    }, timeoutMs);

    child.stdout.on("data", (chunk) => stdout.push(chunk));
    child.stderr.on("data", (chunk) => stderr.push(chunk));
    child.on("error", (error) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      resolve({
        shell: shell.command,
        command: input.command,
        cwd: input.cwd ?? process.cwd(),
        startedAt,
        completedAt: new Date().toISOString(),
        exitCode: null,
        signal: null,
        timedOut,
        stdout: stdout.text(),
        stderr: stderr.text() || error.message,
        stdoutTruncated: stdout.truncated,
        stderrTruncated: stderr.truncated
      });
    });
    child.on("close", (exitCode, signal) => {
      if (settled) return;
      settled = true;
      clearTimeout(timer);
      resolve({
        shell: shell.command,
        command: input.command,
        cwd: input.cwd ?? process.cwd(),
        startedAt,
        completedAt: new Date().toISOString(),
        exitCode,
        signal,
        timedOut,
        stdout: stdout.text(),
        stderr: stderr.text(),
        stdoutTruncated: stdout.truncated,
        stderrTruncated: stderr.truncated
      });
    });
  });
}

function shellCommand(command) {
  if (process.platform === "win32") {
    return {
      command: process.env.ComSpec ?? "cmd.exe",
      args: ["/d", "/s", "/c", String(command)]
    };
  }
  return {
    command: "sh",
    args: ["-lc", String(command)]
  };
}

function killProcessTree(child) {
  if (!child.pid) return;
  if (process.platform === "win32") {
    spawn("taskkill", ["/pid", String(child.pid), "/T", "/F"], {
      windowsHide: true,
      stdio: "ignore"
    });
    return;
  }
  child.kill("SIGTERM");
}

function createLimitedCollector(limitBytes) {
  const chunks = [];
  let size = 0;
  let truncated = false;
  return {
    get truncated() {
      return truncated;
    },
    push(chunk) {
      if (size >= limitBytes) {
        truncated = true;
        return;
      }
      const buffer = Buffer.from(chunk);
      const available = limitBytes - size;
      if (buffer.byteLength > available) {
        chunks.push(buffer.subarray(0, available));
        size += available;
        truncated = true;
        return;
      }
      chunks.push(buffer);
      size += buffer.byteLength;
    },
    text() {
      return Buffer.concat(chunks).toString("utf8");
    }
  };
}

function registrationPayload() {
  refreshInflightState();
  return {
    serviceId: "hwlab-gateway",
    projectId: process.env.HWLAB_PROJECT_ID ?? "prj_mvp_topology",
    gatewayId,
    gatewaySessionId,
    endpoint: state.session.endpoint,
    resourceId,
    boxId,
    capabilities: shellCapabilities(),
    outbound: {
      pollIntervalMs,
      commandExecutionEnabled,
      maxInflightRequests,
      inflightCount: inflightRequests.size,
      inflightRequests: snapshotInflightRequests()
    },
    system: {
      platform: process.platform,
      arch: process.arch,
      hostname: os.hostname(),
      release: os.release()
    }
  };
}

function shellCapabilities() {
  return [
    {
      capabilityId: process.env.HWLAB_GATEWAY_CMD_CAPABILITY_ID ?? "cap_windows_cmd_exec",
      resourceId,
      projectId: process.env.HWLAB_PROJECT_ID ?? "prj_mvp_topology",
      name: "shell.exec",
      description: "Execute a local shell command through an outbound HWLAB gateway demo connection.",
      direction: "bidirectional",
      valueType: "object",
      constraints: {
        gatewayOutboundOnly: true,
        demoOpen: truthy(process.env.HWLAB_GATEWAY_DEMO_OPEN),
        maxInflightRequests
      },
      mutatesState: true
    }
  ];
}

function refreshInflightState() {
  state.outbound.inflightCount = inflightRequests.size;
  state.outbound.inflightRequests = snapshotInflightRequests();
}

function snapshotInflightRequests() {
  const now = Date.now();
  return [...inflightRequests.values()].map((request) => ({
    ...request,
    durationMs: durationSince(request.startedAt, now)
  }));
}

function durationSince(value, now = Date.now()) {
  const started = Date.parse(value ?? "");
  return Number.isFinite(started) ? Math.max(0, now - started) : 0;
}

function normalizeShellInput(params = {}) {
  const input = params.input && typeof params.input === "object" && !Array.isArray(params.input) ? params.input : {};
  const command = params.command ?? input.command ?? input.shell?.command;
  if (!command) {
    throw rpcError(ERROR_CODES.invalidParams, "hardware.invoke.shell requires input.command", {
      required: ["input.command"]
    });
  }
  return {
    command: Array.isArray(command) ? command.map(String).join(" ") : String(command),
    cwd: input.cwd ? String(input.cwd) : undefined,
    timeoutMs: parsePositiveInteger(input.timeoutMs, commandTimeoutMs)
  };
}

function responseMeta(traceId, environment = gatewayEnvironment) {
  return {
    traceId,
    serviceId: "hwlab-gateway",
    environment: normalizeEnvironment(environment)
  };
}

function requestEnvironment(meta = {}) {
  return normalizeEnvironment(meta?.environment || gatewayEnvironment);
}

function normalizeEnvironment(environment) {
  return isProtocolEnvironment(environment) ? environment : ENVIRONMENT_DEV;
}

function rpcError(code, message, data = {}) {
  const error = new Error(message);
  error.code = code;
  error.data = data;
  error.body = createRpcErrorBody({ code, message, data });
  return error;
}

function parsePositiveInteger(value, fallback) {
  const parsed = Number.parseInt(value ?? "", 10);
  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}

function truthy(value) {
  return ["1", "true", "yes", "on", "enabled"].includes(String(value ?? "").trim().toLowerCase());
}

function trimUrl(value) {
  const text = String(value ?? "").trim();
  return text ? text : null;
}

function ensureTrailingSlash(url) {
  return url.endsWith("/") ? url : `${url}/`;
}

function redactCommand(command) {
  const text = String(command ?? "");
  return text.length <= 200 ? text : `${text.slice(0, 200)}...`;
}

function ensureNoProxyForLocalCloud(url) {
  let host = "";
  try {
    host = new URL(url).hostname;
  } catch {
    return;
  }
  const additions = new Set(["localhost", "127.0.0.1", "::1", host].filter(Boolean));
  for (const key of ["NO_PROXY", "no_proxy"]) {
    const current = String(process.env[key] ?? "");
    const values = new Set(current.split(",").map((item) => item.trim()).filter(Boolean));
    for (const addition of additions) values.add(addition);
    process.env[key] = [...values].join(",");
  }
}