meerkat_mobkit/runtime/
cross_mob_remote.rs

1//! Cross-mob remote-handle proxy.
2//!
3//! `cross_mob.rs` historically dispatched all wire/unwire/send operations
4//! against an `Arc<MobHandle>` registered via `register_peer_mob`, which
5//! only works for peers living in the *same* process (inproc transport).
6//! This module introduces the seam for **remote** peers — mobs that live
7//! in a different process or on a different host, reachable over TCP or
8//! UDS.
9//!
10//! # Design (Phase 1)
11//!
12//! Per the 0.6 cross-mob plan we picked design **(A)** — a `LocalOrRemote`
13//! enum at the cross-mob dispatch site (`cross_mob.rs`). The local arm
14//! uses the existing `MobHandle` path; the remote arm uses [`RemoteMobProxy`]
15//! to talk to the peer mob's admin endpoint over TCP/UDS.
16//!
17//! Phase 1 lands the **structural seam**: the `RemoteMobProxy` type, the
18//! `LocalOrRemote` enum, and the contact-directory scheme (`tcp://host:port`,
19//! `uds:///path`) flowing all the way through. Wire setup at the comms
20//! layer already supports those addresses — outbound message routing uses
21//! `meerkat_comms::Router`, which dispatches by `PeerAddr` scheme. The
22//! per-member transport just works once peer specs carry the correct
23//! address.
24//!
25//! What this module does **not** ship in Phase 1:
26//!
27//! 1. **Cross-process control RPC.** A real `RemoteMobProxy::wire` would
28//!    open a TCP/UDS connection to the peer mob's admin endpoint, send a
29//!    `WIRE` request, and await acknowledgment. Today the methods on
30//!    `RemoteMobProxy` return [`RemoteMobError::ControlChannelUnavailable`]
31//!    so callers learn the seam exists but is not yet wired.
32//! 2. **Ed25519-signed peer descriptors.** Sibling Unit 4 lands signed
33//!    descriptor authoring; this unit keeps using
34//!    `TrustedPeerSpec::new(...)` (which is structurally `test_only_unsigned`
35//!    in the 0.5.x core seam). The seam where Unit 4 will plug in is the
36//!    helpers `build_tcp_peer_spec` / `build_uds_peer_spec` in
37//!    `cross_mob.rs`.
38//!
39//! # Phase 2 plan (out of scope here)
40//!
41//! - Add a small JSON-over-CBOR control protocol with three operations:
42//!   `WireRequest { local_member, peer_spec }`,
43//!   `UnwireRequest { local_member, peer_spec }`,
44//!   `InjectRequest { remote_member, content, handling_mode }`.
45//!   Frame with `meerkat_comms::transport::codec::TransportCodec` (length-prefix
46//!   CBOR) — same wire shape as agent envelopes, distinct payload type.
47//! - Bind a control listener on each `UnifiedRuntime` startup when the
48//!   contact directory advertises a TCP/UDS endpoint for *this* mob.
49//! - `RemoteMobProxy` opens the connection lazily and reuses it for
50//!   subsequent calls; reconnect on drop.
51
52use crate::contact_directory::{ContactEntry, MobTransport};
53
54/// Errors raised by the remote-mob proxy.
55#[derive(Debug, Clone, PartialEq, Eq)]
56pub enum RemoteMobError {
57    /// The contact directory entry uses an unsupported transport.
58    UnsupportedTransport { mob_id: String, transport: String },
59    /// The cross-process control channel for this peer mob is not
60    /// reachable: connect failed, write/read failed, or the peer timed
61    /// out. The `operation` field reports which RPC stage hit the wall.
62    ControlChannelUnavailable {
63        mob_id: String,
64        endpoint: String,
65        operation: &'static str,
66    },
67    /// The peer rejected our request with an error response. Carries the
68    /// peer's stable `code` (e.g. `unknown_member`, `mob_error`) and a
69    /// human-readable message.
70    Rejected {
71        mob_id: String,
72        endpoint: String,
73        code: String,
74        message: String,
75    },
76    /// We failed to encode the request payload — usually a programmer
77    /// error (un-serializable JSON value passed in).
78    Encode { endpoint: String, message: String },
79    /// We received a response we couldn't decode — peer is speaking a
80    /// different protocol or sent corrupted data.
81    Decode { endpoint: String, message: String },
82}
83
84impl RemoteMobError {
85    /// Attach IO-error context to a `ControlChannelUnavailable` variant.
86    /// Internal helper for the control-protocol module — keeps the public
87    /// shape stable while letting us record the underlying message.
88    pub(crate) fn with_context(self, context: String) -> Self {
89        match self {
90            Self::ControlChannelUnavailable {
91                mob_id,
92                endpoint,
93                operation,
94            } => Self::ControlChannelUnavailable {
95                mob_id: if mob_id.is_empty() { context } else { mob_id },
96                endpoint,
97                operation,
98            },
99            other => other,
100        }
101    }
102}
103
104impl std::fmt::Display for RemoteMobError {
105    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106        match self {
107            Self::UnsupportedTransport { mob_id, transport } => {
108                write!(
109                    f,
110                    "remote mob '{mob_id}' uses unsupported transport: {transport}"
111                )
112            }
113            Self::ControlChannelUnavailable {
114                mob_id,
115                endpoint,
116                operation,
117            } => {
118                write!(
119                    f,
120                    "remote mob '{mob_id}' control channel ({endpoint}): operation '{operation}' \
121                     failed — confirm the peer gateway is running with a control listener bound \
122                     on this endpoint"
123                )
124            }
125            Self::Rejected {
126                mob_id,
127                endpoint,
128                code,
129                message,
130            } => {
131                write!(
132                    f,
133                    "remote mob '{mob_id}' control channel ({endpoint}) rejected request \
134                     [{code}]: {message}"
135                )
136            }
137            Self::Encode { endpoint, message } => {
138                write!(f, "control request encode failed for {endpoint}: {message}")
139            }
140            Self::Decode { endpoint, message } => {
141                write!(
142                    f,
143                    "control response decode failed for {endpoint}: {message}"
144                )
145            }
146        }
147    }
148}
149
150impl std::error::Error for RemoteMobError {}
151
152/// Endpoint address for the peer mob's control channel.
153#[derive(Debug, Clone, PartialEq, Eq)]
154pub enum RemoteEndpoint {
155    /// TCP `host:port`.
156    Tcp(String),
157    /// Unix-domain-socket path.
158    Uds(String),
159}
160
161impl RemoteEndpoint {
162    /// Return the comms-style transport scheme (`tcp` or `uds`).
163    pub fn scheme(&self) -> &'static str {
164        match self {
165            Self::Tcp(_) => "tcp",
166            Self::Uds(_) => "uds",
167        }
168    }
169
170    /// Return a comms-style address for this endpoint
171    /// (`tcp://host:port` or `uds:///path`).
172    pub fn comms_address(&self) -> String {
173        match self {
174            Self::Tcp(addr) => format!("tcp://{addr}"),
175            Self::Uds(path) => format!("uds://{path}"),
176        }
177    }
178
179    /// Bare endpoint string (without scheme) — `host:port` or `/path`.
180    pub fn raw(&self) -> &str {
181        match self {
182            Self::Tcp(s) | Self::Uds(s) => s.as_str(),
183        }
184    }
185}
186
187/// A proxy for a mob that lives in a different process.
188///
189/// Phase 1: a structural placeholder. The methods are stubs that return
190/// [`RemoteMobError::ControlChannelUnavailable`] so call sites can be
191/// written against the final shape; Phase 2 will replace these with a
192/// real control-channel client.
193#[derive(Debug, Clone)]
194pub struct RemoteMobProxy {
195    mob_id: String,
196    endpoint: RemoteEndpoint,
197}
198
199impl RemoteMobProxy {
200    /// Build a proxy from a contact-directory entry. Returns `None` for
201    /// `Inproc` entries — those are served by an `Arc<MobHandle>` via
202    /// `LocalOrRemote::Local` instead.
203    pub fn from_entry(entry: &ContactEntry) -> Result<Option<Self>, RemoteMobError> {
204        match &entry.transport {
205            MobTransport::Inproc => Ok(None),
206            MobTransport::Tcp(addr) => Ok(Some(Self {
207                mob_id: entry.mob_id.clone(),
208                endpoint: RemoteEndpoint::Tcp(addr.clone()),
209            })),
210            MobTransport::Uds(path) => Ok(Some(Self {
211                mob_id: entry.mob_id.clone(),
212                endpoint: RemoteEndpoint::Uds(path.clone()),
213            })),
214        }
215    }
216
217    /// The peer mob's identifier.
218    pub fn mob_id(&self) -> &str {
219        &self.mob_id
220    }
221
222    /// The peer mob's control endpoint.
223    pub fn endpoint(&self) -> &RemoteEndpoint {
224        &self.endpoint
225    }
226
227    /// Wire a peer on the remote side.
228    ///
229    /// Sends a `Wire` control request to the peer gateway over its
230    /// configured TCP/UDS endpoint. The remote side calls
231    /// `MobHandle::wire(local_member, PeerTarget::External(spec))` and
232    /// responds with `Ok` or `Err`.
233    pub async fn wire_remote(
234        &self,
235        remote_member: &str,
236        local_peer_spec_address: &str,
237        local_comms_name: &str,
238        local_peer_id: &str,
239        local_pubkey_b64: Option<String>,
240    ) -> Result<(), RemoteMobError> {
241        let request = super::cross_mob_control::ControlRequest::Wire {
242            remote_member: remote_member.to_string(),
243            local_peer_spec_address: local_peer_spec_address.to_string(),
244            local_comms_name: local_comms_name.to_string(),
245            local_peer_id: local_peer_id.to_string(),
246            local_pubkey_b64,
247        };
248        self.dispatch_no_payload(request, "wire").await
249    }
250
251    /// Unwire a peer on the remote side. Symmetric with [`Self::wire_remote`].
252    pub async fn unwire_remote(
253        &self,
254        remote_member: &str,
255        local_peer_spec_address: &str,
256        local_comms_name: &str,
257        local_peer_id: &str,
258        local_pubkey_b64: Option<String>,
259    ) -> Result<(), RemoteMobError> {
260        let request = super::cross_mob_control::ControlRequest::Unwire {
261            remote_member: remote_member.to_string(),
262            local_peer_spec_address: local_peer_spec_address.to_string(),
263            local_comms_name: local_comms_name.to_string(),
264            local_peer_id: local_peer_id.to_string(),
265            local_pubkey_b64,
266        };
267        self.dispatch_no_payload(request, "unwire").await
268    }
269
270    /// Inject an external-turn message into a remote member's session.
271    ///
272    /// Returns the bridge session id that accepted the injection, so the
273    /// caller can correlate downstream events. Mirrors the local
274    /// `MobHandle::send` shape used by `send_cross_mob`.
275    pub async fn inject_message(
276        &self,
277        remote_member: &str,
278        content_json: serde_json::Value,
279    ) -> Result<String, RemoteMobError> {
280        let request = super::cross_mob_control::ControlRequest::Inject {
281            remote_member: remote_member.to_string(),
282            content: content_json,
283        };
284        let response = super::cross_mob_control::RemoteControlClient::send(
285            &self.endpoint,
286            &request,
287            super::cross_mob_control::DEFAULT_CONTROL_TIMEOUT,
288        )
289        .await
290        .map_err(|err| self.attach_mob_id(err))?;
291        match response {
292            super::cross_mob_control::ControlResponse::Injected { session_id } => Ok(session_id),
293            super::cross_mob_control::ControlResponse::Err { code, message } => {
294                Err(RemoteMobError::Rejected {
295                    mob_id: self.mob_id.clone(),
296                    endpoint: self.endpoint.comms_address(),
297                    code,
298                    message,
299                })
300            }
301            other => Err(RemoteMobError::Decode {
302                endpoint: self.endpoint.comms_address(),
303                message: format!("expected Injected response, got {other:?}"),
304            }),
305        }
306    }
307
308    async fn dispatch_no_payload(
309        &self,
310        request: super::cross_mob_control::ControlRequest,
311        operation: &'static str,
312    ) -> Result<(), RemoteMobError> {
313        let response = super::cross_mob_control::RemoteControlClient::send(
314            &self.endpoint,
315            &request,
316            super::cross_mob_control::DEFAULT_CONTROL_TIMEOUT,
317        )
318        .await
319        .map_err(|err| self.attach_mob_id(err))?;
320        match response {
321            super::cross_mob_control::ControlResponse::Ok => Ok(()),
322            super::cross_mob_control::ControlResponse::Err { code, message } => {
323                Err(RemoteMobError::Rejected {
324                    mob_id: self.mob_id.clone(),
325                    endpoint: self.endpoint.comms_address(),
326                    code,
327                    message,
328                })
329            }
330            other => Err(RemoteMobError::Decode {
331                endpoint: self.endpoint.comms_address(),
332                message: format!("expected Ok for {operation}, got {other:?}"),
333            }),
334        }
335    }
336
337    /// Look up a remote member's peer info via the control channel.
338    /// Used during `wire_cross_mob` to discover what `TrustedPeerDescriptor`
339    /// to build for the local-side wire without caller-supplied bookkeeping.
340    pub async fn lookup_member(
341        &self,
342        remote_member: &str,
343    ) -> Result<(String, String), RemoteMobError> {
344        let request = super::cross_mob_control::ControlRequest::LookupMember {
345            remote_member: remote_member.to_string(),
346        };
347        let response = super::cross_mob_control::RemoteControlClient::send(
348            &self.endpoint,
349            &request,
350            super::cross_mob_control::DEFAULT_CONTROL_TIMEOUT,
351        )
352        .await
353        .map_err(|err| self.attach_mob_id(err))?;
354        match response {
355            super::cross_mob_control::ControlResponse::Member {
356                peer_id,
357                comms_name,
358            } => Ok((peer_id, comms_name)),
359            super::cross_mob_control::ControlResponse::Err { code, message } => {
360                Err(RemoteMobError::Rejected {
361                    mob_id: self.mob_id.clone(),
362                    endpoint: self.endpoint.comms_address(),
363                    code,
364                    message,
365                })
366            }
367            other => Err(RemoteMobError::Decode {
368                endpoint: self.endpoint.comms_address(),
369                message: format!("expected Member response, got {other:?}"),
370            }),
371        }
372    }
373
374    fn attach_mob_id(&self, err: RemoteMobError) -> RemoteMobError {
375        match err {
376            RemoteMobError::ControlChannelUnavailable {
377                mob_id,
378                endpoint,
379                operation,
380            } if mob_id.is_empty() || mob_id != self.mob_id => {
381                RemoteMobError::ControlChannelUnavailable {
382                    mob_id: self.mob_id.clone(),
383                    endpoint,
384                    operation,
385                }
386            }
387            other => other,
388        }
389    }
390}
391
392#[cfg(test)]
393#[allow(clippy::unwrap_used, clippy::expect_used)]
394mod tests {
395    use super::*;
396
397    #[test]
398    fn from_entry_inproc_returns_none() {
399        let entry = ContactEntry {
400            mob_id: "demo".to_string(),
401            transport: MobTransport::Inproc,
402            pubkey: None,
403        };
404        let proxy = RemoteMobProxy::from_entry(&entry).expect("inproc is supported");
405        assert!(proxy.is_none());
406    }
407
408    #[test]
409    fn from_entry_tcp_round_trip() {
410        let entry = ContactEntry {
411            mob_id: "remote".to_string(),
412            transport: MobTransport::Tcp("127.0.0.1:9001".to_string()),
413            pubkey: None,
414        };
415        let proxy = RemoteMobProxy::from_entry(&entry)
416            .expect("tcp is supported")
417            .expect("tcp returns Some(proxy)");
418        assert_eq!(proxy.mob_id(), "remote");
419        assert_eq!(proxy.endpoint().scheme(), "tcp");
420        assert_eq!(proxy.endpoint().raw(), "127.0.0.1:9001");
421        assert_eq!(proxy.endpoint().comms_address(), "tcp://127.0.0.1:9001");
422    }
423
424    #[test]
425    fn from_entry_uds_round_trip() {
426        let entry = ContactEntry {
427            mob_id: "remote-uds".to_string(),
428            transport: MobTransport::Uds("/tmp/cross-mob.sock".to_string()),
429            pubkey: None,
430        };
431        let proxy = RemoteMobProxy::from_entry(&entry)
432            .expect("uds is supported")
433            .expect("uds returns Some(proxy)");
434        assert_eq!(proxy.endpoint().scheme(), "uds");
435        assert_eq!(proxy.endpoint().raw(), "/tmp/cross-mob.sock");
436        assert_eq!(
437            proxy.endpoint().comms_address(),
438            "uds:///tmp/cross-mob.sock"
439        );
440    }
441
442    /// When no listener is bound at the configured endpoint, `wire_remote`
443    /// surfaces `ControlChannelUnavailable` with the mob id attached.
444    /// (End-to-end happy path lives in `tests/cross_mob_tcp.rs` once
445    /// the unified runtime spins up its own control listener.)
446    #[tokio::test]
447    async fn wire_remote_returns_unavailable_when_no_listener() {
448        let entry = ContactEntry {
449            mob_id: "remote".to_string(),
450            transport: MobTransport::Tcp("127.0.0.1:1".to_string()),
451            pubkey: None,
452        };
453        let proxy = RemoteMobProxy::from_entry(&entry)
454            .expect("tcp ok")
455            .expect("some");
456        let err = proxy
457            .wire_remote(
458                "alice",
459                "tcp://127.0.0.1:9000",
460                "demo/role/alice",
461                "00000000-0000-4000-8000-000000000001",
462                None,
463            )
464            .await
465            .expect_err("no listener");
466        assert!(
467            matches!(err, RemoteMobError::ControlChannelUnavailable { ref mob_id, .. } if mob_id == "remote"),
468            "got {err:?}"
469        );
470    }
471}
meerkat_mobkit/runtime/cross_mob_remote.rs

meerkat_mobkit/runtime/
cross_mob_remote.rs
