Watchdog notice reader: Idunn incidents become owner DMs through Bifrost's own journal
tools/watchdog-notice.mjs is a one-shot reader. It reads Idunn's incident store unlocked with SingleFileMessagePackBackingStore.pullAll, sends one opening and one closing DM per incident through bifrost-bridge.mjs, and keeps a bifrost.watchdog_notice_execution.v1 journal: a running entry becomes unknown and is never re-posted, a failed one is retried with the same nonce up to five times, terminal entries are pruned once Idunn retires the incident. The exit status is the delivery-failure signal. discord-dm accepts --nonce (at most 25 chars) and sends nonce with enforce_nonce so a retry inside Discord's window returns the original message. The tests run the CLI against the store bytes Idunn wrote (fixture from GameCult/Idunn 7f7528b3) with a counting bridge double, and once through the real bridge with fetch stubbed. docs/watchdog-notice.md states the CultLib >= 8cb3b728 floor. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
4152676ebb
commit
417454c28f
5 changed files with 653 additions and 2 deletions
|
|
@ -475,6 +475,10 @@ async function sendDiscordDm(options) {
|
|||
const token = process.env.BIFROST_DISCORD_BOT_TOKEN ?? process.env.DISCORD_BOT_TOKEN;
|
||||
const recipientId = requireOption(options, "recipient-id");
|
||||
const content = await readOptionText(options, "content", "content-file");
|
||||
const nonce = optionalString(options.nonce);
|
||||
if (options.nonce !== undefined && (!nonce || nonce.length > 25)) {
|
||||
throw new Error("--nonce must be 1 to 25 characters.");
|
||||
}
|
||||
const dryRun = options["dry-run"] === "true";
|
||||
|
||||
if (dryRun) {
|
||||
|
|
@ -506,7 +510,7 @@ async function sendDiscordDm(options) {
|
|||
throw new Error("Set BIFROST_DISCORD_BOT_TOKEN or DISCORD_BOT_TOKEN before sending a Discord DM.");
|
||||
}
|
||||
channelId = await openDiscordDmChannel(token, recipientId);
|
||||
result = await postDiscordBotMessage(token, channelId, content, undefined);
|
||||
result = await postDiscordBotMessage(token, channelId, content, undefined, nonce);
|
||||
await bridgeAction?.complete({
|
||||
receiptUrl: `https://discord.com/channels/@me/${channelId}/${result.id}`,
|
||||
externalReceiptId: result.id,
|
||||
|
|
@ -755,7 +759,7 @@ async function openDiscordDmChannel(token, recipientId) {
|
|||
return channel.id;
|
||||
}
|
||||
|
||||
async function postDiscordBotMessage(token, channelId, content, replyToMessageId) {
|
||||
async function postDiscordBotMessage(token, channelId, content, replyToMessageId, nonce) {
|
||||
const response = await fetch(`https://discord.com/api/v10/channels/${channelId}/messages`, {
|
||||
method: "POST",
|
||||
headers: {
|
||||
|
|
@ -773,6 +777,7 @@ async function postDiscordBotMessage(token, channelId, content, replyToMessageId
|
|||
allowed_mentions: {
|
||||
parse: [],
|
||||
},
|
||||
...(nonce ? { nonce, enforce_nonce: true } : {}),
|
||||
}),
|
||||
});
|
||||
|
||||
|
|
|
|||
284
tools/watchdog-notice.mjs
Normal file
284
tools/watchdog-notice.mjs
Normal file
|
|
@ -0,0 +1,284 @@
|
|||
#!/usr/bin/env node
|
||||
// One-shot reader: turns Idunn's operator incidents into owner DMs.
|
||||
//
|
||||
// process --incident-store P --journal-store J --receipt-store R [--bridge-cli B]
|
||||
// status --journal-store J
|
||||
//
|
||||
// Owner of the decision to send, the journal (J), the recipient
|
||||
// (DISCORD_OWNER_ID) and the token (BIFROST_DISCORD_BOT_TOKEN, read only by the
|
||||
// bridge it spawns). Idunn's store P is read unlocked and never written: no
|
||||
// CultMesh node, no lock, only SingleFileMessagePackBackingStore.pullAll().
|
||||
//
|
||||
// Requires CultLib 8cb3b728 or later (package @gamecult/cultcache-ts); see
|
||||
// docs/watchdog-notice.md.
|
||||
//
|
||||
// Nothing here echoes an input value. Errors are fixed codes, and the
|
||||
// journal's lastError is a fixed code, never bridge output.
|
||||
import { createHash } from "node:crypto";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { createRequire } from "node:module";
|
||||
import { mkdtemp, rm, stat, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { resolve } from "node:path";
|
||||
|
||||
const root = resolve(import.meta.dirname, "..");
|
||||
const cult = resolve(process.env.VOIDBOT_CULTLIB_ROOT || resolve(root, "..", "CultLib"));
|
||||
|
||||
const INCIDENT_SCHEMA = "idunn.operator_incident.v1";
|
||||
const JOURNAL_TYPE = "bifrost.watchdog_notice_execution";
|
||||
const JOURNAL_SCHEMA = "bifrost.watchdog_notice_execution.v1";
|
||||
const MAX_ATTEMPTS = 5;
|
||||
const BRIDGE_TIMEOUT_MS = 60_000;
|
||||
const STATUSES = ["running", "completed", "failed", "unknown"];
|
||||
const NAME = /^[a-z][a-z-]{0,63}$/;
|
||||
const SUBJECT = /^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$/;
|
||||
|
||||
class NoticeError extends Error {}
|
||||
const fail = (code) => new NoticeError(code);
|
||||
|
||||
try {
|
||||
process.exitCode = await main(process.argv.slice(2));
|
||||
} catch (error) {
|
||||
process.stderr.write(`watchdog-notice: ${error instanceof NoticeError ? error.message : "unexpected-error"}\n`);
|
||||
process.exitCode = 1;
|
||||
}
|
||||
|
||||
async function main(args) {
|
||||
const [verb, ...rest] = args;
|
||||
const options = parseArgs(rest);
|
||||
if (verb === "status") return status(options);
|
||||
if (verb === "process") return processIncidents(options);
|
||||
throw fail("unknown-verb");
|
||||
}
|
||||
|
||||
async function processIncidents(options) {
|
||||
const incidentStore = requireOption(options, "incident-store");
|
||||
const journalStore = requireOption(options, "journal-store");
|
||||
const receiptStore = requireOption(options, "receipt-store");
|
||||
const bridgeCli = resolve(options["bridge-cli"] || resolve(root, "tools", "bifrost-bridge.mjs"));
|
||||
const recipientId = (process.env.DISCORD_OWNER_ID ?? "").trim();
|
||||
if (!recipientId) throw fail("recipient-missing");
|
||||
|
||||
const runtime = loadRuntime();
|
||||
const { records, skipped, refused } = await readIncidents(runtime, incidentStore);
|
||||
if (records === null) return 0;
|
||||
|
||||
const journal = await openJournal(runtime, journalStore);
|
||||
for (const entry of journal.cache.getAll(journal.definition)) {
|
||||
if (entry.status === "running") {
|
||||
await journal.put({ ...entry, status: "unknown", lastError: "interrupted-before-outcome", updatedAt: now() });
|
||||
}
|
||||
}
|
||||
|
||||
const deliver = async (record, notice) => {
|
||||
const key = `${record.incidentKey}#${notice}`;
|
||||
const previous = journal.cache.get(journal.definition, key);
|
||||
const attempts = (previous?.attempts ?? 0) + 1;
|
||||
const nonce = createHash("sha256").update(key).digest("hex").slice(0, 25);
|
||||
const base = { schemaVersion: JOURNAL_SCHEMA, incidentKey: record.incidentKey, notice, attempts, nonce };
|
||||
await journal.put({ ...base, status: "running", messageId: "", lastError: "", updatedAt: now() });
|
||||
const outcome = await sendNotice({ bridgeCli, recipientId, receiptStore, key, nonce, record, notice });
|
||||
const entry = outcome.messageId
|
||||
? { ...base, status: "completed", messageId: outcome.messageId, lastError: "" }
|
||||
: { ...base, status: "failed", messageId: "", lastError: outcome.error };
|
||||
await journal.put({ ...entry, updatedAt: now() });
|
||||
};
|
||||
const retryable = (entry) => !entry || (entry.status === "failed" && entry.attempts < MAX_ATTEMPTS);
|
||||
const entryFor = (record, notice) => journal.cache.get(journal.definition, `${record.incidentKey}#${notice}`);
|
||||
|
||||
for (const record of records) {
|
||||
if (retryable(entryFor(record, "opened"))) await deliver(record, "opened");
|
||||
if (record.closedAt !== null && entryFor(record, "opened")?.status === "completed" && retryable(entryFor(record, "closed"))) {
|
||||
await deliver(record, "closed");
|
||||
}
|
||||
}
|
||||
|
||||
const present = new Set(records.map((record) => record.incidentKey));
|
||||
for (const entry of journal.cache.getAll(journal.definition)) {
|
||||
if (isTerminal(entry) && !present.has(entry.incidentKey)) {
|
||||
await journal.cache.delete(journal.definition, `${entry.incidentKey}#${entry.notice}`);
|
||||
}
|
||||
}
|
||||
|
||||
const entries = journal.cache.getAll(journal.definition);
|
||||
const unknown = entries.filter((entry) => entry.status === "unknown").length;
|
||||
const exhausted = entries.filter(isExhausted).length;
|
||||
process.stdout.write(`${JSON.stringify({ skipped, refused, unknown, exhausted })}\n`);
|
||||
return unknown + exhausted + refused > 0 ? 1 : 0;
|
||||
}
|
||||
|
||||
async function status(options) {
|
||||
const runtime = loadRuntime();
|
||||
const journal = await openJournal(runtime, requireOption(options, "journal-store"));
|
||||
const entries = journal.cache.getAll(journal.definition);
|
||||
const rows = entries
|
||||
.map(({ incidentKey, notice, status, attempts, updatedAt, lastError }) => ({ incidentKey, notice, status, attempts, updatedAt, lastError }))
|
||||
.sort((a, b) => (a.incidentKey + a.notice).localeCompare(b.incidentKey + b.notice));
|
||||
process.stdout.write(`${JSON.stringify(rows, null, 2)}\n`);
|
||||
return entries.some((entry) => entry.status === "unknown" || isExhausted(entry)) ? 1 : 0;
|
||||
}
|
||||
|
||||
const isExhausted = (entry) => entry.status === "failed" && entry.attempts >= MAX_ATTEMPTS;
|
||||
const isTerminal = (entry) => entry.status === "completed" || entry.status === "unknown" || isExhausted(entry);
|
||||
const now = () => new Date().toISOString();
|
||||
|
||||
// Reads P unlocked. An absent file is "nothing to do" (records: null). A
|
||||
// record of another schema is counted as skipped; a record of this schema
|
||||
// that breaks the contract is counted as refused. Neither is echoed.
|
||||
async function readIncidents(runtime, path) {
|
||||
try {
|
||||
await stat(path);
|
||||
} catch (error) {
|
||||
if (error?.code === "ENOENT") return { records: null, skipped: 0, refused: 0 };
|
||||
throw fail("incident-store-unreadable");
|
||||
}
|
||||
let envelopes;
|
||||
try {
|
||||
envelopes = await new runtime.SingleFileMessagePackBackingStore(path).pullAll();
|
||||
} catch {
|
||||
throw fail("incident-store-unreadable");
|
||||
}
|
||||
const records = [];
|
||||
let skipped = 0;
|
||||
let refused = 0;
|
||||
for (const envelope of envelopes) {
|
||||
let tuple;
|
||||
try {
|
||||
tuple = runtime.decode(envelope.payload);
|
||||
} catch {
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
if (!Array.isArray(tuple) || tuple[0] !== INCIDENT_SCHEMA) {
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
const record = parseIncident(tuple);
|
||||
if (record) records.push(record);
|
||||
else refused += 1;
|
||||
}
|
||||
records.sort((a, b) => a.openedAt - b.openedAt || a.incidentKey.localeCompare(b.incidentKey));
|
||||
return { records, skipped, refused };
|
||||
}
|
||||
|
||||
// [schema_version, incident_key, condition, subject, opened_at, closed_at, close_reason]
|
||||
function parseIncident(tuple) {
|
||||
if (tuple.length !== 7) return null;
|
||||
const [, incidentKey, condition, subject, openedAt, closedAt, closeReason] = tuple;
|
||||
if (typeof condition !== "string" || !NAME.test(condition)) return null;
|
||||
if (typeof subject !== "string" || !SUBJECT.test(subject)) return null;
|
||||
if (!Number.isSafeInteger(openedAt) || openedAt <= 0) return null;
|
||||
if (incidentKey !== `${condition}:${subject}:${openedAt}`) return null;
|
||||
if (closedAt === null) {
|
||||
if (closeReason !== null) return null;
|
||||
} else {
|
||||
if (!Number.isSafeInteger(closedAt) || closedAt < openedAt) return null;
|
||||
if (typeof closeReason !== "string" || !NAME.test(closeReason)) return null;
|
||||
}
|
||||
return { incidentKey, condition, subject, openedAt, closedAt, closeReason };
|
||||
}
|
||||
|
||||
function noticeText(record, notice) {
|
||||
const opened = new Date(record.openedAt).toISOString();
|
||||
if (notice === "opened") return `Idunn incident opened: ${record.condition} on ${record.subject} at ${opened}`;
|
||||
const closed = new Date(record.closedAt).toISOString();
|
||||
return `Idunn incident closed (${record.closeReason}): ${record.condition} on ${record.subject}, opened ${opened}, closed ${closed}`;
|
||||
}
|
||||
|
||||
// The one send primitive. The outcome carries a message id, or a fixed error
|
||||
// code; bridge stdout and stderr are never copied into the journal.
|
||||
async function sendNotice({ bridgeCli, recipientId, receiptStore, key, nonce, record, notice }) {
|
||||
const dir = await mkdtemp(resolve(tmpdir(), "watchdog-notice-"));
|
||||
try {
|
||||
const contentFile = resolve(dir, "content.txt");
|
||||
await writeFile(contentFile, noticeText(record, notice), "utf8");
|
||||
const result = spawnSync(process.execPath, [
|
||||
bridgeCli, "discord-dm",
|
||||
"--recipient-id", recipientId,
|
||||
"--content-file", contentFile,
|
||||
"--nonce", nonce,
|
||||
"--cultmesh-command-id", key,
|
||||
"--receipt-store", receiptStore,
|
||||
"--source-kind", "idunn-operator-incident",
|
||||
"--source-id", record.incidentKey,
|
||||
], { encoding: "utf8", env: process.env, windowsHide: true, timeout: BRIDGE_TIMEOUT_MS, stdio: ["ignore", "pipe", "ignore"] });
|
||||
if (result.error) return { error: "bridge-spawn-failed" };
|
||||
if (result.status !== 0) return { error: "bridge-exit-nonzero" };
|
||||
try {
|
||||
const value = JSON.parse(result.stdout);
|
||||
if (value.action === "discord-dm" && value.ok === true && typeof value.messageId === "string" && value.messageId) {
|
||||
return { messageId: value.messageId };
|
||||
}
|
||||
} catch {
|
||||
// falls through to the fixed code
|
||||
}
|
||||
return { error: "bridge-output-invalid" };
|
||||
} finally {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
}
|
||||
|
||||
async function openJournal(runtime, path) {
|
||||
const definition = runtime.defineDocumentType({
|
||||
type: JOURNAL_TYPE,
|
||||
schemaName: JOURNAL_TYPE,
|
||||
schemaId: JOURNAL_SCHEMA,
|
||||
schemaVersion: JOURNAL_SCHEMA,
|
||||
contentHash: JOURNAL_SCHEMA,
|
||||
global: false,
|
||||
schema: { parse: parseEntry },
|
||||
members: ["schemaVersion", "incidentKey", "notice", "status", "attempts", "nonce", "messageId", "lastError", "updatedAt"]
|
||||
.map((memberName, slot) => ({ slot, memberName, typeName: memberName === "attempts" ? "uint32" : "string" })),
|
||||
});
|
||||
const cache = runtime.CultCache.builder()
|
||||
.withDocumentType(definition)
|
||||
.withGenericStore(new runtime.SingleFileMessagePackBackingStore(path))
|
||||
.build();
|
||||
try {
|
||||
await cache.pullAllBackingStores();
|
||||
} catch {
|
||||
throw fail("journal-unreadable");
|
||||
}
|
||||
return { cache, definition, put: (entry) => cache.put(definition, `${entry.incidentKey}#${entry.notice}`, entry) };
|
||||
}
|
||||
|
||||
function parseEntry(value) {
|
||||
const ok = value && value.schemaVersion === JOURNAL_SCHEMA
|
||||
&& typeof value.incidentKey === "string" && value.incidentKey
|
||||
&& (value.notice === "opened" || value.notice === "closed")
|
||||
&& STATUSES.includes(value.status)
|
||||
&& Number.isSafeInteger(value.attempts) && value.attempts >= 0
|
||||
&& ["nonce", "messageId", "lastError", "updatedAt"].every((field) => typeof value[field] === "string");
|
||||
if (!ok) throw fail("journal-entry-invalid");
|
||||
return value;
|
||||
}
|
||||
|
||||
function loadRuntime() {
|
||||
try {
|
||||
const cache = createRequire(resolve(cult, "packages", "cultcache-ts", "package.json"));
|
||||
const net = createRequire(resolve(cult, "packages", "cultnet-ts", "package.json"));
|
||||
const cc = cache("@gamecult/cultcache-ts");
|
||||
return {
|
||||
CultCache: cc.CultCache,
|
||||
SingleFileMessagePackBackingStore: cc.SingleFileMessagePackBackingStore,
|
||||
defineDocumentType: cc.defineDocumentType,
|
||||
decode: net("@msgpack/msgpack").decode,
|
||||
};
|
||||
} catch {
|
||||
throw fail("cultlib-unavailable");
|
||||
}
|
||||
}
|
||||
|
||||
function parseArgs(args) {
|
||||
const options = {};
|
||||
for (let i = 0; i < args.length; i += 2) {
|
||||
if (!args[i].startsWith("--") || args[i + 1] === undefined) throw fail("bad-arguments");
|
||||
options[args[i].slice(2)] = args[i + 1];
|
||||
}
|
||||
return options;
|
||||
}
|
||||
|
||||
function requireOption(options, name) {
|
||||
if (!options[name]) throw fail(`missing-option-${name}`);
|
||||
return options[name];
|
||||
}
|
||||
Loading…
Reference in a new issue