persona matrix mouth: a delivery's context is read from Matrix, not remembered

The per-room in-memory history map is gone: a restart emptied it and /sync resumed from the
cursor, so every room (a DM first) lost its past. When a delivery is built the mouth now reads
the room's timeline from the homeserver as that Persona (GET /messages dir=b, or the thread's
root plus /relations m.thread for a threaded message), keeps m.room.message from everyone
including her own, oldest first, ending before the delivered event. Newest lines are kept to a
16 KiB byte budget (each clipped to limits.contextBytes), at most the 20 lines the delivery
tuple carries. One fetch per room or thread per sync round. A failed fetch gives empty context
and a redacted log line; the delivery still goes out.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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(); }
+});

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);