From 3629fa8b78688afa24187c35b50f0442cf3c24d5 Mon Sep 17 00:00:00 2001 From: Fimeg Date: Fri, 15 May 2026 16:01:21 -0400 Subject: [PATCH] =?UTF-8?q?feat(federation):=20Phase=208=20=E2=80=94=20hea?= =?UTF-8?q?rth=20&=20limb=20role=20model?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FederationRole (hearth/limb) in config; bridge carries role in device_announce; DeviceRegistry tracks role per peer with a split-brain guard (two hearths → ERROR!). Role defaults to hearth: a standalone machine is its own home. --- src/core/config.rs | 30 ++++++++++++++++++++++++++++++ src/server/device_registry.rs | 32 +++++++++++++++++++++++++++++++- src/server/federation/bridge.rs | 15 ++++++++++++--- src/server/listener.rs | 2 +- src/server/mod.rs | 4 +++- 5 files changed, 77 insertions(+), 6 deletions(-) diff --git a/src/core/config.rs b/src/core/config.rs index 3d0531d..6aac39f 100644 --- a/src/core/config.rs +++ b/src/core/config.rs @@ -662,12 +662,41 @@ fn default_stale_timeout_secs() -> u64 { 90 } // ── Federation ── +/// A machine's role in the federation. +/// +/// - `Hearth` — where the agent lives. Full engine always up, the autonomous +/// rhythm runs here, the memfs HEAD is authoritative. One per agent. +/// - `Limb` — a place she can reach to. At rest it's the lite listener; an +/// authorized summon wakes the full engine, it acts, reports up, and rests. +/// A limb does not run the general autonomous rhythm — that keeps one +/// heartbeat at the hearth and no split-brain across machines. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +#[serde(rename_all = "lowercase")] +pub enum FederationRole { + #[default] + Hearth, + Limb, +} + +impl FederationRole { + pub fn as_str(&self) -> &'static str { + match self { + Self::Hearth => "hearth", + Self::Limb => "limb", + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct FederationConfig { #[serde(default)] pub enabled: bool, #[serde(default)] pub instance_label: Option, + /// This machine's role — hearth (the agent's home) or limb (a place she + /// reaches to). Defaults to hearth: a standalone machine is its own home. + #[serde(default)] + pub role: FederationRole, /// Peers this instance federates with. Each peer connection is a signed /// WS stream to the peer's federation endpoint. #[serde(default)] @@ -690,6 +719,7 @@ impl Default for FederationConfig { Self { enabled: false, instance_label: None, + role: FederationRole::Hearth, peers: Vec::new(), authorized_summoners: Vec::new(), auto_wake: false, diff --git a/src/server/device_registry.rs b/src/server/device_registry.rs index eece7a2..f433b9e 100644 --- a/src/server/device_registry.rs +++ b/src/server/device_registry.rs @@ -1,3 +1,4 @@ +use crate::core::config::FederationRole; use crate::core::nervous::SensorEvent; use chrono::{DateTime, Utc}; use dashmap::DashMap; @@ -5,6 +6,10 @@ use serde::{Deserialize, Serialize}; use std::path::PathBuf; use std::sync::Arc; +fn default_role() -> String { + "hearth".to_string() +} + /// A peer device known to this federation. Updated on every `device_announce` /// and `device_leave` event from the peer's bridge. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -15,6 +20,9 @@ pub struct PeerEntry { pub label: Option, /// The peer's federation endpoint (ws://host:port). pub url: String, + /// The peer's federation role — "hearth" or "limb". + #[serde(default = "default_role")] + pub role: String, /// First time we saw this peer's announce. pub first_seen: DateTime, /// Most recent announce. @@ -30,12 +38,14 @@ pub struct DeviceRegistry { known_peers_path: PathBuf, /// This instance's own seed_id (pubkey hex) — filters self-announcements. local_seed_id: String, + /// This machine's role — used to detect a hearth/hearth split-brain. + local_role: FederationRole, } impl DeviceRegistry { /// `base_dir` = `~/.souveraine/`. The registry writes to /// `{base_dir}/federation/known_peers.json`. - pub fn new(base_dir: PathBuf, local_seed_id: String) -> Self { + pub fn new(base_dir: PathBuf, local_seed_id: String, local_role: FederationRole) -> Self { let fed_dir = base_dir.join("federation"); let known_peers_path = fed_dir.join("known_peers.json"); let peers = DashMap::new(); @@ -53,6 +63,7 @@ impl DeviceRegistry { peers, known_peers_path, local_seed_id, + local_role, } } @@ -81,6 +92,24 @@ impl DeviceRegistry { .and_then(|p| p.get("label")) .and_then(|v| v.as_str()) .map(|s| s.to_string()); + let role = event + .payload + .as_ref() + .and_then(|p| p.get("role")) + .and_then(|v| v.as_str()) + .unwrap_or("hearth") + .to_string(); + // Split-brain guard: two hearths for one agent diverge memory. + // We refuse to auto-resolve — surface it loudly for the human. + if role == "hearth" && self.local_role == FederationRole::Hearth { + tracing::error!( + peer = %seed_id, + "federation: HEARTH CONFLICT — this machine and {} both \ + claim hearth. Only one machine should be the hearth; \ + set [federation].role = \"limb\" on one of them.", + seed_id, + ); + } let now = Utc::now(); // Preserve the original first_seen across re-announces. let first_seen = self.peers.get(&seed_id) @@ -93,6 +122,7 @@ impl DeviceRegistry { seed_id: seed_id.clone(), label, url, + role, first_seen, last_seen: now, alive: true, diff --git a/src/server/federation/bridge.rs b/src/server/federation/bridge.rs index 3b7b175..54b7d5b 100644 --- a/src/server/federation/bridge.rs +++ b/src/server/federation/bridge.rs @@ -6,7 +6,7 @@ use std::time::Duration; use futures::SinkExt; use tokio_tungstenite::tungstenite::Message; -use crate::core::config::PeerConfig; +use crate::core::config::{FederationRole, PeerConfig}; use crate::core::identity::SeedId; use crate::core::nervous::{EventBus, SensorEvent}; @@ -20,14 +20,16 @@ use super::types::SignedEvent; pub struct FederationBridge { event_bus: EventBus, seed: Arc, + role: FederationRole, peers: Vec, } impl FederationBridge { - pub fn new(event_bus: EventBus, seed: Arc) -> Self { + pub fn new(event_bus: EventBus, seed: Arc, role: FederationRole) -> Self { Self { event_bus, seed, + role, peers: Vec::new(), } } @@ -45,6 +47,7 @@ impl FederationBridge { peer, self.event_bus.clone(), self.seed.clone(), + self.role, )); } } @@ -52,7 +55,12 @@ impl FederationBridge { /// Connect to one peer and forward signed events, reconnecting with /// exponential backoff whenever the link drops. -async fn peer_outbound_task(peer: PeerConfig, event_bus: EventBus, seed: Arc) { +async fn peer_outbound_task( + peer: PeerConfig, + event_bus: EventBus, + seed: Arc, + role: FederationRole, +) { let endpoint = federation_endpoint(&peer.url); let mut retry: u32 = 0; @@ -73,6 +81,7 @@ async fn peer_outbound_task(peer: PeerConfig, event_bus: EventBus, seed: Arc, "pubkey": seed.public_key_hex(), + "role": role.as_str(), })), seed_id: None, reply_to: None, diff --git a/src/server/listener.rs b/src/server/listener.rs index 9b69b87..5fb2f89 100644 --- a/src/server/listener.rs +++ b/src/server/listener.rs @@ -74,7 +74,7 @@ impl LiteListener { // Outbound federation bridge — so this instance can also reach peers. if !config.peers.is_empty() { - let mut bridge = FederationBridge::new(event_bus.clone(), Arc::new(seed)); + let mut bridge = FederationBridge::new(event_bus.clone(), Arc::new(seed), config.role); for peer in &config.peers { bridge.add_peer(peer.clone()); } diff --git a/src/server/mod.rs b/src/server/mod.rs index a995a4a..b622ea6 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -220,8 +220,9 @@ impl SouveraineServer { Err(_) => None, }; let sb_for_device_reg = souveraine_base.clone(); + let local_role = config.federation.role; let device_registry = local_seed_id.clone().map(|seed_id| { - let reg = Arc::new(DeviceRegistry::new(sb_for_device_reg, seed_id)); + let reg = Arc::new(DeviceRegistry::new(sb_for_device_reg, seed_id, local_role)); // Subscribe the registry to the event bus for live updates. let reg_clone = reg.clone(); let mut rx = event_bus.subscribe(); @@ -311,6 +312,7 @@ impl SouveraineServer { let mut bridge = federation::FederationBridge::new( self.event_bus.clone(), Arc::new(seed), + fed.role, ); for peer in fed.peers { bridge.add_peer(peer);