feat: add hwpod node native websocket dispatch

cmd/hwlab-cloud-api/main.ts

@@ -1,5 +1,5 @@
 #!/usr/bin/env bun
-import { createCloudApiServer } from "../../internal/cloud/server.ts";
+import { createCloudApiBunServer } from "../../internal/cloud/bun-server.ts";
 import { applyCloudApiImageEnv, parsePort, parseTimeout } from "./runtime-options.ts";
 
 const host = process.env.HWLAB_CLOUD_API_HOST || "0.0.0.0";
@@ -14,28 +14,26 @@ const codeAgentHardTimeoutMs = parseTimeout(process.env.HWLAB_CODE_AGENT_HARD_TI
 });
 applyCloudApiImageEnv(process.env);
 
-const server = createCloudApiServer({
+const runtime = await createCloudApiBunServer({
+  host,
+  port,
   codeAgentTimeoutMs,
   codeAgentHardTimeoutMs
 });
 
-server.listen(port, host, () => {
-  const address = server.address();
-  const resolvedPort = typeof address === "object" && address ? address.port : port;
-  process.stdout.write(
-    `${JSON.stringify({
-      serviceId: "hwlab-cloud-api",
-      status: "listening",
-      host,
-      port: resolvedPort
-    })}\n`
-  );
-});
+process.stdout.write(
+  `${JSON.stringify({
+    serviceId: "hwlab-cloud-api",
+    status: "listening",
+    host,
+    port: runtime.port,
+    hwpodNodeWs: "/v1/hwpod-node/ws"
+  })}\n`
+);
 
 for (const signal of ["SIGINT", "SIGTERM"]) {
   process.on(signal, () => {
-    server.close(() => {
-      process.exit(0);
-    });
+    runtime.stop();
+    process.exit(0);
   });
 }

internal/cloud/bun-server.ts

@@ -0,0 +1,72 @@
+import { createCloudApiServer } from "./server.ts";
+import { createHwpodNodeWsRegistry } from "./hwpod-node-ws-registry.ts";
+
+export async function createCloudApiBunServer(options: any = {}) {
+  const env = options.env ?? process.env;
+  const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry || createHwpodNodeWsRegistry({ ...options, env });
+  const innerServer = createCloudApiServer({ ...options, env, hwpodNodeWsRegistry });
+  await new Promise((resolve) => innerServer.listen(0, "127.0.0.1", resolve));
+  const innerAddress = innerServer.address();
+  const innerPort = typeof innerAddress === "object" && innerAddress ? innerAddress.port : 0;
+  if (!innerPort) throw new Error("cloud-api inner HTTP server did not expose a port");
+
+  const host = options.host ?? "0.0.0.0";
+  const port = options.port ?? 0;
+  const server = Bun.serve({
+    hostname: host,
+    port,
+    async fetch(request, bunServer) {
+      const url = new URL(request.url);
+      if (url.pathname === "/v1/hwpod-node/ws") {
+        const expectedToken = String(env.HWLAB_HWPOD_NODE_WS_TOKEN ?? "").trim();
+        const token = String(url.searchParams.get("token") ?? request.headers.get("x-hwpod-node-token") ?? "");
+        if (expectedToken && token !== expectedToken) return json({ ok: false, error: { code: "hwpod_node_ws_token_invalid" } }, 401);
+        const upgraded = bunServer.upgrade(request, { data: { hwpodNodeWsRegistry } });
+        return upgraded ? undefined : json({ ok: false, status: "upgrade_required", route: "/v1/hwpod-node/ws" }, 426);
+      }
+      return proxyHttpToInnerServer(request, innerPort);
+    },
+    websocket: {
+      open(socket: any) {
+        socket.data.hwpodNodeWsRegistry.openBunSocket(socket);
+      },
+      message(socket: any, message: string | Buffer) {
+        socket.data.hwpodNodeWsRegistry.handleBunMessage(socket, message);
+      },
+      close(socket: any) {
+        socket.data.hwpodNodeWsRegistry.closeBunSocket(socket);
+      }
+    }
+  });
+
+  return {
+    server,
+    innerServer,
+    hwpodNodeWsRegistry,
+    port: server.port,
+    url: `http://${host === "0.0.0.0" ? "127.0.0.1" : host}:${server.port}`,
+    stop() {
+      server.stop(true);
+      innerServer.close();
+    }
+  };
+}
+
+async function proxyHttpToInnerServer(request: Request, innerPort: number) {
+  const sourceUrl = new URL(request.url);
+  const targetUrl = new URL(sourceUrl.pathname + sourceUrl.search, `http://127.0.0.1:${innerPort}`);
+  const method = request.method.toUpperCase();
+  return fetch(targetUrl, {
+    method,
+    headers: request.headers,
+    body: method === "GET" || method === "HEAD" ? undefined : request.body,
+    redirect: "manual"
+  });
+}
+
+function json(body: any, status = 200) {
+  return new Response(`${JSON.stringify(body)}\n`, {
+    status,
+    headers: { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" }
+  });
+}

internal/cloud/hwpod-node-ops.test.ts

@@ -1,8 +1,12 @@
 import assert from "node:assert/strict";
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import os from "node:os";
+import path from "node:path";
 import { test } from "bun:test";
 
+import { createCloudApiBunServer } from "./bun-server.ts";
 import { createCloudApiServer } from "./server.ts";
-import { createHwpodNodeServer } from "../../tools/src/hwpod-node-lib.ts";
+import { connectHwpodNodeWs, createHwpodNodeServer } from "../../tools/src/hwpod-node-lib.ts";
 
 test("cloud-api exposes hwpod-node-ops contract on /v1", async () => {
   const server = createCloudApiServer();
@@ -102,6 +106,34 @@ test("cloud-api forwards hwpod-node-ops to configured thin hwpod-node URL", asyn
   }
 });
 
+test("cloud-api dispatches hwpod-node-ops to outbound WebSocket hwpod-node by nodeId", async () => {
+  const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-hwpod-node-ws-"));
+  const runtime = await createCloudApiBunServer({ host: "127.0.0.1", port: 0, env: { PATH: process.env.PATH } });
+  const connector = connectHwpodNodeWs({ cloudUrl: runtime.url, nodeId: "pc-host-1", reconnect: false, heartbeatIntervalMs: 1000 });
+  try {
+    await writeFile(path.join(root, "main.c"), "int main(void) { return 0; }\n", "utf8");
+    await withTimeout(connector.ready, 3000, "hwpod-node WebSocket registration timed out");
+    const plan = samplePlan();
+    plan.ops[0].args.workspacePath = root;
+    const response = await fetch(`${runtime.url}/v1/hwpod-node-ops`, {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify(plan)
+    });
+    const payload = await response.json();
+    assert.equal(response.status, 200);
+    assert.equal(payload.ok, true);
+    assert.equal(payload.status, "completed");
+    assert.equal(payload.nodeId, "pc-host-1");
+    assert.equal(payload.results[0].op, "workspace.ls");
+    assert.equal(payload.results[0].output.entries.some((entry: any) => entry.name === "main.c"), true);
+  } finally {
+    connector.close();
+    runtime.stop();
+    await rm(root, { recursive: true, force: true });
+  }
+});
+
 test("cloud-api rejects unsupported hwpod-node ops before forwarding", async () => {
   const server = createCloudApiServer({
     hwpodNodeOpsHandler: async () => {
@@ -150,3 +182,17 @@ async function close(server: any) {
 function serverUrl(server: any) {
   return `http://127.0.0.1:${server.address().port}`;
 }
+
+async function withTimeout(promise: Promise<unknown>, timeoutMs: number, message: string) {
+  let timer: ReturnType<typeof setTimeout> | null = null;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise((_resolve, reject) => {
+        timer = setTimeout(() => reject(new Error(message)), timeoutMs);
+      })
+    ]);
+  } finally {
+    if (timer !== null) clearTimeout(timer);
+  }
+}

internal/cloud/hwpod-node-ws-registry.ts

@@ -0,0 +1,199 @@
+import { randomUUID } from "node:crypto";
+
+const DEFAULT_DISPATCH_TIMEOUT_MS = 30000;
+
+type HwpodNodeConnection = {
+  id: string;
+  socket: any;
+  nodeId: string | null;
+  name: string | null;
+  capabilities: string[];
+  labels: Record<string, unknown>;
+  firstSeenAt: string;
+  lastSeenAt: string;
+};
+
+type PendingDispatch = {
+  nodeId: string;
+  requestId: string;
+  timer: ReturnType<typeof setTimeout>;
+  resolve: (value: any) => void;
+};
+
+export function createHwpodNodeWsRegistry(options: any = {}) {
+  const now = options.now ?? (() => new Date().toISOString());
+  const clock = options.clock ?? (() => Date.now());
+  const connections = new Map<string, HwpodNodeConnection>();
+  const pending = new Map<string, PendingDispatch>();
+  const socketConnections = new WeakMap<object, HwpodNodeConnection>();
+
+  function openBunSocket(socket: any) {
+    const connection: HwpodNodeConnection = {
+      id: `hwpod_node_conn_${randomUUID()}`,
+      socket,
+      nodeId: null,
+      name: null,
+      capabilities: [],
+      labels: {},
+      firstSeenAt: now(),
+      lastSeenAt: now()
+    };
+    socketConnections.set(socket, connection);
+    return connection;
+  }
+
+  function handleBunMessage(socket: any, raw: unknown) {
+    const connection = socketConnections.get(socket);
+    if (!connection) return;
+    const text = typeof raw === "string" ? raw : Buffer.from(raw as any).toString("utf8");
+    handleText(connection, text);
+  }
+
+  function closeBunSocket(socket: any) {
+    const connection = socketConnections.get(socket);
+    if (!connection) return;
+    socketConnections.delete(socket);
+    removeConnection(connection);
+  }
+
+  function dispatch(plan: any, requestMeta: any = {}, dispatchOptions: any = {}) {
+    const nodeId = safeNodeId(plan?.nodeId);
+    const connection = nodeId ? connections.get(nodeId) : null;
+    if (!nodeId || !connection) {
+      return Promise.resolve(blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId || "<missing>"} is not connected through outbound WebSocket`));
+    }
+    const requestId = safeText(requestMeta.requestId) || `req_hwpod_ws_${randomUUID()}`;
+    const timeoutMs = positiveInteger(dispatchOptions.timeoutMs, DEFAULT_DISPATCH_TIMEOUT_MS);
+    return new Promise((resolve) => {
+      const timer = setTimeout(() => {
+        pending.delete(requestId);
+        resolve(blockedDispatch(plan, requestMeta, `hwpod-node ${nodeId} WebSocket dispatch timed out after ${timeoutMs}ms`));
+      }, timeoutMs);
+      pending.set(requestId, { nodeId, requestId, timer, resolve });
+      try {
+        sendJson(connection, { type: "hwpod-node-ops", requestId, plan, requestMeta });
+      } catch (error) {
+        clearTimeout(timer);
+        pending.delete(requestId);
+        resolve(blockedDispatch(plan, requestMeta, error instanceof Error ? error.message : String(error)));
+      }
+    });
+  }
+
+  function describe() {
+    const observedAt = now();
+    return {
+      mode: "hwpod-node-outbound-native-ws",
+      route: "/v1/hwpod-node/ws",
+      connectedCount: connections.size,
+      pendingCount: pending.size,
+      nodes: [...connections.values()].map((connection) => ({
+        nodeId: connection.nodeId,
+        name: connection.name,
+        status: "online",
+        capabilityCount: connection.capabilities.length,
+        capabilities: connection.capabilities,
+        labels: connection.labels,
+        firstSeenAt: connection.firstSeenAt,
+        lastSeenAt: connection.lastSeenAt,
+        ageMs: Math.max(0, clock() - Date.parse(connection.lastSeenAt || observedAt))
+      }))
+    };
+  }
+
+  function hasNode(nodeId: string) {
+    return connections.has(nodeId);
+  }
+
+  function handleText(connection: HwpodNodeConnection, text: string) {
+    let message: any;
+    try {
+      message = JSON.parse(text);
+    } catch (error) {
+      sendJson(connection, { type: "ack", requestId: "message", ok: false, message: "invalid JSON message" });
+      return;
+    }
+    const messageType = safeText(message.type);
+    const nodeId = safeNodeId(message.nodeId);
+    if (messageType === "register") {
+      if (!nodeId) {
+        sendJson(connection, { type: "ack", requestId: "register", ok: false, message: "nodeId is required" });
+        return;
+      }
+      connection.nodeId = nodeId;
+      connection.name = safeText(message.name) || nodeId;
+      connection.capabilities = Array.isArray(message.capabilities) ? message.capabilities.map(String).filter(Boolean).slice(0, 64) : [];
+      connection.labels = message.labels && typeof message.labels === "object" && !Array.isArray(message.labels) ? message.labels : {};
+      connection.lastSeenAt = now();
+      const previous = connections.get(nodeId);
+      if (previous && previous !== connection) previous.socket.close();
+      connections.set(nodeId, connection);
+      sendJson(connection, { type: "ack", requestId: "register", ok: true, message: "registered", nodeId });
+      return;
+    }
+    if (messageType === "heartbeat") {
+      if (nodeId && connection.nodeId === nodeId) connection.lastSeenAt = now();
+      return;
+    }
+    if (messageType === "hwpod-node-ops-result") completeDispatch(message);
+  }
+
+  function completeDispatch(message: any) {
+    const requestId = safeText(message.requestId);
+    const waiter = requestId ? pending.get(requestId) : null;
+    if (!waiter) return;
+    clearTimeout(waiter.timer);
+    pending.delete(requestId);
+    const result = message.result && typeof message.result === "object" && !Array.isArray(message.result)
+      ? message.result
+      : { ok: false, status: "failed", results: [], blocker: { code: "hwpod_node_result_invalid", layer: "hwpod-node", retryable: true, summary: "hwpod-node WebSocket result payload is invalid" } };
+    waiter.resolve(result);
+  }
+
+  function removeConnection(connection: HwpodNodeConnection) {
+    if (connection.nodeId && connections.get(connection.nodeId) === connection) connections.delete(connection.nodeId);
+    for (const [requestId, waiter] of pending) {
+      if (waiter.nodeId !== connection.nodeId) continue;
+      clearTimeout(waiter.timer);
+      pending.delete(requestId);
+      waiter.resolve(blockedDispatch({ nodeId: waiter.nodeId, ops: [] }, { requestId }, `hwpod-node ${waiter.nodeId} WebSocket disconnected before returning result`));
+    }
+  }
+
+  return { openBunSocket, handleBunMessage, closeBunSocket, dispatch, describe, hasNode };
+}
+
+function sendJson(connection: HwpodNodeConnection, value: any) {
+  connection.socket.send(JSON.stringify(value));
+}
+
+function blockedDispatch(plan: any, requestMeta: any, summary: string) {
+  const ops = Array.isArray(plan?.ops) ? plan.ops : [];
+  return {
+    ok: false,
+    status: "blocked",
+    results: ops.map((op: any, index: number) => ({
+      opId: safeText(op?.opId) || `op_${index + 1}`,
+      op: safeText(op?.op) || "unknown",
+      ok: false,
+      status: "blocked",
+      blocker: { code: "hwpod_node_unavailable", layer: "hwpod-node", retryable: true, summary }
+    })),
+    blocker: { code: "hwpod_node_unavailable", layer: "hwpod-node", retryable: true, summary },
+    requestMeta
+  };
+}
+
+function safeText(value: unknown) {
+  return typeof value === "string" ? value.trim() : "";
+}
+
+function safeNodeId(value: unknown) {
+  const text = safeText(value);
+  return /^[A-Za-z0-9._:-]{1,128}$/u.test(text) ? text : "";
+}
+
+function positiveInteger(value: unknown, fallback: number) {
+  const numberValue = Number(value);
+  return Number.isInteger(numberValue) && numberValue > 0 ? numberValue : fallback;
+}

internal/cloud/server.ts

@@ -68,6 +68,7 @@ import {
   configureSkillRuntime,
   ensureCodexSkillsAggregationSync
 } from "./skills-store.ts";
+import { createHwpodNodeWsRegistry } from "./hwpod-node-ws-registry.ts";
 import { HWPOD_NODE_OPS, HWPOD_NODE_OPS_CONTRACT_VERSION } from "../../tools/src/hwpod-node-ops-contract.ts";
 
 const DEFAULT_BODY_LIMIT_BYTES = 1024 * 1024;
@@ -92,13 +93,14 @@ export function createCloudApiServer(options = {}) {
     staleMs: parsePositiveInteger(env.HWLAB_GATEWAY_DEMO_STALE_MS, 30000),
     dispatchTimeoutMs: parsePositiveInteger(env.HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS, 120000)
   });
+  const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry || createHwpodNodeWsRegistry({ ...options, env });
   const accessController = options.accessController || createAccessController({ ...options, env, runtimeStore, gatewayRegistry, traceStore, codeAgentChatResults });
   if (typeof accessController.configureCodeAgentWorkspaceContext === "function") {
     accessController.configureCodeAgentWorkspaceContext({ env, fetchImpl: options.fetchImpl, traceStore, codeAgentChatResults });
   }
   return createServer(async (request, response) => {
     try {
-      await routeRequest(request, response, { ...options, env, runtimeStore, gatewayRegistry, accessController, sessionRegistry, traceStore, codexStdioManager, codeAgentChatResults, webPerformanceStore });
+      await routeRequest(request, response, { ...options, env, runtimeStore, gatewayRegistry, hwpodNodeWsRegistry, accessController, sessionRegistry, traceStore, codexStdioManager, codeAgentChatResults, webPerformanceStore });
     } catch (error) {
       if (error?.alreadySent) return;
       sendJson(response, 500, {
@@ -423,6 +425,11 @@ async function handleRestAdapter(request, response, url, options) {
     return;
   }
 
+  if (request.method === "GET" && url.pathname === "/v1/hwpod-node/ws") {
+    sendJson(response, 426, { ok: false, status: "upgrade_required", route: "/v1/hwpod-node/ws", websocket: options.hwpodNodeWsRegistry.describe() });
+    return;
+  }
+
   if (request.method === "GET" && url.pathname === M3_IO_CONTROL_ROUTE) {
     sendJson(response, 200, await describeM3IoControlLive(options));
     return;
@@ -591,7 +598,8 @@ async function handleHwpodNodeOpsHttp(request, response, options) {
       compiler: "hwpod-compiler-cli",
       apiRole: "node-ops-forwarder",
       nodeRole: "thin-hwpod-node-executor",
-      supportedOps: Array.from(HWPOD_NODE_OPS)
+      supportedOps: Array.from(HWPOD_NODE_OPS),
+      websocket: options.hwpodNodeWsRegistry.describe()
     });
     return;
   }
@@ -624,15 +632,18 @@ async function handleHwpodNodeOpsHttp(request, response, options) {
     environment: runtimeEnvironment(options.env ?? process.env)
   };
   const hwpodNodeOpsUrl = normalizedHwpodNodeOpsUrl(options.env ?? process.env);
-  if (typeof options.hwpodNodeOpsHandler !== "function" && !hwpodNodeOpsUrl) {
-    sendJson(response, 200, hwpodNodeOpsBlockedPayload(plan, requestMeta, "HWLAB_HWPOD_NODE_OPS_URL is not configured; cloud-api is only validating the hwpod-node-ops contract"));
+  const hwpodNodeWsRegistry = options.hwpodNodeWsRegistry;
+  if (typeof options.hwpodNodeOpsHandler !== "function" && !hwpodNodeWsRegistry?.hasNode?.(plan.nodeId) && !hwpodNodeOpsUrl) {
+    sendJson(response, 200, hwpodNodeOpsBlockedPayload(plan, requestMeta, "no outbound WebSocket hwpod-node is connected and HWLAB_HWPOD_NODE_OPS_URL is not configured; cloud-api is only validating the hwpod-node-ops contract"));
     return;
   }
 
   try {
     const handled = typeof options.hwpodNodeOpsHandler === "function"
       ? await options.hwpodNodeOpsHandler(plan, { request, requestMeta, env: options.env ?? process.env })
-      : await forwardHwpodNodeOpsPlan(hwpodNodeOpsUrl, plan, requestMeta, options);
+      : hwpodNodeWsRegistry?.hasNode?.(plan.nodeId)
+        ? await hwpodNodeWsRegistry.dispatch(plan, requestMeta, { timeoutMs: parsePositiveInteger(options.env?.HWLAB_HWPOD_NODE_OPS_TIMEOUT_MS, 30000) })
+        : await forwardHwpodNodeOpsPlan(hwpodNodeOpsUrl, plan, requestMeta, options);
     const results = Array.isArray(handled?.results) ? handled.results : [];
     const failed = results.some((item) => item?.ok === false);
     sendJson(response, handled?.httpStatus ?? (failed ? 409 : 200), {

tools/src/hwpod-node-lib.ts

@@ -12,12 +12,23 @@ 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);
+  const stdinText = options.stdinText ?? await readCliStdinForRun(argv);
+  const result = await runHwpodNode(argv, { ...options, stdinText });
   if (result?.keepAlive) return;
   console.log(JSON.stringify(result.payload, null, 2));
   process.exitCode = result.exitCode;
 }
 
+async function readCliStdinForRun(argv: string[]) {
+  const command = argv.find((token) => !token.startsWith("-")) || "help";
+  const hasInlinePlan = argv.some((token) => token === "--plan-json" || token === "--plan" || token.startsWith("--plan-json=") || token.startsWith("--plan="));
+  if (command !== "run" || hasInlinePlan || process.stdin.isTTY) return undefined;
+  const chunks: Buffer[] = [];
+  for await (const chunk of process.stdin) chunks.push(Buffer.from(chunk));
+  const textValue = Buffer.concat(chunks).toString("utf8");
+  return textValue.trim() ? textValue : undefined;
+}
+
 export async function runHwpodNode(argv: string[], options: { stdinText?: string; now?: () => string } = {}) {
   const now = options.now ?? (() => new Date().toISOString());
   try {
@@ -36,12 +47,142 @@ export async function runHwpodNode(argv: string[], options: { stdinText?: string
       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 };
     }
+    if (command === "connect") {
+      const nodeId = text(parsed.nodeId ?? parsed.node) || process.env.HWPOD_NODE_ID || os.hostname();
+      const connector = connectHwpodNodeWs({
+        wsUrl: text(parsed.wsUrl),
+        cloudUrl: text(parsed.cloudUrl ?? parsed.apiUrl),
+        token: text(parsed.token) || process.env.HWPOD_NODE_WS_TOKEN || process.env.HWLAB_HWPOD_NODE_WS_TOKEN || "",
+        nodeId,
+        name: text(parsed.name) || nodeId,
+        heartbeatIntervalMs: numberValue(parsed.heartbeatIntervalMs) ?? 15000,
+        reconnectBaseMs: numberValue(parsed.reconnectBaseMs) ?? 1000,
+        reconnectMaxMs: numberValue(parsed.reconnectMaxMs) ?? 30000,
+        now
+      });
+      console.log(JSON.stringify({ ok: true, action: "hwpod-node.connect", status: "connecting", nodeId, wsUrl: connector.wsUrl, observedAt: now() }, null, 2));
+      return { exitCode: 0, payload: null, keepAlive: true, connector };
+    }
     throw cliError("unsupported_hwpod_node_command", `unsupported hwpod-node command: ${command}`);
   } catch (error) {
     return result(1, failure("hwpod-node", error), now);
   }
 }
 
+export function connectHwpodNodeWs(config: any, options: any = {}) {
+  const now = config.now ?? options.now ?? (() => new Date().toISOString());
+  const nodeId = requiredText(config.nodeId || process.env.HWPOD_NODE_ID || os.hostname(), "nodeId");
+  const wsUrl = hwpodNodeWsUrl(config);
+  const heartbeatIntervalMs = Math.max(1000, numberValue(config.heartbeatIntervalMs) ?? 15000);
+  const reconnectBaseMs = Math.max(250, numberValue(config.reconnectBaseMs) ?? 1000);
+  const reconnectMaxMs = Math.max(reconnectBaseMs, numberValue(config.reconnectMaxMs) ?? 30000);
+  const reconnect = config.reconnect !== false;
+  const executePlan = options.executePlan ?? ((plan: any) => executeHwpodNodeOpsPlan(plan, { now }));
+  let socket: WebSocket | null = null;
+  let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
+  let reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  let closed = false;
+  let attempt = 0;
+  let readyResolve: ((value: unknown) => void) | null = null;
+  const ready = new Promise((resolve) => { readyResolve = resolve; });
+
+  function connect() {
+    if (closed) return;
+    socket = new WebSocket(wsUrl);
+    socket.addEventListener("open", () => {
+      attempt = 0;
+      sendJson({
+        type: "register",
+        nodeId,
+        name: text(config.name) || nodeId,
+        capabilities: ["hwpod-node-ops", "cmd.run", "workspace.ls", "workspace.cat", "workspace.rg", "workspace.apply-patch", "debug.build", "debug.download", "debug.reset"],
+        labels: { platform: process.platform, arch: process.arch, hostname: os.hostname(), version: NODE_VERSION, startedAt: now() }
+      });
+      sendHeartbeat();
+      heartbeatTimer = setInterval(sendHeartbeat, heartbeatIntervalMs);
+    });
+    socket.addEventListener("message", (event: MessageEvent) => {
+      void handleWsMessage(event.data).catch((error) => {
+        sendJson({ type: "hwpod-node-error", nodeId, error: { message: error?.message ?? String(error), code: error?.code ?? "hwpod_node_ws_message_failed" }, observedAt: now() });
+      });
+    });
+    socket.addEventListener("close", scheduleReconnect);
+    socket.addEventListener("error", () => undefined);
+  }
+
+  function scheduleReconnect() {
+    clearTimers();
+    if (closed || !reconnect) return;
+    attempt += 1;
+    const delayMs = Math.min(reconnectMaxMs, reconnectBaseMs * 2 ** Math.min(attempt, 8));
+    reconnectTimer = setTimeout(connect, delayMs);
+  }
+
+  async function handleWsMessage(raw: any) {
+    const textValue = typeof raw === "string" ? raw : await new Response(raw).text();
+    const message = JSON.parse(textValue);
+    if (message?.type === "ack" && message.requestId === "register" && readyResolve) {
+      readyResolve(message);
+      readyResolve = null;
+      return;
+    }
+    if (message?.type !== "hwpod-node-ops") return;
+    const requestId = text(message.requestId) || "req_unknown";
+    let resultPayload;
+    try {
+      resultPayload = await executePlan(message.plan);
+    } catch (error) {
+      resultPayload = failure("hwpod-node.ws.run", error);
+    }
+    sendJson({ type: "hwpod-node-ops-result", nodeId, requestId, result: resultPayload, observedAt: now() });
+  }
+
+  function sendHeartbeat() {
+    sendJson({ type: "heartbeat", nodeId, at: now() });
+  }
+
+  function sendJson(value: any) {
+    if (!socket || socket.readyState !== WebSocket.OPEN) return false;
+    socket.send(JSON.stringify(value));
+    return true;
+  }
+
+  function clearTimers() {
+    if (heartbeatTimer !== null) clearInterval(heartbeatTimer);
+    if (reconnectTimer !== null) clearTimeout(reconnectTimer);
+    heartbeatTimer = null;
+    reconnectTimer = null;
+  }
+
+  function close() {
+    closed = true;
+    clearTimers();
+    socket?.close();
+  }
+
+  connect();
+  return { wsUrl, nodeId, ready, close };
+}
+
+function hwpodNodeWsUrl(config: any) {
+  const explicit = text(config.wsUrl);
+  if (explicit) return appendToken(explicit, text(config.token));
+  const cloudUrl = text(config.cloudUrl).replace(/\/+$/u, "");
+  if (!cloudUrl) throw cliError("hwpod_node_ws_url_required", "connect requires --ws-url or --cloud-url");
+  const parsed = new URL(cloudUrl);
+  parsed.protocol = parsed.protocol === "https:" ? "wss:" : "ws:";
+  parsed.pathname = "/v1/hwpod-node/ws";
+  parsed.search = "";
+  return appendToken(parsed.toString(), text(config.token));
+}
+
+function appendToken(urlValue: string, token: string) {
+  if (!token) return urlValue;
+  const parsed = new URL(urlValue);
+  parsed.searchParams.set("token", token);
+  return parsed.toString();
+}
+
 export function createHwpodNodeServer(options: any = {}) {
   const now = options.now ?? (() => new Date().toISOString());
   return createServer(async (request, response) => {
@@ -320,7 +461,8 @@ function help() {
     role: "Single host-side executor for stable hwpod-node-ops; hwpod-spec, hwpod-cli, hwpod-ctl, and hwpod-compiler stay in the Code Agent workspace.",
     usage: [
       "bun tools/hwpod-node.ts run --plan-json '{...}'",
-      "bun tools/hwpod-node.ts serve --host 127.0.0.1 --port 19678"
+      "bun tools/hwpod-node.ts serve --host 127.0.0.1 --port 19678",
+      "bun tools/hwpod-node.ts connect --cloud-url http://74.48.78.17:19667 --node-id node-d601-f103-v2"
     ]
   };
 }