instance awareness: device_announce carries hosted_agents
The full server's bridge now announces which agent pubkeys this box hosts (read fresh per reconnect), same list the lite listener already computed — loader moved to federation/mod.rs and shared. DeviceRegistry records it (serde default keeps old known_peers.json parsing) and souveraine peers shows a hosts: line. Durable memfs-side instance log still to come.
This commit is contained in:
parent
a62379ad8a
commit
8e540c773d
6 changed files with 108 additions and 21 deletions
|
|
@ -878,6 +878,14 @@ async fn run_peers(json: bool) -> anyhow::Result<()> {
|
|||
}
|
||||
println!(" url: {}", p.url);
|
||||
println!(" seen: {}", p.last_seen.format("%Y-%m-%d %H:%M:%S UTC"));
|
||||
if !p.hosted_agents.is_empty() {
|
||||
let short: Vec<&str> = p
|
||||
.hosted_agents
|
||||
.iter()
|
||||
.map(|a| a.get(..12).unwrap_or(a))
|
||||
.collect();
|
||||
println!(" hosts: {}", short.join(", "));
|
||||
}
|
||||
}
|
||||
println!();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -28,6 +28,10 @@ pub struct PeerEntry {
|
|||
pub last_seen: DateTime<Utc>,
|
||||
/// Is the peer considered alive?
|
||||
pub alive: bool,
|
||||
/// Agent pubkeys (hex) the peer hosts, from its announce — the live half
|
||||
/// of instance awareness ("which agents live where").
|
||||
#[serde(default)]
|
||||
pub hosted_agents: Vec<String>,
|
||||
}
|
||||
|
||||
/// Tracks known federated peers for the local instance. Written to a JSON file
|
||||
|
|
@ -98,6 +102,17 @@ impl DeviceRegistry {
|
|||
.and_then(|v| v.as_str())
|
||||
.unwrap_or("hearth")
|
||||
.to_string();
|
||||
let hosted_agents = event
|
||||
.payload
|
||||
.as_ref()
|
||||
.and_then(|p| p.get("hosted_agents"))
|
||||
.and_then(|v| v.as_array())
|
||||
.map(|a| {
|
||||
a.iter()
|
||||
.filter_map(|v| v.as_str().map(String::from))
|
||||
.collect()
|
||||
})
|
||||
.unwrap_or_default();
|
||||
// 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 {
|
||||
|
|
@ -125,6 +140,7 @@ impl DeviceRegistry {
|
|||
first_seen,
|
||||
last_seen: now,
|
||||
alive: true,
|
||||
hosted_agents,
|
||||
},
|
||||
);
|
||||
self.persist();
|
||||
|
|
@ -182,3 +198,47 @@ impl DeviceRegistry {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn announce_records_hosted_agents() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let registry = DeviceRegistry::new(
|
||||
dir.path().to_path_buf(),
|
||||
"local-seed".into(),
|
||||
FederationRole::Hearth,
|
||||
);
|
||||
let event = SensorEvent {
|
||||
sensor_name: "federation".into(),
|
||||
timestamp: Utc::now(),
|
||||
event_type: "device_announce".into(),
|
||||
target: None,
|
||||
urgency: 0.0,
|
||||
payload: Some(serde_json::json!({
|
||||
"federation_url": "ws://peer:7777",
|
||||
"role": "limb",
|
||||
"hosted_agents": ["aabbcc", "ddeeff"],
|
||||
})),
|
||||
seed_id: Some("peer-seed".into()),
|
||||
reply_to: None,
|
||||
};
|
||||
assert!(registry.handle_event(&event));
|
||||
let peers = registry.list();
|
||||
assert_eq!(peers.len(), 1);
|
||||
assert_eq!(peers[0].hosted_agents, vec!["aabbcc", "ddeeff"]);
|
||||
|
||||
// Old announces without the field must still parse — empty roster.
|
||||
let bare = SensorEvent {
|
||||
payload: Some(serde_json::json!({ "federation_url": "ws://old:7777" })),
|
||||
seed_id: Some("old-peer".into()),
|
||||
..event
|
||||
};
|
||||
registry.handle_event(&bare);
|
||||
let peers = registry.list();
|
||||
let old = peers.iter().find(|p| p.seed_id == "old-peer").unwrap();
|
||||
assert!(old.hosted_agents.is_empty());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,15 +21,23 @@ pub struct FederationBridge {
|
|||
event_bus: EventBus,
|
||||
signer: Arc<MachineSigner>,
|
||||
role: FederationRole,
|
||||
/// `~/.souveraine` — announces read hosted-agent seeds beneath it.
|
||||
base: std::path::PathBuf,
|
||||
peers: Vec<PeerConfig>,
|
||||
}
|
||||
|
||||
impl FederationBridge {
|
||||
pub fn new(event_bus: EventBus, signer: Arc<MachineSigner>, role: FederationRole) -> Self {
|
||||
pub fn new(
|
||||
event_bus: EventBus,
|
||||
signer: Arc<MachineSigner>,
|
||||
role: FederationRole,
|
||||
base: std::path::PathBuf,
|
||||
) -> Self {
|
||||
Self {
|
||||
event_bus,
|
||||
signer,
|
||||
role,
|
||||
base,
|
||||
peers: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
|
@ -48,6 +56,7 @@ impl FederationBridge {
|
|||
self.event_bus.clone(),
|
||||
self.signer.clone(),
|
||||
self.role,
|
||||
self.base.clone(),
|
||||
));
|
||||
}
|
||||
}
|
||||
|
|
@ -60,6 +69,7 @@ async fn peer_outbound_task(
|
|||
event_bus: EventBus,
|
||||
signer: Arc<MachineSigner>,
|
||||
role: FederationRole,
|
||||
base: std::path::PathBuf,
|
||||
) {
|
||||
let endpoint = federation_endpoint(&peer.url);
|
||||
let mut retry: u32 = 0;
|
||||
|
|
@ -70,7 +80,9 @@ async fn peer_outbound_task(
|
|||
retry = 0;
|
||||
tracing::info!(peer = %endpoint, "federation: outbound connected");
|
||||
|
||||
// Announce our presence to the peer.
|
||||
// Announce our presence to the peer. Hosted agents are read
|
||||
// fresh on every (re)connect — agents created since the last
|
||||
// announce are picked up on the next reconnect cycle.
|
||||
let announce = SensorEvent {
|
||||
sensor_name: "federation".into(),
|
||||
timestamp: chrono::Utc::now(),
|
||||
|
|
@ -82,6 +94,7 @@ async fn peer_outbound_task(
|
|||
"label": None::<String>,
|
||||
"pubkey": signer.pubkey_hex(),
|
||||
"role": role.as_str(),
|
||||
"hosted_agents": super::load_hosted_agent_pubkeys(&base),
|
||||
})),
|
||||
seed_id: None,
|
||||
reply_to: None,
|
||||
|
|
|
|||
|
|
@ -15,3 +15,24 @@ pub mod types;
|
|||
|
||||
pub use bridge::FederationBridge;
|
||||
pub use types::SignedEvent;
|
||||
|
||||
/// Read the agent pubkeys this device hosts from
|
||||
/// `{base}/server/agents/*/seed/public.key`. Cheap — a handful of 32-byte
|
||||
/// files, no DB, no memfs load. Announced to peers so the federation knows
|
||||
/// which agents live where (instance awareness, FEDERATION.md).
|
||||
pub fn load_hosted_agent_pubkeys(base: &std::path::Path) -> Vec<String> {
|
||||
let agents_dir = base.join("server").join("agents");
|
||||
let mut out = Vec::new();
|
||||
if let Ok(entries) = std::fs::read_dir(&agents_dir) {
|
||||
for entry in entries.flatten() {
|
||||
let pk = entry.path().join("seed").join("public.key");
|
||||
if let Ok(bytes) = std::fs::read(&pk) {
|
||||
if bytes.len() == 32 {
|
||||
out.push(hex::encode(bytes));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
out.sort();
|
||||
out
|
||||
}
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ use crate::core::config::FederationConfig;
|
|||
use crate::core::identity::verify_summon;
|
||||
use crate::core::nervous::{EventBus, SensorEvent};
|
||||
use crate::machined::signer::MachineSigner;
|
||||
use crate::server::federation::{FederationBridge, SignedEvent};
|
||||
use crate::server::federation::{load_hosted_agent_pubkeys, FederationBridge, SignedEvent};
|
||||
|
||||
/// Shared state for the lite route set — only what the minimal handlers and
|
||||
/// the wake watcher need.
|
||||
|
|
@ -39,23 +39,6 @@ pub struct LiteListener {
|
|||
pub auto_wake: bool,
|
||||
}
|
||||
|
||||
/// Read the agent pubkeys this device hosts from `server/agents/*/seed/public.key`.
|
||||
/// Cheap enough for the lite path — a handful of 32-byte files, no DB.
|
||||
fn load_hosted_agent_pubkeys(base: &Path) -> Vec<String> {
|
||||
let agents_dir = base.join("server").join("agents");
|
||||
let mut out = Vec::new();
|
||||
if let Ok(entries) = std::fs::read_dir(&agents_dir) {
|
||||
for entry in entries.flatten() {
|
||||
let pk = entry.path().join("seed").join("public.key");
|
||||
if let Ok(bytes) = std::fs::read(&pk) {
|
||||
if bytes.len() == 32 {
|
||||
out.push(hex::encode(bytes));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
impl LiteListener {
|
||||
/// Build the listener: load the seed identity, start the federation
|
||||
|
|
@ -77,7 +60,8 @@ 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(), signer, config.role);
|
||||
let mut bridge =
|
||||
FederationBridge::new(event_bus.clone(), signer, config.role, base.clone());
|
||||
for peer in &config.peers {
|
||||
bridge.add_peer(peer.clone());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -402,6 +402,7 @@ impl SouveraineServer {
|
|||
self.event_bus.clone(),
|
||||
signer.clone(),
|
||||
fed.role,
|
||||
dirs::home_dir().unwrap_or_default().join(".souveraine"),
|
||||
);
|
||||
for peer in fed.peers {
|
||||
bridge.add_peer(peer);
|
||||
|
|
|
|||
Loading…
Reference in a new issue