Publish Weksa health over Idunn RUDP

docs/verse-service-contract.md

@@ -68,10 +68,13 @@ commit when the runtime lands.
 
 The local Starfire daemon is launched by `scripts/start-weksa-daemon.ps1`,
 checked by `scripts/health-weksa-daemon.cmd`, and restarted by
-`scripts/restart-weksa-daemon.cmd`. Its compatibility HTTP surface listens on
+`scripts/restart-weksa-daemon.cmd`. Its daemon-owned health path publishes
-`http://127.0.0.1:8813` with `/health`, `/provider-advertisement`,
+`weksa.cultnet-rudp-provider-health` to Idunn over
-`/operator-state`, `/eve/operator`, `/cultmesh/publications`, and
+`cultnet.transport.rudp.v0` after each serialized witness refresh. Its
-`POST /speech-provider/mimo/voicedesign`.
+compatibility HTTP surface still listens on `http://127.0.0.1:8813` with
+`/health`, `/provider-advertisement`, `/operator-state`, `/eve/operator`,
+`/cultmesh/publications`, and `POST /speech-provider/mimo/voicedesign`, but
+those endpoints are operator/debug projections, not daemon liveness owners.
 
 ## CultMesh Verses
 

scripts/idunn-rudp-health.cjs

@@ -0,0 +1,233 @@
+"use strict";
+
+const dgram = require("dgram");
+const path = require("path");
+
+const {
+  CultNetRudpSession,
+  decodeRudpPacket,
+  encodeCultNetMessageForWire,
+  encodeRudpPacket,
+} = require("cultnet-ts");
+
+const { encode } = require(path.resolve(
+  __dirname,
+  "..",
+  "..",
+  "CultLib",
+  "node_modules",
+  "@msgpack",
+  "msgpack",
+));
+
+const CULTNET_RUDP_PROTOCOL_ID = "cultnet.transport.rudp.v0";
+const IDUNN_HEALTH_RUDP_CONNECTION_ID = 0x1d0d0001;
+
+function createIdunnRudpHealthPublisher(options) {
+  if (!options?.endpoint) return null;
+  return {
+    daemonId: options.daemonId,
+    endpoint: parseEndpoint(options.endpoint),
+    healthContract: options.healthContract,
+  };
+}
+
+async function publishIdunnRudpHealth(publisher, health) {
+  if (!publisher) return;
+
+  const socket = dgram.createSocket(endpointFamily(publisher.endpoint.host));
+  await bindSocket(socket, publisher.endpoint);
+  const receiver = createPacketReceiver(socket);
+  const session = new CultNetRudpSession({
+    connectionId: IDUNN_HEALTH_RUDP_CONNECTION_ID,
+    initialSequence: 1,
+    resendDelayMs: 100,
+  });
+
+  try {
+    const connect = session.createConnect(Date.now(), new Uint8Array());
+    await sendPacket(socket, publisher.endpoint, connect);
+    await receiveUntil(receiver, session, publisher.endpoint, (packet) => packet.packetType === "accept", 5000, "accept");
+
+    const observedAt = health.observedAt || new Date().toISOString();
+    const record = {
+      daemonId: publisher.daemonId,
+      state: health.state,
+      detail: health.detail,
+      observedAt,
+      healthContract: publisher.healthContract,
+      publicationSource: "daemon-published",
+      transport: CULTNET_RUDP_PROTOCOL_ID,
+    };
+    const payload = encode([
+      record.daemonId,
+      record.state,
+      record.detail,
+      record.observedAt,
+      record.healthContract,
+      record.publicationSource,
+      record.transport,
+    ]);
+    const message = {
+      schemaVersion: "cultnet.document_put_raw.v0",
+      messageId: `weksa-health:${publisher.daemonId}:${observedAt.replace(/[:.]/g, "-")}`,
+      document: {
+        schemaId: "idunn.daemon_health",
+        recordKey: publisher.daemonId,
+        storedAt: observedAt,
+        payloadEncoding: "messagepack",
+        payload,
+        sourceRuntimeId: "weksa-daemon",
+        sourceRole: "daemon-health-publisher",
+        tags: [CULTNET_RUDP_PROTOCOL_ID],
+      },
+    };
+    const wirePayload = encode(encodeCultNetMessageForWire(message, "cultnet.schema.v0"));
+    const dataPackets = session.sendMany("schema", wirePayload, {
+      reliable: true,
+      ordered: true,
+      nowMs: Date.now(),
+    });
+    const ack = receiveUntil(receiver, session, publisher.endpoint, (packet) => packet.packetType === "ack", 500, "ack")
+      .catch(() => undefined);
+    for (const packet of dataPackets) {
+      await sendPacket(socket, publisher.endpoint, packet);
+    }
+    await ack;
+  } finally {
+    receiver.close();
+    socket.close();
+  }
+}
+
+async function bindSocket(socket, endpoint) {
+  await new Promise((resolve, reject) => {
+    socket.once("error", reject);
+    socket.bind(0, endpoint.host.includes(":") ? "::" : "0.0.0.0", () => {
+      socket.off("error", reject);
+      resolve();
+    });
+  });
+}
+
+function parseEndpoint(value) {
+  const text = String(value || "").trim();
+  const ipv6 = text.match(/^\[([^\]]+)\]:(\d+)$/);
+  if (ipv6) return { host: ipv6[1], port: parsePort(ipv6[2]) };
+  const index = text.lastIndexOf(":");
+  if (index <= 0) {
+    throw new Error(`Idunn RUDP endpoint must be host:port, got "${value}".`);
+  }
+  return { host: text.slice(0, index), port: parsePort(text.slice(index + 1)) };
+}
+
+function parsePort(value) {
+  const port = Number(value);
+  if (!Number.isInteger(port) || port <= 0 || port > 65535) {
+    throw new Error(`Idunn RUDP endpoint port is invalid: ${value}`);
+  }
+  return port;
+}
+
+function endpointFamily(host) {
+  return host.includes(":") ? "udp6" : "udp4";
+}
+
+async function receiveUntil(receiver, session, endpoint, predicate, timeoutMs, label) {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    try {
+      const packet = await receiver.next(Math.min(100, deadline - Date.now()), label);
+      const result = session.receive(packet, Date.now());
+      if (result.reply) {
+        throw new Error("Weksa RUDP health publisher received an unexpected reply-required packet.");
+      }
+      if (predicate(packet)) return;
+    } catch (error) {
+      if (error.code !== "ETIMEDOUT") throw error;
+    }
+    for (const packet of session.dueResends(Date.now())) {
+      await sendPacket(receiver.socket, endpoint, packet);
+    }
+  }
+  throw new Error(`timed out waiting for Idunn RUDP ${label} response after ${timeoutMs}ms`);
+}
+
+function createPacketReceiver(socket) {
+  const packets = [];
+  const waiters = [];
+  const errors = [];
+
+  const resolveNext = () => {
+    while (waiters.length > 0 && (packets.length > 0 || errors.length > 0)) {
+      const waiter = waiters.shift();
+      clearTimeout(waiter.timer);
+      if (errors.length > 0) waiter.reject(errors.shift());
+      else waiter.resolve(packets.shift());
+    }
+  };
+  const onMessage = (wire) => {
+    try {
+      packets.push(decodeRudpPacket(wire));
+    } catch (error) {
+      errors.push(error);
+    }
+    resolveNext();
+  };
+  const onError = (error) => {
+    errors.push(error);
+    resolveNext();
+  };
+
+  socket.on("message", onMessage);
+  socket.on("error", onError);
+
+  return {
+    socket,
+    next(timeoutMs, label = "packet") {
+      if (packets.length > 0) return Promise.resolve(packets.shift());
+      if (errors.length > 0) return Promise.reject(errors.shift());
+      return new Promise((resolve, reject) => {
+        const waiter = {
+          resolve,
+          reject,
+          timer: setTimeout(() => {
+            const index = waiters.indexOf(waiter);
+            if (index >= 0) waiters.splice(index, 1);
+            const error = new Error(`timed out waiting for Idunn RUDP ${label}`);
+            error.code = "ETIMEDOUT";
+            reject(error);
+          }, Math.max(1, timeoutMs)),
+        };
+        waiters.push(waiter);
+      });
+    },
+    close() {
+      socket.off("message", onMessage);
+      socket.off("error", onError);
+      while (waiters.length > 0) {
+        const waiter = waiters.shift();
+        clearTimeout(waiter.timer);
+        const error = new Error("Weksa RUDP health publisher closed.");
+        error.code = "ECLOSED";
+        waiter.reject(error);
+      }
+    },
+  };
+}
+
+async function sendPacket(socket, endpoint, packet) {
+  const wire = encodeRudpPacket(packet);
+  await new Promise((resolve, reject) => {
+    socket.send(wire, endpoint.port, endpoint.host, (error) => {
+      if (error) reject(error);
+      else resolve();
+    });
+  });
+}
+
+module.exports = {
+  CULTNET_RUDP_PROTOCOL_ID,
+  createIdunnRudpHealthPublisher,
+  publishIdunnRudpHealth,
+};

scripts/restart-weksa-daemon.ps1

@@ -2,7 +2,10 @@ param(
   [int] $Port = 8813,
   [string] $HostName = "127.0.0.1",
   [string] $StateRoot = "E:\Projects\weksa\.weksa",
-  [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt"
+  [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt",
+  [string] $IdunnRudpHealth = "127.0.0.1:17870",
+  [string] $IdunnDaemon = "weksa",
+  [string] $IdunnHealthContract = "weksa.cultnet-rudp-provider-health"
 )
 
 $ErrorActionPreference = "Stop"
@@ -18,5 +21,5 @@ if (Test-Path -LiteralPath $pidPath) {
   Remove-Item -LiteralPath $pidPath -Force -ErrorAction SilentlyContinue
 }
 
-& powershell.exe -NoProfile -ExecutionPolicy Bypass -File (Join-Path $repoRoot "scripts\start-weksa-daemon.ps1") -Port $Port -HostName $HostName -StateRoot $StateRoot -MiMoApiKeyPath $MiMoApiKeyPath
+& powershell.exe -NoProfile -ExecutionPolicy Bypass -File (Join-Path $repoRoot "scripts\start-weksa-daemon.ps1") -Port $Port -HostName $HostName -StateRoot $StateRoot -MiMoApiKeyPath $MiMoApiKeyPath -IdunnRudpHealth $IdunnRudpHealth -IdunnDaemon $IdunnDaemon -IdunnHealthContract $IdunnHealthContract
 exit $LASTEXITCODE

scripts/start-weksa-daemon.ps1

@@ -2,7 +2,10 @@ param(
   [int] $Port = 8813,
   [string] $HostName = "127.0.0.1",
   [string] $StateRoot = "E:\Projects\weksa\.weksa",
-  [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt"
+  [string] $MiMoApiKeyPath = "E:\Projects\gamecult-ops\mimo-api-Weksa.txt",
+  [string] $IdunnRudpHealth = "127.0.0.1:17870",
+  [string] $IdunnDaemon = "weksa",
+  [string] $IdunnHealthContract = "weksa.cultnet-rudp-provider-health"
 )
 
 $ErrorActionPreference = "Stop"
@@ -13,6 +16,8 @@ $outLog = Join-Path $StateRoot "weksa-daemon.out.log"
 $errLog = Join-Path $StateRoot "weksa-daemon.err.log"
 
 New-Item -ItemType Directory -Force -Path $StateRoot | Out-Null
+$cultLibPackages = Resolve-Path -LiteralPath (Join-Path $repoRoot "..\CultLib\packages")
+$env:NODE_PATH = (@($cultLibPackages.Path, $env:NODE_PATH) | Where-Object { $_ }) -join [IO.Path]::PathSeparator
 
 if (Test-Path -LiteralPath $pidPath) {
   $pidText = (Get-Content -LiteralPath $pidPath -Raw).Trim()
@@ -38,7 +43,10 @@ $arguments = @(
   "--host", $HostName,
   "--port", [string] $Port,
   "--state-root", $StateRoot,
-  "--mimo-api-key-path", $MiMoApiKeyPath
+  "--mimo-api-key-path", $MiMoApiKeyPath,
+  "--idunn-rudp-health", $IdunnRudpHealth,
+  "--idunn-daemon", $IdunnDaemon,
+  "--idunn-health-contract", $IdunnHealthContract
 )
 
 $process = Start-Process -FilePath "node" `

scripts/weksa-daemon.mjs

@@ -7,6 +7,13 @@ import { basename, delimiter, dirname, extname, resolve } from "node:path";
 import { fileURLToPath } from "node:url";
 
 const repoRoot = resolve(dirname(fileURLToPath(import.meta.url)), "..");
+configureCultLibModulePath();
+const require = Module.createRequire(import.meta.url);
+const {
+  CULTNET_RUDP_PROTOCOL_ID,
+  createIdunnRudpHealthPublisher,
+  publishIdunnRudpHealth,
+} = require("./idunn-rudp-health.cjs");
 const options = parseArgs(process.argv.slice(2));
 const port = Number(options.port ?? process.env.WEKSA_DAEMON_PORT ?? 8813);
 const host = options.host ?? process.env.WEKSA_DAEMON_HOST ?? "127.0.0.1";
@@ -15,11 +22,18 @@ const cultMeshStorePath = resolve(stateRoot, "provider-advertisement-store.cc");
 const pidPath = resolve(stateRoot, "weksa-daemon.pid");
 const logPath = resolve(stateRoot, "weksa-daemon.log");
 const mimoApiKeyPath = options.mimoApiKeyPath ?? process.env.WEKSA_MIMO_API_KEY_FILE ?? "E:/Projects/gamecult-ops/mimo-api-Weksa.txt";
+const idunnHealthPublisher = createIdunnRudpHealthPublisher({
+  endpoint: options.idunnRudpHealth ?? process.env.WEKSA_IDUNN_RUDP_HEALTH,
+  daemonId: options.idunnDaemon ?? process.env.WEKSA_IDUNN_DAEMON ?? "weksa",
+  healthContract: options.idunnHealthContract ?? process.env.WEKSA_IDUNN_HEALTH_CONTRACT ?? "weksa.cultnet-rudp-provider-health",
+});
 const startedAt = new Date().toISOString();
 let tick = 0;
 let lastPublishedAt = "";
 let lastError = undefined;
 let cachedSnapshot = undefined;
+let witnessRefreshRunning = false;
+let lastRudpHealthSuccessMs = 0;
 
 if (options.health) {
   await runHealthCheck();
@@ -29,7 +43,7 @@ if (options.health) {
 await mkdir(stateRoot, { recursive: true });
 await writeFile(pidPath, `${process.pid}\n`, "ascii");
 await appendLog(`started pid=${process.pid} host=${host} port=${port}`);
-await publishWitnesses();
+await refreshWitnesses("startup");
 
 const server = http.createServer(async (request, response) => {
   try {
@@ -61,7 +75,7 @@ const server = http.createServer(async (request, response) => {
     if (url.pathname === "/speech-provider/mimo/voicedesign" && request.method === "POST") {
       const body = await readJsonBody(request);
       const result = await handleMimoVoiceDesign(body);
-      await publishWitnesses();
+      await refreshWitnesses("mimo-voicedesign-command");
       return sendJson(response, 200, result);
     }
     sendJson(response, 404, { ok: false, error: "unknown endpoint" });
@@ -77,9 +91,10 @@ server.listen(port, host, () => {
 });
 
 const interval = setInterval(() => {
-  publishWitnesses().catch(async (error) => {
+  refreshWitnesses("interval").catch(async (error) => {
     lastError = String(error?.stack ?? error);
     await appendLog(`publish error: ${lastError}`);
+    await publishWeksaRudpHealth("failed", `Weksa witness refresh failed: ${String(error?.message ?? error)}`);
   });
 }, 30_000);
 
@@ -131,15 +146,48 @@ async function publishWitnesses() {
   cachedSnapshot = snapshot;
   lastPublishedAt = snapshot.operatorState.updated_at;
   tick += 1;
+  return snapshot;
 }
 
 async function currentSnapshot() {
   if (!cachedSnapshot) {
-    await publishWitnesses();
+    await refreshWitnesses("snapshot-demand");
   }
   return cachedSnapshot;
 }
 
+async function refreshWitnesses(reason) {
+  if (witnessRefreshRunning) {
+    await appendLog(`skipped overlapping witness refresh reason=${reason}`);
+    return cachedSnapshot;
+  }
+  witnessRefreshRunning = true;
+  try {
+    const snapshot = await publishWitnesses();
+    await publishWeksaRudpHealth(
+      "healthy",
+      `Weksa refreshed witnesses reason=${reason} tick=${tick} provider=${snapshot.providerAdvertisement.providerId}`,
+      snapshot.operatorState.updated_at,
+    );
+    return snapshot;
+  } finally {
+    witnessRefreshRunning = false;
+  }
+}
+
+async function publishWeksaRudpHealth(state, detail, observedAt = new Date().toISOString()) {
+  if (!idunnHealthPublisher) return;
+  try {
+    await publishIdunnRudpHealth(idunnHealthPublisher, { state, detail, observedAt });
+    lastRudpHealthSuccessMs = Date.now();
+  } catch (error) {
+    const hasRecentSuccess = lastRudpHealthSuccessMs > 0 && Date.now() - lastRudpHealthSuccessMs < 300_000;
+    if (!hasRecentSuccess) {
+      await appendLog(`Idunn RUDP health publish failed: ${String(error?.message ?? error)}`);
+    }
+  }
+}
+
 async function buildSnapshot() {
   const updatedAt = new Date().toISOString();
   const providerAdvertisement = loadProviderAdvertisement(updatedAt);
@@ -182,7 +230,8 @@ async function buildSnapshot() {
     updated_at: updatedAt,
     transport: {
       cultmesh_store: ".weksa/provider-advertisement-store.cc",
-      native_cultnet_peer: "pending",
+      native_cultnet_peer: idunnHealthPublisher ? `${CULTNET_RUDP_PROTOCOL_ID}:idunn-health` : "pending",
+      idunn_rudp_health: idunnHealthPublisher ? `${idunnHealthPublisher.endpoint.host}:${idunnHealthPublisher.endpoint.port}` : "unconfigured",
       compatibility_http: `http://${host}:${port}`,
     },
     publications: [
@@ -251,12 +300,7 @@ async function createCultMeshNodeWithOwnedStoreReset(CultMesh, storePath, docume
 
 function loadCultRuntime() {
   try {
-    process.env.NODE_PATH = [
+    configureCultLibModulePath();
-      resolve(repoRoot, "..", "CultLib", "packages"),
-      process.env.NODE_PATH || "",
-    ].filter(Boolean).join(delimiter);
-    Module._initPaths();
-    const require = Module.createRequire(import.meta.url);
     const { CultMesh } = require("cultmesh-ts");
     const { defineDocumentType } = require("cultcache-ts");
     return { CultMesh, defineDocumentType, error: null };
@@ -265,6 +309,14 @@ function loadCultRuntime() {
   }
 }
 
+function configureCultLibModulePath() {
+  process.env.NODE_PATH = [
+    resolve(repoRoot, "..", "CultLib", "packages"),
+    process.env.NODE_PATH || "",
+  ].filter(Boolean).join(delimiter);
+  Module._initPaths();
+}
+
 function defineCultMeshDocuments(defineDocumentType) {
   const surfaceDefinition = defineDocumentType({
     type: "gamecult.eve.surface_state",
@@ -928,6 +980,18 @@ function parseArgs(args) {
         parsed.mimoApiKeyPath = args[index + 1];
         index += 1;
         break;
+      case "--idunn-rudp-health":
+        parsed.idunnRudpHealth = args[index + 1];
+        index += 1;
+        break;
+      case "--idunn-daemon":
+        parsed.idunnDaemon = args[index + 1];
+        index += 1;
+        break;
+      case "--idunn-health-contract":
+        parsed.idunnHealthContract = args[index + 1];
+        index += 1;
+        break;
       case "--health":
         parsed.health = true;
         break;