From 417454c28f8776b5ec0b56a52d411194e6ebce40 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 30 Sep 2026 21:15:18 +0200 Subject: [PATCH] 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 --- docs/watchdog-notice.md | 39 +++ tests/fixtures/idunn.operator_incident.v1.cc | Bin 0 -> 737 bytes tests/watchdog-notice.test.mjs | 323 +++++++++++++++++++ tools/bifrost-bridge.mjs | 9 +- tools/watchdog-notice.mjs | 284 ++++++++++++++++ 5 files changed, 653 insertions(+), 2 deletions(-) create mode 100644 docs/watchdog-notice.md create mode 100644 tests/fixtures/idunn.operator_incident.v1.cc create mode 100644 tests/watchdog-notice.test.mjs create mode 100644 tools/watchdog-notice.mjs diff --git a/docs/watchdog-notice.md b/docs/watchdog-notice.md new file mode 100644 index 0000000..50e8235 --- /dev/null +++ b/docs/watchdog-notice.md @@ -0,0 +1,39 @@ +# Watchdog notice reader + +`tools/watchdog-notice.mjs` is a one-shot reader. It turns Idunn's operator +incidents into Discord DMs to the owner. Run it from a timer. + +``` +node tools/watchdog-notice.mjs process --incident-store P --journal-store J --receipt-store R [--bridge-cli B] +node tools/watchdog-notice.mjs status --journal-store J +``` + +## Requirements + +- **CultLib at `8cb3b728` or later.** The reader loads `@gamecult/cultcache-ts`, + the name CultLib gave the package in `8cb3b728` (2026-09-04), from + `$VOIDBOT_CULTLIB_ROOT` or the sibling `../CultLib`. An older CultLib fails at + import with `MODULE_NOT_FOUND`; the reader reports `cultlib-unavailable`. The + same floor applies to every Bifrost tool that loads CultCache + (`tools/persona-feedback.mjs` and the others). A deployment pairs a Bifrost + release with a CultLib commit at or after this one. +- `DISCORD_OWNER_ID` in the environment: the only source of the recipient. The + reader exits non-zero before writing the journal if it is missing. +- `BIFROST_DISCORD_BOT_TOKEN` in the environment, read only by + `bifrost-bridge.mjs`, which the reader spawns. + +## Behaviour + +- P is Idunn's `idunn.operator_incident.v1` store. The reader reads it without a + lock and never writes it. An absent P is a clean no-op. +- One notice when an incident opens and one when it closes, each at most once. + The closing notice is sent only after the opening notice completed. +- J is Bifrost's journal, `bifrost.watchdog_notice_execution.v1`, keyed + `#opened` or `#closed`. A `running` entry found at + start becomes `unknown` and is never sent again. A `failed` entry is retried on + later runs with the same nonce, up to five attempts. Terminal entries are + deleted once Idunn no longer lists the incident. +- The exit status is non-zero while any entry is `unknown` or out of attempts, or + while P holds a record of this schema that breaks its contract. systemd shows + that; `status` lists the entries. +- Notice text is built from `condition`, `subject` and ISO UTC times only. diff --git a/tests/fixtures/idunn.operator_incident.v1.cc b/tests/fixtures/idunn.operator_incident.v1.cc new file mode 100644 index 0000000000000000000000000000000000000000..180c6f0f2ae325984fa3c77bc4f3d193916ac76d GIT binary patch literal 737 zcmbQdDY-PKBsnoTBUP`sB)=$Cugq}b^j(=LrFnUJ`30#(i9o^l%)I2xl+?TupyYNO zQhRZT1GNAV9s_S?R4Ww&tZVp?B$gJJq^4LU=j0csrs$R=7Nw__SQ(ld zAOIfob{ZKN8S5Gv>KdAa7#Ud^npzncXd4&+$s@h^f|qnhY{BO0wJ5GW&%nUgeg0_% z1BCJH-GY@xsmb|esYR(NQ*P>EGr$TM_jzcZHU#^eJWp5P^)!i2!0P4ING^u^^S}WB D;7=HM literal 0 HcmV?d00001 diff --git a/tests/watchdog-notice.test.mjs b/tests/watchdog-notice.test.mjs new file mode 100644 index 0000000..3bf1dd2 --- /dev/null +++ b/tests/watchdog-notice.test.mjs @@ -0,0 +1,323 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createHash } from "node:crypto"; +import { createRequire } from "node:module"; +import { spawnSync } from "node:child_process"; +import { mkdtemp, mkdir, readFile, readdir, stat, writeFile, copyFile } from "node:fs/promises"; +import { existsSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { resolve } from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; + +// The fixture is the store Idunn's incident cut wrote: GameCult/Idunn branch +// hands/watchdog-idunn-incident, commit 7f7528b33e8e5c607c19cc7f8f8a8da4def50fca +// ("Add the idunn.operator_incident.v1 wire fixture"), tests/fixtures/ there. +// It holds two records: closed-target (closed, recovered) and open-target. +const FIXTURE = resolve(import.meta.dirname, "fixtures", "idunn.operator_incident.v1.cc"); +const FIXTURE_SHA256 = "f446770b07da99204e9036ef521f3136be6251e5b0a68e4e9533a89f25f18cab"; +const CLOSED_KEY = "continuity-exhausted:closed-target:1700000000000"; +const OPEN_KEY = "continuity-exhausted:open-target:1700000100000"; +const TOOL = resolve(import.meta.dirname, "..", "tools", "watchdog-notice.mjs"); +const BRIDGE = resolve(import.meta.dirname, "..", "tools", "bifrost-bridge.mjs"); +const RECIPIENT = "424242RECIPIENTCANARY"; + +const OPENED_TEXT = (subject, at) => `Idunn incident opened: continuity-exhausted on ${subject} at ${at}`; +const T_OPEN = "2023-11-14T22:13:20.000Z"; +const T_CLOSE = "2023-11-14T22:15:00.000Z"; +const CLOSED_TEXT = `Idunn incident closed (recovered): continuity-exhausted on closed-target, opened ${T_OPEN}, closed ${T_CLOSE}`; + +// CultLib is the sibling ../CultLib, like the other tests. +const cult = resolve(import.meta.dirname, "..", "..", "CultLib"); +const cache = createRequire(resolve(cult, "packages", "cultcache-ts", "package.json"))("@gamecult/cultcache-ts"); +const msgpack = createRequire(resolve(cult, "packages", "cultnet-ts", "package.json"))("@msgpack/msgpack"); + +// A bridge double that logs each call (with the content it was handed) and +// behaves as the `mode` file says: ok, fail, or crash (kills its parent, the +// reader, after logging, as a power cut between send and journal would). +const FAKE_BRIDGE = ` +import { appendFileSync, readFileSync } from "node:fs"; +import { dirname, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; +const here = dirname(fileURLToPath(import.meta.url)); +const args = process.argv.slice(2); +const opt = (name) => args[args.indexOf("--" + name) + 1]; +const calls = resolve(here, "calls.jsonl"); +const n = (() => { try { return readFileSync(calls, "utf8").trim().split("\\n").length; } catch { return 0; } })(); +appendFileSync(calls, JSON.stringify({ args, content: readFileSync(opt("content-file"), "utf8") }) + "\\n"); +const mode = readFileSync(resolve(here, "mode"), "utf8").trim(); +if (mode === "crash") { process.kill(process.ppid, "SIGKILL"); process.exit(0); } +if (mode === "fail") { process.stderr.write("boom " + opt("recipient-id")); process.stdout.write("boom " + opt("recipient-id")); process.exit(1); } +process.stdout.write(JSON.stringify({ action: "discord-dm", ok: true, messageId: "m" + (n + 1) })); +`; + +async function world(prefix, { incidents = FIXTURE, mode = "ok" } = {}) { + const dir = await mkdtemp(resolve(tmpdir(), prefix)); + const w = { + dir, + incidents: resolve(dir, "idunn", "incidents.cc"), + journal: resolve(dir, "bifrost", "journal.cc"), + receipts: resolve(dir, "bifrost", "receipts.cc"), + bridge: resolve(dir, "fake-bridge.mjs"), + calls: resolve(dir, "calls.jsonl"), + }; + await mkdir(resolve(dir, "idunn")); + await writeFile(w.bridge, FAKE_BRIDGE); + await writeFile(resolve(dir, "mode"), mode); + if (incidents) await copyFile(incidents, w.incidents); + w.setMode = (value) => writeFile(resolve(dir, "mode"), value); + w.run = (extra = {}) => run(w, extra); + w.status = () => runStatus(w); + w.posts = async () => (existsSync(w.calls) ? (await readFile(w.calls, "utf8")).trim().split("\n").map((line) => JSON.parse(line)) : []); + return w; +} + +function run(w, { env = {}, bridge = w.bridge } = {}) { + return spawnSync(process.execPath, [TOOL, "process", "--incident-store", w.incidents, "--journal-store", w.journal, "--receipt-store", w.receipts, "--bridge-cli", bridge], { + encoding: "utf8", + env: { ...process.env, DISCORD_OWNER_ID: RECIPIENT, ...env }, + }); +} + +function runStatus(w) { + const result = spawnSync(process.execPath, [TOOL, "status", "--journal-store", w.journal], { encoding: "utf8", env: process.env }); + return { ...result, rows: result.status === null || !result.stdout ? [] : JSON.parse(result.stdout) }; +} + +const opt = (post, name) => post.args[post.args.indexOf(`--${name}`) + 1]; +const notice = (post) => opt(post, "cultmesh-command-id").split("#")[1]; + +// Builds an incident store from the fixture's own envelopes with the payload +// tuples rewritten by `edit`. +async function variantStore(path, edit) { + const source = await new cache.SingleFileMessagePackBackingStore(FIXTURE).pullAll(); + const tuples = edit(source.map((envelope) => msgpack.decode(envelope.payload))); + const envelopes = tuples.map((tuple, index) => ({ ...source[0], key: `variant-${index}-${tuple[1]}`, payload: msgpack.encode(tuple), schemaId: source[0].schemaId, catalogEntry: source[0].catalogEntry })); + await new cache.SingleFileMessagePackBackingStore(path).pushAll(envelopes); +} + +test("the Idunn fixture decodes as the incidents Idunn wrote", async () => { + const bytes = await readFile(FIXTURE); + assert.equal(createHash("sha256").update(bytes).digest("hex"), FIXTURE_SHA256); + const w = await world("wn-fixture-"); + const result = w.run(); + assert.equal(result.status, 0, result.stderr); + const posts = await w.posts(); + const byKey = Object.fromEntries(posts.map((post) => [opt(post, "cultmesh-command-id"), post.content])); + assert.deepEqual(byKey, { + [`${CLOSED_KEY}#opened`]: OPENED_TEXT("closed-target", T_OPEN), + [`${CLOSED_KEY}#closed`]: CLOSED_TEXT, + [`${OPEN_KEY}#opened`]: OPENED_TEXT("open-target", T_CLOSE), + }); +}); + +test("one incident yields one opening post across runs", async () => { + const w = await world("wn-once-"); + assert.equal(w.run().status, 0); + assert.equal(w.run().status, 0); + const posts = await w.posts(); + const openings = posts.filter((post) => notice(post) === "opened"); + assert.equal(openings.length, 2); + assert.deepEqual(openings.map((post) => opt(post, "source-id")).sort(), [CLOSED_KEY, OPEN_KEY]); + assert.equal(posts.length, 3, "no second run posts again"); + for (const post of posts) { + const key = opt(post, "cultmesh-command-id"); + assert.equal(opt(post, "recipient-id"), RECIPIENT); + assert.equal(opt(post, "nonce"), createHash("sha256").update(key).digest("hex").slice(0, 25)); + assert.equal(opt(post, "receipt-store"), w.receipts); + assert.equal(opt(post, "source-kind"), "idunn-operator-incident"); + } +}); + +test("closure is posted once and only after a completed opening", async () => { + const w = await world("wn-closure-"); + assert.equal(w.run().status, 0); + assert.equal(w.run().status, 0); + const closures = (await w.posts()).filter((post) => notice(post) === "closed"); + assert.equal(closures.length, 1); + assert.equal(opt(closures[0], "source-id"), CLOSED_KEY); + + // An opening that never completes blocks its closure, now and later. + const failing = await world("wn-closure-failed-", { mode: "fail" }); + for (let run = 0; run < 5; run += 1) failing.run(); + await failing.setMode("ok"); + failing.run(); + failing.run(); + assert.equal((await failing.posts()).filter((post) => notice(post) === "closed").length, 0); + const rows = (await failing.status()).rows; + assert.deepEqual(rows.map((row) => `${row.notice}:${row.status}:${row.attempts}`), ["opened:failed:5", "opened:failed:5"]); + + const crashed = await world("wn-closure-unknown-", { mode: "crash" }); + assert.equal(crashed.run().status, null, "the reader died mid-send"); + await crashed.setMode("ok"); + crashed.run(); + crashed.run(); + assert.equal((await crashed.posts()).filter((post) => notice(post) === "closed").length, 0); +}); + +test("a running entry becomes unknown and is never re-posted", async () => { + const w = await world("wn-running-", { mode: "crash" }); + assert.equal(w.run().status, null); + assert.equal((await w.status()).rows[0].status, "running"); + await w.setMode("ok"); + const recovered = w.run(); + assert.notEqual(recovered.status, 0, "an unknown entry is a delivery failure"); + assert.equal(JSON.parse(recovered.stdout).unknown, 1); + const posts = await w.posts(); + assert.equal(posts.filter((post) => opt(post, "cultmesh-command-id") === `${CLOSED_KEY}#opened`).length, 1, "the interrupted notice is not sent again"); + const rows = (await w.status()).rows; + assert.equal(rows.find((row) => row.incidentKey === CLOSED_KEY && row.notice === "opened").status, "unknown"); + assert.notEqual(w.run().status, 0, "still failing on the next run"); + assert.equal((await w.posts()).length, posts.length); +}); + +test("a failed send is retried with the same nonce up to five attempts", async () => { + const w = await world("wn-retry-", { mode: "fail" }); + const exits = []; + for (let run = 0; run < 5; run += 1) exits.push(w.run().status); + assert.deepEqual(exits, [0, 0, 0, 0, 1], "non-zero once attempts are exhausted"); + const open = (await w.posts()).filter((post) => opt(post, "source-id") === OPEN_KEY); + assert.equal(open.length, 5); + assert.equal(new Set(open.map((post) => opt(post, "nonce"))).size, 1); + const before = (await w.posts()).length; + const again = w.run(); + assert.notEqual(again.status, 0); + assert.equal((await w.posts()).length, before, "no spawn after the fifth attempt"); + const status = await w.status(); + assert.notEqual(status.status, 0); + for (const text of [again.stdout, again.stderr, status.stdout, await readFile(w.journal, "latin1")]) { + assert.ok(!text.includes("RECIPIENTCANARY"), "bridge output and recipient never reach the reader's output or journal"); + } +}); + +test("the reader never writes or locks the incident store", async () => { + const w = await world("wn-readonly-"); + const before = { bytes: await readFile(w.incidents), mtime: (await stat(w.incidents)).mtimeMs }; + assert.equal(w.run().status, 0); + assert.equal(w.run().status, 0); + assert.deepEqual(await readFile(w.incidents), before.bytes); + assert.equal((await stat(w.incidents)).mtimeMs, before.mtime); + assert.deepEqual(await readdir(resolve(w.dir, "idunn")), ["incidents.cc"], "no lock or temp file beside the store"); +}); + +test("notice content carries no sensitive text", async () => { + const w = await world("wn-egress-", { incidents: null }); + await variantStore(w.incidents, (tuples) => { + const [closed, open] = tuples; + const longer = [...open, "extra-CANARYFIELD"]; + const pathSubject = ["idunn.operator_incident.v1", `continuity-exhausted:/etc/CANARYPATH:${open[4]}`, "continuity-exhausted", "/etc/CANARYPATH", open[4], null, null]; + const otherSchema = ["idunn.operator_incident.v0", "CANARYSCHEMA", "continuity-exhausted", "x", 1, null, null]; + return [closed, longer, pathSubject, otherSchema]; + }); + const result = w.run(); + assert.notEqual(result.status, 0, "refused records fail the run"); + assert.deepEqual(JSON.parse(result.stdout), { skipped: 1, refused: 2, unknown: 0, exhausted: 0 }); + const posts = await w.posts(); + assert.deepEqual(posts.map((post) => opt(post, "source-id")), [CLOSED_KEY, CLOSED_KEY]); + for (const post of posts) assert.match(post.content, /^Idunn incident (opened: continuity-exhausted on closed-target at \d{4}-[\d-]+T[\d:.]+Z|closed \(recovered\): continuity-exhausted on closed-target, opened [\dTZ:.-]+, closed [\dTZ:.-]+)$/); + for (const text of [result.stdout, result.stderr, (await w.status()).stdout, await readFile(w.journal, "latin1"), JSON.stringify(posts)]) { + assert.ok(!text.includes("CANARY"), "no byte of a refused record is echoed"); + } +}); + +test("no recipient means no journal write and a non-zero exit", async () => { + for (const value of [undefined, "", " "]) { + const w = await world("wn-recipient-"); + const env = value === undefined ? { DISCORD_OWNER_ID: undefined } : { DISCORD_OWNER_ID: value }; + const result = spawnSync(process.execPath, [TOOL, "process", "--incident-store", w.incidents, "--journal-store", w.journal, "--receipt-store", w.receipts, "--bridge-cli", w.bridge], { encoding: "utf8", env: { ...process.env, ...env } }); + assert.notEqual(result.status, 0); + assert.ok(!existsSync(w.journal)); + assert.deepEqual(await w.posts(), []); + } +}); + +test("an absent incident store is a clean no-op", async () => { + const w = await world("wn-absent-", { incidents: null }); + const result = w.run(); + assert.equal(result.status, 0, result.stderr); + assert.ok(!existsSync(w.journal)); + assert.deepEqual(await w.posts(), []); +}); + +test("terminal entries are pruned only after Idunn retires the incident", async () => { + const w = await world("wn-prune-"); + assert.equal(w.run().status, 0); + const keys = async () => (await w.status()).rows.map((row) => `${row.incidentKey}#${row.notice}`).sort(); + const all = [`${CLOSED_KEY}#closed`, `${CLOSED_KEY}#opened`, `${OPEN_KEY}#opened`]; + assert.deepEqual(await keys(), all); + assert.equal(w.run().status, 0); + assert.deepEqual(await keys(), all, "entries stay while Idunn still lists the incident"); + await variantStore(w.incidents, (tuples) => [tuples[1]]); + assert.equal(w.run().status, 0); + assert.deepEqual(await keys(), [`${OPEN_KEY}#opened`]); + await variantStore(w.incidents, () => []); + assert.equal(w.run().status, 0); + assert.deepEqual(await keys(), []); + assert.equal((await w.posts()).length, 3, "retirement never reposts"); +}); + +test("errors name a code and never echo a path or an input", async () => { + const w = await world("wn-canary-", { incidents: null }); + await mkdir(resolve(w.dir, "CANARYDIR")); + const corrupt = resolve(w.dir, "CANARYDIR", "incidents.cc"); + await writeFile(corrupt, "not a cultcache store CANARYBYTES"); + const result = spawnSync(process.execPath, [TOOL, "process", "--incident-store", corrupt, "--journal-store", resolve(w.dir, "CANARYDIR", "j.cc"), "--receipt-store", w.receipts, "--bridge-cli", w.bridge], { encoding: "utf8", env: { ...process.env, DISCORD_OWNER_ID: RECIPIENT } }); + assert.notEqual(result.status, 0); + assert.match(result.stderr, /incident-store-unreadable/); + assert.ok(!(result.stdout + result.stderr).includes("CANARY")); +}); + +// The real bridge, reached through the real reader, with Discord's HTTP API +// stubbed at fetch (NODE_OPTIONS reaches the spawned bridge). +async function fetchStub(dir) { + const log = resolve(dir, "fetch.jsonl"); + const stub = resolve(dir, "fetch-stub.mjs"); + await writeFile(stub, ` +import { appendFileSync } from "node:fs"; +globalThis.fetch = async (url, init) => { + appendFileSync(${JSON.stringify(log)}, JSON.stringify({ url: String(url), body: JSON.parse(init.body) }) + "\\n"); + const body = String(url).endsWith("/users/@me/channels") ? { id: "dm-channel" } : { id: "message-1" }; + return new Response(JSON.stringify(body), { status: 200 }); +}; +`); + return { log, options: `--import=${pathToFileURL(stub).href}` }; +} +const fetches = async (log) => (existsSync(log) ? (await readFile(log, "utf8")).trim().split("\n").map((line) => JSON.parse(line)) : []); + +test("discord-dm carries nonce and enforce_nonce when asked", async () => { + const dir = await mkdtemp(resolve(tmpdir(), "wn-bridge-")); + const stub = await fetchStub(dir); + const env = { ...process.env, NODE_OPTIONS: stub.options, BIFROST_DISCORD_BOT_TOKEN: "test-token-not-real" }; + const dm = (...extra) => spawnSync(process.execPath, [BRIDGE, "discord-dm", "--recipient-id", "1", "--content", "hi", "--cultmesh-command-id", "cmd", "--receipt-store", resolve(dir, "r.cc"), ...extra], { encoding: "utf8", env }); + + const with_nonce = dm("--nonce", "abc123"); + assert.equal(with_nonce.status, 0, with_nonce.stderr); + let posted = (await fetches(stub.log)).filter((entry) => entry.url.endsWith("/channels/dm-channel/messages")); + assert.equal(posted.at(-1).body.nonce, "abc123"); + assert.equal(posted.at(-1).body.enforce_nonce, true); + + const without = dm(); + assert.equal(without.status, 0, without.stderr); + posted = (await fetches(stub.log)).filter((entry) => entry.url.endsWith("/channels/dm-channel/messages")); + assert.ok(!("nonce" in posted.at(-1).body) && !("enforce_nonce" in posted.at(-1).body)); + + const before = (await fetches(stub.log)).length; + const tooLong = dm("--nonce", "x".repeat(26)); + assert.notEqual(tooLong.status, 0); + assert.equal((await fetches(stub.log)).length, before, "an over-long nonce posts nothing"); +}); + +test("the reader posts through the real bridge with its nonce", async () => { + const w = await world("wn-real-bridge-"); + const stub = await fetchStub(w.dir); + const result = w.run({ bridge: BRIDGE, env: { NODE_OPTIONS: stub.options, BIFROST_DISCORD_BOT_TOKEN: "test-token-not-real" } }); + assert.equal(result.status, 0, result.stderr); + const posted = (await fetches(stub.log)).filter((entry) => entry.url.endsWith("/messages")); + assert.deepEqual(posted.map((entry) => entry.body.content).sort(), [CLOSED_TEXT, OPENED_TEXT("closed-target", T_OPEN), OPENED_TEXT("open-target", T_CLOSE)].sort()); + for (const entry of posted) { + assert.equal(entry.body.nonce.length, 25); + assert.equal(entry.body.enforce_nonce, true); + } + assert.ok(existsSync(w.receipts), "the bridge wrote its crossing receipts to the reader's receipt store"); + assert.deepEqual((await w.status()).rows.map((row) => row.status), ["completed", "completed", "completed"]); +}); + diff --git a/tools/bifrost-bridge.mjs b/tools/bifrost-bridge.mjs index 5b4179b..1c4f0cd 100644 --- a/tools/bifrost-bridge.mjs +++ b/tools/bifrost-bridge.mjs @@ -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 } : {}), }), }); diff --git a/tools/watchdog-notice.mjs b/tools/watchdog-notice.mjs new file mode 100644 index 0000000..18fa6a7 --- /dev/null +++ b/tools/watchdog-notice.mjs @@ -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]; +}