diff --git a/docs/bridge.md b/docs/bridge.md
index d66f9a5..5872f22 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 follow state.
+
+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`, 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.
+
### 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..c5bd5f2
--- /dev/null
+++ b/tests/persona-matrix-mouth-follow.test.mjs
@@ -0,0 +1,378 @@
+// 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";
+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 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 = {}, 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;
+ 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 && 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]] ?? []);
+ 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", ...(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 } }])) } });
+ }
+ if (p.includes("/messages")) return reply(200, { chunk: [] });
+ 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,
+ });
+ 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; };
+
+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 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("$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].sort(), [["$e", null], ["$k", null], ["$l", "follow"], ["$s", null]]);
+});
+
+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("$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)] },
+ ] });
+ assert.deepEqual((await run(w, 3))[2], [["$k", "follow"], ["$s", null]]);
+});
+
+
+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("$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)] },
+ ] });
+ 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: 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]]);
+ }
+ 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 sender or event is neither counted nor followed", async () => {
+ const puppet = `@_${PLATFORM}_123:${HS}`;
+ 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 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("$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, 3))[2], [["$k", null]]);
+ assert.ok(!off.seen.some(l => l.endsWith("/state")), "off: no state read");
+ 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);
+});
+
+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");
+});
+
+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 });
+});
+
+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, []);
+});
+
+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");
+});
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..77b45e2 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,11 @@
// 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 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
// (persona-chat-images.mjs): the delivery then carries a reference to the stored file, or the line says why not.
@@ -68,6 +73,12 @@ 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 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.
@@ -126,6 +137,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 +315,96 @@ export function createMouth({
return lines;
}
+ // 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 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 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 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();
+ 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);
+ }
+ 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) 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) {
+ 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 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 - 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);
+ 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
@@ -440,6 +542,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
@@ -474,7 +577,10 @@ 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) 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;
processed.add(event.event_id);
if (deliverable) {
@@ -647,7 +753,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(() => {}); } // 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}`);
@@ -887,7 +996,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).
@@ -1074,6 +1185,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 +1240,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 }),