diff --git a/tests/persona-matrix-mouth.test.mjs b/tests/persona-matrix-mouth.test.mjs index 610dd3b..7975976 100644 --- a/tests/persona-matrix-mouth.test.mjs +++ b/tests/persona-matrix-mouth.test.mjs @@ -14,8 +14,12 @@ 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, users, member, direct } = {}) { +// It is also the record of every room: events the sync script emits and messages sent are kept, and +// `preload` (room id -> events, oldest first) stands for history from before the mouth's process began. +async function fakeHomeserver({ sync, failWith, users, member, direct, preload = {}, failContext = false } = {}) { const seen = []; + const record_ = new Map(Object.entries(preload).map(([room, events]) => [room, [...events]])); + const remember = (room, event) => { const list = record_.get(room) ?? []; if (!list.some(e => e.event_id === event.event_id)) list.push(event); record_.set(room, list); }; const sent = new Map(); // txnId -> event_id const server = createServer((req, res) => { let raw = ""; @@ -26,7 +30,7 @@ async function fakeHomeserver({ sync, failWith, users, member, direct } = {}) { seen.push(record); 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", ""); + const p = record.path.replace("/_matrix/client/v3", "").replace("/_matrix/client/v1", ""); 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) }); @@ -35,10 +39,26 @@ async function fakeHomeserver({ sync, failWith, users, member, direct } = {}) { 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"), bearer)); + if (p === "/sync") { + const out = sync(record.query.get("since"), bearer); + for (const [room, r] of Object.entries(out.rooms?.join ?? {})) for (const e of r.timeline?.events ?? []) remember(room, e); + return reply(200, out); + } + const ctx = p.match(/^\/rooms\/(.+?)\/(messages|event\/.+|relations\/.+)$/); + if (ctx && req.method === "GET") { + if (failContext) return reply(500, { errcode: "M_UNKNOWN", error: `down ${record.auth}` }); + const events = record_.get(ctx[1]) ?? []; + const limit = Number(record.query.get("limit") ?? 10); + const threadOf = e => (e.content?.["m.relates_to"]?.rel_type === "m.thread" ? e.content["m.relates_to"].event_id : null); + const newestFirst = list => list.slice().reverse().slice(0, limit); + if (ctx[2] === "messages") return reply(200, { chunk: newestFirst(events) }); + if (ctx[2].startsWith("event/")) { const e = events.find(x => x.event_id === ctx[2].slice(6)); return e ? reply(200, e) : reply(404, { errcode: "M_NOT_FOUND" }); } + const root = ctx[2].match(/^relations\/(.+)\/m\.thread$/)?.[1]; + return reply(200, { chunk: newestFirst(events.filter(e => threadOf(e) === root)) }); + } 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}`); + if (!sent.has(m[2])) { sent.set(m[2], `$event${sent.size + 1}`); remember(m[1], { type: "m.room.message", event_id: sent.get(m[2]), sender: users?.[bearer] ?? SELF, origin_server_ts: 5000 + sent.size, content: record.body }); } return reply(200, { event_id: sent.get(m[2]) }); } reply(404, { errcode: "M_NOT_FOUND" }); @@ -598,3 +618,80 @@ test("send to another Persona's project room is refused as UnknownRoom", async ( assert.equal(hs.puts().length, 0); } finally { await hs.close(); } }); + +// Context is read from the homeserver, not remembered: these probes start the mouth with a +// room history the process never saw. +const ROOM = "!swarm:gamecult.test"; +const DM = "!dm:gamecult.test"; +const ALICE = `@alice:${HS_NAME}`; +const onlyAfterCursor = (roomId, events) => since => since === "c0" ? { next_batch: "c1", ...joined({ [roomId]: { timeline: { events } } }) } : { next_batch: since }; +const persistedCursor = () => memoryState({ nextBatch: "c0", seq: 7 }); + +test("context_survives_restart: a mouth that never saw the room delivers with the exchange read back from the homeserver", async () => { + const past = [message("$p1", ALICE, "first", 10), message("$p2", SELF, "my answer", 11), message("$p3", ALICE, "second", 12)]; + const hs = await fakeHomeserver({ preload: { [ROOM]: past }, sync: onlyAfterCursor(ROOM, [message("$n", ALICE, "now", 20)]) }); + try { + const [d] = await mouthFor(hs, { state: persistedCursor() }).syncOnce(); + assert.deepEqual(d.context.map(c => [c.sender, c.text]), [[ALICE, "first"], [SELF, "my answer"], [ALICE, "second"]]); + const asked = hs.seen.find(r => r.path.endsWith("/messages")); + assert.equal(asked.query.get("dir"), "b"); + assert.equal(asked.auth, `Bearer ${TOKEN}`); + } finally { await hs.close(); } +}); + +test("dm_context_holds_the_whole_exchange: her own earlier replies are in it, and it ends before the delivered message", async () => { + const past = [message("$d1", ALICE, "hi", 1), message("$d2", SELF, "hello, Deru", 2), message("$d3", ALICE, "do you remember", 3)]; + const hs = await fakeHomeserver({ preload: { [DM]: past }, direct: { [ALICE]: [DM] }, member: [...Object.values(ROOMS), DM], + sync: onlyAfterCursor(DM, [message("$d4", ALICE, "the thing", 4), message("$d5", ALICE, "after it", 5)]) }); + try { + const made = await mouthFor(hs, { state: persistedCursor() }).syncOnce(); + assert.equal(made.length, 2); + assert.ok(made[0].isDirect); + assert.deepEqual(made[0].context.map(c => c.text), ["hi", "hello, Deru", "do you remember"]); + assert.deepEqual(made[1].context.map(c => c.text), ["hi", "hello, Deru", "do you remember", "the thing"]); + assert.equal(hs.seen.filter(r => r.path.endsWith("/messages")).length, 1, "one fetch serves every delivery of the room in a round"); + } finally { await hs.close(); } +}); + +test("thread_context_is_the_thread: root and replies before the delivered reply, not the room's other messages", async () => { + const thread = { rel_type: "m.thread", event_id: "$root" }; + const inThread = (id, sender, body, ts) => ({ ...message(id, sender, body, ts), content: { msgtype: "m.text", body, "m.relates_to": thread } }); + const past = [message("$root", ALICE, "thread root", 1), message("$x", ALICE, "elsewhere in the room", 2), inThread("$t1", ALICE, "in thread", 3), inThread("$t2", SELF, "my thread reply", 4)]; + const hs = await fakeHomeserver({ preload: { [ROOM]: past }, sync: onlyAfterCursor(ROOM, [inThread("$t3", ALICE, "latest", 5)]) }); + try { + const [d] = await mouthFor(hs, { state: persistedCursor() }).syncOnce(); + assert.deepEqual(d.context.map(c => c.text), ["thread root", "in thread", "my thread reply"]); + assert.ok(hs.seen.some(r => r.path.includes("/relations/$root/m.thread"))); + assert.ok(!hs.seen.some(r => r.path.endsWith("/messages"))); + } finally { await hs.close(); } +}); + +test("context_byte_budget: the newest lines that fit 16 KiB, each clipped to a context line, at most 20", async () => { + const wide = Array.from({ length: 40 }, (_, i) => message(`$w${i}`, ALICE, `${String(i).padStart(2, "0")}${"é".repeat(2000)}`, 10 + i)); + const hs = await fakeHomeserver({ preload: { [ROOM]: wide }, sync: onlyAfterCursor(ROOM, [message("$n", ALICE, "now", 99)]) }); + try { + const [d] = await mouthFor(hs, { state: persistedCursor() }).syncOnce(); + const bytes = d.context.reduce((n, c) => n + Buffer.byteLength(c.text) + Buffer.byteLength(c.sender), 0); + assert.ok(bytes <= 16 * 1024, `${bytes} bytes`); + assert.ok(d.context.every(c => Buffer.byteLength(c.text) <= 1000)); + assert.ok(d.context.length > 0 && d.context.length < 20, "bytes, not count, bound it"); + assert.ok(d.context.at(-1).text.startsWith("39"), "the newest line is kept"); + assert.ok(d.context.at(0).text.startsWith(String(40 - d.context.length)), "the lines are contiguous to the newest"); + const short = Array.from({ length: 40 }, (_, i) => message(`$s${i}`, ALICE, `s${i}`, 10 + i)); + const hs2 = await fakeHomeserver({ preload: { [ROOM]: short }, sync: onlyAfterCursor(ROOM, [message("$n", ALICE, "now", 99)]) }); + try { assert.equal((await mouthFor(hs2, { state: persistedCursor() }).syncOnce())[0].context.length, 20, "the wire carries 20 lines"); } finally { await hs2.close(); } + } finally { await hs.close(); } +}); + +test("context_fetch_failure: empty context, one logged line without the token, and the delivery still goes out", async () => { + const lines = []; + const hs = await fakeHomeserver({ failContext: true, preload: { [ROOM]: [message("$p", ALICE, "past", 1)] }, sync: onlyAfterCursor(ROOM, [message("$n", ALICE, "now", 9)]) }); + try { + const [d] = await mouthFor(hs, { state: persistedCursor(), log: l => lines.push(l) }).syncOnce(); + assert.equal(d.text, "now"); + assert.deepEqual(d.context, []); + const logged = lines.filter(l => l.includes("context for")); + assert.equal(logged.length, 1); + assert.ok(!logged[0].includes(TOKEN)); + } finally { await hs.close(); } +}); diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 6ea8246..222c7c6 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -10,6 +10,12 @@ // Mouth: a send request becomes one PUT with txnId = request_id (a retry is // idempotent), and a room is joined, only while that Persona's brake is released. // +// Context: Matrix is the record. A delivery's `context` is read from the homeserver when the delivery +// is built (GET /messages, or the thread's relations for a threaded message), as this Persona's +// account, and holds m.room.message events from everyone, her own included, oldest first, ending +// just before the delivered event. Nothing about a room's past is kept in this process, so a restart +// loses none of it. A failed fetch gives empty context and a log line; the delivery still goes out. +// // 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. @@ -26,7 +32,11 @@ import * as docs from "./persona-chat-documents.mjs"; const API = "/_matrix/client/v3"; const BRAKE_TYPE = "eureka.control.brake.v1"; -const CONTEXT = 20; +const CONTEXT = 20; // /sync timeline limit per round +const CONTEXT_LINES = 20; // the most context lines the delivery tuple carries (persona-chat-documents maxContext) +const CONTEXT_BYTES = 16 * 1024; // text and sender bytes over all lines of one delivery: under CONTEXT_LINES x limits.contextBytes +const FETCH_LIMIT = 100; // events asked of the homeserver per context fetch +const V1 = "/_matrix/client/v1"; // /relations lives only under v1 const MAX_PENDING = 500; const SLUG = /^[A-Za-z0-9_-]{1,64}$/; @@ -41,7 +51,6 @@ export function createMouth({ 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; let cursor = null; // the /sync position we are reading from @@ -54,10 +63,10 @@ export function createMouth({ 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) { + async function matrix(method, path, body, prefix = API) { let response; try { - response = await httpFetch(`${homeserver}${API}${path}`, { + response = await httpFetch(`${homeserver}${prefix}${path}`, { method, headers: { authorization: `Bearer ${token}`, "content-type": "application/json" }, body: body === undefined ? undefined : JSON.stringify(body), @@ -150,6 +159,50 @@ export function createMouth({ if (unjoined()) await ensureRooms(); } + const isLine = e => e?.type === "m.room.message" && typeof e.content?.body === "string" && typeof e.sender === "string" && e.sender !== "" && e.content?.["m.relates_to"]?.rel_type !== "m.replace"; + const threadRootOf = e => { const r = e?.content?.["m.relates_to"]; return r?.rel_type === "m.thread" && typeof r.event_id === "string" && r.event_id !== "" ? r.event_id : null; }; + const lineOf = e => ({ sender: e.sender, text: docs.clip(e.content.body, docs.limits.contextBytes), atUnixMs: Number.isSafeInteger(e.origin_server_ts) && e.origin_server_ts >= 0 ? e.origin_server_ts : 0 }); + + // The room's recent messages outside threads, oldest first. + async function roomEvents(roomId) { + const filter = encodeURIComponent(JSON.stringify({ types: ["m.room.message"] })); + const reply = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/messages?dir=b&limit=${FETCH_LIMIT}&filter=${filter}`); + return (reply.chunk ?? []).filter(e => isLine(e) && threadRootOf(e) === null).reverse(); + } + // One thread's messages, root first. + async function threadEvents(roomId, rootId) { + const room = encodeURIComponent(roomId); + const root = await matrix("GET", `/rooms/${room}/event/${encodeURIComponent(rootId)}`); + const replies = await matrix("GET", `/rooms/${room}/relations/${encodeURIComponent(rootId)}/m.thread?dir=b&limit=${FETCH_LIMIT}`, undefined, V1); + return [root, ...(replies.chunk ?? []).reverse()].filter(isLine); + } + + // The context of one delivered event: the messages before it, the newest kept to the byte budget, each line clipped. + // `round` caches the fetches of one /sync round, so a room or thread is read once however many deliveries it made. + async function contextFor(roomId, event, round) { + const root = threadRootOf(event); + const key = root ? `${roomId}|${root}` : roomId; + if (!round.has(key)) { + round.set(key, (root ? threadEvents(roomId, root) : roomEvents(roomId)).catch(error => { + say(`context for ${roomId}${root ? ` thread ${root}` : ""} not fetched: ${error.message}`); + return []; + })); + } + const events = await round.get(key); + let cut = events.findIndex(e => e.event_id === event.event_id); + if (cut < 0) cut = events.filter(e => e.origin_server_ts < event.origin_server_ts).length; + const lines = []; + let bytes = 0; + for (const e of events.slice(0, cut).reverse()) { + const l = lineOf(e); + const size = Buffer.byteLength(l.text) + Buffer.byteLength(l.sender); + if (lines.length >= CONTEXT_LINES || bytes + size > CONTEXT_BYTES) break; + bytes += size; + lines.unshift(l); + } + return lines; + } + // One /sync round. Returns the deliveries it made. async function syncOnce() { if (!started || unjoined()) await start(); @@ -162,6 +215,7 @@ export function createMouth({ noteInvites(reply.rooms?.invite); await answerInvites(); const made = []; + const round = new Map(); 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); @@ -176,21 +230,20 @@ export function createMouth({ 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); 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 context = await contextFor(roomId, event, round); 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, text: docs.clip(c.text, docs.limits.contextBytes) })), + context, 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)); } } pending.push(...made);