From f9e5775ea38ebc2847d25d6805f1857583d5c6f6 Mon Sep 17 00:00:00 2001 From: Heimdall Date: Thu, 12 Nov 2026 11:53:59 +0000 Subject: [PATCH 1/5] Mouth: basis follow for untagged messages from a Mind's engagement peers (CHAT_FOLLOW, default off) Co-Authored-By: Claude Sonnet 5.5 --- docs/bridge.md | 15 ++- tests/persona-matrix-mouth-follow.test.mjs | 148 +++++++++++++++++++++ tools/persona-chat-documents.mjs | 4 +- tools/persona-matrix-mouth.mjs | 71 +++++++++- 4 files changed, 232 insertions(+), 6 deletions(-) create mode 100644 tests/persona-matrix-mouth-follow.test.mjs diff --git a/docs/bridge.md b/docs/bridge.md index d66f9a5..1dc7df6 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -206,7 +206,7 @@ A client asks for the facts by the kind it polls with: `gamecult.persona.chat_pe | Field | Type | Meaning | |---|---|---| -| 12 `basis` | `"direct"`, `"tag"`, `"reply"`, `"room"` or nil | Why the message is for this Persona, by what Matrix says, first that holds: `direct` the room is one of her DMs; `tag` `m.mentions.user_ids` lists her id, or the client sent no `m.mentions` at all and an `` in the HTML body is exactly her `matrix.to` link; `reply` it replies to, or is in a thread under, an event her account wrote (her Self run speaks as her account, so its words count; any member's message in a thread under one of her events is a `reply` too, and the target is read once per room and event and kept for a minute, hit or failure); `room` an `@room` mention. A `notice` or an `edit` costs no reply lookup. Known limit: lookups run one at a time inside the `/sync` round (10 s each, at most two per reply), so a slow homeserver slows the round; they are not parallelised or capped per round. | +| 12 `basis` | `"direct"`, `"tag"`, `"reply"`, `"room"`, `"follow"` (only with `CHAT_FOLLOW=1`, see Follow) or nil | Why the message is for this Persona, by what Matrix says, first that holds: `direct` the room is one of her DMs; `tag` `m.mentions.user_ids` lists her id, or the client sent no `m.mentions` at all and an `` in the HTML body is exactly her `matrix.to` link; `reply` it replies to, or is in a thread under, an event her account wrote (her Self run speaks as her account, so its words count; any member's message in a thread under one of her events is a `reply` too, and the target is read once per room and event and kept for a minute, hit or failure); `room` an `@room` mention. A `notice` or an `edit` costs no reply lookup. Known limit: lookups run one at a time inside the `/sync` round (10 s each, at most two per reply), so a slow homeserver slows the round; they are not parallelised or capped per round. | | 13 `form` | `"message"`, `"notice"` or `"edit"` | `m.notice` is an automated line; `m.replace` is an edit of an earlier message. Only a `message` can address anyone. | What the mouth cannot know, and so what a nil `basis` means: **unproven, not "not for her"**. Her handle and public name in the text (any casing, with or without `@`, `'s`), the names her stewardships list, a project's repo name, and whether the sender is the operator or a Mind are all matched by Huginn, not here. A reply whose target the homeserver will not give is nil and logged. The mouth also cannot see whether a message was meant for her without naming her. Every delivery is still delivered, claimed tags included: nothing is dropped in the mouth, so the Huginn rule cannot lose a message the mouth had no way to judge. @@ -253,6 +253,19 @@ A line a delivery already shows and a room she cannot speak in change nothing he Golden bytes: `tests/fixtures/persona-chat-v5/` (`chat_pending.v3.tail` carries two lines and no gap, `chat_pending.v3.tail-gaps` one line and two gap rooms; the earlier v3 samples gain an empty fourth element); Huginn pins the same bytes. +### Follow (`basis` `"follow"`, `CHAT_FOLLOW=1`) + +A message from someone she is in conversation with is for her even when it names no one. **It is off unless the mouth runs with `CHAT_FOLLOW=1`** (exactly `1`; default off), because Huginn's parser refuses an unknown basis until its `Basis::Follow` (and the one `is_for` line) lands. With the flag off the mouth emits no `follow`, makes no extra read, and keeps no engagement state. + +With it on, the mouth gives a `message` the basis `follow` when it would otherwise have none or `room`, and: + +- the sender is a peer of an engagement in that room: someone her own post tagged (`m.mentions.user_ids`) or replied to (the author of the quoted `m.in_reply_to`, not a thread fallback pointer). Her post is read from /sync, so a post from any client of her account counts; +- the message is at most W = 600 s after her latest such post (event timestamps). The window anchors on her own post only: a follow delivery never extends it, and her posting again re-anchors it; +- fewer than K = 3 follows were already given for that (Mind, peer, room) in the hour before the message. Past the cap the message keeps the basis it had; +- the room is a public room: not one of her DMs, and its state has no `m.room.encryption`, `m.bridge` or `uk.half-shot.bridge` (read once per room, kept 5 minutes; unreadable is not public). A bridged sender or event (the same identifier rule as `platformEvent`) is neither a peer nor a follower. No engagement is opened from, and no follow is delivered in, such a room. + +State is two maps in memory (`room -> peer -> window end`, and follow times per `room|peer`); nothing is stored, and a restart only shortens windows. A tag or reply to her keeps `tag` or `reply`. A mouth `send` writes a plain `m.text` with no `m.mentions` or reply relation, so a post opens an engagement only when the event that comes back through /sync carries them (a post from another client, or a send that gains them). + ### Reply link (`chat_pending.v4`) A reply to anyone but the Persona arrives bare in v3: `basis` is `reply` only when the quoted author is her account. v4 adds what the message quotes. **It is off unless the mouth runs with `CHAT_PENDING_V4=1`** (exactly `1`; default off), because Huginn pins the tuple bytes and must read v4 first. With the flag off, `gamecult.persona.chat_pending_request.v3` is not a served kind: it is refused as any unknown kind is (`chat_refusal` `UnknownKind`, logged), byte for byte, and v1/v2/v3 requests are answered as before. With it on, that request kind (the v1 tuple `[persona, after_seq]` again; the kind asked by names the answer) is answered with `gamecult.persona.chat_pending.v4`. diff --git a/tests/persona-matrix-mouth-follow.test.mjs b/tests/persona-matrix-mouth-follow.test.mjs new file mode 100644 index 0000000..a541c5e --- /dev/null +++ b/tests/persona-matrix-mouth-follow.test.mjs @@ -0,0 +1,148 @@ +// Follow (docs/bridge.md, "Follow"): an untagged message from someone she tagged or replied to, within W of her own post, is basis "follow", +// at most K an hour per (peer, room), public rooms only, and only with the mouth's follow flag. The homeserver is an in-process fetch port: +// a scripted /sync (one reply per round), the room state and the events a reply target is read from. +import test from "node:test"; +import assert from "node:assert/strict"; +import { createMouth, followFromEnv } from "../tools/persona-matrix-mouth.mjs"; +import * as docs from "../tools/persona-chat-documents.mjs"; + +const HS = "gamecult.test"; +const ALIAS = `#swarm:${HS}`; +const ROOM = "!swarm:gamecult.test"; +const OTHER_ALIAS = `#side:${HS}`; +const OTHER_ROOM = "!side:gamecult.test"; +const ME = `@eureka:${HS}`; +const KIKO = `@kiko:${HS}`; +const STRANGER = `@stranger:${HS}`; +const PLATFORM = ["disc", "ord"].join(""); +const SEC = 1000, W = 600 * SEC, T0 = 1_000_000_000_000; +const msg = (id, sender, ts, content = {}) => ({ type: "m.room.message", event_id: id, sender, origin_server_ts: ts, content: { msgtype: "m.text", body: `line ${id}`, ...content } }); +const tagKiko = { "m.mentions": { user_ids: [KIKO] } }; + +// rounds: sync replies after the baseline, each { [roomId]: events }. state: room id -> state events. +function world({ rounds, state = {}, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { + const seen = []; + const known = new Map(); + let round = 0; + const reply = (status, body) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } }); + const hsFetch = async (url, init = {}) => { + const u = new URL(url); + const p = decodeURIComponent(u.pathname).replace(/^\/_matrix\/client\/v[13]/, ""); + seen.push(`${init.method ?? "GET"} ${p}`); + if (p === "/account/whoami") return reply(200, { user_id: ME }); + if (p === "/joined_rooms") return reply(200, { joined_rooms: [ROOM, OTHER_ROOM] }); + 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: { [ALIAS]: ROOM, [OTHER_ALIAS]: OTHER_ROOM }[p.slice(16)] }); + if (p.startsWith("/presence/") || p.includes("/receipt/")) return reply(200, {}); + const st = p.match(/^\/rooms\/([^/]+)\/state$/); + if (st) return reply(200, state[st[1]] ?? []); + const ev = p.match(/^\/rooms\/([^/]+)\/event\/(.+)$/); + if (ev) return known.has(ev[2]) ? reply(200, known.get(ev[2])) : reply(404, { errcode: "M_NOT_FOUND" }); + if (p === "/sync") { + if (u.searchParams.get("since") === null) return reply(200, { next_batch: "b0" }); + const events = rounds[round++] ?? {}; + for (const list of Object.values(events)) for (const e of list) known.set(e.event_id, e); + return reply(200, { next_batch: `b${round}`, rooms: { join: Object.fromEntries(Object.entries(events).map(([r, list]) => [r, { timeline: { events: list } }])) } }); + } + if (p.includes("/messages")) return reply(200, { chunk: [] }); + return reply(404, { errcode: "M_NOT_FOUND" }); + }; + let clock = T0; + const mouth = createMouth({ + persona: "eureka", homeserver: "http://hs.test", serverName: HS, rooms, token: "tok", fetch: hsFetch, follow, + state: { load: async () => null, save: async () => {} }, brake: async () => true, log: () => {}, syncTimeoutMs: 1, now: () => clock, + }); + return { mouth, seen, at: t => { clock = t; }, basisOfRound: async () => (await mouth.syncOnce()).map(d => [d.eventId, d.basis]) }; +} +const run = async (w, n) => { await w.mouth.syncOnce(); const out = []; for (let i = 0; i < n; i++) out.push(await w.basisOfRound()); return out; }; + +test("follow: a peer she tagged, untagged and within W of her post, is follow; a stranger and a peer elsewhere are not", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, + { [ROOM]: [msg("$kiko", KIKO, T0 + 60 * SEC), msg("$stranger", STRANGER, T0 + 61 * SEC)], [OTHER_ROOM]: [msg("$kikoside", KIKO, T0 + 62 * SEC)] }, + ] }); + const [, second] = await run(w, 2); + assert.deepEqual(second.sort(), [["$kiko", "follow"], ["$kikoside", null], ["$stranger", null]]); +}); + +test("follow: a reply of hers makes the author of the quoted event a peer", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$theirs", KIKO, T0 - 5 * SEC)] }, + { [ROOM]: [msg("$mine", ME, T0, { "m.relates_to": { "m.in_reply_to": { event_id: "$theirs" } } })] }, + { [ROOM]: [msg("$next", KIKO, T0 + 30 * SEC)] }, + ] }); + const out = await run(w, 3); + assert.deepEqual(out[2], [["$next", "follow"]]); +}); + +test("follow: the chain dies when she stops speaking; a follow does not extend W; speaking again re-anchors", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, + { [ROOM]: [msg("$a", KIKO, T0 + 9 * 60 * SEC)] }, // inside W: follow + { [ROOM]: [msg("$b", KIKO, T0 + W + SEC)] }, // after W from her post, though a follow came at 9 min: not follow + { [ROOM]: [msg("$again", ME, T0 + 11 * 60 * SEC, tagKiko)] }, + { [ROOM]: [msg("$c", KIKO, T0 + 20 * 60 * SEC)] }, // inside W of the second post: follow + ] }); + const out = await run(w, 5); + assert.deepEqual(out.flat(), [["$a", "follow"], ["$b", null], ["$c", "follow"]]); +}); + +test("follow: a tag or reply to her keeps its own basis", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, + { [ROOM]: [msg("$t", KIKO, T0 + SEC, { "m.mentions": { user_ids: [ME] } })] }, + ] }); + assert.deepEqual((await run(w, 2))[1], [["$t", "tag"]]); +}); + +test("follow: a DM, an encrypted room and a bridged room open nothing and receive nothing", async () => { + for (const state of [[{ type: "m.room.encryption", content: {} }], [{ type: "m.bridge", content: {} }], [{ type: "uk.half-shot.bridge", content: {} }]]) { + const w = world({ state: { [ROOM]: state }, rounds: [ + { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, + { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }, + ] }); + assert.deepEqual((await run(w, 2))[1], [["$kiko", null]]); + } + const dm = world({ direct: { [KIKO]: [ROOM] }, rounds: [{ [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }] }); + assert.deepEqual((await run(dm, 2))[1], [["$kiko", "direct"]]); +}); + +test("follow: a bridged room, a bridged sender as peer, and a bridged event deliver no follow", async () => { + const w2 = world({ state: { [ROOM]: [{ type: "m.bridge", content: {} }] }, rounds: [{ [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }] }); + assert.deepEqual((await run(w2, 2))[1], [["$kiko", null]]); + const puppet = `@_${PLATFORM}_123:${HS}`; + const w3 = world({ rounds: [ + { [ROOM]: [msg("$mine", ME, T0, { "m.mentions": { user_ids: [puppet] } })] }, + { [ROOM]: [msg("$p", puppet, T0 + SEC)] }, + ] }); + assert.deepEqual((await run(w3, 2))[1], [["$p", null]]); + const w4 = world({ rounds: [ + { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, + { [ROOM]: [msg("$kiko", KIKO, T0 + SEC, { [`${PLATFORM}.relay`]: true })] }, + ] }); + assert.deepEqual((await run(w4, 2))[1], [["$kiko", null]]); +}); + +test("follow: at most three an hour per peer and room; the fourth keeps its basis; an hour later it is follow again", async () => { + const posts = n => ({ [ROOM]: [msg(`$me${n}`, ME, T0 + n * 5 * 60 * SEC, tagKiko), msg(`$k${n}`, KIKO, T0 + n * 5 * 60 * SEC + SEC)] }); + const w = world({ rounds: [posts(0), posts(1), posts(2), posts(3), { [ROOM]: [msg("$me9", ME, T0 + 70 * 60 * SEC, tagKiko), msg("$k9", KIKO, T0 + 70 * 60 * SEC + SEC)] }] }); + const out = (await run(w, 5)).flat(); + assert.deepEqual(out, [["$k0", "follow"], ["$k1", "follow"], ["$k2", "follow"], ["$k3", null], ["$k9", "follow"]]); +}); + +test("follow: flag off emits nothing new and reads nothing extra; a fresh mouth has no engagement", async () => { + const rounds = [{ [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }]; + const off = world({ follow: false, rounds }); + assert.deepEqual((await run(off, 2))[1], [["$kiko", null]]); + assert.ok(!off.seen.some(l => l.endsWith("/state")), "off: no state read"); + const restarted = world({ rounds: [{ [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }] }); + assert.deepEqual((await run(restarted, 1))[0], [["$kiko", null]]); + assert.equal(followFromEnv({ CHAT_FOLLOW: "1" }), true); + for (const v of [undefined, "0", "true", "yes", ""]) assert.equal(followFromEnv({ CHAT_FOLLOW: v }), false); +}); + +test("follow: the basis is a name the delivery document accepts", () => { + assert.ok(docs.bases.includes("follow")); + const d = { seq: 1, roomAlias: ALIAS, roomId: ROOM, eventId: "$e", sender: KIKO, senderDisplay: "kiko", text: "x", atUnixMs: 1, context: [], image: null, mentions: [], isDirect: false, basis: "follow", form: "message" }; + assert.equal(docs.parseDeliveryV3(docs.deliveryV3Tuple(d)).basis, "follow"); +}); diff --git a/tools/persona-chat-documents.mjs b/tools/persona-chat-documents.mjs index 059e595..afdc7f3 100644 --- a/tools/persona-chat-documents.mjs +++ b/tools/persona-chat-documents.mjs @@ -68,8 +68,8 @@ export function parsePendingRequest(t) { // What the mouth can establish about a delivery from Matrix alone, for the Persona it is delivered to (see docs/bridge.md, "Addressing facts"). // `basis` is why the message is for her, or null when the mouth has no proof (not that it is not for her: her names, her projects' names and -// the sender's role are not the mouth's to know). `form` is the event's kind: only a "message" can address anyone. -export const bases = Object.freeze(["direct", "tag", "reply", "room"]); +// the sender's role are not the mouth's to know). "follow" is emitted only under the mouth's CHAT_FOLLOW flag. `form` is the event's kind: only a "message" can address anyone. +export const bases = Object.freeze(["direct", "tag", "reply", "room", "follow"]); export const forms = Object.freeze(["message", "notice", "edit"]); export function ackTuple({ persona, throughSeq }) { diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 10af28a..647e4e8 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -6,7 +6,7 @@ // its m.mentions user ids and whether the room is a DM. The /sync position // (next_batch) is persisted only when an ack covers it, so a restart before the // ack re-delivers instead of losing. A v3 delivery also carries what Matrix proves about why it is for her (basis: direct, tag, -// reply, room; form: message, notice, edit); names and roles are the heartbeat's to match, not the mouth's (docs/bridge.md, +// reply, room, and with CHAT_FOLLOW=1 follow; form: message, notice, edit); names and roles are the heartbeat's to match, not the mouth's (docs/bridge.md, // "Addressing facts"). Reading wakes nothing: the heartbeat polls // with a non-consuming chat_pending_request and decides who runs. An ack posts nothing to Matrix. Her public // m.read receipt follows a chat_read document from Huginn, sent when a delivery has gone into the model's @@ -23,6 +23,10 @@ // reads the room's head from Matrix and answers RoomMoved, naming the messages that moved it, instead // of the PUT. A v1 send, and a v2 send with no basis, are not checked; both still pass through the gate. // +// Follow (CHAT_FOLLOW=1, default off; docs/bridge.md "Follow"): her own post that tags or replies to someone opens a 10 minute engagement +// with that peer in that public room; the peer's next untagged messages inside it carry basis "follow", at most 3 an hour per peer and room. +// Memory only. Off, no basis "follow" is ever emitted: Huginn's parser refuses it until its Basis::Follow lands. +// // Media: an m.image/m.file/m.audio/m.video event is delivered as one typed line (see bodyText), not its bare filename. // An m.image event in a delivered message is also fetched, as this Persona and only when an image store is configured // (persona-chat-images.mjs): the delivery then carries a reference to the stored file, or the line says why not. @@ -68,6 +72,9 @@ const FETCH_DEADLINE_MS = 15000; // one fetch, whole: it ends before Huginn's 20 const FETCH_PER_HOUR = 30; // fetches one Persona may ask in an hour const RELATION_PAGES = 3; // pages of a receipt target's relations read before the answer is called unproven const PLATFORM = ["disc", "ord"].join(""); // the bridged platform's name, as Huginn's room_scope spells it: never written out whole +const FOLLOW_WINDOW_MS = 10 * 60 * 1000; // W: an engagement lasts this long from the Mind's own post; a follow never extends it +const FOLLOW_PER_HOUR = 3; // K: follow bases given per (Mind, peer, room) in an hour; past it the message keeps the basis it had +const FOLLOW_ROOM_TTL_MS = 5 * 60 * 1000; // how long a room's public verdict (not encrypted, not bridged) is kept 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. @@ -126,6 +133,7 @@ export function createMouth({ persona, homeserver, serverName, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake, now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, gate = createRoomGate(), sendTimeoutMs = SEND_TIMEOUT_MS, presenceGraceMs = PRESENCE_GRACE_MS, fetchDeadlineMs = FETCH_DEADLINE_MS, + follow = false, // the follow basis (CHAT_FOLLOW): an untagged message from someone she tagged or replied to, within FOLLOW_WINDOW_MS of her own post, is for her replyLink = false, // resolve each reply's quoted event into delivery.replyTo (the v4 answer; one more event read per reply to a non-Mind) images = null, // this Persona's image store (createImageStore), or null: images stay the typed text line isServed = async () => false, // (async) whether a Matrix user id is an account this process speaks for: a Mind is not a person to open a DM with @@ -303,6 +311,59 @@ export function createMouth({ return lines; } + // Follow (docs/bridge.md, "Follow"). Memory only: room id -> peer id -> the time her engagement with that peer ends, and room id|peer id -> + // the times follows were given. A restart empties both, which only shortens windows. Opened by her own post (seen in /sync, so a post from + // any client of her account counts) that tags or replies to a peer; never by a follow delivery; never in a DM, an encrypted room or a bridged room. + const engagements = new Map(); + const followsGiven = new Map(); + const publicRooms = new Map(); // room id -> { ok, at } + const platformId = id => platformEvent({ sender: id }); + async function publicRoom(roomId) { + const known = publicRooms.get(roomId); + if (known && now() - known.at < FOLLOW_ROOM_TTL_MS) return known.ok; + try { + const state = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/state`, undefined, API, AbortSignal.timeout(sendTimeoutMs)); + if (!Array.isArray(state)) return false; + const ok = !state.some(e => e?.type === "m.room.encryption" || e?.type === "m.bridge" || e?.type === "uk.half-shot.bridge"); + publicRooms.set(roomId, { ok, at: now() }); + return ok; + } catch (error) { + say(`follow: room ${roomId} state not read: ${error.message}`); + return false; // unknown is not public + } + } + const eventAt = event => Number.isSafeInteger(event.origin_server_ts) && event.origin_server_ts >= 0 ? event.origin_server_ts : now(); + // Her own post: whoever it tags (m.mentions) or replies to becomes a peer for W from this post, re-anchoring one she already had. + async function noteOwnPost(roomId, event) { + if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event)) return; + const peers = new Set(Array.isArray(event.content["m.mentions"]?.user_ids) ? event.content["m.mentions"].user_ids : []); + const relates = event.content["m.relates_to"]; + const replied = relates?.["m.in_reply_to"]?.event_id; + if (typeof replied === "string" && replied !== "" && relates.is_falling_back !== true) { + const author = await authorOf(roomId, replied); + if (author !== null) peers.add(author); + } + const ends = eventAt(event) + FOLLOW_WINDOW_MS; + const ids = [...peers].filter(id => docs.isMatrixId("@", id) && id !== selfId && !platformId(id)).slice(0, docs.limits.mentions); + if (ids.length === 0 || ends <= now() || !(await publicRoom(roomId))) return; + const room = engagements.get(roomId) ?? new Map(); + for (const [id, until] of room) if (until <= now()) room.delete(id); + for (const id of ids) room.set(id, ends); + engagements.set(roomId, room); + } + // Whether this message is a follow, and if so spends one of the peer's K for the hour. Called only for a message that has no better basis. + async function followed(roomId, event) { + if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event)) return false; + const at = eventAt(event); + if (!((engagements.get(roomId)?.get(event.sender) ?? 0) > at)) return false; + if (!(await publicRoom(roomId))) return false; + const key = `${roomId}|${event.sender}`; + const recent = (followsGiven.get(key) ?? []).filter(t => at - t < 60 * 60 * 1000); + if (recent.length >= FOLLOW_PER_HOUR) { followsGiven.set(key, recent); return false; } + followsGiven.set(key, [...recent, at]); + return true; + } + // Why a delivered event is for this Persona, from what Matrix says and nothing else: "direct" (a DM), "tag" (m.mentions lists her // id, or, from a client that sends no m.mentions at all, the body links her matrix.to pill), "reply" (it replies to, or sits in a // thread under, an event her account wrote, so her Self run's words count as hers), "room" (an @room mention). In that order; null @@ -474,7 +535,9 @@ export function createMouth({ const deliverable = !baseline && !freshRooms.has(roomId) && event.sender !== selfId; if (deliverable && pending.length + made.length >= maxPending) { truncated = true; break rounds; } if (deliverable && replyLink && replyReads + targetsToRead(roomId, event) > REPLY_READS_PER_ROUND) { say(`event ${event.event_id} not read: round cap on target reads, the rest of the round waits for the next`); truncated = true; break rounds; } // not processed: the next round meets it again, with its read count reset - const basis = deliverable ? await basisOf(roomId, event) : null; + if (!baseline && !freshRooms.has(roomId) && event.sender === selfId && event.type === "m.room.message") await noteOwnPost(roomId, event); + let basis = deliverable ? await basisOf(roomId, event) : null; + if (deliverable && (basis === null || basis === "room") && await followed(roomId, event)) basis = "follow"; const replyTo = deliverable ? await replyToOf(roomId, event) : null; processed.add(event.event_id); if (deliverable) { @@ -1074,6 +1137,8 @@ export function loadRuntime(cultRoot) { export const replyLinkFromEnv = env => env.CHAT_PENDING_V4 === "1"; // The fetch flag, the same way: on only for exactly "1". Off, the fetch request kind is an unknown kind. export const fetchFromEnv = env => env.CHAT_FETCH === "1"; +// The follow flag, the same way. Off, the mouth never emits basis "follow" (Huginn's parser refuses an unknown basis until its Basis::Follow lands). +export const followFromEnv = env => env.CHAT_FOLLOW === "1"; function required(env, name) { if (!env[name]) throw new Error(`${name} is required`); @@ -1127,7 +1192,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, replyLink, isServed: async id => { await Promise.all([...mouths.values()].map(m => m.start().catch(() => {}))); return [...mouths.values()].some(m => m.userId === id); }, log: line => log(`${persona}: ${line}`), + persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, replyLink, follow: followFromEnv(env), isServed: async id => { await Promise.all([...mouths.values()].map(m => m.start().catch(() => {}))); return [...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 }), -- 2.47.3 From 7f3e4f961a1ae7ba07c655cc860f7f054239123e Mon Sep 17 00:00:00 2001 From: Heimdall Date: Thu, 12 Nov 2026 12:00:30 +0000 Subject: [PATCH 2/5] Follow: anchor on her own send; count recent speakers, not tags alone Co-Authored-By: Claude Sonnet 5.5 --- docs/bridge.md | 14 +- tests/persona-matrix-mouth-follow.test.mjs | 177 ++++++++++++++------- tools/persona-matrix-mouth.mjs | 86 ++++++---- 3 files changed, 182 insertions(+), 95 deletions(-) diff --git a/docs/bridge.md b/docs/bridge.md index 1dc7df6..1341260 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -255,16 +255,16 @@ Golden bytes: `tests/fixtures/persona-chat-v5/` (`chat_pending.v3.tail` carries ### Follow (`basis` `"follow"`, `CHAT_FOLLOW=1`) -A message from someone she is in conversation with is for her even when it names no one. **It is off unless the mouth runs with `CHAT_FOLLOW=1`** (exactly `1`; default off), because Huginn's parser refuses an unknown basis until its `Basis::Follow` (and the one `is_for` line) lands. With the flag off the mouth emits no `follow`, makes no extra read, and keeps no engagement state. +A message from someone she is in conversation with is for her even when it names no one. **It is off unless the mouth runs with `CHAT_FOLLOW=1`** (exactly `1`; default off), because Huginn's parser refuses an unknown basis until its `Basis::Follow` (and the one `is_for` line) lands. With the flag off the mouth emits no `follow`, makes no extra read, and keeps no follow state. -With it on, the mouth gives a `message` the basis `follow` when it would otherwise have none or `room`, and: +With it on: -- the sender is a peer of an engagement in that room: someone her own post tagged (`m.mentions.user_ids`) or replied to (the author of the quoted `m.in_reply_to`, not a thread fallback pointer). Her post is read from /sync, so a post from any client of her account counts; -- the message is at most W = 600 s after her latest such post (event timestamps). The window anchors on her own post only: a follow delivery never extends it, and her posting again re-anchors it; -- fewer than K = 3 follows were already given for that (Mind, peer, room) in the hour before the message. Past the cap the message keeps the basis it had; -- the room is a public room: not one of her DMs, and its state has no `m.room.encryption`, `m.bridge` or `uk.half-shot.bridge` (read once per room, kept 5 minutes; unreadable is not public). A bridged sender or event (the same identifier rule as `platformEvent`) is neither a peer nor a follower. No engagement is opened from, and no follow is delivered in, such a room. +- **Anchor.** Her own post in a public room anchors that room for W = 600 s from the post. The mouth anchors at send time (the send's own time, in `send`) and also from the /sync echo of her own events, so a post from another client of her account anchors too; one event anchors once. Only her posting again re-anchors the room and re-snapshots who is counted; a follow never extends the window. +- **Counted senders** (snapshotted at the post): those who spoke in that room in the 10 minutes before the post (a per-room ring of recent senders with event times, 200 entries), plus anyone the post tagged (`m.mentions.user_ids`) or replied to (the author of the quoted `m.in_reply_to`, not a thread fallback pointer) when the event carries it. Someone who first speaks after her post is not counted. Nothing reads message text; no ids are scraped from bodies. She herself, and bridged senders (the `platformEvent` identifier rule), never count. +- **Follow.** A `message` (not an edit, not a notice) from a counted sender gets `follow` when it would otherwise have no basis or `room`; `tag`, `reply` and `direct` are never replaced. Its timestamp must lie inside the window and this process's clock must not be past the window's end by more than 60 s, so an old or forged `origin_server_ts` reuses nothing. At most K = 3 per (Mind, peer, room) in the hour before the message; past it the message keeps the basis it had. +- **"Public" means non-DM, non-encrypted, non-bridged.** The room is not one of her DMs and its state has no `m.room.encryption`, `m.bridge` or `uk.half-shot.bridge` (read once per room, kept 5 minutes; unreadable or malformed state is not public). An invite-only plaintext room counts as public. Nothing is anchored in, and no follow is delivered in, any other room. -State is two maps in memory (`room -> peer -> window end`, and follow times per `room|peer`); nothing is stored, and a restart only shortens windows. A tag or reply to her keeps `tag` or `reply`. A mouth `send` writes a plain `m.text` with no `m.mentions` or reply relation, so a post opens an engagement only when the event that comes back through /sync carries them (a post from another client, or a send that gains them). +State is in memory only (per-room anchor, sender ring, follow times), purged on every sync pass; nothing is stored, and a restart only shortens windows. ### Reply link (`chat_pending.v4`) diff --git a/tests/persona-matrix-mouth-follow.test.mjs b/tests/persona-matrix-mouth-follow.test.mjs index a541c5e..aa2ae1a 100644 --- a/tests/persona-matrix-mouth-follow.test.mjs +++ b/tests/persona-matrix-mouth-follow.test.mjs @@ -1,5 +1,5 @@ -// Follow (docs/bridge.md, "Follow"): an untagged message from someone she tagged or replied to, within W of her own post, is basis "follow", -// at most K an hour per (peer, room), public rooms only, and only with the mouth's follow flag. The homeserver is an in-process fetch port: +// Follow (docs/bridge.md, "Follow"): her own post in a public room anchors it for W; those who spoke there in the W before it (and whoever it tagged or +// replied to) are counted, and their later messages are basis "follow", at most K an hour per (peer, room), and only with the mouth's follow flag. The homeserver is an in-process fetch port: // a scripted /sync (one reply per round), the room state and the events a reply target is read from. import test from "node:test"; import assert from "node:assert/strict"; @@ -14,13 +14,14 @@ const OTHER_ROOM = "!side:gamecult.test"; const ME = `@eureka:${HS}`; const KIKO = `@kiko:${HS}`; const STRANGER = `@stranger:${HS}`; +const LENA = `@lena:${HS}`; const PLATFORM = ["disc", "ord"].join(""); const SEC = 1000, W = 600 * SEC, T0 = 1_000_000_000_000; const msg = (id, sender, ts, content = {}) => ({ type: "m.room.message", event_id: id, sender, origin_server_ts: ts, content: { msgtype: "m.text", body: `line ${id}`, ...content } }); const tagKiko = { "m.mentions": { user_ids: [KIKO] } }; // rounds: sync replies after the baseline, each { [roomId]: events }. state: room id -> state events. -function world({ rounds, state = {}, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { +function world({ rounds, state = {}, stateStatus = {}, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { const seen = []; const known = new Map(); let round = 0; @@ -35,7 +36,8 @@ function world({ rounds, state = {}, follow = true, rooms = [ALIAS, OTHER_ALIAS] if (p.startsWith("/directory/room/")) return reply(200, { room_id: { [ALIAS]: ROOM, [OTHER_ALIAS]: OTHER_ROOM }[p.slice(16)] }); if (p.startsWith("/presence/") || p.includes("/receipt/")) return reply(200, {}); const st = p.match(/^\/rooms\/([^/]+)\/state$/); - if (st) return reply(200, state[st[1]] ?? []); + if (st) return stateStatus[st[1]] ? reply(stateStatus[st[1]], { errcode: "M_UNKNOWN" }) : reply(200, state[st[1]] ?? []); + if (init.method === "PUT" && p.includes("/send/m.room.message/")) return reply(200, { event_id: "$sent1" }); const ev = p.match(/^\/rooms\/([^/]+)\/event\/(.+)$/); if (ev) return known.has(ev[2]) ? reply(200, known.get(ev[2])) : reply(404, { errcode: "M_NOT_FOUND" }); if (p === "/sync") { @@ -52,91 +54,144 @@ function world({ rounds, state = {}, follow = true, rooms = [ALIAS, OTHER_ALIAS] persona: "eureka", homeserver: "http://hs.test", serverName: HS, rooms, token: "tok", fetch: hsFetch, follow, state: { load: async () => null, save: async () => {} }, brake: async () => true, log: () => {}, syncTimeoutMs: 1, now: () => clock, }); - return { mouth, seen, at: t => { clock = t; }, basisOfRound: async () => (await mouth.syncOnce()).map(d => [d.eventId, d.basis]) }; + return { mouth, seen, clock: () => clock, at: t => { clock = t; }, basisOfRound: async () => (await mouth.syncOnce()).map(d => [d.eventId, d.basis]) }; } const run = async (w, n) => { await w.mouth.syncOnce(); const out = []; for (let i = 0; i < n; i++) out.push(await w.basisOfRound()); return out; }; -test("follow: a peer she tagged, untagged and within W of her post, is follow; a stranger and a peer elsewhere are not", async () => { - const w = world({ rounds: [ - { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, - { [ROOM]: [msg("$kiko", KIKO, T0 + 60 * SEC), msg("$stranger", STRANGER, T0 + 61 * SEC)], [OTHER_ROOM]: [msg("$kikoside", KIKO, T0 + 62 * SEC)] }, - ] }); - const [, second] = await run(w, 2); - assert.deepEqual(second.sort(), [["$kiko", "follow"], ["$kikoside", null], ["$stranger", null]]); -}); +const go = async (w, t) => { w.at(t); return w.basisOfRound(); }; // the process clock moves with the round +const mine = (id, ts, content = {}) => msg(id, ME, ts, content); +const MIN = 60 * SEC; -test("follow: a reply of hers makes the author of the quoted event a peer", async () => { +test("follow: a counted sender is one who spoke in the 10 minutes before her post; 9 minutes counts, 11 does not; one who first speaks after does not", async () => { const w = world({ rounds: [ - { [ROOM]: [msg("$theirs", KIKO, T0 - 5 * SEC)] }, - { [ROOM]: [msg("$mine", ME, T0, { "m.relates_to": { "m.in_reply_to": { event_id: "$theirs" } } })] }, - { [ROOM]: [msg("$next", KIKO, T0 + 30 * SEC)] }, + { [ROOM]: [msg("$old", KIKO, T0 - 11 * MIN), msg("$edge", `@edge:${HS}`, T0 - 10.5 * MIN), msg("$recent", LENA, T0 - 9 * MIN)] }, + { [ROOM]: [mine("$mine", T0)] }, + { [ROOM]: [msg("$k", KIKO, T0 + MIN), msg("$l", LENA, T0 + 2 * MIN), msg("$s", STRANGER, T0 + 3 * MIN), msg("$e", `@edge:${HS}`, T0 + 3 * MIN)] }, ] }); const out = await run(w, 3); - assert.deepEqual(out[2], [["$next", "follow"]]); + assert.deepEqual(out[2].sort(), [["$e", null], ["$k", null], ["$l", "follow"], ["$s", null]]); }); -test("follow: the chain dies when she stops speaking; a follow does not extend W; speaking again re-anchors", async () => { +test("follow: a post's tag and reply target are counted though they never spoke; she herself never is", async () => { const w = world({ rounds: [ - { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, - { [ROOM]: [msg("$a", KIKO, T0 + 9 * 60 * SEC)] }, // inside W: follow - { [ROOM]: [msg("$b", KIKO, T0 + W + SEC)] }, // after W from her post, though a follow came at 9 min: not follow - { [ROOM]: [msg("$again", ME, T0 + 11 * 60 * SEC, tagKiko)] }, - { [ROOM]: [msg("$c", KIKO, T0 + 20 * 60 * SEC)] }, // inside W of the second post: follow + { [ROOM]: [msg("$theirs", LENA, T0 - 5 * MIN)] }, + { [ROOM]: [mine("$mine", T0, { "m.mentions": { user_ids: [KIKO, ME] }, "m.relates_to": { "m.in_reply_to": { event_id: "$theirs" } } })] }, + { [ROOM]: [msg("$k", KIKO, T0 + MIN), msg("$s", STRANGER, T0 + MIN)] }, ] }); - const out = await run(w, 5); - assert.deepEqual(out.flat(), [["$a", "follow"], ["$b", null], ["$c", "follow"]]); + assert.deepEqual((await run(w, 3))[2], [["$k", "follow"], ["$s", null]]); }); -test("follow: a tag or reply to her keeps its own basis", async () => { + +test("follow: the chain dies when she stops speaking; a follow does not extend W; speaking again re-anchors and re-snapshots", async () => { const w = world({ rounds: [ - { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, - { [ROOM]: [msg("$t", KIKO, T0 + SEC, { "m.mentions": { user_ids: [ME] } })] }, + { [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, + { [ROOM]: [mine("$mine", T0)] }, + { [ROOM]: [msg("$a", KIKO, T0 + 9 * MIN)] }, // inside W: follow + { [ROOM]: [msg("$b", KIKO, T0 + 10 * MIN + SEC)] }, // past W from her post though a follow came at 9 min + { [ROOM]: [mine("$again", T0 + 11 * MIN)] }, + { [ROOM]: [msg("$c", KIKO, T0 + 20 * MIN), msg("$d", LENA, T0 + 20 * MIN)] }, ] }); - assert.deepEqual((await run(w, 2))[1], [["$t", "tag"]]); + await w.mouth.syncOnce(); + assert.deepEqual(await go(w, T0 - MIN), [["$k0", null]]); + assert.deepEqual(await go(w, T0), []); + assert.deepEqual(await go(w, T0 + 9 * MIN), [["$a", "follow"]]); + assert.deepEqual(await go(w, T0 + 10 * MIN + SEC), [["$b", null]]); + assert.deepEqual(await go(w, T0 + 11 * MIN), []); + assert.deepEqual(await go(w, T0 + 20 * MIN), [["$c", "follow"], ["$d", null]]); // lena never spoke before the second post either }); -test("follow: a DM, an encrypted room and a bridged room open nothing and receive nothing", async () => { - for (const state of [[{ type: "m.room.encryption", content: {} }], [{ type: "m.bridge", content: {} }], [{ type: "uk.half-shot.bridge", content: {} }]]) { - const w = world({ state: { [ROOM]: state }, rounds: [ - { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, - { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }, - ] }); - assert.deepEqual((await run(w, 2))[1], [["$kiko", null]]); +test("follow: the window needs this process's clock too: a message stamped inside it but read long after is not a follow", async () => { + const w = world({ rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$forged", KIKO, T0 + MIN)] }] }); + await w.mouth.syncOnce(); + await go(w, T0 - MIN); await go(w, T0); + assert.deepEqual(await go(w, T0 + 30 * MIN), [["$forged", null]]); +}); + +test("follow: her mouth send anchors at send time, and its /sync echo does not anchor again", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, + { [ROOM]: [msg("$sent1", ME, T0 + 8 * MIN, { body: "echo" })] }, + { [ROOM]: [msg("$late", KIKO, T0 + 12 * MIN)] }, + ] }); + await w.mouth.syncOnce(); await w.basisOfRound(); + const sent = await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + assert.equal(sent.refusal, null); + w.at(T0 + 8 * MIN); + await w.basisOfRound(); // the echo, 8 minutes on: a second anchor would carry the window to 18 minutes + w.at(T0 + 12 * MIN); + assert.deepEqual(await w.basisOfRound(), [["$late", null]]); +}); + +test("follow: a send from the mouth anchors; a send into a DM or an unreadable room does not", async () => { + const mk = opts => world({ ...opts, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$k1", KIKO, T0 + MIN)] }] }); + const ok = mk({}); + await ok.mouth.syncOnce(); await ok.basisOfRound(); + await ok.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + assert.deepEqual(await ok.basisOfRound(), [["$k1", "follow"]]); + const dm = mk({ direct: { [KIKO]: [ROOM] } }); + await dm.mouth.syncOnce(); await dm.basisOfRound(); + await dm.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ROOM, text: "hello" }); + assert.deepEqual(await dm.basisOfRound(), [["$k1", "direct"]]); +}); + +test("follow: a DM, an encrypted room and a bridged room anchor nothing; an unreadable or non-array state is not public", async () => { + const rounds = () => [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$k", KIKO, T0 + MIN)] }]; + for (const state of [[{ type: "m.room.encryption", content: {} }], [{ type: "m.bridge", content: {} }], [{ type: "uk.half-shot.bridge", content: {} }], { not: "an array" }]) { + assert.deepEqual((await run(world({ state: { [ROOM]: state }, rounds: rounds() }), 3))[2], [["$k", null]]); } - const dm = world({ direct: { [KIKO]: [ROOM] }, rounds: [{ [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }] }); - assert.deepEqual((await run(dm, 2))[1], [["$kiko", "direct"]]); + assert.deepEqual((await run(world({ stateStatus: { [ROOM]: 500 }, rounds: rounds() }), 3))[2], [["$k", null]]); + assert.deepEqual((await run(world({ direct: { [KIKO]: [ROOM] }, rounds: rounds() }), 3))[2], [["$k", "direct"]]); + // plaintext and unbridged is public, whatever else it is (an invite-only room included): the state holds no join rule the mouth reads + assert.deepEqual((await run(world({ state: { [ROOM]: [{ type: "m.room.join_rules", content: { join_rule: "invite" } }] }, rounds: rounds() }), 3))[2], [["$k", "follow"]]); }); -test("follow: a bridged room, a bridged sender as peer, and a bridged event deliver no follow", async () => { - const w2 = world({ state: { [ROOM]: [{ type: "m.bridge", content: {} }] }, rounds: [{ [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }] }); - assert.deepEqual((await run(w2, 2))[1], [["$kiko", null]]); +test("follow: a bridged sender or event is neither counted nor followed", async () => { const puppet = `@_${PLATFORM}_123:${HS}`; - const w3 = world({ rounds: [ - { [ROOM]: [msg("$mine", ME, T0, { "m.mentions": { user_ids: [puppet] } })] }, - { [ROOM]: [msg("$p", puppet, T0 + SEC)] }, - ] }); - assert.deepEqual((await run(w3, 2))[1], [["$p", null]]); - const w4 = world({ rounds: [ - { [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, - { [ROOM]: [msg("$kiko", KIKO, T0 + SEC, { [`${PLATFORM}.relay`]: true })] }, - ] }); - assert.deepEqual((await run(w4, 2))[1], [["$kiko", null]]); + const w = world({ rounds: [{ [ROOM]: [msg("$p0", puppet, T0 - MIN), msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0, { "m.mentions": { user_ids: [puppet] } })] }, + { [ROOM]: [msg("$p", puppet, T0 + MIN), msg("$k", KIKO, T0 + MIN, { [`${PLATFORM}.relay`]: true })] }] }); + assert.deepEqual((await run(w, 3))[2].sort(), [["$k", null], ["$p", null]]); }); -test("follow: at most three an hour per peer and room; the fourth keeps its basis; an hour later it is follow again", async () => { - const posts = n => ({ [ROOM]: [msg(`$me${n}`, ME, T0 + n * 5 * 60 * SEC, tagKiko), msg(`$k${n}`, KIKO, T0 + n * 5 * 60 * SEC + SEC)] }); - const w = world({ rounds: [posts(0), posts(1), posts(2), posts(3), { [ROOM]: [msg("$me9", ME, T0 + 70 * 60 * SEC, tagKiko), msg("$k9", KIKO, T0 + 70 * 60 * SEC + SEC)] }] }); - const out = (await run(w, 5)).flat(); - assert.deepEqual(out, [["$k0", "follow"], ["$k1", "follow"], ["$k2", "follow"], ["$k3", null], ["$k9", "follow"]]); +test("follow: at most three an hour per room and peer, counted independently; a half hour does not reset it; an hour later it is follow again", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$k0", KIKO, T0 - SEC), msg("$l0", LENA, T0 - SEC)], [OTHER_ROOM]: [msg("$ko0", KIKO, T0 - SEC)] }, + { [ROOM]: [mine("$m1", T0 + MIN)], [OTHER_ROOM]: [msg("$mo1", ME, T0 + MIN)] }, + { [ROOM]: [msg("$k1", KIKO, T0 + MIN + SEC)], [OTHER_ROOM]: [msg("$ko1", KIKO, T0 + MIN + SEC)] }, + { [ROOM]: [mine("$m2", T0 + 2 * MIN)] }, { [ROOM]: [msg("$k2", KIKO, T0 + 2 * MIN + SEC)] }, + { [ROOM]: [mine("$m3", T0 + 3 * MIN)] }, { [ROOM]: [msg("$k3", KIKO, T0 + 3 * MIN + SEC)] }, + { [ROOM]: [mine("$m4", T0 + 4 * MIN)] }, { [ROOM]: [msg("$k4", KIKO, T0 + 4 * MIN + SEC), msg("$l4", LENA, T0 + 4 * MIN + SEC)] }, + { [ROOM]: [msg("$k30", KIKO, T0 + 30 * MIN), mine("$m31", T0 + 31 * MIN)] }, { [ROOM]: [msg("$k31", KIKO, T0 + 31 * MIN + SEC)] }, + { [ROOM]: [msg("$k60", KIKO, T0 + 61 * MIN), mine("$m62", T0 + 62 * MIN)] }, { [ROOM]: [msg("$k62", KIKO, T0 + 62 * MIN + SEC)] }, + ] }); + await w.mouth.syncOnce(); + const seen = []; + const steps = [0, 1, 1, 2, 2, 3, 3, 4, 4, 31, 31, 62, 62]; + for (const m of steps) seen.push(...await go(w, T0 + m * MIN + 2 * SEC)); + const got = Object.fromEntries(seen.filter(([id]) => /^\$(k|ko|l)\d/.test(id)).map(([id, b]) => [id, b])); + assert.deepEqual(got, { $k0: null, $l0: null, $ko0: null, $k1: "follow", $ko1: "follow", $k2: "follow", $k3: "follow", $k4: null, $l4: "follow", $k30: null, $k31: null, $k60: null, $k62: "follow" }); +}); + +test("follow: an edit and a notice are never follow; a room-basis message from a counted sender is; a tag keeps its own basis", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, + { [ROOM]: [mine("$mine", T0)] }, + { [ROOM]: [ + msg("$edit", KIKO, T0 + SEC, { "m.new_content": { msgtype: "m.text", body: "x" }, "m.relates_to": { rel_type: "m.replace", event_id: "$k0" } }), + msg("$notice", KIKO, T0 + SEC, { msgtype: "m.notice" }), + msg("$room", KIKO, T0 + SEC, { "m.mentions": { room: true } }), + msg("$tag", KIKO, T0 + SEC, { "m.mentions": { user_ids: [ME] } }), + ] }, + ] }); + assert.deepEqual((await run(w, 3))[2].sort(), [["$edit", null], ["$notice", null], ["$room", "follow"], ["$tag", "tag"]]); }); test("follow: flag off emits nothing new and reads nothing extra; a fresh mouth has no engagement", async () => { - const rounds = [{ [ROOM]: [msg("$mine", ME, T0, tagKiko)] }, { [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }]; + const rounds = [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$k", KIKO, T0 + MIN)] }]; const off = world({ follow: false, rounds }); - assert.deepEqual((await run(off, 2))[1], [["$kiko", null]]); + assert.deepEqual((await run(off, 3))[2], [["$k", null]]); assert.ok(!off.seen.some(l => l.endsWith("/state")), "off: no state read"); - const restarted = world({ rounds: [{ [ROOM]: [msg("$kiko", KIKO, T0 + SEC)] }] }); - assert.deepEqual((await run(restarted, 1))[0], [["$kiko", null]]); + const restarted = world({ rounds: [{ [ROOM]: [msg("$k", KIKO, T0 + MIN)] }] }); + assert.deepEqual((await run(restarted, 1))[0], [["$k", null]]); assert.equal(followFromEnv({ CHAT_FOLLOW: "1" }), true); for (const v of [undefined, "0", "true", "yes", ""]) assert.equal(followFromEnv({ CHAT_FOLLOW: v }), false); }); diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 647e4e8..67346bc 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -23,9 +23,10 @@ // reads the room's head from Matrix and answers RoomMoved, naming the messages that moved it, instead // of the PUT. A v1 send, and a v2 send with no basis, are not checked; both still pass through the gate. // -// Follow (CHAT_FOLLOW=1, default off; docs/bridge.md "Follow"): her own post that tags or replies to someone opens a 10 minute engagement -// with that peer in that public room; the peer's next untagged messages inside it carry basis "follow", at most 3 an hour per peer and room. -// Memory only. Off, no basis "follow" is ever emitted: Huginn's parser refuses it until its Basis::Follow lands. +// Follow (CHAT_FOLLOW=1, default off; docs/bridge.md "Follow"): her own send in a public room (not a DM, not encrypted, not bridged) anchors the room +// for 10 minutes; those who spoke there in the 10 minutes before it, and whoever it tagged or replied to, get basis "follow" on their later messages +// inside the window, at most 3 an hour per peer and room. Memory only. Off, no basis "follow" is ever emitted: Huginn's parser refuses it until its +// Basis::Follow lands. // // Media: an m.image/m.file/m.audio/m.video event is delivered as one typed line (see bodyText), not its bare filename. // An m.image event in a delivered message is also fetched, as this Persona and only when an image store is configured @@ -75,6 +76,9 @@ const PLATFORM = ["disc", "ord"].join(""); // the bridged platform's name, as Hu const FOLLOW_WINDOW_MS = 10 * 60 * 1000; // W: an engagement lasts this long from the Mind's own post; a follow never extends it const FOLLOW_PER_HOUR = 3; // K: follow bases given per (Mind, peer, room) in an hour; past it the message keeps the basis it had const FOLLOW_ROOM_TTL_MS = 5 * 60 * 1000; // how long a room's public verdict (not encrypted, not bridged) is kept +const FOLLOW_SLACK_MS = 60 * 1000; // clock disagreement tolerated between this process and the homeserver +const FOLLOW_HOUR_MS = 60 * 60 * 1000; +const FOLLOW_RING = 200; // recent senders kept per room 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. @@ -311,14 +315,19 @@ export function createMouth({ return lines; } - // Follow (docs/bridge.md, "Follow"). Memory only: room id -> peer id -> the time her engagement with that peer ends, and room id|peer id -> - // the times follows were given. A restart empties both, which only shortens windows. Opened by her own post (seen in /sync, so a post from - // any client of her account counts) that tags or replies to a peer; never by a follow delivery; never in a DM, an encrypted room or a bridged room. + // Follow (docs/bridge.md, "Follow"). Memory only, bounded, purged on every sync pass. Her own send in a public room (not a DM, not encrypted, not + // bridged) anchors that room for W from the send: `engagements` room id -> { from, ends, counted }, counted being the senders who spoke there in the + // W before the send (`spoke`, a per-room ring of recent senders with event times) plus whoever the post tagged or replied to, if the event says so. + // Someone who first speaks after the send is not counted. Only her speaking again re-anchors (and re-snapshots); a follow never extends the window. + // `followsGiven` room|peer -> times follows were given. A restart empties all of it, which only shortens windows. const engagements = new Map(); + const spoke = new Map(); const followsGiven = new Map(); + const anchored = new Set(); // event ids of her own posts already anchored: the send path and its /sync echo anchor once const publicRooms = new Map(); // room id -> { ok, at } const platformId = id => platformEvent({ sender: id }); async function publicRoom(roomId) { + if (directRooms.has(roomId)) return false; const known = publicRooms.get(roomId); if (known && now() - known.at < FOLLOW_ROOM_TTL_MS) return known.ok; try { @@ -333,32 +342,53 @@ export function createMouth({ } } const eventAt = event => Number.isSafeInteger(event.origin_server_ts) && event.origin_server_ts >= 0 ? event.origin_server_ts : now(); - // Her own post: whoever it tags (m.mentions) or replies to becomes a peer for W from this post, re-anchoring one she already had. - async function noteOwnPost(roomId, event) { - if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event)) return; - const peers = new Set(Array.isArray(event.content["m.mentions"]?.user_ids) ? event.content["m.mentions"].user_ids : []); - const relates = event.content["m.relates_to"]; - const replied = relates?.["m.in_reply_to"]?.event_id; - if (typeof replied === "string" && replied !== "" && relates.is_falling_back !== true) { - const author = await authorOf(roomId, replied); - if (author !== null) peers.add(author); + function purgeFollow() { + const t = now(); + for (const [room, e] of engagements) if (e.ends + FOLLOW_SLACK_MS <= t) engagements.delete(room); + for (const [room, ring] of spoke) { + const kept = ring.filter(r => r.at > t - FOLLOW_WINDOW_MS - FOLLOW_SLACK_MS); + if (kept.length === 0) spoke.delete(room); else spoke.set(room, kept); } - const ends = eventAt(event) + FOLLOW_WINDOW_MS; - const ids = [...peers].filter(id => docs.isMatrixId("@", id) && id !== selfId && !platformId(id)).slice(0, docs.limits.mentions); - if (ids.length === 0 || ends <= now() || !(await publicRoom(roomId))) return; - const room = engagements.get(roomId) ?? new Map(); - for (const [id, until] of room) if (until <= now()) room.delete(id); - for (const id of ids) room.set(id, ends); - engagements.set(roomId, room); + for (const [key, times] of followsGiven) { + const kept = times.filter(x => t - x < FOLLOW_HOUR_MS + FOLLOW_SLACK_MS); + if (kept.length === 0) followsGiven.delete(key); else followsGiven.set(key, kept); + } + } + // Someone spoke in the room (a message from another account, not an edit or a notice): the ring behind the snapshot. + function noteSpeaker(roomId, event) { + if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event) || event.sender === selfId) return; + const ring = spoke.get(roomId) ?? []; + ring.push({ sender: event.sender, at: eventAt(event) }); + if (ring.length > FOLLOW_RING) ring.shift(); + spoke.set(roomId, ring); + } + // Her post at time `at` (the send's own time, or the echo's origin time): snapshot the room. `tagged`: ids from the event's m.mentions, `replyTo` the quoted event id. + async function anchor(roomId, eventId, at, tagged = [], replyTo = null) { + if (!follow || anchored.has(eventId) || !(await publicRoom(roomId))) return; + anchored.add(eventId); + if (anchored.size > 200) anchored.delete(anchored.values().next().value); + const counted = new Set((spoke.get(roomId) ?? []).filter(r => r.at >= at - FOLLOW_WINDOW_MS && r.at <= at).map(r => r.sender)); + for (const id of tagged) counted.add(id); + if (typeof replyTo === "string" && replyTo !== "") { const author = await authorOf(roomId, replyTo); if (author !== null) counted.add(author); } + for (const id of [...counted]) if (!docs.isMatrixId("@", id) || id === selfId || platformId(id)) counted.delete(id); + engagements.set(roomId, { from: at, ends: at + FOLLOW_WINDOW_MS, counted }); + } + // Her own event seen in /sync (a send from another client, or the echo of a mouth send, which is then not anchored twice). + async function noteOwnPost(roomId, event) { + if (!follow || formOf(event.content) !== "message" || platformEvent(event)) return; + const relates = event.content["m.relates_to"]; + const tagged = Array.isArray(event.content["m.mentions"]?.user_ids) ? event.content["m.mentions"].user_ids : []; + await anchor(roomId, event.event_id, eventAt(event), tagged.slice(0, docs.limits.mentions), relates?.is_falling_back === true ? null : relates?.["m.in_reply_to"]?.event_id); } // Whether this message is a follow, and if so spends one of the peer's K for the hour. Called only for a message that has no better basis. async function followed(roomId, event) { if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event)) return false; - const at = eventAt(event); - if (!((engagements.get(roomId)?.get(event.sender) ?? 0) > at)) return false; + const e = engagements.get(roomId), at = eventAt(event), t = now(); + // Both clocks must agree: the message's own time inside the window, and this process's time not past it (a forged or old timestamp reuses nothing). + if (!e || !e.counted.has(event.sender) || at < e.from || at >= e.ends || t >= e.ends + FOLLOW_SLACK_MS) return false; if (!(await publicRoom(roomId))) return false; const key = `${roomId}|${event.sender}`; - const recent = (followsGiven.get(key) ?? []).filter(t => at - t < 60 * 60 * 1000); + const recent = (followsGiven.get(key) ?? []).filter(x => at - x < FOLLOW_HOUR_MS); if (recent.length >= FOLLOW_PER_HOUR) { followsGiven.set(key, recent); return false; } followsGiven.set(key, [...recent, at]); return true; @@ -501,6 +531,7 @@ export function createMouth({ // One /sync round. Returns the deliveries it made. async function syncOnce() { if (!started || unjoined()) await start(); + purgeFollow(); replyReads = 0; held.clear(); const free = await released(); // the round's brake read for presence, taken before any early return @@ -535,7 +566,8 @@ export function createMouth({ const deliverable = !baseline && !freshRooms.has(roomId) && event.sender !== selfId; if (deliverable && pending.length + made.length >= maxPending) { truncated = true; break rounds; } if (deliverable && replyLink && replyReads + targetsToRead(roomId, event) > REPLY_READS_PER_ROUND) { say(`event ${event.event_id} not read: round cap on target reads, the rest of the round waits for the next`); truncated = true; break rounds; } // not processed: the next round meets it again, with its read count reset - if (!baseline && !freshRooms.has(roomId) && event.sender === selfId && event.type === "m.room.message") await noteOwnPost(roomId, event); + if (!baseline && !freshRooms.has(roomId) && event.sender === selfId) await noteOwnPost(roomId, event); + noteSpeaker(roomId, event); let basis = deliverable ? await basisOf(roomId, event) : null; if (deliverable && (basis === null || basis === "room") && await followed(roomId, event)) basis = "follow"; const replyTo = deliverable ? await replyToOf(roomId, event) : null; @@ -710,7 +742,7 @@ export function createMouth({ const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }, API, AbortSignal.timeout(sendTimeoutMs)); return { requestId, eventId: sent.event_id, refusal: null, newer: [] }; }); - if (result.refusal === null) { touch(); setPresence(true).catch(() => {}); } // outside the room gate and not awaited: a slow presence write holds neither another send nor this receipt + if (result.refusal === null) { touch(); setPresence(true).catch(() => {}); await anchor(roomId, result.eventId, now()).catch(error => say(`follow: send not anchored: ${error.message}`)); } // outside the room gate and not awaited: a slow presence write holds neither another send nor this receipt return result; } catch (error) { say(`send failed: ${error.message}`); -- 2.47.3 From f66b222ea87e461553efe965c84b9b7dae2221be Mon Sep 17 00:00:00 2001 From: Heimdall Date: Thu, 12 Nov 2026 12:05:51 +0000 Subject: [PATCH 3/5] Follow: anchor off the send path, clock slack on the early side, mutation-proof tests Co-Authored-By: Claude Sonnet 5.5 --- docs/bridge.md | 6 +- tests/persona-matrix-mouth-follow.test.mjs | 133 ++++++++++++++++++++- tools/persona-matrix-mouth.mjs | 16 ++- 3 files changed, 145 insertions(+), 10 deletions(-) diff --git a/docs/bridge.md b/docs/bridge.md index 1341260..5872f22 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -259,9 +259,9 @@ A message from someone she is in conversation with is for her even when it names With it on: -- **Anchor.** Her own post in a public room anchors that room for W = 600 s from the post. The mouth anchors at send time (the send's own time, in `send`) and also from the /sync echo of her own events, so a post from another client of her account anchors too; one event anchors once. Only her posting again re-anchors the room and re-snapshots who is counted; a follow never extends the window. -- **Counted senders** (snapshotted at the post): those who spoke in that room in the 10 minutes before the post (a per-room ring of recent senders with event times, 200 entries), plus anyone the post tagged (`m.mentions.user_ids`) or replied to (the author of the quoted `m.in_reply_to`, not a thread fallback pointer) when the event carries it. Someone who first speaks after her post is not counted. Nothing reads message text; no ids are scraped from bodies. She herself, and bridged senders (the `platformEvent` identifier rule), never count. -- **Follow.** A `message` (not an edit, not a notice) from a counted sender gets `follow` when it would otherwise have no basis or `room`; `tag`, `reply` and `direct` are never replaced. Its timestamp must lie inside the window and this process's clock must not be past the window's end by more than 60 s, so an old or forged `origin_server_ts` reuses nothing. At most K = 3 per (Mind, peer, room) in the hour before the message; past it the message keeps the basis it had. +- **Anchor.** Her own post in a public room anchors that room for W = 600 s from the post. The mouth anchors at send time (the send's own time, in `send`, in the background: the send receipt never waits on the room-state read) and also from the /sync echo of her own events, so a post from another client of her account anchors too; one event anchors once. Only her posting again re-anchors the room and re-snapshots who is counted; a follow never extends the window. +- **Counted senders** (snapshotted at the post): those who spoke in that room in the 10 minutes before the post (a per-room ring of recent senders with event times, 200 entries), plus anyone the post tagged (`m.mentions.user_ids`) or replied to (the author of the quoted `m.in_reply_to`, not a thread fallback pointer) when the event carries it. Someone who first speaks after her post is not counted. A member who spoke in the 10 minutes before her post is counted, bounded by 3 follows per hour per (peer, room). Nothing reads message text; no ids are scraped from bodies. She herself, and bridged senders (the `platformEvent` identifier rule), never count. +- **Follow.** A `message` (not an edit, not a notice) from a counted sender gets `follow` when it would otherwise have no basis or `room`; `tag`, `reply` and `direct` are never replaced. Its timestamp must lie inside the window (a homeserver clock up to 60 s behind still follows) and this process's clock must not be past the window's end by more than 60 s, so an old or forged `origin_server_ts` reuses nothing. At most K = 3 per (Mind, peer, room) in the hour before the message; past it the message keeps the basis it had. - **"Public" means non-DM, non-encrypted, non-bridged.** The room is not one of her DMs and its state has no `m.room.encryption`, `m.bridge` or `uk.half-shot.bridge` (read once per room, kept 5 minutes; unreadable or malformed state is not public). An invite-only plaintext room counts as public. Nothing is anchored in, and no follow is delivered in, any other room. State is in memory only (per-room anchor, sender ring, follow times), purged on every sync pass; nothing is stored, and a restart only shortens windows. diff --git a/tests/persona-matrix-mouth-follow.test.mjs b/tests/persona-matrix-mouth-follow.test.mjs index aa2ae1a..7e82dd7 100644 --- a/tests/persona-matrix-mouth-follow.test.mjs +++ b/tests/persona-matrix-mouth-follow.test.mjs @@ -21,7 +21,7 @@ const msg = (id, sender, ts, content = {}) => ({ type: "m.room.message", event_i const tagKiko = { "m.mentions": { user_ids: [KIKO] } }; // rounds: sync replies after the baseline, each { [roomId]: events }. state: room id -> state events. -function world({ rounds, state = {}, stateStatus = {}, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { +function world({ rounds, state = {}, stateStatus = {}, hangState = false, baselineEvents = null, onSync = null, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { const seen = []; const known = new Map(); let round = 0; @@ -36,12 +36,14 @@ function world({ rounds, state = {}, stateStatus = {}, follow = true, rooms = [A if (p.startsWith("/directory/room/")) return reply(200, { room_id: { [ALIAS]: ROOM, [OTHER_ALIAS]: OTHER_ROOM }[p.slice(16)] }); if (p.startsWith("/presence/") || p.includes("/receipt/")) return reply(200, {}); const st = p.match(/^\/rooms\/([^/]+)\/state$/); + if (st && hangState) return new Promise(() => {}); if (st) return stateStatus[st[1]] ? reply(stateStatus[st[1]], { errcode: "M_UNKNOWN" }) : reply(200, state[st[1]] ?? []); if (init.method === "PUT" && p.includes("/send/m.room.message/")) return reply(200, { event_id: "$sent1" }); const ev = p.match(/^\/rooms\/([^/]+)\/event\/(.+)$/); if (ev) return known.has(ev[2]) ? reply(200, known.get(ev[2])) : reply(404, { errcode: "M_NOT_FOUND" }); if (p === "/sync") { - if (u.searchParams.get("since") === null) return reply(200, { next_batch: "b0" }); + if (u.searchParams.get("since") === null) return reply(200, { next_batch: "b0", ...(baselineEvents ? { rooms: { join: Object.fromEntries(Object.entries(baselineEvents).map(([r, list]) => [r, { timeline: { events: list } }])) } } : {}) }); + onSync?.(setClock); const events = rounds[round++] ?? {}; for (const list of Object.values(events)) for (const e of list) known.set(e.event_id, e); return reply(200, { next_batch: `b${round}`, rooms: { join: Object.fromEntries(Object.entries(events).map(([r, list]) => [r, { timeline: { events: list } }])) } }); @@ -50,6 +52,7 @@ function world({ rounds, state = {}, stateStatus = {}, follow = true, rooms = [A return reply(404, { errcode: "M_NOT_FOUND" }); }; let clock = T0; + const setClock = t => { clock = t; }; const mouth = createMouth({ persona: "eureka", homeserver: "http://hs.test", serverName: HS, rooms, token: "tok", fetch: hsFetch, follow, state: { load: async () => null, save: async () => {} }, brake: async () => true, log: () => {}, syncTimeoutMs: 1, now: () => clock, @@ -201,3 +204,129 @@ test("follow: the basis is a name the delivery document accepts", () => { const d = { seq: 1, roomAlias: ALIAS, roomId: ROOM, eventId: "$e", sender: KIKO, senderDisplay: "kiko", text: "x", atUnixMs: 1, context: [], image: null, mentions: [], isDirect: false, basis: "follow", form: "message" }; assert.equal(docs.parseDeliveryV3(docs.deliveryV3Tuple(d)).basis, "follow"); }); + +const tick = async () => { for (let i = 0; i < 20; i++) await new Promise(r => setImmediate(r)); }; + +test("follow: a send receipt is not delayed by a room-state read that hangs or fails; the anchor happens in the background", async () => { + for (const opts of [{ hangState: true }, { stateStatus: { [ROOM]: 500 } }]) { + const w = world({ ...opts, rounds: [] }); + await w.mouth.syncOnce(); + const sent = await Promise.race([w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }), new Promise(r => setTimeout(() => r("late"), 500))]); + assert.equal(sent.refusal, null); + } + const ok = world({ rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$k1", KIKO, T0 + MIN)] }] }); + await ok.mouth.syncOnce(); await ok.basisOfRound(); + await ok.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + await tick(); + assert.deepEqual(await ok.basisOfRound(), [["$k1", "follow"]]); +}); + +test("follow: the flag off makes a send cost no extra call and no extra turn", async () => { + const w = world({ follow: false, rounds: [] }); + await w.mouth.syncOnce(); + const before = w.seen.length; + await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + await tick(); + assert.ok(!w.seen.slice(before).some(l => l.endsWith("/state"))); +}); + +test("follow: a homeserver clock up to 60 s behind still follows; 61 s early does not; the very end of the window is out, 1 ms before it is in", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$k0", KIKO, T0 - MIN), msg("$l0", LENA, T0 - MIN), msg("$e0", `@e:${HS}`, T0 - MIN), msg("$f0", `@f:${HS}`, T0 - MIN)] }, + { [ROOM]: [mine("$mine", T0)] }, + { [ROOM]: [msg("$x1", KIKO, T0 - 60 * SEC), msg("$x2", LENA, T0 - 61 * SEC), msg("$x3", `@e:${HS}`, T0 + W - 1), msg("$x4", `@f:${HS}`, T0 + W)] }, + ] }); + await w.mouth.syncOnce(); await go(w, T0 - MIN); await go(w, T0); + assert.deepEqual((await go(w, T0 + 5 * SEC)).sort(), [["$x1", "follow"], ["$x2", null], ["$x3", "follow"], ["$x4", null]]); +}); + +test("follow: a window that ended while the pass was being read is out, whatever the message says", async () => { + let jump = null; + const w = world({ onSync: set => { if (jump !== null) set(jump); }, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$x", KIKO, T0 + MIN)] }] }); + await w.mouth.syncOnce(); await go(w, T0 - MIN); + const keep = w.mouth.followState; + w.at(T0); await w.basisOfRound(); // her post; the next sync moves the clock after the purge + assert.equal(keep().engagements.length, 1); + jump = T0 + 30 * MIN; w.at(T0); // the purge at the start of the pass sees T0; the clock moves while the pass reads + assert.deepEqual(await w.basisOfRound(), [["$x", null]]); +}); + +test("follow: the cap counts only the hour before the message", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$k0", KIKO, T0 - SEC)] }, + ...[1, 2, 3].flatMap(n => [{ [ROOM]: [mine(`$m${n}`, T0 + n * MIN)] }, { [ROOM]: [msg(`$x${n}`, KIKO, T0 + n * MIN + SEC)] }]), + { [ROOM]: [msg("$k90", KIKO, T0 + 89 * MIN), mine("$m90", T0 + 90 * MIN)] }, { [ROOM]: [msg("$x90", KIKO, T0 + 90 * MIN + SEC)] }, + ] }); + await w.mouth.syncOnce(); await go(w, T0); + for (const n of [1, 1, 2, 2, 3, 3]) await go(w, T0 + n * MIN + SEC); + assert.deepEqual(await go(w, T0 + 30 * MIN), [["$k90", null]]); // stamps run ahead of this process's clock, so only the cap's own hour bound can tell + assert.deepEqual(await go(w, T0 + 30 * MIN), [["$x90", "follow"]]); +}); + +test("follow: whoever her reply quotes is counted though they spoke long ago; a thread fallback pointer names no one", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$old", LENA, T0 - 30 * MIN), msg("$ghost", `@ghost:${HS}`, T0 - 30 * MIN)] }, + { [ROOM]: [mine("$mine", T0, { "m.relates_to": { "m.in_reply_to": { event_id: "$old" } } })] }, + { [ROOM]: [msg("$x1", LENA, T0 + MIN)] }, + { [ROOM]: [msg("$old2", KIKO, T0 + 2 * MIN)] }, + { [ROOM]: [mine("$mine2", T0 + 3 * MIN, { "m.relates_to": { rel_type: "m.thread", event_id: "$old", is_falling_back: true, "m.in_reply_to": { event_id: "$ghost" } } })] }, + { [ROOM]: [msg("$x2", STRANGER, T0 + 4 * MIN)] }, + ] }); + await w.mouth.syncOnce(); + await go(w, T0 - 30 * MIN); await go(w, T0); + assert.deepEqual(await go(w, T0 + MIN), [["$x1", "follow"]]); + await go(w, T0 + 2 * MIN); + await go(w, T0 + 3 * MIN); + const [engagement] = w.mouth.followState().engagements; + assert.ok(engagement.counted.includes(KIKO), "kiko spoke 1 minute before the second post"); + assert.ok(!engagement.counted.includes(`@ghost:${HS}`), "a fallback pointer names no one"); +}); + +test("follow: someone stamped after her send is not counted as having spoken before it", async () => { + const w = world({ rounds: [{ [ROOM]: [msg("$ahead", KIKO, T0 + MIN)] }, { [ROOM]: [msg("$x", KIKO, T0 + 2 * MIN)] }] }); + await w.mouth.syncOnce(); await w.basisOfRound(); + await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + await tick(); + assert.deepEqual(w.mouth.followState().engagements[0].counted, []); + assert.deepEqual(await go(w, T0 + 2 * MIN), [["$x", null]]); +}); + +test("follow: the ring keeps the newest 200 senders of a room; a flood pushes out the oldest, not the newest", async () => { + const flood = Array.from({ length: 250 }, (_, i) => msg(`$f${i}`, `@u${i}:${HS}`, T0 - 5 * MIN + i)); + const w = world({ rounds: [{ [ROOM]: flood }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$x0", `@u0:${HS}`, T0 + MIN), msg("$x249", `@u249:${HS}`, T0 + MIN)] }] }); + await w.mouth.syncOnce(); await w.basisOfRound(); + assert.equal(w.mouth.followState().spoke[0].senders.length, 200); + await go(w, T0); + assert.deepEqual(await go(w, T0 + MIN + SEC), [["$x0", null], ["$x249", "follow"]]); + assert.equal(w.mouth.followState().engagements[0].counted.length, 200); +}); + +test("follow: only senders, and only real ones, are remembered and counted: not her, not a bridged id, not an edit, not a notice, not a malformed id", async () => { + const puppet = `@_${PLATFORM}_9:${HS}`; + const w = world({ rounds: [ + { [ROOM]: [msg("$a", KIKO, T0 - 2 * MIN), mine("$self", T0 - 2 * MIN), msg("$b", puppet, T0 - 2 * MIN), + msg("$c", LENA, T0 - 2 * MIN, { "m.relates_to": { rel_type: "m.replace", event_id: "$a" } }), msg("$d", STRANGER, T0 - 2 * MIN, { msgtype: "m.notice" })] }, + { [ROOM]: [mine("$mine", T0, { "m.mentions": { user_ids: [ME, puppet, "not-an-id", `@${"x".repeat(300)}:${HS}`, LENA] } })] }, + ] }); + await w.mouth.syncOnce(); await go(w, T0 - MIN); + assert.deepEqual(w.mouth.followState().spoke, [{ room: ROOM, senders: [KIKO] }]); + await go(w, T0); + assert.deepEqual(w.mouth.followState().engagements[0].counted.sort(), [KIKO, LENA].sort()); +}); + +test("follow: her own post in the baseline sync, which is history, anchors nothing", async () => { + const w = world({ baselineEvents: { [ROOM]: [msg("$k0", KIKO, T0 - MIN), mine("$mine", T0)] }, rounds: [] }); + await w.mouth.syncOnce(); + assert.deepEqual(w.mouth.followState().engagements, []); +}); + +test("follow: every pass forgets what has aged out: anchors, speakers and cap times", async () => { + const w = world({ rounds: [ + { [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$x", KIKO, T0 + MIN)] }, {}, + ] }); + await w.mouth.syncOnce(); await go(w, T0 - MIN); await go(w, T0); await go(w, T0 + MIN + SEC); + const s1 = w.mouth.followState(); + assert.equal(s1.engagements.length, 1); assert.equal(s1.followsGiven, 1); assert.equal(s1.spoke.length, 1); + await go(w, T0 + 3 * 60 * MIN); + assert.deepEqual(w.mouth.followState(), { engagements: [], spoke: [], followsGiven: 0 }); +}); diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 67346bc..8f0fb73 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -364,9 +364,10 @@ export function createMouth({ } // Her post at time `at` (the send's own time, or the echo's origin time): snapshot the room. `tagged`: ids from the event's m.mentions, `replyTo` the quoted event id. async function anchor(roomId, eventId, at, tagged = [], replyTo = null) { - if (!follow || anchored.has(eventId) || !(await publicRoom(roomId))) return; - anchored.add(eventId); + if (!follow || anchored.has(eventId)) return; + anchored.add(eventId); // before the room read: the send path and its echo race, and one event anchors once if (anchored.size > 200) anchored.delete(anchored.values().next().value); + if (!(await publicRoom(roomId))) return; const counted = new Set((spoke.get(roomId) ?? []).filter(r => r.at >= at - FOLLOW_WINDOW_MS && r.at <= at).map(r => r.sender)); for (const id of tagged) counted.add(id); if (typeof replyTo === "string" && replyTo !== "") { const author = await authorOf(roomId, replyTo); if (author !== null) counted.add(author); } @@ -385,7 +386,7 @@ export function createMouth({ if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event)) return false; const e = engagements.get(roomId), at = eventAt(event), t = now(); // Both clocks must agree: the message's own time inside the window, and this process's time not past it (a forged or old timestamp reuses nothing). - if (!e || !e.counted.has(event.sender) || at < e.from || at >= e.ends || t >= e.ends + FOLLOW_SLACK_MS) return false; + if (!e || !e.counted.has(event.sender) || at < e.from - FOLLOW_SLACK_MS || at >= e.ends || t >= e.ends + FOLLOW_SLACK_MS) return false; if (!(await publicRoom(roomId))) return false; const key = `${roomId}|${event.sender}`; const recent = (followsGiven.get(key) ?? []).filter(x => at - x < FOLLOW_HOUR_MS); @@ -742,7 +743,10 @@ export function createMouth({ const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }, API, AbortSignal.timeout(sendTimeoutMs)); return { requestId, eventId: sent.event_id, refusal: null, newer: [] }; }); - if (result.refusal === null) { touch(); setPresence(true).catch(() => {}); await anchor(roomId, result.eventId, now()).catch(error => say(`follow: send not anchored: ${error.message}`)); } // outside the room gate and not awaited: a slow presence write holds neither another send nor this receipt + if (result.refusal === null) { + touch(); setPresence(true).catch(() => {}); // outside the room gate and not awaited: a slow presence write holds neither another send nor this receipt + if (follow) { const at = now(); anchor(roomId, result.eventId, at).catch(error => say(`follow: send not anchored: ${error.message}`)); } // in the background too: a room-state read that hangs or fails never delays the receipt + } return result; } catch (error) { say(`send failed: ${error.message}`); @@ -982,7 +986,9 @@ export function createMouth({ } } - return { start, syncOnce, ack, read, pendingAfter, tail, speakableRooms, send, openDirect, fetchAttachment, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } }; + // A plain copy of the follow memory, for tests: anchors with their counted senders, the sender ring, and how many cap entries are held. + const followState = () => ({ engagements: [...engagements].map(([room, e]) => ({ room, from: e.from, ends: e.ends, counted: [...e.counted] })), spoke: [...spoke].map(([room, ring]) => ({ room, senders: ring.map(r => r.sender) })), followsGiven: followsGiven.size }); + return { followState, start, syncOnce, ack, read, pendingAfter, tail, speakableRooms, send, openDirect, fetchAttachment, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } }; } // The log's word for a failure: error class and HTTP status only, never the message (it can hold a URL, an mxc or a sender-chosen id). -- 2.47.3 From 76cdf13e2a51bae5b15be5ab65306d9ad3372ee5 Mon Sep 17 00:00:00 2001 From: Heimdall Date: Thu, 12 Nov 2026 12:08:42 +0000 Subject: [PATCH 4/5] Follow: a failed room-state read leaves the post retryable; a no verdict stays recorded Co-Authored-By: Claude Sonnet 5.5 --- tests/persona-matrix-mouth-follow.test.mjs | 37 +++++++++++++++++++++- tools/persona-matrix-mouth.mjs | 34 +++++++++++++------- 2 files changed, 58 insertions(+), 13 deletions(-) diff --git a/tests/persona-matrix-mouth-follow.test.mjs b/tests/persona-matrix-mouth-follow.test.mjs index 7e82dd7..2109682 100644 --- a/tests/persona-matrix-mouth-follow.test.mjs +++ b/tests/persona-matrix-mouth-follow.test.mjs @@ -21,7 +21,7 @@ const msg = (id, sender, ts, content = {}) => ({ type: "m.room.message", event_i const tagKiko = { "m.mentions": { user_ids: [KIKO] } }; // rounds: sync replies after the baseline, each { [roomId]: events }. state: room id -> state events. -function world({ rounds, state = {}, stateStatus = {}, hangState = false, baselineEvents = null, onSync = null, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { +function world({ rounds, state = {}, stateStatus = {}, hangState = false, stateFailures = 0, baselineEvents = null, onSync = null, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { const seen = []; const known = new Map(); let round = 0; @@ -37,6 +37,7 @@ function world({ rounds, state = {}, stateStatus = {}, hangState = false, baseli if (p.startsWith("/presence/") || p.includes("/receipt/")) return reply(200, {}); const st = p.match(/^\/rooms\/([^/]+)\/state$/); if (st && hangState) return new Promise(() => {}); + if (st && stateFailures > 0) { stateFailures--; return reply(500, { errcode: "M_UNKNOWN" }); } if (st) return stateStatus[st[1]] ? reply(stateStatus[st[1]], { errcode: "M_UNKNOWN" }) : reply(200, state[st[1]] ?? []); if (init.method === "PUT" && p.includes("/send/m.room.message/")) return reply(200, { event_id: "$sent1" }); const ev = p.match(/^\/rooms\/([^/]+)\/event\/(.+)$/); @@ -330,3 +331,37 @@ test("follow: every pass forgets what has aged out: anchors, speakers and cap ti await go(w, T0 + 3 * 60 * MIN); assert.deepEqual(w.mouth.followState(), { engagements: [], spoke: [], followsGiven: 0 }); }); + +test("follow: a failed room-state read at send time leaves the post retryable: its echo anchors it", async () => { + const w = world({ stateFailures: 1, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0, { body: "hello" })] }, { [ROOM]: [msg("$x", KIKO, T0 + MIN)] }] }); + await w.mouth.syncOnce(); await w.basisOfRound(); + await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + await tick(); + assert.deepEqual(w.mouth.followState().engagements, [], "the read failed: nothing anchored yet"); + await go(w, T0); + assert.equal(w.mouth.followState().engagements.length, 1, "the echo anchored it"); + assert.deepEqual(await go(w, T0 + MIN), [["$x", "follow"]]); +}); + +test("follow: a send and its echo in flight together anchor once, and a failure of that one read leaves the echo to retry", async () => { + const w = world({ stateFailures: 1, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0, { body: "hello" })] }] }); + await w.mouth.syncOnce(); await w.basisOfRound(); + const sending = w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + await sending; + await w.basisOfRound(); // the echo, possibly while the send's read is still in flight + await tick(); + assert.equal(w.mouth.followState().engagements.length, 1); + assert.equal(w.seen.filter(l => l.endsWith("/state")).length, 2, "one failed read, one that worked; no third"); +}); + +test("follow: a not-public verdict is recorded: the room is not read again for each post or pass", async () => { + const w = world({ state: { [ROOM]: [{ type: "m.bridge", content: {} }] }, rounds: [ + { [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0)] }, { [ROOM]: [mine("$a", T0 + SEC), msg("$x", KIKO, T0 + 2 * SEC)] }, { [ROOM]: [mine("$b", T0 + 3 * SEC)] }, + ] }); + await w.mouth.syncOnce(); await w.basisOfRound(); + await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); + await tick(); + for (const t of [0, 0, 0]) await go(w, T0 + t); + assert.equal(w.seen.filter(l => l.endsWith("/state")).length, 1); + assert.deepEqual(w.mouth.followState().engagements, []); +}); diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 8f0fb73..77b45e2 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -324,21 +324,23 @@ export function createMouth({ const spoke = new Map(); const followsGiven = new Map(); const anchored = new Set(); // event ids of her own posts already anchored: the send path and its /sync echo anchor once - const publicRooms = new Map(); // room id -> { ok, at } + const anchoring = new Map(); // event ids whose anchor is in flight: the send path and its echo never snapshot twice + const publicRooms = new Map(); // room id -> { ok, at }: the verdicts, a "no" included, so a not-public room is read once per FOLLOW_ROOM_TTL_MS, not every pass const platformId = id => platformEvent({ sender: id }); + // true or false is a verdict (kept); null is a failed read: unknown, retryable, kept nowhere. async function publicRoom(roomId) { if (directRooms.has(roomId)) return false; const known = publicRooms.get(roomId); if (known && now() - known.at < FOLLOW_ROOM_TTL_MS) return known.ok; try { const state = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/state`, undefined, API, AbortSignal.timeout(sendTimeoutMs)); - if (!Array.isArray(state)) return false; + if (!Array.isArray(state)) return null; const ok = !state.some(e => e?.type === "m.room.encryption" || e?.type === "m.bridge" || e?.type === "uk.half-shot.bridge"); publicRooms.set(roomId, { ok, at: now() }); return ok; } catch (error) { say(`follow: room ${roomId} state not read: ${error.message}`); - return false; // unknown is not public + return null; // unknown is not public, and not a verdict either } } const eventAt = event => Number.isSafeInteger(event.origin_server_ts) && event.origin_server_ts >= 0 ? event.origin_server_ts : now(); @@ -364,15 +366,23 @@ export function createMouth({ } // Her post at time `at` (the send's own time, or the echo's origin time): snapshot the room. `tagged`: ids from the event's m.mentions, `replyTo` the quoted event id. async function anchor(roomId, eventId, at, tagged = [], replyTo = null) { - if (!follow || anchored.has(eventId)) return; - anchored.add(eventId); // before the room read: the send path and its echo race, and one event anchors once - if (anchored.size > 200) anchored.delete(anchored.values().next().value); - if (!(await publicRoom(roomId))) return; - const counted = new Set((spoke.get(roomId) ?? []).filter(r => r.at >= at - FOLLOW_WINDOW_MS && r.at <= at).map(r => r.sender)); - for (const id of tagged) counted.add(id); - if (typeof replyTo === "string" && replyTo !== "") { const author = await authorOf(roomId, replyTo); if (author !== null) counted.add(author); } - for (const id of [...counted]) if (!docs.isMatrixId("@", id) || id === selfId || platformId(id)) counted.delete(id); - engagements.set(roomId, { from: at, ends: at + FOLLOW_WINDOW_MS, counted }); + if (!follow) return; + while (anchoring.has(eventId)) await anchoring.get(eventId).catch(() => {}); // the send and its echo meet here: the second waits, then sees whether the first succeeded + if (anchored.has(eventId)) return; + const run = (async () => { + const verdict = await publicRoom(roomId); + if (verdict === null) return; // the read failed: this post stays retryable, its echo tries again + anchored.add(eventId); + if (anchored.size > 200) anchored.delete(anchored.values().next().value); + if (!verdict) return; // a "no" stays recorded + const counted = new Set((spoke.get(roomId) ?? []).filter(r => r.at >= at - FOLLOW_WINDOW_MS && r.at <= at).map(r => r.sender)); + for (const id of tagged) counted.add(id); + if (typeof replyTo === "string" && replyTo !== "") { const author = await authorOf(roomId, replyTo); if (author !== null) counted.add(author); } + for (const id of [...counted]) if (!docs.isMatrixId("@", id) || id === selfId || platformId(id)) counted.delete(id); + engagements.set(roomId, { from: at, ends: at + FOLLOW_WINDOW_MS, counted }); + })(); + anchoring.set(eventId, run); + try { await run; } finally { anchoring.delete(eventId); } } // Her own event seen in /sync (a send from another client, or the echo of a mouth send, which is then not anchored twice). async function noteOwnPost(roomId, event) { -- 2.47.3 From b130c2a4184182e0269e7ee9c63fa8fae6ebe5e4 Mon Sep 17 00:00:00 2001 From: Heimdall Date: Thu, 12 Nov 2026 12:08:55 +0000 Subject: [PATCH 5/5] Follow: test the in-flight send and echo meeting Co-Authored-By: Claude Sonnet 5.5 --- tests/persona-matrix-mouth-follow.test.mjs | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/tests/persona-matrix-mouth-follow.test.mjs b/tests/persona-matrix-mouth-follow.test.mjs index 2109682..c5bd5f2 100644 --- a/tests/persona-matrix-mouth-follow.test.mjs +++ b/tests/persona-matrix-mouth-follow.test.mjs @@ -21,7 +21,7 @@ const msg = (id, sender, ts, content = {}) => ({ type: "m.room.message", event_i const tagKiko = { "m.mentions": { user_ids: [KIKO] } }; // rounds: sync replies after the baseline, each { [roomId]: events }. state: room id -> state events. -function world({ rounds, state = {}, stateStatus = {}, hangState = false, stateFailures = 0, baselineEvents = null, onSync = null, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { +function world({ rounds, state = {}, stateStatus = {}, hangState = false, stateDelay = 0, stateFailures = 0, baselineEvents = null, onSync = null, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) { const seen = []; const known = new Map(); let round = 0; @@ -36,6 +36,7 @@ function world({ rounds, state = {}, stateStatus = {}, hangState = false, stateF if (p.startsWith("/directory/room/")) return reply(200, { room_id: { [ALIAS]: ROOM, [OTHER_ALIAS]: OTHER_ROOM }[p.slice(16)] }); if (p.startsWith("/presence/") || p.includes("/receipt/")) return reply(200, {}); const st = p.match(/^\/rooms\/([^/]+)\/state$/); + if (st && stateDelay > 0) await new Promise(r => setTimeout(r, stateDelay)); if (st && hangState) return new Promise(() => {}); if (st && stateFailures > 0) { stateFailures--; return reply(500, { errcode: "M_UNKNOWN" }); } if (st) return stateStatus[st[1]] ? reply(stateStatus[st[1]], { errcode: "M_UNKNOWN" }) : reply(200, state[st[1]] ?? []); @@ -365,3 +366,13 @@ test("follow: a not-public verdict is recorded: the room is not read again for e assert.equal(w.seen.filter(l => l.endsWith("/state")).length, 1); assert.deepEqual(w.mouth.followState().engagements, []); }); + +test("follow: an echo that arrives while the send's anchor is still reading waits for it: one read, one snapshot", async () => { + const w = world({ stateDelay: 80, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0 + 5 * SEC, { body: "hello" })] }] }); + await w.mouth.syncOnce(); await w.basisOfRound(); + await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); // its anchor is now waiting on the state read + await w.basisOfRound(); // the echo, meanwhile + await new Promise(r => setTimeout(r, 200)); + assert.equal(w.seen.filter(l => l.endsWith("/state")).length, 1); + assert.equal(w.mouth.followState().engagements[0].from, T0, "anchored at the send's time, not the echo's"); +}); -- 2.47.3