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

persona matrix mouth: one session per client on the many-peer RUDP server; every request answered on the connection it arrived on

Browse source 
Two Persona turns running at once no longer take each other's session. Requests run in order per client, not across clients.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-10-31 01:33:36 +01:00
commit b892c64f00
2 changed files with 64 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -87,6 +87,50 @@ test("the brake reader reads a real CultCache control store", async () => {
assert.equal(await reader(resolve(dir, "garbage.cc"))(), false);
});
test("two clients served at once each get their own answer on their own session", async () => {
const mouth = {
pendingAfter: () => [],
ack: async () => {},
// Slow enough that the two requests are in flight together.
send: async request => { await new Promise(r => setTimeout(r, 150)); return { requestId: request.requestId, eventId: `$${request.requestId}`, refusal: null }; },
};
const key = "connection-key-for-concurrency";
const server = await startChatEndpoint({ mouths: new Map([["eureka", mouth]]), runtime, port: 0, connectionKey: key, runtimeId: "mouth-test" });
const dial = async name => {
const socket = dgram.createSocket("udp4");
await new Promise(ok => socket.bind(0, "127.0.0.1", ok));
const connection = new runtime.cultnet.CultNetRudpSocketTransportConnection({
runtimeId: name, socket, mode: "client", remoteHost: "127.0.0.1", remotePort: server.address.port,
connectionId: connectionIdFor(key), resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200,
});
const receipts = [];
connection.on("frame", value => {
const message = runtime.cultnet.parseCultNetMessage(msgpack.decode(value.payload), "cultnet.schema.v0");
receipts.push(docs.parseSendReceipt(msgpack.decode(message.document.payload)));
});
connection.connect();
for (let i = 0; i < 100 && !connection.connected; i++) await new Promise(r => setTimeout(r, 20));
assert.ok(connection.connected, `${name} opened its session`);
const send = requestId => connection.send("schema", msgpack.encode(runtime.cultnet.encodeCultNetMessageForWire(
docs.envelope({ kind: docs.kinds.sendRequest, recordKey: requestId, payload: msgpack.encode(docs.sendRequestTuple({ requestId, persona: "eureka", roomAlias: "#operator:gamecult.org", text: name })), sourceRuntimeId: name, sourceAgentId: "eureka", sourceRole: "persona-organ" }), "cultnet.schema.v0")));
return { connection, receipts, send };
};
const first = await dial("mind-persona-one");
const second = await dial("mind-heartbeat");
try {
first.send("req-first");
second.send("req-second");
for (let i = 0; i < 250 && (first.receipts.length < 1 || second.receipts.length < 1); i++) await new Promise(r => setTimeout(r, 20));
assert.deepEqual(first.receipts, [{ requestId: "req-first", eventId: "$req-first", refusal: null }]);
assert.deepEqual(second.receipts, [{ requestId: "req-second", eventId: "$req-second", refusal: null }]);
assert.ok(first.connection.connected && second.connection.connected, "neither session took the other's");
} finally {
server.close();
first.connection.close();
second.connection.close();
}
});
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 mouth = {

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

@ -317,7 +317,9 @@ export function pageOf(persona, deliveries, msgpack, log = () => {}) {
}
// CultNet RUDP endpoint on loopback, one for every Persona in `mouths` (persona -> mouth).
// Requests arrive as document_put_raw messages. A request naming a Persona this
// Every client (each Persona process and heartbeat dials with its own socket) gets its own
// session, and a request is answered on the session it arrived on. Requests arrive as
// document_put_raw messages. A request naming a Persona this
// process does not serve is answered with a typed UnknownPersona refusal.
export async function startChatEndpoint({ mouths, runtime, port, connectionKey, runtimeId, socketFactory = createSocket, host = "127.0.0.1", log = () => {} }) {
if (!/^127\.\d+\.\d+\.\d+$/.test(host)) throw new Error("the chat endpoint listens on loopback only");
@ -325,40 +327,43 @@ export async function startChatEndpoint({ mouths, runtime, port, connectionKey,
const socket = socketFactory("udp4");
await new Promise((ok, fail) => { socket.once("error", fail); socket.once("listening", ok); socket.bind(port, host); });
const connectionId = connectionIdFor(connectionKey);
const connection = new cultnet.CultNetRudpSocketTransportConnection({ runtimeId, socket, mode: "server", connectionId, initialSequence: 100, resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200 });
const respond = (persona, kind, recordKey, tuple) => connection.send("schema", msgpack.encode(cultnet.encodeCultNetMessageForWire(
const server = new cultnet.CultNetRudpSocketTransportServer({ runtimeId, socket, connectionId, initialSequence: 100, resendDelayMs: 50, resendPollMs: 10, maxFragmentBytes: 1200,
// A client keeps its link between requests and does not redial, so only a client gone for an hour (a crash that sent no goodbye) is reaped.
peerTimeoutMs: 3_600_000 });
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 = persona => respond(persona, docs.kinds.refusal, persona, docs.refusalTuple({ persona, refusal: "UnknownPersona" }));
let busy = Promise.resolve();
const handle = async value => {
const refuseUnknown = (peer, persona) => respond(peer, persona, docs.kinds.refusal, persona, docs.refusalTuple({ persona, refusal: "UnknownPersona" }));
const handle = async ({ peer, channelId, payload }) => {
try {
if (value.channelId !== "schema") throw new Error("chat request arrived on a foreign channel");
const message = cultnet.parseCultNetMessage(msgpack.decode(value.payload), "cultnet.schema.v0");
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.sendRequest]);
const tuple = msgpack.decode(document.payload);
if (document.schemaId === docs.kinds.pendingRequest) {
const request = docs.parsePendingRequest(tuple);
const mouth = mouths.get(request.persona);
if (!mouth) return refuseUnknown(request.persona);
respond(request.persona, docs.kinds.pending, request.persona, pageOf(request.persona, mouth.pendingAfter(request.afterSeq), msgpack, log));
if (!mouth) return refuseUnknown(peer, request.persona);
respond(peer, request.persona, docs.kinds.pending, request.persona, pageOf(request.persona, mouth.pendingAfter(request.afterSeq), msgpack, log));
} else if (document.schemaId === docs.kinds.ack) {
const request = docs.parseAck(tuple);
const mouth = mouths.get(request.persona);
if (!mouth) return refuseUnknown(request.persona);
if (!mouth) return refuseUnknown(peer, request.persona);
await mouth.ack(request.throughSeq);
} else {
const request = docs.parseSendRequest(tuple);
const mouth = mouths.get(request.persona);
const receipt = mouth ? await mouth.send(request) : { requestId: request.requestId, eventId: null, refusal: "UnknownPersona" };
respond(request.persona, docs.kinds.sendReceipt, receipt.requestId, docs.sendReceiptTuple(receipt));
respond(peer, request.persona, docs.kinds.sendReceipt, receipt.requestId, docs.sendReceiptTuple(receipt));
}
} catch (error) {
log(`chat endpoint rejected a frame: ${error?.message ?? error}`);
}
};
const frame = value => { busy = busy.then(() => handle(value)); };
connection.on("frame", frame);
return { address: socket.address(), close() { connection.off?.("frame", frame); connection.close(); } };
// A peer's requests run in order; peers do not wait on one another.
const busy = new WeakMap();
const frame = value => { busy.set(value.peer, (busy.get(value.peer) ?? Promise.resolve()).then(() => handle(value))); };
server.on("frame", frame);
return { address: socket.address(), close() { server.off("frame", frame); server.close(); } };
}
export function cultcacheState({ runtime, path }) {