pikasTech-HWLAB/internal/cloud/hwpod-topology-read-model.ts

/*
 * SPEC: PJ2026-010405 云端控制台；PJ2026-010103 HWPOD 服务。
 * Implementation reference: draft-2026-07-13-p0-cloud-console。
 * Responsibility: 由 Cloud API HWPOD domain 生成有界 typed Node/设备 topology，不让浏览器拼装业务真相。
 */

export const HWPOD_TOPOLOGY_CONTRACT_VERSION = "hwpod-topology-v1";

const DEFAULT_PAGE_LIMIT = 50;
const MAX_PAGE_LIMIT = 100;

export const HWPOD_NODE_READINESS_DEFINITIONS = [
  { id: "downloaded", label: "已下载", authority: "user-device", nextAction: { label: "从节点接入页下载并校验已发布 Python 单文件", entrypoint: "cloud-web-hwpod-onboarding" } },
  { id: "installed", label: "已安装", authority: "python-node-runtime", nextAction: { label: "按 owning YAML 选择的交互用户与原生 Python 启动节点", entrypoint: "yaml-managed-hwpod-node" } },
  { id: "desktop-visible", label: "桌面可见", authority: "python-node-runtime", nextAction: { label: "在目标交互登录会话确认窗口或托盘可见", entrypoint: "hwlab-node-ui" } },
  { id: "registered", label: "已注册", authority: "websocket-registry", nextAction: { label: "保持节点运行并通过受控节点状态入口检查注册", entrypoint: "hwpod-node-status" } },
  { id: "workspace-ready", label: "工作区就绪", authority: "hwpod-spec-probe", nextAction: { label: "通过显式 readiness 探测检查 owning workspace", entrypoint: "hwpod-readiness-probe" } },
  { id: "hwpod-ready", label: "HWPOD 就绪", authority: "hwpod-topology", nextAction: { label: "检查挂载 HWPOD 的 blocker、能力与占用状态", entrypoint: "hwpod-topology" } }
] as const;

export function buildHwpodTopologyReadModel(specs: any[], registrySnapshot: any = {}, options: any = {}) {
  const observedAt = text(options.observedAt) || new Date().toISOString();
  const onlineNodes = normalizeOnlineNodes(registrySnapshot);
  const invalidSpecs = specs.filter((spec) => spec?.ok === false).map((spec) => invalidSpecSummary(spec));
  const validSpecs = specs.filter((spec) => spec?.ok !== false && text(spec?.nodeId) && text(spec?.hwpodId ?? spec?.name));
  const devices = validSpecs.map((spec) => buildDevice(spec, onlineNodes.get(text(spec.nodeId)), observedAt));
  const nodeIds = new Set([...onlineNodes.keys(), ...validSpecs.map((spec) => text(spec.nodeId)).filter(Boolean)]);
  const nodes = [...nodeIds].map((nodeId) => buildNode(nodeId, onlineNodes.get(nodeId), devices.filter((device) => device.nodeId === nodeId), observedAt));
  const resource = text(options.resource) === "device" ? "device" : "node";
  const sourceItems = resource === "device" ? devices : nodes;
  const filteredItems = filterItems(sourceItems, options);
  const sortedItems = sortItems(filteredItems, options);
  const page = paginate(sortedItems, options);
  const partial = invalidSpecs.length > 0;

  return {
    ok: true,
    status: partial ? "partial" : "ready",
    contractVersion: HWPOD_TOPOLOGY_CONTRACT_VERSION,
    route: "/v1/hwpod/topology",
    owner: "cloud-api-hwpod-domain",
    resource,
    service: {
      ready: true,
      status: partial ? "partial" : "ready",
      websocketMode: text(registrySnapshot?.mode) || "hwpod-node-outbound-native-ws",
      connectedNodeCount: onlineNodes.size,
      pendingOperationCount: integer(registrySnapshot?.pendingCount),
      invalidSpecCount: invalidSpecs.length
    },
    summary: {
      nodeCount: nodes.length,
      onlineNodeCount: nodes.filter((node) => node.online).length,
      offlineNodeCount: nodes.filter((node) => !node.online).length,
      unconfiguredNodeCount: nodes.filter((node) => node.deviceCount === 0).length,
      deviceCount: devices.length,
      availableDeviceCount: devices.filter((device) => device.available).length,
      busyDeviceCount: devices.filter((device) => device.busy).length,
      capabilityMismatchCount: devices.filter((device) => device.capabilityMismatch).length,
      invalidSpecCount: invalidSpecs.length
    },
    items: page.items,
    page: page.page,
    invalidSpecs,
    valuesRedacted: true,
    observedAt
  };
}

function normalizeOnlineNodes(snapshot: any) {
  const map = new Map<string, any>();
  for (const raw of array(snapshot?.nodes)) {
    const nodeId = text(raw?.nodeId);
    if (!nodeId) continue;
    const actualCapabilities = uniqueStrings(raw?.capabilities);
    const inFlightCount = integer(raw?.inFlightCount ?? raw?.pendingCount);
    const maxInFlight = positiveInteger(raw?.maxInFlight, 1);
    map.set(nodeId, {
      nodeId,
      name: text(raw?.name) || nodeId,
      online: raw?.status !== "offline",
      platform: text(raw?.platform ?? raw?.labels?.platform) || "unknown",
      version: text(raw?.version ?? raw?.labels?.version) || null,
      runtimeKind: text(raw?.runtimeKind ?? raw?.runtime?.kind ?? raw?.labels?.runtimeKind) || null,
      installed: booleanOrNull(raw?.installed ?? raw?.runtime?.installed),
      desktopVisible: booleanOrNull(raw?.desktopVisible ?? raw?.runtime?.desktopVisible),
      actualCapabilities,
      inFlightCount,
      maxInFlight,
      busy: raw?.busy === true || inFlightCount >= maxInFlight,
      firstSeenAt: text(raw?.firstSeenAt) || null,
      lastSeenAt: text(raw?.lastSeenAt) || null,
      ageMs: integer(raw?.ageMs),
      latestOperation: operationSummary(raw?.latestOperation),
      latestReadinessProbe: operationSummary(raw?.latestReadinessProbe),
      diagnostic: diagnosticSummary(raw?.lastDiagnostic)
    });
  }
  return map;
}

function buildDevice(spec: any, node: any, observedAt: string) {
  const document = object(spec.document);
  const body = object(document.spec);
  const declaredCapabilities = declaredCapabilitiesFor(body);
  const actualCapabilities = node?.actualCapabilities ?? [];
  const missingCapabilities = node ? declaredCapabilities.filter((capability) => !actualCapabilities.includes(capability)) : declaredCapabilities;
  const capabilityCoverage = buildCapabilityCoverage(declaredCapabilities, actualCapabilities);
  const availability = availabilitySummary(spec.availability);
  const online = Boolean(node?.online);
  const busy = Boolean(node?.busy);
  const capabilityMismatch = online && missingCapabilities.length > 0;
  const blockers = deviceBlockers({ online, busy, capabilityMismatch, missingCapabilities, availability });
  const status = online ? "online" : "offline";
  const available = online && !busy && !capabilityMismatch && availability.ok === true;
  return {
    resourceType: "hwpod",
    hwpodId: text(spec.hwpodId ?? spec.name),
    uid: text(spec.uid) || null,
    name: text(spec.name ?? spec.hwpodId),
    nodeId: text(spec.nodeId),
    nodeName: node?.name ?? text(spec.nodeId),
    status,
    online,
    available,
    busy,
    capabilityMismatch,
    declaredCapabilities,
    actualCapabilities,
    missingCapabilities,
    capabilityCoverage,
    elements: hwpodElements(body),
    availability,
    operation: node?.latestOperation?.hwpodId === text(spec.hwpodId ?? spec.name) ? node.latestOperation : null,
    diagnostic: node?.diagnostic ?? null,
    blockers,
    source: {
      authority: text(spec.authority) || "workspace-or-registry",
      kind: text(spec.source?.kind) || "workspace-or-registry",
      specPath: text(spec.specPath) || null
    },
    observedAt
  };
}

function buildNode(nodeId: string, onlineNode: any, devices: any[], observedAt: string) {
  const declaredCapabilities = uniqueStrings(devices.flatMap((device) => device.declaredCapabilities));
  const actualCapabilities = onlineNode?.actualCapabilities ?? [];
  const missingCapabilities = onlineNode ? declaredCapabilities.filter((capability) => !actualCapabilities.includes(capability)) : declaredCapabilities;
  const capabilityCoverage = buildCapabilityCoverage(declaredCapabilities, actualCapabilities);
  const busy = Boolean(onlineNode?.busy);
  const online = Boolean(onlineNode?.online);
  const capabilityMismatch = online && missingCapabilities.length > 0;
  const hasSpec = devices.length > 0;
  const status = online ? "online" : "offline";
  const blockers = nodeBlockers({ online, hasSpec, busy, capabilityMismatch, missingCapabilities, devices });
  const workspaceReady = devices.some((device) => device.availability.ok === true);
  const hwpodReady = devices.some((device) => device.available);
  return {
    resourceType: "hwpod-node",
    nodeId,
    name: onlineNode?.name ?? nodeId,
    status,
    online,
    platform: onlineNode?.platform ?? "unknown",
    version: onlineNode?.version ?? null,
    runtimeKind: onlineNode?.runtimeKind ?? null,
    firstSeenAt: onlineNode?.firstSeenAt ?? null,
    lastSeenAt: onlineNode?.lastSeenAt ?? null,
    ageMs: onlineNode?.ageMs ?? null,
    deviceCount: devices.length,
    hwpods: devices.map((device) => ({
      hwpodId: device.hwpodId,
      name: device.name,
      status: device.status,
      busy: device.busy,
      capabilityMismatch: device.capabilityMismatch
    })),
    declaredCapabilities,
    actualCapabilities,
    missingCapabilities,
    capabilityCoverage,
    capabilityMismatch,
    inFlight: {
      count: onlineNode?.inFlightCount ?? 0,
      max: onlineNode?.maxInFlight ?? 1,
      busy
    },
    operation: onlineNode?.latestOperation ?? null,
    readinessProbe: onlineNode?.latestReadinessProbe ?? null,
    diagnostic: onlineNode?.diagnostic ?? null,
    readiness: nodeReadiness({ onlineNode, online, workspaceReady, hwpodReady, busy, blockers }),
    blockers,
    observedAt
  };
}

function declaredCapabilitiesFor(spec: any) {
  const explicit = uniqueStrings([...array(spec?.capabilities), ...array(spec?.nodeBinding?.capabilities)]);
  if (explicit.length > 0) return explicit;
  const capabilities = ["node.health", "node.version", "node.inventory", "node.diagnostics"];
  if (Object.keys(object(spec?.workspace)).length > 0) {
    capabilities.push("workspace.ls", "workspace.cat", "workspace.rg", "workspace.apply-patch", "workspace.write", "workspace.replace", "workspace.insert-after", "cmd.run");
  }
  if (Object.keys(object(spec?.debugProbe)).length > 0) capabilities.push("debug.build", "debug.download", "debug.reset");
  if (Object.keys(object(spec?.ioProbe?.uart)).length > 0) capabilities.push("io.uart.read", "io.uart.write", "io.uart.jsonrpc");
  return uniqueStrings(capabilities);
}

function hwpodElements(spec: any) {
  const target = object(spec.targetDevice);
  const workspace = object(spec.workspace);
  const debugProbe = object(spec.debugProbe);
  const ioProbe = object(spec.ioProbe);
  const uart = object(ioProbe.uart);
  return {
    target: {
      label: text(target.board ?? target.name ?? target.id) || "未声明目标板",
      mcu: text(target.mcu) || null
    },
    workspace: {
      label: text(workspace.projectRoot ?? workspace.path) || "未声明工作区",
      path: text(workspace.path) || null,
      toolchain: text(workspace.toolchain) || null
    },
    debugProbe: {
      label: text(debugProbe.probeName ?? debugProbe.name ?? debugProbe.type ?? debugProbe.id) || "未声明调试探头",
      type: text(debugProbe.type ?? debugProbe.kind) || null
    },
    ioProbe: {
      label: text(uart.id ?? uart.port ?? ioProbe.name ?? ioProbe.id) || "未声明 IO 探头",
      kind: Object.keys(uart).length > 0 ? "uart" : text(ioProbe.kind) || "unknown"
    }
  };
}

function availabilitySummary(value: any) {
  const availability = object(value);
  const blocker = object(availability.blocker);
  return {
    ok: booleanOrNull(availability.ok),
    status: text(availability.status) || "unprobed",
    checkedAt: text(availability.checkedAt) || null,
    blocker: Object.keys(blocker).length > 0 ? {
      code: text(blocker.code) || "hwpod_unavailable",
      summary: text(blocker.summary ?? blocker.message) || "HWPOD 不可用",
      retryable: blocker.retryable !== false
    } : null
  };
}

function buildCapabilityCoverage(declaredCapabilities: string[], actualCapabilities: string[]) {
  const matched = declaredCapabilities.filter((capability) => actualCapabilities.includes(capability)).length;
  const extra = actualCapabilities.filter((capability) => !declaredCapabilities.includes(capability)).length;
  return {
    matched,
    declared: declaredCapabilities.length,
    actual: actualCapabilities.length,
    missing: declaredCapabilities.length - matched,
    extra
  };
}

function deviceBlockers(input: any) {
  const blockers: any[] = [];
  if (!input.online) blockers.push(blocker("hwpod_node_offline", "声明的 HWPOD Node 当前离线。", true));
  if (input.capabilityMismatch) blockers.push(blocker("hwpod_capability_mismatch", `节点缺少 ${input.missingCapabilities.length} 项声明能力。`, false));
  if (input.busy) blockers.push(blocker("hwpod_node_busy", "节点已达 in-flight 并发上限。", true));
  if (input.availability.blocker) blockers.push(input.availability.blocker);
  return dedupeBlockers(blockers);
}

function nodeBlockers(input: any) {
  const blockers: any[] = [];
  if (!input.online) blockers.push(blocker("hwpod_node_offline", "Owning 配置已声明该节点，但未观测到主动出站 WebSocket。", true));
  if (input.online && !input.hasSpec) blockers.push(blocker("hwpod_node_spec_missing", "节点已在线，但没有任何 HWPOD spec 绑定。", false));
  if (input.capabilityMismatch) blockers.push(blocker("hwpod_capability_mismatch", `节点缺少 ${input.missingCapabilities.length} 项所属 HWPOD 声明能力。`, false));
  if (input.busy) blockers.push(blocker("hwpod_node_busy", "节点已达 in-flight 并发上限。", true));
  for (const device of input.devices) blockers.push(...device.blockers);
  return dedupeBlockers(blockers);
}

function nodeReadiness(input: any) {
  const localObserved = Boolean(input.onlineNode);
  const stageBlocker = input.blockers[0] ?? null;
  return [
    readinessStage("downloaded", input.onlineNode?.installed === true ? "ready" : "unknown", input.onlineNode?.installed === true ? null : blocker("hwpod_download_not_observed", "云端不以浏览器下载动作作为安装事实。", false)),
    readinessStage("installed", input.onlineNode?.installed === true ? "ready" : "unknown", input.onlineNode?.installed === true ? null : blocker("hwpod_install_not_observed", "等待 Python 节点报告安装运行时。", true)),
    readinessStage("desktop-visible", input.onlineNode?.desktopVisible === true ? "ready" : localObserved ? "blocked" : "unknown", input.onlineNode?.desktopVisible === true ? null : blocker("hwpod_desktop_not_visible", "尚未证明图形界面与托盘属于交互登录会话。", true)),
    readinessStage("registered", input.online ? "ready" : "blocked", input.online ? null : blocker("hwpod_node_offline", "未观测到已认证的主动出站 WebSocket。", true)),
    readinessStage("workspace-ready", input.workspaceReady ? "ready" : "blocked", input.workspaceReady ? null : stageBlocker ?? blocker("hwpod_workspace_not_ready", "工作区尚未通过显式 readiness 探测。", true)),
    readinessStage("hwpod-ready", input.hwpodReady && !input.busy ? "ready" : "blocked", input.hwpodReady && !input.busy ? null : stageBlocker ?? blocker("hwpod_not_ready", "尚未观测到可用且空闲的挂载 HWPOD。", true))
  ];
}

function readinessStage(id: string, status: string, stageBlocker: any) {
  const definition = HWPOD_NODE_READINESS_DEFINITIONS.find((stage) => stage.id === id)!;
  return { ...definition, status, blocker: stageBlocker, nextAction: status === "ready" ? null : definition.nextAction };
}

function invalidSpecSummary(spec: any) {
  return {
    specPath: text(spec?.specPath) || null,
    status: "invalid",
    blocker: blocker(text(spec?.error?.code) || "invalid_hwpod_spec", text(spec?.error?.message) || "HWPOD spec 无法解析。", false)
  };
}

function operationSummary(value: any) {
  const operation = object(value);
  if (Object.keys(operation).length === 0) return null;
  return {
    requestId: text(operation.requestId) || null,
    planId: text(operation.planId) || null,
    hwpodId: text(operation.hwpodId) || null,
    status: text(operation.status) || "unknown",
    opCount: integer(operation.opCount),
    startedAt: text(operation.startedAt) || null,
    finishedAt: text(operation.finishedAt) || null,
    blockerCode: text(operation.blockerCode) || null
  };
}

function diagnosticSummary(value: any) {
  const diagnostic = object(value);
  if (Object.keys(diagnostic).length === 0) return null;
  return {
    level: text(diagnostic.level) || "ERROR",
    source: text(diagnostic.source) || "unknown",
    message: redactDiagnosticMessage(diagnostic.message),
    observedAt: text(diagnostic.observedAt ?? diagnostic.receivedAt) || null,
    valuesRedacted: true
  };
}

function redactDiagnosticMessage(value: unknown) {
  return text(value)
    .replace(/\b(Bearer|Basic)\s+\S+/giu, "$1 [REDACTED]")
    .replace(/([?&](?:api[_-]?key|token|secret|password|authorization)=)[^&\s]+/giu, "$1[REDACTED]")
    .replace(/\b(api[_-]?key|token|secret|password|authorization)\s*[:=]\s*\S+/giu, "$1=[REDACTED]")
    .slice(0, 500);
}

function filterItems(items: any[], options: any) {
  const query = text(options.q).toLowerCase();
  const statuses = new Set(uniqueStrings(String(options.status ?? "").split(",")));
  return items.filter((item) => {
    if (statuses.size > 0 && !statuses.has(text(item.status))) return false;
    if (!query) return true;
    const haystack = [item.nodeId, item.name, item.hwpodId, item.status, ...array(item.missingCapabilities)].map(text).join(" ").toLowerCase();
    return haystack.includes(query);
  });
}

function sortItems(items: any[], options: any) {
  const sort = ["name", "status", "lastSeenAt", "deviceCount"].includes(text(options.sort)) ? text(options.sort) : "name";
  const direction = text(options.direction) === "desc" ? -1 : 1;
  return [...items].sort((left, right) => compareValues(left?.[sort], right?.[sort]) * direction);
}

function paginate(items: any[], options: any) {
  const limit = Math.min(positiveInteger(options.limit, DEFAULT_PAGE_LIMIT), MAX_PAGE_LIMIT);
  const offset = decodeCursor(options.cursor);
  const pageItems = items.slice(offset, offset + limit);
  const nextOffset = offset + pageItems.length;
  return {
    items: pageItems,
    page: {
      limit,
      returned: pageItems.length,
      total: items.length,
      cursor: offset > 0 ? encodeCursor(offset) : null,
      previousCursor: offset > 0 ? encodeCursor(Math.max(0, offset - limit)) : null,
      nextCursor: nextOffset < items.length ? encodeCursor(nextOffset) : null
    }
  };
}

function encodeCursor(offset: number) {
  return Buffer.from(`offset:${offset}`, "utf8").toString("base64url");
}

function decodeCursor(value: unknown) {
  const cursor = text(value);
  if (!cursor) return 0;
  try {
    const parsed = Buffer.from(cursor, "base64url").toString("utf8").match(/^offset:(\d+)$/u);
    return parsed ? integer(parsed[1]) : 0;
  } catch {
    return 0;
  }
}

function blocker(code: string, summary: string, retryable: boolean) {
  return { code, summary, retryable };
}

function dedupeBlockers(values: any[]) {
  const seen = new Set<string>();
  return values.filter((value) => {
    const code = text(value?.code);
    if (!code || seen.has(code)) return false;
    seen.add(code);
    return true;
  }).slice(0, 8);
}

function compareValues(left: unknown, right: unknown) {
  if (typeof left === "number" || typeof right === "number") return Number(left ?? 0) - Number(right ?? 0);
  return text(left).localeCompare(text(right), "zh-CN");
}

function uniqueStrings(values: unknown) {
  return [...new Set(array(values).map(text).filter(Boolean))].sort();
}

function array(value: unknown): any[] {
  return Array.isArray(value) ? value : [];
}

function object(value: unknown): any {
  return value && typeof value === "object" && !Array.isArray(value) ? value : {};
}

function text(value: unknown) {
  return typeof value === "string" && value.trim() ? value.trim() : "";
}

function integer(value: unknown) {
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : 0;
}

function positiveInteger(value: unknown, fallback: number) {
  const parsed = Number(value);
  return Number.isInteger(parsed) && parsed > 0 ? parsed : fallback;
}

function booleanOrNull(value: unknown) {
  return typeof value === "boolean" ? value : null;
}