mouth-all-personas: one Matrix mouth serves every Persona; deliveries carry mentions and is_direct, refusals are typed; same-server DM invites accepted; the ring is deleted (Persona turns are the heartbeat's) (squash of eureka-body/mouth-all-personas 7b63031..5b628e2; Soul s1-s3)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
18bbe6c312
commit
cc76f0f2d0
7 changed files with 634 additions and 160 deletions
|
|
@ -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-<persona>`) 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/<persona>/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.
|
||||
|
||||
|
|
|
|||
Binary file not shown.
1
tests/fixtures/persona-chat-v1/gamecult.persona.chat_refusal.v1.msgpack
vendored
Normal file
1
tests/fixtures/persona-chat-v1/gamecult.persona.chat_refusal.v1.msgpack
vendored
Normal file
|
|
@ -0,0 +1 @@
|
|||
’¥ghost®UnknownPersona
|
||||
|
|
@ -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");
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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(); }
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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(/<2F>$/, "");
|
||||
}
|
||||
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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-<slug>), its cursor directory under
|
||||
// STATE_DIRECTORY and its brake at CONTROL_ROOT/<slug>/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) {
|
||||
|
|
|
|||
Loading…
Reference in a new issue