feat(orchestration): surface inbound relay DMs via mailbox poll-drain + fix Signal card key (#4564)

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();

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<usize, String> {
+    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");

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};

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).

src/openhuman/tinyplace/manifest.rs

@@ -2991,13 +2991,11 @@ pub(crate) fn handle_tinyplace_signal_key_status(_params: Map<String, Value>) ->
                     .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 == &current_key_b64)
                     .unwrap_or(false);
@@ -3611,33 +3609,36 @@ pub(crate) fn handle_tinyplace_contacts_stats(_params: Map<String, Value>) -> 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<String, Value>,
 ) -> 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/<cryptoId>/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

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",