pikasTech-HWLAB/cmd/hwlab-device-pod/main.ts

#!/usr/bin/env node
import { createServer } from "node:http";
import { randomUUID } from "node:crypto";

import { jsonResponse, listen, parsePort, readJson } from "../../internal/sim/http.mjs";

const SERVICE_ID = "hwlab-device-pod";
const CONTRACT_VERSION = "device-pod-executor-v1";
const DEFAULT_DEVICE_POD_ID = "device-pod-71-freq";
const port = parsePort(process.env.HWLAB_DEVICE_POD_PORT, parsePort(process.env.PORT, 7601));
const devicePodId = process.env.HWLAB_DEVICE_POD_ID || DEFAULT_DEVICE_POD_ID;
const environment = process.env.HWLAB_ENVIRONMENT || process.env.HWLAB_GITOPS_PROFILE || "dev";
const cloudApiInternalUrl = normalizeBaseUrl(process.env.HWLAB_CLOUD_API_INTERNAL_URL || process.env.HWLAB_CLOUD_API_URL);
const internalToken = textOr(process.env.HWLAB_DEVICE_POD_INTERNAL_TOKEN, "");
const dispatchTimeoutMs = numberOr(process.env.HWLAB_DEVICE_POD_GATEWAY_DISPATCH_TIMEOUT_MS, 120000);
const DEVICE_JOB_OUTPUT_MAX_BYTES = 12000;
const jobs = new Map();

listen(createServer(async (request, response) => {
  try {
    const url = new URL(request.url ?? "/", "http://localhost");
    if (request.method === "GET" && (url.pathname === "/health" || url.pathname === "/health/live")) {
      jsonResponse(response, 200, healthPayload());
      return;
    }
    if (url.pathname === "/health" || url.pathname === "/health/live") {
      jsonResponse(response, 405, { error: "method_not_allowed", method: request.method ?? "UNKNOWN", path: url.pathname });
      return;
    }
    if (request.method === "GET" && url.pathname === "/v1/device-pods") {
      jsonResponse(response, 200, listPayload());
      return;
    }
    const route = parseDevicePodPath(url.pathname);
    if (!route) {
      jsonResponse(response, 404, { error: "not_found", path: url.pathname });
      return;
    }
    if (request.method === "GET" && route.subpath === "status") {
      jsonResponse(response, 200, statusPayload(route.devicePodId));
      return;
    }
    if (request.method === "GET" && route.subpath === "events") {
      jsonResponse(response, 200, eventsPayload(route.devicePodId, url.searchParams));
      return;
    }
    if (request.method === "POST" && route.subpath === "jobs") {
      await handleInternalJob(request, response, route.devicePodId);
      return;
    }
    if (request.method === "GET" && route.subpath.startsWith("jobs/")) {
      handleGetInternalJob(request, response, route.devicePodId, route.subpath);
      return;
    }
    if (request.method === "POST" && route.subpath.startsWith("jobs/") && route.subpath.endsWith("/cancel")) {
      handleCancelInternalJob(request, response, route.devicePodId, route.subpath);
      return;
    }
    jsonResponse(response, 405, { error: "method_not_allowed", method: request.method ?? "UNKNOWN", path: url.pathname });
  } catch (error) {
    jsonResponse(response, 500, { error: "internal_error", message: error?.message ?? String(error), valuesRedacted: true });
  }
}), port);

function healthPayload() {
  return {
    serviceId: SERVICE_ID,
    environment,
    status: "live",
    ready: true,
    contractVersion: CONTRACT_VERSION,
    devicePodId,
    observedAt: new Date().toISOString(),
    role: "device-pod-internal-executor",
    authority: "hwlab-cloud-api",
    acceptsUserAuthority: false,
    fake: false,
    gatewayAdapter: gatewayAdapterState(),
    source: sourcePayload(),
    note: "Profile, grant, lease, and user-facing job authority live in hwlab-cloud-api; this service only exposes the internal executor boundary."
  };
}

function listPayload() {
  return {
    serviceId: SERVICE_ID,
    contractVersion: CONTRACT_VERSION,
    status: "ok",
    source: sourcePayload(),
    selectedDevicePodId: devicePodId,
    devicePods: [{
      devicePodId,
      status: "executor-ready",
      authority: "hwlab-cloud-api",
      fake: false,
      gatewayAdapter: gatewayAdapterState(),
      routes: {
        cloudAuthority: "/v1/device-pods",
        internalJob: `/v1/device-pods/${encodeURIComponent(devicePodId)}/jobs`
      }
    }]
  };
}

function statusPayload(id) {
  return {
    serviceId: SERVICE_ID,
    contractVersion: CONTRACT_VERSION,
    status: "blocked",
    devicePodId: id,
    observedAt: new Date().toISOString(),
    source: sourcePayload(),
    summary: {
      devicePodId: id,
      status: "blocked",
      blocker: gatewayAdapterBlocker("device-host-cli dispatch is owned by hwlab-cloud-api gateway routing")
    }
  };
}

function eventsPayload(id, searchParams) {
  const limit = Math.min(Math.max(Number.parseInt(searchParams.get("limit") ?? "80", 10) || 80, 1), 1000);
  const event = {
    eventId: `evt_${id}_executor_boundary`,
    devicePodId: id,
    ts: new Date().toISOString(),
    level: "warn",
    scope: "executor",
    status: "blocked",
    summary: "No standalone fake device events are emitted by hwlab-device-pod; use hwlab-cloud-api job authority.",
    blocker: gatewayAdapterBlocker("standalone device-pod executor has no gateway session")
  };
  return {
    serviceId: SERVICE_ID,
    contractVersion: CONTRACT_VERSION,
    status: "ok",
    devicePodId: id,
    source: sourcePayload(),
    events: limit > 0 ? [event] : [],
    truncation: { limit, returned: limit > 0 ? 1 : 0, truncated: false }
  };
}

async function handleInternalJob(request, response, id) {
  if (!isInternalCaller(request)) {
    jsonResponse(response, 403, {
      accepted: false,
      status: "blocked",
      error: {
        code: "cloud_api_authority_required",
        message: "Device Pod jobs must be authorized by hwlab-cloud-api before reaching the internal executor."
      },
      source: sourcePayload()
    });
    return;
  }
  const body = await readJson(request);
  const traceId = typeof body.traceId === "string" && body.traceId.startsWith("trc_") ? body.traceId : `trc_devicepod_${randomUUID()}`;
  const operationId = typeof body.operationId === "string" ? body.operationId : `op_devicepod_${randomUUID()}`;
  const now = new Date().toISOString();
  const profile = normalizeObject(body.profile);
  const route = normalizeObject(profile.route);
  const jobId = typeof body.jobId === "string" && body.jobId.startsWith("job_") ? body.jobId : `job_devicepod_${randomUUID()}`;
  const command = deviceHostCommand(profile, {
    id: jobId,
    devicePodId: id,
    intent: typeof body.intent === "string" ? body.intent : "unknown",
    args: normalizeObject(body.args)
  });
  const dispatchBlocker = gatewayDispatchPreflightBlocker(route, command);
  const job = {
    id: jobId,
    devicePodId: id,
    status: dispatchBlocker ? "blocked" : "running",
    intent: typeof body.intent === "string" ? body.intent : "unknown",
    reason: typeof body.reason === "string" ? body.reason : "",
    traceId,
    operationId,
    createdAt: now,
    updatedAt: now,
    completedAt: dispatchBlocker ? now : null,
    targetId: typeof body.targetId === "string" ? body.targetId : targetIdFromProfile(profile),
    profileHash: typeof body.profileHash === "string" ? body.profileHash : "",
    ownerUserId: typeof body.ownerUserId === "string" ? body.ownerUserId : "",
    command,
    output: dispatchBlocker ? boundedOutput({ text: "", summary: dispatchBlocker.summary }) : boundedOutput({ text: "", summary: "gateway/device-host-cli dispatch queued" }),
    blocker: dispatchBlocker,
    source: sourcePayload()
  };
  jobs.set(jobKey(id, job.id), job);
  if (!dispatchBlocker) dispatchGatewayJob({ job, route });
  jsonResponse(response, dispatchBlocker ? 409 : 202, jobPayload(job, { accepted: !dispatchBlocker }));
}

function handleGetInternalJob(request, response, id, subpath) {
  if (!isInternalCaller(request)) return rejectExternalJobRoute(response);
  const parts = subpath.split("/");
  const job = jobs.get(jobKey(id, decodeURIComponent(parts[1] ?? "")));
  if (!job) {
    jsonResponse(response, 404, { error: { code: "device_job_not_found", message: "Device Pod executor job was not found" }, source: sourcePayload() });
    return;
  }
  if (parts[2] === "output") {
    jsonResponse(response, 200, jobOutputPayload(job));
    return;
  }
  jsonResponse(response, 200, jobPayload(job));
}

function handleCancelInternalJob(request, response, id, subpath) {
  if (!isInternalCaller(request)) return rejectExternalJobRoute(response);
  const jobId = decodeURIComponent(subpath.split("/")[1] ?? "");
  const job = jobs.get(jobKey(id, jobId));
  if (!job) {
    jsonResponse(response, 404, { error: { code: "device_job_not_found", message: "Device Pod executor job was not found" }, source: sourcePayload() });
    return;
  }
  if (terminalJobStatus(job.status)) {
    jsonResponse(response, 200, jobPayload(job));
    return;
  }
  const now = new Date().toISOString();
  const canceled = { ...job, status: "canceled", updatedAt: now, completedAt: now, blocker: { code: "device_job_canceled", layer: "device-pod", retryable: false, summary: "Device Pod executor job was canceled by cloud-api" } };
  jobs.set(jobKey(id, jobId), canceled);
  jsonResponse(response, 200, jobPayload(canceled));
}

async function dispatchGatewayJob({ job, route }) {
  const target = `${cloudApiInternalUrl}/v1/internal/device-pod/gateway-dispatch`;
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), dispatchTimeoutMs + 10000);
  try {
    const response = await fetch(target, {
      method: "POST",
      signal: controller.signal,
      headers: {
        accept: "application/json",
        "content-type": "application/json",
        "x-hwlab-internal-service": SERVICE_ID,
        "x-hwlab-internal-token": internalToken,
        "x-trace-id": job.traceId,
        "x-request-id": `req_${job.id}`
      },
      body: JSON.stringify({
        id: `req_${job.id}`,
        actorId: job.ownerUserId || "svc_hwlab-device-pod",
        params: {
          projectId: "prj_v02_device_pod",
          gatewaySessionId: route.gatewaySessionId,
          resourceId: route.resourceId ?? "res_device_pod",
          capabilityId: route.capabilityId ?? "cap_device_host_cli",
          operationId: job.operationId,
          traceId: job.traceId,
          input: {
            command: job.command,
            cwd: route.hostWorkspaceRoot,
            timeoutMs: dispatchTimeoutMs
          }
        }
      })
    });
    const payload = await response.json().catch(() => ({}));
    const current = jobs.get(jobKey(job.devicePodId, job.id));
    if (current && terminalJobStatus(current.status)) return;
    const next = jobFromGatewayDispatch(job, payload, response.status);
    jobs.set(jobKey(job.devicePodId, job.id), next);
  } catch (error) {
    const current = jobs.get(jobKey(job.devicePodId, job.id));
    if (current && terminalJobStatus(current.status)) return;
    const now = new Date().toISOString();
    jobs.set(jobKey(job.devicePodId, job.id), {
      ...job,
      status: "failed",
      updatedAt: now,
      completedAt: now,
      output: boundedOutput({ text: "", error: error?.message ?? "gateway dispatch failed" }),
      blocker: gatewayDispatchFailedBlocker(error?.message ?? "gateway dispatch failed")
    });
  } finally {
    clearTimeout(timer);
  }
}

function jobPayload(job, { accepted = !["blocked", "failed"].includes(job.status) } = {}) {
  return {
    accepted,
    status: job.status,
    contractVersion: CONTRACT_VERSION,
    devicePodId: job.devicePodId,
    targetId: job.targetId ?? null,
    profileHash: job.profileHash ?? "",
    traceId: job.traceId,
    operationId: job.operationId,
    job: publicJob(job),
    blocker: job.blocker,
    freshness: freshness(job.updatedAt, job.blocker),
    outputUrl: `/v1/device-pods/${encodeURIComponent(job.devicePodId)}/jobs/${encodeURIComponent(job.id)}/output`,
    cancelUrl: `/v1/device-pods/${encodeURIComponent(job.devicePodId)}/jobs/${encodeURIComponent(job.id)}/cancel`,
    source: job.source ?? sourcePayload()
  };
}

function rejectExternalJobRoute(response) {
  jsonResponse(response, 403, {
    accepted: false,
    status: "blocked",
    error: {
      code: "cloud_api_authority_required",
      message: "Device Pod jobs must be authorized by hwlab-cloud-api before reaching the internal executor."
    },
    source: sourcePayload()
  });
}

function parseDevicePodPath(pathname) {
  const prefix = "/v1/device-pods/";
  if (!pathname.startsWith(prefix)) return null;
  const [id, ...rest] = pathname.slice(prefix.length).split("/").filter(Boolean).map((part) => decodeURIComponent(part));
  return id ? { devicePodId: id, subpath: rest.join("/") } : null;
}

function isInternalCaller(request) {
  if (!internalToken) return false;
  const serviceHeader = request.headers["x-hwlab-internal-service"];
  const tokenHeader = request.headers["x-hwlab-internal-token"];
  return String(Array.isArray(serviceHeader) ? serviceHeader[0] : serviceHeader ?? "").trim() === "hwlab-cloud-api"
    && String(Array.isArray(tokenHeader) ? tokenHeader[0] : tokenHeader ?? "").trim() === internalToken;
}

function sourcePayload() {
  return { kind: "INTERNAL_EXECUTOR", serviceId: SERVICE_ID, authority: "hwlab-cloud-api", fake: false, devLiveEvidence: false };
}

function gatewayAdapterBlocker(summary) {
  return { code: "gateway_dispatch_unavailable", layer: "device-pod", retryable: true, summary, userMessage: "Device Pod executor boundary is present, but gateway/device-host-cli dispatch is not connected here." };
}

function gatewayDispatchFailedBlocker(summary) {
  return { code: "gateway_dispatch_failed", layer: "device-pod", retryable: true, summary, userMessage: "Device Pod gateway/device-host-cli dispatch failed." };
}

function gatewayDispatchPreflightBlocker(route, command) {
  if (!cloudApiInternalUrl) return gatewayAdapterBlocker("HWLAB_CLOUD_API_INTERNAL_URL is not configured for gateway dispatch");
  if (!route.gatewaySessionId) return gatewayAdapterBlocker("profile route.gatewaySessionId is missing");
  if (!command) return gatewayAdapterBlocker("device-host-cli command mapping is not available for this intent");
  return null;
}

function gatewayAdapterState() {
  return { mode: cloudApiInternalUrl ? "cloud-api-internal-dispatch" : "blocked", cloudApiInternalUrlConfigured: Boolean(cloudApiInternalUrl) };
}

function jobFromGatewayDispatch(job, payload, httpStatus) {
  const result = normalizeObject(payload.result);
  const dispatch = normalizeObject(result.dispatch);
  const blocker = payload.blocker ?? result.blocker ?? (payload.error ? gatewayDispatchFailedBlocker(payload.error.message ?? "gateway dispatch failed") : null);
  const completed = !blocker && httpStatus < 400 && (result.status === "completed" || result.status === "succeeded" || dispatch.dispatchStatus === "succeeded" || dispatch.dispatchStatus === "completed");
  const status = completed ? "completed" : blocker?.code === "gateway_dispatch_unavailable" ? "blocked" : "failed";
  const now = new Date().toISOString();
  return {
    ...job,
    status,
    updatedAt: now,
    completedAt: now,
    output: boundedOutput({
      text: gatewayDispatchText(result, dispatch),
      httpStatus,
      dispatch,
      gateway: result.gateway ?? null,
      auditId: result.auditId ?? null,
      evidenceId: result.evidenceId ?? null,
      summary: completed ? "gateway/device-host-cli dispatch completed" : blocker?.summary ?? "gateway/device-host-cli dispatch failed"
    }),
    blocker: completed ? null : blocker ?? gatewayDispatchFailedBlocker(dispatch.message ?? dispatch.stderr ?? result.status ?? "gateway dispatch failed")
  };
}

function gatewayDispatchText(result, dispatch) {
  if (typeof dispatch.stdout === "string" && dispatch.stdout) return dispatch.stdout;
  if (typeof dispatch.stderr === "string" && dispatch.stderr) return dispatch.stderr;
  if (typeof result.text === "string") return result.text;
  return JSON.stringify(result);
}

function deviceHostCommand(profile, job) {
  const args = deviceHostArgs(job.intent, job.args);
  if (!args) return null;
  const payload = {
    devicePodId: profile.devicePodId || job.devicePodId,
    profile,
    hostCli: profile.route?.hostCli || "node tools/device-host-cli.mjs",
    args
  };
  const profileJsonB64 = Buffer.from(JSON.stringify(payload.profile), "utf8").toString("base64");
  return shellJoin([...splitCommand(payload.hostCli), "--profile-json-b64", profileJsonB64, "--pod-id", payload.devicePodId, ...payload.args]);
}

function splitCommand(value) {
  const out = [];
  let quote = null;
  let buffer = "";
  for (const char of String(value || "")) {
    if (quote) {
      if (char === quote) quote = null;
      else buffer += char;
    } else if (char === "'" || char === '"') {
      quote = char;
    } else if (/\s/u.test(char)) {
      if (buffer) {
        out.push(buffer);
        buffer = "";
      }
    } else {
      buffer += char;
    }
  }
  if (buffer) out.push(buffer);
  return out.length ? out : ["node", "tools/device-host-cli.mjs"];
}

function shellJoin(parts) {
  return parts.map(shellQuote).join(" ");
}

function shellQuote(value) {
  const text = String(value ?? "");
  if (/^[A-Za-z0-9_./:=@\\-]+$/u.test(text)) return text;
  return `"${text.replace(/(["\\])/gu, "\\$1")}"`;
}

function deviceHostArgs(intent, args = {}) {
  if (intent === "workspace.ls") return ["workspace", "ls", textOr(args.path, ".")];
  if (intent === "workspace.cat") {
    const path = textOr(args.path, "");
    return path ? ["workspace", "cat", path, "--limit", String(numberOr(args.limit ?? args.maxBytes, 40000))] : null;
  }
  if (intent === "workspace.rg") {
    const pattern = textOr(args.pattern ?? args.query, "");
    return pattern ? ["workspace", "rg", pattern, textOr(args.path, "."), ...hostOptionArgs(args, ["glob", "g", "filesWithMatches", "l", "ignoreCase", "i", "maxCount"])] : null;
  }
  if (intent === "workspace.apply-patch") return ["workspace", "apply-patch", textOr(args.base, "."), "--patch-b64", patchBase64(args)];
  if (intent === "workspace.build") return ["workspace", "build", textOr(args.action, "start"), ...hostOptionArgs(args, ["target", "timeoutMs"]), ...jobIdArgs(args)];
  if (intent === "debug.status") return ["debug-probe", "status"];
  if (intent === "debug.chip-id") return ["debug-probe", "chip-id"];
  if (intent === "debug.download") return ["debug-probe", "download", textOr(args.action, "start"), ...hostOptionArgs(args, ["target", "timeoutMs", "captureUart", "captureDurationMs", "durationMs", "port", "baudRate"]), ...jobIdArgs(args)];
  if (intent === "debug.reset") return ["debug-probe", "reset"];
  if (intent === "io.ports") return ["io-probe", textOr(args.uartId, "uart/1"), "ports"];
  if (intent === "io.uart.read") return ["io-probe", textOr(args.uartId, "uart/1"), "read", "--duration-ms", String(numberOr(args.durationMs ?? args["duration-ms"], 1000)), ...hostOptionArgs(args, ["port", "baudRate"])] ;
  if (intent === "io.uart.read-after-launch-flash") return ["io-probe", textOr(args.uartId, "uart/1"), "read-after-launch-flash", ...hostOptionArgs(args, ["durationMs", "port", "baudRate", "flashBase", "skipHardwareReset", "connectMode", "timeoutMs"])] ;
  if (intent === "io.uart.write") return ["io-probe", textOr(args.uartId, "uart/1"), "write", ...(args.hex ? ["--hex"] : []), ...hostOptionArgs(args, ["port", "baudRate"]), textOr(args.message ?? args.text ?? args.data, "")];
  return null;
}

function jobIdArgs(args = {}) {
  const jobId = textOr(args.jobId, "");
  return jobId ? [jobId] : [];
}

function hostOptionArgs(args = {}, keys = []) {
  const out = [];
  for (const key of keys) {
    const value = args[key];
    if (value === undefined || value === null || value === "" || value === false) continue;
    const name = `--${key.replace(/[A-Z]/gu, (match) => `-${match.toLowerCase()}`)}`;
    if (value === true) out.push(name);
    else out.push(name, String(value));
  }
  return out;
}

function patchBase64(args = {}) {
  if (typeof args.patchB64 === "string") return args.patchB64;
  if (typeof args["patch-b64"] === "string") return args["patch-b64"];
  return Buffer.from(String(args.patch ?? ""), "utf8").toString("base64");
}

function targetIdFromProfile(profile = {}) {
  return profile.target?.id ?? profile.targetId ?? null;
}

function normalizeObject(value) {
  return value && typeof value === "object" && !Array.isArray(value) ? value : {};
}

function textOr(value, fallback) {
  return typeof value === "string" && value.trim() ? value : fallback;
}

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

function normalizeBaseUrl(value) {
  const text = String(value ?? "").trim();
  return text ? text.replace(/\/+$/u, "") : "";
}

function jobOutputPayload(job) {
  return {
    ...jobPayload(job),
    output: job.output,
    text: job.output.text,
    bytes: job.output.bytes,
    truncation: job.output.truncation
  };
}

function publicJob(job) {
  return { id: job.id, devicePodId: job.devicePodId, status: job.status, intent: job.intent, reason: job.reason, traceId: job.traceId, operationId: job.operationId, createdAt: job.createdAt, updatedAt: job.updatedAt, completedAt: job.completedAt };
}

function boundedOutput(output, maxBytes = DEVICE_JOB_OUTPUT_MAX_BYTES) {
  const text = String(output?.text ?? "");
  const buffer = Buffer.from(text, "utf8");
  const clipped = buffer.length > maxBytes;
  const boundedText = clipped ? buffer.subarray(0, maxBytes).toString("utf8") : text;
  const bounded = { ...output, text: boundedText };
  if (clipped) {
    delete bounded.dispatch;
    bounded.omitted = { reason: "device_job_output_truncated", originalBytes: buffer.length };
  }
  return { ...bounded, bytes: Math.min(buffer.length, maxBytes), truncation: { maxBytes, truncated: clipped, originalBytes: buffer.length } };
}

function freshness(observedAt, blocker) {
  return { observedAt, ageMs: 0, stale: Boolean(blocker), source: blocker ? "blocked" : "device-pod-executor" };
}

function terminalJobStatus(status) {
  return ["completed", "failed", "blocked", "canceled"].includes(status);
}

function jobKey(id, jobId) {
  return `${id}\u0000${jobId}`;
}