GameCult/Bifrost
Watch
9
0
Fork
You've already forked Bifrost
 0
Code Issues 7 Pull requests 5 Projects Releases Packages Wiki Activity Actions

Mouth read-through: serialize reports per room, keep the last landed receipt, re-read the brake, presence off the send path

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Heimdall 2026-11-10 14:33:25 +00:00
commit 0a44bf88e5
4 changed files with 139 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/bridge.md
View file

@ -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-<persona>`) 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/<server>/<mediaId>` 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/<persona>/<hex sha256 of the room id>/<hex sha256 of the bytes>[.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 `<room hash>/<file>` 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 posts nothing to Matrix. That Persona's public `m.read` receipt follows `gamecult.persona.chat_read.v1` (`[persona, room_alias, event_id, at_unix_ms]`), which Huginn sends when a delivery has gone into the model's input: the mouth posts `m.read` at that event id only, for an event it delivered to that Persona in that room, never behind a receipt already posted there, and only while her brake is released (fire and forget on the endpoint: a failure is logged once and not retried; pending polls and acks post none). Her Matrix presence follows a turn and her brake: `online` while a read report, a send or an ack came within `presenceGraceMs` (5 minutes), `unavailable` after that, `offline` while the brake is held. Each `/sync` carries the state as `set_presence`, and a change is also `PUT /presence/<user>/status` at once (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/<persona>/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-<persona>`) 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/<server>/<mediaId>` 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/<persona>/<hex sha256 of the room id>/<hex sha256 of the bytes>[.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 `<room hash>/<file>` 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 posts nothing to Matrix. That Persona's public `m.read` receipt follows `gamecult.persona.chat_read.v1` (`[persona, room_alias, event_id, at_unix_ms]`), which Huginn sends when a delivery has gone into the model's input: the mouth posts `m.read` at that event id only, for an event it delivered to that Persona in that room (to the delivered room id, whatever alias the report named), never behind the last receipt that landed there, one report per room at a time, and only while her brake is released (re-read after the presence write; fire and forget on the endpoint: a failure is logged once and not retried; pending polls and acks post none). Her Matrix presence follows a turn and her brake: `online` while a read report, a send or an ack came within `presenceGraceMs` (5 minutes), `unavailable` after that, `offline` while the brake is held. Each `/sync` carries the state as `set_presence`, and a change is also `PUT /presence/<user>/status` at once (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/<persona>/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.

13
tests/persona-matrix-mouth-runtime.test.mjs
View file

@ -132,10 +132,11 @@ test("two clients served at once each get their own answer on their own session"
});
test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a held brake with a Brake refusal", async () => {
const calls = { acks: [], sends: [] };
const calls = { acks: [], sends: [], reads: [] };
const mouth = {
pendingAfter: after => (after < 7 ? [delivery] : []),
ack: async through => { calls.acks.push(through); },
read: async request => { calls.reads.push(request); return null; },
send: async request => { calls.sends.push(request); return request.text === "muted" ? { requestId: request.requestId, eventId: null, refusal: "Brake" } : { requestId: request.requestId, eventId: "$sent", refusal: null }; },
};
const key = "connection-key-for-test";
@ -181,6 +182,16 @@ test("the RUDP endpoint serves pending, ack and send over CultNet, and answers a
assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload))], [docs.kinds.refusal, { persona: "other", refusal: "UnknownPersona" }], "an ack for an unlisted Persona is refused, typed, and not applied");
assert.deepEqual(calls.acks, [7]);
reply = nextReply();
put(docs.kinds.read, "other:#operator:gamecult.org", docs.readTuple({ persona: "other", roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 5 }));
document = await reply;
assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload))], [docs.kinds.refusal, { persona: "other", refusal: "UnknownPersona" }], "a read report for an unlisted Persona is refused, typed, and not applied");
assert.deepEqual(calls.reads, []);
put(docs.kinds.read, "eureka:#operator:gamecult.org", docs.readTuple({ persona: "eureka", roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 5 }));
for (let i = 0; i < 100 && calls.reads.length < 1; i++) await new Promise(r => setTimeout(r, 20));
assert.deepEqual(calls.reads, [{ persona: "eureka", roomAlias: "#operator:gamecult.org", eventId: "$e1", atUnixMs: 5 }], "her own report reaches her mouth with its persona");
reply = nextReply();
put(docs.kinds.pendingRequest, "other", docs.pendingRequestTuple({ persona: "other", afterSeq: 0 }));
document = await reply;

90
tests/persona-matrix-mouth.test.mjs
View file

@ -973,7 +973,7 @@ const withMail = async (hs, overrides) => {
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]);
const presencesOf = hs => hs.presencePuts().map(r => r.body.presence);
const syncsOf = hs => hs.seen.filter(r => r.path.endsWith("/sync")).map(r => r.query.get("set_presence"));
const reportRead = (mouth, roomAlias, eventId) => mouth.read({ roomAlias, eventId, atUnixMs: 1 });
const reportRead = (mouth, roomAlias, eventId, persona = "eureka") => mouth.read({ persona, roomAlias, eventId, atUnixMs: 1 });
test("delivery, pending and an ack post no receipt: she has not read what the model has not seen", async () => {
const hs = await fakeHomeserver({ sync: receiptScript });
@ -1026,6 +1026,88 @@ test("a read report posts nothing while braked, and nothing is queued for the re
assert.equal(await reportRead(mouth, A, "$a1"), null, "a fresh report after the release posts");
});
test("a read report names her persona: another persona's report posts nothing", async () => {
const hs = await fakeHomeserver({ sync: receiptScript });
const mouth = await withMail(hs);
assert.equal(await reportRead(mouth, A, "$a1", "someone-else"), "PersonaMismatch");
assert.deepEqual(hs.receipts(), []);
});
test("the receipt goes to the delivered room id, not to the alias the report named", async () => {
const hs = await fakeHomeserver({ sync: receiptScript });
const mouth = await withMail(hs);
assert.equal(await reportRead(mouth, `#swarm:${HS_NAME}`, "$a1"), null);
assert.deepEqual(receiptsOf(hs), [[A, "$a1", `Bearer ${TOKEN}`, {}]]);
});
test("concurrent reports for one room are judged in arrival order: the newer lands and the older is behind", async () => {
const hs = await fakeHomeserver({ sync: receiptScript });
const mouth = await withMail(hs);
const [newer, older] = await Promise.all([reportRead(mouth, A, "$a2"), reportRead(mouth, A, "$a1")]);
assert.deepEqual([newer, older], [null, "Behind"]);
assert.deepEqual(receiptsOf(hs).map(r => r[1]), ["$a2"], "never a1 after a2");
const [first, second] = await Promise.all([reportRead(mouth, B, "$b1"), reportRead(mouth, B, "$b1")]);
assert.deepEqual([first, second], [null, "Behind"], "the same event twice posts once");
});
test("a failed receipt leaves the last landed receipt in place: an older event is still behind", async () => {
let receiptCalls = 0;
const hs = await fakeHomeserver({ sync: receiptScript });
const fetch = (url, init) => String(url).includes("/receipt/") && ++receiptCalls === 2 ? Promise.resolve(new Response("{}", { status: 500 })) : hs.fetch(url, init);
const mouth = await withMail(hs, { fetch });
assert.equal(await reportRead(mouth, A, "$a1"), null);
assert.equal(await reportRead(mouth, A, "$a2"), "Failed");
assert.equal(await reportRead(mouth, A, "$a1"), "Behind", "a1 landed; the failure of a2 does not reopen the room to a backward post");
assert.equal(await reportRead(mouth, A, "$a2"), null, "a2 itself can still land");
assert.deepEqual(receiptsOf(hs).map(r => r[1]), ["$a1", "$a2"]);
});
test("a brake placed after the first brake read stops the receipt and no present state is PUT while braked", async () => {
let used = 0, allowed = Infinity;
const hs = await fakeHomeserver({ sync: receiptScript });
const mouth = await withMail(hs, { brake: async () => used++ < allowed });
used = 0; allowed = 1; // read() sees it released once; every read after that finds it held
assert.equal(await reportRead(mouth, A, "$a1"), "Brake");
assert.deepEqual(hs.receipts(), [], "the receipt re-reads the brake after the presence write");
assert.ok(!presencesOf(hs).includes("online"), "online is never PUT on a stale brake read");
assert.equal(presencesOf(hs).at(-1), "offline");
});
test("a slow presence write after a send holds no other send in the room", async () => {
const hs = await fakeHomeserver({ sync: receiptScript });
const fetch = (url, init) => String(url).includes("/presence/") && init?.method === "PUT"
? new Promise((_, reject) => init.signal.addEventListener("abort", () => reject(new Error("aborted")))) : hs.fetch(url, init);
const mouth = await withMail(hs, { fetch, sendTimeoutMs: 400 });
const send = id => mouth.send({ requestId: id, persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: id });
void send("s1"); // its presence write hangs
const second = await Promise.race([send("s2").then(r => r.refusal), new Promise(r => setTimeout(() => r("held behind the first send's presence write"), 150))]);
assert.equal(second, null);
});
test("the grace ends exactly at its boundary: one millisecond inside is online, the boundary is unavailable", async () => {
let t = 9_000_000;
const hs = await fakeHomeserver({ sync: receiptScript });
const mouth = await withMail(hs, { now: () => t, presenceGraceMs: 60_000 });
await mouth.ack(3); // touched at t
t += 59_999;
await mouth.syncOnce();
assert.equal(syncsOf(hs).at(-1), "online");
t += 1;
await mouth.syncOnce();
assert.equal(syncsOf(hs).at(-1), "unavailable");
});
test("the state a poll carries is fixed before the poll: a grace that passes while it is open is PUT next round", async () => {
let t = 3_000_000;
const hs = await fakeHomeserver({ sync: receiptScript });
const fetch = async (url, init) => { const res = await hs.fetch(url, init); if (String(url).includes("/sync")) t += 100_000; return res; };
const mouth = mouthFor(hs, { now: () => t, presenceGraceMs: 60_000, fetch });
await mouth.ack(0); // a turn: online
await mouth.syncOnce();
assert.equal(syncsOf(hs).at(-1), "online");
await mouth.syncOnce();
assert.deepEqual(presencesOf(hs), ["online", "unavailable"], "the homeserver was told online by the poll; the next round must tell it unavailable");
});
test("a failed receipt is logged once, not retried, and the next report for the room posts", async () => {
const hs = await fakeHomeserver({ sync: receiptScript, failReceipts: true });
const lines = [];
@ -1144,11 +1226,11 @@ test("two Personas on one homeserver: receipts and presence carry only her own b
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 reportRead(a, A, "$a1");
assert.equal(await reportRead(b, A, "$a1"), "Brake");
await reportRead(a, A, "$a1", "pa");
assert.equal(await reportRead(b, A, "$a1", "pb"), "Brake");
assert.ok(hs.receipts().every(r => r.auth === "Bearer tokA"), "only A read");
bOpen = true; // her receipts go out only while released
assert.equal(await reportRead(b, A, "$a1"), null);
assert.equal(await reportRead(b, A, "$a1", "pb"), null);
assert.deepEqual([...new Set(hs.receipts().map(r => r.auth))].sort(), ["Bearer tokA", "Bearer tokB"]);
assert.ok(hs.presencePuts().some(r => r.path.endsWith(`/presence/${BOB}/status`) && r.auth === "Bearer tokB"));
assert.ok(hs.presencePuts().filter(r => r.path.endsWith(`/presence/${BOB}/status`)).every(r => r.auth === "Bearer tokB"));

66
tools/persona-matrix-mouth.mjs
View file

@ -133,7 +133,8 @@ export function createMouth({
let lastPresence = null; // 'online' | 'unavailable' | 'offline' | null: command-only, suppresses repeat presence PUTs
let activeUntil = 0; // her last sign of life in a turn, plus the grace
const delivered = new Map(); // event id -> { roomId, roomAlias, seq }: what the mouth handed her, newest last, bounded
const readThrough = new Map(); // room id -> the highest delivery seq a receipt was posted for
const readThrough = new Map(); // room id -> { seq, eventId } of the last receipt that landed
const readTails = new Map(); // room id -> the read report in flight: reports for one room run one at a time
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
@ -294,7 +295,9 @@ export function createMouth({
const presenceFor = free => !free ? "offline" : now() < activeUntil ? "online" : "unavailable";
const touch = () => { activeUntil = now() + presenceGraceMs; };
async function setPresence(free) {
const wanted = presenceFor(free);
let wanted = presenceFor(free);
if (wanted === lastPresence) return;
if (wanted !== "offline" && !(await released())) wanted = "offline"; // never PUT a present state on a stale brake read
if (wanted === lastPresence) return;
try {
await matrix("PUT", `/presence/${encodeURIComponent(selfId)}/status`, { presence: wanted }, API, AbortSignal.timeout(sendTimeoutMs));
@ -310,8 +313,9 @@ export function createMouth({
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}&set_presence=${presenceFor(free)}&filter=${filter}${since}`);
if (free) lastPresence = presenceFor(free); // a braked round's offline is only believed once its own PUT lands
const stateSent = presenceFor(free); // the state this poll carries, fixed before the poll: the grace may pass while it is open
const reply = await matrix("GET", `/sync?timeout=${cursor === null ? 0 : syncTimeoutMs}&set_presence=${stateSent}&filter=${filter}${since}`);
if (free) lastPresence = stateSent; // a braked round's offline is only believed once its own PUT lands
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);
@ -378,27 +382,37 @@ export function createMouth({
batches = batches.filter(b => b.maxSeq > throughSeq);
}
// Her m.read receipt, display-only: public, unthreaded, at the event Huginn reports she has taken in. Only an event
// this mouth delivered to her, in the room it was delivered for, and only forward of what was already posted in
// that room. Posted only while her brake is released, read fresh. Returns the reason nothing was posted, or null.
async function read({ roomAlias, eventId }) {
// Her m.read receipt, display-only: public, unthreaded, at the event Huginn reports she has taken in. Only for her own
// persona, an event this mouth delivered to her, in the room it was delivered for (the receipt goes to the delivered
// room id, never to the reported alias), and only forward of the last receipt that landed in that room. Reports for
// one room run one at a time, so the order they arrive in is the order they are judged in. Posted only while her
// brake is released, read fresh before the presence write and again before the POST. Returns the reason nothing
// was posted, or null.
function read({ persona: asked, roomAlias, eventId }) {
const quiet = message => { try { say(message); } catch { /* a throwing logger must not reject a caller that does not await */ } };
if (asked !== persona) return Promise.resolve("PersonaMismatch");
const known = delivered.get(eventId);
if (!known) { quiet("read report names an event this mouth never delivered"); return "UnknownEvent"; }
if (roomAlias !== known.roomAlias && roomAlias !== known.roomId) { quiet("read report names another room than the delivery"); return "RoomMismatch"; }
if ((readThrough.get(known.roomId) ?? 0) >= known.seq) return "Behind";
if (!(await released())) { quiet("receipt not posted: brake held"); return "Brake"; } // none is queued
readThrough.set(known.roomId, known.seq); // claimed before the POST: a concurrent report for an older event posts nothing
touch();
await setPresence(true);
try {
await matrix("POST", `/rooms/${encodeURIComponent(known.roomId)}/receipt/m.read/${encodeURIComponent(eventId)}`, {}, API, AbortSignal.timeout(sendTimeoutMs));
} catch (error) {
if (readThrough.get(known.roomId) === known.seq) readThrough.delete(known.roomId);
quiet(`receipt in ${known.roomId} not posted: ${error.message}`);
return "Failed";
}
return null;
if (!known) { quiet("read report names an event this mouth never delivered"); return Promise.resolve("UnknownEvent"); }
if (roomAlias !== known.roomAlias && roomAlias !== known.roomId) { quiet("read report names another room than the delivery"); return Promise.resolve("RoomMismatch"); }
const run = async () => {
if ((readThrough.get(known.roomId)?.seq ?? 0) >= known.seq) return "Behind";
if (!(await released())) { quiet("receipt not posted: brake held"); return "Brake"; } // none is queued
touch();
await setPresence(true);
if (!(await released())) { quiet("receipt not posted: brake held"); return "Brake"; }
try {
await matrix("POST", `/rooms/${encodeURIComponent(known.roomId)}/receipt/m.read/${encodeURIComponent(eventId)}`, {}, API, AbortSignal.timeout(sendTimeoutMs));
} catch (error) {
quiet(`receipt in ${known.roomId} not posted: ${error.message}`);
return "Failed"; // the last receipt that landed stays what it was: a later report for an older event is still Behind
}
readThrough.set(known.roomId, { seq: known.seq, eventId });
return null;
};
const tail = (readTails.get(known.roomId) ?? Promise.resolve()).then(run);
readTails.set(known.roomId, tail);
tail.then(() => { if (readTails.get(known.roomId) === tail) readTails.delete(known.roomId); });
return tail;
}
async function ack(throughSeq) {
@ -435,17 +449,17 @@ export function createMouth({
if (!started || unjoined()) await start();
const roomId = joinedAliases.has(roomAlias) ? roomIdByAlias.get(roomAlias) : directRooms.has(roomAlias) ? roomAlias : null;
if (!roomId) return refuse("UnknownRoom");
return await gate.run(roomId, async () => {
const result = await gate.run(roomId, async () => {
if (basis) {
const newer = await newerThan(roomId, basis);
if (newer.length > 0) return refuse("RoomMoved", newer);
}
if (!(await released())) return refuse("Brake"); // a brake placed during the gate wait or the head read stops this send
const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }, API, AbortSignal.timeout(sendTimeoutMs));
touch();
await setPresence(true);
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
return result;
} catch (error) {
say(`send failed: ${error.message}`);
return refuse("SendFailed");