Drain operator DM CultMesh requests

This commit is contained in:
Meta Vens 2026-06-04 14:42:35 +01:00
commit 6fcb35ec72
3 changed files with 271 additions and 121 deletions

View file

@ -1,6 +1,7 @@
import "dotenv/config";
import { createHash } from "node:crypto";
import { createRequire } from "node:module";
import { mkdir, writeFile } from "node:fs/promises";
import { spawnSync } from "node:child_process";
import { isAbsolute, join, relative, resolve } from "node:path";
@ -132,6 +133,8 @@ let providerRegistry: ProviderRegistry;
let systemMessages: SystemMessageCatalog;
let jobQueue: JobQueue;
let auditLog: AuditLog;
let isDrainingOperatorDm = false;
let operatorDmDrainUnavailableLogged = false;
void main().catch((error) => {
console.error(error);
@ -195,6 +198,14 @@ async function pollPendingJobs(): Promise<void> {
error instanceof Error ? `${error.name}: ${error.message}` : "Unexpected worker poll failure.";
console.error(`Worker poll failed: ${message}`);
}
try {
await drainOperatorDmRequests();
} catch (error) {
const message =
error instanceof Error ? `${error.name}: ${error.message}` : "Unexpected operator DM drain failure.";
console.error(`Operator DM drain failed: ${message}`);
}
}
async function buildProviderRegistry(systemMessagesCatalog: SystemMessageCatalog): Promise<ProviderRegistry> {
@ -2491,6 +2502,235 @@ async function postOwnerNotification(content: string): Promise<void> {
await postDiscordBotMessage(dmChannelId, fitDiscordMessage(content));
}
type OperatorDmRequestStatus = "pending" | "sent" | "failed";
interface OperatorDmRequest {
requestId?: string;
command?: string;
message?: string;
severity?: string;
service?: string;
sourceId?: string;
status?: OperatorDmRequestStatus | string;
reason?: string;
createdAt?: string;
updatedAt?: string;
sentAt?: string;
failedAt?: string;
attempts?: number;
channelId?: string;
messageId?: string;
error?: string;
}
async function drainOperatorDmRequests(): Promise<void> {
if (isDrainingOperatorDm) {
return;
}
isDrainingOperatorDm = true;
try {
const runtime = loadCultMeshRuntime();
if (!runtime) {
if (!operatorDmDrainUnavailableLogged) {
operatorDmDrainUnavailableLogged = true;
console.warn("Operator DM CultMesh drain skipped: CultMesh/CultCache runtime unavailable.");
}
return;
}
const { CultMesh, defineDocumentType } = runtime;
const documents = operatorDmCultMeshDocuments(defineDocumentType);
const node = await CultMesh.createNode(operatorDmStorePath(), {
documents: documents.all,
});
const requests = node.cache.getAll(documents.requestDefinition) as OperatorDmRequest[];
for (const request of requests) {
if (!shouldSendOperatorDmRequest(request)) {
continue;
}
const requestId = operatorDmRequestId(request);
const attempts = (typeof request.attempts === "number" ? request.attempts : 0) + 1;
try {
const dmChannelId = await openOwnerDmChannel();
const posted = await postDiscordBotMessage(dmChannelId, renderOperatorDmRequest(request));
await node.put(documents.requestDefinition, requestId, {
...request,
requestId,
status: "sent",
attempts,
channelId: dmChannelId,
messageId: posted.id,
sentAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
error: undefined,
});
console.log(`Sent operator DM request ${requestId} as Discord message ${posted.id}.`);
} catch (error) {
await node.put(documents.requestDefinition, requestId, {
...request,
requestId,
status: "failed",
attempts,
failedAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
error: error instanceof Error ? error.message : String(error),
});
console.error(`Operator DM request ${requestId} failed: ${error instanceof Error ? error.message : String(error)}`);
}
}
await node.flush?.();
} finally {
isDrainingOperatorDm = false;
}
}
function operatorDmCultMeshDocuments(defineDocumentType: any): {
requestDefinition: any;
all: any[];
} {
const objectSchema = (label: string) => ({
parse(input: unknown): unknown {
if (!input || typeof input !== "object") {
throw new Error(`${label} must be an object.`);
}
return input;
},
});
const requestDefinition = defineDocumentType({
type: "gamecult.operator_dm_request",
schemaName: "gamecult.operator_dm_request",
schemaId: "gamecult.operator_dm_request.v1",
schemaVersion: "gamecult.operator_dm_request.v1",
contentHash: "gamecult.operator_dm_request.v1",
global: false,
name: (value: OperatorDmRequest) => value.requestId ?? "operator-dm-request",
schema: {
parse(input: unknown): OperatorDmRequest {
if (!input || typeof input !== "object") {
throw new Error("Operator DM request must be an object.");
}
return input as OperatorDmRequest;
},
},
});
return {
requestDefinition,
all: [
defineDocumentType({
type: "voidbot.swarm_state_snapshot",
schemaName: "voidbot.swarm_state_snapshot",
schemaId: "voidbot.swarm_state_snapshot.v1",
schemaVersion: "voidbot.swarm_state_snapshot.v1",
contentHash: "voidbot.swarm_state_snapshot.v1",
global: false,
schema: objectSchema("VoidBot swarm snapshot"),
}),
defineDocumentType({
type: "gamecult.eve.provider_advertisement",
schemaName: "gamecult.eve.provider_advertisement",
schemaId: "gamecult.eve.provider_advertisement.v1",
schemaVersion: "gamecult.eve.provider_advertisement.v1",
contentHash: "gamecult.eve.provider_advertisement.v1",
global: false,
name: "providerId",
schema: objectSchema("Eve provider advertisement"),
}),
defineDocumentType({
type: "gamecult.eve.surface_state",
schemaName: "gamecult.eve.surface_state",
schemaId: "gamecult.eve.surface_state.v1",
schemaVersion: "gamecult.eve.surface_state.v1",
contentHash: "gamecult.eve.surface_state.v1",
global: false,
name: "providerId",
schema: objectSchema("Eve surface state"),
}),
defineDocumentType({
type: "gamecult.eve.interface_binding",
schemaName: "gamecult.eve.interface_binding",
schemaId: "gamecult.eve.interface_binding.v1",
schemaVersion: "gamecult.eve.interface_binding.v1",
contentHash: "gamecult.eve.interface_binding.v1",
global: false,
name: "bindingId",
schema: objectSchema("Eve interface binding"),
}),
requestDefinition,
],
};
}
function shouldSendOperatorDmRequest(request: OperatorDmRequest): boolean {
if (!request || typeof request !== "object") {
return false;
}
if (request.command && request.command !== "owner.dm.send") {
return false;
}
if (!request.message || !request.message.trim()) {
return false;
}
return !request.status || request.status === "pending";
}
function renderOperatorDmRequest(request: OperatorDmRequest): string {
const message = request.message?.trim() ?? "";
const header = [
request.severity ? `[${request.severity.toUpperCase()}]` : "[ALERT]",
request.service ?? "GameCult service",
request.sourceId ? `source ${request.sourceId}` : "",
].filter(Boolean).join(" ");
const reason = request.reason ? `\nReason: ${request.reason}` : "";
return `${header}\n${message}${reason}`.trim();
}
function operatorDmRequestId(request: OperatorDmRequest): string {
if (request.requestId?.trim()) {
return request.requestId.trim();
}
return `operator-dm-${createHash("sha1")
.update(JSON.stringify({
command: request.command ?? "owner.dm.send",
message: request.message ?? "",
severity: request.severity ?? "",
service: request.service ?? "",
sourceId: request.sourceId ?? "",
createdAt: request.createdAt ?? "",
}))
.digest("hex")
.slice(0, 16)}`;
}
function operatorDmStorePath(): string {
return resolve(config.storageRoot, "status", "cultmesh", "voidbot-swarm-state.cc");
}
function loadCultMeshRuntime(): { CultMesh: any; defineDocumentType: any } | null {
const candidates = [
resolve(process.cwd(), "..", "CultLib", "packages", "cultmesh-ts", "package.json"),
resolve(process.cwd(), "..", "CultMeshTS", "package.json"),
];
for (const packageJson of candidates) {
try {
const requireCult = createRequire(packageJson);
const { CultMesh } = requireCult("cultmesh-ts/dist/index.js");
const { defineDocumentType } = requireCult("cultcache-ts/dist/index.js");
if (CultMesh && defineDocumentType) {
return { CultMesh, defineDocumentType };
}
} catch {
continue;
}
}
return null;
}
async function postDiscordBotMessage(channelId: string, content: string): Promise<{ id: string }> {
if (!config.botToken) {
throw new Error("DISCORD_BOT_TOKEN is required for the worker to post Discord messages.");

View file

@ -22,6 +22,8 @@ const eveBindingDocumentType = "gamecult.eve.interface_binding";
const eveBindingSchemaId = "gamecult.eve.interface_binding.v1";
const surfaceDocumentType = "gamecult.eve.surface_state";
const surfaceSchemaId = "gamecult.eve.surface_state.v1";
const operatorDmRequestDocumentType = "gamecult.operator_dm_request";
const operatorDmRequestSchemaId = "gamecult.operator_dm_request.v1";
const args = parseArgs(process.argv.slice(2));
@ -701,7 +703,7 @@ async function writeCultMeshPublication(snapshot, eveState, storePath) {
schemaVersion: snapshotSchemaId,
description: "Redacted read-only VoidBot swarm state snapshot for CultMesh distribution.",
}),
global: true,
global: false,
schema: {
parse(input) {
if (!input || typeof input !== "object") {
@ -717,7 +719,7 @@ async function writeCultMeshPublication(snapshot, eveState, storePath) {
schemaId: providerAdvertisementSchemaId,
schemaVersion: providerAdvertisementSchemaId,
contentHash: providerAdvertisementSchemaId,
global: true,
global: false,
name: "providerId",
schema: { parse: parseObjectDocument("Eve provider advertisement") },
});
@ -727,7 +729,7 @@ async function writeCultMeshPublication(snapshot, eveState, storePath) {
schemaId: surfaceSchemaId,
schemaVersion: surfaceSchemaId,
contentHash: surfaceSchemaId,
global: true,
global: false,
name: "providerId",
schema: { parse: parseObjectDocument("Eve surface state") },
});
@ -737,30 +739,38 @@ async function writeCultMeshPublication(snapshot, eveState, storePath) {
schemaId: eveBindingSchemaId,
schemaVersion: eveBindingSchemaId,
contentHash: eveBindingSchemaId,
global: true,
global: false,
name: "bindingId",
schema: { parse: parseObjectDocument("Eve interface binding") },
});
const operatorDmRequestDefinition = defineDocumentType({
type: operatorDmRequestDocumentType,
schemaName: operatorDmRequestDocumentType,
schemaId: operatorDmRequestSchemaId,
schemaVersion: operatorDmRequestSchemaId,
contentHash: operatorDmRequestSchemaId,
global: false,
name: "requestId",
schema: { parse: parseObjectDocument("Operator DM request") },
});
const node = await CultMesh.createNode(storePath, {
documents: [snapshotDefinition, providerDefinition, surfaceDefinition, eveBindingDefinition],
documents: [snapshotDefinition, providerDefinition, surfaceDefinition, eveBindingDefinition, operatorDmRequestDefinition],
});
const providerAdvertisement = buildProviderAdvertisement(snapshot);
const eveBinding = buildEveInterfaceBinding(snapshot, eveState);
const operatorDmState = buildOperatorDmProviderState(snapshot);
const operatorDmAdvertisement = buildOperatorDmProviderAdvertisement(snapshot);
const operatorDmBinding = buildOperatorDmInterfaceBinding(snapshot, operatorDmState);
await node.put(snapshotDefinition, "voidbot-swarm", snapshot);
await node.put(providerDefinition, providerId, providerAdvertisement);
await node.put(providerDefinition, operatorDmProviderId, operatorDmAdvertisement);
await node.put(surfaceDefinition, providerId, eveState);
await node.put(surfaceDefinition, operatorDmProviderId, operatorDmState);
await node.put(eveBindingDefinition, providerId, eveBinding);
await node.put(eveBindingDefinition, operatorDmProviderId, operatorDmBinding);
await node.flush?.(true);
await node.delete(surfaceDefinition, operatorDmProviderId);
await node.delete(eveBindingDefinition, operatorDmProviderId);
await node.flush?.();
return {
writeStatus: "ok",
storePath,
documents: [snapshotDocumentType, providerAdvertisementDocumentType, surfaceDocumentType, eveBindingDocumentType],
documents: [snapshotDocumentType, providerAdvertisementDocumentType, surfaceDocumentType, eveBindingDocumentType, operatorDmRequestDocumentType],
providerId,
verseId,
eveBinding: {
@ -975,116 +985,14 @@ function buildOperatorDmProviderAdvertisement(snapshot) {
type: eveBindingDocumentType,
schemaId: eveBindingSchemaId,
key: operatorDmProviderId,
}, {
type: operatorDmRequestDocumentType,
schemaId: operatorDmRequestSchemaId,
key: "requestId",
}],
};
}
function buildOperatorDmInterfaceBinding(snapshot, eveState) {
return {
schemaVersion: eveBindingSchemaId,
bindingId: operatorDmProviderId,
providerId: operatorDmProviderId,
verseId,
title: "VoidBot Operator DM",
kind: "eve-cultui-command-surface",
updatedAt: snapshot.generatedAt,
authority: {
owner: "VoidBot Discord owner-DM helper",
commandOwner: "VoidBot worker notify_owner path",
sourceDocuments: [{
type: providerAdvertisementDocumentType,
schemaId: providerAdvertisementSchemaId,
key: operatorDmProviderId,
}],
surfaceDocument: {
type: surfaceDocumentType,
schemaId: surfaceSchemaId,
key: operatorDmProviderId,
},
forbiddenWriters: ["Mimir Well", "Odin", "Eve renderers"],
},
surface: eveState.surface,
stateSummary: {
state: ownerDmStatus(snapshot),
botTokenConfigured: Boolean(env.DISCORD_BOT_TOKEN),
ownerConfigured: Boolean(env.DISCORD_OWNER_ID),
command: "owner.dm.send",
},
controls: [ownerDmCommandDescriptor()],
commands: [ownerDmCommandDescriptor()],
rendererHints: {
preferredLowerings: ["nightwing-tui", "eve-native"],
viewportMode: "operator-alerts",
tileId: "voidbot-operator-dm",
minWidth: 44,
minHeight: 8,
preferredWidth: 72,
preferredHeight: 14,
priority: -80,
density: "compact",
},
};
}
function buildOperatorDmProviderState(snapshot) {
const state = ownerDmStatus(snapshot);
return {
schema: surfaceSchemaId,
providerId: operatorDmProviderId,
title: "VoidBot Operator DM",
version: Number(snapshot.summary?.initiativeClock ?? Date.now()),
updatedAt: snapshot.generatedAt,
surface: {
schema: "gamecult.eve.surface.v1",
id: "voidbot.operator-dm.surface",
title: "VoidBot Operator DM",
commands: [ownerDmCommandDescriptor()],
root: {
id: "voidbot-operator-dm-root",
kind: "pane",
props: {
title: "Operator DM",
providerId: operatorDmProviderId,
status: state,
summary: state === "active"
? "owner.dm.send is advertised; VoidBot owns delivery."
: "owner.dm.send is advertised but Discord credentials are incomplete.",
command: "owner.dm.send",
},
children: [{
id: "voidbot-operator-dm-status",
kind: "metric",
props: {
label: "Owner DM",
value: state,
tone: state === "active" ? "ok" : "warn",
},
children: [],
}, {
id: "voidbot-operator-dm-command",
kind: "button",
props: {
label: "Send owner DM",
command: "owner.dm.send",
action: ownerDmCommandAction(),
status: state,
},
children: [],
}, {
id: "voidbot-operator-dm-contract",
kind: "text",
props: {
text: "Mimir may request operator intervention through gamecult.operator_dm_request.v1; VoidBot owns Discord delivery.",
},
children: [],
}],
},
assets: [],
},
commands: [ownerDmCommandDescriptor()],
};
}
function ownerDmCommandDescriptor() {
return {
id: "owner.dm.send",
@ -1092,7 +1000,7 @@ function ownerDmCommandDescriptor() {
command: "owner.dm.send",
transport: "cultmesh-command",
payloadSchema: {
schemaId: "gamecult.operator_dm_request.v1",
schemaId: operatorDmRequestSchemaId,
type: "object",
required: ["message"],
properties: {
@ -1104,8 +1012,8 @@ function ownerDmCommandDescriptor() {
},
},
writes: {
documentType: "gamecult.operator_dm_request",
schemaId: "gamecult.operator_dm_request.v1",
documentType: operatorDmRequestDocumentType,
schemaId: operatorDmRequestSchemaId,
owner: operatorDmProviderId,
},
action: ownerDmCommandAction(),
@ -1118,7 +1026,7 @@ function ownerDmCommandAction() {
command: "owner.dm.send",
transport: "cultmesh-command",
providerId: operatorDmProviderId,
payloadSchemaId: "gamecult.operator_dm_request.v1",
payloadSchemaId: operatorDmRequestSchemaId,
};
}

View file

@ -332,3 +332,5 @@
{"ts":"2026-06-04T10:51:11Z","type":"implementation","status":"ok","note":"Fixed repo Face active-stack fault: worker now claims one runnable job per poll, and heartbeat reconciliation treats running jobs as active while approved jobs only block duplicate queueing. Reset eight over-claimed running jobs to approved and restarted worker PID 36736; visible active count dropped to one."}
{"ts":"2026-06-04T12:24:21Z","type":"implementation","status":"ok","note":"Watchdog backup discovery now ignores non-timestamped restorepoint directories when choosing latest local backups, so ad hoc Face-state restorepoints under .voidbot/backups no longer masquerade as incomplete scheduled backups. The 2026-06-04 ops smoke still fails only on Raven/192.168.1.84 reachability for remote Ollama and offsite SSH."}
{"ts":"2026-06-04T12:52:27Z","type":"implementation","status":"ok","note":"VoidBot now publishes a provider-owned CultMesh/Eve command surface as voidbot.operator-dm in .voidbot/status/cultmesh/voidbot-swarm-state.cc. It advertises owner.dm.send with payload schema gamecult.operator_dm_request.v1 for Mimir Well intervention alerts; Odin discovers it as an active command-surface provider."}
{"ts":"2026-06-04T13:37:20Z","type":"implementation","status":"ok","note":"VoidBot worker now drains gamecult.operator_dm_request.v1 documents from the provider-owned CultMesh store and delivers pending owner.dm.send requests through the existing Discord owner-DM path, marking requests sent or failed in place. Eve provider documents are keyed, not global, so one store can hold both voidbot.swarm and voidbot.operator-dm."}
{"ts":"2026-06-04T13:42:16Z","type":"correction","status":"ok","note":"Corrected VoidBot operator-DM exposure: voidbot.operator-dm is a provider advertisement and command inbox, not an Eve dashboard/interface slot. Odin lists the capability in /eve/deck/providers while the Odin surface only lowers real provider interfaces such as voidbot.swarm."}