import { randomUUID } from "node:crypto"; import { ENVIRONMENT_DEV, ERROR_CODES, HwlabProtocolError, JSON_RPC_VERSION, validateRequest, validateResponse } from "../protocol/index.mjs"; import { AUDIT_FIELD_NAMES, CLOUD_API_SERVICE_ID, createAuditRecord, deriveActorFromMeta } from "../audit/index.mjs"; import { buildDbRuntimeReadiness } from "./db-contract.mjs"; 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", "cloud.adapter.describe", M3_IO_RPC_METHODS.doWrite, M3_IO_RPC_METHODS.diRead, "gateway.session.register", "box.resource.register", "box.capability.report", "hardware.operation.request", "hardware.invoke.shell", "audit.event.write", "audit.event.query", "evidence.record.write", "evidence.record.query" ]); const rpcHandlers = new Map([ ["system.health", handleSystemHealth], ["cloud.adapter.describe", handleAdapterDescribe], [M3_IO_RPC_METHODS.doWrite, async (params, envelope, context) => handleM3IoRpc("do.write", params, envelope, context)], [M3_IO_RPC_METHODS.diRead, async (params, envelope, context) => handleM3IoRpc("di.read", params, envelope, context)], ["gateway.session.register", async (params, envelope, context) => getRuntimeStore(context).registerGatewaySession(params, envelope.meta)], ["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", 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)], ["evidence.record.query", async (params, envelope, context) => getRuntimeStore(context).queryEvidenceRecords(params)] ]); const defaultRuntimeStore = createCloudRuntimeStore(); export async function handleJsonRpcRequest(envelope, context = {}) { const requestMeta = extractMeta(envelope); const requestId = envelope?.id ?? "req_unassigned"; const method = typeof envelope?.method === "string" ? envelope.method : "method_unspecified"; try { validateRequest(envelope); } catch (error) { return createErrorEnvelope({ id: requestId, code: ERROR_CODES.invalidRequest, message: "Invalid JSON-RPC request", data: { reason: error.message }, requestMeta, context: { ...context, operation: "json_rpc.validate", target: { type: "rpc_method", id: method } } }); } const handler = rpcHandlers.get(envelope.method); if (!handler) { return createErrorEnvelope({ id: envelope.id, code: ERROR_CODES.methodNotFound, message: `Method ${envelope.method} is not implemented by hwlab-cloud-api`, data: { method: envelope.method }, requestMeta: envelope.meta, context: { ...context, operation: envelope.method, target: { type: "rpc_method", id: envelope.method }, result: "rejected" } }); } try { const result = await handler(envelope.params ?? {}, envelope, context); return createResultEnvelope({ id: envelope.id, result, requestMeta: envelope.meta }); } catch (error) { if (error instanceof HwlabProtocolError) { return createErrorEnvelope({ id: envelope.id, code: error.code, message: error.message, data: error.data, requestMeta: envelope.meta, context: { ...context, ...error.context, operation: envelope.method, target: { type: "rpc_method", id: envelope.method }, result: "rejected" } }); } return createErrorEnvelope({ id: envelope.id, code: ERROR_CODES.internalError, message: "hwlab-cloud-api failed to process the request", data: { reason: error.message }, requestMeta: envelope.meta, context: { ...context, operation: envelope.method, target: { type: "rpc_method", id: envelope.method }, result: "failed" } }); } } export function createResultEnvelope({ id, result, requestMeta = {} }) { const envelope = { jsonrpc: JSON_RPC_VERSION, id, result, meta: createResponseMeta(requestMeta) }; validateResponse(envelope); return envelope; } export function createErrorEnvelope({ id = "req_unassigned", code = ERROR_CODES.hwlabUnknown, message = "HWLAB runtime error", data = {}, requestMeta = {}, context = {} } = {}) { const responseId = id ?? "req_unassigned"; const meta = createResponseMeta(requestMeta); const audit = createAuditRecord({ requestId: responseId, actor: deriveActorFromMeta(requestMeta), source: { serviceId: CLOUD_API_SERVICE_ID, environment: ENVIRONMENT_DEV, traceId: meta.traceId, adapter: context.adapter || "json-rpc" }, operation: context.operation || "json_rpc.error", target: context.target || { type: "cloud_api", id: CLOUD_API_SERVICE_ID }, result: context.result || "failed", timestamp: context.timestamp }); const envelope = { jsonrpc: JSON_RPC_VERSION, id: responseId, error: { code, message, data: { ...normalizeErrorData(data), audit } }, meta }; validateResponse(envelope); return envelope; } export function createResponseMeta(requestMeta = {}) { const meta = { traceId: requestMeta.traceId || `trc_${randomUUID()}`, serviceId: CLOUD_API_SERVICE_ID, environment: ENVIRONMENT_DEV }; if (requestMeta.actorId) { meta.actorId = requestMeta.actorId; } return meta; } async function handleSystemHealth(params, envelope, context) { const env = context.env ?? process.env; const db = await buildDbRuntimeReadiness(env, context.dbProbe); const codeAgent = await describeCodeAgentAvailability(env, context); const runtime = await runtimeReadiness(getRuntimeStore(context)); const readiness = buildCloudApiReadiness({ db, codeAgent, runtime }); return { status: readiness.status, ready: readiness.ready, serviceId: CLOUD_API_SERVICE_ID, environment: ENVIRONMENT_DEV, db, codeAgent, runtime, readiness, blockers: readiness.blockers, blockerCodes: readiness.blockerCodes }; } async function handleAdapterDescribe(params, envelope, context) { const env = context.env ?? process.env; const db = await buildDbRuntimeReadiness(env, context.dbProbe); const codeAgent = await describeCodeAgentAvailability(env, context); const runtime = await runtimeReadiness(getRuntimeStore(context)); const readiness = buildCloudApiReadiness({ db, codeAgent, runtime }); return { serviceId: CLOUD_API_SERVICE_ID, status: readiness.status, ready: readiness.ready, rpcEndpoint: "POST /rpc", restEndpoint: "POST /v1/rpc/{method}", methods: SUPPORTED_RPC_METHODS, auditFields: AUDIT_FIELD_NAMES, db, codeAgent, runtime, readiness, blockers: readiness.blockers, blockerCodes: readiness.blockerCodes }; } 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(); } return runtimeStore.summary(); } function extractMeta(envelope) { if (envelope && typeof envelope === "object" && !Array.isArray(envelope) && envelope.meta) { return envelope.meta; } return {}; } function normalizeErrorData(data) { if (!data || typeof data !== "object" || Array.isArray(data)) { return {}; } return data; }