diff --git a/scripts/serve-voidbot-swarm-cultmesh.cjs b/scripts/serve-voidbot-swarm-cultmesh.cjs index 729c7d4..e4f55fa 100644 --- a/scripts/serve-voidbot-swarm-cultmesh.cjs +++ b/scripts/serve-voidbot-swarm-cultmesh.cjs @@ -4,6 +4,12 @@ const fs = require("fs"); const path = require("path"); const { createRequire } = require("module"); +const { + operatorViewPublication, + projectSwarmOperatorView, + readSwarmOperatorInputs, + watchSwarmOperatorInputs, +} = require("./voidbot-swarm-operator-view.cjs"); const repoRoot = path.resolve(__dirname, ".."); const defaultStorePath = path.join(repoRoot, ".voidbot", "status", "cultmesh", "voidbot-swarm-state.cc"); @@ -17,6 +23,12 @@ const odinConnectionId = 0x0d1d0002; const args = parseArgs(process.argv.slice(2)); const storePath = path.resolve(args.store || process.env.VOIDBOT_SWARM_CULTMESH_STORE || defaultStorePath); const controlStorePath = path.resolve(args.controlStore || process.env.VOIDBOT_SWARM_CONTROL_STORE || defaultControlStorePath); +const operatorInputPaths = { + heartbeat: path.resolve(process.env.REPO_FACE_HEARTBEAT_STATE_PATH || path.join(repoRoot, ".voidbot", "status", "repo-face-heartbeats.json")), + orchestrator: path.resolve(process.env.VOIDBOT_ORCHESTRATOR_STATE_PATH || path.join(repoRoot, ".voidbot", "status", "gamecult-orchestrator.json")), + pause: path.resolve(process.env.VOIDBOT_SWARM_PAUSE_STATE_PATH || path.join(repoRoot, "state", "agent-swarm-paused.json")), + control: controlStorePath, +}; const bind = parseBind(args.bind || process.env.VOIDBOT_SWARM_CULTMESH_BIND || defaultBind); const odinCultMeshUri = args.odinCultMeshUri || args["odin-cultmesh-uri"] || process.env.VOIDBOT_ODIN_CULTMESH_URI || process.env.ODIN_CULTMESH_URI || defaultOdinCultMeshUri; const odinRudpEndpoint = args.odinRudpEndpoint || args["odin-rudp-endpoint"] || process.env.VOIDBOT_ODIN_RUDP || process.env.CULTMESH_URI_ODIN_RUDP || defaultOdinRudpEndpoint; @@ -42,6 +54,8 @@ const documentRegistry = new CultNetDocumentRegistry( let server = null; let announceTimer = null; let liveProviderSession = null; +let stopOperatorWatch = null; +let operatorPublishChain = Promise.resolve(); main().catch((error) => { console.error(error instanceof Error ? error.stack || error.message : String(error)); @@ -71,6 +85,8 @@ async function main() { await server.start(); liveProviderSession = await startLiveProviderSession(); + await publishCurrentOperatorView(); + stopOperatorWatch = watchSwarmOperatorInputs(operatorInputPaths, queueOperatorViewPublication); console.log(`VoidBot swarm CultMesh/RUDP serving ${storePath} at rudp://${bind.host}:${bind.port}`); announceToOdin().catch((error) => { console.error(`VoidBot swarm Odin announcement failed: ${error instanceof Error ? error.message : String(error)}`); @@ -86,6 +102,7 @@ async function main() { try { if (announceTimer) clearInterval(announceTimer); server?.close?.(); + stopOperatorWatch?.(); void liveProviderSession?.stop?.(); } finally { process.exit(0); @@ -151,6 +168,33 @@ async function projectAcceptedHeat(control) { await node.put(documents.swarmControl, "voidbot-swarm", control); await node.flush?.(true); await liveProviderSession?.upsertPublication?.(controlPublication(control)); + await queueOperatorViewPublication(); +} + +function queueOperatorViewPublication() { + operatorPublishChain = operatorPublishChain + .then(publishCurrentOperatorView) + .catch((error) => console.error(`VoidBot swarm operator view publication failed: ${error instanceof Error ? error.message : String(error)}`)); + return operatorPublishChain; +} + +async function publishCurrentOperatorView() { + const { control: _controlPath, ...jsonInputPaths } = operatorInputPaths; + const inputs = await readSwarmOperatorInputs(jsonInputPaths); + const controlNode = await CultMesh.createNode(controlStorePath, { documents: [documents.swarmControl] }); + const control = controlNode.get(documents.swarmControl, "voidbot-swarm"); + const view = projectSwarmOperatorView({ + ...inputs, + control, + route: { + providerStatus: liveProviderSession ? "connected" : "local-only", + endpoint: `rudp://${bind.host}:${bind.port}`, + }, + }); + const node = await CultMesh.createNode(storePath, { documents: documentDefinitions }); + await node.put(documents.swarmOperatorView, "voidbot-swarm", view); + await node.flush?.(true); + await liveProviderSession?.upsertPublication?.(operatorViewPublication(view)); } async function startLiveProviderSession() { @@ -332,6 +376,15 @@ function defineDocuments(defineDocumentType) { name: () => "voidbot-swarm", schema: objectSchema("VoidBot swarm control state"), }), + swarmOperatorView: defineDocumentType({ + type: "voidbot.swarm_operator_view", + schemaName: "voidbot.swarm_operator_view", + schemaId: "voidbot.swarm_operator_view.v1", + schemaVersion: "voidbot.swarm_operator_view.v1", + global: false, + name: () => "voidbot-swarm", + schema: objectSchema("VoidBot swarm operator view"), + }), }; } diff --git a/scripts/voidbot-swarm-operator-view.cjs b/scripts/voidbot-swarm-operator-view.cjs new file mode 100644 index 0000000..218a2fd --- /dev/null +++ b/scripts/voidbot-swarm-operator-view.cjs @@ -0,0 +1,221 @@ +"use strict"; + +const fs = require("fs"); +const pathModule = require("path"); + +const SCHEMA_ID = "voidbot.swarm_operator_view.v1"; +const RECORD_KEY = "voidbot-swarm"; + +function projectSwarmOperatorView({ heartbeat, orchestrator, pause, control, observedAt = new Date().toISOString(), sources = {}, route = {} }) { + const initiativeClock = number(heartbeat?.initiativeClock); + const participants = array(heartbeat?.participants).map((participant) => { + const identityId = text(participant?.identityId) || "unknown"; + const pendingMentionCount = array(heartbeat?.pendingMentions) + .filter((mention) => text(mention?.identityId) === identityId).length; + return { + identityId, + displayName: text(participant?.displayName) || identityId, + repoName: text(participant?.repoName) || "unknown", + status: text(participant?.status) || "unknown", + activeJobId: text(participant?.activeJobId), + nextTurnInMinutes: number(participant?.nextTurnInMinutes) ?? subtract(number(participant?.nextTurnAt), initiativeClock), + heat: number(participant?.heat), + avatarUri: text(participant?.avatarUri) || text(participant?.avatarUrl), + memoryCount: number(participant?.memoryCount) ?? array(participant?.memories).length, + pressureCount: array(participant?.responsePressureEvidence).length, + description: text(participant?.description), + statePath: text(participant?.statePath) || text(participant?.faceStatePath), + statePreview: participant?.statePreview ?? null, + pendingMentionCount, + }; + }); + const organs = Object.entries(object(orchestrator?.organs)).map(([id, organ]) => ({ + id, + label: text(organ?.label) || id, + status: text(organ?.lastStatus) || "unknown", + lastFinishedAt: text(organ?.lastFinishedAt), + durationSeconds: number(organ?.lastDurationSeconds), + })); + const activeTurnCount = participants.filter((participant) => participant.activeJobId).length; + const readyNowCount = participants.filter((participant) => + participant.status === "active" && !participant.activeJobId && participant.nextTurnInMinutes !== null && participant.nextTurnInMinutes <= 0, + ).length; + const paused = pause?.paused === true; + const heartbeatReadable = sources.heartbeat?.readable !== false; + const orchestratorReadable = sources.orchestrator?.readable !== false; + + const nextTurn = selectNextTurn(participants); + const selected = participants.find((participant) => participant.identityId === nextTurn?.identityId) ?? participants[0] ?? null; + const surfaceOrgan = object(orchestrator?.organs)["voidbot-swarm-surface"]; + const cultMeshServer = object(surfaceOrgan?.cultMeshServer); + const alerts = buildAlerts({ sources, organs }); + return { + version: SCHEMA_ID, + observedAt, + summary: { + state: deriveSwarmState({ heartbeatReadable, paused, activeTurnCount, organs }), + paused, + participantCount: participants.length, + activeTurnCount, + readyNowCount, + pendingMentionCount: array(heartbeat?.pendingMentions).length, + nextIdentityId: nextTurn?.identityId ?? null, + nextDisplayName: nextTurn?.displayName ?? null, + lastTickAt: text(heartbeat?.lastTickAt), + }, + route: { + storeWriteStatus: text(route.storeWriteStatus) || "ok", + providerStatus: text(route.providerStatus) || "unknown", + odinStatus: text(route.odinStatus) || text(cultMeshServer.status) || deriveOrchestratorState(organs, orchestratorReadable), + endpoint: text(route.endpoint) || text(cultMeshServer.bind), + lastAnnouncementAt: text(route.lastAnnouncementAt) || text(surfaceOrgan?.lastFinishedAt), + alerts, + }, + queue: [...participants] + .sort((left, right) => nullableOrder(left.nextTurnInMinutes, right.nextTurnInMinutes)) + .map((participant, slot) => ({ + slot: slot + 1, + identityId: participant.identityId, + displayName: participant.displayName, + repoName: participant.repoName, + state: participant.activeJobId ? "active" : participant.status, + nextTurnInMinutes: participant.nextTurnInMinutes, + heat: participant.heat ?? number(control?.globalHeat) ?? number(heartbeat?.globalHeat), + label: participant.displayName, + status: participant.activeJobId ? "active" : participant.status, + detail: `${participant.repoName} ยท ${formatTurnDistance(participant.nextTurnInMinutes)}`, + badges: compact([ + participant.heat === null ? null : `heat ${participant.heat}`, + participant.pendingMentionCount > 0 ? `${participant.pendingMentionCount} mention${participant.pendingMentionCount === 1 ? "" : "s"}` : null, + ]), + })), + selectedFace: selected ? { + identityId: selected.identityId, + displayName: selected.displayName, + repoName: selected.repoName, + avatarUri: selected.avatarUri, + status: selected.status, + nextTurnInMinutes: selected.nextTurnInMinutes, + heat: selected.heat ?? number(control?.globalHeat) ?? number(heartbeat?.globalHeat), + memoryCount: selected.memoryCount, + pressureCount: selected.pressureCount, + description: selected.description, + statePath: selected.statePath, + statePreview: selected.statePreview, + } : null, + }; +} + +async function readSwarmOperatorInputs(paths) { + const entries = await Promise.all(Object.entries(paths).map(async ([key, path]) => { + try { + const [raw, stats] = await Promise.all([ + fs.promises.readFile(path, "utf8"), + fs.promises.stat(path), + ]); + return [key, { value: JSON.parse(raw), status: { readable: true, updatedAt: stats.mtime.toISOString() } }]; + } catch (error) { + return [key, { value: null, status: { readable: false, error: error instanceof Error ? error.message : String(error) } }]; + } + })); + const read = Object.fromEntries(entries); + return { + heartbeat: read.heartbeat.value, + orchestrator: read.orchestrator.value, + pause: read.pause.value, + control: read.control?.value ?? null, + sources: Object.fromEntries(Object.entries(read).map(([key, result]) => [key, result.status])), + }; +} + +function watchSwarmOperatorInputs(paths, onChange, { debounceMs = 75 } = {}) { + const watched = [...new Set(Object.values(paths).map(pathModule.resolve))]; + const byDirectory = new Map(); + for (const path of watched) { + const directory = pathModule.dirname(path); + const names = byDirectory.get(directory) ?? new Set(); + names.add(pathModule.basename(path).toLowerCase()); + byDirectory.set(directory, names); + } + let timer = null; + const schedule = () => { + if (timer) clearTimeout(timer); + timer = setTimeout(() => { + timer = null; + Promise.resolve(onChange()).catch(() => {}); + }, debounceMs); + timer.unref?.(); + }; + const watchers = [...byDirectory].map(([directory, names]) => fs.watch(directory, { persistent: false }, (_event, filename) => { + if (filename === null || names.has(String(filename).toLowerCase())) schedule(); + })); + return () => { + if (timer) clearTimeout(timer); + for (const watcher of watchers) watcher.close(); + }; +} + +function operatorViewPublication(value) { + return { + publicationId: "voidbot.swarm.operator-view", + documentType: "voidbot.swarm_operator_view", + schemaId: SCHEMA_ID, + recordKey: RECORD_KEY, + value, + }; +} + +function deriveSwarmState({ heartbeatReadable, paused, activeTurnCount, organs }) { + if (!heartbeatReadable) return "missing"; + if (paused) return "paused"; + if (organs.some((organ) => ["failed", "error", "stalled"].includes(organ.status))) return "warning"; + if (activeTurnCount > 0) return "running"; + return "ready"; +} + +function deriveOrchestratorState(organs, readable) { + if (!readable || organs.length === 0) return "missing"; + if (organs.some((organ) => ["failed", "error", "stalled"].includes(organ.status))) return "warning"; + if (organs.some((organ) => organ.status === "running")) return "running"; + return "ready"; +} + +function selectNextTurn(participants) { + const next = [...participants] + .filter((participant) => participant.nextTurnInMinutes !== null) + .sort((left, right) => left.nextTurnInMinutes - right.nextTurnInMinutes)[0]; + return next ? { + identityId: next.identityId, + displayName: next.displayName, + inMinutes: next.nextTurnInMinutes, + } : null; +} + +function buildAlerts({ sources, organs }) { + const alerts = []; + for (const [source, status] of Object.entries(sources)) { + if (status?.readable === false) alerts.push({ code: `source.${source}.unreadable`, severity: "error", detail: text(status.error) || `${source} is unreadable` }); + } + for (const organ of organs.filter((entry) => ["failed", "error", "stalled"].includes(entry.status))) { + alerts.push({ code: `organ.${organ.id}.${organ.status}`, severity: "warning", detail: `${organ.label} is ${organ.status}` }); + } + return alerts; +} + +function array(value) { return Array.isArray(value) ? value : []; } +function object(value) { return value && typeof value === "object" && !Array.isArray(value) ? value : {}; } +function text(value) { return typeof value === "string" && value.trim() ? value : null; } +function number(value) { const parsed = Number(value); return Number.isFinite(parsed) ? parsed : null; } +function subtract(left, right) { return left === null || right === null ? null : Math.round((left - right) * 1000) / 1000; } +function nullableOrder(left, right) { return left === null ? 1 : right === null ? -1 : left - right; } +function compact(values) { return values.filter((value) => value !== null); } +function formatTurnDistance(value) { return value === null ? "unscheduled" : value <= 0 ? "ready now" : `turn in ${value}m`; } + +module.exports = { + RECORD_KEY, + SCHEMA_ID, + operatorViewPublication, + projectSwarmOperatorView, + readSwarmOperatorInputs, + watchSwarmOperatorInputs, +}; diff --git a/scripts/voidbot-swarm-operator-view.test.cjs b/scripts/voidbot-swarm-operator-view.test.cjs new file mode 100644 index 0000000..15b7717 --- /dev/null +++ b/scripts/voidbot-swarm-operator-view.test.cjs @@ -0,0 +1,72 @@ +"use strict"; + +const assert = require("node:assert/strict"); +const fs = require("node:fs/promises"); +const os = require("node:os"); +const path = require("node:path"); +const test = require("node:test"); +const { operatorViewPublication, projectSwarmOperatorView, readSwarmOperatorInputs } = require("./voidbot-swarm-operator-view.cjs"); + +test("projects bounded operator state and lets control own heat", () => { + const view = projectSwarmOperatorView({ + heartbeat: { + globalHeat: 0.5, + initiativeClock: 12, + lastTickAt: "2026-07-15T10:00:00.000Z", + pendingMentions: [{ identityId: "eve" }], + participants: [ + { identityId: "eve", displayName: "Eve", status: "active", nextTurnInMinutes: 3 }, + { identityId: "odin", displayName: "Odin", status: "active", activeJobId: "job-1", nextTurnInMinutes: 8 }, + ], + }, + orchestrator: { organs: { heartbeat: { label: "Heartbeat", lastStatus: "ok", lastDurationSeconds: 2 } } }, + pause: { paused: false }, + control: { globalHeat: 0.85 }, + observedAt: "2026-07-15T10:01:00.000Z", + sources: { heartbeat: { readable: true }, orchestrator: { readable: true } }, + }); + + assert.equal(view.version, "voidbot.swarm_operator_view.v1"); + assert.equal(view.summary.activeTurnCount, 1); + assert.equal(view.summary.pendingMentionCount, 1); + assert.equal(view.summary.nextIdentityId, "eve"); + assert.equal(view.queue[0].heat, 0.85); + assert.equal(view.selectedFace.identityId, "eve"); + assert.equal(view.recentEvents, undefined); +}); + +test("publication exposes the typed record without dashboard-shaped authority", () => { + const value = projectSwarmOperatorView({ + heartbeat: null, + orchestrator: null, + pause: { paused: true, reason: "maintenance" }, + control: null, + sources: { heartbeat: { readable: false }, orchestrator: { readable: false } }, + }); + const publication = operatorViewPublication(value); + assert.equal(publication.publicationId, "voidbot.swarm.operator-view"); + assert.equal(publication.documentType, "voidbot.swarm_operator_view"); + assert.equal(publication.schemaId, "voidbot.swarm_operator_view.v1"); + assert.equal(publication.recordKey, "voidbot-swarm"); + assert.equal(value.summary.state, "missing"); + assert.equal(value.selectedFace, null); +}); + +test("reads source files independently and reports a missing owner", async () => { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "voidbot-operator-view-")); + try { + const heartbeat = path.join(root, "heartbeat.json"); + await fs.writeFile(heartbeat, JSON.stringify({ participants: [] })); + const inputs = await readSwarmOperatorInputs({ + heartbeat, + orchestrator: path.join(root, "missing.json"), + pause: path.join(root, "pause.json"), + }); + assert.deepEqual(inputs.heartbeat, { participants: [] }); + assert.equal(inputs.sources.heartbeat.readable, true); + assert.equal(inputs.sources.orchestrator.readable, false); + assert.equal(inputs.control, null); + } finally { + await fs.rm(root, { recursive: true, force: true }); + } +});