Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/core/jsonrpc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
49 changes: 49 additions & 0 deletions src/openhuman/orchestration/ingest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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::*;
Expand All @@ -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");
Expand Down
1 change: 1 addition & 0 deletions src/openhuman/orchestration/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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};
27 changes: 27 additions & 0 deletions src/openhuman/orchestration/ops.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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).
Expand Down
45 changes: 23 additions & 22 deletions src/openhuman/tinyplace/manifest.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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)");

Comment thread
coderabbitai[bot] marked this conversation as resolved.
// 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
Expand Down
7 changes: 4 additions & 3 deletions src/openhuman/tinyplace/schemas.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
Loading