diff --git a/docs/bridge.md b/docs/bridge.md index cdac2eb..c2bc284 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -196,7 +196,7 @@ The command is a dry-run shape probe for public surfaces that are real enough to Bifrost's native Discord gateway is the conversation observer. A direct mention of the Bifrost bot, or a reply to one of its messages, becomes an immutable typed `bifrost.discord.persona_conversation_event.v0` only when the guild and channel have one exact Persona target binding. Bot-authored, ambient, empty, over-1200-character, unbound, and ambiguously bound messages do not enter the conversation path. The 1200-character cap is the shared signed alpha boundary with Epiphany. The observation store is Bifrost-private working state, never an actuator queue, and insert collisions must contain exactly equal content. VoidBot does not observe, schedule, or answer for the target Persona. -Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process for every Persona (`PERSONAS`, a list of slugs): it holds each Persona's Matrix access token as its own systemd credential (`matrix-token-`) and nobody else holds any. Per Persona it keeps one `/sync` loop over the configured rooms and the account's direct rooms (a DM invitation from an account on `MATRIX_SERVER_NAME` is accepted while that Persona's brake is released, and one from any other server is declined), one cursor and one delivery sequence. A delivery carries the sender's Matrix id, the event's `m.mentions` user ids and whether the room is a DM. A media event (`m.image`, `m.file`, `m.audio`, `m.video`) is delivered as a typed text line, `[image filename="cat.png" mime=image/png size=2048 url=mxc://server/id]`, followed by its caption when it has one, in `text` and in context. Only `m.image` is also fetched, and only when `IMAGE_STORE_ROOT` is set (unset, the line above is all there is and nothing is downloaded). The fetch is made as the receiving Persona, with its own token, from `/_matrix/client/v1/media/download//` on its homeserver, for a message the Persona is delivered (not its own words, a room it is not joined to, or history before the ear), and only from the event's own `content.url`; a URL in the message text is never followed. An encrypted attachment (`m.room.encrypted`, or `content.file`) is never fetched. The type is decided by the magic bytes alone (png, jpeg, gif, webp; SVG and anything else is refused, the declared mime is ignored), the body is capped at 5 MiB by `Content-Length` and by the bytes actually read, and the pixel count read from the image header (never decoded) is capped at 40 MP. A refused or failed fetch leaves the typed line with a fixed-word reason (`refused=encrypted|no-url|fetch-failed|too-big|wrong-type|bad-header|too-many-pixels|store-failed`) and no image. A stored image is written to `IMAGE_STORE_ROOT///[.png|.jpg|.gif|.webp]`, the name generated by the mouth, never the sender's: files `0640`, room directories `2750` (the Persona directory is setgid to the Persona's group, so the Persona's reader opens them by group; the mouth is the only writer), each written under a temporary name and renamed. It is deleted once it is at least 24 hours old and its delivery has been acked, whichever is later; a room keeps at most 50 MiB, the oldest evicted first. The sweep runs when an image is stored and when an ack arrives. A delivery is the 12-field tuple of `gamecult.persona.chat_pending.v2`: field 11, after `is_direct`, is nil or `[path, mime, size, sha256]`, a reference and never bytes; `path` is the store-relative `/` above, `mime` the sniffed type, `size` the file length (at most 5 MiB) and `sha256` its lowercase hex digest. The reader re-checks all four against the file. A Persona turn that has no reader for it still sees the typed line. Reading wakes nothing and acks nothing: the heartbeat polls with the non-consuming `gamecult.persona.chat_pending_request.v1`, and only `chat_ack` advances a position. The mouth speaks on a `gamecult.persona.chat_send_request.v1` and joins rooms only while that Persona's brake (`/etc/gamecult/minds//control.cc`) is released. One loopback CultNet RUDP endpoint serves all of them; a request naming an unlisted Persona gets a typed `chat_refusal` (`UnknownPersona`). Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`, `-v2/` and `-v3/` (the pending answer; Huginn pins the same bytes). The Discord posting actuator, its permit and its RUDP listener are removed. +Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process for every Persona (`PERSONAS`, a list of slugs): it holds each Persona's Matrix access token as its own systemd credential (`matrix-token-`) and nobody else holds any. Per Persona it keeps one `/sync` loop over the configured rooms and the account's direct rooms (a DM invitation from an account on `MATRIX_SERVER_NAME` is accepted while that Persona's brake is released, and one from any other server is declined), one cursor and one delivery sequence. A delivery carries the sender's Matrix id, the event's `m.mentions` user ids and whether the room is a DM. A media event (`m.image`, `m.file`, `m.audio`, `m.video`) is delivered as a typed text line, `[image filename="cat.png" mime=image/png size=2048 url=mxc://server/id]`, followed by its caption when it has one, in `text` and in context. Only `m.image` is also fetched, and only when `IMAGE_STORE_ROOT` is set (unset, the line above is all there is and nothing is downloaded). The fetch is made as the receiving Persona, with its own token, from `/_matrix/client/v1/media/download//` on its homeserver, for a message the Persona is delivered (not its own words, a room it is not joined to, or history before the ear), and only from the event's own `content.url`; a URL in the message text is never followed. An encrypted attachment (`m.room.encrypted`, or `content.file`) is never fetched. The type is decided by the magic bytes alone (png, jpeg, gif, webp; SVG and anything else is refused, the declared mime is ignored), the body is capped at 5 MiB by `Content-Length` and by the bytes actually read, and the pixel count read from the image header (never decoded) is capped at 40 MP. A refused or failed fetch leaves the typed line with a fixed-word reason (`refused=encrypted|no-url|fetch-failed|too-big|wrong-type|bad-header|too-many-pixels|store-failed`) and no image. A stored image is written to `IMAGE_STORE_ROOT///[.png|.jpg|.gif|.webp]`, the name generated by the mouth, never the sender's: files `0640`, room directories `2750` (the Persona directory is setgid to the Persona's group, so the Persona's reader opens them by group; the mouth is the only writer), each written under a temporary name and renamed. It is deleted once it is at least 24 hours old and its delivery has been acked, whichever is later; a room keeps at most 50 MiB, the oldest evicted first. The sweep runs when an image is stored and when an ack arrives. A delivery is the 12-field tuple of `gamecult.persona.chat_pending.v2`: field 11, after `is_direct`, is nil or `[path, mime, size, sha256]`, a reference and never bytes; `path` is the store-relative `/` above, `mime` the sniffed type, `size` the file length (at most 5 MiB) and `sha256` its lowercase hex digest. The reader re-checks all four against the file. A Persona turn that has no reader for it still sees the typed line. Reading wakes nothing and acks nothing: the heartbeat polls with the non-consuming `gamecult.persona.chat_pending_request.v1`, and only `chat_ack` advances a position. An ack also posts that Persona's public `m.read` receipt, per room, at the newest acked event (fire and forget: the ack never waits on it, a failure is logged once and not retried; the room's next ack covers it); the heartbeat's and a turn's pending calls post none. Her Matrix presence follows her brake: each `/sync` carries `set_presence=online` while it is released and `offline` while it is held, and a round that finds her braked makes one `PUT /presence//status` to offline first (a brake placed mid long-poll shows within `syncTimeoutMs`, 30 s, plus one PUT; a stopped mouth reads offline after Synapse's sync timeout). The mouth speaks on a `gamecult.persona.chat_send_request.v1` and joins rooms only while that Persona's brake (`/etc/gamecult/minds//control.cc`) is released. One loopback CultNet RUDP endpoint serves all of them; a request naming an unlisted Persona gets a typed `chat_refusal` (`UnknownPersona`). Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`, `-v2/` and `-v3/` (the pending answer; Huginn pins the same bytes). The Discord posting actuator, its permit and its RUDP listener are removed. Bifrost admits an observation only when guild, channel, target runtime, canonical repository, Persona, producer, and producer runtime match a preconfigured `bifrost.persona_feedback.target_binding.v0` document in Bifrost's private provider store. That binding also owns source visibility, data classification, and the delivery-store target. Bifrost never infers public visibility from Discord identifiers. It classifies the author using its own `bifrost.discord.actor_link.v0` state: a match yields `linked_governance_feedback`; absence yields `unlinked_social_feedback`. Neither class grants adoption, work, release, or deployment authority. diff --git a/tests/persona-matrix-mouth-runtime.test.mjs b/tests/persona-matrix-mouth-runtime.test.mjs index a78d22c..e4dcf88 100644 --- a/tests/persona-matrix-mouth-runtime.test.mjs +++ b/tests/persona-matrix-mouth-runtime.test.mjs @@ -10,7 +10,7 @@ import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { resolve } from "node:path"; import * as docs from "../tools/persona-chat-documents.mjs"; -import { cultcacheState, connectionIdFor, createBrakeReader, loadRuntime, pageOf, startChatEndpoint } from "../tools/persona-matrix-mouth.mjs"; +import { createMouth, cultcacheState, connectionIdFor, createBrakeReader, loadRuntime, pageOf, startChatEndpoint } from "../tools/persona-matrix-mouth.mjs"; const root = resolve(import.meta.dirname, ".."); const runtime = loadRuntime(resolve(process.env.CULTLIB_ROOT ?? resolve(root, "..", "CultLib"))); @@ -370,3 +370,55 @@ test("the RUDP endpoint answers a v1 send with the three-field receipt and a v2 try { client.close(); } catch { /* the connection already closed its socket */ } } }); + +test("over the RUDP endpoint: pending posts no receipt, and an ack never holds up the same peer's next request", async () => { + const receipts = []; + let release; + const held = new Promise(r => { release = r; }); + const json = (body, status = 200) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } }); + const message = { type: "m.room.message", event_id: "$m1", sender: "@alice:t", origin_server_ts: 5, content: { msgtype: "m.text", body: "hi" } }; + const fetch = async (url, init = {}) => { + const u = new URL(url), p = u.pathname.replace("/_matrix/client/v3", ""); + if (p === "/account/whoami") return json({ user_id: "@eureka:t" }); + if (p === "/joined_rooms") return json({ joined_rooms: ["!r:t"] }); + if (p.startsWith("/directory/room/")) return json({ room_id: "!r:t" }); + if (p.includes("/account_data/")) return json({}, 404); + if (p === "/sync") return json(u.searchParams.get("since") === "b0" ? { next_batch: "b1", rooms: { join: { "!r:t": { timeline: { events: [message] } } } } } : { next_batch: "b0" }); + if (p.includes("/messages")) return json({ chunk: [] }); + if (p.includes("/receipt/")) { receipts.push(p); await held; return json({}); } + return json({}, 404); + }; + const state = { load: async () => null, save: async () => {} }; + const mouth = createMouth({ persona: "eureka", homeserver: "http://hs.test", rooms: ["#r:t"], token: "tok", state, fetch, serverName: "t", brake: async () => true, syncTimeoutMs: 1 }); + await mouth.syncOnce(); await mouth.syncOnce(); + const key = "connection-key-for-receipts"; + const server = await startChatEndpoint({ mouths: new Map([["eureka", mouth]]), runtime, port: 0, connectionKey: key, runtimeId: "mouth-test" }); + const client = dgram.createSocket("udp4"); + await new Promise(ok => client.bind(0, "127.0.0.1", ok)); + const connection = new runtime.cultnet.CultNetRudpSocketTransportConnection({ + runtimeId: "huginn-test", socket: client, mode: "client", remoteHost: "127.0.0.1", remotePort: server.address.port, + connectionId: connectionIdFor(key), initialSequence: 1, resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200, + }); + const replies = []; + connection.on("frame", value => replies.push(runtime.cultnet.parseCultNetMessage(msgpack.decode(value.payload), "cultnet.schema.v0").document)); + const put = (kind, recordKey, tuple) => connection.send("schema", msgpack.encode(runtime.cultnet.encodeCultNetMessageForWire( + docs.envelope({ kind, recordKey, payload: msgpack.encode(tuple), sourceRuntimeId: "huginn-test", sourceAgentId: "eureka", sourceRole: "persona-organ" }), "cultnet.schema.v0"))); + const waitFor = async (check, what) => { for (let i = 0; i < 250 && !check(); i++) await new Promise(r => setTimeout(r, 20)); assert.ok(check(), what); }; + try { + connection.connect(); + await waitFor(() => connection.connected, "session opens"); + put(docs.kinds.pendingRequest, "eureka", docs.pendingRequestTuple({ persona: "eureka", afterSeq: 0 })); + await waitFor(() => replies.length === 1, "pending answered"); + assert.deepEqual(receipts, [], "a pending request posts no receipt"); + put(docs.kinds.ack, "eureka:1", docs.ackTuple({ persona: "eureka", throughSeq: 1 })); + await waitFor(() => receipts.length === 1, "the ack posted her receipt"); + put(docs.kinds.pendingRequest, "eureka", docs.pendingRequestTuple({ persona: "eureka", afterSeq: 0 })); + await waitFor(() => replies.length === 2, "the next request is answered while the receipt response is still held open"); + assert.deepEqual(docs.parsePending(msgpack.decode(replies[1].payload)).deliveries, [], "the ack applied"); + } finally { + release(); + server.close(); + connection.close(); + try { client.close(); } catch { /* already closed */ } + } +}); diff --git a/tests/persona-matrix-mouth.test.mjs b/tests/persona-matrix-mouth.test.mjs index 680eef8..9245f9d 100644 --- a/tests/persona-matrix-mouth.test.mjs +++ b/tests/persona-matrix-mouth.test.mjs @@ -18,12 +18,14 @@ const ROOMS = { [`#swarm:${HS_NAME}`]: "!swarm:gamecult.test", [`#operator:${HS_ const SELF = `@eureka:${HS_NAME}`; const message = (id, sender, body, ts) => ({ type: "m.room.message", event_id: id, sender, origin_server_ts: ts, content: { msgtype: "m.text", body } }); +// The one write a braked round makes by design; an online presence PUT or any other write is not exempt. +const offlinePresence = r => r.method === "PUT" && r.path.includes("/presence/") && r.body?.presence === "offline"; const tick = () => new Promise(r => setImmediate(r)); // A fake Matrix client-server API: just the endpoints the mouth uses. // It is also the record of every room: events the sync script emits and messages sent are kept, and // `preload` (room id -> events, oldest first) stands for history from before the mouth's process began. -async function fakeHomeserver({ sync, failWith, users, member, direct, preload = {}, failContext = false, media } = {}) { +async function fakeHomeserver({ sync, failWith, users, member, direct, preload = {}, failContext = false, media, failReceipts = false, deferReceipts = null, failPresence = 0 } = {}) { const seen = []; const record_ = new Map(Object.entries(preload).map(([room, events]) => [room, [...events]])); const remember = (room, event) => { const list = record_.get(room) ?? []; if (!list.some(e => e.event_id === event.event_id)) list.push(event); record_.set(room, list); }; @@ -34,12 +36,17 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload = req.on("data", c => (raw += c)); req.on("end", () => { const url = new URL(req.url, "http://x"); - const record = { method: req.method, path: decodeURIComponent(url.pathname), query: url.searchParams, auth: req.headers.authorization, body: raw ? JSON.parse(raw) : null }; + const record = { method: req.method, rawPath: url.pathname, path: decodeURIComponent(url.pathname), query: url.searchParams, auth: req.headers.authorization, body: raw ? JSON.parse(raw) : null }; seen.push(record); const reply = (status, body) => { res.writeHead(status, { "content-type": "application/json" }); res.end(JSON.stringify(body)); }; if (failWith) return reply(failWith, { errcode: "M_UNKNOWN_TOKEN", error: `bad token ${record.auth}` }); const p = record.path.replace("/_matrix/client/v3", "").replace("/_matrix/client/v1", ""); const bearer = (record.auth ?? "").replace("Bearer ", ""); + if (req.method === "POST" && p.includes("/receipt/m.read/")) { + if (deferReceipts) return deferReceipts.then(() => reply(200, {})); + return failReceipts ? reply(500, { errcode: "M_UNKNOWN", error: `down ${record.auth}` }) : reply(200, {}); + } + if (req.method === "PUT" && p.includes("/presence/")) return failPresence-- > 0 ? reply(500, { errcode: "M_UNKNOWN" }) : reply(200, {}); if (p.startsWith("/media/download/")) { const served = media?.(bearer, ...p.slice("/media/download/".length).split("/")) ?? { status: 404, body: "" }; res.writeHead(served.status ?? 200, served.headers ?? {}); @@ -77,7 +84,9 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload = reply(404, { errcode: "M_NOT_FOUND" }); }); }; - const fetch = (url, init = {}) => new Promise(resolve => { + const fetch = (url, init = {}) => new Promise((resolve, reject) => { + if (init.signal?.aborted) return reject(init.signal.reason); + init.signal?.addEventListener("abort", () => reject(init.signal.reason), { once: true }); const u = new URL(url); const req = Object.assign(new EventEmitter(), { method: init.method ?? "GET", url: u.pathname + u.search, headers: init.headers ?? {} }); const res = { @@ -91,7 +100,9 @@ async function fakeHomeserver({ sync, failWith, users, member, direct, preload = url: "http://hs.test", fetch, seen, - puts: () => seen.filter(r => r.method === "PUT"), + puts: () => seen.filter(r => r.method === "PUT" && !offlinePresence(r)), + receipts: () => seen.filter(r => r.path.includes("/receipt/")), + presencePuts: () => seen.filter(r => r.path.includes("/presence/")), downloads: () => seen.filter(r => r.path.includes("/media/download/")), close: async () => {}, }; @@ -443,7 +454,7 @@ test("join_while_braked: no join and no send while the brake is held; joining ha const mouth = mouthFor(hs, { brake: async () => open, log: l => logs.push(l) }); await mouth.syncOnce(); await mouth.send({ requestId: "r", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "speak" }); - assert.equal(hs.seen.filter(r => r.method === "POST" || r.method === "PUT").length, 0, "braked: nothing is joined or sent"); + assert.equal(hs.seen.filter(r => (r.method === "POST" || r.method === "PUT") && !offlinePresence(r)).length, 0, "braked: nothing is joined or sent"); open = true; await mouth.syncOnce(); assert.deepEqual(hs.seen.filter(r => r.method === "POST").map(r => r.path.split("/join/")[1]).sort(), ["!operator:gamecult.test", "!swarm:gamecult.test"]); @@ -588,7 +599,7 @@ test("dm_invite_waits_for_release: an invitation received while braked is not jo try { const mouth = mouthFor(hs, { brake: async () => open }); await mouth.syncOnce(); await mouth.syncOnce(); await mouth.syncOnce(); - assert.deepEqual(hs.seen.filter(r => r.method === "POST" || r.method === "PUT"), [], "braked: no join, no leave, nothing written"); + assert.deepEqual(hs.seen.filter(r => (r.method === "POST" || r.method === "PUT") && !offlinePresence(r)), [], "braked: no join, no leave, nothing written"); open = true; await mouth.syncOnce(); assert.deepEqual(joinsOf(hs), [DMROOM]); @@ -941,3 +952,218 @@ test("image_delivery: a store that cannot write leaves the line and a nil image, assert.ok(lines.some(l => l.includes("ENOSPC"))); } finally { await hs.close(); } }); + +// Read receipts and presence: her ack posts her m.read; her brake sets her presence. +const A = "!swarm:gamecult.test", B = "!operator:gamecult.test"; +const receiptScript = since => since === null ? { next_batch: "b0", ...joined({}) } : since === "b0" ? { + next_batch: "b1", + ...joined({ + [A]: { timeline: { events: [message("$a1", `@alice:${HS_NAME}`, "one", 10), message("$a2", `@alice:${HS_NAME}`, "two", 11)] } }, + [B]: { timeline: { events: [message("$b1", `@bob:${HS_NAME}`, "three", 12)] } }, + }), +} : { next_batch: since }; +const until = async check => { for (let i = 0; i < 200 && !check(); i++) await new Promise(r => setImmediate(r)); }; +const withMail = async (hs, overrides) => { + const mouth = mouthFor(hs, overrides); + await mouth.syncOnce(); + assert.equal((await mouth.syncOnce()).map(d => d.seq).join(), "1,2,3"); + return mouth; +}; +const receiptsOf = hs => hs.receipts().map(r => [r.path.split("/rooms/")[1].split("/receipt/")[0], r.path.split("/m.read/")[1], r.auth, r.body]); + +test("an ack posts one m.read per room at the newest covered event, as that Persona", async () => { + const hs = await fakeHomeserver({ sync: receiptScript }); + const mouth = await withMail(hs); + await mouth.ack(2); + await until(() => hs.receipts().length >= 1); + await new Promise(r => setTimeout(r, 20)); + assert.deepEqual(receiptsOf(hs), [[A, "$a2", `Bearer ${TOKEN}`, {}]], "room A at its second event; room B untouched"); + await mouth.ack(3); + await until(() => hs.receipts().length >= 2); + assert.deepEqual(receiptsOf(hs).slice(1), [[B, "$b1", `Bearer ${TOKEN}`, {}]]); + assert.equal(hs.receipts().length, 2); +}); + +test("pending posts no receipt", async () => { + const hs = await fakeHomeserver({ sync: receiptScript }); + const mouth = await withMail(hs); + for (const after of [0, 1, 2, 3]) mouth.pendingAfter(after); + await new Promise(r => setTimeout(r, 20)); + assert.deepEqual(hs.receipts(), []); +}); + +test("a failed receipt leaves the ack whole", async () => { + const hs = await fakeHomeserver({ sync: receiptScript, failReceipts: true }); + const lines = []; + const state = memoryState(); + const mouth = await withMail(hs, { state, log: l => lines.push(l) }); + const saves = []; + const save = state.save; + state.save = async v => { saves.push(v); return save(v); }; + await mouth.ack(3); + await until(() => lines.some(l => l.includes("receipt in"))); + assert.equal(mouth.pendingCount, 0); + assert.equal(state.box.value.nextBatch, "b1", "the cursor committed"); + assert.ok(saves.length >= 1, "state.save called"); + const failures = lines.filter(l => l.startsWith("receipt in ")); + assert.equal(failures.length, 2, "one line per room, nothing retried"); + assert.ok(failures.every(l => l.includes("not posted") && !l.includes(TOKEN))); + await new Promise(r => setTimeout(r, 20)); + assert.equal(hs.receipts().length, 2, "no retry"); +}); + +test("the ack does not wait for the receipt", async () => { + let release; + const hs = await fakeHomeserver({ sync: receiptScript, deferReceipts: new Promise(r => { release = r; }) }); + const mouth = await withMail(hs); + const outcome = await Promise.race([mouth.ack(3).then(() => "acked"), new Promise(r => setTimeout(() => r("waited on the receipt"), 500))]); + assert.equal(outcome, "acked", "the ack resolves while the receipt response is held open"); + assert.equal(mouth.pendingCount, 0); + release(); + await until(() => hs.receipts().length >= 2); +}); + +test("presence follows the brake: online while released, offline and one PUT on the transition", async () => { + let open = true; + const hs = await fakeHomeserver({ sync: receiptScript }); + const mouth = mouthFor(hs, { brake: async () => open }); + const syncs = () => hs.seen.filter(r => r.path.endsWith("/sync")).map(r => r.query.get("set_presence")); + await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual(syncs(), ["online", "online"]); + assert.equal(hs.presencePuts().length, 0, "released: no PUT"); + open = false; + await mouth.syncOnce(); await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual(syncs().slice(2), ["offline", "offline", "offline"]); + assert.deepEqual(hs.presencePuts().map(r => [r.path, r.body]), [[`/_matrix/client/v3/presence/${SELF}/status`, { presence: "offline" }]], "exactly one PUT"); + assert.equal(hs.presencePuts()[0].rawPath, `/_matrix/client/v3/presence/${encodeURIComponent(SELF)}/status`, "the user id is percent-encoded on the wire"); + open = true; + await mouth.syncOnce(); + assert.equal(syncs().at(-1), "online"); + open = false; + await mouth.syncOnce(); + assert.equal(hs.presencePuts().length, 2, "a second transition PUTs once more"); +}); + +test("the first braked round of a fresh mouth PUTs offline once; a failed PUT is tried again next round", async () => { + const hs = await fakeHomeserver({ sync: receiptScript, failPresence: 1 }); + const lines = []; + const mouth = mouthFor(hs, { brake: async () => false, log: l => lines.push(l) }); + await mouth.syncOnce(); + assert.equal(hs.presencePuts().length, 1); + assert.ok(lines.some(l => l.startsWith("presence not set")), "one log line"); + await mouth.syncOnce(); + assert.equal(hs.presencePuts().length, 2, "retried the next round"); + await mouth.syncOnce(); + assert.equal(hs.presencePuts().length, 2, "then settled"); +}); + +test("two Personas on one homeserver: receipts and presence carry only her own bearer and user id", async () => { + const ALICE = `@pa:${HS_NAME}`, BOB = `@pb:${HS_NAME}`; + const hs = await fakeHomeserver({ sync: receiptScript, users: { tokA: ALICE, tokB: BOB } }); + const a = await withMail(hs, { persona: "pa", token: "tokA" }); + let bOpen = false; + const b = await withMail(hs, { persona: "pb", token: "tokB", brake: async () => bOpen }); + await a.ack(3); + await b.syncOnce(); + await until(() => hs.receipts().length >= 2); + assert.ok(hs.receipts().every(r => r.auth === "Bearer tokA"), "only A acked"); + bOpen = true; // her receipts go out only while released + await b.ack(3); + await until(() => hs.receipts().length >= 4); + assert.deepEqual([...new Set(hs.receipts().map(r => r.auth))].sort(), ["Bearer tokA", "Bearer tokB"]); + assert.ok(hs.presencePuts().length >= 1 && hs.presencePuts().every(r => r.path.endsWith(`/presence/${BOB}/status`) && r.auth === "Bearer tokB")); +}); + +// ---- review fixes ------------------------------------------------------------------ + +test("a brake placed during the long-poll stops the join the poll's invite asks for", async () => { + let open = true; + const script = inviteSync(`@meta:${HS_NAME}`); + const hs = await fakeHomeserver({ sync: since => { const out = script(since); if (since === "b0") open = false; return out; } }); // the brake lands while this poll is open + const mouth = mouthFor(hs, { brake: async () => open }); + await mouth.syncOnce(); // baseline, released + await mouth.syncOnce(); // the poll that carries the invite + assert.deepEqual(joinsOf(hs), [], "no join"); + assert.equal(hs.seen.filter(r => r.path.endsWith("/account_data/m.direct") && r.method === "PUT").length, 0, "no m.direct write"); +}); + +test("no m.read receipt while braked: the ack proceeds, nothing is queued", async () => { + let open = true; + const hs = await fakeHomeserver({ sync: receiptScript }); + const lines = []; + const state = memoryState(); + const mouth = await withMail(hs, { brake: async () => open, state, log: l => lines.push(l) }); + open = false; + await mouth.ack(3); + await until(() => lines.some(l => l.includes("receipts not posted"))); + assert.equal(mouth.pendingCount, 0); + assert.equal(state.box.value.nextBatch, "b1", "the cursor committed"); + assert.equal(lines.filter(l => l.includes("receipts not posted")).length, 1, "one line"); + open = true; + await mouth.syncOnce(); + await new Promise(r => setTimeout(r, 20)); + assert.deepEqual(hs.receipts(), [], "released again: nothing deferred is posted"); +}); + +test("the receipt path is percent-encoded on the wire", async () => { + const ROOM = "!od d/room:gamecult.test", EVENT = "$ev/with+odd:chars"; + const hs = await fakeHomeserver({ sync: since => since === null ? { next_batch: "b0" } : { next_batch: "b1", account_data: { events: [{ type: "m.direct", content: { [`@meta:${HS_NAME}`]: [ROOM] } }] }, ...joined({ [ROOM]: { timeline: { events: [message(EVENT, `@meta:${HS_NAME}`, "x", 9)] } } }) } }); + const mouth = mouthFor(hs); + await mouth.syncOnce(); await mouth.syncOnce(); + await mouth.ack(1); + await until(() => hs.receipts().length >= 1); + assert.equal(hs.receipts()[0].rawPath, `/_matrix/client/v3/rooms/${encodeURIComponent(ROOM)}/receipt/m.read/${encodeURIComponent(EVENT)}`); +}); + +test("a receipt that never answers is abandoned after sendTimeoutMs", async () => { + const hs = await fakeHomeserver({ sync: receiptScript, deferReceipts: new Promise(() => {}) }); + const lines = []; + const mouth = await withMail(hs, { sendTimeoutMs: 40, log: l => lines.push(l) }); + await mouth.ack(3); + await until(() => lines.length >= 1); + await new Promise(r => setTimeout(r, 150)); + assert.ok(lines.some(l => l.startsWith("receipt in ") && l.includes("not posted")), "abandoned and logged"); + assert.equal(hs.receipts().length, 2, "the next room's receipt went out after the first was abandoned"); +}); + +test("a throwing logger on a failed receipt causes no unhandled rejection", async () => { + const hs = await fakeHomeserver({ sync: receiptScript, failReceipts: true }); + const mouth = await withMail(hs, { log: () => { throw new Error("logger down"); } }); + const rejected = []; + const note = e => rejected.push(e); + process.on("unhandledRejection", note); + try { + await mouth.ack(3); + await until(() => hs.receipts().length >= 2); + await new Promise(r => setTimeout(r, 50)); + } finally { process.off("unhandledRejection", note); } + assert.deepEqual(rejected, []); +}); + +test("a full backlog still sets presence from the brake", async () => { + let open = true; + const hs = await fakeHomeserver({ sync: receiptScript }); + const mouth = mouthFor(hs, { brake: async () => open, maxPending: 3 }); + await mouth.syncOnce(); + assert.equal((await mouth.syncOnce()).length, 3); + const syncsBefore = hs.seen.filter(r => r.path.endsWith("/sync")).length; + open = false; + assert.deepEqual(await mouth.syncOnce(), [], "backlog full: no /sync"); + assert.equal(hs.seen.filter(r => r.path.endsWith("/sync")).length, syncsBefore); + assert.deepEqual(hs.presencePuts().map(r => r.body), [{ presence: "offline" }]); +}); + +test("a brake placed during the first of two joins stops the second join and its m.direct write", async () => { + const DM2 = "!newdm2:gamecult.test"; + const invites2 = { ...invite(`@meta:${HS_NAME}`), [DM2]: invite(`@meta:${HS_NAME}`)[DMROOM] }; + const hs = await fakeHomeserver({ sync: since => since === null ? { next_batch: "b0" } : since === "b0" ? { next_batch: "b1", rooms: { invite: invites2 } } : { next_batch: since } }); + let open = true; + const fetch = async (url, init) => { const res = await hs.fetch(url, init); if (init?.method === "POST" && String(url).includes("/join/")) open = false; return res; }; // the brake lands during the first join + const mouth = mouthFor(hs, { fetch, brake: async () => open }); + await mouth.syncOnce(); await mouth.syncOnce(); + assert.deepEqual(joinsOf(hs), [DMROOM], "only the join already started"); + assert.equal(hs.seen.filter(r => r.method === "PUT" && r.path.endsWith("/account_data/m.direct")).length, 1, "one m.direct, for the first join"); + open = true; + await mouth.syncOnce(); + assert.deepEqual(joinsOf(hs), [DMROOM, DM2], "the second waits for the release"); +}); diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs index 57b7405..21f1c7c 100644 --- a/tools/persona-matrix-mouth.mjs +++ b/tools/persona-matrix-mouth.mjs @@ -6,7 +6,12 @@ // 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. Reading wakes nothing: the heartbeat polls -// with a non-consuming chat_pending_request and decides who runs. +// with a non-consuming chat_pending_request and decides who runs. An ack also posts this Persona's public +// m.read receipt per room, at the newest event the ack covers (fire and forget: the ack never waits on it, +// a failure is logged once and not retried, the room's next ack covers it). Her presence follows her +// brake: every /sync says online while it is released and offline while it is held, and a round that +// finds her braked makes one presence PUT to offline first. A brake placed mid long-poll shows at the next +// round, within syncTimeoutMs plus one PUT; a stopped mouth reads offline after the homeserver's sync timeout. // Mouth: a send request becomes one PUT with txnId = request_id (a retry is idempotent), and a room // is joined, only while that Persona's brake is released. A v2 send carries the basis of her turn (the // newest room event she saw): inside the room gate, one step per room across every Persona, the mouth @@ -119,6 +124,7 @@ export function createMouth({ let batches = []; // { token, maxSeq }: positions awaiting an ack let committed = null; // the position that is persisted let started = false; + let lastPresence = null; // 'online' | 'offline' | null: command-only, suppresses repeat presence PUTs const directContent = {}; // the account's m.direct content: user id -> room ids const invites = new Map(); // room id -> { inviter, direct }: invitations not yet answered const processed = new Set(); // event ids read since the position last advanced @@ -159,7 +165,6 @@ export function createMouth({ // released; one received while braked waits for the release. An invitation from any other server is // declined. A same-server invitation to a room that is not a DM is not ours to accept and is dropped. async function answerInvites() { - let free = null; for (const [roomId, invite] of [...invites]) { try { const room = encodeURIComponent(roomId); @@ -169,8 +174,7 @@ export function createMouth({ } else if (!invite.direct) { invites.delete(roomId); } else { - free ??= await released(); - if (!free) continue; + if (!(await released())) continue; // read fresh before every join: a brake placed mid-pass stops the next one await matrix("POST", `/join/${room}`, {}); invites.delete(roomId); directRooms.add(roomId); @@ -280,14 +284,22 @@ export function createMouth({ // One /sync round. Returns the deliveries it made. async function syncOnce() { if (!started || unjoined()) await start(); + const free = await released(); // the round's brake read for presence, taken before any early return + if (!free && lastPresence !== "offline") { + try { + await matrix("PUT", `/presence/${encodeURIComponent(selfId)}/status`, { presence: "offline" }, API, AbortSignal.timeout(sendTimeoutMs)); + lastPresence = "offline"; + } catch (error) { say(`presence not set: ${error.message}`); } + } if (pending.length >= maxPending) return []; // unanswered mail is not outrun: the position stays where it is const filter = encodeURIComponent(JSON.stringify({ room: { timeline: { limit: CONTEXT } } })); const since = cursor === null ? "" : `&since=${encodeURIComponent(cursor)}`; - const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&filter=${filter}${since}`); + const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&set_presence=${free ? "online" : "offline"}&filter=${filter}${since}`); + if (free) lastPresence = "online"; const baseline = cursor === null; // first ever sync: history is context, not mail for (const event of reply.account_data?.events ?? []) if (event.type === "m.direct") noteDirect(event.content); noteInvites(reply.rooms?.invite); - await answerInvites(); + await answerInvites(); // each join re-reads the brake, as send does const made = []; const round = new Map(); let truncated = false; // the pending cap was reached inside this round @@ -346,10 +358,26 @@ export function createMouth({ batches = batches.filter(b => b.maxSeq > throughSeq); } + // Her m.read receipt, display-only: public, unthreaded, at the newest event each room's ack covers. + // Never awaited by the ack; a failure is one log line and nothing else. Posted only while her brake is + // released, read fresh when each receipt would go out; a held brake drops it (the next ack covers it). + async function readUpTo(newest) { + const quiet = message => { try { say(message); } catch { /* a throwing logger must not reject this unawaited call */ } }; + for (const { roomId, eventId } of newest.values()) { + if (!(await released())) { quiet("receipts not posted: brake held"); return; } // no receipt while braked; none is queued + try { + await matrix("POST", `/rooms/${encodeURIComponent(roomId)}/receipt/m.read/${encodeURIComponent(eventId)}`, {}, API, AbortSignal.timeout(sendTimeoutMs)); + } catch (error) { quiet(`receipt in ${roomId} not posted: ${error.message}`); } + } + } + async function ack(throughSeq) { + const newest = new Map(); // roomId -> the highest-seq delivery this ack covers + for (const d of pending) if (d.seq <= throughSeq && (newest.get(d.roomId)?.seq ?? 0) < d.seq) newest.set(d.roomId, d); pending = pending.filter(d => d.seq > throughSeq); images?.acked(throughSeq); await commit(pending.length ? pending[0].seq - 1 : seqCounter); + readUpTo(newest); } function pendingAfter(afterSeq) {