merge: reuse accepted AI provider setup in publishing worktree
This commit is contained in:
@@ -352,20 +352,19 @@ impl RpcHandler {
|
||||
(Some(u), Some(t)) if t > 0 => {
|
||||
serde_json::json!((u as f64 / t as f64 * 100.0).round())
|
||||
}
|
||||
_ => serde_json::json!(0),
|
||||
_ => serde_json::Value::Null,
|
||||
}
|
||||
};
|
||||
let apps = state.map(|s| s.apps.as_slice()).unwrap_or(&[]);
|
||||
let reported_at = state
|
||||
.map(|s| s.timestamp.clone())
|
||||
.or_else(|| n.last_seen.clone())
|
||||
.unwrap_or_else(|| n.added_at.clone());
|
||||
.or_else(|| n.last_seen.clone());
|
||||
|
||||
let mut report = serde_json::json!({
|
||||
"node_id": n.did,
|
||||
"node_name": state.and_then(|s| s.node_name.clone()).or_else(|| n.name.clone()),
|
||||
"uptime_secs": state.and_then(|s| s.uptime_secs).unwrap_or(0),
|
||||
"cpu_pct": state.and_then(|s| s.cpu_usage_percent).map(|v| v.round()).unwrap_or(0.0),
|
||||
"uptime_secs": state.and_then(|s| s.uptime_secs),
|
||||
"cpu_pct": state.and_then(|s| s.cpu_usage_percent).filter(|v| v.is_finite() && (0.0..=100.0).contains(v)).map(|v| v.round()),
|
||||
"mem_pct": pct(state.and_then(|s| s.mem_used_bytes), state.and_then(|s| s.mem_total_bytes)),
|
||||
"disk_pct": pct(state.and_then(|s| s.disk_used_bytes), state.and_then(|s| s.disk_total_bytes)),
|
||||
"container_count": apps.len(),
|
||||
@@ -561,7 +560,7 @@ fn annotate_fleet_report(report: &mut serde_json::Value) {
|
||||
let is_online = reported
|
||||
.map(|dt| {
|
||||
let age = chrono::Utc::now().signed_duration_since(dt);
|
||||
age.num_minutes() < 30
|
||||
age.num_seconds() >= -60 && age.num_seconds() < 1800
|
||||
})
|
||||
.unwrap_or(false);
|
||||
|
||||
@@ -569,7 +568,9 @@ fn annotate_fleet_report(report: &mut serde_json::Value) {
|
||||
.map(|dt| {
|
||||
let age = chrono::Utc::now().signed_duration_since(dt);
|
||||
let mins = age.num_minutes();
|
||||
if mins < 1 {
|
||||
if age.num_seconds() < -60 {
|
||||
"unknown (clock ahead)".to_string()
|
||||
} else if mins < 1 {
|
||||
"just now".to_string()
|
||||
} else if mins < 60 {
|
||||
format!("{}m ago", mins)
|
||||
|
||||
@@ -602,14 +602,39 @@ impl RpcHandler {
|
||||
None
|
||||
};
|
||||
|
||||
// Reuse the minute collector instead of running expensive probes for
|
||||
// every peer. An absent/stalled collector is unknown, never zero load.
|
||||
let now = chrono::Utc::now().timestamp();
|
||||
let latest = self
|
||||
.metrics_store
|
||||
.latest()
|
||||
.await
|
||||
.filter(|sample| (0..=180).contains(&now.saturating_sub(sample.timestamp)));
|
||||
let metrics = latest.as_ref().map(|sample| &sample.system);
|
||||
let uptime = tokio::fs::read_to_string("/proc/uptime")
|
||||
.await
|
||||
.ok()
|
||||
.and_then(|s| s.split_whitespace().next()?.parse::<f64>().ok())
|
||||
.filter(|v| v.is_finite() && *v >= 0.0)
|
||||
.map(|v| v as u64);
|
||||
let state = federation::build_local_state(
|
||||
apps,
|
||||
0.0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
metrics
|
||||
.map(|m| m.cpu_percent)
|
||||
.filter(|v| v.is_finite() && (0.0..=100.0).contains(v)),
|
||||
metrics
|
||||
.filter(|m| m.mem_total_bytes > 0)
|
||||
.map(|m| m.mem_used_bytes),
|
||||
metrics
|
||||
.filter(|m| m.mem_total_bytes > 0)
|
||||
.map(|m| m.mem_total_bytes),
|
||||
metrics
|
||||
.filter(|m| m.disk_total_bytes > 0)
|
||||
.map(|m| m.disk_used_bytes),
|
||||
metrics
|
||||
.filter(|m| m.disk_total_bytes > 0)
|
||||
.map(|m| m.disk_total_bytes),
|
||||
uptime,
|
||||
tor_active,
|
||||
server_name,
|
||||
nostr_npub,
|
||||
@@ -1254,76 +1279,120 @@ impl RpcHandler {
|
||||
);
|
||||
}
|
||||
|
||||
let reply = self.prepare_peer_approval_reply(&req).await?;
|
||||
// Persist the operator decision before transport. A relay outage must
|
||||
// not require another approval or lose the already-authorized reply.
|
||||
pending::decide(&self.config.data_dir, id, pending::PendingState::Approved).await?;
|
||||
let delivered = self.deliver_peer_approval_reply(&reply).await?;
|
||||
Ok(serde_json::json!({ "approved": true, "id": id, "delivery_pending": !delivered }))
|
||||
}
|
||||
|
||||
async fn prepare_peer_approval_reply(
|
||||
&self,
|
||||
req: &pending::PendingPeerRequest,
|
||||
) -> Result<federation::handshake_delivery::ApprovalReply> {
|
||||
use federation::handshake_delivery::{self, ApprovalReply};
|
||||
if let Some(reply) = handshake_delivery::find(&self.config.data_dir, &req.id).await? {
|
||||
anyhow::ensure!(
|
||||
reply.recipient == req.from_nostr_pubkey && reply.expected_did == req.from_did,
|
||||
"Approval recipient changed"
|
||||
);
|
||||
return Ok(reply);
|
||||
}
|
||||
let (data, _) = self.state_manager.get_snapshot().await;
|
||||
let local_did = identity::did_key_from_pubkey_hex(&data.server_info.pubkey)?;
|
||||
let local_onion = data
|
||||
.server_info
|
||||
.tor_address
|
||||
.clone()
|
||||
.as_deref()
|
||||
.ok_or_else(|| anyhow::anyhow!("Tor address not available"))?;
|
||||
let local_pubkey = data.server_info.pubkey.clone();
|
||||
|
||||
// Generate a one-shot federation invite. The code embeds OUR onion
|
||||
// and OUR pubkey, but it leaves this box only inside the NIP-44
|
||||
// ciphertext below.
|
||||
let identity_dir = self.config.data_dir.join("identity");
|
||||
let local_fips_npub = identity::fips_npub(&identity_dir).await.unwrap_or(None);
|
||||
// Discovery/connection-request approvals admit the requester as
|
||||
// Observer — the invite itself now carries that level, so both
|
||||
// sides converge on Observer without post-hoc demotion.
|
||||
let local_fips_npub = identity::fips_npub(&self.config.data_dir.join("identity"))
|
||||
.await
|
||||
.unwrap_or(None);
|
||||
let invite_code = federation::create_invite(
|
||||
&self.config.data_dir,
|
||||
&local_did,
|
||||
&local_onion,
|
||||
&local_pubkey,
|
||||
local_onion,
|
||||
&data.server_info.pubkey,
|
||||
local_fips_npub.as_deref(),
|
||||
TrustLevel::Observer,
|
||||
)
|
||||
.await?;
|
||||
handshake_delivery::stage(
|
||||
&self.config.data_dir,
|
||||
ApprovalReply {
|
||||
request_id: req.id.clone(),
|
||||
recipient: req.from_nostr_pubkey.clone(),
|
||||
expected_did: req.from_did.clone(),
|
||||
invite_code,
|
||||
attempts: 0,
|
||||
next_attempt: 0,
|
||||
},
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
// Pre-add the requester to OUR federation list as Observer so that
|
||||
// when their `federation.peer-joined` callback arrives over Tor we
|
||||
// already trust their pubkey enough to accept the join. Their DID
|
||||
// and pubkey come from the request — we'll cross-check the pubkey
|
||||
// against the eventual peer-joined signature in the existing
|
||||
// verification path (handlers.rs line ~365).
|
||||
if !req.from_did.is_empty() {
|
||||
// We don't know the requester's onion or ed25519 pubkey yet —
|
||||
// they'll send those in the federation.peer-joined callback
|
||||
// after they apply our invite. Until then we can't add a real
|
||||
// FederatedNode entry. We just store the pending row as
|
||||
// Approved so the UI shows progress, and trust the existing
|
||||
// peer-joined handler to admit them as Observer when they call.
|
||||
//
|
||||
// Caveat: peer-joined currently hardcodes TrustLevel::Trusted.
|
||||
// We override that below by demoting on success.
|
||||
debug!(
|
||||
requester_did = %req.from_did,
|
||||
"Approval pending — waiting for federation.peer-joined callback over Tor"
|
||||
);
|
||||
}
|
||||
|
||||
// Encrypt + send the invite over NIP-44 to the requester.
|
||||
let identity_dir = self.config.data_dir.join("identity");
|
||||
nostr_handshake::send_peer_invite(
|
||||
&identity_dir,
|
||||
&req.from_nostr_pubkey,
|
||||
&invite_code,
|
||||
&self.config.nostr_relays,
|
||||
async fn deliver_peer_approval_reply(
|
||||
&self,
|
||||
reply: &federation::handshake_delivery::ApprovalReply,
|
||||
) -> Result<bool> {
|
||||
let Some(claimed) = federation::handshake_delivery::claim(
|
||||
&self.config.data_dir,
|
||||
&reply.request_id,
|
||||
chrono::Utc::now().timestamp(),
|
||||
)
|
||||
.await?
|
||||
else {
|
||||
return Ok(false);
|
||||
};
|
||||
let result = nostr_handshake::send_peer_invite(
|
||||
&self.config.data_dir.join("identity"),
|
||||
&claimed.recipient,
|
||||
&claimed.invite_code,
|
||||
&self.handshake_relays().await,
|
||||
self.config.nostr_tor_proxy.as_deref(),
|
||||
)
|
||||
.await?;
|
||||
.await;
|
||||
if result.is_err() {
|
||||
warn!(request_id = %reply.request_id, "Peer approval delivery deferred; durable retry scheduled");
|
||||
}
|
||||
Ok(result.is_ok())
|
||||
}
|
||||
|
||||
pending::set_state(&self.config.data_dir, id, pending::PendingState::Approved).await?;
|
||||
info!(
|
||||
id = %id,
|
||||
from = %req.from_nostr_pubkey,
|
||||
"Approved peer request and shipped invite over NIP-44"
|
||||
);
|
||||
Ok(serde_json::json!({
|
||||
"approved": true,
|
||||
"id": id,
|
||||
}))
|
||||
/// Recover relay loss and legacy approvals without changing trust or
|
||||
/// resurrecting a node that the operator explicitly removed.
|
||||
pub(in crate::api::rpc) async fn retry_peer_approval_replies(&self) -> Result<()> {
|
||||
let requests = pending::load_pending(&self.config.data_dir).await?;
|
||||
let nodes = federation::load_nodes(&self.config.data_dir).await?;
|
||||
let removed = federation::load_removed_dids(&self.config.data_dir).await?;
|
||||
let cutoff = chrono::Utc::now() - chrono::Duration::days(30);
|
||||
let mut sent = 0;
|
||||
for req in requests {
|
||||
if req.outbound || req.state != pending::PendingState::Approved {
|
||||
federation::handshake_delivery::remove(&self.config.data_dir, &req.id).await?;
|
||||
continue;
|
||||
}
|
||||
let expired = chrono::DateTime::parse_from_rfc3339(&req.received_at)
|
||||
.map(|time| time < cutoff)
|
||||
.unwrap_or(true);
|
||||
if expired
|
||||
|| removed.contains(&req.from_did)
|
||||
|| nodes.iter().any(|node| node.did == req.from_did)
|
||||
{
|
||||
federation::handshake_delivery::remove(&self.config.data_dir, &req.id).await?;
|
||||
continue;
|
||||
}
|
||||
if req.from_did.is_empty() || sent >= 4 {
|
||||
continue;
|
||||
}
|
||||
let reply = self.prepare_peer_approval_reply(&req).await?;
|
||||
if reply.next_attempt > chrono::Utc::now().timestamp() {
|
||||
continue;
|
||||
}
|
||||
self.deliver_peer_approval_reply(&reply).await?;
|
||||
sent += 1;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// federation.reject-request — drop a pending request and, if requested,
|
||||
@@ -1353,19 +1422,19 @@ impl RpcHandler {
|
||||
);
|
||||
}
|
||||
|
||||
pending::decide(&self.config.data_dir, id, pending::PendingState::Rejected).await?;
|
||||
if notify {
|
||||
let identity_dir = self.config.data_dir.join("identity");
|
||||
let _ = nostr_handshake::send_peer_reject(
|
||||
&identity_dir,
|
||||
&req.from_nostr_pubkey,
|
||||
reason,
|
||||
&self.config.nostr_relays,
|
||||
&self.handshake_relays().await,
|
||||
self.config.nostr_tor_proxy.as_deref(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
pending::set_state(&self.config.data_dir, id, pending::PendingState::Rejected).await?;
|
||||
info!(id = %id, from = %req.from_nostr_pubkey, "Rejected peer request");
|
||||
Ok(serde_json::json!({ "rejected": true, "id": id }))
|
||||
}
|
||||
@@ -1410,7 +1479,7 @@ impl RpcHandler {
|
||||
&identity_dir,
|
||||
&req.from_nostr_pubkey,
|
||||
reason,
|
||||
&self.config.nostr_relays,
|
||||
&self.handshake_relays().await,
|
||||
self.config.nostr_tor_proxy.as_deref(),
|
||||
)
|
||||
.await
|
||||
|
||||
@@ -0,0 +1,276 @@
|
||||
//! Exercise encrypted replies through a relay configured in the UI only.
|
||||
use crate::federation::pending::{self, PendingState};
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use nostr_sdk::prelude::{nip44, Event, Keys};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
#[tokio::test]
|
||||
async fn managed_relay_receives_approval_rejection_and_cancellation() {
|
||||
for (operation, accepted) in [
|
||||
("approve", true),
|
||||
("reject", true),
|
||||
("cancel", true),
|
||||
("approve", false),
|
||||
("retry", true),
|
||||
] {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let relay_url = format!("ws://{}", listener.local_addr().unwrap());
|
||||
let relay = tokio::spawn(async move {
|
||||
let (socket, _) = listener.accept().await.unwrap();
|
||||
let mut ws = tokio_tungstenite::accept_async(socket).await.unwrap();
|
||||
while let Some(Ok(message)) = ws.next().await {
|
||||
if !message.is_text() {
|
||||
continue;
|
||||
}
|
||||
let value: serde_json::Value =
|
||||
serde_json::from_str(message.to_text().unwrap()).unwrap();
|
||||
if value[0] != "EVENT" {
|
||||
continue;
|
||||
}
|
||||
let event: Event = serde_json::from_value(value[1].clone()).unwrap();
|
||||
event.verify().unwrap();
|
||||
ws.send(tokio_tungstenite::tungstenite::Message::Text(
|
||||
serde_json::json!([
|
||||
"OK",
|
||||
event.id.to_hex(),
|
||||
accepted,
|
||||
"blocked: fixture rejection"
|
||||
])
|
||||
.to_string(),
|
||||
))
|
||||
.await
|
||||
.unwrap();
|
||||
return event;
|
||||
}
|
||||
panic!("relay closed without a signed event");
|
||||
});
|
||||
let mut config = crate::config::Config::default();
|
||||
config.data_dir = tmp.path().to_path_buf();
|
||||
config.nostr_relays.clear();
|
||||
config.nostr_tor_proxy = None;
|
||||
crate::nostr_relays::save_relays(
|
||||
tmp.path(),
|
||||
&crate::nostr_relays::RelayStore {
|
||||
relays: vec![crate::nostr_relays::RelayConfig {
|
||||
url: relay_url,
|
||||
enabled: true,
|
||||
added_at: chrono::Utc::now().to_rfc3339(),
|
||||
}],
|
||||
},
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let sender = Keys::parse(&"11".repeat(32)).unwrap();
|
||||
let recipient = Keys::parse(&"22".repeat(32)).unwrap();
|
||||
let identity_dir = tmp.path().join("identity");
|
||||
tokio::fs::create_dir_all(&identity_dir).await.unwrap();
|
||||
tokio::fs::write(identity_dir.join("nostr_secret"), "11".repeat(32))
|
||||
.await
|
||||
.unwrap();
|
||||
tokio::fs::write(identity_dir.join("node_key"), [0x33; 32])
|
||||
.await
|
||||
.unwrap();
|
||||
let state = Arc::new(crate::state::StateManager::new());
|
||||
state
|
||||
.mutate_data(|data| {
|
||||
data.server_info.pubkey = "33".repeat(32);
|
||||
data.server_info.tor_address = Some(format!("{}.onion", "a".repeat(56)));
|
||||
})
|
||||
.await;
|
||||
let handler = crate::api::rpc::RpcHandler::new(
|
||||
config,
|
||||
state,
|
||||
Arc::new(crate::monitoring::MetricsStore::new()),
|
||||
crate::session::SessionStore::new_for_tests(tmp.path().join("sessions.json")),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let row = if operation == "cancel" {
|
||||
pending::insert_outbound(
|
||||
tmp.path(),
|
||||
recipient.public_key().to_hex(),
|
||||
String::new(),
|
||||
crate::identity::did_key_from_pubkey_hex(&"44".repeat(32)).unwrap(),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
} else {
|
||||
pending::insert_inbound(
|
||||
tmp.path(),
|
||||
recipient.public_key().to_hex(),
|
||||
String::new(),
|
||||
crate::identity::did_key_from_pubkey_hex(&"44".repeat(32)).unwrap(),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap()
|
||||
};
|
||||
if operation == "retry" {
|
||||
pending::decide(tmp.path(), &row.id, PendingState::Approved)
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
let params = Some(serde_json::json!({"id": row.id, "notify": true}));
|
||||
let action = async {
|
||||
match operation {
|
||||
"approve" => handler.handle_federation_approve_request(params).await,
|
||||
"reject" => handler.handle_federation_reject_request(params).await,
|
||||
"retry" => handler
|
||||
.retry_peer_approval_replies()
|
||||
.await
|
||||
.map(|_| serde_json::json!({"ok": true})),
|
||||
_ => handler.handle_federation_cancel_request(params).await,
|
||||
}
|
||||
};
|
||||
let outcome = tokio::time::timeout(Duration::from_secs(20), action)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(outcome.is_ok(), accepted || operation == "approve");
|
||||
let event = tokio::time::timeout(Duration::from_secs(5), relay)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(event.pubkey, sender.public_key());
|
||||
let plaintext =
|
||||
nip44::decrypt(recipient.secret_key(), &event.pubkey, &event.content).unwrap();
|
||||
let message: serde_json::Value = serde_json::from_str(&plaintext).unwrap();
|
||||
let expected = match operation {
|
||||
"approve" | "retry" => "peer-invite",
|
||||
"reject" => "peer-reject",
|
||||
_ => "peer-cancel",
|
||||
};
|
||||
assert_eq!(message["type"], expected);
|
||||
let saved = pending::find_by_id(tmp.path(), &row.id).await.unwrap();
|
||||
if !accepted {
|
||||
assert_eq!(
|
||||
saved.unwrap().state,
|
||||
if operation == "approve" {
|
||||
PendingState::Approved
|
||||
} else {
|
||||
PendingState::Pending
|
||||
}
|
||||
);
|
||||
if operation == "approve" {
|
||||
assert_eq!(outcome.unwrap()["delivery_pending"], true);
|
||||
let durable = crate::federation::handshake_delivery::find(tmp.path(), &row.id)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(durable.recipient, recipient.public_key().to_hex());
|
||||
assert_eq!(durable.attempts, 1);
|
||||
}
|
||||
assert!(crate::federation::load_nodes(tmp.path())
|
||||
.await
|
||||
.unwrap()
|
||||
.is_empty());
|
||||
continue;
|
||||
}
|
||||
match operation {
|
||||
"approve" | "retry" => {
|
||||
assert_eq!(saved.unwrap().state, PendingState::Approved);
|
||||
let first = crate::federation::handshake_delivery::find(tmp.path(), &row.id)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(first.attempts, 1);
|
||||
// Concurrent/background polling honors the persisted backoff.
|
||||
handler.retry_peer_approval_replies().await.unwrap();
|
||||
let second = crate::federation::handshake_delivery::find(tmp.path(), &row.id)
|
||||
.await
|
||||
.unwrap()
|
||||
.unwrap();
|
||||
assert_eq!(second.attempts, 1);
|
||||
assert_eq!(first.invite_code, second.invite_code);
|
||||
let invite =
|
||||
crate::federation::parse_invite(message["invite_code"].as_str().unwrap())
|
||||
.unwrap();
|
||||
assert_eq!(invite.trust_level, crate::federation::TrustLevel::Observer);
|
||||
assert!(!event.content.contains(".onion"));
|
||||
}
|
||||
"reject" => assert_eq!(saved.unwrap().state, PendingState::Rejected),
|
||||
_ => assert!(saved.is_none()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn federation_metrics_are_collected_values_or_unknown_never_placeholders() {
|
||||
for age in [None, Some(0), Some(181), Some(-120)] {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let mut config = crate::config::Config::default();
|
||||
config.data_dir = tmp.path().to_path_buf();
|
||||
let metrics = Arc::new(crate::monitoring::MetricsStore::new());
|
||||
if let Some(age) = age {
|
||||
metrics
|
||||
.push(
|
||||
serde_json::from_value(serde_json::json!({
|
||||
"timestamp": chrono::Utc::now().timestamp() - age,
|
||||
"system": {"cpu_percent": 37.5, "mem_used_bytes": 200,
|
||||
"mem_total_bytes": 800, "disk_used_bytes": 600,
|
||||
"disk_total_bytes": 1000, "net_rx_bytes": 0, "net_tx_bytes": 0,
|
||||
"load_avg_1": 0.0, "load_avg_5": 0.0, "load_avg_15": 0.0},
|
||||
"containers": [], "rpc_latency_ms": 0.0, "ws_connections": 0
|
||||
}))
|
||||
.unwrap(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
let handler = crate::api::rpc::RpcHandler::new(
|
||||
config,
|
||||
Arc::new(crate::state::StateManager::new()),
|
||||
metrics,
|
||||
crate::session::SessionStore::new_for_tests(tmp.path().join("sessions.json")),
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let snapshot = handler.handle_federation_get_state().await.unwrap();
|
||||
if age == Some(0) {
|
||||
assert_eq!(snapshot["cpu_usage_percent"], 37.5);
|
||||
assert_eq!(snapshot["mem_used_bytes"], 200);
|
||||
assert_eq!(snapshot["disk_total_bytes"], 1000);
|
||||
} else {
|
||||
for field in [
|
||||
"cpu_usage_percent",
|
||||
"mem_used_bytes",
|
||||
"mem_total_bytes",
|
||||
"disk_used_bytes",
|
||||
"disk_total_bytes",
|
||||
] {
|
||||
assert!(
|
||||
snapshot.get(field).is_none_or(|v| v.is_null()),
|
||||
"{age:?}: {field}"
|
||||
);
|
||||
}
|
||||
}
|
||||
let peer = serde_json::from_value(serde_json::json!({
|
||||
"did": "did:key:test", "pubkey": "11".repeat(32), "onion": "test.onion",
|
||||
"trust_level": "trusted", "added_at": chrono::Utc::now().to_rfc3339(),
|
||||
"last_state": snapshot
|
||||
}))
|
||||
.unwrap();
|
||||
crate::federation::save_nodes(tmp.path(), &[peer])
|
||||
.await
|
||||
.unwrap();
|
||||
let fleet = handler.handle_telemetry_fleet_status().await.unwrap();
|
||||
let report = &fleet["nodes"][0];
|
||||
if age == Some(0) {
|
||||
assert_eq!(report["cpu_pct"], 38.0);
|
||||
assert_eq!(report["mem_pct"], 25.0);
|
||||
assert_eq!(report["disk_pct"], 60.0);
|
||||
} else {
|
||||
for field in ["cpu_pct", "mem_pct", "disk_pct"] {
|
||||
assert!(report[field].is_null(), "{age:?}: {field}");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,4 +1,6 @@
|
||||
mod handlers;
|
||||
#[cfg(test)]
|
||||
mod handshake_tests;
|
||||
|
||||
use anyhow::Result;
|
||||
|
||||
|
||||
@@ -276,6 +276,11 @@ impl RpcHandler {
|
||||
}
|
||||
Err(e) => tracing::debug!("background handshake poll failed: {e:#}"),
|
||||
}
|
||||
if load_discovery_state(&self.config.data_dir).await.enabled {
|
||||
if let Err(error) = self.retry_peer_approval_replies().await {
|
||||
tracing::warn!("Peer approval retry could not complete: {error:#}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) async fn handle_handshake_poll(&self) -> Result<serde_json::Value> {
|
||||
@@ -356,6 +361,16 @@ impl RpcHandler {
|
||||
);
|
||||
continue;
|
||||
};
|
||||
let scoped_invite = match crate::federation::restrict_discovery_invite(
|
||||
invite_code,
|
||||
&row.from_did,
|
||||
) {
|
||||
Ok(code) => code,
|
||||
Err(_) => {
|
||||
tracing::warn!("Rejected peer invite with mismatched identity");
|
||||
continue;
|
||||
}
|
||||
};
|
||||
let row_id = row.id.clone();
|
||||
let (data, _) = self.state_manager.get_snapshot().await;
|
||||
let local_did =
|
||||
@@ -373,7 +388,7 @@ impl RpcHandler {
|
||||
let local_name = data.server_info.name.clone();
|
||||
match crate::federation::accept_invite(
|
||||
&self.config.data_dir,
|
||||
invite_code,
|
||||
&scoped_invite,
|
||||
&local_did,
|
||||
&local_onion,
|
||||
&local_pubkey,
|
||||
|
||||
@@ -1025,10 +1025,46 @@ impl RpcHandler {
|
||||
.ok_or_else(|| anyhow::anyhow!("Missing key"))?;
|
||||
|
||||
match key {
|
||||
"claude_api_key_set" => {
|
||||
let key_file = self.config.data_dir.join("secrets/claude-api-key");
|
||||
let has_key = tokio::fs::metadata(&key_file).await.is_ok();
|
||||
Ok(serde_json::json!({ "value": has_key }))
|
||||
"claude_api_key_set" | "openai_api_key_set" => {
|
||||
let provider = if key == "claude_api_key_set" {
|
||||
"claude"
|
||||
} else {
|
||||
"openai"
|
||||
};
|
||||
Ok(
|
||||
serde_json::json!({ "value": crate::settings::model_provider::has_key(&self.config.data_dir, provider).await }),
|
||||
)
|
||||
}
|
||||
"ai_provider" => {
|
||||
let settings =
|
||||
crate::settings::model_provider::ModelProvider::load(&self.config.data_dir)
|
||||
.await?;
|
||||
Ok(serde_json::json!({ "value": settings }))
|
||||
}
|
||||
"ai_provider_status" => {
|
||||
let settings =
|
||||
crate::settings::model_provider::ModelProvider::load(&self.config.data_dir)
|
||||
.await?;
|
||||
let local = tokio::time::timeout(std::time::Duration::from_secs(4), async {
|
||||
let (detected, _) = crate::api::rpc::mesh::assistant::detect_ollama().await;
|
||||
detected
|
||||
&& crate::assistant::backends::ollama::model_supports_tools(
|
||||
crate::assistant::backends::ollama::OLLAMA_BASE_URL,
|
||||
crate::assistant::backends::ollama::OLLAMA_DEFAULT_MODEL,
|
||||
)
|
||||
.await
|
||||
});
|
||||
let (claude, openai, local) = tokio::join!(
|
||||
crate::settings::model_provider::has_key(&self.config.data_dir, "claude"),
|
||||
crate::settings::model_provider::has_key(&self.config.data_dir, "openai"),
|
||||
local,
|
||||
);
|
||||
let budget = crate::assistant::AssistantBudget::load(&self.config.data_dir).await;
|
||||
Ok(serde_json::json!({ "value": {
|
||||
"schema": 1, "settings": settings, "claude_configured": claude,
|
||||
"openai_configured": openai, "local_ready": local.ok(),
|
||||
"routstr_remaining_sats": budget.remaining_sats(),
|
||||
}}))
|
||||
}
|
||||
_ => Ok(serde_json::json!({ "value": null })),
|
||||
}
|
||||
@@ -1210,38 +1246,21 @@ impl RpcHandler {
|
||||
let value = params.get("value").and_then(|v| v.as_str()).unwrap_or("");
|
||||
|
||||
match key {
|
||||
"claude_api_key" => {
|
||||
let secrets_dir = self.config.data_dir.join("secrets");
|
||||
tokio::fs::create_dir_all(&secrets_dir)
|
||||
.await
|
||||
.context("Failed to create secrets dir")?;
|
||||
let key_file = secrets_dir.join("claude-api-key");
|
||||
|
||||
if value.is_empty() {
|
||||
// Remove key
|
||||
tokio::fs::remove_file(&key_file).await.ok();
|
||||
info!("Claude API key removed");
|
||||
"claude_api_key" | "openai_api_key" => {
|
||||
let provider = if key == "claude_api_key" {
|
||||
"claude"
|
||||
} else {
|
||||
// Save key
|
||||
tokio::fs::write(&key_file, value)
|
||||
.await
|
||||
.context("Failed to write API key")?;
|
||||
#[cfg(unix)]
|
||||
{
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
std::fs::set_permissions(&key_file, std::fs::Permissions::from_mode(0o600))
|
||||
.ok();
|
||||
}
|
||||
info!("Claude API key saved");
|
||||
}
|
||||
|
||||
// `secrets/claude-api-key` (above) is deliberately the ONLY
|
||||
// Claude key ledger on this node (13-02-PLAN.md). A second
|
||||
// copy used to be written alongside it for a standalone,
|
||||
// unauthenticated sidecar process on port 3142 — that
|
||||
// sidecar and its key copy are retired; the session-gated
|
||||
// Rust daemon reads this one file directly.
|
||||
|
||||
"openai"
|
||||
};
|
||||
crate::settings::model_provider::save_key(&self.config.data_dir, provider, value)
|
||||
.await?;
|
||||
info!(provider, "AI provider credential updated");
|
||||
Ok(serde_json::json!({ "saved": true }))
|
||||
}
|
||||
"ai_provider" => {
|
||||
let settings: crate::settings::model_provider::ModelProvider =
|
||||
serde_json::from_str(value).context("Invalid AI provider settings")?;
|
||||
settings.save(&self.config.data_dir).await?;
|
||||
Ok(serde_json::json!({ "saved": true }))
|
||||
}
|
||||
_ => anyhow::bail!("Unknown setting: {}", key),
|
||||
|
||||
Reference in New Issue
Block a user