diff --git a/core/archipelago/src/api/rpc/content.rs b/core/archipelago/src/api/rpc/content.rs index 0b0d0762..2daf9a70 100644 --- a/core/archipelago/src/api/rpc/content.rs +++ b/core/archipelago/src/api/rpc/content.rs @@ -1257,37 +1257,57 @@ impl RpcHandler { }) .collect(); + // CONCURRENT with a cap, mirroring Cloud.vue's peer-files fan-out + // (BROWSE_PEER_CONCURRENCY = 3, 10s per peer, one attempt). That is + // the implementation the operator already trusts, and it is why the + // Cloud tab answers while a sequential version here did not: with 16 + // peers at 8s each, going one at a time reached only one or two inside + // any sane budget. + // + // 02-08 is the reason for the CAP rather than an unbounded fan-out — + // 13 of 14 simultaneous browse-peer calls never settled and starved + // the connection pool. Three at a time keeps a dead peer from costing + // anything but its own slot. + // Cloud.vue uses 3, but that cap exists because CHROMIUM's connection + // pool was being starved (02-08) — a browser constraint the daemon does + // not share. Measured here: at 3, a 20s budget only got through 2 + // batches of 16 peers and reached none. At 8 every peer is attempted + // inside the budget, which is the point. + const BROWSE_PEER_CONCURRENCY: usize = 8; + const PER_PEER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + let overall = std::time::Duration::from_secs(20); + let mut items = Vec::new(); let mut reached = 0usize; let mut unreachable = 0usize; - // OVERALL budget, not just per-peer. With a dozen federated peers an - // 8s per-peer timeout still adds up past any usable answer — measured - // on the node: this call returned nothing after 45 seconds, which the - // assistant reports to the user as "having trouble accessing the peer - // content list". Partial results beat a timeout: whatever answered - // inside the budget is returned, and the counts say what was missed. - let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(20); - - // Sequential with a per-peer timeout rather than an unbounded fan-out: - // 02-08 traced a real UI stall to content.browse-peer starving the - // connection pool, and the assistant is not latency-critical. - for onion in &onions { - if tokio::time::Instant::now() >= deadline { - // Everything not yet tried counts as unreachable for this call - // rather than being silently omitted. - unreachable += onions.len() - reached - unreachable; - break; + let results = tokio::time::timeout(overall, async { + let mut out: Vec<(String, Option)> = Vec::new(); + for chunk in onions.chunks(BROWSE_PEER_CONCURRENCY) { + let mut set = Vec::new(); + for onion in chunk { + let params = Some(serde_json::json!({ "onion": onion })); + set.push(async move { + let v = tokio::time::timeout( + PER_PEER_TIMEOUT, + self.handle_content_browse_peer(params), + ) + .await + .ok() + .and_then(|r| r.ok()); + (onion.clone(), v) + }); + } + out.extend(futures_util::future::join_all(set).await); } - let params = Some(serde_json::json!({ "onion": onion })); - // Never wait past the overall deadline for a single peer. - let per_peer = std::cmp::min( - std::time::Duration::from_secs(8), - deadline.saturating_duration_since(tokio::time::Instant::now()), - ); - match tokio::time::timeout(per_peer, self.handle_content_browse_peer(params)).await - { - Ok(Ok(v)) => { + out + }) + .await + .unwrap_or_default(); + + for (onion, v) in &results { + match v { + Some(v) => { reached += 1; if let Some(arr) = v.get("items").and_then(|i| i.as_array()) { for it in arr { @@ -1299,17 +1319,18 @@ impl RpcHandler { } } } - _ => unreachable += 1, + None => unreachable += 1, } } + // Peers the overall budget never got to are unreachable for this call, + // not silently absent. + unreachable += onions.len().saturating_sub(results.len()); Ok(serde_json::json!({ "items": items, "peers_reached": reached, "peers_unreachable": unreachable, "peers_total": onions.len(), - // Explicit so the assistant can say "3 of 12 peers answered" - // instead of implying the empty ones have nothing to share. "partial": unreachable > 0, })) } diff --git a/core/archipelago/src/api/rpc/system/handlers.rs b/core/archipelago/src/api/rpc/system/handlers.rs index 74ba6bff..cd097797 100644 --- a/core/archipelago/src/api/rpc/system/handlers.rs +++ b/core/archipelago/src/api/rpc/system/handlers.rs @@ -1018,8 +1018,8 @@ impl RpcHandler { pub(in crate::api::rpc) async fn handle_ai_permissions_get( &self, ) -> Result { - let perms = crate::settings::ai_permissions::load(&self.config.data_dir).await; - Ok(serde_json::json!({ "granted": perms.granted })) + let granted = Self::ai_grants_unified(&self.config.data_dir).await; + Ok(serde_json::json!({ "granted": granted })) } /// ai.permissions.set — replace the grant set. @@ -1045,12 +1045,58 @@ impl RpcHandler { }) .ok_or_else(|| anyhow::anyhow!("Missing granted (array of category ids)"))?; - let saved = crate::settings::ai_permissions::save( + // ONE store, not two kept in sync. The assistant's grants file is the + // authority — it is what actually gates the tool list, so a value that + // does not reach it is a setting that does nothing. That was the live + // bug: Settings wrote only the UI store, assistant/grants.json stayed + // {"apps","system"}, and the model truthfully answered "I don't have a + // tool for that" no matter what the operator toggled. + let mut grants = crate::assistant::grants::Grants::default_closed(); + for name in &granted { + if let Ok(cat) = serde_json::from_value::( + serde_json::Value::String(name.clone()), + ) { + grants.set(cat, true); + } + } + grants.save(&self.config.data_dir).await?; + + // Keep the legacy file in step so a downgrade does not lose grants. + let _ = crate::settings::ai_permissions::save( &self.config.data_dir, - crate::settings::ai_permissions::AiPermissions { granted }, + crate::settings::ai_permissions::AiPermissions { + granted: granted.clone(), + }, ) - .await?; - Ok(serde_json::json!({ "granted": saved.granted })) + .await; + + let saved = Self::ai_grants_unified(&self.config.data_dir).await; + Ok(serde_json::json!({ "granted": saved })) + } + + /// The node's AI grants, as category id strings. + /// + /// Reads the assistant's grants file — the one that actually gates tools — + /// and folds in anything still recorded only in the older + /// settings/ai_permissions.json, so grants made before the two were + /// unified are not silently revoked on upgrade. + async fn ai_grants_unified(data_dir: &std::path::Path) -> Vec { + let grants = crate::assistant::grants::Grants::load(data_dir).await; + let mut out: Vec = grants + .categories() + .iter() + .filter_map(|c| serde_json::to_value(c).ok()?.as_str().map(str::to_string)) + .collect(); + + let legacy = crate::settings::ai_permissions::load(data_dir).await; + for name in legacy.granted { + if !out.contains(&name) { + out.push(name); + } + } + out.sort(); + out.dedup(); + out } /// auth.session-policy.get — how long a login lasts on this node.