diff --git a/cmd/hwlab-cloud-api/main.ts b/cmd/hwlab-cloud-api/main.ts index 5d038f83..94a230d9 100644 --- a/cmd/hwlab-cloud-api/main.ts +++ b/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); }); } diff --git a/internal/cloud/bun-server.ts b/internal/cloud/bun-server.ts new file mode 100644 index 00000000..3757f7e0 --- /dev/null +++ b/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" } + }); +} diff --git a/internal/cloud/hwpod-node-ops.test.ts b/internal/cloud/hwpod-node-ops.test.ts index 9c70d584..f5e9bf1b 100644 --- a/internal/cloud/hwpod-node-ops.test.ts +++ b/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, timeoutMs: number, message: string) { + let timer: ReturnType | 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); + } +} diff --git a/internal/cloud/hwpod-node-ws-registry.ts b/internal/cloud/hwpod-node-ws-registry.ts new file mode 100644 index 00000000..e268f4f9 --- /dev/null +++ b/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; + firstSeenAt: string; + lastSeenAt: string; +}; + +type PendingDispatch = { + nodeId: string; + requestId: string; + timer: ReturnType; + 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(); + const pending = new Map(); + const socketConnections = new WeakMap(); + + 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 || ""} 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; +} diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index 89b6ca4d..e5ea6bac 100644 --- a/internal/cloud/server.ts +++ b/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), { diff --git a/tools/src/hwpod-node-lib.ts b/tools/src/hwpod-node-lib.ts index 959f6465..ade65596 100644 --- a/tools/src/hwpod-node-lib.ts +++ b/tools/src/hwpod-node-lib.ts @@ -12,12 +12,23 @@ const BOOLEAN_OPTIONS = new Set(["help", "h", "json"]); type ParsedArgs = Record & { _: 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 | null = null; + let reconnectTimer: ReturnType | 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" ] }; }