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

import assert from "node:assert/strict";
import { spawn } from "node:child_process";
import { createServer } from "node:http";
import test from "node:test";

const bunCommand = process.env.HWLAB_TEST_BUN_COMMAND || process.env.HWLAB_BUN_COMMAND || (process.versions.bun ? process.execPath : "bun");
const INTERNAL_TOKEN = "test-internal-token";

test("device pod executor exposes cloud-api authority boundary and method guards", async () => {
  const service = await startDevicePod("device-pod-test");
  try {
    const health = await fetchJson(`http://127.0.0.1:${service.port}/health/live`);
    assert.equal(health.serviceId, "hwlab-device-pod");
    assert.equal(health.status, "live");
    assert.equal(health.devicePodId, "device-pod-test");
    assert.equal(health.contractVersion, "device-pod-executor-v1");
    assert.equal(health.role, "device-pod-internal-executor");
    assert.equal(health.authority, "hwlab-cloud-api");
    assert.equal(health.fake, false);

    const list = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods`);
    assert.equal(list.serviceId, "hwlab-device-pod");
    assert.equal(list.selectedDevicePodId, "device-pod-test");
    assert.equal(list.devicePods.length, 1);
    assert.equal(list.source.fake, false);

    const status = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/status`);
    assert.equal(status.devicePodId, "device-pod-test");
    assert.equal(status.status, "blocked");
    assert.equal(status.summary.blocker.code, "gateway_dispatch_unavailable");
    assert.equal(status.targetId, null);
    assert.equal(status.profileHash, "");
    assert.match(status.traceId, /^trc_devicepod_/u);
    assert.match(status.operationId, /^op_devicepod_/u);
    assert.equal(status.freshness.stale, true);
    assert.equal(status.truncation.truncated, false);
    assert.match(status.output.summary, /device-host-cli dispatch is owned/u);

    const events = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/events?limit=1`);
    assert.equal(events.events.length, 1);
    assert.equal(events.truncation.limit, 1);
    assert.equal(events.truncation.truncated, false);
    assert.equal(events.events[0].blocker.code, "gateway_dispatch_unavailable");

    const externalJob = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs`, {
      method: "POST",
      headers: { "content-type": "application/json" },
      body: JSON.stringify({ intent: "workspace.ls" })
    });
    assert.equal(externalJob.status, 403);
    assert.equal((await externalJob.json()).error.code, "cloud_api_authority_required");

    const headerOnlyJob = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs`, {
      method: "POST",
      headers: { "content-type": "application/json", "x-hwlab-internal-service": "hwlab-cloud-api" },
      body: JSON.stringify({ intent: "workspace.ls" })
    });
    assert.equal(headerOnlyJob.status, 403);
    assert.equal((await headerOnlyJob.json()).error.code, "cloud_api_authority_required");

    const internalJob = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs`, {
      method: "POST",
      headers: internalHeaders({ "content-type": "application/json" }),
      body: JSON.stringify({ jobId: "job_devicepod_test", intent: "workspace.ls", traceId: "trc_device_pod_test", operationId: "op_device_pod_test" })
    });
    assert.equal(internalJob.status, 409);
    const internalJobPayload = await internalJob.json();
    assert.equal(internalJobPayload.blocker.code, "gateway_dispatch_unavailable");
    assert.equal(internalJobPayload.job.id, "job_devicepod_test");
    assert.equal(internalJobPayload.outputUrl, "/v1/device-pods/device-pod-test/jobs/job_devicepod_test/output");

    const storedJob = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test`, {
      headers: internalHeaders()
    });
    assert.equal(storedJob.job.id, "job_devicepod_test");
    assert.equal(storedJob.status, "blocked");

    const storedOutput = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test/output`, {
      headers: internalHeaders()
    });
    assert.equal(storedOutput.job.id, "job_devicepod_test");
    assert.equal(storedOutput.output.summary, "HWLAB_CLOUD_API_INTERNAL_URL is not configured for gateway dispatch");
    assert.equal(storedOutput.truncation.truncated, false);
    assert.equal(storedOutput.truncation.originalBytes, 0);

    const externalJobRead = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test`);
    assert.equal(externalJobRead.status, 403);
    assert.equal((await externalJobRead.json()).error.code, "cloud_api_authority_required");

    const cancelStored = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs/job_devicepod_test/cancel`, {
      method: "POST",
      headers: internalHeaders()
    });
    assert.equal(cancelStored.status, 200);
    assert.equal((await cancelStored.json()).job.id, "job_devicepod_test");

    const missing = await fetch(`http://127.0.0.1:${service.port}/missing`);
    assert.equal(missing.status, 404);
    assert.deepEqual(await missing.json(), { error: "not_found", path: "/missing" });

    const methodGuard = await fetch(`http://127.0.0.1:${service.port}/health`, { method: "POST" });
    assert.equal(methodGuard.status, 405);
    assert.deepEqual(await methodGuard.json(), { error: "method_not_allowed", method: "POST", path: "/health" });
  } finally {
    await service.stop();
  }
});

test("device pod executor lists and isolates multiple logical device pods", async () => {
  const service = await startDevicePod("device-pod-main", {
    HWLAB_DEVICE_POD_IDS: "device-pod-main, device-pod-alt"
  });

  try {
    const list = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods`);
    assert.equal(list.selectedDevicePodId, "device-pod-main");
    assert.deepEqual(list.devicePods.map((pod) => pod.devicePodId), ["device-pod-main", "device-pod-alt"]);
    assert.equal(list.devicePods[1].routes.internalJob, "/v1/device-pods/device-pod-alt/jobs");

    const mainJobResponse = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-main/jobs`, {
      method: "POST",
      headers: { "content-type": "application/json", ...internalHeaders() },
      body: JSON.stringify({ jobId: "job_multi_main", intent: "workspace.ls", traceId: "trc_multi_main", operationId: "op_multi_main" })
    });
    assert.equal(mainJobResponse.status, 409);
    const altJobResponse = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-alt/jobs`, {
      method: "POST",
      headers: { "content-type": "application/json", ...internalHeaders() },
      body: JSON.stringify({ jobId: "job_multi_alt", intent: "workspace.ls", traceId: "trc_multi_alt", operationId: "op_multi_alt" })
    });
    assert.equal(altJobResponse.status, 409);

    const mainJob = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-main/jobs/job_multi_main`, { headers: internalHeaders() });
    assert.equal(mainJob.job.devicePodId, "device-pod-main");
    const altJob = await fetchJson(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-alt/jobs/job_multi_alt`, { headers: internalHeaders() });
    assert.equal(altJob.job.devicePodId, "device-pod-alt");
    const crossRead = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-alt/jobs/job_multi_main`, { headers: internalHeaders() });
    assert.equal(crossRead.status, 404);
  } finally {
    await service.stop();
  }
});

test("device-pod executor bounds gateway output text", async () => {
  const longText = "z".repeat(13000);
  const cloudApi = createServer(async (request, response) => {
    const body = await requestJson(request);
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify({
      jsonrpc: "2.0",
      id: body.id,
      result: {
        accepted: true,
        status: "completed",
        dispatch: { shellExecuted: true, dispatchStatus: "succeeded", stdout: longText, exitCode: 0 }
      }
    }));
  });
  await new Promise((resolve) => cloudApi.listen(0, "127.0.0.1", resolve));
  const cloudPort = cloudApi.address().port;
  const service = await startDevicePod("device-pod-test", {
    HWLAB_CLOUD_API_INTERNAL_URL: `http://127.0.0.1:${cloudPort}`
  });

  try {
    const jobResponse = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs`, {
      method: "POST",
      headers: internalHeaders({ "content-type": "application/json" }),
      body: JSON.stringify({
        jobId: "job_bounded_output_test",
        intent: "workspace.ls",
        args: { path: "src" },
        traceId: "trc_bounded_output_test",
        operationId: "op_bounded_output_test",
        targetId: "target-device-pod-test",
        profileHash: "sha256:test",
        profile: {
          schemaVersion: 1,
          devicePodId: "device-pod-test",
          target: { id: "target-device-pod-test" },
          route: { gatewaySessionId: "gws_devicepod_test", resourceId: "res_devicepod_test", capabilityId: "cap_device_host_cli" }
        }
      })
    });
    assert.equal(jobResponse.status, 202);
    const output = await waitForJobStatus(service.port, "device-pod-test", "job_bounded_output_test", "completed");
    assert.equal(output.output.text.length, 12000);
    assert.equal(output.text.length, 12000);
    assert.equal(output.bytes, 12000);
    assert.equal(output.truncation.truncated, true);
    assert.equal(output.truncation.originalBytes, 13000);
    assert.equal(JSON.stringify(output).includes(longText), false);
  } finally {
    await service.stop();
    await new Promise((resolve, reject) => cloudApi.close((error) => (error ? reject(error) : resolve())));
  }
});

test("device pod executor dispatches internal jobs through cloud-api gateway adapter", async () => {
  const dispatches = [];
  const cloudApi = createServer(async (request, response) => {
    const body = await requestJson(request);
    dispatches.push({ method: request.method, url: request.url, internalService: request.headers["x-hwlab-internal-service"], internalToken: request.headers["x-hwlab-internal-token"], body });
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify({
      jsonrpc: "2.0",
      id: body.id,
      result: {
        accepted: true,
        status: "completed",
        operationId: body.params.operationId,
        traceId: body.params.traceId,
        gatewaySessionId: body.params.gatewaySessionId,
        resourceId: body.params.resourceId,
        capabilityId: body.params.capabilityId,
        dispatch: {
          shellExecuted: true,
          dispatchStatus: "succeeded",
          stdout: "device-host-cli ok",
          exitCode: 0
        },
        auditId: "aud_devicepod_test",
        evidenceId: "evd_devicepod_test",
        gateway: { gatewayId: "gtw_devicepod_test" }
      }
    }));
  });
  await new Promise((resolve) => cloudApi.listen(0, "127.0.0.1", resolve));
  const cloudPort = cloudApi.address().port;
  const service = await startDevicePod("device-pod-test", {
    HWLAB_CLOUD_API_INTERNAL_URL: `http://127.0.0.1:${cloudPort}`
  });

  try {
    const jobResponse = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs`, {
      method: "POST",
      headers: internalHeaders({ "content-type": "application/json" }),
      body: JSON.stringify({
        jobId: "job_devicepod_gateway_test",
        intent: "workspace.ls",
        args: { path: "src" },
        traceId: "trc_device_pod_gateway_test",
        operationId: "op_device_pod_gateway_test",
        targetId: "target-device-pod-test",
        profileHash: "sha256:test",
        ownerUserId: "usr_device_pod_test",
        profile: {
          schemaVersion: 1,
          devicePodId: "device-pod-test",
          target: { id: "target-device-pod-test" },
          projectWorkspace: { workspaceRoot: "F:\\Work\\Project" },
          route: {
            gatewaySessionId: "gws_devicepod_test",
            resourceId: "res_devicepod_test",
            capabilityId: "cap_device_host_cli",
            hostWorkspaceRoot: "F:\\Work\\Project",
            hostCli: "node tools\\device-host-cli.mjs"
          }
        }
      })
    });
    assert.equal(jobResponse.status, 202);
    const job = await jobResponse.json();
    assert.equal(job.status, "running");
    assert.equal(job.targetId, "target-device-pod-test");
    assert.equal(job.profileHash, "sha256:test");

    const output = await waitForJobStatus(service.port, "device-pod-test", "job_devicepod_gateway_test", "completed");
    assert.equal(output.status, "completed");
    assert.equal(output.output.text, "device-host-cli ok");
    assert.equal(output.output.auditId, "aud_devicepod_test");
    assert.equal(dispatches.length, 1);
    assert.equal(dispatches[0].internalService, "hwlab-device-pod");
    assert.equal(dispatches[0].internalToken, INTERNAL_TOKEN);
    assert.equal(dispatches[0].body.params.gatewaySessionId, "gws_devicepod_test");
    assert.equal(dispatches[0].body.params.input.cwd, "F:\\Work\\Project");
    assert.match(dispatches[0].body.params.input.command, /^node tools\\device-host-cli\.mjs --profile-json-b64 /u);
    assert.equal(dispatches[0].body.params.input.command.includes("node -e"), false);
  } finally {
    await service.stop();
    await new Promise((resolve, reject) => cloudApi.close((error) => (error ? reject(error) : resolve())));
  }
});

test("device-pod executor maps v0.1 CLI options to device-host-cli argv", async () => {
  let dispatchedBody = null;
  const cloudApi = createServer(async (request, response) => {
    dispatchedBody = await requestJson(request);
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify({
      ok: true,
      result: {
        status: "completed",
        dispatch: { shellExecuted: true, dispatchStatus: "succeeded", stdout: "ok", exitCode: 0 },
        auditId: "aud_options",
        evidenceId: "ev_options"
      }
    }));
  });
  await new Promise((resolve) => cloudApi.listen(0, "127.0.0.1", resolve));
  const cloudPort = cloudApi.address().port;
  const service = await startDevicePod("device-pod-test", {
    HWLAB_CLOUD_API_INTERNAL_URL: `http://127.0.0.1:${cloudPort}`
  });

  try {
    const jobResponse = await fetch(`http://127.0.0.1:${service.port}/v1/device-pods/device-pod-test/jobs`, {
      method: "POST",
      headers: internalHeaders({ "content-type": "application/json" }),
      body: JSON.stringify({
        jobId: "job_options_test",
        intent: "debug.download",
        args: { action: "start", captureUart: "uart/1", captureDurationMs: 8000, port: "COM4", baudRate: 921600 },
        profileHash: "sha256:profile",
        profile: {
          devicePodId: "device-pod-test",
          target: { id: "target-test" },
          route: { gatewaySessionId: "gws_test", resourceId: "res_test", capabilityId: "cap_test", hostCli: "node tools/device-host-cli.mjs" }
        }
      })
    });
    assert.equal(jobResponse.status, 202);
    await waitForJobStatus(service.port, "device-pod-test", "job_options_test", "completed");
    const command = dispatchedBody.params.input.command;
    assert.equal(command.includes("node -e"), false);
    assert.match(command, /^node tools\/device-host-cli\.mjs --profile-json-b64 /u);
    assert.match(command, / --pod-id device-pod-test debug-probe download start /u);
    assert.match(command, / --capture-uart uart\/1 --capture-duration-ms 8000 --port COM4 --baud-rate 921600$/u);
    const encoded = command.match(/--profile-json-b64 ([A-Za-z0-9+/=]+)/u)[1];
    const decoded = JSON.parse(Buffer.from(encoded, "base64").toString("utf8"));
    assert.equal(decoded.devicePodId, "device-pod-test");
    assert.equal(decoded.route.gatewaySessionId, "gws_test");
  } finally {
    await service.stop();
    await new Promise((resolve, reject) => cloudApi.close((error) => (error ? reject(error) : resolve())));
  }
});

test("device-pod executor maps absorbed G14 device-pod operations to host CLI argv", async () => {
  const commands = [];
  const cloudApi = createServer(async (request, response) => {
    const body = await requestJson(request);
    commands.push(body.params.input.command);
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify({
      jsonrpc: "2.0",
      id: body.id,
      result: { status: "completed", dispatch: { dispatchStatus: "succeeded", stdout: "ok", exitCode: 0 } }
    }));
  });
  await new Promise((resolve) => cloudApi.listen(0, "127.0.0.1", resolve));
  const cloudPort = cloudApi.address().port;
  const service = await startDevicePod("device-pod-test", { HWLAB_CLOUD_API_INTERNAL_URL: `http://127.0.0.1:${cloudPort}` });
  const baseBody = {
    profile: { devicePodId: "device-pod-test", target: { id: "target-test" }, route: { gatewaySessionId: "gws_test", hostCli: "node tools/device-host-cli.mjs" } }
  };

  try {
    await submitAndWait(service.port, "workspace.put", {
      path: "User/new.c",
      contentB64: Buffer.from("int x;\n").toString("base64"),
      createDirs: true
    }, "job_map_put", baseBody);
    await submitAndWait(service.port, "workspace.keil", {
      action: "add-source",
      base: "projects/app",
      path: "User/new.c",
      group: "User"
    }, "job_map_keil", baseBody);
    await submitAndWait(service.port, "io.uart.jsonrpc", {
      uartId: "uart/1",
      method: "gpio.read",
      params: "{\"pin\":\"PB5\"}",
      requireJsonrpcResult: true,
      expectResultField: ["pin", "value"]
    }, "job_map_jsonrpc", baseBody);

    assert.match(commands[0], / workspace put User\/new\.c --content-b64 /u);
    assert.match(commands[0], / --create-dirs$/u);
    assert.match(commands[1], / workspace keil add-source User\/new\.c --base projects\/app --group User$/u);
    assert.match(commands[2], / io-probe uart\/1 jsonrpc gpio\.read /u);
    assert.match(commands[2], / --params "\{\\"pin\\":\\"PB5\\"\}" /u);
    assert.match(commands[2], / --require-jsonrpc-result --expect-result-field pin --expect-result-field value$/u);
  } finally {
    await service.stop();
    await new Promise((resolve, reject) => cloudApi.close((error) => (error ? reject(error) : resolve())));
  }
});

test("device-pod executor fails mismatched gateway JSON-RPC response ids", async () => {
  const cloudApi = createServer(async (_request, response) => {
    response.writeHead(200, { "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify({ jsonrpc: "2.0", id: "wrong-id", result: { status: "completed", dispatch: { dispatchStatus: "succeeded", stdout: "ok", exitCode: 0 } } }));
  });
  await new Promise((resolve) => cloudApi.listen(0, "127.0.0.1", resolve));
  const cloudPort = cloudApi.address().port;
  const service = await startDevicePod("device-pod-test", { HWLAB_CLOUD_API_INTERNAL_URL: `http://127.0.0.1:${cloudPort}` });

  try {
    const output = await submitAndWait(service.port, "workspace.ls", { path: "." }, "job_bad_id", {
      profile: { devicePodId: "device-pod-test", target: { id: "target-test" }, route: { gatewaySessionId: "gws_test", hostCli: "node tools/device-host-cli.mjs" } }
    }, "failed");
    assert.equal(output.status, "failed");
    assert.match(output.blocker.summary, /JSON-RPC id mismatch/u);
  } finally {
    await service.stop();
    await new Promise((resolve, reject) => cloudApi.close((error) => (error ? reject(error) : resolve())));
  }
});

async function submitAndWait(port, intent, args, jobId, extraBody = {}, expectedStatus = "completed") {
  const response = await fetch(`http://127.0.0.1:${port}/v1/device-pods/device-pod-test/jobs`, {
    method: "POST",
    headers: internalHeaders({ "content-type": "application/json" }),
    body: JSON.stringify({ jobId, intent, args, traceId: `trc_${jobId}`, operationId: `op_${jobId}`, ...extraBody })
  });
  assert.equal(response.status, 202);
  return await waitForJobStatus(port, "device-pod-test", jobId, expectedStatus);
}

async function startDevicePod(devicePodId, extraEnv = {}) {
  const port = await freePort();
  const child = spawn(bunCommand, ["run", "cmd/hwlab-device-pod/main.ts"], {
    cwd: process.cwd(),
    env: {
      ...process.env,
      HWLAB_DEVICE_POD_PORT: String(port),
      HWLAB_DEVICE_POD_ID: devicePodId,
      HWLAB_ENVIRONMENT: "v02",
      HWLAB_DEVICE_POD_INTERNAL_TOKEN: INTERNAL_TOKEN,
      ...extraEnv
    },
    stdio: ["ignore", "pipe", "pipe"]
  });
  await waitForListening(child, "device pod");
  return {
    port,
    stop: () => stopChild(child)
  };
}

async function fetchJson(url, options = {}) {
  const response = await fetch(url, options);
  assert.equal(response.ok, true, `${url} returned ${response.status}`);
  return response.json();
}

async function waitForJobStatus(port, devicePodId, jobId, status) {
  let last;
  for (let attempt = 0; attempt < 30; attempt += 1) {
    last = await fetchJson(`http://127.0.0.1:${port}/v1/device-pods/${devicePodId}/jobs/${jobId}/output`, {
      headers: internalHeaders()
    });
    if (last.status === status) return last;
    await new Promise((resolve) => setTimeout(resolve, 50));
  }
  assert.equal(last?.status, status);
  return last;
}

function internalHeaders(extra = {}) {
  return { ...extra, "x-hwlab-internal-service": "hwlab-cloud-api", "x-hwlab-internal-token": INTERNAL_TOKEN };
}

async function requestJson(request) {
  const chunks = [];
  for await (const chunk of request) chunks.push(chunk);
  const text = Buffer.concat(chunks).toString("utf8").trim();
  return text ? JSON.parse(text) : {};
}

async function freePort() {
  const server = createServer();
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
  const port = server.address().port;
  await new Promise((resolve) => server.close(resolve));
  return port;
}

async function waitForListening(child, label) {
  let stderr = "";
  child.stderr.on("data", (chunk) => {
    stderr += chunk;
  });
  await new Promise((resolve, reject) => {
    const timer = setTimeout(() => reject(new Error(`${label} did not start: ${stderr}`)), 5000);
    child.stdout.on("data", (chunk) => {
      if (chunk.toString().includes("listening")) {
        clearTimeout(timer);
        resolve();
      }
    });
    child.on("exit", (code) => {
      clearTimeout(timer);
      reject(new Error(`${label} exited early code=${code}: ${stderr}`));
    });
  });
}

async function stopChild(child) {
  if (child.exitCode !== null) return;
  child.kill("SIGTERM");
  await new Promise((resolve) => child.once("exit", resolve));
}