diff --git a/docs/bridge.md b/docs/bridge.md index 0e5c56d..d971d0c 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -196,7 +196,7 @@ The command is a dry-run shape probe for public surfaces that are real enough to Bifrost's native Discord gateway is the conversation observer. A direct mention of the Bifrost bot, or a reply to one of its messages, becomes an immutable typed `bifrost.discord.persona_conversation_event.v0` only when the guild and channel have one exact Persona target binding. Bot-authored, ambient, empty, over-1200-character, unbound, and ambiguously bound messages do not enter the conversation path. The 1200-character cap is the shared signed alpha boundary with Epiphany. The observation store is Bifrost-private working state, never an actuator queue, and insert collisions must contain exactly equal content. VoidBot does not observe, schedule, or answer for the target Persona. -Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process per Persona that holds the Persona's Matrix access token (a systemd credential) and nobody else does. It reads the configured rooms by `/sync`, keeps each message from another sender as a delivery, and speaks on a `gamecult.persona.chat_send_request.v1` over a loopback CultNet RUDP endpoint, only while the instance brake is released. Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`. The Discord posting actuator, its permit and its RUDP listener are removed. +Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process for every Persona (`PERSONAS`, a list of slugs): it holds each Persona's Matrix access token as its own systemd credential (`matrix-token-`) and nobody else holds any. Per Persona it keeps one `/sync` loop over the configured rooms and the account's direct rooms (a DM invitation from an account on `MATRIX_SERVER_NAME` is accepted while that Persona's brake is released, and one from any other server is declined), one cursor and one delivery sequence. A delivery carries the sender's Matrix id, the event's `m.mentions` user ids and whether the room is a DM. Reading wakes nothing and acks nothing: the heartbeat polls with the non-consuming `gamecult.persona.chat_pending_request.v1`, and only `chat_ack` advances a position. The mouth speaks on a `gamecult.persona.chat_send_request.v1` and joins rooms only while that Persona's brake (`/etc/gamecult/minds//control.cc`) is released. One loopback CultNet RUDP endpoint serves all of them; a request naming an unlisted Persona gets a typed `chat_refusal` (`UnknownPersona`). Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`. The Discord posting actuator, its permit and its RUDP listener are removed. Bifrost admits an observation only when guild, channel, target runtime, canonical repository, Persona, producer, and producer runtime match a preconfigured `bifrost.persona_feedback.target_binding.v0` document in Bifrost's private provider store. That binding also owns source visibility, data classification, and the delivery-store target. Bifrost never infers public visibility from Discord identifiers. It classifies the author using its own `bifrost.discord.actor_link.v0` state: a match yields `linked_governance_feedback`; absence yields `unlinked_social_feedback`. Neither class grants adoption, work, release, or deployment authority. diff --git a/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack index 0bdbdfa..0e7c7a0 100644 Binary files a/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack and b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack differ diff --git a/tests/fixtures/persona-chat-v1/gamecult.persona.chat_refusal.v1.msgpack b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_refusal.v1.msgpack new file mode 100644 index 0000000..75727ec --- /dev/null +++ b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_refusal.v1.msgpack @@ -0,0 +1 @@ +’¥ghost®UnknownPersona \ No newline at end of file diff --git a/tests/persona-matrix-mouth-runtime.test.mjs b/tests/persona-matrix-mouth-runtime.test.mjs index abf755f..8a71588 100644 --- a/tests/persona-matrix-mouth-runtime.test.mjs +++ b/tests/persona-matrix-mouth-runtime.test.mjs @@ -4,11 +4,13 @@ import test from "node:test"; import assert from "node:assert/strict"; import dgram from "node:dgram"; +import { createServer } from "node:http"; +import { spawn } from "node:child_process"; import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { resolve } from "node:path"; import * as docs from "../tools/persona-chat-documents.mjs"; -import { cultcacheState, connectionIdFor, createBrakeReader, loadRuntime, startChatEndpoint } from "../tools/persona-matrix-mouth.mjs"; +import { cultcacheState, connectionIdFor, createBrakeReader, loadRuntime, pageOf, startChatEndpoint } from "../tools/persona-matrix-mouth.mjs"; const root = resolve(import.meta.dirname, ".."); const runtime = loadRuntime(resolve(process.env.CULTLIB_ROOT ?? resolve(root, "..", "CultLib"))); @@ -19,6 +21,7 @@ const delivery = { seq: 7, roomAlias: "#operator:gamecult.org", roomId: "!abc:gamecult.org", eventId: "$e1", sender: "@meta:gamecult.org", senderDisplay: "Meta", text: "hello eureka", atUnixMs: 1761700000000, context: [{ sender: "@eureka:gamecult.org", text: "earlier", atUnixMs: 1761699999000 }], + mentions: ["@eureka:gamecult.org", "@nova:gamecult.org"], isDirect: true, }; const samples = { [docs.kinds.pendingRequest]: { tuple: docs.pendingRequestTuple({ persona: "eureka", afterSeq: 3 }), parse: docs.parsePendingRequest }, @@ -26,6 +29,7 @@ const samples = { [docs.kinds.ack]: { tuple: docs.ackTuple({ persona: "eureka", throughSeq: 7 }), parse: docs.parseAck }, [docs.kinds.sendRequest]: { tuple: docs.sendRequestTuple({ requestId: "req-1", persona: "eureka", roomAlias: "#operator:gamecult.org", text: "hi" }), parse: docs.parseSendRequest }, [docs.kinds.sendReceipt]: { tuple: docs.sendReceiptTuple({ requestId: "req-1", eventId: "$sent" }), parse: docs.parseSendReceipt }, + [docs.kinds.refusal]: { tuple: docs.refusalTuple({ persona: "ghost", refusal: "UnknownPersona" }), parse: docs.parseRefusal }, }; for (const [kind, { tuple, parse }] of Object.entries(samples)) { @@ -44,9 +48,17 @@ test("the chat documents refuse malformed tuples", () => { assert.throws(() => docs.parsePendingRequest(["", 0])); assert.throws(() => docs.parseSendRequest(["r", "eureka", "#a:b", "x".repeat(4001)])); assert.throws(() => docs.parseSendRequest(["r", "eureka", "#a:b", 5])); + const wire = docs.deliveryTuple(delivery); + assert.equal(wire.length, 11); + assert.deepEqual(docs.parseDelivery(wire), delivery); + assert.throws(() => docs.parseDelivery(wire.slice(0, 9)), /11-field/, "the nine-field delivery is gone: one shape"); + assert.throws(() => docs.parseDelivery([...wire.slice(0, 10), "yes"]), /boolean/); + assert.throws(() => docs.parseDelivery([...wire.slice(0, 9), [""], true])); + assert.throws(() => docs.refusalTuple({ persona: "p", refusal: "Invented" })); assert.throws(() => docs.parseSendReceipt(["r", "$e", "Brake"]), /exactly one/); assert.throws(() => docs.parseSendReceipt(["r", null, null]), /exactly one/); assert.throws(() => docs.sendReceiptTuple({ requestId: "r", refusal: "Invented" })); + assert.throws(() => docs.parseSendReceipt(["r", null, "Invented"]), /unknown refusal/); assert.deepEqual(docs.parseSendReceipt(["r", null, "Brake"]), { requestId: "r", eventId: null, refusal: "Brake" }); }); @@ -84,7 +96,7 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a }; const key = "connection-key-for-test"; const rejected = []; - const server = await startChatEndpoint({ mouth, persona: "eureka", runtime, port: 0, connectionKey: key, runtimeId: "mouth-test", log: line => rejected.push(line) }); + const server = await startChatEndpoint({ mouths: new Map([["eureka", mouth]]), runtime, port: 0, connectionKey: key, runtimeId: "mouth-test", log: line => rejected.push(line) }); const client = dgram.createSocket("udp4"); await new Promise(ok => client.bind(0, "127.0.0.1", ok)); const connection = new runtime.cultnet.CultNetRudpSocketTransportConnection({ @@ -111,7 +123,7 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a assert.equal(document.schemaId, docs.kinds.pending); const pending = docs.parsePending(msgpack.decode(document.payload)); assert.deepEqual(pending.deliveries.map(d => d.eventId), ["$e1"]); - assert.deepEqual(pending.deliveries[0].context, delivery.context); + assert.deepEqual(pending.deliveries[0], delivery, "sender, mentions and the DM flag cross the wire exactly"); put(docs.kinds.ack, "eureka:7", docs.ackTuple({ persona: "eureka", throughSeq: 7 })); reply = nextReply(); @@ -119,16 +131,22 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a document = await reply; assert.equal(document.schemaId, docs.kinds.sendReceipt); assert.deepEqual(docs.parseSendReceipt(msgpack.decode(document.payload)), { requestId: "req-2", eventId: null, refusal: "Brake" }); + reply = nextReply(); put(docs.kinds.ack, "other:9", docs.ackTuple({ persona: "other", throughSeq: 9 })); + document = await reply; + assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload))], [docs.kinds.refusal, { persona: "other", refusal: "UnknownPersona" }], "an ack for an unlisted Persona is refused, typed, and not applied"); assert.deepEqual(calls.acks, [7]); reply = nextReply(); put(docs.kinds.pendingRequest, "other", docs.pendingRequestTuple({ persona: "other", afterSeq: 0 })); - put(docs.kinds.sendRequest, "req-3", docs.sendRequestTuple({ requestId: "req-3", persona: "eureka", roomAlias: "#operator:gamecult.org", text: "speak" })); document = await reply; - assert.equal(document.schemaId, docs.kinds.sendReceipt, "a request naming another Persona gets no answer; the next request is still served"); - assert.equal(docs.parseSendReceipt(msgpack.decode(document.payload)).eventId, "$sent"); - assert.deepEqual(calls.acks, [7], "an ack naming another Persona is not applied"); + assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload)).refusal], [docs.kinds.refusal, "UnknownPersona"], "unknown_persona_refused"); + + reply = nextReply(); + put(docs.kinds.sendRequest, "req-9", docs.sendRequestTuple({ requestId: "req-9", persona: "other", roomAlias: "#operator:gamecult.org", text: "speak" })); + document = await reply; + assert.deepEqual(docs.parseSendReceipt(msgpack.decode(document.payload)), { requestId: "req-9", eventId: null, refusal: "UnknownPersona" }); + assert.equal(calls.sends.length, 1, "nothing reached a mouth for the unlisted Persona"); } finally { server.close(); connection.close(); @@ -137,13 +155,83 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a }); test("the RUDP endpoint refuses to listen beyond loopback", async () => { - await assert.rejects(startChatEndpoint({ mouth: {}, persona: "eureka", runtime, port: 0, host: "0.0.0.0", connectionKey: "k", runtimeId: "m" }), /loopback/); + await assert.rejects(startChatEndpoint({ mouths: new Map(), runtime, port: 0, host: "0.0.0.0", connectionKey: "k", runtimeId: "m" }), /loopback/); }); test("loopback means exactly an IPv4 127.x.y.z address", async () => { - const refuses = host => assert.rejects(startChatEndpoint({ mouth: {}, persona: "eureka", runtime, port: 0, host, connectionKey: "k", runtimeId: "m" }), /loopback/); + const refuses = host => assert.rejects(startChatEndpoint({ mouths: new Map(), runtime, port: 0, host, connectionKey: "k", runtimeId: "m" }), /loopback/); for (const host of ["9127.0.0.1", "127.0.0.1x", "10.127.0.0.1", "127.0.0", ""]) await refuses(host); - const ok = await startChatEndpoint({ mouth: {}, persona: "eureka", runtime, port: 0, host: "127.10.20.30", connectionKey: "k", runtimeId: "m" }).catch(e => e); + const ok = await startChatEndpoint({ mouths: new Map(), runtime, port: 0, host: "127.10.20.30", connectionKey: "k", runtimeId: "m" }).catch(e => e); assert.ok(!/loopback/.test(ok?.message ?? ""), "a multi-digit loopback address is accepted by the check"); ok.close?.(); }); + +test("one process serves N Personas from systemd credentials: each token only in its own requests, nothing joined while braked, and no token on stderr", async () => { + const dir = await mkdtemp(resolve(tmpdir(), "mouth-main-")); + const creds = resolve(dir, "creds"), control = resolve(dir, "control"), stateDir = resolve(dir, "state"); + const tokens = { eureka: "syt_AAA_eureka_token", nova: "syt_BBB_nova_token" }; + await mkdir(creds, { recursive: true }); + for (const [persona, token] of Object.entries(tokens)) await writeFile(resolve(creds, `matrix-token-${persona}`), `${token}\n`); + await writeFile(resolve(creds, "chat-key"), "chat-key-credential\n"); + await mkdir(resolve(control, "eureka"), { recursive: true }); + await mkdir(resolve(control, "nova"), { recursive: true }); + await writeControl(resolve(control, "eureka", "control.cc"), [true, "2026-10-29T00:00:00Z", "op"]); + await writeControl(resolve(control, "nova", "control.cc"), [false, "2026-10-29T00:00:00Z", "op"]); // nova is braked + + const seen = []; + const joinedBy = new Map(); + const hs = createServer((req, res) => { + const auth = req.headers.authorization ?? ""; + const path = decodeURIComponent(new URL(req.url, "http://x").pathname).replace("/_matrix/client/v3", ""); + seen.push({ auth, method: req.method, path }); + const reply = body => { res.writeHead(200, { "content-type": "application/json" }); res.end(JSON.stringify(body)); }; + if (path === "/account/whoami") return reply({ user_id: auth.includes(tokens.eureka) ? "@eureka:t" : "@nova:t" }); + if (path === "/joined_rooms") return reply({ joined_rooms: [...(joinedBy.get(auth) ?? [])] }); + if (path.startsWith("/join/")) { joinedBy.set(auth, (joinedBy.get(auth) ?? new Set()).add(path.slice(6))); return reply({ room_id: path.slice(6) }); } + if (path.startsWith("/directory/room/")) return reply({ room_id: "!swarm:t" }); + if (path === "/sync") return reply({ next_batch: "b0" }); + res.writeHead(404); res.end("{}"); + }); + await new Promise(ok => hs.listen(0, "127.0.0.1", ok)); + const probe = dgram.createSocket("udp4"); + await new Promise(ok => probe.bind(0, "127.0.0.1", ok)); + const port = probe.address().port; + probe.close(); + const child = spawn(process.execPath, [resolve(root, "tools", "persona-matrix-mouth.mjs")], { + env: { + PATH: process.env.PATH, CULTLIB_ROOT: resolve(process.env.CULTLIB_ROOT ?? resolve(root, "..", "CultLib")), PERSONAS: "eureka,nova", CREDENTIALS_DIRECTORY: creds, CONTROL_ROOT: control, + STATE_DIRECTORY: stateDir, MATRIX_HOMESERVER: `http://127.0.0.1:${hs.address().port}`, MATRIX_ROOMS: "#swarm:t", MATRIX_SERVER_NAME: "t", CHAT_RUDP_LISTEN: `127.0.0.1:${port}`, + }, + }); + let stderr = ""; + child.stderr.on("data", c => (stderr += c)); + try { + for (let i = 0; i < 100 && !(seen.some(r => r.path === "/sync" && r.auth.includes(tokens.nova)) && seen.some(r => r.path === "/sync" && r.auth.includes(tokens.eureka))); i++) await new Promise(r => setTimeout(r, 50)); + const by = persona => seen.filter(r => r.auth === `Bearer ${tokens[persona]}`); + assert.ok(by("eureka").some(r => r.path === "/sync") && by("nova").some(r => r.path === "/sync"), `both accounts sync from one process; stderr: ${stderr}`); + assert.equal(by("eureka").length + by("nova").length, seen.length, "every request carries exactly one Persona's token"); + assert.deepEqual(by("eureka").filter(r => r.method === "POST").map(r => r.path), ["/join/!swarm:t"], "the released Persona joins"); + assert.deepEqual(by("nova").filter(r => r.method === "POST"), [], "the braked Persona joins nothing"); + } finally { + child.kill(); + await new Promise(ok => hs.close(ok)); + } + assert.ok(!Object.values(tokens).some(t => stderr.includes(t)) && !stderr.includes("chat-key-credential"), "no secret on stderr"); +}); + +test("a reply is paged to the byte budget, oldest first, and one delivery that will not encode is skipped and logged without silencing the rest", () => { + const mk = (n, body) => ({ ...delivery, seq: n, eventId: `$e${n}`, text: body }); + const logs = []; + const page = pageOf("eureka", [mk(1, "ok"), { ...mk(2, "bad"), sender: 123 }, mk(3, "fine")], msgpack, l => logs.push(l)); + assert.deepEqual(page[1].map(t => t[0]), [1, 3]); + assert.equal(logs.length, 1); + assert.ok(logs[0].includes("delivery 2 skipped")); + + const bodies = Array.from({ length: 200 }, (_, i) => mk(i + 1, "x".repeat(3900))); + const paged = pageOf("eureka", bodies, msgpack); + const size = msgpack.encode(paged).length; + assert.ok(size <= docs.limits.replyBytes + 1000, `page of ${size} bytes is over the budget`); + assert.ok(paged[1].length > 1 && paged[1].length < 200); + assert.deepEqual(paged[1].map(t => t[0]), paged[1].map((_, i) => i + 1), "the page is the oldest deliveries"); + assert.equal(pageOf("eureka", [mk(1, "x".repeat(4000))], msgpack)[1].length, 1, "a single delivery is always served"); +}); diff --git a/tests/persona-matrix-mouth.test.mjs b/tests/persona-matrix-mouth.test.mjs index 5f336eb..463429e 100644 --- a/tests/persona-matrix-mouth.test.mjs +++ b/tests/persona-matrix-mouth.test.mjs @@ -3,7 +3,7 @@ import assert from "node:assert/strict"; import { createServer } from "node:http"; import { readFileSync, readdirSync, statSync } from "node:fs"; import { resolve, join } from "node:path"; -import { createMouth, createBrakeReader, createRing, redact } from "../tools/persona-matrix-mouth.mjs"; +import { createMouth, createBrakeReader, guarded, redact } from "../tools/persona-matrix-mouth.mjs"; const TOKEN = "syt_SECRET_TOKEN_9f3"; const HS_NAME = "gamecult.test"; @@ -14,7 +14,7 @@ const message = (id, sender, body, ts) => ({ type: "m.room.message", event_id: i const tick = () => new Promise(r => setImmediate(r)); // A fake Matrix client-server API: just the endpoints the mouth uses. -async function fakeHomeserver({ sync, failWith } = {}) { +async function fakeHomeserver({ sync, failWith, users, member, direct } = {}) { const seen = []; const sent = new Map(); // txnId -> event_id const server = createServer((req, res) => { @@ -27,10 +27,15 @@ async function fakeHomeserver({ sync, failWith } = {}) { const reply = (status, body) => { res.writeHead(status, { "content-type": "application/json" }); res.end(JSON.stringify(body)); }; if (failWith) return reply(failWith, { errcode: "M_UNKNOWN_TOKEN", error: `bad token ${record.auth}` }); const p = record.path.replace("/_matrix/client/v3", ""); - if (p === "/account/whoami") return reply(200, { user_id: SELF }); + const bearer = (record.auth ?? "").replace("Bearer ", ""); + if (p === "/account/whoami") return reply(200, { user_id: users?.[bearer] ?? SELF }); + if (p === "/joined_rooms") return reply(200, { joined_rooms: member ?? Object.values(ROOMS) }); + if (req.method === "PUT" && p.includes("/account_data/m.direct")) return reply(200, {}); + if (p.endsWith("/leave")) return reply(200, {}); + if (p.includes("/account_data/m.direct")) return direct ? reply(200, direct) : reply(404, { errcode: "M_NOT_FOUND" }); if (p.startsWith("/directory/room/")) return reply(200, { room_id: ROOMS[p.slice("/directory/room/".length)] }); if (p.startsWith("/join/")) return reply(200, { room_id: p.slice(6) }); - if (p === "/sync") return reply(200, sync(record.query.get("since"))); + if (p === "/sync") return reply(200, sync(record.query.get("since"), bearer)); const m = p.match(/^\/rooms\/(.+)\/send\/m\.room\.message\/(.+)$/); if (m && req.method === "PUT") { if (!sent.has(m[2])) sent.set(m[2], `$event${sent.size + 1}`); @@ -57,7 +62,7 @@ const joined = events => ({ rooms: { join: events } }); function mouthFor(hs, overrides = {}) { return createMouth({ persona: "eureka", homeserver: hs.url, rooms: Object.keys(ROOMS), token: TOKEN, state: memoryState(), - brake: async () => true, ring: async () => {}, log: () => {}, syncTimeoutMs: 1, ...overrides, + serverName: HS_NAME, brake: async () => true, log: () => {}, syncTimeoutMs: 1, ...overrides, }); } @@ -229,59 +234,6 @@ test("brake reader: only a decoded released brake counts as released", async () assert.equal(await readerFor(store({ type: "eureka.control.brake.v1", key: "brake", payload: Buffer.from("not json") }))(), false, "undecodable bytes"); }); -test("ring_starts_persona_unit_once: two deliveries in one idle window cause one start; a drained mailbox rings again", async () => { - const hs = await fakeHomeserver({ sync: script }); - try { - let rings = 0; - const mouth = mouthFor(hs, { ring: async () => { rings++; } }); - await mouth.syncOnce(); // baseline - await mouth.syncOnce(); // delivery 1 rings - await mouth.syncOnce(); // delivery 2, same window, still unacked - await tick(); - assert.equal(rings, 1); - } finally { await hs.close(); } - - const every = await fakeHomeserver({ sync: since => ({ next_batch: `n${since}`, ...joined({ "!swarm:gamecult.test": { timeline: { events: [message(`$e${since}`, `@alice:${HS_NAME}`, "again", 5)] } } }) }) }); - try { - let clock = 0, rings = 0; - const mouth = mouthFor(every, { ring: async () => { rings++; }, now: () => clock }); - await mouth.syncOnce(); // baseline - await mouth.syncOnce(); await tick(); - assert.equal(rings, 1); - await mouth.syncOnce(); await tick(); - assert.equal(rings, 1, "inside the window and unacked: no second start"); - await mouth.ack(2); - await mouth.syncOnce(); await tick(); - assert.equal(rings, 2, "after a drain a new delivery rings again"); - clock = 6 * 60 * 1000; - await mouth.syncOnce(); await tick(); - assert.equal(rings, 3, "a delivery left unanswered past the window rings again"); - } finally { await every.close(); } -}); - -test("a failed ring is logged once and never retried", async () => { - const logs = []; - let calls = 0; - const hs = await fakeHomeserver({ sync: script }); - try { - const mouth = mouthFor(hs, { ring: async () => { calls++; throw new Error("systemctl refused"); }, log: l => logs.push(l) }); - await mouth.syncOnce(); await mouth.syncOnce(); await tick(); await tick(); - assert.equal(calls, 1); - assert.equal(logs.filter(l => l.includes("ring failed")).length, 1); - } finally { await hs.close(); } -}); - -test("ring runs exactly systemctl start --no-block on the instance's unit, with no token in the child's environment", async () => { - const calls = []; - await createRing({ instance: "eureka", run: (cmd, args, options, cb) => { calls.push({ cmd, args, options }); cb(null); } })(); - assert.equal(calls.length, 1); - assert.deepEqual([calls[0].cmd, calls[0].args], ["systemctl", ["start", "--no-block", "mind-persona@eureka.service"]]); - assert.deepEqual(Object.keys(calls[0].options.env), ["PATH"]); - const failed = []; - await createRing({ instance: "eureka", run: (c, a, o, cb) => cb(new Error("denied")), log: l => failed.push(l) })(); - assert.deepEqual(failed, ["ring failed: denied"]); -}); - test("the Matrix token goes only in the Authorization header: never in logs, errors, deliveries or receipts", async () => { const logs = []; const bad = await fakeHomeserver({ sync: script, failWith: 401 }); @@ -316,6 +268,7 @@ test("the Matrix token goes only in the Authorization header: never in logs, err assert.ok(logs.length > 0, "failures were logged"); assert.ok(logs.every(l => !l.includes(TOKEN)), `a log line carried the token: ${logs.find(l => l.includes(TOKEN))}`); assert.equal(redact(`x ${TOKEN} y`, TOKEN), "x [redacted] y"); + assert.equal(redact("a1 b2 a1", ["a1", "b2"]), "[redacted] [redacted] [redacted]"); }); test("a homeserver error is reported with its status and none of its body", async () => { @@ -345,3 +298,282 @@ test("the Discord posting actuator is gone and the Matrix path names no Discord" assert.throws(() => statSync(resolve(root, "tools", gone))); } }); + +// ---- one mouth, many Personas ---------------------------------------------------- + +const NOVA = `@nova:${HS_NAME}`; +const TOKEN_NOVA = "syt_OTHER_TOKEN_77x"; + +test("two_personas_two_queues: one message in a room both joined yields one delivery per Persona with separate seqs and only its own token", async () => { + const both = since => since === null + ? { next_batch: "b0", ...joined({ "!swarm:gamecult.test": { timeline: { events: [] } } }) } + : { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { timeline: { events: [message("$m1", `@alice:${HS_NAME}`, "hello both", 1000)] } } }) }; + const hs = await fakeHomeserver({ sync: both, users: { [TOKEN]: SELF, [TOKEN_NOVA]: NOVA } }); + try { + const eureka = mouthFor(hs); + const nova = mouthFor(hs, { persona: "nova", token: TOKEN_NOVA }); + await eureka.syncOnce(); await nova.syncOnce(); + await eureka.syncOnce(); + await eureka.syncOnce(); // a second delivery for eureka only + await nova.syncOnce(); + assert.deepEqual(eureka.pendingAfter(0).map(d => d.seq), [1, 2]); + assert.deepEqual(nova.pendingAfter(0).map(d => d.seq), [1], "separate queue, separate seq"); + await eureka.ack(2); + assert.equal(eureka.pendingCount, 0); + assert.equal(nova.pendingCount, 1, "one Persona's ack leaves the other's mail alone"); + for (const r of hs.seen) assert.ok([`Bearer ${TOKEN}`, `Bearer ${TOKEN_NOVA}`].includes(r.auth)); + const novaCalls = hs.seen.filter(r => r.auth === `Bearer ${TOKEN_NOVA}`); + assert.ok(novaCalls.length >= 3 && novaCalls.every(r => !JSON.stringify(r).includes(TOKEN)), "a token never appears in another account's request"); + } finally { await hs.close(); } +}); + +test("a Persona's own words are not mail for it, but are for the other Persona", async () => { + const hs = await fakeHomeserver({ + users: { [TOKEN]: SELF, [TOKEN_NOVA]: NOVA }, + sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { timeline: { events: [message("$e", SELF, "eureka speaking", 5)] } } }) }, + }); + try { + const eureka = mouthFor(hs); + const nova = mouthFor(hs, { persona: "nova", token: TOKEN_NOVA }); + for (const m of [eureka, nova]) { await m.syncOnce(); await m.syncOnce(); } + assert.equal(eureka.pendingCount, 0); + assert.deepEqual(nova.pendingAfter(0).map(d => d.sender), [SELF]); + } finally { await hs.close(); } +}); + +test("mentions_carried: m.mentions user ids and the exact sender id are on the delivery; absent or malformed mentions are empty", async () => { + const withMentions = (id, mentions) => ({ ...message(id, `@meta:${HS_NAME}`, "hi", 7), content: { msgtype: "m.text", body: "hi", ...(mentions === undefined ? {} : { "m.mentions": mentions }) } }); + const hs = await fakeHomeserver({ + sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { timeline: { events: [ + withMentions("$a", { user_ids: [SELF, NOVA] }), withMentions("$b", undefined), withMentions("$c", { user_ids: "nope" }), withMentions("$d", { room: true }), + ] } } }) }, + }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); + const made = await mouth.syncOnce(); + assert.deepEqual(made.map(d => d.mentions), [[SELF, NOVA], [], [], []]); + assert.ok(made.every(d => d.sender === `@meta:${HS_NAME}`)); + assert.ok(made.every(d => d.isDirect === false)); + } finally { await hs.close(); } +}); + +test("dm_flag: a joined room the account marks in m.direct is delivered as a DM, answerable by its room id; other rooms are not", async () => { + const DM = "!dm:gamecult.test"; + const hs = await fakeHomeserver({ + direct: { [`@meta:${HS_NAME}`]: [DM] }, + sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", account_data: { events: [{ type: "m.direct", content: { [`@meta:${HS_NAME}`]: [DM, "!dm2:gamecult.test"] } }] }, ...joined({ + [DM]: { timeline: { events: [message("$dm", `@meta:${HS_NAME}`, "private", 9)] } }, + "!dm2:gamecult.test": { timeline: { events: [message("$dm2", `@meta:${HS_NAME}`, "second", 10)] } }, + "!swarm:gamecult.test": { timeline: { events: [message("$pub", `@meta:${HS_NAME}`, "public", 11)] } }, + }) }, + }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); + const made = await mouth.syncOnce(); + assert.deepEqual(made.map(d => [d.eventId, d.roomAlias, d.isDirect]), [["$dm", DM, true], ["$dm2", "!dm2:gamecult.test", true], ["$pub", `#swarm:${HS_NAME}`, false]]); + assert.equal((await mouth.send({ requestId: "dm-1", persona: "eureka", roomAlias: DM, text: "reply" })).refusal, null); + assert.equal(hs.puts().at(-1).path, `/_matrix/client/v3/rooms/${DM}/send/m.room.message/dm-1`); + } finally { await hs.close(); } +}); + +test("pending_is_not_ack: repeated reads return the same deliveries until an ack", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); await mouth.syncOnce(); + const first = mouth.pendingAfter(0), second = mouth.pendingAfter(0); + assert.deepEqual(first, second); + assert.equal(first.length, 1); + await mouth.ack(first[0].seq); + assert.deepEqual(mouth.pendingAfter(0), []); + } finally { await hs.close(); } +}); + +test("join_while_braked: no join and no send while the brake is held; joining happens on the round after release; a failing join is logged, redacted, and never throws", async () => { + let open = false; + const logs = []; + const hs = await fakeHomeserver({ sync: script, member: [] }); + try { + const mouth = mouthFor(hs, { brake: async () => open, log: l => logs.push(l) }); + await mouth.syncOnce(); + await mouth.send({ requestId: "r", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "speak" }); + assert.equal(hs.seen.filter(r => r.method === "POST" || r.method === "PUT").length, 0, "braked: nothing is joined or sent"); + open = true; + await mouth.syncOnce(); + assert.deepEqual(hs.seen.filter(r => r.method === "POST").map(r => r.path.split("/join/")[1]).sort(), ["!operator:gamecult.test", "!swarm:gamecult.test"]); + await mouth.syncOnce(); + assert.equal(hs.seen.filter(r => r.method === "POST").length, 2, "joined rooms are not joined again"); + } finally { await hs.close(); } + + const failing = await fakeHomeserver({ sync: script, member: [] }); + try { + const mouth = mouthFor(failing, { fetch: (url, init) => String(url).includes("/join/") ? Promise.reject(new Error(`join refused for ${init.headers.authorization}`)) : fetch(url, init), log: l => logs.push(l) }); + await mouth.syncOnce(); // no throw: the unit does not crash-loop on a failed join + assert.ok(logs.some(l => l.includes("not joined")) && logs.every(l => !l.includes(TOKEN))); + } finally { await failing.close(); } +}); + +test("a freshly joined room's history is context, not mail; an already joined room's backlog still is mail after a restart", async () => { + const hs = await fakeHomeserver({ + member: ["!operator:gamecult.test"], + sync: since => since === "b0" + ? { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { timeline: { events: [message("$hist", `@alice:${HS_NAME}`, "before I joined", 1)] } }, "!operator:gamecult.test": { timeline: { events: [message("$mail", `@alice:${HS_NAME}`, "waiting for me", 2)] } } }) } + : { next_batch: "b0" }, + }); + try { + const mouth = mouthFor(hs, { state: memoryState({ nextBatch: "b0", seq: 4 }) }); + const made = await mouth.syncOnce(); + assert.deepEqual(made.map(d => d.eventId), ["$mail"]); + assert.equal(made[0].seq, 5); + } finally { await hs.close(); } +}); + +test("pending is bounded: at the bound the mouth stops reading, keeps its position, and resumes after an ack", async () => { + const hs = await fakeHomeserver({ sync: since => ({ next_batch: `n${since}`, ...joined({ "!swarm:gamecult.test": { timeline: { events: [message(`$e${since}`, `@alice:${HS_NAME}`, "again", 5)] } } }) }) }); + try { + const mouth = mouthFor(hs, { maxPending: 2 }); + await mouth.syncOnce(); // baseline + await mouth.syncOnce(); await mouth.syncOnce(); + const syncs = () => hs.seen.filter(r => r.path.endsWith("/sync")).length; + const before = syncs(); + assert.deepEqual(await mouth.syncOnce(), []); + assert.equal(syncs(), before, "no /sync while the mailbox is at the bound"); + assert.equal(mouth.pendingCount, 2); + await mouth.ack(2); + assert.equal((await mouth.syncOnce()).length, 1); + } finally { await hs.close(); } +}); + +test("guarded: a failure leaves with every secret redacted, however many tokens the process holds", async () => { + await assert.rejects(guarded([TOKEN, TOKEN_NOVA], async () => { throw new Error(`boom ${TOKEN} then ${TOKEN_NOVA}`); }), err => err.message === "boom [redacted] then [redacted]"); + assert.equal(await guarded([TOKEN], async () => 5), 5); +}); + +test("the ring is gone: the mouth names no unit manager and spawns no child", () => { + const source = readFileSync(resolve(import.meta.dirname, "..", "tools", "persona-matrix-mouth.mjs"), "utf8"); + assert.ok(!/systemctl|child_process|execFile|mind-persona@/.test(source)); +}); + +test("delivery_poison_pill: a non-string display name or malformed event fields never reach a delivery; the sender falls back to its localpart", async () => { + const hs = await fakeHomeserver({ + sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { + state: { events: [{ type: "m.room.member", state_key: `@alice:${HS_NAME}`, content: { displayname: 123 } }] }, + timeline: { events: [ + message("$a", `@alice:${HS_NAME}`, "numeric name", "not-a-number"), + { type: "m.room.message", event_id: "$b", sender: 5, content: { msgtype: "m.text", body: "numeric sender" } }, + { type: "m.room.message", sender: `@alice:${HS_NAME}`, content: { msgtype: "m.text", body: "no event id" } }, + ] }, + } }) }, + }); + try { + const mouth = mouthFor(hs, { now: () => 42 }); + await mouth.syncOnce(); + const made = await mouth.syncOnce(); + assert.deepEqual(made.map(d => [d.eventId, d.senderDisplay, d.atUnixMs]), [["$a", "alice", 42]]); + } finally { await hs.close(); } +}); + +test("delivery size: body, context lines and mentions are bounded per delivery, on a character boundary", async () => { + const big = "é".repeat(30000); + const hs = await fakeHomeserver({ + sync: since => since === null ? { next_batch: "b0", ...joined({ "!swarm:gamecult.test": { timeline: { events: [message("$h", `@alice:${HS_NAME}`, big, 1)] } } }) } + : { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { timeline: { events: [{ ...message("$n", `@bob:${HS_NAME}`, big, 2), content: { msgtype: "m.text", body: big, "m.mentions": { user_ids: Array.from({ length: 500 }, (_, i) => `@u${i}:x`) } } }] } } }) }, + }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); + const [d] = await mouth.syncOnce(); + assert.ok(Buffer.byteLength(d.text) <= 4000 && d.text === "é".repeat(2000)); + assert.ok(d.context.every(c => Buffer.byteLength(c.text) <= 1000)); + assert.equal(d.mentions.length, 64); + } finally { await hs.close(); } +}); + +test("bound_is_a_hard_cap: a round that would pass the cap enqueues only what fits, keeps its position, and the rest arrives once after an ack", async () => { + const four = ["$w", "$x", "$y", "$z"].map((id, i) => message(id, `@alice:${HS_NAME}`, id, 10 + i)); + const state = memoryState(); + const hs = await fakeHomeserver({ + sync: since => since === null ? { next_batch: "b0" } : since === "b0" ? { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { timeline: { events: four } } }) } : { next_batch: since }, + }); + try { + const mouth = mouthFor(hs, { maxPending: 3, state }); + await mouth.syncOnce(); + assert.deepEqual((await mouth.syncOnce()).map(d => d.eventId), ["$w", "$x", "$y"]); + assert.equal(mouth.pendingCount, 3, "never above the cap"); + assert.deepEqual(await mouth.syncOnce(), [], "full: no read"); + await mouth.ack(2); + const rest = await mouth.syncOnce(); + assert.deepEqual(rest.map(d => d.eventId), ["$z"], "the four-message round is read again; the three already taken are not delivered twice"); + assert.deepEqual(mouth.pendingAfter(0).map(d => d.eventId), ["$y", "$z"]); + assert.equal(state.box.value.nextBatch, "b0", "the position moves only once the whole round is acked"); + await mouth.ack(4); + assert.equal(state.box.value.nextBatch, "b1"); + } finally { await hs.close(); } +}); + +// ---- DM invitations ---------------------------------------------------------------- + +const DMROOM = "!newdm:gamecult.test"; +const invite = (inviter, isDirect = true) => ({ [DMROOM]: { invite_state: { events: [{ type: "m.room.member", state_key: SELF, sender: inviter, content: { membership: "invite", is_direct: isDirect } }] } } }); +const inviteSync = (inviter, isDirect) => since => since === null ? { next_batch: "b0" } + : since === "b0" ? { next_batch: "b1", rooms: { invite: invite(inviter, isDirect) } } + : since === "b1" ? { next_batch: "b2", ...joined({ [DMROOM]: { timeline: { events: [message("$first", `@meta:${HS_NAME}`, "hello in private", 3)] } } }) } + : { next_batch: since }; +const joinsOf = hs => hs.seen.filter(r => r.method === "POST" && r.path.includes("/join/")).map(r => r.path.split("/join/")[1]); +const leavesOf = hs => hs.seen.filter(r => r.method === "POST" && r.path.endsWith("/leave")); + +test("dm_invite_accepted: a same-server DM invitation is joined, remembered in m.direct, and its messages arrive as is_direct deliveries", async () => { + const hs = await fakeHomeserver({ sync: inviteSync(`@meta:${HS_NAME}`) }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual(joinsOf(hs), [DMROOM]); + const saved = hs.puts().find(r => r.path.endsWith("/account_data/m.direct")); + assert.deepEqual(saved.body, { [`@meta:${HS_NAME}`]: [DMROOM] }); + const [d] = await mouth.syncOnce(); + assert.deepEqual([d.eventId, d.roomAlias, d.isDirect, d.sender], ["$first", DMROOM, true, `@meta:${HS_NAME}`], "the message sent before the join is mail, not history"); + assert.equal((await mouth.send({ requestId: "dm", persona: "eureka", roomAlias: DMROOM, text: "hi" })).refusal, null); + } finally { await hs.close(); } +}); + +test("dm_invite_waits_for_release: an invitation received while braked is not joined until the brake is released", async () => { + let open = false; + const hs = await fakeHomeserver({ sync: inviteSync(`@meta:${HS_NAME}`) }); + try { + const mouth = mouthFor(hs, { brake: async () => open }); + await mouth.syncOnce(); await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual(hs.seen.filter(r => r.method === "POST" || r.method === "PUT"), [], "braked: no join, no leave, nothing written"); + open = true; + await mouth.syncOnce(); + assert.deepEqual(joinsOf(hs), [DMROOM]); + } finally { await hs.close(); } +}); + +test("dm_invite_other_server_declined: an invitation from another server is declined, never joined, brake or no brake", async () => { + const hs = await fakeHomeserver({ sync: inviteSync("@mallory:evil.example") }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual(joinsOf(hs), []); + assert.deepEqual(leavesOf(hs).map(r => r.path), [`/_matrix/client/v3/rooms/${DMROOM}/leave`]); + await mouth.syncOnce(); + assert.equal(leavesOf(hs).length, 1, "answered once"); + } finally { await hs.close(); } + const lookalike = await fakeHomeserver({ sync: inviteSync(`@mallory:${HS_NAME}.evil.example`) }); + try { + const mouth = mouthFor(lookalike); + await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual(joinsOf(lookalike), [], "the server name must match exactly"); + } finally { await lookalike.close(); } +}); + +test("a same-server invitation to a room that is not a DM is neither joined nor declined", async () => { + const hs = await fakeHomeserver({ sync: inviteSync(`@meta:${HS_NAME}`, false) }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual([joinsOf(hs), leavesOf(hs)], [[], []]); + } finally { await hs.close(); } +}); diff --git a/tools/persona-chat-documents.mjs b/tools/persona-chat-documents.mjs index 3778379..5516f7a 100644 --- a/tools/persona-chat-documents.mjs +++ b/tools/persona-chat-documents.mjs @@ -9,13 +9,24 @@ export const kinds = Object.freeze({ ack: "gamecult.persona.chat_ack.v1", sendRequest: "gamecult.persona.chat_send_request.v1", sendReceipt: "gamecult.persona.chat_send_receipt.v1", + refusal: "gamecult.persona.chat_refusal.v1", }); -export const refusals = Object.freeze(["Brake", "PersonaMismatch", "UnknownRoom", "EmptyText", "SendFailed"]); +export const refusals = Object.freeze(["Brake", "PersonaMismatch", "UnknownRoom", "EmptyText", "SendFailed", "UnknownPersona"]); const maxTextBytes = 4000; const maxContext = 20; +// Per-delivery bounds, applied where the mouth reads Matrix: the body is bounded like a send, +// a context line and a display name are smaller, and a reply is paged to a byte budget. +export const limits = Object.freeze({ textBytes: maxTextBytes, contextBytes: 1000, displayBytes: 256, mentions: 64, replyBytes: 256 * 1024 }); + +// `value` cut to at most `maxBytes` of UTF-8 without splitting a character. +export function clip(value, maxBytes) { + if (Buffer.byteLength(value, "utf8") <= maxBytes) return value; + return Buffer.from(value, "utf8").subarray(0, maxBytes).toString("utf8").replace(/�$/, ""); +} + export function pendingRequestTuple({ persona, afterSeq }) { return [text(persona, "persona"), seq(afterSeq, "after_seq")]; } @@ -50,23 +61,38 @@ export function sendReceiptTuple({ requestId, eventId = null, refusal = null }) export function parseSendReceipt(t) { const [requestId, eventId, refusal] = tuple(t, 3, "chat_send_receipt"); if ((eventId === null) === (refusal === null)) throw new Error("a send receipt carries exactly one of event_id and refusal"); + if (refusal !== null && !refusals.includes(refusal)) throw new Error(`unknown refusal ${String(refusal)}`); return { requestId: text(requestId, "request_id"), eventId: eventId === null ? null : text(eventId, "event_id"), refusal }; } +// A typed refusal for a request that names a Persona the mouth does not serve. +export function refusalTuple({ persona, refusal }) { + if (!refusals.includes(refusal)) throw new Error(`unknown refusal ${refusal}`); + return [text(persona, "persona"), refusal]; +} +export function parseRefusal(t) { + const [persona, refusal] = tuple(t, 2, "chat_refusal"); + if (!refusals.includes(refusal)) throw new Error(`unknown refusal ${String(refusal)}`); + return { persona: text(persona, "persona"), refusal }; +} + export function deliveryTuple(d) { return [ seq(d.seq, "seq"), text(d.roomAlias, "room_alias"), text(d.roomId, "room_id"), text(d.eventId, "event_id"), text(d.sender, "sender"), text(d.senderDisplay, "sender_display"), String(d.text), seq(d.atUnixMs, "at_unix_ms"), d.context.slice(-maxContext).map(c => [text(c.sender, "sender"), String(c.text), seq(c.atUnixMs, "at_unix_ms")]), + d.mentions.map(id => text(id, "mention")), flag(d.isDirect, "is_direct"), ]; } export function parseDelivery(t) { - const [s, roomAlias, roomId, eventId, sender, senderDisplay, body, atUnixMs, context] = tuple(t, 9, "delivery"); + const [s, roomAlias, roomId, eventId, sender, senderDisplay, body, atUnixMs, context, mentions, isDirect] = tuple(t, 11, "delivery"); if (!Array.isArray(context)) throw new Error("delivery context must be a list"); + if (!Array.isArray(mentions)) throw new Error("delivery mentions must be a list"); return { seq: seq(s, "seq"), roomAlias: text(roomAlias, "room_alias"), roomId: text(roomId, "room_id"), eventId: text(eventId, "event_id"), sender: text(sender, "sender"), senderDisplay: text(senderDisplay, "sender_display"), text: String(body), atUnixMs: seq(atUnixMs, "at_unix_ms"), context: context.map(c => { const [cs, ct, ca] = tuple(c, 3, "context"); return { sender: text(cs, "sender"), text: String(ct), atUnixMs: seq(ca, "at_unix_ms") }; }), + mentions: mentions.map(id => text(id, "mention")), isDirect: flag(isDirect, "is_direct"), }; } @@ -108,6 +134,10 @@ function text(v, what) { if (typeof v !== "string" || v === "") throw new Error(`${what} must be a non-empty string`); return v; } +function flag(v, what) { + if (typeof v !== "boolean") throw new Error(`${what} must be a boolean`); + return v; +} function seq(v, what) { if (!Number.isSafeInteger(v) || v < 0) throw new Error(`${what} must be a non-negative integer`); return v; diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 22429cb..ca50f36 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -1,21 +1,23 @@ -// The Matrix ear and mouth of one Persona. One process per Persona. +// The Matrix ear and mouth of every Persona. One process: one account (token) and +// one /sync loop per Persona, one CultNet endpoint for all of them. // -// Ear: /sync the configured rooms; every message from another sender becomes a -// delivery kept in memory. The /sync position (next_batch) is persisted only when -// an ack covers it, so a restart before the ack re-delivers instead of losing. +// Ear: /sync the configured rooms (and the account's direct rooms); every message +// from another sender becomes a delivery kept in memory, carrying its sender id, +// its m.mentions user ids and whether the room is a DM. The /sync position +// (next_batch) is persisted only when an ack covers it, so a restart before the +// ack re-delivers instead of losing. Reading wakes nothing: the heartbeat polls +// with a non-consuming chat_pending_request and decides who runs. // Mouth: a send request becomes one PUT with txnId = request_id (a retry is -// idempotent) and only while the instance brake is released. -// Ring: the first delivery of an idle window starts the Persona's one-shot unit. -// Starting is a request; the unit's own gate decides. +// idempotent), and a room is joined, only while that Persona's brake is released. // -// The Matrix access token is read here, from a systemd credential, and nowhere -// else. It is never put in a document, an environment passed to a child, or a log. +// Each Matrix access token is read here, from its own systemd credential, and +// nowhere else. It is never put in a document, an environment passed to a child, +// or a log, and it is only ever presented in its own Persona's Authorization header. // -// Everything outside this process is a port (fetch, state, brake, ring, clock), so -// the core runs against a fake homeserver with no CultLib and no daemon. -import { execFile } from "node:child_process"; +// Everything outside this process is a port (fetch, state, brake, clock), so the +// core runs against a fake homeserver with no CultLib and no daemon. import { createHash } from "node:crypto"; -import { readFileSync } from "node:fs"; +import { mkdirSync, readFileSync } from "node:fs"; import { createRequire } from "node:module"; import { createSocket } from "node:dgram"; import { resolve, join } from "node:path"; @@ -25,16 +27,20 @@ import * as docs from "./persona-chat-documents.mjs"; const API = "/_matrix/client/v3"; const BRAKE_TYPE = "eureka.control.brake.v1"; const CONTEXT = 20; -const RING_WINDOW_MS = 5 * 60 * 1000; +const MAX_PENDING = 500; +const SLUG = /^[A-Za-z0-9_-]{1,64}$/; export function createMouth({ - persona, homeserver, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake, ring, - now = Date.now, log = () => {}, syncTimeoutMs = 30000, + persona, homeserver, serverName, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake, + now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, }) { if (!persona || !homeserver || !token || !rooms?.length) throw new Error("persona, homeserver, rooms and token are required"); const say = message => log(redact(String(message), token)); const roomIdByAlias = new Map(); const aliasByRoomId = new Map(); + const joinedAliases = new Set(); + const freshRooms = new Set(); // joined by this process since the last sync: their history is context, not mail + const directRooms = new Set(); // room ids the account marks as DMs (m.direct) const history = new Map(); // room_id -> last CONTEXT messages, oldest first const displayNames = new Map(); let selfId = null; @@ -43,8 +49,10 @@ export function createMouth({ let pending = []; // deliveries not yet acked let batches = []; // { token, maxSeq }: positions awaiting an ack let committed = null; // the position that is persisted - let lastRingAt = -Infinity; let started = false; + const directContent = {}; // the account's m.direct content: user id -> room ids + const invites = new Map(); // room id -> { inviter, direct }: invitations not yet answered + const processed = new Set(); // event ids read since the position last advanced async function matrix(method, path, body) { let response; @@ -61,58 +69,139 @@ export function createMouth({ return response.json(); } - async function start() { - const saved = await state.load(); - cursor = committed = saved?.nextBatch ?? null; - seqCounter = saved?.seq ?? 0; - selfId = (await matrix("GET", "/account/whoami")).user_id; - for (const alias of rooms) { - const { room_id: roomId } = await matrix("GET", `/directory/room/${encodeURIComponent(alias)}`); - await matrix("POST", `/join/${encodeURIComponent(roomId)}`, {}); - roomIdByAlias.set(alias, roomId); - aliasByRoomId.set(roomId, alias); + const released = async () => { try { return (await brake()) === true; } catch { return false; } }; + const noteDirect = content => { + for (const [user, ids] of Object.entries(content ?? {})) { + if (!Array.isArray(ids)) continue; + directContent[user] = ids.filter(id => typeof id === "string"); + directContent[user].forEach(id => directRooms.add(id)); } - started = true; + }; + + function noteInvites(invited) { + for (const [roomId, room] of Object.entries(invited ?? {})) { + const event = (room.invite_state?.events ?? []).find(e => e.type === "m.room.member" && e.state_key === selfId && typeof e.sender === "string"); + if (event) invites.set(roomId, { inviter: event.sender, direct: event.content?.is_direct === true }); + } + } + + // A DM invitation from an account on our own homeserver is accepted, and only while the brake is + // released; one received while braked waits for the release. An invitation from any other server is + // declined. A same-server invitation to a room that is not a DM is not ours to accept and is dropped. + async function answerInvites() { + let free = null; + for (const [roomId, invite] of [...invites]) { + try { + const room = encodeURIComponent(roomId); + if (!serverName || invite.inviter.slice(invite.inviter.indexOf(":") + 1) !== serverName) { + await matrix("POST", `/rooms/${room}/leave`, {}); + invites.delete(roomId); + } else if (!invite.direct) { + invites.delete(roomId); + } else { + free ??= await released(); + if (!free) continue; + await matrix("POST", `/join/${room}`, {}); + invites.delete(roomId); + directRooms.add(roomId); + directContent[invite.inviter] = [...new Set([...(directContent[invite.inviter] ?? []), roomId])]; + await matrix("PUT", `/user/${encodeURIComponent(selfId)}/account_data/m.direct`, directContent).catch(error => say(`m.direct not saved: ${error.message}`)); + } + } catch (error) { + say(`invite to ${roomId} not answered: ${error.message}`); + } + } + } + const unjoined = () => joinedAliases.size < rooms.length; + + // Join the configured rooms the account is not in, only while the brake is released. + // One room failing is logged and retried on the next round; it never stops the others. + async function ensureRooms() { + const member = new Set((await matrix("GET", "/joined_rooms")).joined_rooms ?? []); + let free = null; + for (const alias of rooms.filter(a => !joinedAliases.has(a))) { + try { + let roomId = roomIdByAlias.get(alias); + if (!roomId) ({ room_id: roomId } = await matrix("GET", `/directory/room/${encodeURIComponent(alias)}`)); + roomIdByAlias.set(alias, roomId); + if (!member.has(roomId)) { + free ??= await released(); + if (!free) continue; + await matrix("POST", `/join/${encodeURIComponent(roomId)}`, {}); + freshRooms.add(roomId); + } + aliasByRoomId.set(roomId, alias); + joinedAliases.add(alias); + } catch (error) { + say(`room ${alias} not joined: ${error.message}`); + } + } + } + + async function start() { + if (!started) { + const saved = await state.load(); + cursor = committed = saved?.nextBatch ?? null; + seqCounter = saved?.seq ?? 0; + selfId = (await matrix("GET", "/account/whoami")).user_id; + try { noteDirect(await matrix("GET", `/user/${encodeURIComponent(selfId)}/account_data/m.direct`)); } catch { /* no m.direct: no DMs yet */ } + started = true; + } + if (unjoined()) await ensureRooms(); } // One /sync round. Returns the deliveries it made. async function syncOnce() { - if (!started) await start(); + if (!started || unjoined()) await start(); + if (pending.length >= maxPending) return []; // unanswered mail is not outrun: the position stays where it is const filter = encodeURIComponent(JSON.stringify({ room: { timeline: { limit: CONTEXT } } })); const since = cursor === null ? "" : `&since=${encodeURIComponent(cursor)}`; const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&filter=${filter}${since}`); const baseline = cursor === null; // first ever sync: history is context, not mail + for (const event of reply.account_data?.events ?? []) if (event.type === "m.direct") noteDirect(event.content); + noteInvites(reply.rooms?.invite); + await answerInvites(); const made = []; - for (const [roomId, room] of Object.entries(reply.rooms?.join ?? {})) { - const alias = aliasByRoomId.get(roomId); + let truncated = false; // the pending cap was reached inside this round + rounds: for (const [roomId, room] of Object.entries(reply.rooms?.join ?? {})) { + const alias = aliasByRoomId.get(roomId) ?? (directRooms.has(roomId) ? roomId : undefined); if (!alias) continue; for (const event of [...(room.state?.events ?? []), ...(room.timeline?.events ?? [])]) { - if (event.type === "m.room.member" && event.content?.displayname) displayNames.set(event.state_key, event.content.displayname); + const shown = event.content?.displayname; + if (event.type === "m.room.member" && typeof event.state_key === "string" && typeof shown === "string" && shown !== "") displayNames.set(event.state_key, docs.clip(shown, docs.limits.displayBytes)); } for (const event of room.timeline?.events ?? []) { if (event.type !== "m.room.message" || typeof event.content?.body !== "string") continue; - const entry = { sender: event.sender, text: event.content.body, atUnixMs: event.origin_server_ts ?? now() }; + if (typeof event.sender !== "string" || event.sender === "" || typeof event.event_id !== "string" || event.event_id === "") continue; + if (processed.has(event.event_id)) continue; // read already in a round the cap kept from advancing + const ts = event.origin_server_ts; + const entry = { sender: event.sender, text: docs.clip(event.content.body, docs.limits.textBytes), atUnixMs: Number.isSafeInteger(ts) && ts >= 0 ? ts : now() }; const context = (history.get(roomId) ?? []).slice(-CONTEXT); - if (!baseline && event.sender !== selfId) { + const deliverable = !baseline && !freshRooms.has(roomId) && event.sender !== selfId; + if (deliverable && pending.length + made.length >= maxPending) { truncated = true; break rounds; } + processed.add(event.event_id); + if (deliverable) { + const mentioned = event.content["m.mentions"]?.user_ids; made.push({ seq: ++seqCounter, roomAlias: alias, roomId, eventId: event.event_id, sender: event.sender, senderDisplay: displayNames.get(event.sender) ?? localpart(event.sender), text: entry.text, atUnixMs: entry.atUnixMs, - context: context.map(c => ({ ...c })), + context: context.map(c => ({ ...c, text: docs.clip(c.text, docs.limits.contextBytes) })), + mentions: Array.isArray(mentioned) ? mentioned.filter(id => typeof id === "string" && id !== "" && id.length <= 255).slice(0, docs.limits.mentions) : [], + isDirect: directRooms.has(roomId), }); } history.set(roomId, [...context, entry].slice(-CONTEXT)); } } - const wasIdle = pending.length === 0; pending.push(...made); - cursor = reply.next_batch; - batches.push({ token: reply.next_batch, maxSeq: seqCounter }); + if (!truncated) { // a truncated round leaves the position where it was: the rest stays on the server + cursor = reply.next_batch; + batches.push({ token: reply.next_batch, maxSeq: seqCounter }); + processed.clear(); + freshRooms.clear(); + } await commit(); if (made.length > 0) await state.save({ nextBatch: committed, seq: seqCounter }); // sequence numbers never repeat across a restart - if (made.length > 0 && (wasIdle || now() - lastRingAt >= RING_WINDOW_MS)) { - lastRingAt = now(); - Promise.resolve().then(ring).catch(error => say(`ring failed: ${error?.message ?? error}`)); - } return made; } @@ -133,20 +222,18 @@ export function createMouth({ } function pendingAfter(afterSeq) { - return pending.filter(d => d.seq > afterSeq).map(d => ({ ...d })); + return pending.filter(d => d.seq > afterSeq).map(d => ({ ...d, context: d.context.map(c => ({ ...c })), mentions: [...d.mentions] })); } async function send({ requestId, persona: asked, roomAlias, text }) { const refuse = refusal => ({ requestId, eventId: null, refusal }); if (asked !== persona) return refuse("PersonaMismatch"); - let released = false; - try { released = (await brake()) === true; } catch { released = false; } - if (!released) return refuse("Brake"); + if (!(await released())) return refuse("Brake"); if (typeof text !== "string" || text.trim() === "") return refuse("EmptyText"); - if (!started) await start(); - const roomId = roomIdByAlias.get(roomAlias); - if (!roomId) return refuse("UnknownRoom"); try { + if (!started || unjoined()) await start(); + const roomId = joinedAliases.has(roomAlias) ? roomIdByAlias.get(roomAlias) : directRooms.has(roomAlias) ? roomAlias : null; + if (!roomId) return refuse("UnknownRoom"); const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }); return { requestId, eventId: sent.event_id, refusal: null }; } catch (error) { @@ -156,10 +243,14 @@ export function createMouth({ } async function run(signal) { + const pause = ms => new Promise(r => { const t = setTimeout(r, ms); signal?.addEventListener("abort", () => { clearTimeout(t); r(); }, { once: true }); }); while (!signal?.aborted) { - try { await syncOnce(); } catch (error) { + try { + await syncOnce(); + if (pending.length >= maxPending) await pause(1000); + } catch (error) { say(`sync failed: ${error.message}`); - await new Promise(r => { const t = setTimeout(r, 5000); signal?.addEventListener("abort", () => { clearTimeout(t); r(); }, { once: true }); }); + await pause(5000); } } } @@ -167,24 +258,22 @@ export function createMouth({ return { start, syncOnce, ack, pendingAfter, send, run, get pendingCount() { return pending.length; } }; } -export function redact(message, token) { - return token ? message.split(token).join("[redacted]") : message; +// Replace every secret in a message. `secrets` is one string or a list. +export function redact(message, secrets) { + let out = String(message); + for (const secret of [secrets].flat()) if (secret) out = out.split(secret).join("[redacted]"); + return out; +} + +// Run `fn`; whatever it throws leaves with every secret redacted. +export async function guarded(secrets, fn) { + try { return await fn(); } catch (error) { throw new Error(redact(error?.message ?? String(error), secrets)); } } function localpart(userId) { return userId.replace(/^@/, "").split(":")[0]; } -// Ring: start the Persona's one-shot unit. No retry loop, no token in its environment. -export function createRing({ instance, run = execFile, log = () => {} }) { - return () => new Promise(done => { - run("systemctl", ["start", "--no-block", `mind-persona@${instance}.service`], { env: { PATH: process.env.PATH ?? "" } }, error => { - if (error) log(`ring failed: ${error.message}`); - done(); - }); - }); -} - // Brake: released only when the instance's control store decodes to a released // brake. Absent, undecodable or held reads are all "not released". export function createBrakeReader({ controlPath, readStore, decode }) { @@ -208,16 +297,38 @@ export function connectionIdFor(connectionKey) { return createHash("sha256").update(connectionKey).digest().readUInt32BE(0); } -// CultNet RUDP endpoint on loopback. Requests arrive as document_put_raw messages. -export async function startChatEndpoint({ mouth, persona, runtime, port, connectionKey, runtimeId, socketFactory = createSocket, host = "127.0.0.1", log = () => {} }) { +// The reply to a chat_pending_request: deliveries oldest first, each encoded on its own so one that +// will not encode is skipped and logged without silencing the rest, up to the reply byte budget. +// The caller acks what it has read and asks again. +export function pageOf(persona, deliveries, msgpack, log = () => {}) { + const page = []; + let bytes = 0; + for (const delivery of deliveries) { + let tuple, size; + try { tuple = docs.deliveryTuple(delivery); size = msgpack.encode(tuple).length; } catch (error) { + log(`delivery ${delivery?.seq} skipped: ${error?.message ?? error}`); + continue; + } + if (page.length > 0 && bytes + size > docs.limits.replyBytes) break; + page.push(tuple); + bytes += size; + } + return [persona, page]; +} + +// CultNet RUDP endpoint on loopback, one for every Persona in `mouths` (persona -> mouth). +// Requests arrive as document_put_raw messages. A request naming a Persona this +// process does not serve is answered with a typed UnknownPersona refusal. +export async function startChatEndpoint({ mouths, runtime, port, connectionKey, runtimeId, socketFactory = createSocket, host = "127.0.0.1", log = () => {} }) { if (!/^127\.\d+\.\d+\.\d+$/.test(host)) throw new Error("the chat endpoint listens on loopback only"); const { msgpack, cultnet } = runtime; const socket = socketFactory("udp4"); await new Promise((ok, fail) => { socket.once("error", fail); socket.once("listening", ok); socket.bind(port, host); }); const connectionId = connectionIdFor(connectionKey); const connection = new cultnet.CultNetRudpSocketTransportConnection({ runtimeId, socket, mode: "server", connectionId, initialSequence: 100, resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200 }); - const respond = (kind, recordKey, tuple) => connection.send("schema", msgpack.encode(cultnet.encodeCultNetMessageForWire( + const respond = (persona, kind, recordKey, tuple) => connection.send("schema", msgpack.encode(cultnet.encodeCultNetMessageForWire( docs.envelope({ kind, recordKey, payload: msgpack.encode(tuple), sourceRuntimeId: runtimeId, sourceAgentId: persona, sourceRole: "bifrost-persona-matrix-mouth" }), "cultnet.schema.v0"))); + const refuseUnknown = persona => respond(persona, docs.kinds.refusal, persona, docs.refusalTuple({ persona, refusal: "UnknownPersona" })); let busy = Promise.resolve(); const handle = async value => { try { @@ -227,16 +338,19 @@ export async function startChatEndpoint({ mouth, persona, runtime, port, connect const tuple = msgpack.decode(document.payload); if (document.schemaId === docs.kinds.pendingRequest) { const request = docs.parsePendingRequest(tuple); - if (request.persona !== persona) throw new Error("chat request names another Persona"); - respond(docs.kinds.pending, persona, docs.pendingTuple({ persona, deliveries: mouth.pendingAfter(request.afterSeq) })); + const mouth = mouths.get(request.persona); + if (!mouth) return refuseUnknown(request.persona); + respond(request.persona, docs.kinds.pending, request.persona, pageOf(request.persona, mouth.pendingAfter(request.afterSeq), msgpack, log)); } else if (document.schemaId === docs.kinds.ack) { const request = docs.parseAck(tuple); - if (request.persona !== persona) throw new Error("chat ack names another Persona"); + const mouth = mouths.get(request.persona); + if (!mouth) return refuseUnknown(request.persona); await mouth.ack(request.throughSeq); } else { const request = docs.parseSendRequest(tuple); - const receipt = await mouth.send(request); - respond(docs.kinds.sendReceipt, receipt.requestId, docs.sendReceiptTuple(receipt)); + const mouth = mouths.get(request.persona); + const receipt = mouth ? await mouth.send(request) : { requestId: request.requestId, eventId: null, refusal: "UnknownPersona" }; + respond(request.persona, docs.kinds.sendReceipt, receipt.requestId, docs.sendReceiptTuple(receipt)); } } catch (error) { log(`chat endpoint rejected a frame: ${error?.message ?? error}`); @@ -275,27 +389,36 @@ function credential(env, name) { return readFileSync(join(required(env, "CREDENTIALS_DIRECTORY"), name), "utf8").trim(); } +// PERSONAS is the comma-separated list of Persona slugs this process speaks for. +// Each slug names its credential (matrix-token-), its cursor directory under +// STATE_DIRECTORY and its brake at CONTROL_ROOT//control.cc. export async function main(env = process.env) { - const persona = required(env, "PERSONA"); - const instance = env.MIND_INSTANCE ?? persona; // the unit template is instanced by the Persona's mind - if (!/^[A-Za-z0-9_-]{1,64}$/.test(instance)) throw new Error("MIND_INSTANCE is not a slug"); - const token = credential(env, "matrix-token"); - const runtime = loadRuntime(resolve(env.CULTLIB_ROOT ?? resolve(import.meta.dirname, "..", "..", "CultLib"))); - const log = line => console.error(line); - const controlPath = join(env.CONTROL_ROOT ?? "/etc/gamecult/minds", instance, "control.cc"); - const mouth = createMouth({ - persona, token, log, - homeserver: required(env, "MATRIX_HOMESERVER").replace(/\/$/, ""), - rooms: required(env, "MATRIX_ROOMS").split(",").map(s => s.trim()).filter(Boolean), - state: cultcacheState({ runtime, path: env.STATE_PATH ?? join(required(env, "STATE_DIRECTORY"), "cursor.cc") }), - brake: createBrakeReader({ controlPath, readStore: runtime.readSingleFileStore, decode: runtime.msgpack.decode }), - ring: createRing({ instance, log }), + const personas = required(env, "PERSONAS").split(",").map(s => s.trim()).filter(Boolean); + if (personas.length === 0 || new Set(personas).size !== personas.length || !personas.every(p => SLUG.test(p))) throw new Error("PERSONAS must be distinct slugs"); + const tokens = new Map(personas.map(p => [p, credential(env, `matrix-token-${p}`)])); + const chatKey = credential(env, "chat-key"); + const secrets = [...tokens.values(), chatKey]; + return guarded(secrets, async () => { + const runtime = loadRuntime(resolve(env.CULTLIB_ROOT ?? resolve(import.meta.dirname, "..", "..", "CultLib"))); + const log = line => console.error(redact(line, secrets)); + const homeserver = required(env, "MATRIX_HOMESERVER").replace(/\/$/, ""); + const serverName = required(env, "MATRIX_SERVER_NAME"); + const rooms = required(env, "MATRIX_ROOMS").split(",").map(s => s.trim()).filter(Boolean); + const stateRoot = required(env, "STATE_DIRECTORY"); + const controlRoot = env.CONTROL_ROOT ?? "/etc/gamecult/minds"; + const mouths = new Map(personas.map(persona => { + mkdirSync(join(stateRoot, persona), { recursive: true }); + return [persona, createMouth({ + persona, homeserver, serverName, rooms, token: tokens.get(persona), log: line => log(`${persona}: ${line}`), + state: cultcacheState({ runtime, path: join(stateRoot, persona, "cursor.cc") }), + brake: createBrakeReader({ controlPath: join(controlRoot, persona, "control.cc"), readStore: runtime.readSingleFileStore, decode: runtime.msgpack.decode }), + })]; + })); + const listen = required(env, "CHAT_RUDP_LISTEN"); // host:port, loopback + const colon = listen.lastIndexOf(":"); + await startChatEndpoint({ mouths, runtime, host: listen.slice(0, colon), port: Number(listen.slice(colon + 1)), connectionKey: chatKey, runtimeId: "bifrost-persona-mouth", log }); + await Promise.all([...mouths.values()].map(mouth => mouth.run())); }); - await mouth.start(); - const listen = required(env, "CHAT_RUDP_LISTEN"); // host:port, loopback - const colon = listen.lastIndexOf(":"); - await startChatEndpoint({ mouth, persona, runtime, host: listen.slice(0, colon), port: Number(listen.slice(colon + 1)), connectionKey: credential(env, "chat-key"), runtimeId: `bifrost-persona-mouth-${persona}`, log }); - await mouth.run(); } if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {