From 7b630312e1b05bf70fb45a9ab0f57ead3a1d3c1a Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 29 Oct 2026 19:41:30 +0100 Subject: [PATCH] persona-mouth: Matrix ear and mouth for Personas (brake-refusing sends, token only in the Authorization header, cursor after ack, one ring per idle window); delete the Discord posting actuator (squash of eureka-body/persona-mouth 1894f65..ab2bdc6; Soul s1) Co-Authored-By: Claude Opus 5.5 --- docs/bridge.md | 10 +- .../gamecult.persona.chat_ack.v1.msgpack | 1 + .../gamecult.persona.chat_pending.v1.msgpack | Bin 0 -> 142 bytes ...lt.persona.chat_pending_request.v1.msgpack | 1 + ...ecult.persona.chat_send_receipt.v1.msgpack | 1 + ...ecult.persona.chat_send_request.v1.msgpack | 1 + ...rsona-discord-crossing-rust-smoke.test.mjs | 27 -- tests/persona-discord-delivery.test.mjs | 36 -- tests/persona-discord-permit.test.mjs | 15 - ...rsona-discord-rudp-cross-language.test.mjs | 29 -- tests/persona-feedback-cli.test.mjs | 4 +- tests/persona-matrix-mouth-runtime.test.mjs | 149 ++++++++ tests/persona-matrix-mouth.test.mjs | 347 ++++++++++++++++++ tools/persona-chat-documents.mjs | 114 ++++++ tools/persona-discord-delivery-documents.mjs | 43 --- tools/persona-discord-delivery.mjs | 52 --- tools/persona-discord-permit.mjs | 39 -- tools/persona-discord-rudp.mjs | 31 -- tools/persona-feedback.mjs | 22 +- tools/persona-matrix-mouth.mjs | 303 +++++++++++++++ 20 files changed, 923 insertions(+), 302 deletions(-) create mode 100644 tests/fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack create mode 100644 tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack create mode 100644 tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack create mode 100644 tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack create mode 100644 tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack delete mode 100644 tests/persona-discord-crossing-rust-smoke.test.mjs delete mode 100644 tests/persona-discord-delivery.test.mjs delete mode 100644 tests/persona-discord-permit.test.mjs delete mode 100644 tests/persona-discord-rudp-cross-language.test.mjs create mode 100644 tests/persona-matrix-mouth-runtime.test.mjs create mode 100644 tests/persona-matrix-mouth.test.mjs create mode 100644 tools/persona-chat-documents.mjs delete mode 100644 tools/persona-discord-delivery-documents.mjs delete mode 100644 tools/persona-discord-delivery.mjs delete mode 100644 tools/persona-discord-permit.mjs delete mode 100644 tools/persona-discord-rudp.mjs create mode 100644 tools/persona-matrix-mouth.mjs diff --git a/docs/bridge.md b/docs/bridge.md index 683d7d2..0e5c56d 100644 --- a/docs/bridge.md +++ b/docs/bridge.md @@ -196,15 +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 replies cross back through two explicit CultMesh surfaces. Epiphany is -the sole writer of `epiphany.persona_discord_delivery_request.v0`; Bifrost -reads that surface, authenticates the dedicated Persona-mouth signer, journals -external execution before touching Discord, and is the sole writer of -`bifrost.persona_discord_delivery_receipt.v0`. Epiphany reads receipts but -never receives the Discord token or write access to Bifrost private state. -`unknown` is terminal for automatic transport retry: an external post may have -happened, so restart recovery must not manufacture a duplicate. Failed and -unknown outcomes remain Persona pressure for explicit resolution. +Persona speech goes out through Matrix, not Discord. `tools/persona-matrix-mouth.mjs` is the one process per Persona that holds the Persona's Matrix access token (a systemd credential) and nobody else does. It reads the configured rooms by `/sync`, keeps each message from another sender as a delivery, and speaks on a `gamecult.persona.chat_send_request.v1` over a loopback CultNet RUDP endpoint, only while the instance brake is released. Document shapes: `tools/persona-chat-documents.mjs`; golden bytes: `tests/fixtures/persona-chat-v1/`. 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/fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack new file mode 100644 index 0000000..af22d1e --- /dev/null +++ b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_ack.v1.msgpack @@ -0,0 +1 @@ +’¦eureka \ No newline at end of file diff --git a/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending.v1.msgpack new file mode 100644 index 0000000000000000000000000000000000000000..0bdbdfa4f12ab390f58bc48935e9747290cb7103 GIT binary patch literal 142 zcmbQVEVZ;KH9K+QO!jTc`30#(i6!|(R_TeksmY}|C3^Wq=^GUjlaf(H7OSKhZgR*? zElETXUg8TBS(A~PlasFiG4wnG1LG{+hK&ppCvS0p2%+d$o|;&clbKosRkv0G01G@g AZU6uP literal 0 HcmV?d00001 diff --git a/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack new file mode 100644 index 0000000..0c326f5 --- /dev/null +++ b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_pending_request.v1.msgpack @@ -0,0 +1 @@ +’¦eureka \ No newline at end of file diff --git a/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack new file mode 100644 index 0000000..9b082a3 --- /dev/null +++ b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_receipt.v1.msgpack @@ -0,0 +1 @@ +“¥req-1¥$sentÀ \ No newline at end of file diff --git a/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack new file mode 100644 index 0000000..44f6485 --- /dev/null +++ b/tests/fixtures/persona-chat-v1/gamecult.persona.chat_send_request.v1.msgpack @@ -0,0 +1 @@ +”¥req-1¦eureka¶#operator:gamecult.org¢hi \ No newline at end of file diff --git a/tests/persona-discord-crossing-rust-smoke.test.mjs b/tests/persona-discord-crossing-rust-smoke.test.mjs deleted file mode 100644 index cc9ed61..0000000 --- a/tests/persona-discord-crossing-rust-smoke.test.mjs +++ /dev/null @@ -1,27 +0,0 @@ -import test from "node:test"; -import {existsSync} from "node:fs"; -import assert from "node:assert/strict"; -import {spawnSync} from "node:child_process"; -import {chmod,mkdtemp,writeFile} from "node:fs/promises"; -import {tmpdir} from "node:os"; -import {resolve} from "node:path"; -import {createRequire} from "node:module"; -import {enrollDomainIdentity} from "../tools/bifrost-signed-identity.mjs"; -import {receiptIdentityDomain} from "../tools/persona-discord-delivery-documents.mjs"; -import {processPersonaDiscordDeliveries} from "../tools/persona-discord-delivery.mjs"; - -const epiphany=resolve(import.meta.dirname,"..","..","Epiphany"),cult=resolve(import.meta.dirname,"..","..","CultLib"); -function runtime(){const cr=createRequire(resolve(cult,"packages","cultcache-ts","package.json")),nr=createRequire(resolve(cult,"packages","cultnet-ts","package.json")),cc=cr("@gamecult/cultcache-ts"),mp=nr("@msgpack/msgpack");return {CultCache:cc.CultCache,SingleFileMessagePackBackingStore:cc.SingleFileMessagePackBackingStore,defineDocumentType:cc.defineDocumentType,encode:mp.encode,decode:mp.decode};} - -const rustFixture=existsSync(resolve(epiphany,"epiphany-core","src","bin","epiphany-persona-discord-crossing-fixture.rs"))&&spawnSync("cargo",["--version"]).status===0; -test("Bifrost consumes a Rust-authored request store through its read-only crossing directory",{skip:rustFixture?false:"needs cargo and Epiphany's epiphany-persona-discord-crossing-fixture fixture (removed in Epiphany 387afe49)"},async()=>{ - const root=await mkdtemp(resolve(tmpdir(),"rust-bifrost-crossing-")),requestDir=resolve(root,"request"),receiptDir=resolve(root,"receipt"),requestStore=resolve(requestDir,"requests.cc"),anchor=resolve(root,"mouth-anchor.msgpack"),identity=resolve(root,"mouth.cc"),deliveryKey=resolve(root,"delivery.seed"),bridge=resolve(root,"bridge.mjs"); - const cargo=spawnSync("cargo",["run","--quiet","--manifest-path",resolve(epiphany,"epiphany-core","Cargo.toml"),"--bin","epiphany-persona-discord-crossing-fixture","--","--request-store",requestStore,"--identity-store",identity,"--request-anchor",anchor],{cwd:epiphany,encoding:"utf8",timeout:180000,windowsHide:true}); - assert.equal(cargo.status,0,cargo.stderr||cargo.stdout); - await chmod(requestStore,0o444); - await enrollDomainIdentity(deliveryKey,"delivery",receiptIdentityDomain); - await writeFile(bridge,`console.log(JSON.stringify({action:"discord-post",ok:true,channelId:"123",messageId:"rust-smoke-message",transport:"smoke",crossingReceiptId:"rust-smoke-receipt",url:"https://discord.invalid/rust-smoke"}))`); - const r=runtime(),results=await processPersonaDiscordDeliveries({runtime:r,requestStorePath:requestStore,receiptStorePath:resolve(receiptDir,"receipts.cc"),executionStorePath:resolve(root,"private","executions.cc"),epiphanyTrustAnchorPath:anchor,epiphanyRuntimeId:"epiphany-yggdrasil",privateKeyPath:deliveryKey,targetAdmitted:()=>true,requestPermit:async({requestPayloadSha256})=>({permitId:`permit:${requestPayloadSha256}`,expiresAt:new Date(Date.now()+5000).toISOString()}),bridgeCliPath:bridge,bifrostRoot:root}); - assert.equal(results[0]?.status,"completed"); - assert.equal(results[0]?.requestId,"rust-crossing-smoke-1"); -}); diff --git a/tests/persona-discord-delivery.test.mjs b/tests/persona-discord-delivery.test.mjs deleted file mode 100644 index ca4b65f..0000000 --- a/tests/persona-discord-delivery.test.mjs +++ /dev/null @@ -1,36 +0,0 @@ -import test from "node:test"; -import assert from "node:assert/strict"; -import {createHash} from "node:crypto"; -import {createRequire} from "node:module"; -import {mkdtemp,writeFile} from "node:fs/promises"; -import {tmpdir} from "node:os"; -import {resolve} from "node:path"; -import {buildPersonaDiscordDeliveryDefinitions,executionSchema,parseReceipt,parseRequest,receiptDocumentTuple,receiptIdentityDomain,receiptSignatureDomain,receiptSigningPurpose,receiptSigningTuple,requestIdentityDomain,requestSignatureDomain,requestSigningPurpose,requestSigningTuple} from "../tools/persona-discord-delivery-documents.mjs"; -import {enrollPersonaDiscordDeliveryIdentity,exportPersonaDiscordPermitRequestAnchor,inspectPersonaDiscordDeliveryReadiness,processPersonaDiscordDeliveries} from "../tools/persona-discord-delivery.mjs"; -import {enrollDomainIdentity,signDomainPurpose,verifyDomainPurpose} from "../tools/bifrost-signed-identity.mjs"; - -const signature=()=>new Uint8Array(64); -const request=()=>({schemaVersion:"epiphany.persona_discord_delivery_request.v0",requestId:"request-1",effectDocumentId:"effect-1",targetRuntimeId:"epiphany-yggdrasil",personaAgentId:"epiphany.Persona",channelId:"123",replyToMessageId:"456",content:"hello",contentSha256:`sha256-${"a".repeat(64)}`,issuedAt:"2026-07-21T12:00:00Z",expiresAt:"2026-07-21T12:01:00Z",privateStateExposed:false,signerIdentityId:"signer",signerSignature:signature()}); - -test("Persona delivery request has one exact signature tuple",()=>{const value=parseRequest(request());assert.deepEqual(requestSigningTuple(value),["epiphany.persona_discord_delivery_request.v0","request-1","effect-1","epiphany-yggdrasil","epiphany.Persona","123","456","hello",`sha256-${"a".repeat(64)}`,"2026-07-21T12:00:00Z","2026-07-21T12:01:00Z",false,"signer"]);}); - -test("Persona delivery receipt terminal states cannot counterfeit delivery",()=>{const base={schemaVersion:"bifrost.persona_discord_delivery_receipt.v0",receiptId:"request-1",requestId:"request-1",requestPayloadSha256:`sha256-${"b".repeat(64)}`,status:"unknown",channelId:"123",replyToMessageId:"456",messageId:"",transport:"",crossingReceiptId:"",receiptUrl:"",completedAt:"2026-07-21T12:00:02Z",providerIdentityId:"bifrost",privateStateExposed:false,providerSignature:signature()};assert.equal(receiptSigningTuple(parseReceipt(base)).length,14);assert.equal(receiptDocumentTuple(base).length,15);assert.deepEqual(parseReceipt(receiptDocumentTuple(base)),base);assert.throws(()=>parseReceipt({...base,messageId:"counterfeit"}),/may not claim delivery evidence/);assert.doesNotThrow(()=>parseReceipt({...base,status:"failed"}));assert.doesNotThrow(()=>parseReceipt({...base,status:"completed",messageId:"789",transport:"bifrost.discord-post",crossingReceiptId:"crossing-1",receiptUrl:"https://discord.com/channels/g/123/789"}));}); - -test("Persona delivery documents refuse privacy and identity substitutions",()=>{assert.throws(()=>parseRequest({...request(),privateStateExposed:true}),/must be false/);const receipt={schemaVersion:"bifrost.persona_discord_delivery_receipt.v0",receiptId:"receipt-2",requestId:"request-1",requestPayloadSha256:`sha256-${"b".repeat(64)}`,status:"failed",channelId:"123",replyToMessageId:"",messageId:"",transport:"",crossingReceiptId:"",receiptUrl:"",completedAt:"2026-07-21T12:00:02Z",providerIdentityId:"bifrost",privateStateExposed:false,providerSignature:signature()};assert.throws(()=>parseReceipt(receipt),/must equal requestId/);}); - -test("receipt signer and permit requester are distinct domain identities",async()=>{const root=await mkdtemp(resolve(tmpdir(),"persona-mouth-identities-")),runtime=loadRuntime(),receiptKey=resolve(root,"receipt.seed"),permitKey=resolve(root,"permit.seed"),receiptAnchor=resolve(root,"receipt-anchor.msgpack"),permitAnchor=resolve(root,"permit-anchor.msgpack"),receipt=await enrollPersonaDiscordDeliveryIdentity({runtime,privateKeyPath:receiptKey,trustAnchorOut:receiptAnchor}),permit=await exportPersonaDiscordPermitRequestAnchor({runtime,privateKeyPath:permitKey,trustAnchorOut:permitAnchor});assert.notEqual(receiptKey,permitKey);assert.notEqual(receipt.identityId,permit.identityId);}); - -test("delivery listener admits absent typed stores when their owner directory is writable",async()=>{const root=await mkdtemp(resolve(tmpdir(),"persona-mouth-readiness-")),runtime=loadRuntime(),requestIdentity=await enrollDomainIdentity(resolve(root,"request.seed"),"request identity",requestIdentityDomain),mouthAnchor=resolve(root,"mouth-anchor.msgpack"),permitAnchor=resolve(root,"permit-anchor.msgpack"),receiptKey=resolve(root,"receipt.seed"),permitKey=resolve(root,"permit.seed");await writeFile(mouthAnchor,runtime.encode(requestAnchor(requestIdentity)));await writeFile(permitAnchor,new Uint8Array([1]));await enrollPersonaDiscordDeliveryIdentity({runtime,privateKeyPath:receiptKey,trustAnchorOut:resolve(root,"receipt-anchor.msgpack")});await exportPersonaDiscordPermitRequestAnchor({runtime,privateKeyPath:permitKey,trustAnchorOut:resolve(root,"permit-request-anchor.msgpack")});const readiness=await inspectPersonaDiscordDeliveryReadiness({runtime,requestStorePath:resolve(root,"requests.cc"),receiptStorePath:resolve(root,"receipts.cc"),executionStorePath:resolve(root,"executions.cc"),epiphanyTrustAnchorPath:mouthAnchor,epiphanyRuntimeId:"epiphany-yggdrasil",epiphanyPermitAnchorPath:permitAnchor,epiphanyPermitEndpoint:"rudp://127.0.0.1:9",privateKeyPath:receiptKey,permitRequesterPrivateKeyPath:permitKey});assert.equal(readiness.ready,true);assert.deepEqual(readiness.reasons,[]);}); - -test("Bifrost authenticates one request, publishes one signed receipt, and replays without reposting",async()=>{const root=await mkdtemp(resolve(tmpdir(),"persona-mouth-")),runtime=loadRuntime(),defs=buildPersonaDiscordDeliveryDefinitions(runtime.defineDocumentType),requestStore=resolve(root,"requests.cc"),receiptStore=resolve(root,"receipts.cc"),executionStore=resolve(root,"private","executions.cc"),requestKey=resolve(root,"request.seed"),deliveryKey=resolve(root,"delivery.seed"),anchorPath=resolve(root,"request-anchor.msgpack"),counterPath=resolve(root,"post-count"),bridgePath=resolve(root,"fake-bridge.mjs"),requestIdentity=await enrollDomainIdentity(requestKey,"request identity",requestIdentityDomain),deliveryIdentity=await enrollDomainIdentity(deliveryKey,"delivery identity",receiptIdentityDomain);await writeFile(anchorPath,runtime.encode(requestAnchor(requestIdentity)));await writeFile(bridgePath,`import{readFileSync,writeFileSync}from'node:fs';const p=${JSON.stringify(counterPath)},i=process.argv.indexOf('--cultmesh-command-id');if(i<0||process.argv[i+1]!=='request-1')process.exit(2);let n=0;try{n=Number(readFileSync(p,'utf8'))}catch{}writeFileSync(p,String(n+1));console.log(JSON.stringify({action:'discord-post',ok:true,channelId:'123',messageId:'789',transport:'bifrost.discord-post',crossingReceiptId:'crossing-1',url:'https://discord.com/channels/g/123/789'}));`);const now=Date.now(),value={...request(),issuedAt:new Date(now-1000).toISOString(),expiresAt:new Date(now+60000).toISOString(),signerIdentityId:requestIdentity.identityId};value.contentSha256=`sha256-${createHash("sha256").update(value.content).digest("hex")}`;value.signerSignature=new Uint8Array(signDomainPurpose(requestIdentity,requestSignatureDomain,requestSigningPurpose,runtime.encode(requestSigningTuple(value))));const requests=await open(runtime,defs.request,requestStore);await requests.put(defs.request,value.requestId,value);const options={runtime,requestStorePath:requestStore,receiptStorePath:receiptStore,executionStorePath:executionStore,epiphanyTrustAnchorPath:anchorPath,epiphanyRuntimeId:"epiphany-yggdrasil",privateKeyPath:deliveryKey,targetAdmitted:()=>true,bridgeCliPath:bridgePath,bifrostRoot:root,requestPermit:async()=>({permitId:"permit-1"})};let results=await processPersonaDiscordDeliveries(options);assert.equal(results[0].status,"completed");results=await processPersonaDiscordDeliveries(options);assert.equal(results.length,0);const receipts=await open(runtime,defs.receipt,receiptStore),receipt=receipts.get(defs.receipt,value.requestId);assert.equal(receipt.messageId,"789");assert.equal(verifyDomainPurpose(deliveryIdentity.publicKey,receiptSignatureDomain,receiptSigningPurpose,runtime.encode(receiptSigningTuple(receipt)),receipt.providerSignature),true);assert.equal(await import("node:fs/promises").then(fs=>fs.readFile(counterPath,"utf8")),"1");}); - -test("running journal recovers as terminal unknown without a second Discord post",async()=>{const root=await mkdtemp(resolve(tmpdir(),"persona-mouth-crash-")),runtime=loadRuntime(),defs=buildPersonaDiscordDeliveryDefinitions(runtime.defineDocumentType),requestStore=resolve(root,"requests.cc"),receiptStore=resolve(root,"receipts.cc"),executionStore=resolve(root,"private","executions.cc"),requestIdentity=await enrollDomainIdentity(resolve(root,"request.seed"),"request identity",requestIdentityDomain),deliveryKey=resolve(root,"delivery.seed"),anchorPath=resolve(root,"anchor.msgpack");await enrollDomainIdentity(deliveryKey,"delivery identity",receiptIdentityDomain);await writeFile(anchorPath,runtime.encode(requestAnchor(requestIdentity)));const now=Date.now(),value={...request(),issuedAt:new Date(now-1000).toISOString(),expiresAt:new Date(now+60000).toISOString(),signerIdentityId:requestIdentity.identityId};value.contentSha256=`sha256-${createHash("sha256").update(value.content).digest("hex")}`;value.signerSignature=new Uint8Array(signDomainPurpose(requestIdentity,requestSignatureDomain,requestSigningPurpose,runtime.encode(requestSigningTuple(value))));const requests=await open(runtime,defs.request,requestStore);await requests.put(defs.request,value.requestId,value);const digest=`sha256-${createHash("sha256").update(runtime.encode(requestSigningTuple(value))).digest("hex")}`,executions=await open(runtime,defs.execution,executionStore);await executions.put(defs.execution,value.requestId,{schemaName:"bifrost.persona_discord_delivery_execution",schemaVersion:executionSchema,requestId:value.requestId,requestPayloadSha256:digest,status:"running",recordedAt:new Date().toISOString(),detail:"crash fixture"});const results=await processPersonaDiscordDeliveries({runtime,requestStorePath:requestStore,receiptStorePath:receiptStore,executionStorePath:executionStore,epiphanyTrustAnchorPath:anchorPath,epiphanyRuntimeId:"epiphany-yggdrasil",privateKeyPath:deliveryKey,targetAdmitted:()=>true,bridgeCliPath:resolve(root,"must-not-run.mjs"),bifrostRoot:root});assert.equal(results[0].status,"unknown");assert.equal(results[0].messageId,"");}); - -test("permit intent survives network failure and restart reuses its exact request tuple",async()=>{const f=await deliveryFixture("permit-retry-");const attempts=[];let results=await processPersonaDiscordDeliveries({...f.options,requestPermit:async value=>{attempts.push({...value});throw new Error("permit network unavailable");}});assert.equal(results[0].status,"permit_pending");const executions=await open(f.runtime,f.defs.execution,f.executionStore),pending=executions.get(f.defs.execution,f.value.requestId);assert.equal(pending.status,"permit_pending");assert.equal(pending.permitNonce,attempts[0].nonce);assert.equal(pending.permitIssuedAt,attempts[0].issuedAt);assert.equal(pending.permitExpiresAt,attempts[0].expiresAt);results=await processPersonaDiscordDeliveries({...f.options,requestPermit:async value=>{attempts.push({...value});return {permitId:"permit-after-restart"};}});assert.equal(results[0].status,"completed");assert.equal(attempts.length,2);for(const key of ["requestId","requestPayloadSha256","targetRuntimeId","nonce","issuedAt","expiresAt"])assert.equal(attempts[1][key],attempts[0][key]);assert.equal(await import("node:fs/promises").then(fs=>fs.readFile(f.counterPath,"utf8")),"1");}); - -test("expired permit intent restarts as terminal unknown without requesting or posting",async()=>{const f=await deliveryFixture("permit-expired-"),digest=`sha256-${createHash("sha256").update(f.runtime.encode(requestSigningTuple(f.value))).digest("hex")}`,executions=await open(f.runtime,f.defs.execution,f.executionStore),issuedAt=new Date(Date.now()-6000).toISOString(),expiresAt=new Date(Date.now()-1000).toISOString();await executions.put(f.defs.execution,f.value.requestId,{schemaName:"bifrost.persona_discord_delivery_execution",schemaVersion:executionSchema,requestId:f.value.requestId,requestPayloadSha256:digest,status:"permit_pending",recordedAt:issuedAt,permitNonce:"persisted-nonce",permitIssuedAt:issuedAt,permitExpiresAt:expiresAt,detail:"restart fixture"});let permitCalls=0;const results=await processPersonaDiscordDeliveries({...f.options,requestPermit:async()=>{permitCalls++;throw new Error("must not request");},bridgeCliPath:resolve(f.root,"must-not-post.mjs")});assert.equal(results[0].status,"unknown");assert.equal(results[0].messageId,"");assert.equal(permitCalls,0);await assert.rejects(import("node:fs/promises").then(fs=>fs.readFile(f.counterPath,"utf8")));}); - -function loadRuntime(){const cult=resolve(import.meta.dirname,"..","..","CultLib"),cr=createRequire(resolve(cult,"packages","cultcache-ts","package.json")),nr=createRequire(resolve(cult,"packages","cultnet-ts","package.json")),cc=cr("@gamecult/cultcache-ts"),mp=nr("@msgpack/msgpack");return {CultCache:cc.CultCache,SingleFileMessagePackBackingStore:cc.SingleFileMessagePackBackingStore,defineDocumentType:cc.defineDocumentType,encode:mp.encode,decode:mp.decode};} -async function open(runtime,definition,path){const cache=runtime.CultCache.builder().withDocumentType(definition).withGenericStore(new runtime.SingleFileMessagePackBackingStore(path)).build();await cache.pullAllBackingStores();return cache;} -function requestAnchor(identity){return ["gamecult.service_trust_anchor.v1","epiphany-persona-mouth:epiphany-yggdrasil:v0","epiphany-persona-mouth","epiphany-yggdrasil",identity.identityId,new Uint8Array(identity.publicKey),"ed25519",requestSigningPurpose,"epiphany.persona_discord_delivery_request.v0","root",Date.now(),null,false];} -async function deliveryFixture(prefix){const root=await mkdtemp(resolve(tmpdir(),prefix)),runtime=loadRuntime(),defs=buildPersonaDiscordDeliveryDefinitions(runtime.defineDocumentType),requestStore=resolve(root,"requests.cc"),receiptStore=resolve(root,"receipts.cc"),executionStore=resolve(root,"private","executions.cc"),deliveryKey=resolve(root,"delivery.seed"),anchorPath=resolve(root,"request-anchor.msgpack"),counterPath=resolve(root,"post-count"),bridgeCliPath=resolve(root,"fake-bridge.mjs"),requestIdentity=await enrollDomainIdentity(resolve(root,"request.seed"),"request identity",requestIdentityDomain);await enrollDomainIdentity(deliveryKey,"delivery identity",receiptIdentityDomain);await writeFile(anchorPath,runtime.encode(requestAnchor(requestIdentity)));await writeFile(bridgeCliPath,`import{readFileSync,writeFileSync}from'node:fs';const p=${JSON.stringify(counterPath)};let n=0;try{n=Number(readFileSync(p,'utf8'))}catch{}writeFileSync(p,String(n+1));console.log(JSON.stringify({action:'discord-post',ok:true,channelId:'123',messageId:'789',transport:'bifrost.discord-post',crossingReceiptId:'crossing-1',url:'https://discord.com/channels/g/123/789'}));`);const now=Date.now(),value={...request(),issuedAt:new Date(now-1000).toISOString(),expiresAt:new Date(now+60000).toISOString(),signerIdentityId:requestIdentity.identityId};value.contentSha256=`sha256-${createHash("sha256").update(value.content).digest("hex")}`;value.signerSignature=new Uint8Array(signDomainPurpose(requestIdentity,requestSignatureDomain,requestSigningPurpose,runtime.encode(requestSigningTuple(value))));const requests=await open(runtime,defs.request,requestStore);await requests.put(defs.request,value.requestId,value);return {root,runtime,defs,value,executionStore,counterPath,options:{runtime,requestStorePath:requestStore,receiptStorePath:receiptStore,executionStorePath:executionStore,epiphanyTrustAnchorPath:anchorPath,epiphanyRuntimeId:"epiphany-yggdrasil",privateKeyPath:deliveryKey,targetAdmitted:()=>true,bridgeCliPath,bifrostRoot:root}};} diff --git a/tests/persona-discord-permit.test.mjs b/tests/persona-discord-permit.test.mjs deleted file mode 100644 index fbba89a..0000000 --- a/tests/persona-discord-permit.test.mjs +++ /dev/null @@ -1,15 +0,0 @@ -import test from "node:test"; -import assert from "node:assert/strict"; -import { createRequire } from "node:module"; -import { mkdtemp } from "node:fs/promises"; -import { tmpdir } from "node:os"; -import { resolve } from "node:path"; -import { enrollDomainIdentity, signDomainPurpose } from "../tools/bifrost-signed-identity.mjs"; -import { permitIdentityDomain, permitPurpose, permitRequestSchema, permitRequestWireTuple, permitSchema, permitSignatureDomain, permitTuple, parseEpiphanyPermitAnchorTuple, parsePermitWireTuple, validatePermit } from "../tools/persona-discord-permit.mjs"; - -const requireCultNet=createRequire(resolve(import.meta.dirname,"..","..","CultLib","packages","cultnet-ts","package.json")); -const {encode}=requireCultNet("@msgpack/msgpack"); - -test("Rust-shaped permit anchors and signatures retain byte authority",async()=>{const root=await mkdtemp(resolve(tmpdir(),"persona-permit-rust-shape-")),identity=await enrollDomainIdentity(resolve(root,"permit.seed"),"permit identity",permitIdentityDomain),now=Date.now(),anchor=parseEpiphanyPermitAnchorTuple(["gamecult.service_trust_anchor.v1","anchor","epiphany-persona-discord-permit","epiphany-starfire",identity.identityId,Array.from(identity.publicKey),"ed25519",permitPurpose,permitSchema,"root",now-1000,null,false],"epiphany-starfire",now),request={requestId:"request-1",requestPayloadSha256:`sha256-${"a".repeat(64)}`,targetRuntimeId:"epiphany-starfire",nonce:"nonce-1",requesterIdentityId:"requester"},permit={schemaVersion:permitSchema,permitId:`permit:${request.requestPayloadSha256}`,requestId:request.requestId,requestPayloadSha256:request.requestPayloadSha256,targetRuntimeId:request.targetRuntimeId,nonce:request.nonce,requesterIdentityId:request.requesterIdentityId,brakeStateDocumentId:"brake",brakeStateDocumentSha256:`sha256-${"b".repeat(64)}`,brakeObservedAt:new Date(now-100).toISOString(),issuedAt:new Date(now).toISOString(),expiresAt:new Date(now+4000).toISOString(),providerIdentityId:identity.identityId,privateStateExposed:false,providerSignature:[]};permit.providerSignature=Array.from(signDomainPurpose(identity,permitSignatureDomain,permitPurpose,encode(permitTuple(permit))));const parsed=validatePermit(permit,request,anchor,encode);assert.ok(parsed.providerSignature instanceof Uint8Array);assert.equal(parsed.providerSignature.length,64);}); - -test("Bifrost and Rust exchange exact positional permit bodies",()=>{const request={schemaVersion:permitRequestSchema,requestId:"request-1",requestPayloadSha256:`sha256-${"a".repeat(64)}`,targetRuntimeId:"epiphany-starfire",nonce:"nonce-1",issuedAt:"2026-08-08T20:00:00.000Z",expiresAt:"2026-08-08T20:00:05.000Z",requesterIdentityId:"requester",privateStateExposed:false,requesterSignature:Uint8Array.from({length:64},(_,i)=>i)};const requestWire=permitRequestWireTuple(request);assert.equal(requestWire.length,10);assert.deepEqual(requestWire.slice(0,9),[request.schemaVersion,request.requestId,request.requestPayloadSha256,request.targetRuntimeId,request.nonce,request.issuedAt,request.expiresAt,request.requesterIdentityId,false]);assert.deepEqual(requestWire[9],request.requesterSignature);const permitWire=[permitSchema,`permit:${request.requestPayloadSha256}`,request.requestId,request.requestPayloadSha256,request.targetRuntimeId,request.nonce,request.requesterIdentityId,"brake",`sha256-${"b".repeat(64)}`,"2026-08-08T20:00:00.000Z","2026-08-08T20:00:00.000Z","2026-08-08T20:00:04.000Z","provider",false,Array.from({length:64},(_,i)=>63-i)];const permit=parsePermitWireTuple(permitWire);assert.equal(permit.providerIdentityId,"provider");assert.deepEqual(permit.providerSignature,permitWire[14]);}); diff --git a/tests/persona-discord-rudp-cross-language.test.mjs b/tests/persona-discord-rudp-cross-language.test.mjs deleted file mode 100644 index f22f93f..0000000 --- a/tests/persona-discord-rudp-cross-language.test.mjs +++ /dev/null @@ -1,29 +0,0 @@ -import test from "node:test"; -import {existsSync} from "node:fs"; -import assert from "node:assert/strict"; -import {spawn,spawnSync} from "node:child_process"; -import {createHash} from "node:crypto"; -import {createRequire} from "node:module"; -import {mkdtemp,readFile,writeFile} from "node:fs/promises"; -import {tmpdir} from "node:os"; -import {resolve} from "node:path"; -import {enrollDomainIdentity,signDomainPurpose} from "../tools/bifrost-signed-identity.mjs"; -import {receiptIdentityDomain,receiptSignatureDomain,receiptSigningPurpose,receiptSigningTuple,requestSigningTuple} from "../tools/persona-discord-delivery-documents.mjs"; -import {startPersonaDiscordDeliveryRudpServer} from "../tools/persona-discord-rudp.mjs"; - -const epiphany=resolve(import.meta.dirname,"..","..","Epiphany"); - -const rustFixture=existsSync(resolve(epiphany,"epiphany-core","src","bin","epiphany-persona-discord-rudp-client-fixture.rs"))&&spawnSync("cargo",["--version"]).status===0; -test("Rust Persona mouth and Node Bifrost exchange one signed request and terminal receipt over CultNet RUDP",{skip:rustFixture?false:"needs cargo and Epiphany's epiphany-persona-discord-rudp-client-fixture fixture (removed in Epiphany 387afe49)"},async()=>{ - const root=await mkdtemp(resolve(tmpdir(),"persona-rudp-cross-language-")),runtime=loadRuntime(),deliveryIdentity=await enrollDomainIdentity(resolve(root,"delivery.seed"),"delivery identity",receiptIdentityDomain),receiptAnchor=resolve(root,"receipt-anchor.msgpack"),requestAnchor=resolve(root,"request-anchor.msgpack"),requestStore=resolve(root,"requests.cc"),receiptStore=resolve(root,"receipts.cc"),identityStore=resolve(root,"mouth.cc"); - await writeFile(receiptAnchor,runtime.msgpack.encode(["gamecult.service_trust_anchor.v1","bifrost-persona-discord-delivery:bifrost-discord-yggdrasil:v0","bifrost-persona-discord-delivery","bifrost-discord-yggdrasil",deliveryIdentity.identityId,new Uint8Array(deliveryIdentity.publicKey),"ed25519",receiptSigningPurpose,"bifrost.persona_discord_delivery_receipt.v0","root",Date.now(),null,false])); - let admitted; - const server=await startPersonaDiscordDeliveryRudpServer({runtime,endpoint:"rudp://127.0.0.1:0",onRequest:async request=>{admitted=request;const digest=`sha256-${createHash("sha256").update(runtime.msgpack.encode(requestSigningTuple(request))).digest("hex")}`,receipt={schemaVersion:"bifrost.persona_discord_delivery_receipt.v0",receiptId:request.requestId,requestId:request.requestId,requestPayloadSha256:digest,status:"completed",channelId:request.channelId,replyToMessageId:request.replyToMessageId,messageId:"node-rudp-message",transport:"bifrost.discord-post",crossingReceiptId:"node-rudp-crossing",receiptUrl:`https://discord.com/channels/g/${request.channelId}/node-rudp-message`,completedAt:new Date().toISOString(),providerIdentityId:deliveryIdentity.identityId,privateStateExposed:false,providerSignature:new Uint8Array()};receipt.providerSignature=new Uint8Array(signDomainPurpose(deliveryIdentity,receiptSignatureDomain,receiptSigningPurpose,runtime.msgpack.encode(receiptSigningTuple(receipt))));return receipt;}}); - try{ - const output=await run("cargo",["run","--quiet","--manifest-path",resolve(epiphany,"epiphany-core","Cargo.toml"),"--bin","epiphany-persona-discord-rudp-client-fixture","--","--request-store",requestStore,"--receipt-store",receiptStore,"--identity-store",identityStore,"--request-anchor",requestAnchor,"--receipt-anchor",receiptAnchor,"--endpoint",`127.0.0.1:${server.address.port}`],{cwd:epiphany}); - assert.equal(output.code,0,`${output.stderr}\nserver diagnostics: ${JSON.stringify(server.diagnostics)}`);const result=JSON.parse(output.stdout.trim());assert.equal(result.status,"completed");assert.equal(result.messageId,"node-rudp-message");assert.equal(admitted.targetRuntimeId,"epiphany-starfire");assert.ok((await readFile(requestStore)).length>0);assert.ok((await readFile(receiptStore)).length>0); - }finally{server.close();} -}); - -function loadRuntime(){const cult=resolve(import.meta.dirname,"..","..","CultLib"),nr=createRequire(resolve(cult,"packages","cultnet-ts","package.json")),cultnet=nr("cultnet-ts"),msgpack=nr("@msgpack/msgpack");return {cultnet,msgpack};} -function run(command,args,options){return new Promise((resolveRun,reject)=>{const child=spawn(command,args,{...options,stdio:["ignore","pipe","pipe"]});let stdout="",stderr="";child.stdout.on("data",chunk=>stdout+=chunk);child.stderr.on("data",chunk=>stderr+=chunk);child.once("error",reject);child.once("close",code=>resolveRun({code,stdout,stderr}));});} diff --git a/tests/persona-feedback-cli.test.mjs b/tests/persona-feedback-cli.test.mjs index ff6f749..cf9c026 100644 --- a/tests/persona-feedback-cli.test.mjs +++ b/tests/persona-feedback-cli.test.mjs @@ -56,8 +56,8 @@ test("Bifrost admits bound feedback as pressure only and exports CultNet",async( const beforeStatus=await Promise.all([store,observationStore,deliveryStore,key].map(path=>readFile(path))); const readiness=cli("status",store,"--observation-store",observationStore,"--private-key",key); assert.equal(readiness.schemaVersion,"bifrost.persona_feedback.readiness.v0");assert.equal(readiness.status,"ready");assert.equal(readiness.bindingCount,1);assert.equal(readiness.pendingFailedCount,0);assert.equal(readiness.privateStateExposed,false); - const staged=cli("status",store,"--observation-store",observationStore,"--private-key",key,"--epiphany-persona-mouth-trust-anchor",resolve(dir,"missing-mouth-anchor.msgpack"),"--epiphany-runtime-id","epiphany-yggdrasil","--epiphany-persona-permit-trust-anchor",resolve(dir,"missing-permit-anchor.msgpack"),"--epiphany-persona-permit-rudp","rudp://127.0.0.1:9","--persona-delivery-private-key",resolve(dir,"missing-delivery.seed"),"--persona-permit-request-private-key",resolve(dir,"missing-permit-request.seed")); - assert.equal(staged.ingressReady,true);assert.deepEqual(staged.ingressReasons,[]);assert.equal(staged.personaDelivery.ready,false);assert.equal(staged.status,"degraded"); + const staged=cli("status",store,"--observation-store",observationStore,"--private-key",key); + assert.equal(staged.ingressReady,true);assert.deepEqual(staged.ingressReasons,[]);assert.equal(staged.personaDelivery,undefined);assert.equal(staged.status,"ready"); const afterStatus=await Promise.all([store,observationStore,deliveryStore,key].map(path=>readFile(path)));assert.deepEqual(afterStatus,beforeStatus); const Store=cr("@gamecult/cultcache-ts").SingleFileMessagePackBackingStore; assert.deepEqual([...new Set((await new Store(observationStore).pullAll()).map(entry=>entry.type))],["bifrost.discord.persona_conversation_event"]); diff --git a/tests/persona-matrix-mouth-runtime.test.mjs b/tests/persona-matrix-mouth-runtime.test.mjs new file mode 100644 index 0000000..abf755f --- /dev/null +++ b/tests/persona-matrix-mouth-runtime.test.mjs @@ -0,0 +1,149 @@ +// The parts of the Matrix mouth that sit on CultLib: the chat_* wire documents, the +// CultCache cursor and brake stores, and the CultNet RUDP endpoint. Needs the built +// CultLib TypeScript packages (CULTLIB_ROOT, else ../CultLib), like the other Bifrost tests. +import test from "node:test"; +import assert from "node:assert/strict"; +import dgram from "node:dgram"; +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, startChatEndpoint } from "../tools/persona-matrix-mouth.mjs"; + +const root = resolve(import.meta.dirname, ".."); +const runtime = loadRuntime(resolve(process.env.CULTLIB_ROOT ?? resolve(root, "..", "CultLib"))); +const { msgpack } = runtime; +const fixtureDir = resolve(import.meta.dirname, "fixtures", "persona-chat-v1"); + +const delivery = { + seq: 7, roomAlias: "#operator:gamecult.org", roomId: "!abc:gamecult.org", eventId: "$e1", sender: "@meta:gamecult.org", senderDisplay: "Meta", + text: "hello eureka", atUnixMs: 1761700000000, + context: [{ sender: "@eureka:gamecult.org", text: "earlier", atUnixMs: 1761699999000 }], +}; +const samples = { + [docs.kinds.pendingRequest]: { tuple: docs.pendingRequestTuple({ persona: "eureka", afterSeq: 3 }), parse: docs.parsePendingRequest }, + [docs.kinds.pending]: { tuple: docs.pendingTuple({ persona: "eureka", deliveries: [delivery] }), parse: docs.parsePending }, + [docs.kinds.ack]: { tuple: docs.ackTuple({ persona: "eureka", throughSeq: 7 }), parse: docs.parseAck }, + [docs.kinds.sendRequest]: { tuple: docs.sendRequestTuple({ requestId: "req-1", persona: "eureka", roomAlias: "#operator:gamecult.org", text: "hi" }), parse: docs.parseSendRequest }, + [docs.kinds.sendReceipt]: { tuple: docs.sendReceiptTuple({ requestId: "req-1", eventId: "$sent" }), parse: docs.parseSendReceipt }, +}; + +for (const [kind, { tuple, parse }] of Object.entries(samples)) { + test(`golden fixture: ${kind} is byte-stable and parses back`, async () => { + const bytes = Buffer.from(msgpack.encode(tuple)); + const file = resolve(fixtureDir, `${kind}.msgpack`); + if (process.env.UPDATE_FIXTURES === "1") { await mkdir(fixtureDir, { 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 chat documents refuse malformed tuples", () => { + assert.throws(() => docs.parseAck(["eureka"])); + assert.throws(() => docs.parseAck(["eureka", -1])); + assert.throws(() => docs.parsePendingRequest(["", 0])); + assert.throws(() => docs.parseSendRequest(["r", "eureka", "#a:b", "x".repeat(4001)])); + assert.throws(() => docs.parseSendRequest(["r", "eureka", "#a:b", 5])); + assert.throws(() => docs.parseSendReceipt(["r", "$e", "Brake"]), /exactly one/); + assert.throws(() => docs.parseSendReceipt(["r", null, null]), /exactly one/); + assert.throws(() => docs.sendReceiptTuple({ requestId: "r", refusal: "Invented" })); + assert.deepEqual(docs.parseSendReceipt(["r", null, "Brake"]), { requestId: "r", eventId: null, refusal: "Brake" }); +}); + +test("the cursor is a CultCache file that survives a restart", async () => { + const path = resolve(await mkdtemp(resolve(tmpdir(), "mouth-cursor-")), "cursor.cc"); + assert.equal(await cultcacheState({ runtime, path }).load(), null); + await cultcacheState({ runtime, path }).save({ nextBatch: "s72_9", seq: 12 }); + const loaded = await cultcacheState({ runtime, path }).load(); + assert.deepEqual([loaded.nextBatch, loaded.seq], ["s72_9", 12]); +}); + +async function writeControl(path, brakeTuple) { + const store = new runtime.SingleFileMessagePackBackingStore(path); + await store.push({ key: "brake", type: "eureka.control.brake.v1", payload: msgpack.encode(brakeTuple), storedAt: new Date().toISOString() }); +} + +test("the brake reader reads a real CultCache control store", async () => { + const dir = await mkdtemp(resolve(tmpdir(), "mouth-control-")); + const reader = path => createBrakeReader({ controlPath: path, readStore: runtime.readSingleFileStore, decode: msgpack.decode }); + assert.equal(await reader(resolve(dir, "missing.cc"))(), false); + await writeControl(resolve(dir, "held.cc"), [false, "2026-10-29T00:00:00Z", "op"]); + assert.equal(await reader(resolve(dir, "held.cc"))(), false); + await writeControl(resolve(dir, "released.cc"), [true, "2026-10-29T00:00:00Z", "op"]); + assert.equal(await reader(resolve(dir, "released.cc"))(), true); + await writeFile(resolve(dir, "garbage.cc"), "not a store"); + assert.equal(await reader(resolve(dir, "garbage.cc"))(), false); +}); + +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 = { + pendingAfter: after => (after < 7 ? [delivery] : []), + ack: async through => { calls.acks.push(through); }, + 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"; + const rejected = []; + const server = await startChatEndpoint({ mouth, persona: "eureka", runtime, port: 0, connectionKey: key, runtimeId: "mouth-test", log: line => rejected.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 => { + const message = runtime.cultnet.parseCultNetMessage(msgpack.decode(value.payload), "cultnet.schema.v0"); + replies.push(message.document); + waiting.splice(0).forEach(w => w()); + }); + const nextReply = () => new Promise((ok, fail) => { const t = setTimeout(() => fail(new Error(`no reply from the mouth; it logged: ${rejected.join(" | ")}`)), 5000); waiting.push(() => { clearTimeout(t); ok(replies.at(-1)); }); }); + 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"))); + try { + connection.connect(); + for (let i = 0; i < 100 && !connection.connected; i++) await new Promise(r => setTimeout(r, 20)); + assert.ok(connection.connected, "the RUDP session opens with the credential-derived connection id"); + let reply = nextReply(); + put(docs.kinds.pendingRequest, "eureka", docs.pendingRequestTuple({ persona: "eureka", afterSeq: 0 })); + let document = await reply; + assert.equal(document.schemaId, docs.kinds.pending); + const pending = docs.parsePending(msgpack.decode(document.payload)); + assert.deepEqual(pending.deliveries.map(d => d.eventId), ["$e1"]); + assert.deepEqual(pending.deliveries[0].context, delivery.context); + + put(docs.kinds.ack, "eureka:7", docs.ackTuple({ persona: "eureka", throughSeq: 7 })); + reply = nextReply(); + put(docs.kinds.sendRequest, "req-2", docs.sendRequestTuple({ requestId: "req-2", persona: "eureka", roomAlias: "#operator:gamecult.org", text: "muted" })); + document = await reply; + assert.equal(document.schemaId, docs.kinds.sendReceipt); + assert.deepEqual(docs.parseSendReceipt(msgpack.decode(document.payload)), { requestId: "req-2", eventId: null, refusal: "Brake" }); + put(docs.kinds.ack, "other:9", docs.ackTuple({ persona: "other", throughSeq: 9 })); + assert.deepEqual(calls.acks, [7]); + + reply = nextReply(); + put(docs.kinds.pendingRequest, "other", docs.pendingRequestTuple({ persona: "other", afterSeq: 0 })); + put(docs.kinds.sendRequest, "req-3", docs.sendRequestTuple({ requestId: "req-3", persona: "eureka", roomAlias: "#operator:gamecult.org", text: "speak" })); + document = await reply; + assert.equal(document.schemaId, docs.kinds.sendReceipt, "a request naming another Persona gets no answer; the next request is still served"); + assert.equal(docs.parseSendReceipt(msgpack.decode(document.payload)).eventId, "$sent"); + assert.deepEqual(calls.acks, [7], "an ack naming another Persona is not applied"); + } finally { + server.close(); + connection.close(); + try { client.close(); } catch { /* the connection already closed its socket */ } + } +}); + +test("the RUDP endpoint refuses to listen beyond loopback", async () => { + await assert.rejects(startChatEndpoint({ mouth: {}, persona: "eureka", runtime, port: 0, host: "0.0.0.0", connectionKey: "k", runtimeId: "m" }), /loopback/); +}); + +test("loopback means exactly an IPv4 127.x.y.z address", async () => { + const refuses = host => assert.rejects(startChatEndpoint({ mouth: {}, persona: "eureka", runtime, port: 0, host, connectionKey: "k", runtimeId: "m" }), /loopback/); + for (const host of ["9127.0.0.1", "127.0.0.1x", "10.127.0.0.1", "127.0.0", ""]) await refuses(host); + const ok = await startChatEndpoint({ mouth: {}, persona: "eureka", runtime, port: 0, host: "127.10.20.30", connectionKey: "k", runtimeId: "m" }).catch(e => e); + assert.ok(!/loopback/.test(ok?.message ?? ""), "a multi-digit loopback address is accepted by the check"); + ok.close?.(); +}); diff --git a/tests/persona-matrix-mouth.test.mjs b/tests/persona-matrix-mouth.test.mjs new file mode 100644 index 0000000..5f336eb --- /dev/null +++ b/tests/persona-matrix-mouth.test.mjs @@ -0,0 +1,347 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { createServer } from "node:http"; +import { readFileSync, readdirSync, statSync } from "node:fs"; +import { resolve, join } from "node:path"; +import { createMouth, createBrakeReader, createRing, redact } from "../tools/persona-matrix-mouth.mjs"; + +const TOKEN = "syt_SECRET_TOKEN_9f3"; +const HS_NAME = "gamecult.test"; +const ROOMS = { [`#swarm:${HS_NAME}`]: "!swarm:gamecult.test", [`#operator:${HS_NAME}`]: "!operator:gamecult.test" }; +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 } }); +const tick = () => new Promise(r => setImmediate(r)); + +// A fake Matrix client-server API: just the endpoints the mouth uses. +async function fakeHomeserver({ sync, failWith } = {}) { + const seen = []; + const sent = new Map(); // txnId -> event_id + 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 record = { method: req.method, 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", ""); + if (p === "/account/whoami") return reply(200, { user_id: SELF }); + if (p.startsWith("/directory/room/")) return reply(200, { room_id: ROOMS[p.slice("/directory/room/".length)] }); + if (p.startsWith("/join/")) return reply(200, { room_id: p.slice(6) }); + if (p === "/sync") return reply(200, sync(record.query.get("since"))); + const m = p.match(/^\/rooms\/(.+)\/send\/m\.room\.message\/(.+)$/); + if (m && req.method === "PUT") { + if (!sent.has(m[2])) sent.set(m[2], `$event${sent.size + 1}`); + return reply(200, { event_id: sent.get(m[2]) }); + } + 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}`, + seen, + puts: () => seen.filter(r => r.method === "PUT"), + close: () => new Promise(ok => server.close(ok)), + }; +} + +function memoryState(initial = null) { + const box = { value: initial }; + return { box, load: async () => box.value, save: async v => { box.value = v; } }; +} + +const joined = events => ({ rooms: { join: events } }); +function mouthFor(hs, overrides = {}) { + return createMouth({ + persona: "eureka", homeserver: hs.url, rooms: Object.keys(ROOMS), token: TOKEN, state: memoryState(), + brake: async () => true, ring: async () => {}, log: () => {}, syncTimeoutMs: 1, ...overrides, + }); +} + +// The scripted timeline: the sync with no position is the baseline; later positions advance it. +const timeline = { + null: { next_batch: "b0", ...joined({ "!swarm:gamecult.test": { timeline: { events: [message("$old", `@alice:${HS_NAME}`, "history before the ear", 1)] } } }) }, + b0: { + next_batch: "b1", + ...joined({ + "!swarm:gamecult.test": { + state: { events: [{ type: "m.room.member", state_key: `@alice:${HS_NAME}`, content: { displayname: "Alice" } }] }, + timeline: { events: [ + message("$m1", `@alice:${HS_NAME}`, "hello eureka", 1000), + message("$m2", SELF, "my own words", 1001), + { type: "m.room.topic", event_id: "$t", sender: `@alice:${HS_NAME}`, content: { topic: "x" } }, + ] }, + }, + "!elsewhere:gamecult.test": { timeline: { events: [message("$x", `@bob:${HS_NAME}`, "not my room", 1002)] } }, + }), + }, + b1: { next_batch: "b2", ...joined({ "!operator:gamecult.test": { timeline: { events: [message("$m3", `@bob:${HS_NAME}`, "second batch", 2000)] } } }) }, +}; +const script = since => timeline[since] ?? { next_batch: since }; + +test("sync_message_becomes_delivery: another sender's message in a joined room is one delivery with context; own and foreign-room messages are none", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + const mouth = mouthFor(hs); + assert.deepEqual(await mouth.syncOnce(), [], "history at first sync is context, not mail"); + const made = await mouth.syncOnce(); + assert.equal(made.length, 1); + const [d] = made; + assert.deepEqual({ roomAlias: d.roomAlias, roomId: d.roomId, eventId: d.eventId, sender: d.sender, senderDisplay: d.senderDisplay, text: d.text, atUnixMs: d.atUnixMs }, { + roomAlias: `#swarm:${HS_NAME}`, roomId: "!swarm:gamecult.test", eventId: "$m1", sender: `@alice:${HS_NAME}`, senderDisplay: "Alice", text: "hello eureka", atUnixMs: 1000, + }); + assert.deepEqual(d.context, [{ sender: `@alice:${HS_NAME}`, text: "history before the ear", atUnixMs: 1 }]); + assert.equal(d.seq, 1); + assert.deepEqual(mouth.pendingAfter(0).map(x => x.eventId), ["$m1"]); + assert.deepEqual(mouth.pendingAfter(1), []); + } finally { await hs.close(); } +}); + +test("context holds the last 20 messages before the delivery, oldest first", async () => { + const many = Array.from({ length: 25 }, (_, i) => message(`$h${i}`, `@alice:${HS_NAME}`, `m${i}`, 10 + i)); + const hs = await fakeHomeserver({ + sync: since => since === null + ? { next_batch: "b0", ...joined({ "!swarm:gamecult.test": { timeline: { events: many } } }) } + : { next_batch: "b1", ...joined({ "!swarm:gamecult.test": { timeline: { events: [message("$n", `@bob:${HS_NAME}`, "now", 99)] } } }) }, + }); + try { + const mouth = mouthFor(hs); + await mouth.syncOnce(); + const [d] = await mouth.syncOnce(); + assert.equal(d.context.length, 20); + assert.equal(d.context[0].text, "m5"); + assert.equal(d.context[19].text, "m24"); + } finally { await hs.close(); } +}); + +test("next_batch_persisted_only_after_ack: restart before ack re-delivers; after ack does not", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + const state = memoryState(); + const first = mouthFor(hs, { state }); + await first.syncOnce(); // baseline b0, nothing pending: persisted at once + assert.equal(state.box.value.nextBatch, "b0"); + const [d] = await first.syncOnce(); // delivery in the batch ending b1 + assert.equal(state.box.value.nextBatch, "b0", "b1 must not be persisted before an ack"); + + const restarted = mouthFor(hs, { state }); // crash before the ack + const [again] = await restarted.syncOnce(); + assert.equal(again.eventId, "$m1", "the unacked message arrives again"); + assert.ok(again.seq > d.seq, "sequence numbers never repeat across a restart"); + + await restarted.ack(again.seq); + assert.equal(state.box.value.nextBatch, "b1"); + + const afterAck = mouthFor(hs, { state }); + assert.deepEqual((await afterAck.syncOnce()).map(x => x.eventId), ["$m3"], "an acked batch is not delivered again: only the newer message arrives"); + assert.equal(hs.seen.filter(r => r.path.endsWith("/sync")).at(-1).query.get("since"), "b1"); + } finally { await hs.close(); } +}); + +test("an ack covers only the batches whose deliveries it reaches", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + const state = memoryState(); + const mouth = mouthFor(hs, { state }); + await mouth.syncOnce(); + const [a] = await mouth.syncOnce(); // batch ending b1 holds seq 1 + const [b] = await mouth.syncOnce(); // batch ending b2 holds seq 2 + assert.deepEqual([a.seq, b.seq], [1, 2]); + await mouth.ack(1); + assert.equal(state.box.value.nextBatch, "b1", "acking seq 1 persists the batch that ends at seq 1, not the later one"); + await mouth.ack(2); + assert.equal(state.box.value.nextBatch, "b2"); + assert.equal(mouth.pendingCount, 0); + } finally { await hs.close(); } +}); + +test("send_uses_request_id_txn: the PUT carries txnId = request_id and a repeat sends the same txnId", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + const mouth = mouthFor(hs); + const request = { requestId: "req 7f", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "hello world" }; + const one = await mouth.send(request); + const two = await mouth.send(request); + assert.equal(one.refusal, null); + assert.equal(one.eventId, two.eventId, "the homeserver deduplicates by txnId, so a retry is the same event"); + const puts = hs.puts(); + assert.equal(puts.length, 2); + for (const put of puts) { + assert.equal(put.path, "/_matrix/client/v3/rooms/!swarm:gamecult.test/send/m.room.message/req 7f"); + assert.deepEqual(put.body, { msgtype: "m.text", body: "hello world" }); + } + assert.notEqual((await mouth.send({ ...request, requestId: "req-8" })).eventId, one.eventId, "a different request is a different txn"); + } finally { await hs.close(); } +}); + +test("send refuses another Persona's request, an unknown room and empty text, without a PUT", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + const mouth = mouthFor(hs); + assert.equal((await mouth.send({ requestId: "a", persona: "other", roomAlias: `#swarm:${HS_NAME}`, text: "x" })).refusal, "PersonaMismatch"); + assert.equal((await mouth.send({ requestId: "b", persona: "eureka", roomAlias: "#nowhere:gamecult.test", text: "x" })).refusal, "UnknownRoom"); + assert.equal((await mouth.send({ requestId: "c", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: " " })).refusal, "EmptyText"); + assert.equal(hs.puts().length, 0); + } finally { await hs.close(); } +}); + +const payload = value => Buffer.from(JSON.stringify(value)); +const store = (...envelopes) => ({ absent: false, envelopes }); +const brakeEnvelope = value => ({ type: "eureka.control.brake.v1", key: "brake", payload: payload(value) }); +const jsonDecode = b => JSON.parse(Buffer.from(b).toString()); +const readerFor = read => createBrakeReader({ controlPath: "x", readStore: () => read, decode: jsonDecode }); + +test("held_brake_refuses_send: held, throwing, missing and non-boolean brakes refuse with Brake and nothing reaches the homeserver", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + const request = { requestId: "r1", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "speak" }; + const brakes = [ + async () => false, + async () => { throw new Error("control store unreadable"); }, + readerFor({ absent: true, envelopes: [] }), + readerFor(store(brakeEnvelope([false, "t", "op"]))), + async () => "yes", + ]; + for (const brake of brakes) { + const receipt = await mouthFor(hs, { brake }).send(request); + assert.deepEqual({ eventId: receipt.eventId, refusal: receipt.refusal }, { eventId: null, refusal: "Brake" }); + } + assert.equal(hs.puts().length, 0); + assert.equal(hs.seen.length, 0, "a refused send does not even touch the homeserver"); + const released = await mouthFor(hs, { brake: readerFor(store(brakeEnvelope([true, "t", "op"]))) }).send(request); + assert.equal(released.refusal, null, "the same request goes through once the brake is released"); + } finally { await hs.close(); } +}); + +test("brake reader: only a decoded released brake counts as released", async () => { + assert.equal(await readerFor(store(brakeEnvelope([true, "t", "op"])))(), true); + assert.equal(await readerFor(store(brakeEnvelope([false, "t", "op"])))(), false); + assert.equal(await readerFor(store(brakeEnvelope({ 0: true })))(), true, "integer-keyed map form"); + assert.equal(await readerFor(store(brakeEnvelope({ 0: false })))(), false); + assert.equal(await readerFor(store(brakeEnvelope(["true"])))(), false, "a non-boolean is not a release"); + assert.equal(await readerFor(store())(), false, "no brake document"); + assert.equal(await readerFor(store({ ...brakeEnvelope([true]), key: "dial" }))(), false, "wrong key"); + assert.equal(await readerFor(store({ ...brakeEnvelope([true]), type: "other.v1" }))(), false, "wrong type"); + assert.equal(await createBrakeReader({ controlPath: "x", readStore: () => { throw new Error("corrupt"); }, decode: jsonDecode })(), false); + assert.equal(await readerFor(store({ type: "eureka.control.brake.v1", key: "brake", payload: Buffer.from("not json") }))(), false, "undecodable bytes"); +}); + +test("ring_starts_persona_unit_once: two deliveries in one idle window cause one start; a drained mailbox rings again", async () => { + const hs = await fakeHomeserver({ sync: script }); + try { + let rings = 0; + const mouth = mouthFor(hs, { ring: async () => { rings++; } }); + await mouth.syncOnce(); // baseline + await mouth.syncOnce(); // delivery 1 rings + await mouth.syncOnce(); // delivery 2, same window, still unacked + await tick(); + assert.equal(rings, 1); + } finally { await hs.close(); } + + const every = await fakeHomeserver({ sync: since => ({ next_batch: `n${since}`, ...joined({ "!swarm:gamecult.test": { timeline: { events: [message(`$e${since}`, `@alice:${HS_NAME}`, "again", 5)] } } }) }) }); + try { + let clock = 0, rings = 0; + const mouth = mouthFor(every, { ring: async () => { rings++; }, now: () => clock }); + await mouth.syncOnce(); // baseline + await mouth.syncOnce(); await tick(); + assert.equal(rings, 1); + await mouth.syncOnce(); await tick(); + assert.equal(rings, 1, "inside the window and unacked: no second start"); + await mouth.ack(2); + await mouth.syncOnce(); await tick(); + assert.equal(rings, 2, "after a drain a new delivery rings again"); + clock = 6 * 60 * 1000; + await mouth.syncOnce(); await tick(); + assert.equal(rings, 3, "a delivery left unanswered past the window rings again"); + } finally { await every.close(); } +}); + +test("a failed ring is logged once and never retried", async () => { + const logs = []; + let calls = 0; + const hs = await fakeHomeserver({ sync: script }); + try { + const mouth = mouthFor(hs, { ring: async () => { calls++; throw new Error("systemctl refused"); }, log: l => logs.push(l) }); + await mouth.syncOnce(); await mouth.syncOnce(); await tick(); await tick(); + assert.equal(calls, 1); + assert.equal(logs.filter(l => l.includes("ring failed")).length, 1); + } finally { await hs.close(); } +}); + +test("ring runs exactly systemctl start --no-block on the instance's unit, with no token in the child's environment", async () => { + const calls = []; + await createRing({ instance: "eureka", run: (cmd, args, options, cb) => { calls.push({ cmd, args, options }); cb(null); } })(); + assert.equal(calls.length, 1); + assert.deepEqual([calls[0].cmd, calls[0].args], ["systemctl", ["start", "--no-block", "mind-persona@eureka.service"]]); + assert.deepEqual(Object.keys(calls[0].options.env), ["PATH"]); + const failed = []; + await createRing({ instance: "eureka", run: (c, a, o, cb) => cb(new Error("denied")), log: l => failed.push(l) })(); + assert.deepEqual(failed, ["ring failed: denied"]); +}); + +test("the Matrix token goes only in the Authorization header: never in logs, errors, deliveries or receipts", async () => { + const logs = []; + const bad = await fakeHomeserver({ sync: script, failWith: 401 }); + try { + const mouth = mouthFor(bad, { log: l => logs.push(l) }); + await assert.rejects(mouth.syncOnce(), err => { assert.ok(!err.message.includes(TOKEN)); return true; }); + const receipt = await mouth.send({ requestId: "r", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "x" }).catch(e => ({ thrown: e.message })); + assert.ok(!JSON.stringify(receipt).includes(TOKEN)); + const controller = new AbortController(); + const running = mouthFor(bad, { log: l => logs.push(l) }).run(controller.signal); + await new Promise(r => setTimeout(r, 30)); + controller.abort(); + void running; // the run loop is left to its backoff; only its log lines matter here + } finally { await bad.close(); } + + const unreachable = mouthFor({ url: "http://127.0.0.1:1" }, { + log: l => logs.push(l), + fetch: async (url, init) => { throw new Error(`connect failed with ${init.headers.authorization} to ${url}`); }, + }); + await assert.rejects(unreachable.syncOnce(), err => { assert.ok(!err.message.includes(TOKEN), "transport errors are redacted"); return true; }); + + const hs = await fakeHomeserver({ sync: script }); + try { + const mouth = mouthFor(hs, { log: l => logs.push(l) }); + await mouth.syncOnce(); + const made = await mouth.syncOnce(); + const receipt = await mouth.send({ requestId: "ok", persona: "eureka", roomAlias: `#swarm:${HS_NAME}`, text: "hi" }); + assert.ok(!JSON.stringify([made, receipt, mouth.pendingAfter(0)]).includes(TOKEN)); + assert.ok(hs.seen.every(r => r.auth === `Bearer ${TOKEN}`), "the token is presented to the homeserver on every call"); + } finally { await hs.close(); } + + assert.ok(logs.length > 0, "failures were logged"); + assert.ok(logs.every(l => !l.includes(TOKEN)), `a log line carried the token: ${logs.find(l => l.includes(TOKEN))}`); + assert.equal(redact(`x ${TOKEN} y`, TOKEN), "x [redacted] y"); +}); + +test("a homeserver error is reported with its status and none of its body", async () => { + const hs = await fakeHomeserver({ sync: script, failWith: 500 }); + try { + await assert.rejects(mouthFor(hs).syncOnce(), err => /answered 500/.test(err.message) && !/bad token/.test(err.message)); + } finally { await hs.close(); } +}); + +function walk(dir, out = []) { + for (const name of readdirSync(dir)) { + if (name === ".git" || name === "node_modules" || name === "bin" || name === "obj") continue; + const full = join(dir, name); + statSync(full).isDirectory() ? walk(full, out) : out.push(full); + } + return out; +} + +test("the Discord posting actuator is gone and the Matrix path names no Discord", () => { + const root = resolve(import.meta.dirname, ".."); + for (const file of ["tools/persona-matrix-mouth.mjs", "tools/persona-chat-documents.mjs"]) { + assert.ok(!/discord/i.test(readFileSync(resolve(root, file), "utf8")), `${file} mentions Discord`); + } + const survivors = walk(root).filter(f => /\.(mjs|cs|json|ya?ml|ps1|sh)$/.test(f) && /persona-discord-delivery/.test(readFileSync(f, "utf8")) && !/persona-matrix-mouth\.test\.mjs$/.test(f)); + assert.deepEqual(survivors, []); + for (const gone of ["persona-discord-delivery.mjs", "persona-discord-rudp.mjs", "persona-discord-delivery-documents.mjs", "persona-discord-permit.mjs"]) { + assert.throws(() => statSync(resolve(root, "tools", gone))); + } +}); diff --git a/tools/persona-chat-documents.mjs b/tools/persona-chat-documents.mjs new file mode 100644 index 0000000..3778379 --- /dev/null +++ b/tools/persona-chat-documents.mjs @@ -0,0 +1,114 @@ +// gamecult.persona.chat_*.v1: the documents Huginn and the Matrix mouth exchange. +// Each kind is a positional tuple (the CultCache payload shape); this module owns +// the field order, the parsers and the envelope check. It does no I/O and loads no +// runtime, so the wire encoder (MessagePack) stays with the caller. + +export const kinds = Object.freeze({ + pendingRequest: "gamecult.persona.chat_pending_request.v1", + pending: "gamecult.persona.chat_pending.v1", + ack: "gamecult.persona.chat_ack.v1", + sendRequest: "gamecult.persona.chat_send_request.v1", + sendReceipt: "gamecult.persona.chat_send_receipt.v1", +}); + +export const refusals = Object.freeze(["Brake", "PersonaMismatch", "UnknownRoom", "EmptyText", "SendFailed"]); + +const maxTextBytes = 4000; +const maxContext = 20; + +export function pendingRequestTuple({ persona, afterSeq }) { + return [text(persona, "persona"), seq(afterSeq, "after_seq")]; +} +export function parsePendingRequest(t) { + const [persona, afterSeq] = tuple(t, 2, "chat_pending_request"); + return { persona: text(persona, "persona"), afterSeq: seq(afterSeq, "after_seq") }; +} + +export function ackTuple({ persona, throughSeq }) { + return [text(persona, "persona"), seq(throughSeq, "through_seq")]; +} +export function parseAck(t) { + const [persona, throughSeq] = tuple(t, 2, "chat_ack"); + return { persona: text(persona, "persona"), throughSeq: seq(throughSeq, "through_seq") }; +} + +export function sendRequestTuple({ requestId, persona, roomAlias, text: body }) { + return [text(requestId, "request_id"), text(persona, "persona"), text(roomAlias, "room_alias"), String(body)]; +} +export function parseSendRequest(t) { + const [requestId, persona, roomAlias, body] = tuple(t, 4, "chat_send_request"); + if (typeof body !== "string") throw new Error("chat_send_request text must be a string"); + if (Buffer.byteLength(body, "utf8") > maxTextBytes) throw new Error("chat_send_request text is over the byte bound"); + return { requestId: text(requestId, "request_id"), persona: text(persona, "persona"), roomAlias: text(roomAlias, "room_alias"), text: body }; +} + +export function sendReceiptTuple({ requestId, eventId = null, refusal = null }) { + if ((eventId === null) === (refusal === null)) throw new Error("a send receipt carries exactly one of event_id and refusal"); + if (refusal !== null && !refusals.includes(refusal)) throw new Error(`unknown refusal ${refusal}`); + return [text(requestId, "request_id"), eventId, refusal]; +} +export function parseSendReceipt(t) { + const [requestId, eventId, refusal] = tuple(t, 3, "chat_send_receipt"); + if ((eventId === null) === (refusal === null)) throw new Error("a send receipt carries exactly one of event_id and refusal"); + return { requestId: text(requestId, "request_id"), eventId: eventId === null ? null : text(eventId, "event_id"), refusal }; +} + +export function deliveryTuple(d) { + return [ + seq(d.seq, "seq"), text(d.roomAlias, "room_alias"), text(d.roomId, "room_id"), text(d.eventId, "event_id"), + text(d.sender, "sender"), text(d.senderDisplay, "sender_display"), String(d.text), seq(d.atUnixMs, "at_unix_ms"), + d.context.slice(-maxContext).map(c => [text(c.sender, "sender"), String(c.text), seq(c.atUnixMs, "at_unix_ms")]), + ]; +} +export function parseDelivery(t) { + const [s, roomAlias, roomId, eventId, sender, senderDisplay, body, atUnixMs, context] = tuple(t, 9, "delivery"); + if (!Array.isArray(context)) throw new Error("delivery context must be a list"); + return { + seq: seq(s, "seq"), roomAlias: text(roomAlias, "room_alias"), roomId: text(roomId, "room_id"), eventId: text(eventId, "event_id"), + sender: text(sender, "sender"), senderDisplay: text(senderDisplay, "sender_display"), text: String(body), atUnixMs: seq(atUnixMs, "at_unix_ms"), + context: context.map(c => { const [cs, ct, ca] = tuple(c, 3, "context"); return { sender: text(cs, "sender"), text: String(ct), atUnixMs: seq(ca, "at_unix_ms") }; }), + }; +} + +export function pendingTuple({ persona, deliveries }) { + return [text(persona, "persona"), deliveries.map(deliveryTuple)]; +} +export function parsePending(t) { + const [persona, deliveries] = tuple(t, 2, "chat_pending"); + if (!Array.isArray(deliveries)) throw new Error("chat_pending deliveries must be a list"); + return { persona: text(persona, "persona"), deliveries: deliveries.map(parseDelivery) }; +} + +// The CultNet document_put_raw envelope every chat_* document travels in. +export function envelope({ kind, recordKey, payload, sourceRuntimeId, sourceAgentId, sourceRole }) { + return { + schemaVersion: "cultnet.document_put_raw.v0", + messageId: `${kind}:${recordKey}`, + document: { + schemaId: kind, recordKey, storedAt: new Date().toISOString(), payloadEncoding: "messagepack", payload, + sourceRuntimeId, sourceAgentId, sourceRole, tags: ["cultnet.transport.rudp.v0"], + }, + }; +} + +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 (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; +} + +function tuple(t, n, what) { + if (!Array.isArray(t) || t.length !== n) throw new Error(`${what} must be a ${n}-field tuple`); + return t; +} +function text(v, what) { + if (typeof v !== "string" || v === "") throw new Error(`${what} must be a non-empty string`); + return v; +} +function seq(v, what) { + if (!Number.isSafeInteger(v) || v < 0) throw new Error(`${what} must be a non-negative integer`); + return v; +} diff --git a/tools/persona-discord-delivery-documents.mjs b/tools/persona-discord-delivery-documents.mjs deleted file mode 100644 index a99ff3b..0000000 --- a/tools/persona-discord-delivery-documents.mjs +++ /dev/null @@ -1,43 +0,0 @@ -export const requestType = "epiphany.persona_discord_delivery_request"; -export const requestSchema = "epiphany.persona_discord_delivery_request.v0"; -export const requestSigningPurpose = "epiphany.persona-discord.delivery-request.v0"; -export const requestIdentityDomain = "epiphany.persona-discord-delivery-request.identity.v0\0"; -export const requestSignatureDomain = "epiphany.persona-discord-delivery-request.signature.v0\0"; -export const receiptType = "bifrost.persona_discord_delivery_receipt"; -export const receiptSchema = "bifrost.persona_discord_delivery_receipt.v0"; -export const receiptSigningPurpose = "bifrost.persona-discord.delivery-receipt.v0"; -export const receiptIdentityDomain = "bifrost.persona-discord-delivery-receipt.identity.v0\0"; -export const receiptSignatureDomain = "bifrost.persona-discord-delivery-receipt.signature.v0\0"; -export const executionType = "bifrost.persona_discord_delivery_execution"; -export const executionSchema = "bifrost.persona_discord_delivery_execution.v0"; - -export function buildPersonaDiscordDeliveryDefinitions(defineDocumentType) { - const make=(type,schema,name,parse,members)=>defineDocumentType({type,schemaName:type,schemaId:schema,schemaVersion:schema,contentHash:schema,global:false,name,schema:{parse},members}); - return { - request:make(requestType,requestSchema,"requestId",parseRequest,requestMembers), - receipt:make(receiptType,receiptSchema,"receiptId",parseReceipt,receiptMembers), - execution:make(executionType,executionSchema,"requestId",parseExecution), - }; -} - -const requestMembers = ["schemaVersion","requestId","effectDocumentId","targetRuntimeId","personaAgentId","channelId","replyToMessageId","content","contentSha256","issuedAt","expiresAt","privateStateExposed","signerIdentityId","signerSignature"].map((memberName,slot)=>({slot,memberName,typeName:memberName==="privateStateExposed"?"boolean":memberName==="signerSignature"?"byte[]":"string",isName:memberName==="requestId"})); -const receiptMembers = ["schemaVersion","receiptId","requestId","requestPayloadSha256","status","channelId","replyToMessageId","messageId","transport","crossingReceiptId","receiptUrl","completedAt","providerIdentityId","privateStateExposed","providerSignature"].map((memberName,slot)=>({slot,memberName,typeName:memberName==="privateStateExposed"?"boolean":memberName==="providerSignature"?"byte[]":"string",isName:memberName==="receiptId"})); - -export function requestSigningTuple(v){return [v.schemaVersion,v.requestId,v.effectDocumentId,v.targetRuntimeId,v.personaAgentId,v.channelId,v.replyToMessageId,v.content,v.contentSha256,v.issuedAt,v.expiresAt,v.privateStateExposed,v.signerIdentityId];} -export function receiptSigningTuple(v){return [v.schemaVersion,v.receiptId,v.requestId,v.requestPayloadSha256,v.status,v.channelId,v.replyToMessageId,v.messageId,v.transport,v.crossingReceiptId,v.receiptUrl,v.completedAt,v.providerIdentityId,v.privateStateExposed];} -export function receiptDocumentTuple(v){return [...receiptSigningTuple(v),v.providerSignature];} - -export function parseRequest(value){const v=typed(value,requestMembers,"Persona Discord delivery request");closed(v,["schemaName","schemaVersion","requestId","effectDocumentId","targetRuntimeId","personaAgentId","channelId","replyToMessageId","content","contentSha256","issuedAt","expiresAt","privateStateExposed","signerIdentityId","signerSignature"],"request");exact(v.schemaVersion,requestSchema,"request schema");for(const key of ["requestId","effectDocumentId","targetRuntimeId","personaAgentId","channelId","contentSha256","issuedAt","expiresAt","signerIdentityId"])text(v[key],key);content(v.content);optionalText(v.replyToMessageId,"replyToMessageId");if(v.privateStateExposed!==false)throw new Error("request privateStateExposed must be false.");bytes64(v.signerSignature,"signerSignature");sha(v.contentSha256,"contentSha256");const issued=date(v.issuedAt,"issuedAt"),expires=date(v.expiresAt,"expiresAt");if(expires<=issued||expires-issued>120000)throw new Error("request lifetime must be positive and at most 120 seconds.");return v;} -export function parseReceipt(value){const v=typed(value,receiptMembers,"Persona Discord delivery receipt");exact(v.schemaVersion,receiptSchema,"receipt schema");for(const key of ["receiptId","requestId","requestPayloadSha256","status","channelId","completedAt","providerIdentityId"])text(v[key],key);for(const key of ["replyToMessageId","messageId","transport","crossingReceiptId","receiptUrl"])optionalText(v[key],key);if(v.receiptId!==v.requestId)throw new Error("receiptId must equal requestId.");if(!["completed","failed","unknown"].includes(v.status))throw new Error("receipt status is invalid.");if(v.status==="completed"&&[v.messageId,v.transport,v.crossingReceiptId,v.receiptUrl].some(x=>!x))throw new Error("completed receipt lacks exact delivery evidence.");if(v.status!=="completed"&&[v.messageId,v.transport,v.crossingReceiptId,v.receiptUrl].some(Boolean))throw new Error("non-completed receipt may not claim delivery evidence.");if(v.privateStateExposed!==false)throw new Error("receipt privateStateExposed must be false.");bytes64(v.providerSignature,"providerSignature");sha(v.requestPayloadSha256,"requestPayloadSha256");date(v.completedAt,"completedAt");return v;} -export function parseExecution(value){const v=record(value,"Persona Discord execution journal");exact(v.schemaVersion,executionSchema,"execution schema");for(const key of ["requestId","requestPayloadSha256","status","recordedAt"])text(v[key],key);if(!["permit_pending","running","completed","failed","unknown"].includes(v.status))throw new Error("execution status is invalid.");if(v.status==="permit_pending"){text(v.permitNonce,"permitNonce");date(v.permitIssuedAt,"permitIssuedAt");date(v.permitExpiresAt,"permitExpiresAt");if(Date.parse(v.permitExpiresAt)<=Date.parse(v.permitIssuedAt)||Date.parse(v.permitExpiresAt)-Date.parse(v.permitIssuedAt)>5000)throw new Error("permit intent lifetime is invalid");}return v;} -function record(v,label){if(!v||typeof v!=="object"||Array.isArray(v))throw new Error(`${label} must be an object.`);return v;} -function typed(v,members,label){if(Array.isArray(v)){if(v.length!==members.length)throw new Error(`${label} tuple has the wrong arity.`);return Object.fromEntries(members.map(member=>[member.memberName,member.typeName==="byte[]"&&Array.isArray(v[member.slot])?Uint8Array.from(v[member.slot]):v[member.slot]]));}return record(v,label);} -function text(v,label){bounded(v,label,256);} -function bounded(v,label,maxBytes){if(typeof v!=="string"||!v.trim()||Buffer.byteLength(v,"utf8")>maxBytes||/\p{Cc}/u.test(v))throw new Error(`${label} is empty, oversized, or contains control characters.`);} -function content(v){if(typeof v!=="string"||!v.trim()||Buffer.byteLength(v,"utf8")>1900||v.includes("\0"))throw new Error("content is empty, exceeds 1900 UTF-8 bytes, or contains NUL.");} -function optionalText(v,label){if(typeof v!=="string")throw new Error(`${label} must be a string.`);} -function exact(v,w,label){if(v!==w)throw new Error(`${label} must be ${w}.`);} -function bytes64(v,label){if(!(v instanceof Uint8Array)||v.length!==64)throw new Error(`${label} must be 64 bytes.`);} -function sha(v,label){if(!/^sha256-[0-9a-f]{64}$/.test(v))throw new Error(`${label} must be lowercase sha256.`);} -function date(v,label){const parsed=Date.parse(v);if(Number.isNaN(parsed))throw new Error(`${label} must be RFC3339.`);return parsed;} -function closed(v,keys,label){const allowed=new Set(keys);for(const key of Object.keys(v))if(!allowed.has(key))throw new Error(`${label} contains unknown field ${key}.`);} diff --git a/tools/persona-discord-delivery.mjs b/tools/persona-discord-delivery.mjs deleted file mode 100644 index 832bf66..0000000 --- a/tools/persona-discord-delivery.mjs +++ /dev/null @@ -1,52 +0,0 @@ -import {createHash,randomBytes} from "node:crypto"; -import {spawnSync} from "node:child_process"; -import {access,readFile,writeFile,mkdir} from "node:fs/promises"; -import {constants} from "node:fs"; -import {dirname,resolve} from "node:path"; -import {enrollDomainIdentity,loadDomainIdentity,signDomainPurpose,verifyDomainPurpose} from "./bifrost-signed-identity.mjs"; -import {buildPersonaDiscordDeliveryDefinitions,executionSchema,receiptIdentityDomain,receiptSchema,receiptSignatureDomain,receiptSigningPurpose,receiptSigningTuple,requestSignatureDomain,requestSigningPurpose,requestSigningTuple} from "./persona-discord-delivery-documents.mjs"; -import {requestPersonaDiscordPermit} from "./persona-discord-permit.mjs"; -import {permitRequestIdentityDomain,permitRequestPurpose,permitRequestSchema} from "./persona-discord-permit.mjs"; - -export async function enrollPersonaDiscordDeliveryIdentity({runtime,privateKeyPath,trustAnchorOut}){const identity=await enrollDomainIdentity(privateKeyPath,"Bifrost Persona Discord delivery identity",receiptIdentityDomain),anchor=["gamecult.service_trust_anchor.v1","bifrost-persona-discord-delivery:bifrost-discord-yggdrasil:v0","bifrost-persona-discord-delivery","bifrost-discord-yggdrasil",identity.identityId,new Uint8Array(identity.publicKey),"ed25519",receiptSigningPurpose,"bifrost.persona_discord_delivery_receipt.v0","root",Date.now(),null,false];await mkdir(dirname(trustAnchorOut),{recursive:true});await writeFile(trustAnchorOut,runtime.encode(anchor),{flag:"wx",mode:0o640});return {identityId:identity.identityId,privateKeyPath,trustAnchorOut};} -export async function exportPersonaDiscordPermitRequestAnchor({runtime,privateKeyPath,trustAnchorOut}){let identity;try{await access(privateKeyPath,constants.F_OK);identity=await loadDomainIdentity(privateKeyPath,"Bifrost Persona Discord permit requester identity",permitRequestIdentityDomain);}catch(error){if(!error||typeof error!=="object"||error.code!=="ENOENT")throw error;identity=await enrollDomainIdentity(privateKeyPath,"Bifrost Persona Discord permit requester identity",permitRequestIdentityDomain);}const anchor=["gamecult.service_trust_anchor.v1","bifrost-persona-discord-delivery-permit-request:bifrost-discord-yggdrasil:v0","bifrost-persona-discord-delivery","bifrost-discord-yggdrasil",identity.identityId,new Uint8Array(identity.publicKey),"ed25519",permitRequestPurpose,permitRequestSchema,"root",Date.now(),null,false];await mkdir(dirname(trustAnchorOut),{recursive:true});await writeFile(trustAnchorOut,runtime.encode(anchor),{flag:"wx",mode:0o640});return {identityId:identity.identityId,privateKeyPath,trustAnchorOut};} - -export async function processPersonaDiscordDeliveries(options){ - const {runtime,requestStorePath,receiptStorePath,executionStorePath,epiphanyTrustAnchorPath,privateKeyPath,targetAdmitted}=options,defs=buildPersonaDiscordDeliveryDefinitions(runtime.defineDocumentType); - requireSeparate([requestStorePath,receiptStorePath,executionStorePath]); - const requests=await open(runtime,defs.request,requestStorePath),receipts=await open(runtime,defs.receipt,receiptStorePath),executions=await open(runtime,defs.execution,executionStorePath),anchor=await readAnchor(runtime,epiphanyTrustAnchorPath,options.epiphanyRuntimeId),identity=await loadDomainIdentity(privateKeyPath,"Bifrost Persona Discord delivery identity",receiptIdentityDomain); - const results=[]; - for(const request of requests.cache.getAll(defs.request).sort((a,b)=>String(a.issuedAt).localeCompare(String(b.issuedAt)))){ - const unsigned=runtime.encode(requestSigningTuple(request)),digest=`sha256-${sha(unsigned)}`,existingReceipt=unwrap(receipts.cache.get(defs.receipt,request.requestId)),existingExecution=unwrap(executions.cache.get(defs.execution,request.requestId)); - if(existingReceipt){if(existingReceipt.requestPayloadSha256!==digest)throw new Error(`Persona Discord request ${request.requestId} conflicts with its terminal receipt.`);continue;} - if(existingExecution){if(existingExecution.requestPayloadSha256!==digest)throw new Error(`Persona Discord request ${request.requestId} conflicts with its execution journal.`);if(existingExecution.status==="running"){results.push(await terminal("unknown",request,digest,"delivery outcome was indeterminate after service interruption",{runtime,defs,receipts,executions,identity}));continue;}if(existingExecution.status!=="permit_pending")continue;} - let failure=""; - try{validateRequest(request,unsigned,anchor,targetAdmitted);}catch(error){failure=error instanceof Error?error.message:String(error);} - if(failure){results.push(await terminal("failed",request,digest,failure,{runtime,defs,receipts,executions,identity}));continue;} - if(existingExecution&&Date.parse(existingExecution.permitExpiresAt)now+30_000)throw new Error("request issuance is in the future");if(expires<=issued||expires120_000)throw new Error("request lifetime exceeds 120 seconds");if(!targetAdmitted(request))throw new Error("request target is not an exact configured Bifrost Persona binding");} -function post(request,options){const args=[options.bridgeCliPath,"discord-post","--cultmesh-command-id",request.requestId,"--channel-id",request.channelId,"--content",request.content,"--identity",request.personaAgentId,"--source-kind","epiphany_persona_speech","--source-id",request.effectDocumentId,"--authority-ref","epiphany.persona_discord_delivery_request.v0","--epiphany-lane-id","Persona","--epiphany-agent-identity",request.personaAgentId,"--receipt-store",options.crossingReceiptStorePath??resolve(dirname(options.executionStorePath),"persona-discord-crossing-receipts.cc")];if(request.replyToMessageId)args.push("--reply-to-message-id",request.replyToMessageId);const capability=process.env.HEIMDALL_CAPABILITY_REF||process.env.BIFROST_HEIMDALL_CAPABILITY_REF;if(capability)args.push("--heimdall-capability-ref",capability);const result=spawnSync(process.execPath,args,{cwd:options.bifrostRoot,encoding:"utf8",windowsHide:true,env:process.env});if(result.status!==0||result.error)throw new Error(`Bifrost Discord transport returned indeterminate outcome: ${result.error?.message||result.stderr||result.stdout}`);const value=JSON.parse(result.stdout);if(value.action!=="discord-post"||value.ok!==true||value.channelId!==request.channelId||!value.messageId||!value.crossingReceiptId||!value.url)throw new Error("Bifrost Discord transport did not return exact delivery evidence");return value;} -async function terminal(status,request,digest,detail,c){const now=new Date().toISOString(),posted=c.posted??{},receipt={schemaName:"bifrost.persona_discord_delivery_receipt",schemaVersion:receiptSchema,receiptId:request.requestId,requestId:request.requestId,requestPayloadSha256:digest,status,channelId:request.channelId,replyToMessageId:request.replyToMessageId,messageId:status==="completed"?posted.messageId:"",transport:status==="completed"?posted.transport:"",crossingReceiptId:status==="completed"?posted.crossingReceiptId:"",receiptUrl:status==="completed"?posted.url:"",completedAt:now,providerIdentityId:c.identity.identityId,privateStateExposed:false,providerSignature:new Uint8Array()};receipt.providerSignature=new Uint8Array(signDomainPurpose(c.identity,receiptSignatureDomain,receiptSigningPurpose,c.runtime.encode(receiptSigningTuple(receipt))));await c.receipts.cache.put(c.defs.receipt,receipt.receiptId,receipt);await c.executions.cache.put(c.defs.execution,request.requestId,{schemaName:"bifrost.persona_discord_delivery_execution",schemaVersion:executionSchema,requestId:request.requestId,requestPayloadSha256:digest,status,recordedAt:now,detail});return receipt;} -async function open(runtime,definition,path){await mkdir(dirname(path),{recursive:true});const cache=runtime.CultCache.builder().withDocumentType(definition).withGenericStore(new runtime.SingleFileMessagePackBackingStore(path)).build();await cache.pullAllBackingStores();return {cache};} -async function accessStoreOrWritableParent(path,mode){try{await access(path,mode);}catch(error){if(!error||typeof error!=="object"||error.code!=="ENOENT")throw error;await access(dirname(path),constants.W_OK);}} -async function readAnchor(runtime,path,expectedRuntimeId){if(typeof expectedRuntimeId!=="string"||!expectedRuntimeId)throw new Error("Epiphany runtime identity is required for Persona mouth trust");const value=runtime.decode(await readFile(path));if(!Array.isArray(value)||value.length!==13)throw new Error("Epiphany Persona mouth trust anchor must be the exact service-contract tuple");const [schemaVersion,trustAnchorId,serviceId,runtimeId,identityId,rawPublicKey,algorithm,purpose,signedSchema,bindingAuthority,boundAt,expiresAt,privateStateExposed]=value,publicKey=rawPublicKey instanceof Uint8Array?rawPublicKey:Array.isArray(rawPublicKey)?Uint8Array.from(rawPublicKey):null,now=Date.now();if(schemaVersion!=="gamecult.service_trust_anchor.v1"||typeof trustAnchorId!=="string"||serviceId!=="epiphany-persona-mouth"||runtimeId!==expectedRuntimeId||typeof identityId!=="string"||!publicKey||publicKey.length!==32||algorithm!=="ed25519"||purpose!==requestSigningPurpose||signedSchema!=="epiphany.persona_discord_delivery_request.v0"||bindingAuthority!=="root"||!Number.isSafeInteger(boundAt)||boundAt<=0||boundAt>now||(expiresAt!==null&&(!Number.isSafeInteger(expiresAt)||expiresAt<=boundAt||expiresAt<=now))||privateStateExposed!==false)throw new Error("Epiphany Persona mouth service trust anchor is invalid");return {identityId,publicKey};} -function requireSeparate(paths){const keys=paths.map(p=>resolve(p).toLowerCase());if(new Set(keys).size!==keys.length)throw new Error("Persona Discord crossing and private execution stores must be distinct.");} -function sha(value){return createHash("sha256").update(value).digest("hex");} -function unwrap(value){return Array.isArray(value)?value[0]:value?.value??value;} diff --git a/tools/persona-discord-permit.mjs b/tools/persona-discord-permit.mjs deleted file mode 100644 index 9d14e77..0000000 --- a/tools/persona-discord-permit.mjs +++ /dev/null @@ -1,39 +0,0 @@ -import {createHash,randomBytes} from "node:crypto"; -import {createSocket} from "node:dgram"; -import {createRequire} from "node:module"; -import {existsSync} from "node:fs"; -import {readFile} from "node:fs/promises"; -import {resolve} from "node:path"; -import {loadDomainIdentity,signDomainPurpose,verifyDomainPurpose} from "./bifrost-signed-identity.mjs"; - -export const permitRequestSchema="bifrost.persona_discord_delivery_permit_request.v0"; -export const permitSchema="epiphany.persona_discord_delivery_permit.v0"; -export const permitRequestPurpose="bifrost.persona-discord.delivery-permit-request.v0"; -export const permitPurpose="epiphany.persona-discord.delivery-permit.v0"; -export const permitRequestIdentityDomain="bifrost.persona-discord-delivery-permit-request.identity.v0\0"; -export const permitRequestSignatureDomain="bifrost.persona-discord-delivery-permit-request.signature.v0\0"; -export const permitIdentityDomain="epiphany.persona-discord-delivery-permit.identity.v0\0"; -export const permitSignatureDomain="epiphany.persona-discord-delivery-permit.signature.v0\0"; -const connectionId=0xe91f0002; - -export const permitRequestTuple=v=>[v.schemaVersion,v.requestId,v.requestPayloadSha256,v.targetRuntimeId,v.nonce,v.issuedAt,v.expiresAt,v.requesterIdentityId,v.privateStateExposed]; -export const permitTuple=v=>[v.schemaVersion,v.permitId,v.requestId,v.requestPayloadSha256,v.targetRuntimeId,v.nonce,v.requesterIdentityId,v.brakeStateDocumentId,v.brakeStateDocumentSha256,v.brakeObservedAt,v.issuedAt,v.expiresAt,v.providerIdentityId,v.privateStateExposed]; -export const permitRequestWireTuple=v=>[...permitRequestTuple(v),v.requesterSignature]; -export function parsePermitWireTuple(v){if(!Array.isArray(v)||v.length!==15)throw new Error("Epiphany permit must be an exact Rust wire tuple");const [schemaVersion,permitId,requestId,requestPayloadSha256,targetRuntimeId,nonce,requesterIdentityId,brakeStateDocumentId,brakeStateDocumentSha256,brakeObservedAt,issuedAt,expiresAt,providerIdentityId,privateStateExposed,providerSignature]=v;return {schemaVersion,permitId,requestId,requestPayloadSha256,targetRuntimeId,nonce,requesterIdentityId,brakeStateDocumentId,brakeStateDocumentSha256,brakeObservedAt,issuedAt,expiresAt,providerIdentityId,privateStateExposed,providerSignature};} - -export async function requestPersonaDiscordPermit({endpoint,requestId,requestPayloadSha256,targetRuntimeId,nonce,issuedAt,expiresAt,privateKeyPath,epiphanyPermitAnchorPath,timeoutMs=5000,runtime:provided,socketFactory=createSocket,connectionFactory}){ - const runtime=provided??loadRuntime(),identity=await loadDomainIdentity(privateKeyPath,"Bifrost Persona Discord permit requester identity",permitRequestIdentityDomain),anchor=await readAnchor(runtime,epiphanyPermitAnchorPath,targetRuntimeId),now=Date.now(); - const request={schemaVersion:permitRequestSchema,requestId,requestPayloadSha256,targetRuntimeId,nonce:nonce??randomBytes(16).toString("hex"),issuedAt:issuedAt??new Date(now).toISOString(),expiresAt:expiresAt??new Date(now+5000).toISOString(),requesterIdentityId:identity.identityId,privateStateExposed:false,requesterSignature:new Uint8Array()}; - request.requesterSignature=new Uint8Array(signDomainPurpose(identity,permitRequestSignatureDomain,permitRequestPurpose,runtime.msgpack.encode(permitRequestTuple(request)))); - const {host,port}=parseEndpoint(endpoint),socket=socketFactory("udp4");await bind(socket);const connection=(connectionFactory??(o=>new runtime.cultnet.CultNetRudpSocketTransportConnection(o)))({runtimeId:"bifrost-discord-yggdrasil",socket,mode:"client",remoteHost:host,remotePort:port,connectionId,resendDelayMs:50,resendPollMs:10}); - try{connection.connect();await connected(connection,timeoutMs);const response=wait(connection,runtime.msgpack,timeoutMs,request);const message={schemaVersion:"cultnet.document_put_raw.v0",messageId:`permit-request-${requestId}`,document:{schemaId:permitRequestSchema,recordKey:requestId,storedAt:request.issuedAt,payloadEncoding:"messagepack",payload:runtime.msgpack.encode(permitRequestWireTuple(request)),sourceRuntimeId:"bifrost-discord-yggdrasil",sourceAgentId:identity.identityId,sourceRole:"bifrost-persona-discord-delivery",tags:["cultnet.transport.rudp.v0"]}};connection.send("schema",runtime.msgpack.encode(runtime.cultnet.encodeCultNetMessageForWire(message,"cultnet.schema.v0")));const permit=await response;validatePermit(permit,request,anchor,runtime.msgpack.encode);return permit;}finally{connection.close();} -} - -export function validatePermit(value,request,anchor,encode){const v=Array.isArray(value)?parsePermitWireTuple(value):value,providerSignature=v?.providerSignature instanceof Uint8Array?v.providerSignature:Array.isArray(v?.providerSignature)?Uint8Array.from(v.providerSignature):null;if(!v||v.schemaVersion!==permitSchema||v.requestId!==request.requestId||v.requestPayloadSha256!==request.requestPayloadSha256||v.targetRuntimeId!==request.targetRuntimeId||v.nonce!==request.nonce||v.requesterIdentityId!==request.requesterIdentityId||v.providerIdentityId!==anchor.identityId||v.privateStateExposed!==false||!/^sha256-[0-9a-f]{64}$/.test(v.brakeStateDocumentSha256)||!providerSignature||providerSignature.length!==64)throw new Error("Epiphany permit substituted delivery authority");const issued=Date.parse(v.issuedAt),expires=Date.parse(v.expiresAt),observed=Date.parse(v.brakeObservedAt),now=Date.now();if(!Number.isFinite(issued)||!Number.isFinite(expires)||!Number.isFinite(observed)||expires<=issued||expires-issued>5000||expiresissued||issued>now+1000)throw new Error("Epiphany permit is stale or has an invalid lifetime");if(!verifyDomainPurpose(anchor.publicKey,permitSignatureDomain,permitPurpose,encode(permitTuple(v)),providerSignature))throw new Error("Epiphany permit signature is invalid");return {...v,providerSignature};} -export function parseEpiphanyPermitAnchorTuple(a,expectedRuntimeId,now=Date.now()){if(!Array.isArray(a)||a.length!==13)throw new Error("Epiphany permit anchor must be an exact service anchor");const [schemaVersion,,serviceId,runtimeId,identityId,rawPublicKey,algorithm,purpose,signedSchema,authority,boundAt,expiresAt,privateStateExposed]=a,publicKey=rawPublicKey instanceof Uint8Array?rawPublicKey:Array.isArray(rawPublicKey)?Uint8Array.from(rawPublicKey):null;if(schemaVersion!=="gamecult.service_trust_anchor.v1"||serviceId!=="epiphany-persona-discord-permit"||runtimeId!==expectedRuntimeId||!publicKey||publicKey.length!==32||algorithm!=="ed25519"||purpose!==permitPurpose||signedSchema!==permitSchema||authority!=="root"||boundAt>now||(expiresAt!==null&&expiresAt<=now)||privateStateExposed!==false)throw new Error("Epiphany permit anchor is invalid");return {identityId,publicKey};} -async function readAnchor(runtime,path,expectedRuntimeId){return parseEpiphanyPermitAnchorTuple(runtime.msgpack.decode(await readFile(path)),expectedRuntimeId);} -function wait(c,m,t,r){return new Promise((ok,no)=>{const done=(f,v)=>{clearTimeout(timer);c.off?.("frame",frame);c.off?.("error",error);f(v)},error=e=>done(no,e),frame=x=>{try{if(x.channelId!=="schema")throw new Error("permit arrived on foreign channel");const w=m.decode(x.payload),d=w?.document,p=parsePermitWireTuple(m.decode(d?.payload));if(w?.schemaVersion!=="cultnet.document_put_raw.v0"||d?.schemaId!==permitSchema||d?.recordKey!==`permit:${r.requestPayloadSha256}`||d?.payloadEncoding!=="messagepack"||d?.sourceRuntimeId!==r.targetRuntimeId||d?.sourceRole!=="epiphany-persona-discord-permit"||d?.sourceAgentId!==p.providerIdentityId)throw new Error("permit envelope substituted authority");done(ok,p)}catch(e){done(no,e)}};const timer=setTimeout(()=>done(no,new Error("Timed out waiting for Epiphany Persona Discord permit")),t);c.on("frame",frame);c.on("error",error)})} -async function connected(c,t){const s=Date.now();while(!c.connected){if(Date.now()-s>=t)throw new Error("Timed out establishing permit RUDP session");await new Promise(r=>setTimeout(r,5));}} -function bind(s){return new Promise((ok,no)=>{s.once("error",no);s.once("listening",ok);s.bind(0,"0.0.0.0")})} -function parseEndpoint(v){const u=new URL(v);if(u.protocol!=="rudp:"||!u.hostname||!u.port)throw new Error("permit endpoint must be rudp://host:port");return {host:u.hostname,port:Number(u.port)}} -function loadRuntime(){const root=resolve(import.meta.dirname,".."),cult=resolve(process.env.VOIDBOT_CULTLIB_ROOT||resolve(root,"..","CultLib")),pkg=resolve(cult,"packages","cultnet-ts","package.json");if(!existsSync(pkg))throw new Error("CultNet runtime unavailable");const req=createRequire(pkg);return {cultnet:req("cultnet-ts"),msgpack:req("@msgpack/msgpack")};} diff --git a/tools/persona-discord-rudp.mjs b/tools/persona-discord-rudp.mjs deleted file mode 100644 index d3ebfe6..0000000 --- a/tools/persona-discord-rudp.mjs +++ /dev/null @@ -1,31 +0,0 @@ -import {createSocket} from "node:dgram"; -import {parseRequest,receiptDocumentTuple} from "./persona-discord-delivery-documents.mjs"; - -export const personaDiscordDeliveryConnectionId=0xe91f0003; - -export async function startPersonaDiscordDeliveryRudpServer({runtime,endpoint,runtimeId="bifrost-discord-yggdrasil",onRequest,socketFactory=createSocket,connectionFactory}){ - const {host,port}=parseEndpoint(endpoint),socket=socketFactory("udp4"); - await bind(socket,host,port); - const connection=(connectionFactory??(options=>new runtime.cultnet.CultNetRudpSocketTransportConnection(options)))({runtimeId,socket,mode:"server",connectionId:personaDiscordDeliveryConnectionId,initialSequence:100,resendDelayMs:50,resendPollMs:10}); - let busy=false;const diagnostics={datagrams:0,frames:0,responses:0,errors:[]};socket.on("message",()=>diagnostics.datagrams++); - const frame=async value=>{ - diagnostics.frames++; - if(busy)return; - busy=true; - try{ - if(value.channelId!=="schema")throw new Error("Persona delivery request arrived on a foreign channel"); - const message=runtime.cultnet.parseCultNetMessage(runtime.msgpack.decode(value.payload),"cultnet.schema.v0"),document=message.document; - if(message.schemaVersion!=="cultnet.document_put_raw.v0"||document.schemaId!=="epiphany.persona_discord_delivery_request.v0"||document.payloadEncoding!=="messagepack"||document.sourceRole!=="epiphany-persona-mouth"||document.sourceRuntimeId!=="epiphany-starfire"||!Array.isArray(document.tags)||document.tags.length!==1||document.tags[0]!=="cultnet.transport.rudp.v0")throw new Error("Persona delivery request envelope substituted authority"); - const request=parseRequest(runtime.msgpack.decode(document.payload)); - if(document.recordKey!==request?.requestId||document.sourceAgentId!==request?.signerIdentityId||request?.targetRuntimeId!=="epiphany-starfire")throw new Error("Persona delivery request envelope is not bound to its signed payload"); - const receipt=await onRequest(request); - const response={schemaVersion:"cultnet.document_put_raw.v0",messageId:`persona-delivery-receipt-${receipt.requestId}`,document:{schemaId:"bifrost.persona_discord_delivery_receipt.v0",recordKey:receipt.requestId,storedAt:receipt.completedAt,payloadEncoding:"messagepack",payload:runtime.msgpack.encode(receiptDocumentTuple(receipt)),sourceRuntimeId:runtimeId,sourceAgentId:receipt.providerIdentityId,sourceRole:"bifrost-persona-discord-delivery",tags:["cultnet.transport.rudp.v0"]}}; - connection.send("schema",runtime.msgpack.encode(runtime.cultnet.encodeCultNetMessageForWire(response,"cultnet.schema.v0")));diagnostics.responses++; - }catch(error){const message=error instanceof Error?error.message:String(error);diagnostics.errors.push(message);console.error(`Bifrost Persona delivery RUDP rejected frame: ${message}`);}finally{busy=false;} - }; - connection.on("frame",frame); - return {address:socket.address(),diagnostics,close(){connection.off?.("frame",frame);connection.close();}}; -} - -function parseEndpoint(value){const url=new URL(value);if(url.protocol!=="rudp:"||!url.hostname||!url.port)throw new Error("Persona delivery listener must be rudp://host:port");return {host:url.hostname,port:Number(url.port)};} -function bind(socket,host,port){return new Promise((resolve,reject)=>{socket.once("error",reject);socket.once("listening",resolve);socket.bind(port,host);});} diff --git a/tools/persona-feedback.mjs b/tools/persona-feedback.mjs index b586a58..ea8e185 100644 --- a/tools/persona-feedback.mjs +++ b/tools/persona-feedback.mjs @@ -8,18 +8,12 @@ import { buildFeedbackDefinitions, actorLinkSchema, feedbackBindingSchema, feedb import { defaultFeedbackIdentityPath, enrollBifrostFeedbackIdentity, loadBifrostFeedbackIdentity, providerHealthTrustAnchor, signFeedbackAdmission, trustAnchor } from "./bifrost-feedback-identity.mjs"; import { createPersonaFeedbackHealthPublisher, publishPersonaFeedbackHealth } from "./persona-feedback-idunn-health.mjs"; import { discordGatewayReady, serveDiscordPersonaIngress } from "./bifrost-discord-persona-ingress.mjs"; -import { enrollPersonaDiscordDeliveryIdentity, exportPersonaDiscordPermitRequestAnchor, inspectPersonaDiscordDeliveryReadiness, processPersonaDiscordDeliveries } from "./persona-discord-delivery.mjs"; -import { startPersonaDiscordDeliveryRudpServer } from "./persona-discord-rudp.mjs"; -import { buildPersonaDiscordDeliveryDefinitions, parseRequest as parsePersonaDiscordDeliveryRequest } from "./persona-discord-delivery-documents.mjs"; const root=resolve(import.meta.dirname,".."),projects=resolve(root,".."),cult=resolve(process.env.VOIDBOT_CULTLIB_ROOT||resolve(projects,"CultLib")); const [command,...rawArgs]=process.argv.slice(2),options=parseArgs(rawArgs); const privateStorePath=resolve(options.store||(process.platform==="win32"?resolve(root,".bifrost","provider-store.cc"):"/var/lib/gamecult/bifrost/persona-feedback/provider.cc")); const observationStorePath=resolve(options["observation-store"]||(process.platform==="win32"?resolve(root,".bifrost","persona-conversation-observations.cc"):"/srv/bifrost/persona-feedback/observations.cc")); const deliveryStorePath=resolve(options["delivery-store"]||(process.platform==="win32"?resolve(root,".bifrost","epiphany-persona-feedback.cc"):"/srv/bifrost/persona-feedback/deliveries.cc")); -const personaDeliveryRequestStorePath=resolve(options["persona-delivery-request-store"]||(process.platform==="win32"?resolve(root,".bifrost","persona-discord-requests.cc"):"/srv/bifrost/persona-discord-crossing/requests.cc")); -const personaDeliveryReceiptStorePath=resolve(options["persona-delivery-receipt-store"]||(process.platform==="win32"?resolve(root,".bifrost","persona-discord-receipts.cc"):"/srv/bifrost/persona-discord-crossing/receipts.cc")); -const personaDeliveryExecutionStorePath=resolve(options["persona-delivery-execution-store"]||(process.platform==="win32"?resolve(root,".bifrost","private","persona-discord-executions.cc"):"/var/lib/gamecult/bifrost/persona-feedback/persona-discord-executions.cc")); const runtime=loadRuntime(),defs=buildFeedbackDefinitions(runtime.defineDocumentType); try{await main();}catch(error){console.error(error instanceof Error?error.message:String(error));process.exitCode=1;} @@ -30,14 +24,11 @@ async function main(){ if(command==="process")return processEvent(requireOption("event-id")); if(command==="serve")return serve(); if(command==="serve-discord")return serveDiscord(); - if(command==="serve-persona-delivery")return servePersonaDelivery(); if(command==="export")return exportSnapshot(); if(command==="enroll-identity")return enrollIdentity(); - if(command==="enroll-persona-delivery-identity")return enrollPersonaDeliveryIdentity(); - if(command==="export-persona-permit-request-anchor")return exportPersonaPermitRequestAnchor(); if(command==="export-provider-health-anchor")return exportProviderHealthAnchor(); if(command==="status")return status(); - throw new Error("Usage: persona-feedback.mjs bind-target|link-actor|process|serve|serve-discord|serve-persona-delivery|status|export|enroll-identity|enroll-persona-delivery-identity|export-provider-health-anchor"); + throw new Error("Usage: persona-feedback.mjs bind-target|link-actor|process|serve|serve-discord|status|export|enroll-identity|export-provider-health-anchor"); } async function privateNode(){return runtime.CultMesh.createNode(privateStorePath,{documents:[defs.binding,defs.actorLink,defs.receipt,defs.delivery,defs.pendingFailure]});} async function observationNode(){return runtime.CultMesh.createNode(observationStorePath,{documents:[defs.event]});} @@ -77,19 +68,14 @@ async function serveDiscord(){ const publishHealth=async()=>{const value=await readiness(),gatewayReady=discordGatewayReady(gatewayState),active=value.status==="ready"&&gatewayReady;await publishPersonaFeedbackHealth({endpoint:healthEndpoint,daemonId:"yggdrasil-bifrost-persona-feedback",healthContract:"bifrost.cultnet-rudp-persona-feedback-health",state:active?"active":"degraded",detail:active?"ready":"readiness-degraded",publisher:healthPublisher,publishUnsignedDiagnostic:true});return {...value,gatewayReady,gatewayState};}; const initial=await readiness();if(!initial.ingressReady)throw new Error(`Bifrost Discord Persona ingress is not ready: ${initial.ingressReasons.join("; ")}`); const recovered=await drainPendingObservations();if(recovered.some(value=>!value.ok))console.error(`Bifrost Persona ingress startup recovery retained ${recovered.filter(value=>!value.ok).length} failed observation(s) for reconnect retry.`); - await drainPersonaDeliveries().catch(error=>console.error(`Bifrost Persona delivery is staged but not yet ready: ${error instanceof Error?error.message:String(error)}`)); await publishHealth(); const healthTimer=setInterval(()=>publishHealth().catch(error=>console.error(`Bifrost Persona-feedback health publication failed: ${error instanceof Error?error.message:String(error)}`)),10000); let observationBusy=false;const observationTimer=setInterval(async()=>{if(observationBusy)return;observationBusy=true;try{const retried=await drainPendingObservations();if(retried.some(value=>!value.ok))console.error(`Bifrost Persona ingress retained ${retried.filter(value=>!value.ok).length} failed pending observation(s).`);}catch(error){console.error(`Bifrost Persona observation reconciliation failed: ${error instanceof Error?error.message:String(error)}`);}finally{observationBusy=false;}},positiveInteger(options["observation-retry-interval-ms"]||"5000","observation-retry-interval-ms")); - let deliveryBusy=false;const deliveryTimer=setInterval(async()=>{if(deliveryBusy)return;deliveryBusy=true;try{await drainPersonaDeliveries();}catch(error){console.error(`Bifrost Persona delivery poll failed: ${error instanceof Error?error.message:String(error)}`);}finally{deliveryBusy=false;}},positiveInteger(options["persona-delivery-interval-ms"]||"500","persona-delivery-interval-ms")); - try{await serveDiscordPersonaIngress({token,runtimeId:producerRuntimeId,onConversation:conversation=>admitDiscordConversation(conversation,producerRuntimeId),onReady:async state=>{const drained=await drainPendingObservations();print({schemaVersion:"bifrost.discord.persona_ingress_status.v0",status:drained.every(value=>value.ok)?"ready":"degraded",runtimeId:state.runtimeId,botUserId:state.botUserId,recoveredObservationCount:drained.length,privateStateExposed:false});},onState:state=>{gatewayState=state;},onDisconnect:error=>console.error(`Bifrost Discord Persona ingress reconnecting: ${error instanceof Error?error.message:String(error)}`)});}finally{gatewayState={connected:false,sessionId:"",lastHeartbeatAckAtMillis:0,phase:"stopped"};await publishHealth().catch(()=>{});clearInterval(healthTimer);clearInterval(observationTimer);clearInterval(deliveryTimer);} + try{await serveDiscordPersonaIngress({token,runtimeId:producerRuntimeId,onConversation:conversation=>admitDiscordConversation(conversation,producerRuntimeId),onReady:async state=>{const drained=await drainPendingObservations();print({schemaVersion:"bifrost.discord.persona_ingress_status.v0",status:drained.every(value=>value.ok)?"ready":"degraded",runtimeId:state.runtimeId,botUserId:state.botUserId,recoveredObservationCount:drained.length,privateStateExposed:false});},onState:state=>{gatewayState=state;},onDisconnect:error=>console.error(`Bifrost Discord Persona ingress reconnecting: ${error instanceof Error?error.message:String(error)}`)});}finally{gatewayState={connected:false,sessionId:"",lastHeartbeatAckAtMillis:0,phase:"stopped"};await publishHealth().catch(()=>{});clearInterval(healthTimer);clearInterval(observationTimer);} } -async function servePersonaDelivery(){const readinessState=await readiness();if(!readinessState.personaDelivery.ready)throw new Error(`Bifrost Persona delivery is not ready: ${readinessState.personaDelivery.reasons.join("; ")}`);const server=await startPersonaDiscordDeliveryRudpServer({runtime,endpoint:requireOption("persona-delivery-rudp-listen"),onRequest:acceptPersonaDeliveryRequest});print({schemaVersion:"bifrost.persona_discord_delivery_listener.v0",status:"ready",runtimeId:"bifrost-discord-yggdrasil",host:server.address.address,port:server.address.port,requestSignerIdentityId:readinessState.personaDelivery.requestSignerIdentityId,receiptSignerIdentityId:readinessState.personaDelivery.receiptSignerIdentityId,privateStateExposed:false});await new Promise(()=>{});} -async function drainPersonaDeliveries(){const n=await privateNode();await n.cache?.pullAllBackingStores?.();const bindings=n.cache.getAll(defs.binding),epiphanyRuntimeId=requireOption("epiphany-runtime-id");return processPersonaDiscordDeliveries({runtime,permitRuntime:{cultnet:runtime.cultnet,msgpack:runtime.msgpack},requestStorePath:personaDeliveryRequestStorePath,receiptStorePath:personaDeliveryReceiptStorePath,executionStorePath:personaDeliveryExecutionStorePath,epiphanyTrustAnchorPath:resolve(requireOption("epiphany-persona-mouth-trust-anchor")),epiphanyRuntimeId,epiphanyPermitAnchorPath:resolve(requireOption("epiphany-persona-permit-trust-anchor")),epiphanyPermitEndpoint:requireOption("epiphany-persona-permit-rudp"),privateKeyPath:resolve(requireOption("persona-delivery-private-key")),permitRequesterPrivateKeyPath:resolve(requireOption("persona-permit-request-private-key")),bridgeCliPath:resolve(root,"tools","bifrost-bridge.mjs"),bifrostRoot:root,targetAdmitted:request=>request.targetRuntimeId===epiphanyRuntimeId&&request.personaAgentId==="epiphany.Persona"&&bindings.some(binding=>binding.targetRuntimeId===request.targetRuntimeId&&binding.targetPersonaId==="epiphany"&&binding.channelId===request.channelId)});} -async function acceptPersonaDeliveryRequest(request){const deliveryDefs=buildPersonaDiscordDeliveryDefinitions(runtime.defineDocumentType),requests=runtime.CultCache.builder().withDocumentType(deliveryDefs.request).withGenericStore(new runtime.SingleFileMessagePackBackingStore(personaDeliveryRequestStorePath)).build();await requests.pullAllBackingStores();const parsed=parsePersonaDiscordDeliveryRequest(request),existing=unwrap(requests.get(deliveryDefs.request,parsed.requestId));if(existing&&stable(existing)!==stable(parsed))throw new Error(`Persona delivery request ${parsed.requestId} conflicts with durable endpoint state.`);if(!existing)await requests.put(deliveryDefs.request,parsed.requestId,parsed);await drainPersonaDeliveries();const receipts=runtime.CultCache.builder().withDocumentType(deliveryDefs.receipt).withGenericStore(new runtime.SingleFileMessagePackBackingStore(personaDeliveryReceiptStorePath)).build();await receipts.pullAllBackingStores();const receipt=unwrap(receipts.get(deliveryDefs.receipt,parsed.requestId));if(!receipt)throw new Error(`Persona delivery request ${parsed.requestId} has no terminal Bifrost receipt yet.`);return receipt;} async function admitDiscordConversation(conversation,producerRuntimeId){ const privateNodeValue=await privateNode();await privateNodeValue.cache?.pullAllBackingStores?.(); @@ -104,11 +90,9 @@ async function admitDiscordConversation(conversation,producerRuntimeId){ } async function enrollIdentity(){const keyPath=resolve(options["private-key"]||defaultFeedbackIdentityPath()),identity=await enrollBifrostFeedbackIdentity(keyPath),createdAt=new Date().toISOString(),anchor=trustAnchor(identity,createdAt),out=resolve(requireOption("trust-anchor-out")),providerOut=options["provider-health-trust-anchor-out"]?resolve(options["provider-health-trust-anchor-out"]):null;await mkdir(dirname(out),{recursive:true});await writeFile(out,runtime.encode(anchorTuple(anchor)));if(providerOut){if(providerOut===out)throw new Error("Feedback and provider-health trust anchors must use distinct paths.");const providerAnchor=providerHealthTrustAnchor(identity,createdAt);await mkdir(dirname(providerOut),{recursive:true});await writeFile(providerOut,runtime.encode(anchorTuple(providerAnchor)));}print({privateKey:keyPath,trustAnchor:out,identityId:identity.identityId,providerHealthTrustAnchor:providerOut??""});} -async function enrollPersonaDeliveryIdentity(){print(await enrollPersonaDiscordDeliveryIdentity({runtime,privateKeyPath:resolve(requireOption("private-key")),trustAnchorOut:resolve(requireOption("trust-anchor-out"))}));} -async function exportPersonaPermitRequestAnchor(){print(await exportPersonaDiscordPermitRequestAnchor({runtime,privateKeyPath:resolve(requireOption("permit-request-private-key")),trustAnchorOut:resolve(requireOption("trust-anchor-out"))}));} async function exportProviderHealthAnchor(){const keyPath=resolve(options["private-key"]||defaultFeedbackIdentityPath()),identity=await loadBifrostFeedbackIdentity(keyPath),out=resolve(requireOption("trust-anchor-out")),anchor=providerHealthTrustAnchor(identity,new Date().toISOString());await mkdir(dirname(out),{recursive:true});await writeFile(out,runtime.encode(anchorTuple(anchor)),{flag:"wx",mode:0o640});print({privateKey:keyPath,trustAnchor:out,identityId:anchor.identityId});} async function exportSnapshot(){const n=await privateNode();await n.cache?.pullAllBackingStores?.();const out=resolve(requireOption("out")),scratch=`${out}.scratch.cc`,cache=runtime.CultCache.builder().withDocumentType(defs.delivery).withDocumentType(defs.receipt).withGenericStore(new runtime.SingleFileMessagePackBackingStore(scratch)).build();for(const value of n.cache.getAll(defs.delivery))await cache.put(defs.delivery,value.admissionId,value);for(const value of n.cache.getAll(defs.receipt))await cache.put(defs.receipt,value.receiptId,value);const registry=new runtime.CultNetDocumentRegistry([runtime.defineCultNetDocumentBinding({definition:defs.delivery,payloadSchemaVersion:feedbackDeliverySchema}),runtime.defineCultNetDocumentBinding({definition:defs.receipt,payloadSchemaVersion:feedbackReceiptSchema})]),message=registry.createRawSnapshotResponse(cache,options["message-id"]||`persona-feedback-${randomUUID()}`);await mkdir(dirname(out),{recursive:true});await writeFile(out,runtime.encode(message));await rm(scratch,{force:true});print({schemaVersion:message.schemaVersion,messageId:message.messageId,documentCount:message.documents.length,out});} -async function readiness(){const ingressReasons=[];let identity,bindingCount=0,pendingFailedCount=0;try{identity=await loadBifrostFeedbackIdentity(resolve(options["private-key"]||defaultFeedbackIdentityPath()));}catch(error){ingressReasons.push(`identity:${error instanceof Error?error.message:String(error)}`);}let bindings=[];try{const info=await stat(privateStorePath);if(!info.isFile())throw new Error("private provider store is not a regular file");const cache=runtime.CultCache.builder().withDocumentType(defs.binding).withDocumentType(defs.actorLink).withDocumentType(defs.receipt).withDocumentType(defs.delivery).withDocumentType(defs.pendingFailure).withGenericStore(new runtime.SingleFileMessagePackBackingStore(privateStorePath)).build();await cache.pullAllBackingStores();bindings=cache.getAll(defs.binding);bindingCount=bindings.length;const admitted=new Set(cache.getAll(defs.receipt).map(value=>value.eventId));pendingFailedCount=cache.getAll(defs.pendingFailure).filter(value=>!admitted.has(value.eventId)).length;if(pendingFailedCount>0)ingressReasons.push(`pendingFailures:${pendingFailedCount}`);if(bindingCount===0)ingressReasons.push("bindings:none");}catch(error){ingressReasons.push(`privateStore:${error instanceof Error?error.message:String(error)}`);}for(const binding of bindings){try{await assertStoreSeparation(binding.deliveryStorePath);try{const info=await stat(binding.deliveryStorePath);if(!info.isFile())throw new Error("delivery path is not a regular file");await access(binding.deliveryStorePath,constants.W_OK);}catch(error){if(!error||typeof error!=="object"||error.code!=="ENOENT")throw error;await access(dirname(binding.deliveryStorePath),constants.W_OK);}}catch(error){ingressReasons.push(`binding:${binding.bindingId}:${error instanceof Error?error.message:String(error)}`);}}const personaConfigured=command==="serve-discord"||command==="serve-persona-delivery"||Boolean(options["epiphany-persona-mouth-trust-anchor"]);let personaDelivery={ready:false,requestSignerIdentityId:"",receiptSignerIdentityId:"",permitRequesterIdentityId:"",reasons:["not configured"]};if(personaConfigured){personaDelivery=await inspectPersonaDiscordDeliveryReadiness({runtime,requestStorePath:personaDeliveryRequestStorePath,receiptStorePath:personaDeliveryReceiptStorePath,executionStorePath:personaDeliveryExecutionStorePath,epiphanyTrustAnchorPath:resolve(requireOption("epiphany-persona-mouth-trust-anchor")),epiphanyRuntimeId:requireOption("epiphany-runtime-id"),epiphanyPermitAnchorPath:resolve(requireOption("epiphany-persona-permit-trust-anchor")),epiphanyPermitEndpoint:requireOption("epiphany-persona-permit-rudp"),privateKeyPath:resolve(requireOption("persona-delivery-private-key")),permitRequesterPrivateKeyPath:resolve(requireOption("persona-permit-request-private-key"))});}const reasons=[...ingressReasons,...(personaConfigured?personaDelivery.reasons.map(reason=>`personaDelivery:${reason}`):[])],ingressReady=ingressReasons.length===0;return {schemaVersion:"bifrost.persona_feedback.readiness.v0",providerIdentityId:identity?.identityId??"",bindingCount,pendingFailedCount,ingressReady,ingressReasons,personaDelivery,status:ingressReady&&(!personaConfigured||personaDelivery.ready)?"ready":"degraded",reasons,privateStateExposed:false};} +async function readiness(){const ingressReasons=[];let identity,bindingCount=0,pendingFailedCount=0;try{identity=await loadBifrostFeedbackIdentity(resolve(options["private-key"]||defaultFeedbackIdentityPath()));}catch(error){ingressReasons.push(`identity:${error instanceof Error?error.message:String(error)}`);}let bindings=[];try{const info=await stat(privateStorePath);if(!info.isFile())throw new Error("private provider store is not a regular file");const cache=runtime.CultCache.builder().withDocumentType(defs.binding).withDocumentType(defs.actorLink).withDocumentType(defs.receipt).withDocumentType(defs.delivery).withDocumentType(defs.pendingFailure).withGenericStore(new runtime.SingleFileMessagePackBackingStore(privateStorePath)).build();await cache.pullAllBackingStores();bindings=cache.getAll(defs.binding);bindingCount=bindings.length;const admitted=new Set(cache.getAll(defs.receipt).map(value=>value.eventId));pendingFailedCount=cache.getAll(defs.pendingFailure).filter(value=>!admitted.has(value.eventId)).length;if(pendingFailedCount>0)ingressReasons.push(`pendingFailures:${pendingFailedCount}`);if(bindingCount===0)ingressReasons.push("bindings:none");}catch(error){ingressReasons.push(`privateStore:${error instanceof Error?error.message:String(error)}`);}for(const binding of bindings){try{await assertStoreSeparation(binding.deliveryStorePath);try{const info=await stat(binding.deliveryStorePath);if(!info.isFile())throw new Error("delivery path is not a regular file");await access(binding.deliveryStorePath,constants.W_OK);}catch(error){if(!error||typeof error!=="object"||error.code!=="ENOENT")throw error;await access(dirname(binding.deliveryStorePath),constants.W_OK);}}catch(error){ingressReasons.push(`binding:${binding.bindingId}:${error instanceof Error?error.message:String(error)}`);}}const reasons=[...ingressReasons],ingressReady=ingressReasons.length===0;return {schemaVersion:"bifrost.persona_feedback.readiness.v0",providerIdentityId:identity?.identityId??"",bindingCount,pendingFailedCount,ingressReady,ingressReasons,status:ingressReady?"ready":"degraded",reasons,privateStateExposed:false};} async function status(){print(await readiness());} function loadRuntime(){const cr=createRequire(resolve(cult,"packages","cultcache-ts","package.json")),mr=createRequire(resolve(cult,"packages","cultmesh-ts","package.json")),nr=createRequire(resolve(cult,"packages","cultnet-ts","package.json")),cc=cr("@gamecult/cultcache-ts"),cn=nr("cultnet-ts"),mp=nr("@msgpack/msgpack");return {CultMesh:mr("cultmesh-ts").CultMesh,CultCache:cc.CultCache,SingleFileMessagePackBackingStore:cc.SingleFileMessagePackBackingStore,defineDocumentType:cc.defineDocumentType,CultNetDocumentRegistry:cn.CultNetDocumentRegistry,defineCultNetDocumentBinding:cn.defineCultNetDocumentBinding,cultnet:cn,msgpack:mp,encode:mp.encode,decode:mp.decode};} diff --git a/tools/persona-matrix-mouth.mjs b/tools/persona-matrix-mouth.mjs new file mode 100644 index 0000000..22429cb --- /dev/null +++ b/tools/persona-matrix-mouth.mjs @@ -0,0 +1,303 @@ +// The Matrix ear and mouth of one Persona. One process per Persona. +// +// Ear: /sync the configured rooms; every message from another sender becomes a +// delivery kept in memory. The /sync position (next_batch) is persisted only when +// an ack covers it, so a restart before the ack re-delivers instead of losing. +// Mouth: a send request becomes one PUT with txnId = request_id (a retry is +// idempotent) and only while the instance brake is released. +// Ring: the first delivery of an idle window starts the Persona's one-shot unit. +// Starting is a request; the unit's own gate decides. +// +// The Matrix access token is read here, from a systemd credential, and nowhere +// else. It is never put in a document, an environment passed to a child, or a log. +// +// Everything outside this process is a port (fetch, state, brake, ring, clock), so +// the core runs against a fake homeserver with no CultLib and no daemon. +import { execFile } from "node:child_process"; +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import { createRequire } from "node:module"; +import { createSocket } from "node:dgram"; +import { resolve, join } from "node:path"; +import { pathToFileURL } from "node:url"; +import * as docs from "./persona-chat-documents.mjs"; + +const API = "/_matrix/client/v3"; +const BRAKE_TYPE = "eureka.control.brake.v1"; +const CONTEXT = 20; +const RING_WINDOW_MS = 5 * 60 * 1000; + +export function createMouth({ + persona, homeserver, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake, ring, + now = Date.now, log = () => {}, syncTimeoutMs = 30000, +}) { + if (!persona || !homeserver || !token || !rooms?.length) throw new Error("persona, homeserver, rooms and token are required"); + const say = message => log(redact(String(message), token)); + const roomIdByAlias = new Map(); + const aliasByRoomId = new Map(); + const history = new Map(); // room_id -> last CONTEXT messages, oldest first + const displayNames = new Map(); + let selfId = null; + let cursor = null; // the /sync position we are reading from + let seqCounter = 0; + let pending = []; // deliveries not yet acked + let batches = []; // { token, maxSeq }: positions awaiting an ack + let committed = null; // the position that is persisted + let lastRingAt = -Infinity; + let started = false; + + async function matrix(method, path, body) { + let response; + try { + response = await httpFetch(`${homeserver}${API}${path}`, { + method, + headers: { authorization: `Bearer ${token}`, "content-type": "application/json" }, + body: body === undefined ? undefined : JSON.stringify(body), + }); + } catch (error) { + throw new Error(`homeserver unreachable: ${redact(error?.message ?? String(error), token)}`); + } + if (!response.ok) throw new Error(`homeserver answered ${response.status} for ${method} ${path.split("?")[0]}`); + return response.json(); + } + + async function start() { + const saved = await state.load(); + cursor = committed = saved?.nextBatch ?? null; + seqCounter = saved?.seq ?? 0; + selfId = (await matrix("GET", "/account/whoami")).user_id; + for (const alias of rooms) { + const { room_id: roomId } = await matrix("GET", `/directory/room/${encodeURIComponent(alias)}`); + await matrix("POST", `/join/${encodeURIComponent(roomId)}`, {}); + roomIdByAlias.set(alias, roomId); + aliasByRoomId.set(roomId, alias); + } + started = true; + } + + // One /sync round. Returns the deliveries it made. + async function syncOnce() { + if (!started) await start(); + 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 baseline = cursor === null; // first ever sync: history is context, not mail + const made = []; + for (const [roomId, room] of Object.entries(reply.rooms?.join ?? {})) { + const alias = aliasByRoomId.get(roomId); + if (!alias) continue; + for (const event of [...(room.state?.events ?? []), ...(room.timeline?.events ?? [])]) { + if (event.type === "m.room.member" && event.content?.displayname) displayNames.set(event.state_key, event.content.displayname); + } + for (const event of room.timeline?.events ?? []) { + if (event.type !== "m.room.message" || typeof event.content?.body !== "string") continue; + const entry = { sender: event.sender, text: event.content.body, atUnixMs: event.origin_server_ts ?? now() }; + const context = (history.get(roomId) ?? []).slice(-CONTEXT); + if (!baseline && event.sender !== selfId) { + made.push({ + seq: ++seqCounter, roomAlias: alias, roomId, eventId: event.event_id, sender: event.sender, + senderDisplay: displayNames.get(event.sender) ?? localpart(event.sender), text: entry.text, atUnixMs: entry.atUnixMs, + context: context.map(c => ({ ...c })), + }); + } + history.set(roomId, [...context, entry].slice(-CONTEXT)); + } + } + const wasIdle = pending.length === 0; + pending.push(...made); + cursor = reply.next_batch; + batches.push({ token: reply.next_batch, maxSeq: seqCounter }); + await commit(); + if (made.length > 0) await state.save({ nextBatch: committed, seq: seqCounter }); // sequence numbers never repeat across a restart + if (made.length > 0 && (wasIdle || now() - lastRingAt >= RING_WINDOW_MS)) { + lastRingAt = now(); + Promise.resolve().then(ring).catch(error => say(`ring failed: ${error?.message ?? error}`)); + } + return made; + } + + // The persisted position is the newest batch whose deliveries are all acked. + async function commit(throughSeq = pending.length ? pending[0].seq - 1 : seqCounter) { + let covered = null; + for (const b of batches) { if (b.maxSeq <= throughSeq) covered = b; else break; } + if (covered && covered.token !== committed) { + committed = covered.token; + await state.save({ nextBatch: committed, seq: seqCounter }); + } + batches = batches.filter(b => b.maxSeq > throughSeq); + } + + async function ack(throughSeq) { + pending = pending.filter(d => d.seq > throughSeq); + await commit(pending.length ? pending[0].seq - 1 : seqCounter); + } + + function pendingAfter(afterSeq) { + return pending.filter(d => d.seq > afterSeq).map(d => ({ ...d })); + } + + async function send({ requestId, persona: asked, roomAlias, text }) { + const refuse = refusal => ({ requestId, eventId: null, refusal }); + if (asked !== persona) return refuse("PersonaMismatch"); + let released = false; + try { released = (await brake()) === true; } catch { released = false; } + if (!released) return refuse("Brake"); + if (typeof text !== "string" || text.trim() === "") return refuse("EmptyText"); + if (!started) await start(); + const roomId = roomIdByAlias.get(roomAlias); + if (!roomId) return refuse("UnknownRoom"); + try { + const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }); + return { requestId, eventId: sent.event_id, refusal: null }; + } catch (error) { + say(`send failed: ${error.message}`); + return refuse("SendFailed"); + } + } + + async function run(signal) { + while (!signal?.aborted) { + try { await syncOnce(); } catch (error) { + say(`sync failed: ${error.message}`); + await new Promise(r => { const t = setTimeout(r, 5000); signal?.addEventListener("abort", () => { clearTimeout(t); r(); }, { once: true }); }); + } + } + } + + return { start, syncOnce, ack, pendingAfter, send, run, get pendingCount() { return pending.length; } }; +} + +export function redact(message, token) { + return token ? message.split(token).join("[redacted]") : message; +} + +function localpart(userId) { + return userId.replace(/^@/, "").split(":")[0]; +} + +// Ring: start the Persona's one-shot unit. No retry loop, no token in its environment. +export function createRing({ instance, run = execFile, log = () => {} }) { + return () => new Promise(done => { + run("systemctl", ["start", "--no-block", `mind-persona@${instance}.service`], { env: { PATH: process.env.PATH ?? "" } }, error => { + if (error) log(`ring failed: ${error.message}`); + done(); + }); + }); +} + +// Brake: released only when the instance's control store decodes to a released +// brake. Absent, undecodable or held reads are all "not released". +export function createBrakeReader({ controlPath, readStore, decode }) { + return async () => { + try { + const read = readStore(controlPath); + if (read.absent) return false; + const envelope = read.envelopes.find(e => e.type === BRAKE_TYPE && e.key === "brake"); + if (!envelope) return false; + const value = decode(new Uint8Array(envelope.payload)); + const released = Array.isArray(value) ? value[0] : value instanceof Map ? value.get(0) ?? value.get("released") : value?.[0] ?? value?.["0"] ?? value?.released; + return released === true; + } catch { + return false; + } + }; +} + +// The RUDP connection id is derived from the credential, so a peer without it cannot open the session. +export function connectionIdFor(connectionKey) { + return createHash("sha256").update(connectionKey).digest().readUInt32BE(0); +} + +// CultNet RUDP endpoint on loopback. Requests arrive as document_put_raw messages. +export async function startChatEndpoint({ mouth, persona, 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"); + const { msgpack, cultnet } = runtime; + 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 = (kind, recordKey, tuple) => connection.send("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"))); + let busy = Promise.resolve(); + const handle = async value => { + 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"); + 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); + if (request.persona !== persona) throw new Error("chat request names another Persona"); + respond(docs.kinds.pending, persona, docs.pendingTuple({ persona, deliveries: mouth.pendingAfter(request.afterSeq) })); + } else if (document.schemaId === docs.kinds.ack) { + const request = docs.parseAck(tuple); + if (request.persona !== persona) throw new Error("chat ack names another Persona"); + await mouth.ack(request.throughSeq); + } else { + const request = docs.parseSendRequest(tuple); + const receipt = await mouth.send(request); + respond(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(); } }; +} + +export function cultcacheState({ runtime, path }) { + const def = runtime.defineDocumentType({ + type: "gamecult.persona.chat_cursor.v1", schemaName: "gamecult.persona.chat_cursor.v1", schemaId: "gamecult.persona.chat_cursor.v1", + schemaVersion: "gamecult.persona.chat_cursor.v1", contentHash: "gamecult.persona.chat_cursor.v1", global: false, name: "key", + schema: { parse: v => ({ key: String(v.key), nextBatch: v.nextBatch ?? null, seq: Number(v.seq) }) }, + }); + const cache = runtime.CultCache.builder().withDocumentType(def).withGenericStore(new runtime.SingleFileMessagePackBackingStore(path)).build(); + const unwrap = v => (Array.isArray(v) ? v[0] : v?.value ?? v); + return { + async load() { await cache.pullAllBackingStores(); return unwrap(cache.get(def, "cursor")) ?? null; }, + async save({ nextBatch, seq }) { await cache.put(def, "cursor", { key: "cursor", nextBatch, seq }); }, + }; +} + +export function loadRuntime(cultRoot) { + const at = name => createRequire(resolve(cultRoot, "packages", name, "package.json")); + const cc = at("cultcache-ts")("@gamecult/cultcache-ts"), cn = at("cultnet-ts")("cultnet-ts"), mp = at("cultnet-ts")("@msgpack/msgpack"); + return { CultCache: cc.CultCache, SingleFileMessagePackBackingStore: cc.SingleFileMessagePackBackingStore, defineDocumentType: cc.defineDocumentType, readSingleFileStore: cc.readSingleFileStore, cultnet: cn, msgpack: mp }; +} + +function required(env, name) { + if (!env[name]) throw new Error(`${name} is required`); + return env[name]; +} +function credential(env, name) { + return readFileSync(join(required(env, "CREDENTIALS_DIRECTORY"), name), "utf8").trim(); +} + +export async function main(env = process.env) { + const persona = required(env, "PERSONA"); + const instance = env.MIND_INSTANCE ?? persona; // the unit template is instanced by the Persona's mind + if (!/^[A-Za-z0-9_-]{1,64}$/.test(instance)) throw new Error("MIND_INSTANCE is not a slug"); + const token = credential(env, "matrix-token"); + const runtime = loadRuntime(resolve(env.CULTLIB_ROOT ?? resolve(import.meta.dirname, "..", "..", "CultLib"))); + const log = line => console.error(line); + const controlPath = join(env.CONTROL_ROOT ?? "/etc/gamecult/minds", instance, "control.cc"); + const mouth = createMouth({ + persona, token, log, + homeserver: required(env, "MATRIX_HOMESERVER").replace(/\/$/, ""), + rooms: required(env, "MATRIX_ROOMS").split(",").map(s => s.trim()).filter(Boolean), + state: cultcacheState({ runtime, path: env.STATE_PATH ?? join(required(env, "STATE_DIRECTORY"), "cursor.cc") }), + brake: createBrakeReader({ controlPath, readStore: runtime.readSingleFileStore, decode: runtime.msgpack.decode }), + ring: createRing({ instance, log }), + }); + await mouth.start(); + const listen = required(env, "CHAT_RUDP_LISTEN"); // host:port, loopback + const colon = listen.lastIndexOf(":"); + await startChatEndpoint({ mouth, persona, runtime, host: listen.slice(0, colon), port: Number(listen.slice(colon + 1)), connectionKey: credential(env, "chat-key"), runtimeId: `bifrost-persona-mouth-${persona}`, log }); + await mouth.run(); +} + +if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) { + main().catch(error => { console.error(error?.message ?? String(error)); process.exitCode = 1; }); +}