diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 879645ecd3..76c41f2544 100644 --- a/src/core/jsonrpc.rs +++ b/src/core/jsonrpc.rs @@ -2294,6 +2294,10 @@ fn register_domain_subscribers( crate::openhuman::orchestration::register_orchestration_ingest_subscriber(); // Orchestration: wake the split-brain graph on each persisted session DM. crate::openhuman::orchestration::register_orchestration_wake_subscriber(); + // Orchestration: relay DMs are poll-only (`/messages`) and never traverse + // `/inbox/stream`, so this poller is the delivery path that surfaces + // inbound DMs from paired agents into the wake graph. + crate::openhuman::orchestration::start_message_drain_supervisor(); // Task-sources proactive ingestion: connection-created hook + poll. crate::openhuman::task_sources::bus::register_task_sources_subscriber(); crate::openhuman::task_sources::start_periodic_poll(); diff --git a/src/openhuman/orchestration/ingest.rs b/src/openhuman/orchestration/ingest.rs index f475ff03c2..1d55300d27 100644 --- a/src/openhuman/orchestration/ingest.rs +++ b/src/openhuman/orchestration/ingest.rs @@ -197,6 +197,46 @@ async fn ingest_one( Ok(()) } +/// Poll the relay mailbox once and run every delivered envelope through the +/// ingest pipeline. +/// +/// The relay delivers DMs to `/messages` (poll-only) and — unlike inbox items — +/// never publishes them to the `/inbox/stream` WebSocket, so a poller is the +/// only way orchestration learns about inbound DMs. Envelopes from senders that +/// are not orchestration-linked are skipped by [`ingest_one`] WITHOUT being +/// decrypted or acknowledged, so they stay in the mailbox for the Messaging UI. +/// +/// Returns the number of envelopes examined this pass. Best-effort per envelope: +/// a decrypt/persist failure on one is logged and does not abort the batch. +pub async fn drain_mailbox_once(config: &Config) -> Result { + if !config.orchestration.enabled { + return Ok(0); + } + let client = crate::openhuman::tinyplace::ops::global_state() + .client() + .await?; + let signer = client + .http() + .signer() + .ok_or_else(|| "no signer configured".to_string())?; + let agent_id = signer.agent_id(); + let resp = client + .messages + .list(&agent_id, Some(100)) + .await + .map_err(|e| format!("messages.list: {e}"))?; + let count = resp.messages.len(); + if count > 0 { + log::debug!(target: LOG, "[orchestration] drain.fetched count={count}"); + } + for envelope in resp.messages { + if let Err(e) = ingest_one(config, envelope).await { + log::warn!(target: LOG, "[orchestration] drain.ingest_error: {e}"); + } + } + Ok(count) +} + #[cfg(test)] mod tests { use super::*; @@ -220,6 +260,15 @@ mod tests { assert!(!is_dm_stream("inbox", "inbox")); } + #[tokio::test] + async fn drain_is_a_noop_when_orchestration_disabled() { + // Guard short-circuits before touching the tiny.place client, so this + // exercises the early return without any wallet/network. + let mut config = Config::default(); + config.orchestration.enabled = false; + assert_eq!(drain_mailbox_once(&config).await, Ok(0)); + } + #[test] fn classifies_harness_envelope_as_session() { let c = classify_message(ENVELOPE.to_string(), "2026-07-02T09:00:00Z"); diff --git a/src/openhuman/orchestration/mod.rs b/src/openhuman/orchestration/mod.rs index 66f44b3fa0..613f6537c1 100644 --- a/src/openhuman/orchestration/mod.rs +++ b/src/openhuman/orchestration/mod.rs @@ -30,4 +30,5 @@ pub use graph::{ build_orchestration_graph, orchestration_graph_topology, run_orchestration_graph, OrchestrationState, }; +pub use ops::start_message_drain_supervisor; pub use schemas::{all_controller_schemas, all_registered_controllers}; diff --git a/src/openhuman/orchestration/ops.rs b/src/openhuman/orchestration/ops.rs index ef9a04a614..b087b9bd5a 100644 --- a/src/openhuman/orchestration/ops.rs +++ b/src/openhuman/orchestration/ops.rs @@ -117,6 +117,33 @@ pub async fn schedule_wake(agent_id: String, session_id: String, chat_kind: Stri }); } +/// Periodically drain the relay mailbox through orchestration ingest. +/// +/// tiny.place relay DMs are delivered to `/messages` (poll-only) and are NOT +/// published to the `/inbox/stream` WebSocket — the backend only streams inbox +/// items for payments/notifications, never `PUT /messages`. So a poller is the +/// actual delivery path: it lists `/messages` and feeds each envelope through +/// the same decrypt → classify → persist → acknowledge pipeline the wake graph +/// consumes. Unlinked senders are skipped without being consumed, so their DMs +/// remain readable by the Messaging UI. +pub fn start_message_drain_supervisor() { + tokio::spawn(async { + loop { + match Config::load_or_init().await { + Ok(config) => match super::ingest::drain_mailbox_once(&config).await { + Ok(n) if n > 0 => { + log::debug!(target: LOG, "[orchestration] drain: examined {n} envelope(s)") + } + Ok(_) => {} + Err(e) => log::debug!(target: LOG, "[orchestration] drain error: {e}"), + }, + Err(e) => log::debug!(target: LOG, "[orchestration] drain config load: {e}"), + } + tokio::time::sleep(std::time::Duration::from_secs(15)).await; + } + }); +} + /// Seed a wake-cycle [`OrchestrationState`] from the store: the counterpart to /// reply to plus the recent-message window. Returns `None` when the session has /// no persisted messages (nothing to wake for). diff --git a/src/openhuman/tinyplace/manifest.rs b/src/openhuman/tinyplace/manifest.rs index f4c0d98179..82947120f2 100644 --- a/src/openhuman/tinyplace/manifest.rs +++ b/src/openhuman/tinyplace/manifest.rs @@ -2991,13 +2991,11 @@ pub(crate) fn handle_tinyplace_signal_key_status(_params: Map) -> .metadata .as_ref() .and_then(|m| m.get("encryptionPublicKey")); - let current_key_b64 = base64::engine::general_purpose::STANDARD.encode( - store - .identity_x25519_key_pair() - .await - .map(|kp| kp.public_key) - .unwrap_or([0u8; 32]), - ); + // The card advertises the Ed25519 identity key (the addressable + // messaging key where the bundle + mailbox live), not the X25519 + // DH key — see `signal_register_encryption_key`. Compare against + // that so readiness reflects what peers actually resolve. + let current_key_b64 = signer.public_key_base64(); let matches = published_key .map(|pk| pk == ¤t_key_b64) .unwrap_or(false); @@ -3611,33 +3609,36 @@ pub(crate) fn handle_tinyplace_contacts_stats(_params: Map) -> Co // ── Signal: encryption key registration (0D) ──────────────────────────────── -/// Publish the user's X25519 identity public key on their directory card as +/// Publish the user's Ed25519 identity key (the addressable cryptoId, where the +/// Signal prekey bundle + mailbox live) on their directory card as /// `metadata.encryptionPublicKey`. This makes the user discoverable for -/// encrypted DMs via `find_agent_by_encryption_key`. +/// encrypted DMs via `find_agent_by_encryption_key`; peers derive the X25519 DH +/// key from the fetched bundle themselves. /// -/// SECURITY: only the PUBLIC key is published. The private key never leaves -/// the `FileSessionStore`. +/// SECURITY: only the PUBLIC key is published. pub(crate) fn handle_tinyplace_signal_register_encryption_key( _params: Map, ) -> ControllerFuture { Box::pin(async move { log::debug!("{LOG_PREFIX} signal_register_encryption_key"); - // 1. Read identity public key from the signal store. - let store = crate::openhuman::tinyplace::signal_store::global_signal_store().await?; - let identity_kp = store - .identity_x25519_key_pair() - .await - .map_err(|e| format!("identity key: {e}"))?; - let encryption_key_b64 = - base64::engine::general_purpose::STANDARD.encode(identity_kp.public_key); - log::debug!("{LOG_PREFIX} signal_register_encryption_key derived key (not logging value)"); - - // 2. Acquire client and signer. + // 1. Acquire client and signer. let client = global_state().client().await?; let signer = require_signer(client)?; let agent_id = signer.agent_id(); + // 2. The messaging key peers resolve from this directory card must be the + // wallet's **Ed25519 identity key** — that is the addressable identity + // where the Signal prekey bundle is published (`/keys//bundle`) + // and where the mailbox is keyed. Peers derive the X25519 DH key from + // the bundle's identity key themselves (see `decode_identity_key`), so + // they never need a separate X25519 key advertised here. Publishing the + // X25519 key instead made every peer resolve to a *bundle-less* key and + // 404 on the bundle fetch — the exact reason inbound DMs never reached + // this agent. Advertise the identity key so resolution + delivery align. + let encryption_key_b64 = signer.public_key_base64(); + log::debug!("{LOG_PREFIX} signal_register_encryption_key advertising identity key (value not logged)"); + // 3. Fetch current AgentCard to preserve existing fields. A wallet that // has Signal keys but no directory presence yet (e.g. it registered a // @handle but was never upserted as an agent) 404s here — in that diff --git a/src/openhuman/tinyplace/schemas.rs b/src/openhuman/tinyplace/schemas.rs index 2e315401da..407ba2e697 100644 --- a/src/openhuman/tinyplace/schemas.rs +++ b/src/openhuman/tinyplace/schemas.rs @@ -2013,9 +2013,10 @@ fn schema_signal_register_encryption_key() -> ControllerSchema { ControllerSchema { namespace: "tinyplace", function: "signal_register_encryption_key", - description: "Publish the user's Signal X25519 identity public key on their directory \ - card (metadata.encryptionPublicKey). Makes the user discoverable for \ - encrypted DMs. Reads the key from the local Signal store — no params needed.", + description: "Publish the user's Ed25519 identity key (the addressable cryptoId, where \ + the Signal prekey bundle + mailbox live) on their directory card \ + (metadata.encryptionPublicKey). Peers derive the X25519 DH key from the bundle \ + themselves. Makes the user discoverable for encrypted DMs. No params needed.", inputs: vec![], outputs: vec![json_output( "result",