From 052d5b1ec194b9466d33858d2d7de9ec5c1acec3 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 8 Aug 2026 20:10:48 +0100 Subject: [PATCH] persona: connect Starfire cognition to Bifrost mouth --- epiphany-core/Cargo.toml | 4 + ...any-persona-discord-rudp-client-fixture.rs | 82 +++++++++++++ epiphany-core/src/persona_conversation.rs | 15 +++ epiphany-core/src/persona_discord_crossing.rs | 114 +++++++++++++++++- .../src/bin/epiphany-persona-service.rs | 52 +++++++- vendor/cultnet-rs | 2 +- 6 files changed, 260 insertions(+), 9 deletions(-) create mode 100644 epiphany-core/src/bin/epiphany-persona-discord-rudp-client-fixture.rs diff --git a/epiphany-core/Cargo.toml b/epiphany-core/Cargo.toml index 9b0fa03e..6484cadf 100644 --- a/epiphany-core/Cargo.toml +++ b/epiphany-core/Cargo.toml @@ -120,6 +120,10 @@ path = "src/bin/epiphany-persona-discord.rs" name = "epiphany-persona-discord-crossing-fixture" path = "src/bin/epiphany-persona-discord-crossing-fixture.rs" +[[bin]] +name = "epiphany-persona-discord-rudp-client-fixture" +path = "src/bin/epiphany-persona-discord-rudp-client-fixture.rs" + [[bin]] name = "epiphany-persona-discord-permit" path = "src/bin/epiphany-persona-discord-permit.rs" diff --git a/epiphany-core/src/bin/epiphany-persona-discord-rudp-client-fixture.rs b/epiphany-core/src/bin/epiphany-persona-discord-rudp-client-fixture.rs new file mode 100644 index 00000000..3f7362f9 --- /dev/null +++ b/epiphany-core/src/bin/epiphany-persona-discord-rudp-client-fixture.rs @@ -0,0 +1,82 @@ +use std::{env, net::SocketAddr, path::PathBuf, time::Duration}; + +use anyhow::{Result, anyhow}; +use epiphany_core::{ + enroll_persona_discord_request_identity, exchange_persona_discord_delivery_rudp, + export_persona_discord_request_anchor_candidate, insert_persona_discord_delivery_request, + load_persona_discord_receipt_anchor, open_persona_discord_request_identity, + sign_persona_discord_delivery_request, +}; + +fn main() -> Result<()> { + let values = args()?; + let request_store = path(&values, "--request-store")?; + let receipt_store = path(&values, "--receipt-store")?; + let identity_store = path(&values, "--identity-store")?; + let request_anchor = path(&values, "--request-anchor")?; + let receipt_anchor = path(&values, "--receipt-anchor")?; + let endpoint: SocketAddr = value(&values, "--endpoint")?.parse()?; + enroll_persona_discord_request_identity(&identity_store)?; + export_persona_discord_request_anchor_candidate( + &identity_store, + &request_anchor, + "epiphany-starfire", + )?; + let signer = open_persona_discord_request_identity(&identity_store)?; + let issued = chrono::Utc::now(); + let request = sign_persona_discord_delivery_request( + &signer, + "persona-discord:rudp-cross-language:0".into(), + "effect-rudp-cross-language".into(), + "epiphany-starfire".into(), + "epiphany.Persona".into(), + "123".into(), + "456".into(), + "Cross-language Persona nerve smoke.".into(), + issued.to_rfc3339(), + (issued + chrono::Duration::seconds(90)).to_rfc3339(), + )?; + insert_persona_discord_delivery_request(&request_store, &request)?; + let anchor = load_persona_discord_receipt_anchor(&receipt_anchor)?; + let receipt = exchange_persona_discord_delivery_rudp( + endpoint, + "epiphany-starfire", + &request_store, + &receipt_store, + &request, + &anchor, + Duration::from_secs(5), + )?; + println!( + "{}", + serde_json::to_string(&serde_json::json!({ + "requestId": receipt.request_id, + "status": receipt.status, + "messageId": receipt.message_id, + "providerIdentityId": receipt.provider_identity_id, + }))? + ); + Ok(()) +} + +fn args() -> Result> { + let mut values = std::collections::BTreeMap::new(); + let mut args = env::args().skip(1); + while let Some(key) = args.next() { + values.insert( + key, + args.next() + .ok_or_else(|| anyhow!("argument requires value"))?, + ); + } + Ok(values) +} +fn value(values: &std::collections::BTreeMap, key: &str) -> Result { + values + .get(key) + .cloned() + .ok_or_else(|| anyhow!("{key} is required")) +} +fn path(values: &std::collections::BTreeMap, key: &str) -> Result { + Ok(value(values, key)?.into()) +} diff --git a/epiphany-core/src/persona_conversation.rs b/epiphany-core/src/persona_conversation.rs index 3b4485aa..b5d4b11c 100644 --- a/epiphany-core/src/persona_conversation.rs +++ b/epiphany-core/src/persona_conversation.rs @@ -450,6 +450,21 @@ pub fn persona_delivery_receipt_exists_for_turn( Ok(false) } +pub fn pending_persona_discord_delivery_request_for_turn( + request_store: &Path, + request_id: &str, +) -> Result> { + for index in 0..16 { + if let Some(request) = crate::load_persona_discord_delivery_request( + request_store, + &format!("persona-discord:{request_id}:{index}"), + )? { + return Ok(Some(request)); + } + } + Ok(None) +} + fn admit_state_notes( runtime_store: &Path, agent_store: &Path, diff --git a/epiphany-core/src/persona_discord_crossing.rs b/epiphany-core/src/persona_discord_crossing.rs index 5d5a870e..cb4773b2 100644 --- a/epiphany-core/src/persona_discord_crossing.rs +++ b/epiphany-core/src/persona_discord_crossing.rs @@ -1,4 +1,8 @@ -use std::path::Path; +use std::{ + net::{SocketAddr, UdpSocket}, + path::Path, + time::{Duration, Instant}, +}; use anyhow::{Result, anyhow, bail}; use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; @@ -6,8 +10,11 @@ use serde::Serialize; use sha2::{Digest, Sha256}; use cultnet_rs::{ + CultNetMessage, CultNetRawDocumentRecord, CultNetRawPayloadEncoding, + CultNetRudpSocketTransportConnection, CultNetRudpSocketTransportOptions, CultNetWireContract, GAMECULT_SERVICE_TRUST_ANCHOR_SCHEMA, GameCultServiceTrustAnchorRecord, ServiceIdentityProfile, ServiceIdentitySignature, ServiceIdentitySigner, ServiceSignaturePurpose, + decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, verify_service_identity_signature_with_public_key, }; @@ -20,9 +27,9 @@ pub const PERSONA_DISCORD_DELIVERY_REQUEST_SIGNING_PURPOSE: &str = pub const PERSONA_DISCORD_DELIVERY_RECEIPT_SIGNING_PURPOSE: &str = "bifrost.persona-discord.delivery-receipt.v0"; pub const EPIPHANY_PERSONA_MOUTH_SERVICE_ID: &str = "epiphany-persona-mouth"; -pub const EPIPHANY_PERSONA_MOUTH_RUNTIME_ID: &str = "epiphany-yggdrasil"; pub const BIFROST_PERSONA_DELIVERY_SERVICE_ID: &str = "bifrost-persona-discord-delivery"; pub const BIFROST_PERSONA_DELIVERY_RUNTIME_ID: &str = "bifrost-discord-yggdrasil"; +pub const PERSONA_DISCORD_DELIVERY_RUDP_CONNECTION_ID: u32 = 0xe91f_0003; pub enum EpiphanyPersonaDeliveryRequestIdentity {} pub struct EpiphanyPersonaDeliveryRequestPurpose; @@ -315,6 +322,109 @@ pub fn load_persona_discord_delivery_receipt( load_exact(path, id) } +pub fn insert_persona_discord_delivery_receipt( + path: &Path, + receipt: &PersonaDiscordDeliveryReceipt, +) -> Result<()> { + validate_receipt_shape(receipt)?; + let mut cache = CultCache::new(); + cache.register_entry_type::()?; + let (entry, _) = cache.prepare_entry(&receipt.request_id, receipt)?; + if !SingleFileMessagePackBackingStore::new(path).insert_entry_if_absent(entry)? { + let existing = load_persona_discord_delivery_receipt(path, &receipt.request_id)?; + if existing.as_ref() != Some(receipt) { + bail!("Persona delivery receipt identity collision"); + } + } + Ok(()) +} + +pub fn exchange_persona_discord_delivery_rudp( + endpoint: SocketAddr, + runtime_id: &str, + request_store: &Path, + receipt_store: &Path, + request: &PersonaDiscordDeliveryRequest, + receipt_anchor: &GameCultServiceTrustAnchorRecord, + timeout: Duration, +) -> Result { + if load_persona_discord_delivery_request(request_store, &request.request_id)?.as_ref() + != Some(request) + { + bail!("Persona delivery request must be durable before transport"); + } + let message = CultNetMessage::DocumentPutRaw { + message_id: format!("persona-delivery-request-{}", request.request_id), + document: CultNetRawDocumentRecord { + schema_id: PERSONA_DISCORD_DELIVERY_REQUEST_SCHEMA_VERSION.into(), + record_key: request.request_id.clone(), + stored_at: request.issued_at.clone(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload: rmp_serde::to_vec(request)?, + source_runtime_id: Some(runtime_id.into()), + source_agent_id: Some(request.signer_identity_id.clone()), + source_role: Some("epiphany-persona-mouth".into()), + tags: Some(vec!["cultnet.transport.rudp.v0".into()]), + }, + }; + let socket = UdpSocket::bind("0.0.0.0:0")?; + socket.set_read_timeout(Some(Duration::from_millis(50)))?; + let mut transport = + CultNetRudpSocketTransportConnection::new(CultNetRudpSocketTransportOptions::client( + runtime_id, + socket, + endpoint, + PERSONA_DISCORD_DELIVERY_RUDP_CONNECTION_ID, + ))?; + transport.connect(Vec::new())?; + let deadline = Instant::now() + timeout; + while !transport.connected() { + let _ = transport.receive_once()?; + transport.poll_resends()?; + if Instant::now() >= deadline { + bail!("Bifrost Persona delivery handshake unavailable"); + } + } + transport.send( + "schema", + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0)?, + )?; + loop { + if let Some(frame) = transport.receive_once()? { + if frame.channel_id != "schema" { + bail!("Bifrost Persona receipt arrived on a foreign channel"); + } + let response = decode_cultnet_message_from_slice( + &frame.payload, + CultNetWireContract::CultNetSchemaV0, + )?; + let CultNetMessage::DocumentPutRaw { document, .. } = response else { + bail!("Bifrost Persona delivery returned a foreign message"); + }; + let receipt: PersonaDiscordDeliveryReceipt = rmp_serde::from_slice(&document.payload)?; + if document.schema_id != PERSONA_DISCORD_DELIVERY_RECEIPT_SCHEMA_VERSION + || document.record_key != request.request_id + || document.payload_encoding != CultNetRawPayloadEncoding::Messagepack + || document.source_runtime_id.as_deref() + != Some(BIFROST_PERSONA_DELIVERY_RUNTIME_ID) + || document.source_agent_id.as_deref() + != Some(receipt.provider_identity_id.as_str()) + || document.source_role.as_deref() != Some("bifrost-persona-discord-delivery") + || document.tags.as_deref() != Some(&["cultnet.transport.rudp.v0".to_string()]) + { + bail!("Bifrost Persona receipt envelope substituted authority"); + } + verify_persona_discord_delivery_receipt(&receipt, request, receipt_anchor)?; + insert_persona_discord_delivery_receipt(receipt_store, &receipt)?; + return Ok(receipt); + } + transport.poll_resends()?; + if Instant::now() >= deadline { + bail!("Bifrost Persona delivery terminal receipt unavailable"); + } + } +} + pub fn load_persona_discord_receipt_anchor( path: &Path, ) -> Result { diff --git a/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs b/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs index 542c4bfb..12bd588f 100644 --- a/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs +++ b/epiphany-openai-runtime/src/bin/epiphany-persona-service.rs @@ -1,13 +1,14 @@ -use std::{env, path::PathBuf, process::Command, thread, time::Duration}; +use std::{env, net::SocketAddr, path::PathBuf, process::Command, thread, time::Duration}; use anyhow::{Context, Result, anyhow}; use epiphany_core::{ PersonaIdentity, PersonaProjectorInput, PersonaRepoActivity, PersonaSocialAffordance, PersonaTranscriptMessage, PersonaTurnTerminalOptions, complete_persona_turn_request_store, - default_organ_dependencies_for, load_agent_memory_entry_for_role, - load_epiphany_cultmesh_swarm_brake, load_heartbeat_state_entry, - load_persona_discord_receipt_anchor, load_persona_discord_service_anchor, - open_persona_discord_request_identity, persona_delivery_receipt_exists_for_turn, + default_organ_dependencies_for, exchange_persona_discord_delivery_rudp, + load_agent_memory_entry_for_role, load_epiphany_cultmesh_swarm_brake, + load_heartbeat_state_entry, load_persona_discord_receipt_anchor, + load_persona_discord_service_anchor, open_persona_discord_request_identity, + pending_persona_discord_delivery_request_for_turn, persona_delivery_receipt_exists_for_turn, persona_model_terminal_exists, poll_persona_discord_crossing, reconcile_terminal_persona_conversation, semantic_memory_recall_from_heartbeat_action, validate_persona_discord_request_anchor, @@ -173,7 +174,7 @@ async fn poll_once(options: &Options) -> Result { )); } let receipt_anchor = load_persona_discord_receipt_anchor(&options.mouth_receipt_anchor)?; - let result = poll_persona_discord_crossing( + let mut result = poll_persona_discord_crossing( &options.runtime_store, &options.heartbeat_store, &options.agent_store, @@ -186,6 +187,39 @@ async fn poll_once(options: &Options) -> Result { &request.request_id, &model_terminal.effect_document_id, )?; + if result.is_none() { + let crossing = pending_persona_discord_delivery_request_for_turn( + &options.mouth_request_store, + &request.request_id, + )? + .ok_or_else(|| anyhow!("Persona crossing is pending without a durable request"))?; + match exchange_persona_discord_delivery_rudp( + options.mouth_rudp, + &options.runtime_id, + &options.mouth_request_store, + &options.mouth_receipt_store, + &crossing, + &receipt_anchor, + Duration::from_millis(options.mouth_timeout_ms), + ) { + Ok(_) => { + result = poll_persona_discord_crossing( + &options.runtime_store, + &options.heartbeat_store, + &options.agent_store, + &options.cultmesh_store, + &options.runtime_id, + &options.mouth_request_store, + &options.mouth_receipt_store, + &signer, + &receipt_anchor, + &request.request_id, + &model_terminal.effect_document_id, + )?; + } + Err(error) => eprintln!("Persona delivery remains durably pending: {error:#}"), + } + } Ok(result.is_some()) } @@ -207,6 +241,8 @@ struct Options { mouth_identity_store: PathBuf, mouth_request_anchor: PathBuf, mouth_receipt_anchor: PathBuf, + mouth_rudp: SocketAddr, + mouth_timeout_ms: u64, poll_ms: u64, once: bool, } @@ -244,6 +280,10 @@ impl Options { mouth_identity_store: path("--mouth-identity-store")?, mouth_request_anchor: path("--mouth-request-anchor")?, mouth_receipt_anchor: path("--mouth-receipt-anchor")?, + mouth_rudp: value(&values, "--mouth-rudp", "") + .parse() + .context("--mouth-rudp must be an IP:port socket address")?, + mouth_timeout_ms: value(&values, "--mouth-timeout-ms", "10000").parse()?, runtime_id: value(&values, "--runtime-id", "epiphany-local"), provider: value(&values, "--provider", "openai-codex"), model: value(&values, "--model", "gpt-5.4"), diff --git a/vendor/cultnet-rs b/vendor/cultnet-rs index fab5b31d..2d0988ba 160000 --- a/vendor/cultnet-rs +++ b/vendor/cultnet-rs @@ -1 +1 @@ -Subproject commit fab5b31d72a6b1b5abe0787b0036fedb2af639d5 +Subproject commit 2d0988ba84f47e37831e5a4533a652022ef816e4