persona: connect Starfire cognition to Bifrost mouth
This commit is contained in:
parent
58b26c5f81
commit
052d5b1ec1
6 changed files with 260 additions and 9 deletions
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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<std::collections::BTreeMap<String, String>> {
|
||||
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<String, String>, key: &str) -> Result<String> {
|
||||
values
|
||||
.get(key)
|
||||
.cloned()
|
||||
.ok_or_else(|| anyhow!("{key} is required"))
|
||||
}
|
||||
fn path(values: &std::collections::BTreeMap<String, String>, key: &str) -> Result<PathBuf> {
|
||||
Ok(value(values, key)?.into())
|
||||
}
|
||||
|
|
@ -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<Option<crate::PersonaDiscordDeliveryRequest>> {
|
||||
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,
|
||||
|
|
|
|||
|
|
@ -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::<PersonaDiscordDeliveryReceipt>()?;
|
||||
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<PersonaDiscordDeliveryReceipt> {
|
||||
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<GameCultServiceTrustAnchorRecord> {
|
||||
|
|
|
|||
|
|
@ -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<bool> {
|
|||
));
|
||||
}
|
||||
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<bool> {
|
|||
&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"),
|
||||
|
|
|
|||
2
vendor/cultnet-rs
vendored
2
vendor/cultnet-rs
vendored
|
|
@ -1 +1 @@
|
|||
Subproject commit fab5b31d72a6b1b5abe0787b0036fedb2af639d5
|
||||
Subproject commit 2d0988ba84f47e37831e5a4533a652022ef816e4
|
||||
Reference in a new issue