car-server-core 0.55.0

//! Reading inventories — this instance's, and a peer's.
//!
//! The local half walks the state this daemon already keeps: attached agents,
//! the observe-only registry, declarative agents, installed coding CLIs, the
//! runtime's registered tools, learned skills, and the model registry. Nothing
//! new is stored; the inventory is a projection.
//!
//! The remote half asks a peer over the A2A surface, and **degrades honestly**.
//! `car/fleetInventory` is a CAR extension, so a third-party A2A agent (or an
//! older CAR) answers `MethodNotFound` — in which case the public agent card
//! still yields that peer's skills, and the result is marked
//! `InventoryFidelity::CardOnly` so an empty model list reads as *unknown*
//! rather than *none*. That distinction is the whole reason the fidelity field
//! exists: a placement decision made on "no models" when the truth is "we never
//! asked" sends work to the wrong machine.

use std::sync::Arc;

use async_trait::async_trait;
use car_fleet::{
    now_ms, FleetAgent, FleetAgentKind, FleetCapability, FleetCapabilityKind, FleetError,
    FleetModel, FleetModelKind, InstanceInventory, InstanceRef, InventoryProvider,
};

use crate::session::ServerState;

/// How many of an instance's learned skills to report.
///
/// A mature memory graph holds thousands, and a composite is read by humans and
/// by a placement step that only needs to know *what kinds of thing* a host
/// knows. Truncation is visible: the row count in the composite stops here, and
/// `skill.find` on that instance remains the full-fidelity surface.
const MAX_SKILLS_REPORTED: usize = 200;

/// Build this instance's inventory.
///
/// `runtime` supplies the registered tools — which runtime matters: over the
/// WebSocket it is the calling session's (so a client's own `tools.register`d
/// tools appear), and on the peer-facing A2A surface it is that listener's,
/// which is exactly the set a peer could actually reach. Passing `None` reports
/// no tools rather than guessing at another session's.
///
/// `memgine` supplies learned skills and is likewise session-scoped; the
/// peer-facing report passes `None`, so a peer sees this host's tools, agents,
/// and models but not its private skill graph.
pub async fn local_inventory(
    state: &ServerState,
    runtime: Option<&Arc<car_engine::Runtime>>,
    memgine: Option<&tokio::sync::Mutex<car_memgine::MemgineEngine>>,
) -> InstanceInventory {
    let identity = car_identity::IdentityStore::from_home().load_or_default();
    let mut instance = InstanceRef::local(car_a2a::lan::host_label());
    instance.version = Some(env!("CARGO_PKG_VERSION").to_string());
    instance.platform = Some(format!(
        "{}/{}",
        std::env::consts::OS,
        std::env::consts::ARCH
    ));

    let mut inv = InstanceInventory::full(instance, now_ms());

    // --- agents ---------------------------------------------------------
    // The flagship assistant is always present: it is what a peer reaches when
    // it sends a conversational message, so omitting it would misdescribe the
    // instance.
    inv.agents.push(FleetAgent {
        id: crate::assistant::register::ASSISTANT_AGENT_ID.to_string(),
        kind: FleetAgentKind::Assistant,
        display_name: Some(identity.name.clone()),
        status: None,
        capability: None,
        addressable: true,
    });

    for peer in crate::peers::snapshot_attached(state).await {
        inv.agents.push(FleetAgent {
            id: peer.name,
            kind: FleetAgentKind::Attached,
            display_name: peer.display_name,
            status: Some("attached".to_string()),
            capability: peer.capability,
            addressable: true,
        });
    }

    if let Ok(registry) = car_registry::AgentRegistry::user_default() {
        if let Ok(entries) = registry.list() {
            for entry in entries {
                inv.agents.push(FleetAgent {
                    id: entry.name,
                    kind: FleetAgentKind::Supervised,
                    display_name: entry.display_name,
                    status: Some(format!("{:?}", entry.status).to_lowercase()),
                    capability: entry.capability,
                    addressable: false,
                });
            }
        }
    }

    if let Ok(decl) = state.declagents() {
        for spec in decl.list().into_iter().filter(|s| s.enabled) {
            inv.agents.push(FleetAgent {
                id: spec.id,
                kind: FleetAgentKind::Declarative,
                display_name: Some(spec.name),
                status: Some("enabled".to_string()),
                capability: Some(spec.identity),
                addressable: false,
            });
        }
    }

    for spec in super::detected_adapters().await {
        inv.agents.push(FleetAgent {
            id: spec.id.clone(),
            kind: FleetAgentKind::ExternalCli,
            display_name: Some(spec.display_name.clone()),
            status: spec.version.clone(),
            capability: None,
            // A batch CLI has no inbox — same asymmetry `car_peers::PeerKind`
            // encodes, repeated here so a fleet consumer needn't cross-reference.
            addressable: false,
        });
    }

    // --- capabilities ---------------------------------------------------
    if let Some(runtime) = runtime {
        for schema in runtime.tool_schemas().await {
            let mut tags = Vec::new();
            if schema.idempotent {
                tags.push("idempotent".to_string());
            }
            if schema.cache_ttl_secs.is_some() {
                tags.push("cacheable".to_string());
            }
            if schema.rate_limit.is_some() {
                tags.push("rate-limited".to_string());
            }
            inv.capabilities.push(FleetCapability {
                name: schema.name,
                kind: FleetCapabilityKind::Tool,
                description: Some(schema.description),
                tags,
            });
        }
    }

    if let Some(memgine) = memgine {
        let engine = memgine.lock().await;
        let mut skills: Vec<FleetCapability> = engine
            .graph
            .inner
            .node_indices()
            .filter_map(|nix| {
                let node = engine.graph.inner.node_weight(nix)?;
                if node.kind != car_memgine::MemKind::Skill {
                    return None;
                }
                let meta = car_memgine::SkillMeta::from_node(node)?;
                Some(FleetCapability {
                    name: meta.name,
                    kind: FleetCapabilityKind::Skill,
                    description: None,
                    tags: Vec::new(),
                })
            })
            .collect();
        skills.sort_by(|a, b| a.name.cmp(&b.name));
        skills.truncate(MAX_SKILLS_REPORTED);
        inv.capabilities.extend(skills);
    }

    // --- models ---------------------------------------------------------
    for model in crate::handler::get_inference_engine(state).list_models_unified() {
        inv.models.push(FleetModel {
            id: model.id,
            kind: if model.is_local {
                FleetModelKind::Local
            } else {
                FleetModelKind::Cloud
            },
            provider: Some(model.provider),
            available: model.available,
            context_window: Some(model.context_length as u64),
            capabilities: model
                .capabilities
                .iter()
                .map(|c| format!("{c:?}").to_lowercase())
                .collect(),
        });
    }

    inv.worker = Some(super::worker_profile().await);
    inv
}

/// Reads a peer CAR daemon's inventory over the A2A surface.
pub struct PeerInventoryProvider {
    instance: InstanceRef,
    identity: Option<Arc<car_a2a::peer_auth::PeerIdentity>>,
}

impl PeerInventoryProvider {
    pub fn new(
        instance: InstanceRef,
        identity: Option<Arc<car_a2a::peer_auth::PeerIdentity>>,
    ) -> Self {
        Self { instance, identity }
    }

    fn client(&self, base_url: &str) -> car_a2a::client::A2aClient {
        let client = car_a2a::client::A2aClient::new(base_url);
        match &self.identity {
            Some(id) => client.with_peer_identity(Arc::clone(id)),
            // Without an identity the peer will refuse us; the call still runs
            // so the failure is reported against the named instance rather than
            // the instance silently disappearing from the fleet.
            None => client,
        }
    }
}

#[async_trait]
impl InventoryProvider for PeerInventoryProvider {
    fn instance(&self) -> InstanceRef {
        self.instance.clone()
    }

    async fn inventory(&self) -> Result<InstanceInventory, FleetError> {
        let base_url = self
            .instance
            .base_url
            .as_deref()
            .ok_or_else(|| FleetError::Protocol("peer has no base url".into()))?;
        let client = self.client(base_url);

        match client
            .call::<_, serde_json::Value>("car/fleetInventory", &serde_json::json!({}))
            .await
        {
            Ok(value) => {
                let mut inv: InstanceInventory = serde_json::from_value(value)
                    .map_err(|e| FleetError::Protocol(format!("inventory shape: {e}")))?;
                // The peer describes itself as `local`; from here it is a
                // remote, and it is addressed by the name THIS host knows it
                // by (the peer listing's), not the name it calls itself.
                let name = self.instance.name.clone();
                let reported = inv.instance.name.clone();
                inv.instance = InstanceRef {
                    name,
                    version: inv.instance.version,
                    platform: inv.instance.platform,
                    ..self.instance.clone()
                };
                if reported != inv.instance.name {
                    inv.instance.reference = self.instance.reference.clone();
                }
                Ok(inv)
            }
            // -32601 is JSON-RPC method-not-found, matched as a literal rather
            // than a guard so clippy::redundant_guards stays satisfied.
            Err(car_a2a::client::ClientError::Rpc { code: -32601, .. }) => {
                // Not a CAR peer, or an older one. Its public card still lists
                // skills, which is real information — take it and say so.
                self.card_only(&client).await
            }
            // A 401 is the one failure whose fix is not on the peer: it means
            // this host's key is not in that daemon's trust set, which is
            // sourced from the synced device roster. Saying "unauthorized"
            // would send an operator to the wrong machine.
            Err(car_a2a::client::ClientError::Status { code: 401, .. }) => {
                Err(FleetError::Unreachable(format!(
                    "`{}` does not trust this host's peer key — a CAR daemon accepts keys published on the signed-in device roster, so sign in to the same account on both machines",
                    self.instance.name
                )))
            }
            Err(e) => Err(FleetError::Unreachable(e.to_string())),
        }
    }
}

impl PeerInventoryProvider {
    async fn card_only(
        &self,
        client: &car_a2a::client::A2aClient,
    ) -> Result<InstanceInventory, FleetError> {
        let card = client
            .agent_card()
            .await
            .map_err(|e| FleetError::Unreachable(e.to_string()))?;
        let capabilities = card
            .skills
            .into_iter()
            .map(|s| FleetCapability {
                name: s.name,
                kind: FleetCapabilityKind::A2aSkill,
                description: Some(s.description),
                tags: s.tags,
            })
            .collect();
        let mut instance = self.instance.clone();
        instance.version = Some(card.version);
        Ok(InstanceInventory::card_only(
            instance,
            capabilities,
            now_ms(),
            Some(
                "peer does not implement `car/fleetInventory`; agents and models are unknown, \
                 not absent"
                    .to_string(),
            ),
        ))
    }
}