tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.declined.msgpack

@@ -0,0 +1 @@
+”¥req-3¨DeclinedÀÀ

tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.existing.msgpack

@@ -0,0 +1 @@
+”¥req-2¨Existing±!dm1:gamecult.org¦$line2

tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_receipt.v1.opened.msgpack

@@ -0,0 +1 @@
+”¥req-1¦Opened±!dm1:gamecult.org¦$line1

tests/fixtures/persona-chat-v4/gamecult.persona.chat_open_direct_request.v1.msgpack

@@ -0,0 +1 @@
+”¥req-1¦eureka²@meta:gamecult.orgªhello Meta

tests/fixtures/persona-chat-v4/gamecult.persona.chat_refusal.v1.unknown-kind.msgpack

@@ -0,0 +1 @@
+’¦eureka«UnknownKind

tests/persona-matrix-mouth-open-direct.test.mjs

@@ -0,0 +1,344 @@
+// A Persona asks her mouth for a direct room with one person. The homeserver here keeps room state the way Synapse does
+// (create event with its content, one member event per account, encryption event when set), so the mouth's lookup,
+// the consent that lives in the person's own membership, and the order of its writes are exercised, not mocked.
+import test from "node:test";
+import assert from "node:assert/strict";
+import { createServer } from "node:http";
+import { createMouth } from "../tools/persona-matrix-mouth.mjs";
+
+const HS = "gamecult.org";
+const ME = `@eureka:${HS}`;
+const META = `@meta:${HS}`;
+const PROJECT = "!project:gamecult.org";
+const ALIAS = `#swarm:${HS}`;
+
+async function homeserver({ people = [META, `@nova:${HS}`], now = () => 1000, failInvite: initialFailInvite = false } = {}) {
+  const config = { failInvite: initialFailInvite };
+  const rooms = new Map(); // room id -> { events }
+  const stateReads = [];
+  const log = []; // every write, in order, as "METHOD path"
+  let counter = 0;
+  const direct = { value: null };
+  const member = (room, user, membership) => {
+    const events = rooms.get(room).events.filter(e => !(e.type === "m.room.member" && e.state_key === user));
+    events.push({ type: "m.room.member", state_key: user, origin_server_ts: now(), content: { membership } });
+    rooms.get(room).events = events;
+  };
+  const server = createServer((req, res) => {
+    let raw = "";
+    req.on("data", c => (raw += c));
+    req.on("end", () => {
+      const url = new URL(req.url, "http://x");
+      const path = decodeURIComponent(url.pathname).replace("/_matrix/client/v3", "");
+      const body = raw ? JSON.parse(raw) : null;
+      const reply = (status, json) => { res.writeHead(status, { "content-type": "application/json" }); res.end(JSON.stringify(json)); };
+      const write = () => { if (!path.startsWith("/presence/") && !path.includes("/account_data/")) log.push(`${req.method} ${path}`); };
+      if (path === "/account/whoami") return reply(200, { user_id: ME });
+      if (path === "/joined_rooms") return reply(200, { joined_rooms: [PROJECT, ...[...rooms].filter(([, r]) => r.events.some(e => e.type === "m.room.member" && e.state_key === ME && e.content.membership === "join")).map(([id]) => id)] });
+      if (path.includes("/account_data/m.direct")) {
+        if (req.method === "PUT") { direct.value = body; return reply(200, {}); }
+        return direct.value ? reply(200, direct.value) : reply(404, { errcode: "M_NOT_FOUND" });
+      }
+      if (path.startsWith("/directory/room/")) return reply(200, { room_id: PROJECT });
+      if (path.startsWith("/profile/")) return people.includes(path.slice("/profile/".length)) ? reply(200, { displayname: "x" }) : reply(404, { errcode: "M_NOT_FOUND" });
+      const state = path.match(/^\/rooms\/(.+)\/state$/);
+      if (state) stateReads.push(state[1]);
+      if (state) return rooms.has(state[1]) ? reply(200, rooms.get(state[1]).events) : reply(403, {});
+      if (req.method === "POST" && path === "/createRoom") {
+        write();
+        const id = `!dm${++counter}:${HS}`;
+        rooms.set(id, { events: [{ type: "m.room.create", state_key: "", origin_server_ts: now() + counter, content: { creator: ME, ...body.creation_content } }] });
+        if (body.preset !== "private_chat") return reply(400, {});
+        member(id, ME, "join");
+        rooms.get(id).body = body;
+        if (body.initial_state?.some(e => e.type === "m.room.encryption")) rooms.get(id).events.push({ type: "m.room.encryption", state_key: "", content: {} });
+        return reply(200, { room_id: id });
+      }
+      const invite = path.match(/^\/rooms\/(.+)\/invite$/);
+      if (invite) { write(); if (config.failInvite) return reply(403, { errcode: "M_FORBIDDEN" }); member(invite[1], body.user_id, "invite"); rooms.get(invite[1]).invited = body; return reply(200, {}); }
+      const put = path.match(/^\/rooms\/(.+)\/send\/m\.room\.message\/(.+)$/);
+      if (put) {
+        write();
+        const r = rooms.get(put[1]);
+        r.sent ??= new Map();
+        if (!r.sent.has(put[2])) r.sent.set(put[2], { id: `$e${counter}-${r.sent.size + 1}`, body });
+        return reply(200, { event_id: r.sent.get(put[2]).id });
+      }
+      reply(404, { errcode: "M_NOT_FOUND" });
+    });
+  });
+  await new Promise(ok => server.listen(0, "127.0.0.1", ok));
+  return {
+    url: `http://127.0.0.1:${server.address().port}`, rooms, log, direct, stateReads,
+    set failInvite(v) { config.failInvite = v; },
+    seedDm(id, user, membership, { encrypted = false, created = 5, marker } = {}) {
+      rooms.set(id, { events: [{ type: "m.room.create", state_key: "", origin_server_ts: created, content: { creator: ME, ...(marker ? { "org.gamecult.open_request": marker } : {}) } }] });
+      member(id, ME, "join"); member(id, user, membership);
+      if (encrypted) rooms.get(id).events.push({ type: "m.room.encryption", state_key: "", content: {} });
+    },
+    set: member,
+    writes: pattern => log.filter(l => pattern.test(l)),
+    close: () => { server.closeAllConnections(); return new Promise(ok => server.close(ok)); },
+  };
+}
+
+const memoryState = () => { const box = { value: null }; return { load: async () => box.value, save: async v => { box.value = v; } }; };
+const mouthFor = (hs, extra = {}) => createMouth({ persona: "eureka", homeserver: hs.url, serverName: HS, rooms: [ALIAS], token: "tok", state: memoryState(), brake: async () => true, syncTimeoutMs: 1, ...extra });
+const ask = (m, requestId, user = META, text = "hello Meta, may I ask you something?", persona = "eureka") => m.openDirect({ requestId, persona, userId: user, text });
+const outcome = r => r.outcome;
+
+test("a new person: creates an unencrypted private direct room, invites with the line as the reason, says the line with the not-E2EE notice, and answers Opened", async () => {
+  const hs = await homeserver();
+  try {
+    const m = mouthFor(hs);
+    const r = await ask(m, "req-1");
+    assert.equal(r.outcome, "Opened");
+    const [roomId] = [...hs.rooms.keys()];
+    assert.equal(r.roomId, roomId);
+    const room = hs.rooms.get(roomId);
+    assert.equal(room.body.preset, "private_chat");
+    assert.equal(room.body.is_direct, true);
+    assert.equal(room.body.creation_content["m.federate"], false);
+    assert.equal(room.body.creation_content["org.gamecult.open_request"], "req-1");
+    assert.ok(!JSON.stringify(room.body).includes("m.room.encryption"), "the room is created without encryption");
+    assert.ok(!room.events.some(e => e.type === "m.room.encryption"));
+    assert.deepEqual(room.invited, { user_id: META, reason: "hello Meta, may I ask you something?" });
+    const line = room.sent.get("req-1");
+    assert.equal(r.eventId, line.id);
+    assert.match(line.body.body, /not end-to-end encrypted/, "the first message says so");
+    assert.ok(line.body.body.endsWith("hello Meta, may I ask you something?"));
+    assert.deepEqual(hs.direct.value, { [META]: [roomId] }, "the room id is stored in m.direct");
+    assert.deepEqual(hs.log.map(l => l.split(" ")[0] + " " + l.split("/").pop()), ["POST createRoom", "POST invite", "PUT req-1"], "order of writes");
+  } finally { await hs.close(); }
+});
+
+test("the invite reason is the line cut to 300 bytes", async () => {
+  const hs = await homeserver();
+  try {
+    await ask(mouthFor(hs), "req-1", META, "x".repeat(900));
+    assert.equal(Buffer.byteLength([...hs.rooms.values()][0].invited.reason), 300);
+  } finally { await hs.close(); }
+});
+
+test("only her own request is served, and only for a person: mismatch, empty text, a foreign server, herself, another Mind, an unknown account", async () => {
+  const hs = await homeserver();
+  try {
+    const nova = `@nova:${HS}`;
+    const m = mouthFor(hs, { isServed: id => id === nova });
+    assert.equal(outcome(await ask(m, "r1", META, "hi", "nova")), "PersonaMismatch");
+    assert.equal(outcome(await ask(m, "r2", META, "   ")), "EmptyText");
+    assert.equal(outcome(await ask(m, "r3", "@meta:matrix.org")), "ForeignUser");
+    assert.equal(outcome(await ask(m, "r3b", "@meta:gamecult.org.evil.example")), "ForeignUser");
+    assert.equal(outcome(await ask(m, "r4", ME)), "NotAPerson");
+    assert.equal(outcome(await ask(m, "r5", nova)), "NotAPerson");
+    assert.equal(outcome(await ask(m, "r6", `@nobody:${HS}`)), "NoSuchUser");
+    assert.deepEqual(hs.log, [], "no write for any refusal");
+    assert.equal(hs.rooms.size, 0);
+  } finally { await hs.close(); }
+});
+
+test("a held brake refuses before any Matrix call", async () => {
+  const hs = await homeserver();
+  try {
+    const m = mouthFor(hs, { brake: async () => false });
+    assert.equal(outcome(await ask(m, "r1")), "Brake");
+    assert.deepEqual(hs.log, []);
+    assert.equal(hs.rooms.size, 0);
+  } finally { await hs.close(); }
+});
+
+test("the brake is read fresh before createRoom, before the invite and before the line: flipping it at any point stops every later write", async () => {
+  for (let free = 1; free <= 5; free++) {
+    const hs = await homeserver();
+    try {
+      let reads = 0;
+      const events = [];
+      const m = mouthFor(hs, { brake: async () => { const ok = ++reads <= free; events.push(ok ? "free" : "held"); return ok; } });
+      const originalLog = hs.log;
+      const push = originalLog.push.bind(originalLog);
+      originalLog.push = (...x) => { events.push(`write ${x[0]}`); return push(...x); };
+      await ask(m, "r1");
+      let held = false;
+      for (const e of events) {
+        if (e === "held") held = true;
+        else if (held && e.startsWith("write")) assert.fail(`free=${free}: a write after the brake was held: ${e}`);
+      }
+      const writes = events.filter(e => e.startsWith("write"));
+      for (let i = 0; i < events.length; i++) if (events[i].startsWith("write")) assert.equal(events[i - 1], "free", `free=${free}: ${events[i]} was not immediately preceded by a released brake read`);
+      assert.equal(writes.length, Math.min(3, free - 1), `free=${free}: reads 2, 3 and 4 each gate exactly the createRoom, the invite and the line`);
+    } finally { await hs.close(); }
+  }
+});
+
+test("a second request for a person who accepted says the line in the same room and answers Existing, with no new room", async () => {
+  const hs = await homeserver();
+  try {
+    const m = mouthFor(hs);
+    const first = await ask(m, "r1");
+    hs.set(first.roomId, META, "join"); // Meta accepts
+    const again = await ask(m, "r2", META, "second line");
+    assert.deepEqual([again.outcome, again.roomId], ["Existing", first.roomId]);
+    assert.equal(hs.rooms.size, 1);
+    const sent = hs.rooms.get(first.roomId).sent.get("r2");
+    assert.equal(again.eventId, sent.id);
+    assert.equal(sent.body.body, "second line", "no notice again after the first message");
+  } finally { await hs.close(); }
+});
+
+test("a repeat of the same request id after a crash resumes in the room it made: one room, one line, Opened again", async () => {
+  const hs = await homeserver();
+  try {
+    const first = await ask(mouthFor(hs), "r1");
+    const again = await ask(mouthFor(hs), "r1"); // a restarted process: no memory of the first
+    assert.deepEqual([again.outcome, again.roomId, again.eventId], ["Opened", first.roomId, first.eventId]);
+    assert.equal(hs.rooms.size, 1);
+    assert.equal(hs.writes(/createRoom/).length, 1);
+    assert.equal(hs.writes(/\/invite/).length, 1);
+  } finally { await hs.close(); }
+});
+
+test("an invite the person has not answered is InvitePending: nothing new is written", async () => {
+  const hs = await homeserver();
+  try {
+    const m = mouthFor(hs);
+    await ask(m, "r1");
+    const before = hs.log.length;
+    const again = await ask(m, "r2");
+    assert.equal(again.outcome, "InvitePending");
+    assert.equal(hs.log.length, before);
+  } finally { await hs.close(); }
+});
+
+test("a decline stands: leaving or banning is Declined, for a restarted mouth too, until the person invites her", async () => {
+  for (const gone of ["leave", "ban"]) {
+    const hs = await homeserver();
+    try {
+      const first = await ask(mouthFor(hs), "r1");
+      hs.set(first.roomId, META, gone);
+      const before = hs.log.length;
+      for (const m of [mouthFor(hs), mouthFor(hs)]) {
+        assert.equal(outcome(await ask(m, `again-${Math.random()}`)), "Declined", gone);
+      }
+      assert.equal(hs.log.length, before, "no room, invite or line");
+      // the person writes first: she is invited into a new room and joins; now there is a live DM
+      hs.seedDm("!theirs:gamecult.org", META, "join", { created: 99999 });
+      const answer = await ask(mouthFor(hs), "r-after");
+      assert.deepEqual([answer.outcome, answer.roomId], ["Existing", "!theirs:gamecult.org"]);
+    } finally { await hs.close(); }
+  }
+});
+
+test("an invite the homeserver refuses leaves the room made; the retry invites into that same room instead of making another", async () => {
+  const hs = await homeserver({ failInvite: true });
+  try {
+    assert.equal(outcome(await ask(mouthFor(hs), "r1")), "SendFailed");
+    assert.equal(hs.rooms.size, 1);
+    hs.failInvite = false;
+    const retry = await ask(mouthFor(hs), "r2");
+    assert.equal(retry.outcome, "Opened");
+    assert.equal(hs.rooms.size, 1, "no second room");
+    assert.equal(retry.roomId, [...hs.rooms.keys()][0]);
+    assert.equal(hs.rooms.get(retry.roomId).invited.user_id, META);
+  } finally { await hs.close(); }
+});
+
+test("one creation per person per day: a second creation the same UTC day is DailyLimit, the next UTC day it is allowed", async () => {
+  let t = Date.parse("2026-11-10T10:00:00Z");
+  const hs = await homeserver({ now: () => t });
+  try {
+    const m = mouthFor(hs, { now: () => t });
+    assert.equal(outcome(await ask(m, "r1")), "Opened");
+    hs.rooms.clear(); // she left the room: no DM with them is found now, only the day's record stands
+    assert.equal(outcome(await ask(m, "r2")), "DailyLimit");
+    assert.equal(hs.writes(/createRoom/).length, 1);
+    t += 24 * 3600 * 1000;
+    assert.equal(outcome(await ask(m, "r3")), "Opened");
+    assert.equal(outcome(await ask(m, "r4", `@nova:${HS}`)), "Opened", "the limit is per person");
+  } finally { await hs.close(); }
+});
+
+test("an encrypted direct room is not used: the line is never sent into it and a new unencrypted room is made", async () => {
+  const hs = await homeserver();
+  try {
+    hs.seedDm("!old:gamecult.org", META, "join", { encrypted: true });
+    const r = await ask(mouthFor(hs), "r1");
+    assert.equal(r.outcome, "Opened");
+    assert.notEqual(r.roomId, "!old:gamecult.org");
+    assert.equal(hs.rooms.get("!old:gamecult.org").sent, undefined);
+  } finally { await hs.close(); }
+});
+
+test("a homeserver failure answers SendFailed and writes nothing the brake or a refusal should have stopped", async () => {
+  const hs = await homeserver();
+  try {
+    const m = mouthFor(hs, { homeserver: "http://127.0.0.1:9" });
+    assert.equal(outcome(await ask(m, "r1")), "SendFailed");
+  } finally { await hs.close(); }
+});
+
+test("two requests for one person at once make one room: opens for a person run one at a time", async () => {
+  const hs = await homeserver();
+  try {
+    const m = mouthFor(hs);
+    const answers = await Promise.all([ask(m, "a"), ask(m, "b")]);
+    assert.equal(hs.writes(/createRoom/).length, 1);
+    assert.deepEqual(answers.map(outcome).sort(), ["InvitePending", "Opened"]);
+    const other = await Promise.all([ask(m, "c", `@nova:${HS}`), ask(m, "d", META)]);
+    assert.equal(other[0].outcome, "Opened", "another person is not held behind");
+  } finally { await hs.close(); }
+});
+
+test("a decline stands in an encrypted room too, and a pending invite there is still pending: no new room, no line", async () => {
+  for (const [membership, expected] of [["leave", "Declined"], ["invite", "InvitePending"]]) {
+    const hs = await homeserver();
+    try {
+      hs.seedDm("!enc:gamecult.org", META, membership, { encrypted: true });
+      assert.equal(outcome(await ask(mouthFor(hs), "r1")), expected);
+      assert.deepEqual(hs.log, []);
+    } finally { await hs.close(); }
+  }
+});
+
+test("the not-E2EE notice reaches the person even when the first line was held back: a later line into the room this path made carries it once", async () => {
+  const hs = await homeserver();
+  try {
+    let reads = 0;
+    const m = mouthFor(hs, { brake: async () => ++reads <= 3 }); // top, createRoom, invite: then the brake is held before the line
+    assert.equal(outcome(await ask(m, "r1")), "Brake");
+    const roomId = [...hs.rooms.keys()][0];
+    assert.equal(hs.rooms.get(roomId).sent, undefined, "the invite went out, the line did not");
+    hs.set(roomId, META, "join");
+    const answer = await ask(mouthFor(hs), "r2", META, "second");
+    assert.deepEqual([answer.outcome, answer.roomId], ["Existing", roomId]);
+    assert.match(hs.rooms.get(roomId).sent.get("r2").body.body, /^This room is not end-to-end encrypted.*\n\nsecond$/s);
+    await ask(mouthFor(hs), "r3", META, "third"); // another process: the notice is repeated at worst once, never lost
+    const sameProcess = mouthFor(hs);
+    await ask(sameProcess, "r4", META, "fourth");
+    await ask(sameProcess, "r5", META, "fifth");
+    assert.equal(hs.rooms.get(roomId).sent.get("r5").body.body, "fifth", "once the notice went out in this process it is not repeated");
+  } finally { await hs.close(); }
+});
+
+test("the lookup finds a DM only: a group room with the person is not one, and project rooms are never scanned", async () => {
+  const hs = await homeserver();
+  try {
+    hs.seedDm("!group:gamecult.org", META, "join");
+    hs.set("!group:gamecult.org", `@third:${HS}`, "join"); // three members
+    const r = await ask(mouthFor(hs), "r1");
+    assert.equal(r.outcome, "Opened");
+    assert.notEqual(r.roomId, "!group:gamecult.org");
+    assert.equal(hs.rooms.get("!group:gamecult.org").sent, undefined);
+    assert.ok(!hs.stateReads.includes("!project:gamecult.org"), "a configured project room is not read as a DM candidate");
+  } finally { await hs.close(); }
+});
+
+test("another Mind this process serves is not a person, even before that Mind's own mouth has started", async () => {
+  const hs = await homeserver();
+  try {
+    const nova = `@nova:${HS}`;
+    let asked = null;
+    const m = mouthFor(hs, { isServed: async id => { asked = id; return id === nova; } });
+    assert.equal(outcome(await ask(m, "r1", nova)), "NotAPerson");
+    assert.equal(asked, nova);
+  } finally { await hs.close(); }
+});

tests/persona-matrix-mouth-runtime.test.mjs

@@ -438,3 +438,86 @@ test("over the RUDP endpoint: pending and an ack post no receipt, a read report
     try { client.close(); } catch { /* already closed */ }
   }
 });
+
+// The open-direct pair (v4): the bytes Huginn pins in its crates/mind-body/tests/fixtures/persona-chat-v4/.
+const v4Dir = resolve(root, "tests", "fixtures", "persona-chat-v4");
+const v4Samples = {
+  "gamecult.persona.chat_open_direct_request.v1": { tuple: docs.openDirectRequestTuple({ requestId: "req-1", persona: "eureka", userId: "@meta:gamecult.org", text: "hello Meta" }), parse: docs.parseOpenDirectRequest },
+  "gamecult.persona.chat_open_direct_receipt.v1.opened": { tuple: docs.openDirectReceiptTuple({ requestId: "req-1", outcome: "Opened", roomId: "!dm1:gamecult.org", eventId: "$line1" }), parse: docs.parseOpenDirectReceipt },
+  "gamecult.persona.chat_open_direct_receipt.v1.existing": { tuple: docs.openDirectReceiptTuple({ requestId: "req-2", outcome: "Existing", roomId: "!dm1:gamecult.org", eventId: "$line2" }), parse: docs.parseOpenDirectReceipt },
+  "gamecult.persona.chat_open_direct_receipt.v1.declined": { tuple: docs.openDirectReceiptTuple({ requestId: "req-3", outcome: "Declined" }), parse: docs.parseOpenDirectReceipt },
+  "gamecult.persona.chat_refusal.v1.unknown-kind": { tuple: docs.refusalTuple({ persona: "eureka", refusal: "UnknownKind" }), parse: docs.parseRefusal },
+};
+for (const [name, { tuple, parse }] of Object.entries(v4Samples)) {
+  test(`golden fixture: ${name} is byte-stable and parses back`, async () => {
+    const bytes = Buffer.from(msgpack.encode(tuple));
+    const file = resolve(v4Dir, `${name}.msgpack`);
+    if (process.env.UPDATE_FIXTURES === "1") { await mkdir(v4Dir, { recursive: true }); await writeFile(file, bytes); }
+    assert.deepEqual(bytes, await readFile(file), "wire bytes changed: Huginn decodes these fixtures");
+    assert.deepEqual(parse(msgpack.decode(bytes)), parse(tuple));
+  });
+}
+
+test("the open-direct documents are strict: room and event ids exactly for Opened and Existing, a known outcome, a bounded line", () => {
+  for (const outcome of docs.directOutcomes) {
+    const placed = outcome === "Opened" || outcome === "Existing";
+    assert.doesNotThrow(() => docs.openDirectReceiptTuple(placed ? { requestId: "r", outcome, roomId: "!a:b", eventId: "$e" } : { requestId: "r", outcome }), outcome);
+    assert.throws(() => docs.openDirectReceiptTuple(placed ? { requestId: "r", outcome } : { requestId: "r", outcome, roomId: "!a:b", eventId: "$e" }), /exactly for Opened and Existing/, outcome);
+    assert.throws(() => docs.parseOpenDirectReceipt(["r", outcome, placed ? "!a:b" : null, placed ? null : "$e"]), /exactly for/, `${outcome}: event id alone is half a placement`);
+  }
+  assert.throws(() => docs.openDirectReceiptTuple({ requestId: "r", outcome: "Invented" }), /unknown open-direct outcome/);
+  assert.throws(() => docs.parseOpenDirectRequest(["r", "eureka", "@a:b", "x".repeat(4001)]), /byte bound/);
+  assert.throws(() => docs.parseOpenDirectRequest(["r", "eureka", "", "x"]), /user_id/);
+  assert.throws(() => docs.parseOpenDirectRequest(["r", "eureka", "@a:b"]), /4-field/);
+  assert.throws(() => docs.sendReceiptTuple({ requestId: "r", refusal: "UnknownKind" }), /unknown refusal/, "UnknownKind exists only in the typed refusal document");
+});
+
+test("the RUDP endpoint answers an open-direct request with a receipt for its Persona, refuses another Persona's and answers a request of an unknown kind loudly", async () => {
+  const opened = [];
+  const mouth = { openDirect: async request => { opened.push(request); return { requestId: request.requestId, outcome: "Opened", roomId: "!dm1:gamecult.org", eventId: "$line1" }; } };
+  const key = "connection-key-for-test";
+  const logged = [];
+  const server = await startChatEndpoint({ mouths: new Map([["eureka", mouth]]), runtime, port: 0, connectionKey: key, runtimeId: "mouth-test", log: line => logged.push(line) });
+  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 = [];
+  const waiting = [];
+  connection.on("frame", value => {
+    replies.push(runtime.cultnet.parseCultNetMessage(msgpack.decode(value.payload), "cultnet.schema.v0").document);
+    waiting.splice(0).forEach(w => w());
+  });
+  const nextReply = () => new Promise((ok, fail) => { const t = setTimeout(() => fail(new Error(`no reply; the mouth logged: ${logged.join(" | ")}`)), 5000); waiting.push(() => { clearTimeout(t); ok(replies.at(-1)); }); });
+  const put = (kind, recordKey, tuple, agent = "eureka") => connection.send("schema", msgpack.encode(runtime.cultnet.encodeCultNetMessageForWire(
+    docs.envelope({ kind, recordKey, payload: msgpack.encode(tuple), sourceRuntimeId: "huginn-test", sourceAgentId: agent, sourceRole: "persona-organ" }), "cultnet.schema.v0")));
+  try {
+    connection.connect();
+    for (let i = 0; i < 100 && !connection.connected; i++) await new Promise(r => setTimeout(r, 20));
+    assert.ok(connection.connected);
+    let reply = nextReply();
+    put(docs.kinds.openDirectRequest, "req-1", docs.openDirectRequestTuple({ requestId: "req-1", persona: "eureka", userId: "@meta:gamecult.org", text: "hello" }));
+    let document = await reply;
+    assert.equal(document.schemaId, docs.kinds.openDirectReceipt);
+    assert.deepEqual(docs.parseOpenDirectReceipt(msgpack.decode(document.payload)), { requestId: "req-1", outcome: "Opened", roomId: "!dm1:gamecult.org", eventId: "$line1" });
+    assert.deepEqual(opened, [{ requestId: "req-1", persona: "eureka", userId: "@meta:gamecult.org", text: "hello" }]);
+
+    reply = nextReply();
+    put(docs.kinds.openDirectRequest, "req-2", docs.openDirectRequestTuple({ requestId: "req-2", persona: "other", userId: "@meta:gamecult.org", text: "hello" }), "other");
+    document = await reply;
+    assert.deepEqual(docs.parseOpenDirectReceipt(msgpack.decode(document.payload)), { requestId: "req-2", outcome: "UnknownPersona", roomId: null, eventId: null });
+    assert.equal(opened.length, 1, "nothing reached a mouth for the unlisted Persona");
+
+    reply = nextReply();
+    put("gamecult.persona.chat_invented_request.v9", "req-3", ["x"]);
+    document = await reply;
+    assert.deepEqual([document.schemaId, docs.parseRefusal(msgpack.decode(document.payload))], [docs.kinds.refusal, { persona: "eureka", refusal: "UnknownKind" }], "an unknown kind is answered, typed");
+    assert.ok(logged.some(l => /unknown kind: gamecult\.persona\.chat_invented_request\.v9/.test(l)), "and logged");
+  } finally {
+    server.close();
+    connection.close();
+    try { client.close(); } catch { /* the connection already closed its socket */ }
+  }
+});

tools/persona-chat-documents.mjs

@@ -16,6 +16,8 @@ export const kinds = Object.freeze({
   sendRequestV2: "gamecult.persona.chat_send_request.v2",
   sendReceiptV2: "gamecult.persona.chat_send_receipt.v2",
   refusal: "gamecult.persona.chat_refusal.v1",
+  openDirectRequest: "gamecult.persona.chat_open_direct_request.v1",
+  openDirectReceipt: "gamecult.persona.chat_open_direct_receipt.v1",
 });
 
 export const refusals = Object.freeze(["Brake", "PersonaMismatch", "UnknownRoom", "EmptyText", "SendFailed", "UnknownPersona"]);
@@ -23,6 +25,12 @@ export const refusals = Object.freeze(["Brake", "PersonaMismatch", "UnknownRoom"
 // RoomMoved exists only in the v2 receipt: the v1 tuple, a v1 parse and the typed refusal document do not know it.
 export const refusalsV2 = Object.freeze([...refusals, "RoomMoved"]);
 
+// The typed refusal document also answers a request of a kind this mouth does not serve (no Persona is known then).
+const refusalDocument = Object.freeze([...refusals, "UnknownKind"]);
+
+// One outcome word per open-direct receipt. room_id and event_id are present exactly for Opened and Existing.
+export const directOutcomes = Object.freeze(["Opened", "Existing", "Brake", "PersonaMismatch", "UnknownPersona", "EmptyText", "ForeignUser", "NotAPerson", "NoSuchUser", "InvitePending", "Declined", "DailyLimit", "SendFailed"]);
+
 const maxTextBytes = 4000;
 export const maxNewer = 50;
 const maxContext = 20;
@@ -119,15 +127,41 @@ export function parseSendReceiptV2(t) {
 
 // A typed refusal for a request that names a Persona the mouth does not serve.
 export function refusalTuple({ persona, refusal }) {
-  if (!refusals.includes(refusal)) throw new Error(`unknown refusal ${refusal}`);
+  if (!refusalDocument.includes(refusal)) throw new Error(`unknown refusal ${refusal}`);
   return [text(persona, "persona"), refusal];
 }
 export function parseRefusal(t) {
   const [persona, refusal] = tuple(t, 2, "chat_refusal");
-  if (!refusals.includes(refusal)) throw new Error(`unknown refusal ${String(refusal)}`);
+  if (!refusalDocument.includes(refusal)) throw new Error(`unknown refusal ${String(refusal)}`);
   return { persona: text(persona, "persona"), refusal };
 }
 
+// A Persona asks her mouth to open a direct room with one person and say a first line there. user_id is a Matrix user id.
+export function openDirectRequestTuple({ requestId, persona, userId, text: body }) {
+  return [text(requestId, "request_id"), text(persona, "persona"), text(userId, "user_id"), String(body)];
+}
+export function parseOpenDirectRequest(t) {
+  const [requestId, persona, userId, body] = tuple(t, 4, "chat_open_direct_request");
+  if (typeof body !== "string") throw new Error("chat_open_direct_request text must be a string");
+  if (Buffer.byteLength(body, "utf8") > maxTextBytes) throw new Error("chat_open_direct_request text is over the byte bound");
+  return { requestId: text(requestId, "request_id"), persona: text(persona, "persona"), userId: text(userId, "user_id"), text: body };
+}
+
+function checkDirectReceipt(outcome, roomId, eventId) {
+  if (!directOutcomes.includes(outcome)) throw new Error(`unknown open-direct outcome ${String(outcome)}`);
+  const placed = outcome === "Opened" || outcome === "Existing";
+  if (placed !== (roomId !== null && eventId !== null) || (!placed && (roomId !== null || eventId !== null))) throw new Error("room_id and event_id are present exactly for Opened and Existing");
+}
+export function openDirectReceiptTuple({ requestId, outcome, roomId = null, eventId = null }) {
+  checkDirectReceipt(outcome, roomId, eventId);
+  return [text(requestId, "request_id"), outcome, roomId === null ? null : text(roomId, "room_id"), eventId === null ? null : text(eventId, "event_id")];
+}
+export function parseOpenDirectReceipt(t) {
+  const [requestId, outcome, roomId, eventId] = tuple(t, 4, "chat_open_direct_receipt");
+  checkDirectReceipt(outcome, roomId, eventId);
+  return { requestId: text(requestId, "request_id"), outcome, roomId: roomId === null ? null : text(roomId, "room_id"), eventId: eventId === null ? null : text(eventId, "event_id") };
+}
+
 export function deliveryTuple(d) {
   return [
     seq(d.seq, "seq"), text(d.roomAlias, "room_alias"), text(d.roomId, "room_id"), text(d.eventId, "event_id"),
@@ -190,10 +224,11 @@ export function envelope({ kind, recordKey, payload, sourceRuntimeId, sourceAgen
   };
 }
 
+// allowedKinds null: any kind passes (the caller answers a kind it does not serve).
 export function checkEnvelope(message, allowedKinds) {
   const d = message?.document;
   if (message?.schemaVersion !== "cultnet.document_put_raw.v0" || !d) throw new Error("not a document_put_raw message");
-  if (!allowedKinds.includes(d.schemaId)) throw new Error(`document kind ${String(d.schemaId)} is not accepted here`);
+  if (allowedKinds !== null && !allowedKinds.includes(d.schemaId)) throw new Error(`document kind ${String(d.schemaId)} is not accepted here`);
   if (d.payloadEncoding !== "messagepack") throw new Error("chat documents are MessagePack");
   if (!Array.isArray(d.tags) || d.tags.length !== 1 || d.tags[0] !== "cultnet.transport.rudp.v0") throw new Error("chat document tags substituted authority");
   return d;

tools/persona-matrix-mouth.mjs

@@ -58,6 +58,9 @@ const SLUG = /^[A-Za-z0-9_-]{1,64}$/;
 const SEND_TIMEOUT_MS = 10000; // the head read and the PUT of one gated send
 const PRESENCE_GRACE_MS = 5 * 60 * 1000; // a turn's last sign of life keeps her online this long
 const DELIVERED_INDEX = 4000; // delivered event ids remembered so a read report can be checked against them
+const DM_SCAN = 200; // joined rooms read when looking for an existing DM with a person
+const NOT_E2EE = "This room is not end-to-end encrypted: the homeserver can read what is said here.";
+const dayOf = ms => new Date(ms).toISOString().slice(0, 10);
 const HEAD_PAGE = 50; // the room head a gated send reads; also the most ids a RoomMoved receipt names
 
 // A message line: a text event with a sender, not an edit.
@@ -114,6 +117,7 @@ export function createMouth({
   now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, gate = createRoomGate(), sendTimeoutMs = SEND_TIMEOUT_MS,
   presenceGraceMs = PRESENCE_GRACE_MS,
   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
 }) {
   if (!persona || !homeserver || !token || !rooms?.length) throw new Error("persona, homeserver, rooms and token are required");
   const say = message => log(redact(String(message), token));
@@ -138,6 +142,9 @@ export function createMouth({
   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
+  const noticed = new Set(); // rooms this mouth created where the not-E2EE first line has gone out
+  const personTails = new Map(); // person's user id -> the open in flight: opens for one person run one at a time
+  const openedOn = new Map(); // person's user id -> the UTC day this process last created a room for them: one creation per person per day
 
   async function matrix(method, path, body, prefix = API, signal) {
     let response;
@@ -466,6 +473,103 @@ export function createMouth({
     }
   }
 
+  // Every room this account shares with `userId` alone, newest first: { roomId, createdAt, marker, encrypted, theirs }, theirs the
+  // person's membership ("join" | "invite" | "leave" | "ban" | null when they were never invited). Project rooms are not DMs. A
+  // room qualifies when its members are exactly the two of us, or m.direct lists it for the person (a created room whose invite failed).
+  async function directsWith(userId) {
+    const joined = (await matrix("GET", "/joined_rooms")).joined_rooms ?? [];
+    const known = new Set(directContent[userId] ?? []);
+    const candidates = [...known, ...joined.filter(id => !known.has(id) && !aliasByRoomId.has(id))].slice(0, DM_SCAN);
+    const found = [];
+    for (const roomId of candidates) {
+      let events;
+      try { events = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/state`); } catch (error) {
+        if (/ 40[34] /.test(error.message)) continue; // a room m.direct still lists that this account has left or forgotten
+        throw error;
+      }
+      const members = events.filter(e => e.type === "m.room.member");
+      const mine = members.find(e => e.state_key === selfId);
+      if (mine?.content?.membership !== "join") continue;
+      const others = members.filter(e => e.state_key !== selfId);
+      const theirs = others.find(e => e.state_key === userId);
+      if (!(known.has(roomId) || (others.length === 1 && theirs))) continue;
+      const create = events.find(e => e.type === "m.room.create");
+      found.push({
+        roomId, createdAt: create?.origin_server_ts ?? 0, marker: create?.content?.["org.gamecult.open_request"] ?? null,
+        encrypted: events.some(e => e.type === "m.room.encryption"), theirs: theirs?.content?.membership ?? null,
+      });
+    }
+    return found.sort((a, b) => b.createdAt - a.createdAt);
+  }
+
+  // The first line of an open: the invite is the ask and the room is not end-to-end encrypted, and the line says so.
+  // Every line into a room this path created carries the notice until one has gone out (a restart repeats it once at worst).
+  async function putLine(roomId, requestId, text, made) {
+    return gate.run(roomId, async () => {
+      if (!(await released())) return null;
+      const body = made && !noticed.has(roomId) ? `${NOT_E2EE}\n\n${text}` : text;
+      const sent = (await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body }, API, AbortSignal.timeout(sendTimeoutMs))).event_id;
+      if (made) noticed.add(roomId);
+      return sent;
+    });
+  }
+
+  // A Persona's request for a direct room with one person. One outcome word and, for Opened and Existing, the room and the line's event.
+  function openDirect(request) {
+    const key = typeof request.userId === "string" ? request.userId : "";
+    const run = () => openDirectNow(request);
+    const tail = (personTails.get(key) ?? Promise.resolve()).then(run, run);
+    personTails.set(key, tail);
+    const done = () => { if (personTails.get(key) === tail) personTails.delete(key); };
+    tail.then(done, done);
+    return tail;
+  }
+  async function openDirectNow({ requestId, persona: asked, userId, text }) {
+    const answer = (outcome, roomId = null, eventId = null) => { say(`open direct ${requestId}: ${outcome}${roomId ? ` ${roomId}` : ""}`); return { requestId, outcome, roomId, eventId }; };
+    if (asked !== persona) return answer("PersonaMismatch");
+    if (!(await released())) return answer("Brake");
+    if (typeof text !== "string" || text.trim() === "") return answer("EmptyText");
+    try {
+      if (!started || unjoined()) await start();
+      if (!serverName || typeof userId !== "string" || userId.slice(userId.indexOf(":") + 1) !== serverName) return answer("ForeignUser");
+      if (userId === selfId || await isServed(userId)) return answer("NotAPerson");
+      try { await matrix("GET", `/profile/${encodeURIComponent(userId)}`); } catch (error) {
+        if (/ 404 /.test(error.message)) return answer("NoSuchUser");
+        throw error;
+      }
+      const rooms = await directsWith(userId);
+      const usable = rooms.filter(r => !r.encrypted); // a line is never sent into an encrypted room
+      const resumed = usable.find(r => r.marker === requestId); // a repeat of this very request after a crash
+      const live = resumed ?? usable.find(r => r.theirs === "join");
+      if (live) {
+        const eventId = await putLine(live.roomId, requestId, text, live.marker !== null);
+        return eventId === null ? answer("Brake") : answer(resumed ? "Opened" : "Existing", live.roomId, eventId);
+      }
+      const latest = rooms.find(r => ["invite", "leave", "ban"].includes(r.theirs)); // an invite or a decline stands in an encrypted room too
+      if (latest?.theirs === "invite") return answer("InvitePending");
+      if (latest) return answer("Declined"); // left or banned, and nothing newer: only the person reopens it, by inviting her
+      let roomId = usable.find(r => r.theirs === null && r.marker !== null)?.roomId ?? null; // made here earlier, the invite never went out
+      if (roomId === null) {
+        if (openedOn.get(userId) === dayOf(now())) return answer("DailyLimit");
+        if (!(await released())) return answer("Brake");
+        ({ room_id: roomId } = await matrix("POST", "/createRoom", { preset: "private_chat", is_direct: true, creation_content: { "m.federate": false, "org.gamecult.open_request": requestId } }, API, AbortSignal.timeout(sendTimeoutMs)));
+        openedOn.set(userId, dayOf(now()));
+        directRooms.add(roomId);
+        directContent[userId] = [...new Set([...(directContent[userId] ?? []), roomId])];
+        await matrix("PUT", `/user/${encodeURIComponent(selfId)}/account_data/m.direct`, directContent).catch(error => say(`m.direct not saved: ${error.message}`));
+      }
+      if (!(await released())) return answer("Brake");
+      await matrix("POST", `/rooms/${encodeURIComponent(roomId)}/invite`, { user_id: userId, reason: docs.clip(text, 300) }, API, AbortSignal.timeout(sendTimeoutMs));
+      const eventId = await putLine(roomId, requestId, text, true);
+      if (eventId === null) return answer("Brake");
+      touch(); setPresence(true).catch(() => {});
+      return answer("Opened", roomId, eventId);
+    } catch (error) {
+      say(`open direct failed: ${error.message}`);
+      return answer("SendFailed");
+    }
+  }
+
   async function run(signal) {
     const pause = ms => new Promise(r => { const t = setTimeout(r, ms); signal?.addEventListener("abort", () => { clearTimeout(t); r(); }, { once: true }); });
     while (!signal?.aborted) {
@@ -479,7 +583,7 @@ export function createMouth({
     }
   }
 
-  return { start, syncOnce, ack, read, pendingAfter, send, run, get pendingCount() { return pending.length; } };
+  return { start, syncOnce, ack, read, pendingAfter, send, openDirect, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } };
 }
 
 // Replace every secret in a message. `secrets` is one string or a list.
@@ -557,13 +661,24 @@ export async function startChatEndpoint({ mouths, runtime, port, connectionKey,
   const respond = (peer, persona, kind, recordKey, tuple) => server.send(peer, "schema", msgpack.encode(cultnet.encodeCultNetMessageForWire(
     docs.envelope({ kind, recordKey, payload: msgpack.encode(tuple), sourceRuntimeId: runtimeId, sourceAgentId: persona, sourceRole: "bifrost-persona-matrix-mouth" }), "cultnet.schema.v0")));
   const refuseUnknown = (peer, persona) => respond(peer, persona, docs.kinds.refusal, persona, docs.refusalTuple({ persona, refusal: "UnknownPersona" }));
+  const served = [docs.kinds.pendingRequest, docs.kinds.ack, docs.kinds.read, docs.kinds.sendRequest, docs.kinds.sendRequestV2, docs.kinds.openDirectRequest];
   const handle = async ({ peer, channelId, payload }) => {
     try {
       if (channelId !== "schema") throw new Error("chat request arrived on a foreign channel");
       const message = cultnet.parseCultNetMessage(msgpack.decode(payload), "cultnet.schema.v0");
-      const document = docs.checkEnvelope(message, [docs.kinds.pendingRequest, docs.kinds.ack, docs.kinds.read, docs.kinds.sendRequest, docs.kinds.sendRequestV2]);
+      const document = docs.checkEnvelope(message, null);
+      if (!served.includes(document.schemaId)) { // a kind this mouth does not serve is answered, loudly, and logged: never dropped
+        log(`chat endpoint refused a request of an unknown kind: ${String(document.schemaId).slice(0, 120)}`);
+        const who = typeof document.sourceAgentId === "string" && SLUG.test(document.sourceAgentId) ? document.sourceAgentId : "unknown";
+        return respond(peer, who, docs.kinds.refusal, who, docs.refusalTuple({ persona: who, refusal: "UnknownKind" }));
+      }
       const tuple = msgpack.decode(document.payload);
-      if (document.schemaId === docs.kinds.pendingRequest) {
+      if (document.schemaId === docs.kinds.openDirectRequest) {
+        const request = docs.parseOpenDirectRequest(tuple);
+        const mouth = mouths.get(request.persona);
+        const receipt = mouth ? await mouth.openDirect(request) : { requestId: request.requestId, outcome: "UnknownPersona", roomId: null, eventId: null };
+        respond(peer, request.persona, docs.kinds.openDirectReceipt, receipt.requestId, docs.openDirectReceiptTuple(receipt));
+      } else if (document.schemaId === docs.kinds.pendingRequest) {
         const request = docs.parsePendingRequest(tuple);
         const mouth = mouths.get(request.persona);
         if (!mouth) return refuseUnknown(peer, request.persona);
@@ -669,7 +784,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, log: line => log(`${persona}: ${line}`),
+        persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, 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 }),