From 77975b9f0e3481f5f9c7ebabf5a0baa8cdea3abc Mon Sep 17 00:00:00 2001 From: Heimdall Date: Tue, 10 Nov 2026 18:32:05 +0000 Subject: [PATCH] persona mouth: open a direct room for a Persona, answer unknown request kinds New chat_open_direct request/receipt pair on the existing endpoint. The mouth creates an unencrypted private DM, invites the person (the line is the invite reason), says the line with a not-E2EE notice, and records the room in m.direct. Consent is the person's own membership: pending invite and decline are read from room state, and a decline stands until the person invites her. One creation per person per UTC day, the brake read before each write, a repeat by request id resumes. A request of an unknown kind now gets a typed UnknownKind refusal and a log line instead of silence. Golden fixtures in tests/fixtures/persona-chat-v4. Co-Authored-By: Claude Sonnet 5.5 --- ...at_open_direct_receipt.v1.declined.msgpack | 1 + ...at_open_direct_receipt.v1.existing.msgpack | 1 + ...chat_open_direct_receipt.v1.opened.msgpack | 1 + ...ersona.chat_open_direct_request.v1.msgpack | 1 + ...rsona.chat_refusal.v1.unknown-kind.msgpack | 1 + .../persona-matrix-mouth-open-direct.test.mjs | 275 ++++++++++++++++++ tests/persona-matrix-mouth-runtime.test.mjs | 83 ++++++ tools/persona-chat-documents.mjs | 41 ++- tools/persona-matrix-mouth.mjs | 109 ++++++- 9 files changed, 506 insertions(+), 7 deletions(-) create mode 100644 tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.declined.msgpack create mode 100644 tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.existing.msgpack create mode 100644 tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.opened.msgpack create mode 100644 tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_request.v1.msgpack create mode 100644 tests/fixtures/persona-chat-v4/gamecult.persona.chat_refusal.v1.unknown-kind.msgpack create mode 100644 tests/persona-matrix-mouth-open-direct.test.mjs diff --git a/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.declined.msgpack b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.declined.msgpack new file mode 100644 index 0000000..46b2220 --- /dev/null +++ b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.declined.msgpack @@ -0,0 +1 @@ +”¥req-3¨DeclinedÀÀ \ No newline at end of file diff --git a/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.existing.msgpack b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.existing.msgpack new file mode 100644 index 0000000..2a6cf01 --- /dev/null +++ b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.existing.msgpack @@ -0,0 +1 @@ +”¥req-2¨Existing±!dm1:gamecult.org¦$line2 \ No newline at end of file diff --git a/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.opened.msgpack b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.opened.msgpack new file mode 100644 index 0000000..d8dcd38 --- /dev/null +++ b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.opened.msgpack @@ -0,0 +1 @@ +”¥req-1¦Opened±!dm1:gamecult.org¦$line1 \ No newline at end of file diff --git a/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_request.v1.msgpack b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_request.v1.msgpack new file mode 100644 index 0000000..fb97ef6 --- /dev/null +++ b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_request.v1.msgpack @@ -0,0 +1 @@ +”¥req-1¦eureka²@meta:gamecult.orgªhello Meta \ No newline at end of file diff --git a/tests/fixtures/persona-chat-v4/gamecult.persona.chat_refusal.v1.unknown-kind.msgpack b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_refusal.v1.unknown-kind.msgpack new file mode 100644 index 0000000..41b7ade --- /dev/null +++ b/tests/fixtures/persona-chat-v4/gamecult.persona.chat_refusal.v1.unknown-kind.msgpack @@ -0,0 +1 @@ +’¦eureka«UnknownKind \ No newline at end of file diff --git a/tests/persona-matrix-mouth-open-direct.test.mjs b/tests/persona-matrix-mouth-open-direct.test.mjs new file mode 100644 index 0000000..b2c4081 --- /dev/null +++ b/tests/persona-matrix-mouth-open-direct.test.mjs @@ -0,0 +1,275 @@ +// A Persona asks her mouth for a direct room with one person. The homeserver here keeps room state the way Synapse does +// (create event with its content, one member event per account, encryption event when set), so the mouth's lookup, +// the consent that lives in the person's own membership, and the order of its writes are exercised, not mocked. +import test from "node:test"; +import assert from "node:assert/strict"; +import { createServer } from "node:http"; +import { createMouth } from "../tools/persona-matrix-mouth.mjs"; + +const HS = "gamecult.org"; +const ME = `@eureka:${HS}`; +const META = `@meta:${HS}`; +const PROJECT = "!project:gamecult.org"; +const ALIAS = `#swarm:${HS}`; + +async function homeserver({ people = [META, `@nova:${HS}`], now = () => 1000, failInvite: initialFailInvite = false } = {}) { + const config = { failInvite: initialFailInvite }; + const rooms = new Map(); // room id -> { events } + const log = []; // every write, in order, as "METHOD path" + let counter = 0; + const direct = { value: null }; + const member = (room, user, membership) => { + const events = rooms.get(room).events.filter(e => !(e.type === "m.room.member" && e.state_key === user)); + events.push({ type: "m.room.member", state_key: user, origin_server_ts: now(), content: { membership } }); + rooms.get(room).events = events; + }; + const server = createServer((req, res) => { + let raw = ""; + req.on("data", c => (raw += c)); + req.on("end", () => { + const url = new URL(req.url, "http://x"); + const path = decodeURIComponent(url.pathname).replace("/_matrix/client/v3", ""); + const body = raw ? JSON.parse(raw) : null; + const reply = (status, json) => { res.writeHead(status, { "content-type": "application/json" }); res.end(JSON.stringify(json)); }; + const write = () => { if (!path.startsWith("/presence/") && !path.includes("/account_data/")) log.push(`${req.method} ${path}`); }; + if (path === "/account/whoami") return reply(200, { user_id: ME }); + if (path === "/joined_rooms") return reply(200, { joined_rooms: [PROJECT, ...[...rooms].filter(([, r]) => r.events.some(e => e.type === "m.room.member" && e.state_key === ME && e.content.membership === "join")).map(([id]) => id)] }); + if (path.includes("/account_data/m.direct")) { + if (req.method === "PUT") { direct.value = body; return reply(200, {}); } + return direct.value ? reply(200, direct.value) : reply(404, { errcode: "M_NOT_FOUND" }); + } + if (path.startsWith("/directory/room/")) return reply(200, { room_id: PROJECT }); + if (path.startsWith("/profile/")) return people.includes(path.slice("/profile/".length)) ? reply(200, { displayname: "x" }) : reply(404, { errcode: "M_NOT_FOUND" }); + const state = path.match(/^\/rooms\/(.+)\/state$/); + if (state) return rooms.has(state[1]) ? reply(200, rooms.get(state[1]).events) : reply(403, {}); + if (req.method === "POST" && path === "/createRoom") { + write(); + const id = `!dm${++counter}:${HS}`; + rooms.set(id, { events: [{ type: "m.room.create", state_key: "", origin_server_ts: now() + counter, content: { creator: ME, ...body.creation_content } }] }); + if (body.preset !== "private_chat") return reply(400, {}); + member(id, ME, "join"); + rooms.get(id).body = body; + if (body.initial_state?.some(e => e.type === "m.room.encryption")) rooms.get(id).events.push({ type: "m.room.encryption", state_key: "", content: {} }); + return reply(200, { room_id: id }); + } + const invite = path.match(/^\/rooms\/(.+)\/invite$/); + if (invite) { write(); if (config.failInvite) return reply(403, { errcode: "M_FORBIDDEN" }); member(invite[1], body.user_id, "invite"); rooms.get(invite[1]).invited = body; return reply(200, {}); } + const put = path.match(/^\/rooms\/(.+)\/send\/m\.room\.message\/(.+)$/); + if (put) { + write(); + const r = rooms.get(put[1]); + r.sent ??= new Map(); + if (!r.sent.has(put[2])) r.sent.set(put[2], { id: `$e${counter}-${r.sent.size + 1}`, body }); + return reply(200, { event_id: r.sent.get(put[2]).id }); + } + reply(404, { errcode: "M_NOT_FOUND" }); + }); + }); + await new Promise(ok => server.listen(0, "127.0.0.1", ok)); + return { + url: `http://127.0.0.1:${server.address().port}`, rooms, log, direct, + set failInvite(v) { config.failInvite = v; }, + seedDm(id, user, membership, { encrypted = false, created = 5, marker } = {}) { + rooms.set(id, { events: [{ type: "m.room.create", state_key: "", origin_server_ts: created, content: { creator: ME, ...(marker ? { "org.gamecult.open_request": marker } : {}) } }] }); + member(id, ME, "join"); member(id, user, membership); + if (encrypted) rooms.get(id).events.push({ type: "m.room.encryption", state_key: "", content: {} }); + }, + set: member, + writes: pattern => log.filter(l => pattern.test(l)), + close: () => { server.closeAllConnections(); return new Promise(ok => server.close(ok)); }, + }; +} + +const memoryState = () => { const box = { value: null }; return { load: async () => box.value, save: async v => { box.value = v; } }; }; +const mouthFor = (hs, extra = {}) => createMouth({ persona: "eureka", homeserver: hs.url, serverName: HS, rooms: [ALIAS], token: "tok", state: memoryState(), brake: async () => true, syncTimeoutMs: 1, ...extra }); +const ask = (m, requestId, user = META, text = "hello Meta, may I ask you something?", persona = "eureka") => m.openDirect({ requestId, persona, userId: user, text }); +const outcome = r => r.outcome; + +test("a new person: creates an unencrypted private direct room, invites with the line as the reason, says the line with the not-E2EE notice, and answers Opened", async () => { + const hs = await homeserver(); + try { + const m = mouthFor(hs); + const r = await ask(m, "req-1"); + assert.equal(r.outcome, "Opened"); + const [roomId] = [...hs.rooms.keys()]; + assert.equal(r.roomId, roomId); + const room = hs.rooms.get(roomId); + assert.equal(room.body.preset, "private_chat"); + assert.equal(room.body.is_direct, true); + assert.equal(room.body.creation_content["m.federate"], false); + assert.equal(room.body.creation_content["org.gamecult.open_request"], "req-1"); + assert.ok(!JSON.stringify(room.body).includes("m.room.encryption"), "the room is created without encryption"); + assert.ok(!room.events.some(e => e.type === "m.room.encryption")); + assert.deepEqual(room.invited, { user_id: META, reason: "hello Meta, may I ask you something?" }); + const line = room.sent.get("req-1"); + assert.equal(r.eventId, line.id); + assert.match(line.body.body, /not end-to-end encrypted/, "the first message says so"); + assert.ok(line.body.body.endsWith("hello Meta, may I ask you something?")); + assert.deepEqual(hs.direct.value, { [META]: [roomId] }, "the room id is stored in m.direct"); + assert.deepEqual(hs.log.map(l => l.split(" ")[0] + " " + l.split("/").pop()), ["POST createRoom", "POST invite", "PUT req-1"], "order of writes"); + } finally { await hs.close(); } +}); + +test("the invite reason is the line cut to 300 bytes", async () => { + const hs = await homeserver(); + try { + await ask(mouthFor(hs), "req-1", META, "x".repeat(900)); + assert.equal(Buffer.byteLength([...hs.rooms.values()][0].invited.reason), 300); + } finally { await hs.close(); } +}); + +test("only her own request is served, and only for a person: mismatch, empty text, a foreign server, herself, another Mind, an unknown account", async () => { + const hs = await homeserver(); + try { + const nova = `@nova:${HS}`; + const m = mouthFor(hs, { isServed: id => id === nova }); + assert.equal(outcome(await ask(m, "r1", META, "hi", "nova")), "PersonaMismatch"); + assert.equal(outcome(await ask(m, "r2", META, " ")), "EmptyText"); + assert.equal(outcome(await ask(m, "r3", "@meta:matrix.org")), "ForeignUser"); + assert.equal(outcome(await ask(m, "r3b", "@meta:gamecult.org.evil.example")), "ForeignUser"); + assert.equal(outcome(await ask(m, "r4", ME)), "NotAPerson"); + assert.equal(outcome(await ask(m, "r5", nova)), "NotAPerson"); + assert.equal(outcome(await ask(m, "r6", `@nobody:${HS}`)), "NoSuchUser"); + assert.deepEqual(hs.log, [], "no write for any refusal"); + assert.equal(hs.rooms.size, 0); + } finally { await hs.close(); } +}); + +test("a held brake refuses before any Matrix call", async () => { + const hs = await homeserver(); + try { + const m = mouthFor(hs, { brake: async () => false }); + assert.equal(outcome(await ask(m, "r1")), "Brake"); + assert.deepEqual(hs.log, []); + assert.equal(hs.rooms.size, 0); + } finally { await hs.close(); } +}); + +test("the brake is read fresh before createRoom, before the invite and before the line: flipping it at any point stops every later write", async () => { + for (let free = 1; free <= 5; free++) { + const hs = await homeserver(); + try { + let reads = 0; + const events = []; + const m = mouthFor(hs, { brake: async () => { const ok = ++reads <= free; events.push(ok ? "free" : "held"); return ok; } }); + const originalLog = hs.log; + const push = originalLog.push.bind(originalLog); + originalLog.push = (...x) => { events.push(`write ${x[0]}`); return push(...x); }; + await ask(m, "r1"); + let held = false; + for (const e of events) { + if (e === "held") held = true; + else if (held && e.startsWith("write")) assert.fail(`free=${free}: a write after the brake was held: ${e}`); + } + const writes = events.filter(e => e.startsWith("write")); + for (let i = 0; i < events.length; i++) if (events[i].startsWith("write")) assert.equal(events[i - 1], "free", `free=${free}: ${events[i]} was not immediately preceded by a released brake read`); + if (free >= 5) assert.equal(writes.length, 3); + } finally { await hs.close(); } + } +}); + +test("a second request for a person who accepted says the line in the same room and answers Existing, with no new room", async () => { + const hs = await homeserver(); + try { + const m = mouthFor(hs); + const first = await ask(m, "r1"); + hs.set(first.roomId, META, "join"); // Meta accepts + const again = await ask(m, "r2", META, "second line"); + assert.deepEqual([again.outcome, again.roomId], ["Existing", first.roomId]); + assert.equal(hs.rooms.size, 1); + const sent = hs.rooms.get(first.roomId).sent.get("r2"); + assert.equal(again.eventId, sent.id); + assert.equal(sent.body.body, "second line", "no notice again after the first message"); + } finally { await hs.close(); } +}); + +test("a repeat of the same request id after a crash resumes in the room it made: one room, one line, Opened again", async () => { + const hs = await homeserver(); + try { + const first = await ask(mouthFor(hs), "r1"); + const again = await ask(mouthFor(hs), "r1"); // a restarted process: no memory of the first + assert.deepEqual([again.outcome, again.roomId, again.eventId], ["Opened", first.roomId, first.eventId]); + assert.equal(hs.rooms.size, 1); + assert.equal(hs.writes(/createRoom/).length, 1); + assert.equal(hs.writes(/\/invite/).length, 1); + } finally { await hs.close(); } +}); + +test("an invite the person has not answered is InvitePending: nothing new is written", async () => { + const hs = await homeserver(); + try { + const m = mouthFor(hs); + await ask(m, "r1"); + const before = hs.log.length; + const again = await ask(m, "r2"); + assert.equal(again.outcome, "InvitePending"); + assert.equal(hs.log.length, before); + } finally { await hs.close(); } +}); + +test("a decline stands: leaving or banning is Declined, for a restarted mouth too, until the person invites her", async () => { + for (const gone of ["leave", "ban"]) { + const hs = await homeserver(); + try { + const first = await ask(mouthFor(hs), "r1"); + hs.set(first.roomId, META, gone); + const before = hs.log.length; + for (const m of [mouthFor(hs), mouthFor(hs)]) { + assert.equal(outcome(await ask(m, `again-${Math.random()}`)), "Declined", gone); + } + assert.equal(hs.log.length, before, "no room, invite or line"); + // the person writes first: she is invited into a new room and joins; now there is a live DM + hs.seedDm("!theirs:gamecult.org", META, "join", { created: 99999 }); + const answer = await ask(mouthFor(hs), "r-after"); + assert.deepEqual([answer.outcome, answer.roomId], ["Existing", "!theirs:gamecult.org"]); + } finally { await hs.close(); } + } +}); + +test("an invite the homeserver refuses leaves the room made; the retry invites into that same room instead of making another", async () => { + const hs = await homeserver({ failInvite: true }); + try { + assert.equal(outcome(await ask(mouthFor(hs), "r1")), "SendFailed"); + assert.equal(hs.rooms.size, 1); + hs.failInvite = false; + const retry = await ask(mouthFor(hs), "r2"); + assert.equal(retry.outcome, "Opened"); + assert.equal(hs.rooms.size, 1, "no second room"); + assert.equal(retry.roomId, [...hs.rooms.keys()][0]); + assert.equal(hs.rooms.get(retry.roomId).invited.user_id, META); + } finally { await hs.close(); } +}); + +test("one creation per person per day: a second creation the same UTC day is DailyLimit, the next UTC day it is allowed", async () => { + let t = Date.parse("2026-11-10T10:00:00Z"); + const hs = await homeserver({ now: () => t }); + try { + const m = mouthFor(hs, { now: () => t }); + assert.equal(outcome(await ask(m, "r1")), "Opened"); + hs.rooms.clear(); // she left the room: no DM with them is found now, only the day's record stands + assert.equal(outcome(await ask(m, "r2")), "DailyLimit"); + assert.equal(hs.writes(/createRoom/).length, 1); + t += 24 * 3600 * 1000; + assert.equal(outcome(await ask(m, "r3")), "Opened"); + assert.equal(outcome(await ask(m, "r4", `@nova:${HS}`)), "Opened", "the limit is per person"); + } finally { await hs.close(); } +}); + +test("an encrypted direct room is not used: the line is never sent into it and a new unencrypted room is made", async () => { + const hs = await homeserver(); + try { + hs.seedDm("!old:gamecult.org", META, "join", { encrypted: true }); + const r = await ask(mouthFor(hs), "r1"); + assert.equal(r.outcome, "Opened"); + assert.notEqual(r.roomId, "!old:gamecult.org"); + assert.equal(hs.rooms.get("!old:gamecult.org").sent, undefined); + } finally { await hs.close(); } +}); + +test("a homeserver failure answers SendFailed and writes nothing the brake or a refusal should have stopped", async () => { + const hs = await homeserver(); + try { + const m = mouthFor(hs, { homeserver: "http://127.0.0.1:9" }); + assert.equal(outcome(await ask(m, "r1")), "SendFailed"); + } finally { await hs.close(); } +}); diff --git a/tests/persona-matrix-mouth-runtime.test.mjs b/tests/persona-matrix-mouth-runtime.test.mjs index d53de8e..098a422 100644 --- a/tests/persona-matrix-mouth-runtime.test.mjs +++ b/tests/persona-matrix-mouth-runtime.test.mjs @@ -438,3 +438,86 @@ test("over the RUDP endpoint: pending and an ack post no receipt, a read report try { client.close(); } catch { /* already closed */ } } }); + +// The open-direct pair (v4): the bytes Huginn pins in its crates/mind-body/tests/fixtures/persona-chat-v4/. +const v4Dir = resolve(root, "tests", "fixtures", "persona-chat-v4"); +const v4Samples = { + "gamecult.persona.chat_open_direct_request.v1": { tuple: docs.openDirectRequestTuple({ requestId: "req-1", persona: "eureka", userId: "@meta:gamecult.org", text: "hello Meta" }), parse: docs.parseOpenDirectRequest }, + "gamecult.persona.chat_open_direct_receipt.v1.opened": { tuple: docs.openDirectReceiptTuple({ requestId: "req-1", outcome: "Opened", roomId: "!dm1:gamecult.org", eventId: "$line1" }), parse: docs.parseOpenDirectReceipt }, + "gamecult.persona.chat_open_direct_receipt.v1.existing": { tuple: docs.openDirectReceiptTuple({ requestId: "req-2", outcome: "Existing", roomId: "!dm1:gamecult.org", eventId: "$line2" }), parse: docs.parseOpenDirectReceipt }, + "gamecult.persona.chat_open_direct_receipt.v1.declined": { tuple: docs.openDirectReceiptTuple({ requestId: "req-3", outcome: "Declined" }), parse: docs.parseOpenDirectReceipt }, + "gamecult.persona.chat_refusal.v1.unknown-kind": { tuple: docs.refusalTuple({ persona: "eureka", refusal: "UnknownKind" }), parse: docs.parseRefusal }, +}; +for (const [name, { tuple, parse }] of Object.entries(v4Samples)) { + test(`golden fixture: ${name} is byte-stable and parses back`, async () => { + const bytes = Buffer.from(msgpack.encode(tuple)); + const file = resolve(v4Dir, `${name}.msgpack`); + if (process.env.UPDATE_FIXTURES === "1") { await mkdir(v4Dir, { recursive: true }); await writeFile(file, bytes); } + assert.deepEqual(bytes, await readFile(file), "wire bytes changed: Huginn decodes these fixtures"); + assert.deepEqual(parse(msgpack.decode(bytes)), parse(tuple)); + }); +} + +test("the open-direct documents are strict: room and event ids exactly for Opened and Existing, a known outcome, a bounded line", () => { + for (const outcome of docs.directOutcomes) { + const placed = outcome === "Opened" || outcome === "Existing"; + assert.doesNotThrow(() => docs.openDirectReceiptTuple(placed ? { requestId: "r", outcome, roomId: "!a:b", eventId: "$e" } : { requestId: "r", outcome }), outcome); + assert.throws(() => docs.openDirectReceiptTuple(placed ? { requestId: "r", outcome } : { requestId: "r", outcome, roomId: "!a:b", eventId: "$e" }), /exactly for Opened and Existing/, outcome); + assert.throws(() => docs.parseOpenDirectReceipt(["r", outcome, placed ? "!a:b" : null, placed ? null : "$e"]), /exactly for/, `${outcome}: event id alone is half a placement`); + } + assert.throws(() => docs.openDirectReceiptTuple({ requestId: "r", outcome: "Invented" }), /unknown open-direct outcome/); + assert.throws(() => docs.parseOpenDirectRequest(["r", "eureka", "@a:b", "x".repeat(4001)]), /byte bound/); + assert.throws(() => docs.parseOpenDirectRequest(["r", "eureka", "", "x"]), /user_id/); + assert.throws(() => docs.parseOpenDirectRequest(["r", "eureka", "@a:b"]), /4-field/); + assert.throws(() => docs.sendReceiptTuple({ requestId: "r", refusal: "UnknownKind" }), /unknown refusal/, "UnknownKind exists only in the typed refusal document"); +}); + +test("the RUDP endpoint answers an open-direct request with a receipt for its Persona, refuses another Persona's and answers a request of an unknown kind loudly", async () => { + const opened = []; + const mouth = { openDirect: async request => { opened.push(request); return { requestId: request.requestId, outcome: "Opened", roomId: "!dm1:gamecult.org", eventId: "$line1" }; } }; + const key = "connection-key-for-test"; + const logged = []; + const server = await startChatEndpoint({ mouths: new Map([["eureka", mouth]]), runtime, port: 0, connectionKey: key, runtimeId: "mouth-test", log: line => logged.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({ + runtimeId: "huginn-test", socket: client, mode: "client", remoteHost: "127.0.0.1", remotePort: server.address.port, + connectionId: connectionIdFor(key), initialSequence: 1, resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200, + }); + const replies = []; + const waiting = []; + connection.on("frame", value => { + replies.push(runtime.cultnet.parseCultNetMessage(msgpack.decode(value.payload), "cultnet.schema.v0").document); + waiting.splice(0).forEach(w => w()); + }); + const nextReply = () => new Promise((ok, fail) => { const t = setTimeout(() => fail(new Error(`no reply; the mouth logged: ${logged.join(" | ")}`)), 5000); waiting.push(() => { clearTimeout(t); ok(replies.at(-1)); }); }); + const put = (kind, recordKey, tuple, agent = "eureka") => connection.send("schema", msgpack.encode(runtime.cultnet.encodeCultNetMessageForWire( + docs.envelope({ kind, recordKey, payload: msgpack.encode(tuple), sourceRuntimeId: "huginn-test", sourceAgentId: agent, sourceRole: "persona-organ" }), "cultnet.schema.v0"))); + try { + connection.connect(); + for (let i = 0; i < 100 && !connection.connected; i++) await new Promise(r => setTimeout(r, 20)); + assert.ok(connection.connected); + let reply = nextReply(); + put(docs.kinds.openDirectRequest, "req-1", docs.openDirectRequestTuple({ requestId: "req-1", persona: "eureka", userId: "@meta:gamecult.org", text: "hello" })); + let document = await reply; + assert.equal(document.schemaId, docs.kinds.openDirectReceipt); + assert.deepEqual(docs.parseOpenDirectReceipt(msgpack.decode(document.payload)), { requestId: "req-1", outcome: "Opened", roomId: "!dm1:gamecult.org", eventId: "$line1" }); + assert.deepEqual(opened, [{ requestId: "req-1", persona: "eureka", userId: "@meta:gamecult.org", text: "hello" }]); + + reply = nextReply(); + put(docs.kinds.openDirectRequest, "req-2", docs.openDirectRequestTuple({ requestId: "req-2", persona: "other", userId: "@meta:gamecult.org", text: "hello" }), "other"); + document = await reply; + assert.deepEqual(docs.parseOpenDirectReceipt(msgpack.decode(document.payload)), { requestId: "req-2", outcome: "UnknownPersona", roomId: null, eventId: null }); + assert.equal(opened.length, 1, "nothing reached a mouth for the unlisted Persona"); + + reply = nextReply(); + put("gamecult.persona.chat_invented_request.v9", "req-3", ["x"]); + document = await reply; + assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload))], [docs.kinds.refusal, { persona: "eureka", refusal: "UnknownKind" }], "an unknown kind is answered, typed"); + assert.ok(logged.some(l => /unknown kind: gamecult\.persona\.chat_invented_request\.v9/.test(l)), "and logged"); + } finally { + server.close(); + connection.close(); + try { client.close(); } catch { /* the connection already closed its socket */ } + } +}); diff --git a/tools/persona-chat-documents.mjs b/tools/persona-chat-documents.mjs index f20cad4..f5c6e67 100644 --- a/tools/persona-chat-documents.mjs +++ b/tools/persona-chat-documents.mjs @@ -16,6 +16,8 @@ export const kinds = Object.freeze({ sendRequestV2: "gamecult.persona.chat_send_request.v2", sendReceiptV2: "gamecult.persona.chat_send_receipt.v2", refusal: "gamecult.persona.chat_refusal.v1", + openDirectRequest: "gamecult.persona.chat_open_direct_request.v1", + openDirectReceipt: "gamecult.persona.chat_open_direct_receipt.v1", }); export const refusals = Object.freeze(["Brake", "PersonaMismatch", "UnknownRoom", "EmptyText", "SendFailed", "UnknownPersona"]); @@ -23,6 +25,12 @@ export const refusals = Object.freeze(["Brake", "PersonaMismatch", "UnknownRoom" // RoomMoved exists only in the v2 receipt: the v1 tuple, a v1 parse and the typed refusal document do not know it. export const refusalsV2 = Object.freeze([...refusals, "RoomMoved"]); +// The typed refusal document also answers a request of a kind this mouth does not serve (no Persona is known then). +const refusalDocument = Object.freeze([...refusals, "UnknownKind"]); + +// One outcome word per open-direct receipt. room_id and event_id are present exactly for Opened and Existing. +export const directOutcomes = Object.freeze(["Opened", "Existing", "Brake", "PersonaMismatch", "UnknownPersona", "EmptyText", "ForeignUser", "NotAPerson", "NoSuchUser", "InvitePending", "Declined", "DailyLimit", "SendFailed"]); + const maxTextBytes = 4000; export const maxNewer = 50; const maxContext = 20; @@ -119,15 +127,41 @@ export function parseSendReceiptV2(t) { // 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}`); + if (!refusalDocument.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)}`); + if (!refusalDocument.includes(refusal)) throw new Error(`unknown refusal ${String(refusal)}`); return { persona: text(persona, "persona"), refusal }; } +// A Persona asks her mouth to open a direct room with one person and say a first line there. user_id is a Matrix user id. +export function openDirectRequestTuple({ requestId, persona, userId, text: body }) { + return [text(requestId, "request_id"), text(persona, "persona"), text(userId, "user_id"), String(body)]; +} +export function parseOpenDirectRequest(t) { + const [requestId, persona, userId, body] = tuple(t, 4, "chat_open_direct_request"); + if (typeof body !== "string") throw new Error("chat_open_direct_request text must be a string"); + if (Buffer.byteLength(body, "utf8") > maxTextBytes) throw new Error("chat_open_direct_request text is over the byte bound"); + return { requestId: text(requestId, "request_id"), persona: text(persona, "persona"), userId: text(userId, "user_id"), text: body }; +} + +function checkDirectReceipt(outcome, roomId, eventId) { + if (!directOutcomes.includes(outcome)) throw new Error(`unknown open-direct outcome ${String(outcome)}`); + const placed = outcome === "Opened" || outcome === "Existing"; + if (placed !== (roomId !== null && eventId !== null) || (!placed && (roomId !== null || eventId !== null))) throw new Error("room_id and event_id are present exactly for Opened and Existing"); +} +export function openDirectReceiptTuple({ requestId, outcome, roomId = null, eventId = null }) { + checkDirectReceipt(outcome, roomId, eventId); + return [text(requestId, "request_id"), outcome, roomId === null ? null : text(roomId, "room_id"), eventId === null ? null : text(eventId, "event_id")]; +} +export function parseOpenDirectReceipt(t) { + const [requestId, outcome, roomId, eventId] = tuple(t, 4, "chat_open_direct_receipt"); + checkDirectReceipt(outcome, roomId, eventId); + return { requestId: text(requestId, "request_id"), outcome, roomId: roomId === null ? null : text(roomId, "room_id"), eventId: eventId === null ? null : text(eventId, "event_id") }; +} + export function deliveryTuple(d) { return [ seq(d.seq, "seq"), text(d.roomAlias, "room_alias"), text(d.roomId, "room_id"), text(d.eventId, "event_id"), @@ -190,10 +224,11 @@ export function envelope({ kind, recordKey, payload, sourceRuntimeId, sourceAgen }; } +// allowedKinds null: any kind passes (the caller answers a kind it does not serve). export function checkEnvelope(message, allowedKinds) { const d = message?.document; if (message?.schemaVersion !== "cultnet.document_put_raw.v0" || !d) throw new Error("not a document_put_raw message"); - if (!allowedKinds.includes(d.schemaId)) throw new Error(`document kind ${String(d.schemaId)} is not accepted here`); + if (allowedKinds !== null && !allowedKinds.includes(d.schemaId)) throw new Error(`document kind ${String(d.schemaId)} is not accepted here`); if (d.payloadEncoding !== "messagepack") throw new Error("chat documents are MessagePack"); if (!Array.isArray(d.tags) || d.tags.length !== 1 || d.tags[0] !== "cultnet.transport.rudp.v0") throw new Error("chat document tags substituted authority"); return d; diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 658c728..3c80f9a 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -58,6 +58,9 @@ const SLUG = /^[A-Za-z0-9_-]{1,64}$/; const SEND_TIMEOUT_MS = 10000; // the head read and the PUT of one gated send const PRESENCE_GRACE_MS = 5 * 60 * 1000; // a turn's last sign of life keeps her online this long const DELIVERED_INDEX = 4000; // delivered event ids remembered so a read report can be checked against them +const DM_SCAN = 200; // joined rooms read when looking for an existing DM with a person +const NOT_E2EE = "This room is not end-to-end encrypted: the homeserver can read what is said here."; +const dayOf = ms => new Date(ms).toISOString().slice(0, 10); const HEAD_PAGE = 50; // the room head a gated send reads; also the most ids a RoomMoved receipt names // A message line: a text event with a sender, not an edit. @@ -114,6 +117,7 @@ export function createMouth({ now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, gate = createRoomGate(), sendTimeoutMs = SEND_TIMEOUT_MS, presenceGraceMs = PRESENCE_GRACE_MS, images = null, // this Persona's image store (createImageStore), or null: images stay the typed text line + isServed = () => false, // whether a Matrix user id is an account this process speaks for: a Mind is not a person to open a DM with }) { if (!persona || !homeserver || !token || !rooms?.length) throw new Error("persona, homeserver, rooms and token are required"); const say = message => log(redact(String(message), token)); @@ -138,6 +142,7 @@ export function createMouth({ 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 + const openedOn = new Map(); // person's user id -> the UTC day this process last created a room for them: one creation per person per day async function matrix(method, path, body, prefix = API, signal) { let response; @@ -466,6 +471,91 @@ export function createMouth({ } } + // Every room this account shares with `userId` alone, newest first: { roomId, createdAt, marker, encrypted, theirs }, theirs the + // person's membership ("join" | "invite" | "leave" | "ban" | null when they were never invited). Project rooms are not DMs. A + // room qualifies when its members are exactly the two of us, or m.direct lists it for the person (a created room whose invite failed). + async function directsWith(userId) { + const joined = (await matrix("GET", "/joined_rooms")).joined_rooms ?? []; + const known = new Set(directContent[userId] ?? []); + const candidates = [...known, ...joined.filter(id => !known.has(id) && !aliasByRoomId.has(id))].slice(0, DM_SCAN); + const found = []; + for (const roomId of candidates) { + let events; + try { events = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/state`); } catch (error) { + if (/ 40[34] /.test(error.message)) continue; // a room m.direct still lists that this account has left or forgotten + throw error; + } + const members = events.filter(e => e.type === "m.room.member"); + const mine = members.find(e => e.state_key === selfId); + if (mine?.content?.membership !== "join") continue; + const others = members.filter(e => e.state_key !== selfId); + const theirs = others.find(e => e.state_key === userId); + if (!(known.has(roomId) || (others.length === 1 && theirs))) continue; + const create = events.find(e => e.type === "m.room.create"); + found.push({ + roomId, createdAt: create?.origin_server_ts ?? 0, marker: create?.content?.["org.gamecult.open_request"] ?? null, + encrypted: events.some(e => e.type === "m.room.encryption"), theirs: theirs?.content?.membership ?? null, + }); + } + return found.sort((a, b) => b.createdAt - a.createdAt); + } + + // The first line of an open: the invite is the ask and the room is not end-to-end encrypted, and the line says so. + async function putLine(roomId, requestId, text, first) { + return gate.run(roomId, async () => { + if (!(await released())) return null; + const body = first ? `${NOT_E2EE}\n\n${text}` : text; + return (await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body }, API, AbortSignal.timeout(sendTimeoutMs))).event_id; + }); + } + + // A Persona's request for a direct room with one person. One outcome word and, for Opened and Existing, the room and the line's event. + async function openDirect({ requestId, persona: asked, userId, text }) { + const answer = (outcome, roomId = null, eventId = null) => { say(`open direct ${requestId}: ${outcome}${roomId ? ` ${roomId}` : ""}`); return { requestId, outcome, roomId, eventId }; }; + if (asked !== persona) return answer("PersonaMismatch"); + if (!(await released())) return answer("Brake"); + if (typeof text !== "string" || text.trim() === "") return answer("EmptyText"); + try { + if (!started || unjoined()) await start(); + if (!serverName || typeof userId !== "string" || userId.slice(userId.indexOf(":") + 1) !== serverName) return answer("ForeignUser"); + if (userId === selfId || isServed(userId)) return answer("NotAPerson"); + try { await matrix("GET", `/profile/${encodeURIComponent(userId)}`); } catch (error) { + if (/ 404 /.test(error.message)) return answer("NoSuchUser"); + throw error; + } + const rooms = await directsWith(userId); + const usable = rooms.filter(r => !r.encrypted); + const resumed = usable.find(r => r.marker === requestId); // a repeat of this very request after a crash + const live = resumed ?? usable.find(r => r.theirs === "join"); + if (live) { + const eventId = await putLine(live.roomId, requestId, text, resumed !== undefined); + return eventId === null ? answer("Brake") : answer(resumed ? "Opened" : "Existing", live.roomId, eventId); + } + const latest = usable.find(r => r.theirs !== null); + if (latest?.theirs === "invite") return answer("InvitePending"); + if (latest) return answer("Declined"); // left or banned, and nothing newer: only the person reopens it, by inviting her + let roomId = usable.find(r => r.theirs === null)?.roomId ?? null; // made earlier, the invite never went out + if (roomId === null) { + if (openedOn.get(userId) === dayOf(now())) return answer("DailyLimit"); + if (!(await released())) return answer("Brake"); + ({ room_id: roomId } = await matrix("POST", "/createRoom", { preset: "private_chat", is_direct: true, creation_content: { "m.federate": false, "org.gamecult.open_request": requestId } }, API, AbortSignal.timeout(sendTimeoutMs))); + openedOn.set(userId, dayOf(now())); + directRooms.add(roomId); + directContent[userId] = [...new Set([...(directContent[userId] ?? []), roomId])]; + await matrix("PUT", `/user/${encodeURIComponent(selfId)}/account_data/m.direct`, directContent).catch(error => say(`m.direct not saved: ${error.message}`)); + } + if (!(await released())) return answer("Brake"); + await matrix("POST", `/rooms/${encodeURIComponent(roomId)}/invite`, { user_id: userId, reason: docs.clip(text, 300) }, API, AbortSignal.timeout(sendTimeoutMs)); + const eventId = await putLine(roomId, requestId, text, true); + if (eventId === null) return answer("Brake"); + touch(); setPresence(true).catch(() => {}); + return answer("Opened", roomId, eventId); + } catch (error) { + say(`open direct failed: ${error.message}`); + return answer("SendFailed"); + } + } + 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) { @@ -479,7 +569,7 @@ export function createMouth({ } } - return { start, syncOnce, ack, read, pendingAfter, send, run, get pendingCount() { return pending.length; } }; + return { start, syncOnce, ack, read, pendingAfter, send, openDirect, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } }; } // Replace every secret in a message. `secrets` is one string or a list. @@ -557,13 +647,24 @@ export async function startChatEndpoint({ mouths, runtime, port, connectionKey, const respond = (peer, persona, kind, recordKey, tuple) => server.send(peer, "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 = (peer, persona) => respond(peer, persona, docs.kinds.refusal, persona, docs.refusalTuple({ persona, refusal: "UnknownPersona" })); + const served = [docs.kinds.pendingRequest, docs.kinds.ack, docs.kinds.read, docs.kinds.sendRequest, docs.kinds.sendRequestV2, docs.kinds.openDirectRequest]; const handle = async ({ peer, channelId, payload }) => { try { if (channelId !== "schema") throw new Error("chat request arrived on a foreign channel"); const message = cultnet.parseCultNetMessage(msgpack.decode(payload), "cultnet.schema.v0"); - const document = docs.checkEnvelope(message, [docs.kinds.pendingRequest, docs.kinds.ack, docs.kinds.read, docs.kinds.sendRequest, docs.kinds.sendRequestV2]); + const document = docs.checkEnvelope(message, null); + if (!served.includes(document.schemaId)) { // a kind this mouth does not serve is answered, loudly, and logged: never dropped + log(`chat endpoint refused a request of an unknown kind: ${String(document.schemaId).slice(0, 120)}`); + const who = typeof document.sourceAgentId === "string" && document.sourceAgentId !== "" ? document.sourceAgentId : "unknown"; + return respond(peer, who, docs.kinds.refusal, who, docs.refusalTuple({ persona: who, refusal: "UnknownKind" })); + } const tuple = msgpack.decode(document.payload); - if (document.schemaId === docs.kinds.pendingRequest) { + if (document.schemaId === docs.kinds.openDirectRequest) { + const request = docs.parseOpenDirectRequest(tuple); + const mouth = mouths.get(request.persona); + const receipt = mouth ? await mouth.openDirect(request) : { requestId: request.requestId, outcome: "UnknownPersona", roomId: null, eventId: null }; + respond(peer, request.persona, docs.kinds.openDirectReceipt, receipt.requestId, docs.openDirectReceiptTuple(receipt)); + } else if (document.schemaId === docs.kinds.pendingRequest) { const request = docs.parsePendingRequest(tuple); const mouth = mouths.get(request.persona); if (!mouth) return refuseUnknown(peer, request.persona); @@ -669,7 +770,7 @@ export async function main(env = process.env) { const mouths = new Map(personas.map(persona => { mkdirSync(join(stateRoot, persona), { recursive: true }); return [persona, createMouth({ - persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, log: line => log(`${persona}: ${line}`), + persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, isServed: id => [...mouths.values()].some(m => m.userId === id), log: line => log(`${persona}: ${line}`), images: imageRoot ? createImageStore({ root: join(imageRoot, persona), log: line => log(`${persona}: ${line}`) }) : null, state: cultcacheState({ runtime, path: join(stateRoot, persona, "cursor.cc") }), brake: createBrakeReader({ controlPath: join(controlRoot, persona, "control.cc"), readStore: runtime.readSingleFileStore, decode: runtime.msgpack.decode }),