feat: add hwpod harness node-ops loop

2026-06-05 11:47:00 +08:00
parent 45bd95e9d1
commit c8a8aa505e
13 changed files with 1795 additions and 3 deletions
@@ -0,0 +1,397 @@
+import { createServer } from "node:http";
+import { mkdir, readdir, readFile, rm, stat, writeFile } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
+
+import { HWPOD_NODE_OPS_CONTRACT_VERSION } from "./hwpod-harness-lib.ts";
+
+const NODE_VERSION = "0.1.0-thin-node-ops";
+const DEFAULT_BODY_LIMIT_BYTES = 1024 * 1024;
+const BOOLEAN_OPTIONS = new Set(["help", "h", "json"]);
+
+type ParsedArgs = Record<string, unknown> & { _: string[] };
+
+export async function mainHwpodNode(argv = process.argv.slice(2), options: any = {}) {
+  const result = await runHwpodNode(argv, options);
+  if (result?.keepAlive) return;
+  console.log(JSON.stringify(result.payload, null, 2));
+  process.exitCode = result.exitCode;
+}
+
+export async function runHwpodNode(argv: string[], options: { stdinText?: string; now?: () => string } = {}) {
+  const now = options.now ?? (() => new Date().toISOString());
+  try {
+    const parsed = parseOptions(argv);
+    const command = parsed._[0] || "help";
+    if (["help", "--help", "-h"].includes(command)) return result(0, help(), now);
+    if (command === "run") {
+      const plan = planFromParsed(parsed, options.stdinText);
+      return result(0, await executeHwpodNodeOpsPlan(plan, { now }), now);
+    }
+    if (command === "serve") {
+      const server = createHwpodNodeServer({ now });
+      const host = text(parsed.host) || "127.0.0.1";
+      const port = numberValue(parsed.port) ?? 19678;
+      await new Promise((resolve) => server.listen(port, host, resolve));
+      console.log(JSON.stringify({ ok: true, action: "hwpod-node.serve", status: "listening", host, port: server.address()?.port ?? port, observedAt: now() }, null, 2));
+      return { exitCode: 0, payload: null, keepAlive: true, server };
+    }
+    throw cliError("unsupported_hwpod_node_command", `unsupported hwpod-node command: ${command}`);
+  } catch (error) {
+    return result(1, failure("hwpod-node", error), now);
+  }
+}
+
+export function createHwpodNodeServer(options: any = {}) {
+  const now = options.now ?? (() => new Date().toISOString());
+  return createServer(async (request, response) => {
+    try {
+      const url = new URL(request.url || "/", "http://hwpod-node.local");
+      if (request.method === "GET" && (url.pathname === "/health/live" || url.pathname === "/v1/hwpod-node-ops")) {
+        return sendJson(response, 200, {
+          ok: true,
+          status: "ready",
+          serviceId: "hwpod-node",
+          contractVersion: HWPOD_NODE_OPS_CONTRACT_VERSION,
+          version: NODE_VERSION,
+          nodeId: process.env.HWPOD_NODE_ID || os.hostname(),
+          observedAt: now()
+        });
+      }
+      if (request.method === "POST" && url.pathname === "/v1/hwpod-node-ops") {
+        const body = await readBody(request, options.bodyLimitBytes);
+        const plan = body ? JSON.parse(body) : {};
+        return sendJson(response, 200, await executeHwpodNodeOpsPlan(plan, { now }));
+      }
+      return sendJson(response, 404, { ok: false, error: { code: "not_found", message: "hwpod-node route not found" } });
+    } catch (error) {
+      return sendJson(response, 500, failure("hwpod-node.http", error));
+    }
+  });
+}
+
+export async function executeHwpodNodeOpsPlan(plan: any, options: any = {}) {
+  const now = options.now ?? (() => new Date().toISOString());
+  if (!plan || typeof plan !== "object" || Array.isArray(plan)) throw cliError("invalid_hwpod_node_ops_plan", "plan must be a JSON object");
+  if (plan.contractVersion !== HWPOD_NODE_OPS_CONTRACT_VERSION) throw cliError("invalid_hwpod_node_ops_contract", `contractVersion must be ${HWPOD_NODE_OPS_CONTRACT_VERSION}`);
+  if (!Array.isArray(plan.ops) || plan.ops.length === 0) throw cliError("invalid_hwpod_node_ops", "ops must be a non-empty array");
+  const results = [];
+  for (const op of plan.ops) {
+    results.push(await executeOp(op, { plan, now }));
+  }
+  const failed = results.some((item) => item.ok === false);
+  return {
+    ok: !failed,
+    status: failed ? "failed" : "completed",
+    contractVersion: HWPOD_NODE_OPS_CONTRACT_VERSION,
+    serviceId: "hwpod-node",
+    nodeVersion: NODE_VERSION,
+    planId: plan.planId ?? null,
+    hwpodId: plan.hwpodId ?? null,
+    nodeId: plan.nodeId ?? process.env.HWPOD_NODE_ID ?? os.hostname(),
+    results,
+    observedAt: now()
+  };
+}
+
+async function executeOp(op: any, context: any) {
+  const opId = text(op?.opId) || "op_unknown";
+  const name = text(op?.op);
+  const args = op?.args && typeof op.args === "object" && !Array.isArray(op.args) ? op.args : {};
+  try {
+    if (name === "node.health") return opOk(opId, name, { platform: process.platform, arch: process.arch, hostname: os.hostname(), cwd: process.cwd() });
+    if (name === "node.version") return opOk(opId, name, { version: NODE_VERSION });
+    if (name === "node.inventory") return opOk(opId, name, await nodeInventory(args));
+    if (name === "workspace.ls") return opOk(opId, name, await workspaceLs(args));
+    if (name === "workspace.cat") return opOk(opId, name, await workspaceCat(args));
+    if (name === "workspace.rg") return opOk(opId, name, await workspaceRg(args));
+    if (name === "workspace.apply-patch") return opOk(opId, name, await workspaceApplyPatch(args));
+    if (name === "cmd.run") return opOk(opId, name, await cmdRun(args));
+    if (["debug.build", "debug.download", "debug.reset", "io.uart.read", "io.uart.write", "io.uart.jsonrpc"].includes(name)) {
+      return opBlocked(opId, name, "hwpod_node_op_not_configured", `${name} requires node-side tool binding; the thin node contract is present but this local executor has no binding yet`);
+    }
+    return opBlocked(opId, name, "unsupported_hwpod_node_op", `unsupported hwpod-node op: ${name}`);
+  } catch (error) {
+    return opBlocked(opId, name, error?.code || "hwpod_node_op_failed", error?.message || String(error));
+  }
+}
+
+async function nodeInventory(args: any) {
+  const root = workspaceRoot(args);
+  const info = await stat(root).catch(() => null);
+  return { workspaceRoot: root, workspaceExists: Boolean(info), workspaceIsDirectory: info?.isDirectory?.() ?? false };
+}
+
+async function workspaceLs(args: any) {
+  const target = resolveWorkspacePath(args, text(args.path) || ".");
+  const entries = await readdir(target, { withFileTypes: true });
+  return {
+    path: target,
+    entries: entries.map((entry) => ({ name: entry.name, type: entry.isDirectory() ? "dir" : entry.isFile() ? "file" : "other" }))
+  };
+}
+
+async function workspaceCat(args: any) {
+  const target = resolveWorkspacePath(args, requiredText(args.path, "path"));
+  const maxBytes = numberValue(args.maxBytes) ?? 65536;
+  const content = await readFile(target, "utf8");
+  const truncated = Buffer.byteLength(content, "utf8") > maxBytes;
+  return { path: target, content: truncated ? content.slice(0, maxBytes) : content, truncated };
+}
+
+async function workspaceRg(args: any) {
+  const pattern = requiredText(args.pattern, "pattern");
+  const cwd = workspaceRoot(args);
+  const rgPath = text(args.path) || ".";
+  const command = ["rg", "--line-number", "--color", "never", ...(args.ignoreCase === true ? ["--ignore-case"] : []), pattern, rgPath];
+  const output = await spawnOutput(command, { cwd, timeoutMs: numberValue(args.timeoutMs) ?? 10000 });
+  return { cwd, command, ...output };
+}
+
+async function workspaceApplyPatch(args: any) {
+  const patch = text(args.patch) || (text(args.patchBase64) ? Buffer.from(text(args.patchBase64), "base64").toString("utf8") : "");
+  if (!patch) throw cliError("patch_required", "patch or patchBase64 is required");
+  const cwd = workspaceRoot(args);
+  return { cwd, changes: await applyPatchEnvelope(cwd, patch) };
+}
+
+async function applyPatchEnvelope(root: string, patch: string) {
+  const lines = patch.replace(/\r\n?/gu, "\n").split("\n");
+  while (lines.length > 0 && lines[lines.length - 1] === "") lines.pop();
+  if (lines[0] !== "*** Begin Patch" || lines[lines.length - 1] !== "*** End Patch") {
+    throw cliError("invalid_apply_patch_envelope", "patch must start with *** Begin Patch and end with *** End Patch");
+  }
+  const changes = [];
+  for (let index = 1; index < lines.length - 1;) {
+    const header = lines[index];
+    if (header.startsWith("*** Add File: ")) {
+      const relativePath = header.slice("*** Add File: ".length).trim();
+      const filePath = resolvePatchFile(root, relativePath);
+      const addLines = [];
+      index += 1;
+      while (index < lines.length - 1 && !lines[index].startsWith("*** ")) {
+        if (!lines[index].startsWith("+")) throw cliError("invalid_add_file_hunk", `add file lines must start with + for ${relativePath}`);
+        addLines.push(lines[index].slice(1));
+        index += 1;
+      }
+      await mkdir(path.dirname(filePath), { recursive: true });
+      await writeFile(filePath, `${addLines.join("\n")}\n`, "utf8");
+      changes.push({ action: "add", path: relativePath, lines: addLines.length });
+      continue;
+    }
+    if (header.startsWith("*** Delete File: ")) {
+      const relativePath = header.slice("*** Delete File: ".length).trim();
+      await rm(resolvePatchFile(root, relativePath), { force: false });
+      changes.push({ action: "delete", path: relativePath });
+      index += 1;
+      continue;
+    }
+    if (header.startsWith("*** Update File: ")) {
+      const relativePath = header.slice("*** Update File: ".length).trim();
+      const filePath = resolvePatchFile(root, relativePath);
+      const hunks = [];
+      index += 1;
+      while (index < lines.length - 1 && !lines[index].startsWith("*** ")) {
+        if (!lines[index].startsWith("@@")) throw cliError("invalid_update_hunk", `update hunk must start with @@ for ${relativePath}`);
+        index += 1;
+        const hunkLines = [];
+        while (index < lines.length - 1 && !lines[index].startsWith("@@") && !lines[index].startsWith("*** ")) {
+          const prefix = lines[index][0];
+          if (![" ", "-", "+"].includes(prefix)) throw cliError("invalid_update_hunk_line", `hunk lines must start with space, -, or + for ${relativePath}`);
+          hunkLines.push(lines[index]);
+          index += 1;
+        }
+        hunks.push(hunkLines);
+      }
+      const applied = await applyUpdateHunks(filePath, hunks, relativePath);
+      changes.push({ action: "update", path: relativePath, hunks: hunks.length, replacements: applied });
+      continue;
+    }
+    throw cliError("unsupported_apply_patch_operation", `unsupported patch operation: ${header}`);
+  }
+  return changes;
+}
+
+async function applyUpdateHunks(filePath: string, hunks: string[][], relativePath: string) {
+  const original = await readFile(filePath, "utf8");
+  const hadFinalNewline = original.endsWith("\n");
+  let fileLines = original.replace(/\r\n?/gu, "\n").split("\n");
+  if (hadFinalNewline) fileLines.pop();
+  let searchStart = 0;
+  for (const hunk of hunks) {
+    const oldLines = hunk.filter((line) => line[0] !== "+").map((line) => line.slice(1));
+    const newLines = hunk.filter((line) => line[0] !== "-").map((line) => line.slice(1));
+    const at = findLineSequence(fileLines, oldLines, searchStart);
+    if (at < 0) throw cliError("apply_patch_context_not_found", `patch context not found for ${relativePath}`);
+    fileLines = [...fileLines.slice(0, at), ...newLines, ...fileLines.slice(at + oldLines.length)];
+    searchStart = at + newLines.length;
+  }
+  await writeFile(filePath, `${fileLines.join("\n")}${hadFinalNewline ? "\n" : ""}`, "utf8");
+  return hunks.length;
+}
+
+function findLineSequence(lines: string[], sequence: string[], start: number) {
+  if (sequence.length === 0) return start;
+  for (let index = start; index <= lines.length - sequence.length; index += 1) {
+    if (sequence.every((line, offset) => lines[index + offset] === line)) return index;
+  }
+  return -1;
+}
+
+function resolvePatchFile(root: string, relativePath: string) {
+  if (!relativePath || path.isAbsolute(relativePath)) throw cliError("invalid_patch_path", "patch file paths must be relative", { path: relativePath });
+  return resolveWorkspacePath({ workspacePath: root }, relativePath);
+}
+
+async function cmdRun(args: any) {
+  const command = requiredText(args.command, "command");
+  const argv = Array.isArray(args.argv) ? args.argv.map(String) : [];
+  const cwd = workspaceRoot(args);
+  const output = await spawnOutput([command, ...argv], { cwd, timeoutMs: numberValue(args.timeoutMs) ?? 30000 });
+  return { cwd, command: [command, ...argv], ...output };
+}
+
+async function spawnOutput(command: string[], { cwd, stdinText = "", timeoutMs }: any) {
+  const proc = Bun.spawn(command, { cwd, stdin: stdinText ? "pipe" : "ignore", stdout: "pipe", stderr: "pipe" });
+  if (stdinText && proc.stdin) {
+    proc.stdin.write(stdinText);
+    proc.stdin.end();
+  }
+  const timeout = setTimeout(() => proc.kill(), timeoutMs);
+  try {
+    const [stdout, stderr, exitCode] = await Promise.all([new Response(proc.stdout).text(), new Response(proc.stderr).text(), proc.exited]);
+    return { exitCode, stdout, stderr, ok: exitCode === 0 };
+  } finally {
+    clearTimeout(timeout);
+  }
+}
+
+function workspaceRoot(args: any) {
+  return path.resolve(text(args.workspacePath) || process.env.HWPOD_WORKSPACE_PATH || process.cwd());
+}
+
+function resolveWorkspacePath(args: any, relativePath: string) {
+  const root = workspaceRoot(args);
+  const target = path.resolve(root, relativePath);
+  if (target !== root && !target.startsWith(`${root}${path.sep}`)) {
+    throw cliError("workspace_path_outside_root", "workspace path must stay inside workspacePath", { workspacePath: root, path: relativePath });
+  }
+  return target;
+}
+
+function planFromParsed(parsed: ParsedArgs, stdinText?: string) {
+  if (parsed.planJson) return JSON.parse(String(parsed.planJson));
+  if (parsed.plan) return JSON.parse(String(parsed.plan));
+  if (stdinText) return JSON.parse(stdinText);
+  throw cliError("hwpod_node_plan_required", "pass --plan-json or pipe a hwpod-node-ops plan to hwpod-node run");
+}
+
+function help() {
+  return {
+    ok: true,
+    action: "hwpod-node.help",
+    status: "succeeded",
+    contractVersion: HWPOD_NODE_OPS_CONTRACT_VERSION,
+    version: NODE_VERSION,
+    role: "Thin node-side executor for stable hwpod-node-ops.",
+    usage: [
+      "bun tools/hwpod-node.ts run --plan-json '{...}'",
+      "bun tools/hwpod-node.ts serve --host 127.0.0.1 --port 19678"
+    ]
+  };
+}
+
+function result(exitCode: number, payload: any, now: () => string) {
+  return { exitCode, payload: { ...payload, observedAt: payload?.observedAt ?? now() } };
+}
+
+function opOk(opId: string, op: string, output: any) {
+  return { opId, op, ok: true, status: "completed", output };
+}
+
+function opBlocked(opId: string, op: string, code: string, summary: string) {
+  return { opId, op, ok: false, status: "blocked", blocker: { code, layer: "hwpod-node", retryable: true, summary } };
+}
+
+function failure(action: string, error: any) {
+  return { ok: false, action, status: "failed", error: { code: error?.code || "hwpod_node_error", message: error?.message || String(error), details: error?.details || undefined } };
+}
+
+function parseOptions(argv: string[]): ParsedArgs {
+  const parsed: ParsedArgs = { _: [] };
+  for (let index = 0; index < argv.length; index += 1) {
+    const token = argv[index];
+    if (token === "--") {
+      parsed._.push(...argv.slice(index + 1));
+      break;
+    }
+    if (!token.startsWith("-") || token === "-") {
+      parsed._.push(token);
+      continue;
+    }
+    if (token === "-h") {
+      parsed.h = true;
+      parsed.help = true;
+      continue;
+    }
+    const [rawKey, rawInline] = token.slice(2).split(/=(.*)/su, 2);
+    const key = rawKey.replace(/-([a-z0-9])/giu, (_match, char) => char.toUpperCase());
+    if (BOOLEAN_OPTIONS.has(key)) {
+      parsed[key] = rawInline === undefined || rawInline === "" ? true : /^(?:1|true|yes|on)$/iu.test(rawInline);
+      continue;
+    }
+    if (rawInline !== undefined && rawInline !== "") {
+      parsed[key] = rawInline;
+      continue;
+    }
+    const next = argv[index + 1];
+    if (next === undefined || next.startsWith("--")) {
+      parsed[key] = true;
+    } else {
+      parsed[key] = next;
+      index += 1;
+    }
+  }
+  return parsed;
+}
+
+function readBody(request: any, limitBytes = DEFAULT_BODY_LIMIT_BYTES) {
+  const limit = Number.isInteger(limitBytes) && limitBytes > 0 ? limitBytes : DEFAULT_BODY_LIMIT_BYTES;
+  return new Promise<string>((resolve, reject) => {
+    let body = "";
+    request.setEncoding("utf8");
+    request.on("data", (chunk: string) => {
+      body += chunk;
+      if (Buffer.byteLength(body, "utf8") > limit) request.destroy(new Error(`request body exceeds ${limit} bytes`));
+    });
+    request.on("end", () => resolve(body));
+    request.on("error", reject);
+  });
+}
+
+function sendJson(response: any, statusCode: number, body: any) {
+  response.writeHead(statusCode, { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" });
+  response.end(`${JSON.stringify(body)}\n`);
+}
+
+function cliError(code: string, message: string, details: Record<string, unknown> = {}) {
+  const error = new Error(message) as Error & { code?: string; details?: Record<string, unknown> };
+  error.code = code;
+  error.details = details;
+  return error;
+}
+
+function requiredText(value: unknown, name: string) {
+  const normalized = text(value);
+  if (!normalized) throw cliError("required_option_missing", `${name} is required`, { name });
+  return normalized;
+}
+
+function text(value: unknown) {
+  return typeof value === "string" && value.trim() ? value.trim() : "";
+}
+
+function numberValue(value: unknown) {
+  const parsed = Number.parseInt(String(value ?? ""), 10);
+  return Number.isInteger(parsed) ? parsed : undefined;
+}