feat: add outbound gateway demo

2026-05-23 23:39:58 +08:00
parent 194ca572c7
commit ab063880bd
11 changed files with 1149 additions and 65 deletions
@@ -0,0 +1,169 @@
+import { randomUUID } from "node:crypto";
+
+import { ENVIRONMENT_DEV, JSON_RPC_VERSION } from "../protocol/index.mjs";
+import { CLOUD_API_SERVICE_ID } from "../audit/index.mjs";
+
+const DEFAULT_STALE_MS = 30000;
+const DEFAULT_DISPATCH_TIMEOUT_MS = 30000;
+
+export function createGatewayDemoRegistry({
+  now = () => new Date().toISOString(),
+  clock = () => Date.now(),
+  staleMs = DEFAULT_STALE_MS,
+  dispatchTimeoutMs = DEFAULT_DISPATCH_TIMEOUT_MS
+} = {}) {
+  const sessions = new Map();
+  const pending = new Map();
+
+  function updateSession(input = {}) {
+    const gatewaySessionId = input.gatewaySessionId || `gws_${input.gatewayId || "gateway_demo"}`;
+    const gatewayId = input.gatewayId || gatewaySessionId.replace(/^gws_/, "") || "gateway_demo";
+    const previous = sessions.get(gatewaySessionId);
+    const observedAt = now();
+    const session = {
+      ...(previous ?? {}),
+      serviceId: input.serviceId || "hwlab-gateway",
+      projectId: input.projectId || previous?.projectId || "prj_mvp_topology",
+      gatewayId,
+      gatewaySessionId,
+      status: "connected",
+      endpoint: input.endpoint ?? previous?.endpoint ?? null,
+      resourceId: input.resourceId ?? previous?.resourceId ?? "res_windows_host",
+      boxId: input.boxId ?? previous?.boxId ?? "box_windows_host",
+      capabilities: Array.isArray(input.capabilities) ? input.capabilities : previous?.capabilities ?? [],
+      system: input.system ?? previous?.system ?? {},
+      firstSeenAt: previous?.firstSeenAt ?? observedAt,
+      lastSeenAt: observedAt,
+      lastSeenEpochMs: clock(),
+      queue: previous?.queue ?? []
+    };
+    sessions.set(gatewaySessionId, session);
+    return session;
+  }
+
+  function isOnline(gatewaySessionId, { maxAgeMs = staleMs } = {}) {
+    const session = sessions.get(gatewaySessionId);
+    if (!session) return false;
+    return clock() - session.lastSeenEpochMs <= maxAgeMs;
+  }
+
+  function enqueue({ gatewaySessionId, request, timeoutMs = dispatchTimeoutMs }) {
+    const session = sessions.get(gatewaySessionId);
+    if (!session || !isOnline(gatewaySessionId)) {
+      return Promise.resolve({
+        ok: false,
+        status: "not_connected",
+        error: `gatewaySessionId ${gatewaySessionId} is not connected`
+      });
+    }
+
+    const requestId = request.id ?? `req_gateway_demo_${randomUUID()}`;
+    const outbound = {
+      ...request,
+      id: requestId
+    };
+
+    return new Promise((resolve) => {
+      const timer = setTimeout(() => {
+        pending.delete(requestId);
+        resolve({
+          ok: false,
+          status: "timed_out",
+          error: `gateway dispatch timed out after ${timeoutMs}ms`,
+          request: outbound
+        });
+      }, timeoutMs);
+
+      pending.set(requestId, {
+        gatewaySessionId,
+        resolve,
+        timer,
+        request: outbound,
+        createdAt: now()
+      });
+      session.queue.push(outbound);
+    });
+  }
+
+  function nextRequest(gatewaySessionId) {
+    const session = sessions.get(gatewaySessionId);
+    return session?.queue.shift() ?? null;
+  }
+
+  function complete(input = {}) {
+    const response = input.response ?? input;
+    const requestId = response.id;
+    const waiter = pending.get(requestId);
+    if (!waiter) {
+      return {
+        accepted: false,
+        status: "unknown_request",
+        requestId: requestId ?? null
+      };
+    }
+
+    clearTimeout(waiter.timer);
+    pending.delete(requestId);
+    waiter.resolve({
+      ok: !response.error,
+      status: response.error ? "failed" : "completed",
+      response
+    });
+
+    return {
+      accepted: true,
+      status: "completed",
+      requestId
+    };
+  }
+
+  function describe() {
+    return {
+      serviceId: CLOUD_API_SERVICE_ID,
+      environment: ENVIRONMENT_DEV,
+      mode: "gateway-outbound-poll-demo",
+      staleMs,
+      dispatchTimeoutMs,
+      sessions: [...sessions.values()].map((session) => ({
+        serviceId: session.serviceId,
+        projectId: session.projectId,
+        gatewayId: session.gatewayId,
+        gatewaySessionId: session.gatewaySessionId,
+        status: isOnline(session.gatewaySessionId) ? "online" : "stale",
+        resourceId: session.resourceId,
+        boxId: session.boxId,
+        capabilityCount: session.capabilities.length,
+        queueDepth: session.queue.length,
+        firstSeenAt: session.firstSeenAt,
+        lastSeenAt: session.lastSeenAt,
+        system: session.system
+      })),
+      pendingCount: pending.size
+    };
+  }
+
+  return {
+    updateSession,
+    isOnline,
+    enqueue,
+    nextRequest,
+    complete,
+    describe,
+    getSession: (gatewaySessionId) => sessions.get(gatewaySessionId) ?? null
+  };
+}
+
+export function createGatewayShellRequest({ id, params = {}, meta = {} }) {
+  return {
+    jsonrpc: JSON_RPC_VERSION,
+    id: id ?? `req_gateway_shell_${randomUUID()}`,
+    method: "hardware.invoke.shell",
+    params,
+    meta: {
+      traceId: meta.traceId ?? params.traceId ?? `trc_gateway_shell_${randomUUID()}`,
+      actorId: meta.actorId,
+      serviceId: CLOUD_API_SERVICE_ID,
+      environment: ENVIRONMENT_DEV
+    }
+  };
+}
@@ -19,6 +19,7 @@ import { describeCodeAgentAvailability } from "./code-agent-chat.mjs";
 import { buildCloudApiReadiness } from "./health-contract.mjs";
 import { createCloudRuntimeStore } from "../db/runtime-store.mjs";
 import { M3_IO_RPC_METHODS, handleM3IoRpc } from "./m3-io-control.mjs";
+import { createGatewayShellRequest } from "./gateway-demo-registry.mjs";

 export const SUPPORTED_RPC_METHODS = Object.freeze([
  "system.health",
@@ -45,7 +46,7 @@ const rpcHandlers = new Map([
  ["box.resource.register", async (params, envelope, context) => getRuntimeStore(context).registerBoxResource(params, envelope.meta)],
  ["box.capability.report", async (params, envelope, context) => getRuntimeStore(context).reportBoxCapabilities(params, envelope.meta)],
  ["hardware.operation.request", async (params, envelope, context) => getRuntimeStore(context).requestHardwareOperation(params, envelope.meta)],
-  ["hardware.invoke.shell", async (params, envelope, context) => getRuntimeStore(context).invokeHardwareShell(params, envelope.meta)],
+  ["hardware.invoke.shell", handleHardwareInvokeShell],
  ["audit.event.write", async (params, envelope, context) => getRuntimeStore(context).writeAuditEvent(params, envelope.meta)],
  ["audit.event.query", async (params, envelope, context) => getRuntimeStore(context).queryAuditEvents(params)],
  ["evidence.record.write", async (params, envelope, context) => getRuntimeStore(context).writeEvidenceRecord(params, envelope.meta)],
@@ -271,6 +272,71 @@ function getRuntimeStore(context = {}) {
  return context.runtimeStore || defaultRuntimeStore;
 }

+async function handleHardwareInvokeShell(params, envelope, context) {
+  const registry = context.gatewayRegistry;
+  const gatewaySessionId = params.gatewaySessionId;
+  if (!registry?.isOnline?.(gatewaySessionId)) {
+    return getRuntimeStore(context).invokeHardwareShell(params, envelope.meta);
+  }
+
+  const dispatch = await registry.enqueue({
+    gatewaySessionId,
+    request: createGatewayShellRequest({
+      id: envelope.id,
+      params,
+      meta: envelope.meta
+    }),
+    timeoutMs: parseDispatchTimeout(context.env ?? process.env)
+  });
+
+  if (!dispatch.ok) {
+    return {
+      accepted: true,
+      status: dispatch.status === "timed_out" ? "timed_out" : "accepted",
+      operationId: params.operationId ?? null,
+      gatewaySessionId,
+      resourceId: params.resourceId ?? null,
+      capabilityId: params.capabilityId ?? null,
+      dispatch: {
+        shellExecuted: false,
+        dispatchStatus: dispatch.status,
+        message: dispatch.error ?? dispatch.response?.error?.message ?? "gateway dispatch failed",
+        error: dispatch.response?.error ?? null
+      }
+    };
+  }
+
+  const gatewayResult = dispatch.response?.result ?? {};
+  return {
+    accepted: gatewayResult.accepted !== false,
+    status: gatewayResult.status ?? statusFromShellResult(gatewayResult),
+    operationId: gatewayResult.operationId ?? params.operationId ?? null,
+    gatewaySessionId: gatewayResult.gatewaySessionId ?? gatewaySessionId,
+    resourceId: params.resourceId ?? null,
+    capabilityId: params.capabilityId ?? null,
+    dispatch: {
+      shellExecuted: gatewayResult.shellExecuted === true,
+      dispatchStatus: gatewayResult.dispatchStatus ?? statusFromShellResult(gatewayResult),
+      ...gatewayResult
+    },
+    auditId: gatewayResult.auditId ?? null,
+    evidenceId: gatewayResult.evidenceId ?? null,
+    gateway: gatewayResult.gateway ?? null
+  };
+}
+
+function statusFromShellResult(result = {}) {
+  if (result.timedOut) return "timed_out";
+  if (result.shellExecuted === false) return "rejected";
+  if (Number.isInteger(result.exitCode)) return result.exitCode === 0 ? "succeeded" : "failed";
+  return "completed";
+}
+
+function parseDispatchTimeout(env) {
+  const parsed = Number.parseInt(env.HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS ?? "", 10);
+  return Number.isInteger(parsed) && parsed > 0 ? parsed : 30000;
+}
+
 async function runtimeReadiness(runtimeStore) {
  if (typeof runtimeStore?.readiness === "function") {
    return runtimeStore.readiness();
@@ -32,15 +32,20 @@ import {
  handleM3IoControl
 } from "./m3-io-control.mjs";
 import { createConfiguredCloudRuntimeStore } from "../db/runtime-store.mjs";
+import { createGatewayDemoRegistry } from "./gateway-demo-registry.mjs";

 const DEFAULT_BODY_LIMIT_BYTES = 1024 * 1024;

 export function createCloudApiServer(options = {}) {
  const env = options.env ?? process.env;
  const runtimeStore = options.runtimeStore || createConfiguredCloudRuntimeStore({ ...options, env });
+  const gatewayRegistry = options.gatewayRegistry || createGatewayDemoRegistry({
+    staleMs: parsePositiveInteger(env.HWLAB_GATEWAY_DEMO_STALE_MS, 30000),
+    dispatchTimeoutMs: parsePositiveInteger(env.HWLAB_GATEWAY_DISPATCH_TIMEOUT_MS, 30000)
+  });
  return createServer(async (request, response) => {
    try {
-      await routeRequest(request, response, { ...options, env, runtimeStore });
+      await routeRequest(request, response, { ...options, env, runtimeStore, gatewayRegistry });
    } catch (error) {
      sendJson(response, 500, {
        error: {
@@ -166,7 +171,8 @@ async function handleRpcHttpRequest(request, response, options) {
    runtimeStore: options.runtimeStore,
    env: options.env,
    dbProbe: options.dbProbe,
-    m3IoRequestJson: options.m3IoRequestJson
+    m3IoRequestJson: options.m3IoRequestJson,
+    gatewayRegistry: options.gatewayRegistry
  });
  sendJson(response, statusForRpcResponse(rpcResponse), rpcResponse);
 }
@@ -212,6 +218,21 @@ async function handleRestAdapter(request, response, url, options) {
    return;
  }

+  if (request.method === "GET" && url.pathname === "/v1/gateway/sessions") {
+    sendJson(response, 200, options.gatewayRegistry.describe());
+    return;
+  }
+
+  if (request.method === "POST" && url.pathname === "/v1/gateway/poll") {
+    await handleGatewayPollHttp(request, response, options);
+    return;
+  }
+
+  if (request.method === "POST" && url.pathname === "/v1/gateway/result") {
+    await handleGatewayResultHttp(request, response, options);
+    return;
+  }
+
  if (request.method === "POST" && url.pathname === M3_IO_CONTROL_ROUTE) {
    await handleM3IoControlHttp(request, response, options);
    return;
@@ -282,13 +303,145 @@ async function handleRestAdapter(request, response, url, options) {
      runtimeStore: options.runtimeStore,
      env: options.env,
      dbProbe: options.dbProbe,
-      m3IoRequestJson: options.m3IoRequestJson
+      m3IoRequestJson: options.m3IoRequestJson,
+      gatewayRegistry: options.gatewayRegistry
    }
  );

  sendJson(response, statusForRpcResponse(rpcResponse), rpcResponse);
 }

+async function handleGatewayPollHttp(request, response, options) {
+  const body = await readBody(request, options.bodyLimitBytes);
+  let params = {};
+
+  try {
+    params = body ? JSON.parse(body) : {};
+  } catch (error) {
+    sendJson(response, 400, {
+      accepted: false,
+      error: {
+        code: "parse_error",
+        message: "Invalid JSON body",
+        reason: error.message
+      }
+    });
+    return;
+  }
+
+  if (!params || typeof params !== "object" || Array.isArray(params)) {
+    sendJson(response, 400, {
+      accepted: false,
+      error: {
+        code: "invalid_params",
+        message: "gateway poll body must be a JSON object"
+      }
+    });
+    return;
+  }
+
+  const session = options.gatewayRegistry.updateSession(params);
+  await persistGatewayDemoSession({
+    runtimeStore: options.runtimeStore,
+    session,
+    request,
+    env: options.env
+  });
+  const outbound = options.gatewayRegistry.nextRequest(session.gatewaySessionId);
+  sendJson(response, 200, {
+    accepted: true,
+    serviceId: CLOUD_API_SERVICE_ID,
+    mode: "gateway-outbound-poll-demo",
+    gatewayId: session.gatewayId,
+    gatewaySessionId: session.gatewaySessionId,
+    queueDepth: session.queue.length,
+    request: outbound,
+    type: outbound ? "request" : "noop",
+    observedAt: new Date().toISOString()
+  });
+}
+
+async function handleGatewayResultHttp(request, response, options) {
+  const body = await readBody(request, options.bodyLimitBytes);
+  let params = {};
+
+  try {
+    params = body ? JSON.parse(body) : {};
+  } catch (error) {
+    sendJson(response, 400, {
+      accepted: false,
+      error: {
+        code: "parse_error",
+        message: "Invalid JSON body",
+        reason: error.message
+      }
+    });
+    return;
+  }
+
+  const result = options.gatewayRegistry.complete(params);
+  sendJson(response, result.accepted ? 200 : 404, {
+    ...result,
+    serviceId: CLOUD_API_SERVICE_ID,
+    mode: "gateway-outbound-poll-demo",
+    observedAt: new Date().toISOString()
+  });
+}
+
+async function persistGatewayDemoSession({ runtimeStore, session, request, env = process.env }) {
+  if (!runtimeStore) return;
+  const refreshMs = parsePositiveInteger(env.HWLAB_GATEWAY_REGISTRATION_REFRESH_MS, 60000);
+  if (session.runtimeRegisteredEpochMs && Date.now() - session.runtimeRegisteredEpochMs < refreshMs) {
+    return;
+  }
+  const meta = {
+    traceId: getHeader(request, "x-trace-id") || `trc_gateway_poll_${randomUUID()}`,
+    actorId: `svc_${session.serviceId}`,
+    serviceId: CLOUD_API_SERVICE_ID,
+    environment: ENVIRONMENT_DEV
+  };
+
+  try {
+    await runtimeStore.registerGatewaySession?.({
+      projectId: session.projectId,
+      gatewaySessionId: session.gatewaySessionId,
+      serviceId: "hwlab-gateway",
+      gatewayId: session.gatewayId,
+      endpoint: session.endpoint,
+      status: "connected",
+      labels: {
+        demoOutboundPoll: "true",
+        os: String(session.system?.platform ?? "unknown")
+      }
+    }, meta);
+    await runtimeStore.registerBoxResource?.({
+      projectId: session.projectId,
+      gatewaySessionId: session.gatewaySessionId,
+      resourceId: session.resourceId,
+      boxId: session.boxId,
+      resourceType: "simulator_endpoint",
+      state: "available",
+      metadata: {
+        serviceId: "hwlab-gateway",
+        demoShellHost: true
+      }
+    }, meta);
+    if (session.capabilities.length > 0) {
+      await runtimeStore.reportBoxCapabilities?.({
+        capabilities: session.capabilities.map((capability) => ({
+          ...capability,
+          resourceId: capability.resourceId ?? session.resourceId,
+          projectId: capability.projectId ?? session.projectId
+        }))
+      }, meta);
+    }
+    session.runtimeRegisteredAt = new Date().toISOString();
+    session.runtimeRegisteredEpochMs = Date.now();
+  } catch {
+    // Demo polling must keep the gateway online even when durable runtime writes are not ready.
+  }
+}
+
 async function handleCodeAgentChatHttp(request, response, options) {
  const body = await readBody(request, options.bodyLimitBytes);
  let params = {};
@@ -499,6 +652,11 @@ function getHeader(request, name) {
  return value;
 }

+function parsePositiveInteger(value, fallback) {
+  const parsed = Number.parseInt(value ?? "", 10);
+  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
+}
+
 async function runtimeReadiness(runtimeStore) {
  if (typeof runtimeStore?.readiness === "function") {
    return runtimeStore.readiness();