From 8e9ffb32f4b74d43727d4f2126ecde1b8fbbd7ce Mon Sep 17 00:00:00 2001 From: Heimdall Date: Wed, 11 Nov 2026 08:04:06 +0000 Subject: [PATCH] Mouth: room tail on chat_pending.v3 (20 lines a room, 60 in all, unacked past her cursor, speakable rooms only) Co-Authored-By: Claude Sonnet 5.5 --- docs/bridge.md | 19 +++- ...mecult.persona.chat_pending.v3.tag.msgpack | Bin 197 -> 198 bytes ...ecult.persona.chat_pending.v3.tail.msgpack | Bin 0 -> 163 bytes ...na.chat_pending.v3.unproven-notice.msgpack | Bin 153 -> 154 bytes tests/persona-matrix-mouth-runtime.test.mjs | 17 ++- tests/persona-matrix-mouth.test.mjs | 100 ++++++++++++++++++ tools/persona-chat-documents.mjs | 23 +++- tools/persona-matrix-mouth.mjs | 70 ++++++++++-- 8 files changed, 212 insertions(+), 17 deletions(-) create mode 100644 tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tail.msgpack diff --git a/docs/bridge.md b/docs/bridge.md index 4468988..a6430d1 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -216,10 +216,25 @@ Proposed runtime half (follow-up `persona-wakes-on-address-only`): 1. Poll with `chat_pending_request.v2` and parse `chat_pending.v3`. 2. A delivery is *addressed* to a Persona when `form == "message"` and either `basis` is `direct`, `tag` or `reply`, or `address::addresses` matches her names or her projects' names on the text. The off-beat wake keeps the sender gate of `heartbeat.rs` `wakes` (the operator or another Mind of the roster): a stranger's tag, `@room` or DM is delivered but wakes no one off the beat. `basis == "room"` is the runtime's policy to weigh; the mouth only reports it. 3. A real-time wake, a live session's tell and the stagger apply to addressed deliveries only. A `notice` or an `edit` never wakes, whatever its basis. -4. A delivery that is not addressed is acked in sequence with the rest, so the position moves (the mouth stops reading at 500 unacked deliveries); it is not lost: Matrix is the record, and the next addressed delivery's `context` (up to 20 lines) and a beat's own room read show it. Reading the tail on a scheduled beat is a runtime read of Matrix, a follow-up of the ruling `wake-reads-room-tail`; the mouth adds no endpoint for it. +4. A delivery that is not addressed is acked in sequence with the rest, so the position moves (the mouth stops reading at 500 unacked deliveries); it is not lost: Matrix is the record, and the next addressed delivery's `context` (up to 20 lines) and a beat's own room read show it. The room tail below gives a scheduled beat those lines without a Matrix read of its own; the mouth adds no endpoint for it. 5. `basis` and `form` are evidence, not authority: a sender's claim is weighed by the runtime's own rule (`address.rs` for names, the sender gate for who may wake her); the mouth reports what the event says and what only Matrix could look up (the author of a reply target). -Golden bytes: `tests/fixtures/persona-chat-v5/`; Huginn pins the same bytes. +### Room tail (`chat_pending.v3`, field 2) + +The v3 answer is `[persona, deliveries, tail]`. `tail` is a list of recent room lines, oldest first, for a reader that was not woken for them (a scheduled beat). It is built from the same `/messages` read as a delivery's `context`, as her account, with no new endpoint and no new homeserver call per line. A `chat_pending_request.v1` is still answered with the two-field `chat_pending.v2`, and `chat_pending.v3` has not been deployed, so no reader sees the third field before it is ready for it. + +| Bound | Rule | +|---|---| +| Rooms | Only rooms the Persona may speak in: a configured room she has joined, or one of her DMs. The same rule `send` uses (`speakableRoomId`). A room she is in but cannot speak in, and any other room, is absent. | +| Lines | A text line outside threads, not an edit, newest `20` per room, newest `60` over all rooms. Trimmed oldest first if the reply byte budget (256 KiB, deliveries first) would be exceeded. | +| Cursor | Only lines past her ack cursor in that room: the origin time of the newest delivery she has acked there (ties by event id). A room she has acked nothing in since the mouth started returns its recent lines, a quiet room included. The cursor is in memory: a restart forgets it and the next tail shows the recent lines again. | +| Read | A room's read is kept 15 seconds, dropped when a sync brings the room news or an ack moves the cursor. A room that cannot be read is logged and absent; the other rooms and the deliveries are unaffected. | + +A line is the 7-field tuple `[room, event_id, at_unix_ms, sender, form, basis, body]`: `room` is the configured alias (the room id for a DM), `form` and `basis` as for a delivery, `body` clipped to 1000 bytes with the same typed text line for media a delivery has (nothing is fetched). `basis` is what Matrix says on the event itself (`direct` in a DM, `tag`, `room`) and nothing the mouth would have to look up, so a reply is nil here: **nil is unproven, not "not for her"**, and a line in a DM is `direct`. Huginn's rule that an `is_direct` delivery wakes with a nil `basis` is unchanged and lives with Huginn; the tail does not feed it. + +The tail is **data, not instruction**. A line is what a sender typed in a room, and the mouth wakes no one, acks nothing and sends nothing because of it. A reader treats the body as untrusted text and weighs `basis` as evidence, as for a delivery. The tail does not replace a delivery: a message still reaches her as a delivery, with its own `seq`, and only `chat_ack` advances a position. + +Golden bytes: `tests/fixtures/persona-chat-v5/` (`chat_pending.v3.tail` carries two lines); Huginn pins the same bytes. Bifrost admits an observation only when guild, channel, target runtime, canonical repository, Persona, producer, and producer runtime match a preconfigured `bifrost.persona_feedback.target_binding.v0` document in Bifrost's private provider store. That binding also owns source visibility, data classification, and the delivery-store target. Bifrost never infers public visibility from Discord identifiers. It classifies the author using its own `bifrost.discord.actor_link.v0` state: a match yields `linked_governance_feedback`; absence yields `unlinked_social_feedback`. Neither class grants adoption, work, release, or deployment authority. diff --git a/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tag.msgpack b/tests/fixtures/persona-chat-v5/gamecult.persona.chat_pending.v3.tag.msgpack index 006f6cf6f6bbd6d289589bb4ebd1ec33d3ef4b36..6685251d258d7deb362a087216fa725da643e82c 100644 GIT binary patch delta 11 ScmX@gc#M&8@ { test("two clients served at once each get their own answer on their own session", async () => { const mouth = { pendingAfter: () => [], + tail: async () => [], ack: async () => {}, // Slow enough that the two requests are in flight together. send: async request => { await new Promise(r => setTimeout(r, 150)); return { requestId: request.requestId, eventId: `$${request.requestId}`, refusal: null }; }, @@ -138,6 +140,7 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a const calls = { acks: [], sends: [], reads: [] }; const mouth = { pendingAfter: after => (after < 7 ? [delivery] : []), + tail: async () => [tailLine], ack: async through => { calls.acks.push(through); }, read: async request => { calls.reads.push(request); return null; }, send: async request => { calls.sends.push(request); return request.text === "muted" ? { requestId: request.requestId, eventId: null, refusal: "Brake" } : { requestId: request.requestId, eventId: "$sent", refusal: null }; }, @@ -177,7 +180,9 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a put(docs.kinds.pendingRequestV3, "eureka", docs.pendingRequestTuple({ persona: "eureka", afterSeq: 0 })); document = await reply; assert.equal(document.schemaId, docs.kinds.pendingV3, "the kind asked by names the answer"); - assert.deepEqual(docs.parsePendingV3(msgpack.decode(document.payload)).deliveries[0], { ...delivery, basis: null, form: "message" }); + const v3 = docs.parsePendingV3(msgpack.decode(document.payload)); + assert.deepEqual(v3.deliveries[0], { ...delivery, basis: null, form: "message" }); + assert.deepEqual(v3.tail, [tailLine], "the v3 answer carries the room tail"); put(docs.kinds.ack, "eureka:7", docs.ackTuple({ persona: "eureka", throughSeq: 7 })); reply = nextReply(); @@ -322,6 +327,10 @@ const v5Dir = resolve(root, "tests", "fixtures", "persona-chat-v5"); const v5Samples = { "gamecult.persona.chat_pending_request.v2": { tuple: docs.pendingRequestTuple({ persona: "eureka", afterSeq: 3 }), parse: docs.parsePendingRequest }, "gamecult.persona.chat_pending.v3.tag": { tuple: docs.pendingV3Tuple({ persona: "eureka", deliveries: [{ ...delivery, isDirect: false, basis: "tag", form: "message" }] }), parse: docs.parsePendingV3 }, + "gamecult.persona.chat_pending.v3.tail": { tuple: docs.pendingV3Tuple({ persona: "eureka", deliveries: [], tail: [ + { roomAlias: "#operator:gamecult.org", eventId: "$e0", atUnixMs: 1_700_000_000_000, sender: "@meta:gamecult.org", form: "message", basis: null, text: "earlier" }, + { roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 1_700_000_001_000, sender: "@meta:gamecult.org", form: "notice", basis: "tag", text: "hello eureka" }, + ] }), parse: docs.parsePendingV3 }, "gamecult.persona.chat_pending.v3.unproven-notice": { tuple: docs.pendingV3Tuple({ persona: "eureka", deliveries: [{ ...delivery, isDirect: false, mentions: [], basis: null, form: "notice" }] }), parse: docs.parsePendingV3 }, }; for (const [name, { tuple, parse }] of Object.entries(v5Samples)) { @@ -331,7 +340,9 @@ for (const [name, { tuple, parse }] of Object.entries(v5Samples)) { if (process.env.UPDATE_FIXTURES === "1") { await mkdir(v5Dir, { 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)); - if (name.includes("pending.v3")) assert.equal(msgpack.decode(bytes)[1][0].length, 14); + if (name.includes("pending.v3")) assert.equal(msgpack.decode(bytes).length, 3, "persona, deliveries, tail"); + if (name.endsWith("v3.tag") || name.endsWith("unproven-notice")) assert.equal(msgpack.decode(bytes)[1][0].length, 14); + if (name.endsWith("v3.tail")) assert.equal(msgpack.decode(bytes)[2][0].length, 7); }); } @@ -370,7 +381,7 @@ test("RoomMoved is v2 only, and the v2 documents are strict about arity and newe test("the RUDP endpoint answers a v1 send with the three-field receipt and a v2 send with the four-field one", async () => { const seen = []; - const mouth = { pendingAfter: () => [], ack: async () => {}, send: async request => { + const mouth = { pendingAfter: () => [], tail: async () => [], ack: async () => {}, send: async request => { seen.push(request); return request.basis ? { requestId: request.requestId, eventId: null, refusal: "RoomMoved", newer: ["$a"] } : { requestId: request.requestId, eventId: "$sent", refusal: null, newer: [] }; } }; diff --git a/tests/persona-matrix-mouth.test.mjs b/tests/persona-matrix-mouth.test.mjs index 9458370..5558df3 100644 --- a/tests/persona-matrix-mouth.test.mjs +++ b/tests/persona-matrix-mouth.test.mjs @@ -1475,3 +1475,103 @@ test("addressing facts: a target that failed is not refetched within a minute, a assert.equal(fetched(), 2, "past the minute: read again"); } finally { await hs.close(); } }); + +// The room tail (docs/bridge.md, "Room tail"): her rooms' recent talk past her ack cursor, bounded, as data on the v3 answer. +const OPERATOR = "!operator:gamecult.test"; +const lines = (room, n, from = 0, sender = META) => Array.from({ length: n }, (_, i) => message(`$${room.slice(1, 4)}${from + i}`, sender, `line ${from + i}`, 10 + from + i)); +async function tailed({ preload, direct, member, events = [], act = async () => {} } = {}) { + const hs = await fakeHomeserver({ preload, direct, member, sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", ...joined(events.length ? { [ROOM]: { timeline: { events } } } : {}) } }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); + await act(mouth); + return { tail: await mouth.tail(), mouth, hs }; + } finally { await hs.close(); } +} + +test("room tail: 600 lines in a room keep the newest 20, oldest first", async () => { + const { tail } = await tailed({ preload: { [ROOM]: lines(ROOM, 600) } }); + assert.equal(tail.length, 20); + assert.deepEqual(tail.map(l => l.text), Array.from({ length: 20 }, (_, i) => `line ${580 + i}`)); + assert.deepEqual(Object.keys(tail[0]).sort(), ["atUnixMs", "basis", "eventId", "form", "roomAlias", "sender", "text"]); + assert.equal(tail[0].roomAlias, `#swarm:${HS_NAME}`); + assert.equal(tail[0].sender, META); + assert.equal(tail[0].form, "message"); + assert.equal(tail[0].basis, null, "unproven, not 'not for her'"); +}); + +test("room tail: a quiet room, with nothing delivered, still returns its recent lines", async () => { + const { tail } = await tailed({ preload: { [OPERATOR]: lines(OPERATOR, 3) } }); + assert.deepEqual(tail.map(l => l.text), ["line 0", "line 1", "line 2"]); + assert.ok(tail.every(l => l.roomAlias === `#operator:${HS_NAME}`)); +}); + +test("room tail: a room she is not in the speaking set of is absent, and a DM she holds is present with basis direct", async () => { + const STRANGER = "!stranger:gamecult.test", DM = "!mydm:gamecult.test"; + const { tail } = await tailed({ + preload: { [STRANGER]: lines(STRANGER, 2), [DM]: lines(DM, 2), [ROOM]: lines(ROOM, 1) }, + member: [...Object.values(ROOMS), STRANGER, DM], + direct: { [META]: [DM] }, + }); + assert.deepEqual([...new Set(tail.map(l => l.roomAlias))].sort(), [DM, `#swarm:${HS_NAME}`]); + assert.ok(tail.filter(l => l.roomAlias === DM).every(l => l.basis === "direct")); + assert.ok(!tail.some(l => l.eventId.startsWith("$str"))); +}); + +test("room tail: about 60 lines over all rooms, the newest, whatever the room count", async () => { + const dms = ["!d1:gamecult.test", "!d2:gamecult.test", "!d3:gamecult.test", "!d4:gamecult.test"]; + const preload = { [ROOM]: lines(ROOM, 30, 0), [OPERATOR]: lines(OPERATOR, 30, 30) }; + dms.forEach((dm, i) => { preload[dm] = lines(dm, 30, 60 + 30 * i); }); + const { tail } = await tailed({ preload, direct: { [META]: dms }, member: [...Object.values(ROOMS), ...dms] }); + assert.equal(tail.length, 60); + assert.deepEqual(tail.map(l => l.atUnixMs), tail.map(l => l.atUnixMs).sort((a, b) => a - b), "oldest first"); + assert.equal(Math.min(...tail.map(l => l.atUnixMs)), 110, "the oldest kept is the 60th newest of the 120 offered"); +}); + +test("room tail: lines at or before her ack cursor are gone, later ones stay, and mention basis rides on the line", async () => { + const events = [ + withContent("$a", META, { body: "first" }, 3000), + withContent("$b", META, { body: "tag", "m.mentions": { user_ids: [SELF] } }, 3001), + ]; + const hs = await fakeHomeserver({ sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", ...joined({ [ROOM]: { timeline: { events } } }) } }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); + const made = await mouth.syncOnce(); + assert.equal(made.length, 2); + assert.deepEqual((await mouth.tail()).map(l => [l.eventId, l.basis]), [["$a", null], ["$b", "tag"]]); + await mouth.ack(made[0].seq); + assert.deepEqual((await mouth.tail()).map(l => l.eventId), ["$b"], "unacked past the cursor only"); + await mouth.ack(made[1].seq); + assert.deepEqual(await mouth.tail(), []); + } finally { await hs.close(); } +}); + +test("room tail: a body is clipped like a context line, and an unreadable room is absent without silencing the others", async () => { + const { tail } = await tailed({ preload: { [ROOM]: [message("$big", META, "x".repeat(5000), 10)] } }); + assert.equal(Buffer.byteLength(tail[0].text), docs.limits.contextBytes); + const hs = await fakeHomeserver({ failContext: true, sync: since => ({ next_batch: since ?? "b0" }) }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); + assert.deepEqual(await mouth.tail(), []); + } finally { await hs.close(); } +}); + +test("room tail: only the v3 answer carries it; the v2 answer a v1 request gets is unchanged", async () => { + const { tail } = await tailed({ preload: { [ROOM]: lines(ROOM, 2) } }); + const pack = { encode: v => Buffer.from(JSON.stringify(v)) }; + assert.deepEqual(docs.parsePendingV3(pageOf("eureka", [], pack, () => {}, true, tail)).tail.map(l => l.text), ["line 0", "line 1"]); + const v2 = pageOf("eureka", [], pack, () => {}, false, tail); + assert.equal(v2.length, 2); + assert.deepEqual(docs.parsePending(v2), { persona: "eureka", deliveries: [] }); + assert.throws(() => docs.parsePendingV3(["eureka", []]), /3-field/); +}); + +test("room tail: the reply byte budget trims the oldest tail lines, never a delivery", async () => { + const pack = { encode: v => Buffer.from(JSON.stringify(v)) }; + const big = Array.from({ length: 60 }, (_, i) => ({ roomAlias: "#r:x", eventId: `$${i}`, atUnixMs: i, sender: "@a:x", form: "message", basis: null, text: "y".repeat(5000) })); + const kept = docs.parsePendingV3(pageOf("eureka", [], pack, () => {}, true, big)).tail; + assert.ok(kept.length > 0 && kept.length < 60); + assert.equal(kept.at(-1).eventId, "$59", "the newest survive"); +}); diff --git a/tools/persona-chat-documents.mjs b/tools/persona-chat-documents.mjs index 4867901..e47b7c8 100644 --- a/tools/persona-chat-documents.mjs +++ b/tools/persona-chat-documents.mjs @@ -39,7 +39,7 @@ 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 }); +export const limits = Object.freeze({ textBytes: maxTextBytes, contextBytes: 1000, displayBytes: 256, mentions: 64, replyBytes: 256 * 1024, tailPerRoom: 20, tailTotal: 60 }); // `value` cut to at most `maxBytes` of UTF-8 without splitting a character. export function clip(value, maxBytes) { @@ -237,13 +237,26 @@ export function parsePending(t) { return { persona: text(persona, "persona"), deliveries: deliveries.map(parseDelivery) }; } -export function pendingV3Tuple({ persona, deliveries }) { - return [text(persona, "persona"), deliveries.map(deliveryV3Tuple)]; +// One line of a room's recent talk, newest-unacked: room alias (or room id for a DM), event id, origin time, sender, form, basis, body. +// Data for the reader, never an instruction: no delivery, no wake, no ack position hangs on it. +export function tailLineTuple(l) { + return [text(l.roomAlias, "room"), text(l.eventId, "event_id"), seq(l.atUnixMs, "at_unix_ms"), text(l.sender, "sender"), form(l.form), basis(l.basis ?? null), String(l.text)]; +} +function parseTailLine(t) { + const [roomAlias, eventId, atUnixMs, sender, f, b, body] = tuple(t, 7, "tail_line"); + return { roomAlias: text(roomAlias, "room"), eventId: text(eventId, "event_id"), atUnixMs: seq(atUnixMs, "at_unix_ms"), sender: text(sender, "sender"), form: form(f), basis: basis(b), text: String(body) }; +} + +// The v3 answer: the persona, the deliveries, then the tail (at most tailPerRoom lines a room, tailTotal in all, oldest first). +export function pendingV3Tuple({ persona, deliveries, tail = [] }) { + if (tail.length > limits.tailTotal) throw new Error("chat_pending tail is over the line bound"); + return [text(persona, "persona"), deliveries.map(deliveryV3Tuple), tail.map(tailLineTuple)]; } export function parsePendingV3(t) { - const [persona, deliveries] = tuple(t, 2, "chat_pending_v3"); + const [persona, deliveries, tail] = tuple(t, 3, "chat_pending_v3"); if (!Array.isArray(deliveries)) throw new Error("chat_pending deliveries must be a list"); - return { persona: text(persona, "persona"), deliveries: deliveries.map(parseDeliveryV3) }; + if (!Array.isArray(tail)) throw new Error("chat_pending tail must be a list"); + return { persona: text(persona, "persona"), deliveries: deliveries.map(parseDeliveryV3), tail: tail.map(parseTailLine) }; } // The CultNet document_put_raw envelope every chat_* document travels in. diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 439816e..206fce2 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -63,6 +63,7 @@ const DELIVERED_INDEX = 4000; // delivered event ids remembered so a read report 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 TAIL_TTL_MS = 15 * 1000; const HEAD_PAGE = 50; // the room head a gated send reads; also the most ids a RoomMoved receipt names // The kind of an event for a delivery: an edit (m.replace) and an automated notice (m.notice) are not messages anyone is addressed by. @@ -141,6 +142,8 @@ export function createMouth({ let started = false; let lastPresence = null; // 'online' | 'unavailable' | 'offline' | null: command-only, suppresses repeat presence PUTs let activeUntil = 0; // her last sign of life in a turn, plus the grace + const ackedThrough = new Map(); // room id -> { ts, ids }: the newest delivery she has acked there, ids those at exactly that time; the tail is what lies past it + const tailReads = new Map(); // room id -> { at, lines }: a room's tail read, kept TAIL_TTL_MS or until a sync brings the room news or an ack moves the cursor const delivered = new Map(); // event id -> { roomId, roomAlias, seq }: what the mouth handed her, newest last, bounded const readThrough = new Map(); // room id -> { seq, eventId } of the last receipt that landed const readTails = new Map(); // room id -> the read report in flight: reports for one room run one at a time @@ -209,6 +212,8 @@ export function createMouth({ } } const unjoined = () => joinedAliases.size < rooms.length; + // The rooms this Persona may speak in: a configured room she has joined, or one of her DMs. One rule for send and for the tail. + const speakableRoomId = alias => joinedAliases.has(alias) ? roomIdByAlias.get(alias) : directRooms.has(alias) ? alias : null; // 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. @@ -294,11 +299,11 @@ export function createMouth({ // thread under, an event her account wrote, so her Self run's words count as hers), "room" (an @room mention). In that order; null // when none holds. Her names, her projects' names and the sender's role are the heartbeat's to match (Huginn address.rs), never the // mouth's. A reply target that cannot be read is null and logged: unknown, not "not hers". - async function basisOf(roomId, event) { + async function basisOf(roomId, event, lookup = true) { if (directRooms.has(roomId)) return "direct"; const mentions = event.content["m.mentions"]; if (Array.isArray(mentions?.user_ids) ? mentions.user_ids.includes(selfId) : mentions === undefined && pillFor(event.content.formatted_body)) return "tag"; - if (formOf(event.content) === "message" && await repliesToMe(roomId, event.content["m.relates_to"])) return "reply"; // an edit or a notice wakes no one: no fetch for it + if (lookup && formOf(event.content) === "message" && await repliesToMe(roomId, event.content["m.relates_to"])) return "reply"; // an edit or a notice wakes no one: no fetch for it return mentions?.room === true ? "room" : null; } // A pill is an anchor whose href is exactly her matrix.to link; a longer id, an attribute that merely contains one and a quoted reply do not count. @@ -379,6 +384,7 @@ export function createMouth({ 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)); } + if ((room.timeline?.events ?? []).length > 0) tailReads.delete(roomId); for (const event of room.timeline?.events ?? []) { if (event.type !== "m.room.message" || typeof event.content?.body !== "string") continue; if (typeof event.sender !== "string" || event.sender === "" || typeof event.event_id !== "string" || event.event_id === "") continue; @@ -466,12 +472,48 @@ export function createMouth({ } async function ack(throughSeq) { + for (const d of pending) if (d.seq <= throughSeq) { + const through = ackedThrough.get(d.roomId); + if (!through || d.atUnixMs > through.ts) ackedThrough.set(d.roomId, { ts: d.atUnixMs, ids: new Set([d.eventId]) }); + else if (d.atUnixMs === through.ts) through.ids.add(d.eventId); + } pending = pending.filter(d => d.seq > throughSeq); + tailReads.clear(); images?.acked(throughSeq); await commit(pending.length ? pending[0].seq - 1 : seqCounter); touch(); } + // Her rooms' recent talk, for a reader that was not woken for it: in each room she may speak in, the newest tailPerRoom lines past + // her ack cursor there (a room she has acked nothing in yields its recent lines), the newest tailTotal over all rooms, oldest + // first. From the same /messages read as a delivery's context, as her account, bodies clipped as context lines are. The basis is + // what Matrix says on the event itself (no reply lookup: nil is unproven). A room that cannot be read is logged and absent. + async function tail() { + if (!started) return []; + const targets = new Set(); + for (const alias of [...joinedAliases, ...directRooms]) { const id = speakableRoomId(alias); if (id) targets.add(id); } + const perRoom = await Promise.all([...targets].map(async roomId => { + let read = tailReads.get(roomId); + if (!read || now() - read.at >= TAIL_TTL_MS) { + read = { at: now(), lines: roomEvents(roomId).catch(error => { say(`tail for ${roomId} not fetched: ${error.message}`); return null; }) }; + tailReads.set(roomId, read); + } + const events = await read.lines; + if (events === null) { if (tailReads.get(roomId) === read) tailReads.delete(roomId); return []; } + const through = ackedThrough.get(roomId); + const alias = aliasByRoomId.get(roomId) ?? roomId; + const lines = []; + for (const e of events) { + const at = Number.isSafeInteger(e.origin_server_ts) && e.origin_server_ts >= 0 ? e.origin_server_ts : 0; + if (through && (at < through.ts || (at === through.ts && through.ids.has(e.event_id)))) continue; + if (typeof e.event_id !== "string" || e.event_id === "") continue; + lines.push({ roomAlias: alias, eventId: e.event_id, atUnixMs: at, sender: e.sender, form: formOf(e.content), basis: await basisOf(roomId, e, false), text: bodyText(e.content, docs.limits.contextBytes) }); + } + return lines.slice(-docs.limits.tailPerRoom); + })); + return perRoom.flat().sort((a, b) => a.atUnixMs - b.atUnixMs).slice(-docs.limits.tailTotal); + } + function pendingAfter(afterSeq) { return pending.filter(d => d.seq > afterSeq).map(d => ({ ...d, context: d.context.map(c => ({ ...c })), mentions: [...d.mentions], image: d.image && { ...d.image } })); } @@ -497,7 +539,7 @@ export function createMouth({ if (typeof text !== "string" || text.trim() === "") return refuse("EmptyText"); try { if (!started || unjoined()) await start(); - const roomId = joinedAliases.has(roomAlias) ? roomIdByAlias.get(roomAlias) : directRooms.has(roomAlias) ? roomAlias : null; + const roomId = speakableRoomId(roomAlias); if (!roomId) return refuse("UnknownRoom"); const result = await gate.run(roomId, async () => { if (basis) { @@ -626,7 +668,7 @@ export function createMouth({ } } - return { start, syncOnce, ack, read, pendingAfter, send, openDirect, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } }; + return { start, syncOnce, ack, read, pendingAfter, tail, 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. @@ -671,7 +713,8 @@ export function connectionIdFor(connectionKey) { // 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 = () => {}, v3 = false) { +// A v3 reply also carries `tail` (room lines, oldest first), trimmed oldest first to what the byte budget has left after the deliveries. +export function pageOf(persona, deliveries, msgpack, log = () => {}, v3 = false, tail = []) { const page = []; let bytes = 0; for (const delivery of deliveries) { @@ -684,7 +727,19 @@ export function pageOf(persona, deliveries, msgpack, log = () => {}, v3 = false) page.push(tuple); bytes += size; } - return [persona, page]; + if (!v3) return [persona, page]; + const lines = []; + for (const line of tail.slice().reverse()) { + let tuple, size; + try { tuple = docs.tailLineTuple(line); size = msgpack.encode(tuple).length; } catch (error) { + log(`tail line ${line?.eventId} skipped: ${error?.message ?? error}`); + continue; + } + if (bytes + size > docs.limits.replyBytes) break; + lines.unshift(tuple); + bytes += size; + } + return [persona, page, lines]; } // CultNet RUDP endpoint on loopback, one for every Persona in `mouths` (persona -> mouth). @@ -726,7 +781,8 @@ export async function startChatEndpoint({ mouths, runtime, port, connectionKey, const request = docs.parsePendingRequest(tuple); const mouth = mouths.get(request.persona); if (!mouth) return refuseUnknown(peer, request.persona); - respond(peer, request.persona, v3 ? docs.kinds.pendingV3 : docs.kinds.pending, request.persona, pageOf(request.persona, mouth.pendingAfter(request.afterSeq), msgpack, log, v3)); + const tail = v3 ? await mouth.tail().catch(error => { log(`tail not read: ${error?.message ?? error}`); return []; }) : []; + respond(peer, request.persona, v3 ? docs.kinds.pendingV3 : docs.kinds.pending, request.persona, pageOf(request.persona, mouth.pendingAfter(request.afterSeq), msgpack, log, v3, tail)); } else if (document.schemaId === docs.kinds.ack) { const request = docs.parseAck(tuple); const mouth = mouths.get(request.persona); -- 2.47.3