car-server-core 0.51.0

//! Peer messaging between agents — `agents.peers` and `agents.message`.
//!
//! Delivery is a reverse-call down the authenticated connection the daemon
//! already holds for each agent, the same mechanism `agents.chat` uses. There
//! is deliberately no per-agent socket: a path that reached an agent without
//! passing through here would make admission advisory, since nothing would sit
//! between sender and recipient.
//!
//! Design: `docs/proposals/agent-to-agent-messaging.md`.
//!
//! ## What a message is, and is not
//!
//! A peer message is inert data. It is explicitly not a `proposal.submit`, so it
//! cannot reach the executor; whatever the receiving agent decides to *do* about
//! it goes through that agent's own gates unchanged. It cannot answer a pending
//! permission prompt, and a slash command in the body arrives as text.
//!
//! ## Why the sender is not a parameter
//!
//! `from` is derived server-side from the connection's bound `agent_id`. A
//! caller-supplied sender would let any agent attribute a message to any other,
//! which would in turn make the anti-laundering rule — never ask a peer to do
//! what was refused here — unenforceable, because the audit trail would be
//! forgeable.

use crate::session::{ClientSession, ServerState};
use car_peers::{
    DeliveryGuard, DeliveryOutcome, GuardVerdict, PeerAddress, PeerDescriptor, PeerDirectory,
    PeerKind, PeerMessage, PeerSource, StaticProvider,
};
use futures::SinkExt;
use serde_json::Value;
use tokio::sync::oneshot;
use tokio_tungstenite::tungstenite::Message;

/// A peer message set aside for operator approval.
///
/// Carries the resolved target alongside the message because the recipient may
/// have detached by the time a human answers — approval then fails with a
/// structured error naming the agent, rather than silently re-resolving to
/// whatever now answers to that name.
#[derive(Debug, Clone, serde::Serialize)]
pub struct HeldPeerMessage {
    pub message: PeerMessage,
    pub target: PeerDescriptor,
    pub held_at_ms: u64,
    pub reason: String,
}

/// How long to wait for a recipient to acknowledge a peer message.
///
/// Short on purpose. The ack means "your agent took delivery", not "your agent
/// acted on it" — an agent that treats a message as work to do would otherwise
/// hold the sender's call open for the length of a task.
const PEER_ACK_TIMEOUT_SECS: u64 = 5;

/// Snapshot the agents currently attached to this daemon as peers.
///
/// A point-in-time copy rather than a live view: assembling a listing while the
/// connection table shifts underneath would produce a list that never existed.
/// The on-disk agent registry is deliberately not consulted — it is observe-only
/// self-report whose reap sweep tolerates a 900s stale window, so routing on it
/// would address agents that exited a quarter of an hour ago.
pub async fn snapshot_attached(state: &ServerState) -> Vec<PeerDescriptor> {
    let attached = state.attached_agents.lock().await.clone();
    attached
        .into_keys()
        .filter(|id| car_peers::is_valid_peer_name(id))
        .map(|agent_id| PeerDescriptor {
            name: agent_id.clone(),
            reference: None,
            kind: PeerKind::CarAgent,
            source: PeerSource::Attached,
            address: PeerAddress::AttachedAgent { agent_id },
            display_name: None,
            capability: None,
            last_seen_ms: Some(car_peers::now_ms()),
        })
        .collect()
}

/// Build the directory as seen by `session`.
async fn directory_for(state: &ServerState, session: &ClientSession) -> PeerDirectory {
    let self_name = session.agent_id.lock().await.clone().unwrap_or_default();
    let mut dir = PeerDirectory::new(self_name).with_provider(Box::new(StaticProvider::new(
        "attached",
        snapshot_attached(state).await,
    )));
    for (label, peers) in [
        ("parslee", snapshot_parslee(state).await),
        ("lan", snapshot_lan(state)),
    ] {
        if !peers.is_empty() {
            dir = dir.with_provider(Box::new(StaticProvider::new(label, peers)));
        }
    }
    dir
}

/// CAR daemons this user's other devices announced over the synced oplog.
///
/// Authenticated by construction: the oplog is readable only with this user's
/// own credentials and is end-to-end encrypted, so an entry here is a machine
/// they enrolled. Empty when sync is not configured — "find my other Mac
/// through Parslee" needs a login, and without one this is honestly nothing
/// rather than a guess.
pub async fn snapshot_parslee(state: &ServerState) -> Vec<PeerDescriptor> {
    let handle = { state.sync.lock().unwrap_or_else(|e| e.into_inner()).clone() };
    let Some(sync) = handle else {
        return Vec::new();
    };
    let endpoints = { sync.lock().await.host_endpoints() };
    endpoints
        .into_iter()
        .filter(|e| car_peers::is_valid_peer_name(&e.name))
        .map(|e| PeerDescriptor {
            name: e.name,
            reference: None,
            kind: PeerKind::RemoteCar,
            source: PeerSource::Parslee,
            address: PeerAddress::A2a { base_url: e.url },
            display_name: Some(e.device_id),
            capability: None,
            last_seen_ms: None,
        })
        .collect()
}

/// Recompute which peer keys this host accepts.
///
/// Sourced from the oplog only. A key there arrived over an E2E-encrypted
/// channel this login's key material protects, so publishing one requires
/// already being the user's device — that is what makes it trustworthy without
/// an operator comparing fingerprints.
///
/// mDNS keys are deliberately excluded. An advertisement is an unauthenticated
/// claim, and accepting a key because it was broadcast would defeat the entire
/// scheme: anyone on the network could then talk to CAR. A LAN-discovered host
/// on the same login shows up here anyway, through the oplog.
///
/// Replaces the set wholesale so a device removed upstream stops being accepted.
pub async fn refresh_peer_trust(state: &ServerState) {
    let handle = { state.sync.lock().unwrap_or_else(|e| e.into_inner()).clone() };
    let Some(sync) = handle else {
        state.peer_trust.set_trusted(Vec::<String>::new());
        return;
    };
    let keys: Vec<String> = sync
        .lock()
        .await
        .host_endpoints()
        .into_iter()
        .map(|e| e.pubkey)
        .filter(|k| !k.trim().is_empty())
        .collect();
    let n = keys.len();
    state.peer_trust.set_trusted(keys);
    tracing::debug!(trusted_peers = n, "refreshed CAR peer trust set");
}

/// CAR daemons advertising themselves on the local network.
///
/// Unauthenticated: anyone on the network can advertise any name. These are
/// listed so an operator can see them, and `agents.message` refuses them until
/// they are promoted through the A2A peer registry's trust gate — the same one
/// `a2a.peers.add` uses. Discovery makes a peer visible; it does not make it
/// reachable.
pub fn snapshot_lan(state: &ServerState) -> Vec<PeerDescriptor> {
    let guard = state
        .lan_discovery
        .lock()
        .unwrap_or_else(|e| e.into_inner());
    let Some(dir) = guard.as_ref() else {
        return Vec::new();
    };
    let trusted: std::collections::HashSet<String> = car_a2a::peers::PeerRegistry::user_default()
        .map(|r| r.list().into_iter().map(|p| p.url).collect())
        .unwrap_or_default();
    dir.peers()
        .into_iter()
        .filter(|p| car_peers::is_valid_peer_name(&p.name))
        // A LAN peer the operator already promoted is reported under its
        // trusted source instead, so it is addressable and not double-listed.
        .filter(|p| !trusted.contains(&p.url))
        .map(|p| PeerDescriptor {
            name: p.name,
            reference: None,
            kind: PeerKind::RemoteCar,
            source: PeerSource::Lan,
            address: PeerAddress::A2a { base_url: p.url },
            display_name: None,
            capability: None,
            last_seen_ms: None,
        })
        .collect()
}

/// `agents.peers` — who this caller can message.
///
/// Mirrors the shape of Claude Code's `/list-agents`: the caller's own name
/// first (it is the address others use to reach it) and never among the rows,
/// since a message addressed to yourself is an error rather than a loopback.
pub async fn handle_agents_peers(
    state: &ServerState,
    session: &ClientSession,
) -> Result<Value, String> {
    let dir = directory_for(state, session).await;
    let peers = dir.list();
    // Whether OTHER hosts can find this one. Distinct from whether this host can
    // find them: browsing needs nothing, advertising needs an A2A endpoint. A
    // caller seeing an empty peer list needs to know which half is missing.
    let discoverable = state
        .lan_discovery
        .lock()
        .unwrap_or_else(|e| e.into_inner())
        .is_some();
    Ok(serde_json::json!({
        "self": if dir.self_name().is_empty() { Value::Null } else { Value::from(dir.self_name()) },
        "lan_browsing": discoverable,
        "peers": peers.iter().map(|p| serde_json::json!({
            "name": p.name,
            "address": p.address_form(),
            "reference": p.reference,
            "kind": p.kind.as_str(),
            "source": p.source.as_str(),
            "can_receive": p.kind.can_receive(),
            "display_name": p.display_name,
            "capability": p.capability,
            "last_seen_ms": p.last_seen_ms,
        })).collect::<Vec<_>>(),
        "count": peers.len(),
    }))
}

/// `agents.message` — deliver text to one peer.
///
/// Params: `{ to, body, summary? }`. `from` is never read from params.
pub async fn handle_agents_message(
    req: &crate::handler::JsonRpcMessage,
    state: &ServerState,
    session: &ClientSession,
) -> Result<Value, String> {
    let to = req
        .params
        .get("to")
        .and_then(|v| v.as_str())
        .ok_or("missing `to`")?
        .to_string();
    let body = req
        .params
        .get("body")
        .and_then(|v| v.as_str())
        .ok_or("missing `body`")?
        .to_string();
    let summary = req
        .params
        .get("summary")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());

    // Server-derived. A caller-supplied sender would make the audit forgeable.
    let from = crate::handler::session_principal_for_peers(session).await;

    let dir = directory_for(state, session).await;
    let target = dir.resolve(&to).map_err(|e| e.to_string())?;

    if !target.source.is_trusted_by_default() {
        return Err(format!(
            "`{}` was discovered on the local network and is not a trusted peer. Anyone on \
             this network can advertise any name, so discovery makes a peer visible, not \
             reachable. Promote it with `a2a.peers.add` first.",
            target.name
        ));
    }

    if !target.kind.can_receive() {
        return Err(format!(
            "`{}` is a {} — it can message CAR while it runs but has no inbox to deliver into",
            target.name,
            target.kind.as_str()
        ));
    }

    let mut msg = PeerMessage::new(&from, &target.name, &body);
    msg.summary = summary;

    // Stage 1: the channel guard. Runs before policy because it is cheap and
    // because a message loop must terminate even when both ends are fully
    // authorized.
    let verdict = {
        let mut guards = state.peer_guards.lock().await;
        let guard = guards
            .entry(target.name.clone())
            .or_insert_with(DeliveryGuard::new);
        guard.admit(&msg, car_peers::now_ms())
    };
    if !verdict.is_accept() {
        let outcome = DeliveryOutcome::Refused {
            reason: verdict.reason(),
        };
        append_peer_audit(&msg, &target, &outcome);
        return Err(guard_error(&verdict));
    }

    // Stage 2: admission. Whether this sender may say this to this recipient.
    let outcome = admit(state, session, &msg, &target).await;
    append_peer_audit(&msg, &target, &outcome);
    match &outcome {
        DeliveryOutcome::Refused { reason } => {
            release(state, &target.name).await;
            return Err(format!("message refused: {reason}"));
        }
        DeliveryOutcome::Held { reason } => {
            // Release the in-flight slot and move the message to the hold queue.
            // The two bounds are separate on purpose: QUEUE_CAP limits what a
            // recipient has yet to read, HOLD_CAP limits what an operator has
            // yet to decide. Charging a held message against the delivery queue
            // would let a slow human block a healthy channel.
            release(state, &target.name).await;
            let held = HeldPeerMessage {
                message: msg.clone(),
                target: target.clone(),
                held_at_ms: car_peers::now_ms(),
                reason: reason.clone(),
            };
            let dropped = {
                let mut q = state.held_peer_messages.lock().await;
                q.push_back(held);
                if q.len() > car_peers::HOLD_CAP {
                    q.pop_front()
                } else {
                    None
                }
            };
            if let Some(evicted) = dropped {
                // Say so rather than losing it quietly: the operator never saw
                // this one, and the sender was told it was retained.
                tracing::warn!(
                    id = %evicted.message.id,
                    from = %evicted.message.from,
                    to = %evicted.target.name,
                    "hold queue full; dropped the oldest undecided peer message"
                );
                append_peer_audit(
                    &evicted.message,
                    &evicted.target,
                    &DeliveryOutcome::Refused {
                        reason: format!(
                            "evicted from the hold queue at {} undecided messages",
                            car_peers::HOLD_CAP
                        ),
                    },
                );
            }
            return Ok(serde_json::json!({
                "id": msg.id,
                "to": target.name,
                "outcome": "held",
                "retained": true,
                "reason": reason,
            }));
        }
        DeliveryOutcome::Delivered => {}
    }

    let result = deliver(state, &target, &msg).await;
    release(state, &target.name).await;
    result
}

/// `agents.message.pending` — peer messages awaiting an operator decision.
///
/// Host-only, mirroring `agents.chat.approve`: an agent must not be able to
/// read, or later approve, the queue that exists to gate it.
pub async fn handle_agents_message_pending(
    state: &ServerState,
    session: &ClientSession,
) -> Result<Value, String> {
    require_host(session, "agents.message.pending")?;
    Ok(pending_snapshot(state).await)
}

/// [`handle_agents_message_pending`] without the host check.
///
/// Split so the queue's behaviour is testable without constructing a live
/// client session; the host check is tested directly on [`require_host`].
pub async fn pending_snapshot(state: &ServerState) -> Value {
    let q = state.held_peer_messages.lock().await;
    serde_json::json!({
        "held": q.iter().map(|h| serde_json::json!({
            "id": h.message.id,
            "from": h.message.from,
            "to": h.target.name,
            "body": h.message.body,
            "held_at_ms": h.held_at_ms,
            "reason": h.reason,
        })).collect::<Vec<_>>(),
        "count": q.len(),
        "cap": car_peers::HOLD_CAP,
    })
}

/// `agents.message.approve` — release or drop one held message.
///
/// Params: `{ id, decision }`. `decision` is a bool, or a string the operator
/// surface finds natural (`approve`/`approved`/`yes`); anything else denies,
/// and an omitted decision denies. Same convention as `agents.chat.approve`, so
/// an operator does not have to remember two.
pub async fn handle_agents_message_approve(
    req: &crate::handler::JsonRpcMessage,
    state: &ServerState,
    session: &ClientSession,
) -> Result<Value, String> {
    require_host(session, "agents.message.approve")?;
    let id = req
        .params
        .get("id")
        .and_then(|v| v.as_str())
        .ok_or("missing `id`")?
        .to_string();
    let approved = match req.params.get("decision") {
        Some(Value::Bool(b)) => *b,
        Some(Value::String(sv)) => {
            matches!(
                sv.to_ascii_lowercase().as_str(),
                "approve" | "approved" | "yes"
            )
        }
        _ => false,
    };

    decide_held(state, &id, approved).await
}

/// [`handle_agents_message_approve`] without the host check. See
/// [`pending_snapshot`] for why the split exists.
pub async fn decide_held(state: &ServerState, id: &str, approved: bool) -> Result<Value, String> {
    let held = {
        let mut q = state.held_peer_messages.lock().await;
        let pos = q.iter().position(|h| h.message.id == id);
        match pos {
            Some(i) => q.remove(i).expect("position just found"),
            None => return Err(format!("no held message with id `{id}`")),
        }
    };

    if !approved {
        append_peer_audit(
            &held.message,
            &held.target,
            &DeliveryOutcome::Refused {
                reason: "denied by the operator".into(),
            },
        );
        return Ok(serde_json::json!({
            "id": id,
            "outcome": "denied",
        }));
    }

    // Re-admit through the channel guard. The message passed it when it was
    // sent, but time has moved and the recipient may since have been flooded;
    // the guard bounds the channel, and an approval is not a licence to bypass
    // it. Its identical-repeat window has long since expired for anything that
    // sat awaiting a human, so this does not spuriously reject.
    let verdict = {
        let mut guards = state.peer_guards.lock().await;
        guards
            .entry(held.target.name.clone())
            .or_insert_with(DeliveryGuard::new)
            .admit(&held.message, car_peers::now_ms())
    };
    if !verdict.is_accept() {
        append_peer_audit(
            &held.message,
            &held.target,
            &DeliveryOutcome::Refused {
                reason: verdict.reason(),
            },
        );
        return Err(guard_error(&verdict));
    }

    append_peer_audit(&held.message, &held.target, &DeliveryOutcome::Delivered);
    let result = deliver(state, &held.target, &held.message).await;
    release(state, &held.target.name).await;
    result
}

/// Refuse a surface that only the operator's own client may drive.
///
/// Separate helper because the reason matters more than the check: these two
/// methods exist to gate agents, so an agent reaching them would be approving
/// the very messages its posture was set to hold.
fn require_host(session: &ClientSession, method: &str) -> Result<(), String> {
    if session.is_host.load(std::sync::atomic::Ordering::Acquire) {
        return Ok(());
    }
    Err(require_host_message(method))
}

/// The refusal text for a host-only peer surface.
fn require_host_message(method: &str) -> String {
    format!("`{method}` is host-only; an agent cannot approve the messages its own posture held")
}

/// Decrement the recipient's in-flight count.
async fn release(state: &ServerState, recipient: &str) {
    if let Some(g) = state.peer_guards.lock().await.get_mut(recipient) {
        g.consumed();
    }
}

/// Turn a guard verdict into the caller-facing error.
///
/// Named separately so the sender is told *which* limit stopped it and can act:
/// batching is the answer to a rate limit, waiting is the answer to a full
/// queue, and neither is the answer to an oversized body.
fn guard_error(v: &GuardVerdict) -> String {
    match v {
        GuardVerdict::Accept => "accepted".into(),
        GuardVerdict::TooLarge { .. } => {
            format!("{} — send a path or a state handle instead", v.reason())
        }
        GuardVerdict::RateLimited { .. } => {
            format!("{} — batch the rest into one message", v.reason())
        }
        GuardVerdict::DuplicateWithinWindow => {
            format!("{} — it was already delivered; do not resend", v.reason())
        }
        GuardVerdict::QueueFull { .. } => {
            format!("{} — wait for it to drain", v.reason())
        }
        GuardVerdict::InvalidName { .. } => v.reason(),
    }
}

/// Admission: may this sender say this to this recipient?
///
/// Resolved against [`car_policy::AgentPermissionPolicy`] at the
/// [`PermissionTier::ReadOnly`] tier, because that is honestly what a peer
/// message is: it mutates nothing on the recipient, reaches no executor, and
/// grants no authority. Rating it higher would be theatre — and rating it lower
/// than a tier at all would leave operators no knob.
///
/// The tier is resolved for the **sender**. The question a peer message raises
/// is whether this agent may talk to other agents, which is the sender's
/// authority; what the recipient then does is gated by the recipient's own
/// runtime, unchanged.
///
/// Under the Balanced preset `ReadOnly` is `AlwaysAllow`, so the default is
/// permissive. The value is that an operator who sets a specific agent's
/// `ReadOnly` posture to `Deny` actually stops its peer messages, rather than
/// the rule living only in a system prompt the agent may or may not follow.
async fn admit(
    _state: &ServerState,
    session: &ClientSession,
    msg: &PeerMessage,
    _target: &PeerDescriptor,
) -> DeliveryOutcome {
    let sender_agent = session.agent_id.lock().await.clone();
    let is_host = session.is_host.load(std::sync::atomic::Ordering::Acquire);
    admit_with(
        &crate::agent_permissions::load_policy(),
        sender_agent,
        is_host,
        &msg.from,
    )
}

/// [`admit`] against an explicit policy.
///
/// Split so the authorization branches can be tested without writing a policy
/// file under `CAR_HOME`, which is process-global and would race the rest of the
/// test binary. An untested authorization path is the one kind that must not
/// ship on a compile alone.
fn admit_with(
    policy: &car_policy::AgentPermissionPolicy,
    sender_agent: Option<String>,
    is_host: bool,
    from: &str,
) -> DeliveryOutcome {
    let Some(agent_id) = sender_agent else {
        if is_host {
            // The host is the operator's own client; it needs no agent posture.
            return DeliveryOutcome::Delivered;
        }
        return DeliveryOutcome::Refused {
            reason: format!(
                "sender `{from}` is neither a bound agent nor the host; a peer message needs an authenticated principal"
            ),
        };
    };

    match policy.resolve(&agent_id, car_policy::PermissionTier::ReadOnly) {
        car_policy::agent_permissions::ApprovalMode::AlwaysAllow => DeliveryOutcome::Delivered,
        car_policy::agent_permissions::ApprovalMode::RequireApproval => DeliveryOutcome::Held {
            reason: format!(
                "`{agent_id}` is set to require approval; the message is held rather than dropped"
            ),
        },
        car_policy::agent_permissions::ApprovalMode::Deny => DeliveryOutcome::Refused {
            reason: format!("`{agent_id}` is denied at the read_only tier"),
        },
    }
}

/// Reverse-call the recipient's attached channel.
async fn deliver(
    state: &ServerState,
    target: &PeerDescriptor,
    msg: &PeerMessage,
) -> Result<Value, String> {
    let agent_id = match &target.address {
        PeerAddress::AttachedAgent { agent_id } => agent_id,
        PeerAddress::A2a { base_url } => return deliver_remote(state, base_url, target, msg).await,
    };

    let agent_client_id = state
        .attached_agents
        .lock()
        .await
        .get(agent_id)
        .cloned()
        .ok_or_else(|| format!("agent `{agent_id}` detached before the message could be sent"))?;
    let channel = {
        let sessions = state.sessions.lock().await;
        sessions
            .get(&agent_client_id)
            .map(|s| s.channel.clone())
            .ok_or_else(|| format!("agent `{agent_id}` raced with disconnect"))?
    };

    let request_id = channel.next_request_id();
    let (tx, rx) = oneshot::channel();
    channel.pending.lock().await.insert(request_id.clone(), tx);

    let rpc = serde_json::json!({
        "jsonrpc": "2.0",
        "method": "agent.peer_message",
        "params": {
            "id": msg.id,
            "from": msg.from,
            "body": msg.body,
            "sent_at_ms": msg.sent_at_ms,
            "no_reply": msg.no_reply,
        },
        "id": request_id,
    });
    let frame = Message::Text(
        serde_json::to_string(&rpc)
            .map_err(|e| e.to_string())?
            .into(),
    );

    if let Err(e) = channel.write.lock().await.send(frame).await {
        channel.pending.lock().await.remove(&request_id);
        return Err(format!("failed to deliver to `{agent_id}`: {e}"));
    }

    match tokio::time::timeout(std::time::Duration::from_secs(PEER_ACK_TIMEOUT_SECS), rx).await {
        Ok(Ok(_)) => Ok(serde_json::json!({
            "id": msg.id,
            "to": target.name,
            "outcome": "delivered",
        })),
        Ok(Err(_)) => Err(format!("agent `{agent_id}` closed before acknowledging")),
        Err(_) => {
            // Timed out: stop waiting, but do not claim non-delivery. The frame
            // was written; an agent that does not implement `agent.peer_message`
            // simply never answers, and saying "not delivered" would be a guess.
            channel.pending.lock().await.remove(&request_id);
            Ok(serde_json::json!({
                "id": msg.id,
                "to": target.name,
                "outcome": "unacknowledged",
                "detail": format!(
                    "written to `{agent_id}` but not acknowledged within {PEER_ACK_TIMEOUT_SECS}s"
                ),
            }))
        }
    }
}

/// Deliver to a CAR daemon on another host, over A2A.
///
/// The message is addressed to the remote **daemon**, not to one of its agents.
/// That daemon then runs it through its own guard and policy before reverse-
/// calling a local agent, so a cross-host message passes two admissions — the
/// sender's here and the recipient's there — and neither can be skipped by
/// naming an agent directly. That property is why `PeerAddress` has no variant
/// for a remote agent.
///
/// Errors are reported as delivery failures rather than swallowed: a peer that
/// is advertised but unreachable is exactly the case an operator needs to see,
/// and a network that silently drops messages is worse than one that refuses
/// them.
async fn deliver_remote(
    state: &ServerState,
    base_url: &str,
    target: &PeerDescriptor,
    msg: &PeerMessage,
) -> Result<Value, String> {
    use car_a2a::types::{Message as A2aMessage, MessageRole, Part, TextPart};

    // Sign as this daemon. Without an identity the peer will refuse us, so say
    // that here rather than letting it surface as an opaque 401 from the far
    // side — the operator's fix is local, not remote.
    let identity = {
        state
            .peer_identity
            .lock()
            .unwrap_or_else(|e| e.into_inner())
            .clone()
    };
    let Some(identity) = identity else {
        return Err(format!(
            "cannot reach `{}`: this daemon has no peer identity, so a remote CAR would \
             refuse it. The identity is created when the A2A surface starts.",
            target.name
        ));
    };
    let client = car_a2a::client::A2aClient::new(base_url).with_peer_identity(identity);
    // The sender travels in metadata, not in the body: a recipient must be able
    // to tell who sent a message without parsing prose, and the body stays the
    // author's text verbatim. Both keys are surfaced camelCase, matching the
    // `correlationId`/`replyTo` convention the type's own docs pin.
    let mut metadata = std::collections::HashMap::new();
    metadata.insert("carPeerFrom".to_string(), Value::from(msg.from.clone()));
    metadata.insert("carPeerTo".to_string(), Value::from(target.name.clone()));
    let a2a_msg = A2aMessage {
        message_id: msg.id.clone(),
        role: MessageRole::User,
        parts: vec![Part::Text(TextPart {
            text: msg.body.clone(),
            metadata: std::collections::HashMap::new(),
        })],
        task_id: None,
        context_id: None,
        metadata,
    };

    match client.send_message(a2a_msg, true).await {
        Ok(_) => Ok(serde_json::json!({
            "id": msg.id,
            "to": target.name,
            "outcome": "delivered",
            "transport": "a2a",
            "url": base_url,
        })),
        Err(e) => Err(format!(
            "failed to deliver to `{}` at {base_url}: {e}",
            target.name
        )),
    }
}

/// Append a peer-message record to the audit journal.
///
/// Best-effort and non-fatal, mirroring `append_external_agent_audit`: an
/// unwritable journal must not fail the call, but every attempted delivery —
/// refused ones included — leaves a record.
pub fn append_peer_audit(msg: &PeerMessage, target: &PeerDescriptor, outcome: &DeliveryOutcome) {
    let Some(car_dir) = car_home::root() else {
        return;
    };
    if std::fs::create_dir_all(&car_dir).is_err() {
        return;
    }
    append_peer_audit_at(&car_dir.join("peer-messages.jsonl"), msg, target, outcome);
}

/// [`append_peer_audit`] against an explicit journal path.
///
/// Split out so the record shape can be tested without mutating `CAR_HOME`,
/// which is process-global and would race every other test in the binary.
pub fn append_peer_audit_at(
    path: &std::path::Path,
    msg: &PeerMessage,
    target: &PeerDescriptor,
    outcome: &DeliveryOutcome,
) {
    use std::io::Write;
    let record = serde_json::json!({
        "ts": chrono::Utc::now().to_rfc3339(),
        "id": msg.id,
        "from": msg.from,
        "to": target.name,
        "kind": target.kind.as_str(),
        "source": target.source.as_str(),
        "bytes": msg.body.len(),
        "outcome": outcome,
    });
    let Ok(line) = serde_json::to_string(&record) else {
        return;
    };
    if let Ok(mut f) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)
    {
        let _ = writeln!(f, "{line}");
    } else {
        tracing::warn!(path = %path.display(), "failed to append peer-message audit record");
    }
}

/// Record an `agents.chat` reverse-call.
///
/// `agents.chat` has driven another agent's turn since it shipped, with no
/// `is_host` gate at dispatch or inside the handler and — verified by grep over
/// the whole handler range — no eventlog, policy, or audit call of any kind,
/// while its sibling `agents.invoke_external` gets
/// `append_external_agent_audit`. That made agent-to-agent messaging shipped,
/// ungoverned behaviour rather than a design option.
///
/// Adding a governed `agents.message` beside an unrecorded `agents.chat` would
/// be worse than either alone: it would move well-behaved callers onto the
/// audited path and leave the unaudited one as the way to avoid the record. So
/// the record lands on both in the same change.
///
/// This closes the *observability* half only. Whether `agents.chat` should also
/// require `is_host` is an authorization change to a shipped surface with live
/// consumers, and it is deliberately not made here.
pub fn append_agent_chat_audit(principal: &str, agent_id: &str, session_id: &str) {
    use std::io::Write;
    let Some(car_dir) = car_home::root() else {
        return;
    };
    if std::fs::create_dir_all(&car_dir).is_err() {
        return;
    }
    let path = car_dir.join("peer-messages.jsonl");
    let record = serde_json::json!({
        "ts": chrono::Utc::now().to_rfc3339(),
        "surface": "agents.chat",
        "from": principal,
        "to": agent_id,
        "session_id": session_id,
    });
    let Ok(line) = serde_json::to_string(&record) else {
        return;
    };
    if let Ok(mut f) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
    {
        let _ = writeln!(f, "{line}");
    }
}

// ---------------------------------------------------------------------------
// MCP surface — the return path for external CLIs
// ---------------------------------------------------------------------------

/// Register `peer_list` and `peer_message` on the daemon's MCP endpoint.
///
/// This is the whole external-CLI story, and it is deliberately one-directional.
/// CAR invokes Claude Code and friends as batch processes — `runner.rs` writes
/// the task to stdin and closes it so the child sees EOF — so there is no
/// steady state to deliver *into*. What a running CLI does have is CAR's MCP
/// server, already mounted by `build_mcp_config_json` on every invocation. So a
/// CLI can message CAR agents while it works; it just cannot be addressed as a
/// recipient. `PeerKind::can_receive()` carries that asymmetry in the type.
///
/// Registered here rather than in `car-mcp` for the same reason the assistant
/// trio is: the tool list is per-`Server`, so `car-mcp-server` — which has no
/// daemon and no connection table — cannot advertise a tool it could not serve.
///
/// ## Sender identity is coarse here, and that is disclosed rather than hidden
///
/// On the WebSocket surface the sender is unforgeable: it comes from the
/// connection's bound `agent_id`. The MCP endpoint has no equivalent — it is
/// shared across invocations and carries no per-invocation principal — so a
/// message sent through this tool is attributed to `mcp:external-cli` and
/// nothing finer. That is a real limitation: the audit record proves *an* MCP
/// client sent it, not *which*. Threading a per-invocation token from
/// `build_mcp_config_json` through to the MCP session would close it, and is
/// the obvious next step. Until then the coarse principal is recorded honestly
/// rather than a specific-looking name being invented.
pub fn register_peer_tools(
    server: &mut car_mcp::Server,
    state: std::sync::Arc<ServerState>,
) -> Result<(), car_mcp::RegisterError> {
    server.register_tool(
        peer_list_schema(),
        std::sync::Arc::new(PeerListTool(state.clone())),
    )?;
    server.register_tool(
        peer_message_schema(),
        std::sync::Arc::new(PeerMessageTool(state)),
    )?;
    Ok(())
}

/// The principal recorded for anything arriving over the shared MCP endpoint.
const MCP_PRINCIPAL: &str = "mcp:external-cli";

fn peer_list_schema() -> Value {
    serde_json::json!({
        "name": "peer_list",
        "description": "List the CAR agents you can send a message to. Returns each peer's \
                        address, kind, and whether it can receive. Use the returned `address` \
                        verbatim as peer_message's `to` — it carries a disambiguating suffix \
                        when two live agents share a name.",
        "inputSchema": { "type": "object", "properties": {} },
        "annotations": {
            "readOnlyHint": true,
            "destructiveHint": false,
            "idempotentHint": true,
            "openWorldHint": false,
        },
    })
}

fn peer_message_schema() -> Value {
    serde_json::json!({
        "name": "peer_message",
        "description": "Send a short plain-text message to one CAR agent — a finding, a status, \
                        a decision it is blocked on. The message is text only: it cannot run a \
                        command, approve anything, or change the recipient's configuration, and \
                        whatever the recipient does about it goes through its own permissions. \
                        Get `to` from peer_list. Keep it to one self-contained first line; \
                        identical repeats within 10s are dropped.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "to": { "type": "string", "description": "An `address` from peer_list." },
                "body": { "type": "string", "description": "Plain text. First line should stand alone." },
            },
            "required": ["to", "body"],
        },
        "annotations": {
            "readOnlyHint": false,
            "destructiveHint": false,
            "idempotentHint": false,
            "openWorldHint": true,
        },
    })
}

struct PeerListTool(std::sync::Arc<ServerState>);

#[async_trait::async_trait]
impl car_mcp::ToolHandler for PeerListTool {
    async fn call(&self, _args: Value) -> Result<String, car_mcp::ToolError> {
        let peers = snapshot_attached(&self.0).await;
        let rows: Vec<Value> = peers
            .iter()
            .map(|p| {
                serde_json::json!({
                    "address": p.address_form(),
                    "kind": p.kind.as_str(),
                    "can_receive": p.kind.can_receive(),
                })
            })
            .collect();
        serde_json::to_string(&serde_json::json!({ "peers": rows, "count": rows.len() }))
            .map_err(|e| car_mcp::ToolError::Internal(e.to_string()))
    }
}

struct PeerMessageTool(std::sync::Arc<ServerState>);

#[async_trait::async_trait]
impl car_mcp::ToolHandler for PeerMessageTool {
    async fn call(&self, args: Value) -> Result<String, car_mcp::ToolError> {
        let to = args
            .get("to")
            .and_then(|v| v.as_str())
            .ok_or_else(|| car_mcp::ToolError::InvalidParams("missing `to`".into()))?;
        let body = args
            .get("body")
            .and_then(|v| v.as_str())
            .ok_or_else(|| car_mcp::ToolError::InvalidParams("missing `body`".into()))?;

        let dir = PeerDirectory::new(MCP_PRINCIPAL).with_provider(Box::new(StaticProvider::new(
            "attached",
            snapshot_attached(&self.0).await,
        )));
        let target = dir
            .resolve(to)
            .map_err(|e| car_mcp::ToolError::Internal(e.to_string()))?;
        if !target.kind.can_receive() {
            return Err(car_mcp::ToolError::Internal(format!(
                "`{}` has no inbox to deliver into",
                target.name
            )));
        }

        let msg = PeerMessage::new(MCP_PRINCIPAL, &target.name, body);

        let verdict = {
            let mut guards = self.0.peer_guards.lock().await;
            let guard = guards
                .entry(target.name.clone())
                .or_insert_with(DeliveryGuard::new);
            guard.admit(&msg, car_peers::now_ms())
        };
        if !verdict.is_accept() {
            append_peer_audit(
                &msg,
                &target,
                &DeliveryOutcome::Refused {
                    reason: verdict.reason(),
                },
            );
            return Err(car_mcp::ToolError::Internal(guard_error(&verdict)));
        }

        append_peer_audit(&msg, &target, &DeliveryOutcome::Delivered);
        let result = deliver(&self.0, &target, &msg).await;
        release(&self.0, &target.name).await;
        match result {
            Ok(v) => {
                serde_json::to_string(&v).map_err(|e| car_mcp::ToolError::Internal(e.to_string()))
            }
            Err(e) => Err(car_mcp::ToolError::Internal(e)),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;

    async fn test_state() -> (Arc<ServerState>, tempfile::TempDir) {
        let temp = tempfile::tempdir().unwrap();
        let state = Arc::new(ServerState::with_config(
            crate::session::ServerStateConfig::new(temp.path().to_path_buf()),
        ));
        (state, temp)
    }

    async fn attach(state: &ServerState, agent_id: &str) {
        state
            .attached_agents
            .lock()
            .await
            .insert(agent_id.to_string(), format!("client-{agent_id}"));
    }

    fn held_fixture(id: &str, to: &str) -> HeldPeerMessage {
        let mut m = PeerMessage::new("agent:sender", to, format!("body-{id}"));
        m.id = id.to_string();
        HeldPeerMessage {
            message: m,
            target: PeerDescriptor {
                name: to.into(),
                reference: None,
                kind: PeerKind::CarAgent,
                source: PeerSource::Attached,
                address: PeerAddress::AttachedAgent {
                    agent_id: to.into(),
                },
                display_name: None,
                capability: None,
                last_seen_ms: None,
            },
            held_at_ms: 1_000,
            reason: "requires approval".into(),
        }
    }

    #[tokio::test]
    async fn pending_lists_held_messages_oldest_first() {
        let (state, _t) = test_state().await;
        {
            let mut q = state.held_peer_messages.lock().await;
            q.push_back(held_fixture("first", "milo"));
            q.push_back(held_fixture("second", "milo"));
        }
        let snap = pending_snapshot(&state).await;
        assert_eq!(snap["count"], 2);
        assert_eq!(snap["cap"], car_peers::HOLD_CAP);
        assert_eq!(snap["held"][0]["id"], "first");
        assert_eq!(snap["held"][1]["id"], "second");
    }

    #[tokio::test]
    async fn denying_a_held_message_removes_it() {
        let (state, _t) = test_state().await;
        state
            .held_peer_messages
            .lock()
            .await
            .push_back(held_fixture("m1", "milo"));

        let out = decide_held(&state, "m1", false).await.unwrap();
        assert_eq!(out["outcome"], "denied");
        assert_eq!(
            pending_snapshot(&state).await["count"],
            0,
            "a decided message must leave the queue"
        );
    }

    #[tokio::test]
    async fn deciding_an_unknown_id_is_a_named_error() {
        let (state, _t) = test_state().await;
        let err = decide_held(&state, "ghost", true).await.unwrap_err();
        assert!(err.contains("ghost"), "error should name the id: {err}");
    }

    #[tokio::test]
    async fn a_held_message_cannot_be_decided_twice() {
        let (state, _t) = test_state().await;
        state
            .held_peer_messages
            .lock()
            .await
            .push_back(held_fixture("m1", "milo"));
        assert!(decide_held(&state, "m1", false).await.is_ok());
        // The second decision must fail rather than re-deliver: removal on
        // decide is what makes approval idempotent-by-absence.
        assert!(decide_held(&state, "m1", true).await.is_err());
    }

    #[tokio::test]
    async fn approving_a_detached_recipient_fails_loudly() {
        let (state, _t) = test_state().await;
        // Held while attached, decided after the agent went away.
        state
            .held_peer_messages
            .lock()
            .await
            .push_back(held_fixture("m1", "ghost"));
        let err = decide_held(&state, "m1", true).await.unwrap_err();
        assert!(
            err.contains("ghost"),
            "approval of a vanished recipient must name it: {err}"
        );
        assert_eq!(pending_snapshot(&state).await["count"], 0);
    }

    #[test]
    fn only_the_host_may_read_or_decide_the_hold_queue() {
        // The queue exists to gate agents, so an agent reaching these surfaces
        // would be approving the very messages its posture held.
        let msg = require_host_message("agents.message.approve");
        assert!(msg.contains("host-only"), "{msg}");
        assert!(msg.contains("its own posture held"), "{msg}");
    }

    #[test]
    fn an_unauthenticated_sender_is_refused() {
        let policy = car_policy::AgentPermissionPolicy::default();
        let out = admit_with(&policy, None, false, "conn:abc");
        assert!(
            matches!(out, DeliveryOutcome::Refused { .. }),
            "got {out:?}"
        );
    }

    #[test]
    fn the_host_needs_no_agent_posture() {
        let policy = car_policy::AgentPermissionPolicy::default();
        assert_eq!(
            admit_with(&policy, None, true, "conn:host"),
            DeliveryOutcome::Delivered
        );
    }

    #[test]
    fn a_bound_agent_is_allowed_by_default() {
        let policy = car_policy::AgentPermissionPolicy::default();
        assert_eq!(
            admit_with(&policy, Some("milo".into()), false, "agent:milo"),
            DeliveryOutcome::Delivered
        );
    }

    #[test]
    fn denying_an_agent_at_read_only_actually_stops_its_messages() {
        // The whole point of resolving a tier: an operator's setting has to bind,
        // rather than the rule living only in a prompt the agent may ignore.
        let mut policy = car_policy::AgentPermissionPolicy::default();
        policy.set_agent(
            "milo",
            car_policy::PermissionTier::ReadOnly,
            car_policy::agent_permissions::ApprovalMode::Deny,
        );
        let out = admit_with(&policy, Some("milo".into()), false, "agent:milo");
        assert!(
            matches!(out, DeliveryOutcome::Refused { .. }),
            "got {out:?}"
        );
        // A different agent is unaffected by the per-agent override.
        assert_eq!(
            admit_with(&policy, Some("trader".into()), false, "agent:trader"),
            DeliveryOutcome::Delivered
        );
    }

    #[test]
    fn require_approval_holds_rather_than_drops() {
        let mut policy = car_policy::AgentPermissionPolicy::default();
        policy.set_agent(
            "milo",
            car_policy::PermissionTier::ReadOnly,
            car_policy::agent_permissions::ApprovalMode::RequireApproval,
        );
        // Held is a third outcome on purpose: it can still be delivered later,
        // and the sender is told which of the two happened.
        assert!(matches!(
            admit_with(&policy, Some("milo".into()), false, "agent:milo"),
            DeliveryOutcome::Held { .. }
        ));
    }

    #[tokio::test]
    async fn snapshot_lists_attached_agents() {
        let (state, _t) = test_state().await;
        attach(&state, "milo").await;
        attach(&state, "trader").await;
        let peers = snapshot_attached(&state).await;
        assert_eq!(peers.len(), 2);
        assert!(peers.iter().all(|p| p.kind == PeerKind::CarAgent));
        assert!(peers.iter().all(|p| p.source == PeerSource::Attached));
    }

    #[tokio::test]
    async fn snapshot_drops_names_that_are_not_addressable() {
        let (state, _t) = test_state().await;
        attach(&state, "milo").await;
        // A name that would escape the addressing charset must never become a
        // peer, regardless of how it got into the connection table.
        attach(&state, "../escape").await;
        let peers = snapshot_attached(&state).await;
        assert_eq!(peers.len(), 1);
        assert_eq!(peers[0].name, "milo");
    }

    #[tokio::test]
    async fn an_oversized_message_is_refused_before_delivery() {
        let (state, _t) = test_state().await;
        attach(&state, "milo").await;

        let msg = PeerMessage::new("agent:sender", "milo", "x".repeat(2_000_000));
        let verdict = {
            let mut guards = state.peer_guards.lock().await;
            guards
                .entry("milo".to_string())
                .or_insert_with(DeliveryGuard::new)
                .admit(&msg, car_peers::now_ms())
        };
        assert!(matches!(verdict, GuardVerdict::TooLarge { .. }));
        // And the refusal names a remedy rather than just a limit.
        assert!(guard_error(&verdict).contains("state handle"));
    }

    #[tokio::test]
    async fn a_detached_agent_yields_a_structured_error_not_a_hang() {
        let (state, _t) = test_state().await;
        // Present in the connection table but with no live session behind it —
        // exactly the disconnect race. It must resolve to a named error.
        attach(&state, "ghost").await;
        let target = snapshot_attached(&state).await.remove(0);
        let msg = PeerMessage::new("agent:sender", "ghost", "hello");
        let err = deliver(&state, &target, &msg).await.unwrap_err();
        assert!(
            err.contains("ghost") && err.contains("disconnect"),
            "error should name the agent and the cause, got: {err}"
        );
    }

    #[tokio::test]
    async fn guards_are_per_recipient_not_global() {
        let (state, _t) = test_state().await;
        attach(&state, "a").await;
        attach(&state, "b").await;

        let mut guards = state.peer_guards.lock().await;
        let dup = PeerMessage::new("agent:s", "a", "same body");
        assert!(guards
            .entry("a".into())
            .or_insert_with(DeliveryGuard::new)
            .admit(&dup, 1_000)
            .is_accept());
        // The identical body to a DIFFERENT recipient is unaffected: dedupe is
        // about one channel, not about the sender saying a thing twice.
        let to_b = PeerMessage::new("agent:s", "b", "same body");
        assert!(guards
            .entry("b".into())
            .or_insert_with(DeliveryGuard::new)
            .admit(&to_b, 1_000)
            .is_accept());
    }

    #[tokio::test]
    async fn a_refused_message_still_leaves_an_audit_record() {
        let temp = tempfile::tempdir().unwrap();
        let journal = temp.path().join("peer-messages.jsonl");

        let target = PeerDescriptor {
            name: "milo".into(),
            reference: None,
            kind: PeerKind::CarAgent,
            source: PeerSource::Attached,
            address: PeerAddress::AttachedAgent {
                agent_id: "milo".into(),
            },
            display_name: None,
            capability: None,
            last_seen_ms: None,
        };
        let msg = PeerMessage::new("agent:sender", "milo", "hello");
        append_peer_audit_at(
            &journal,
            &msg,
            &target,
            &DeliveryOutcome::Refused {
                reason: "over the rate budget".into(),
            },
        );

        // Refusals are the half an operator most needs to see, so they must be
        // recorded as loudly as deliveries.
        let body = std::fs::read_to_string(&journal).expect("journal written");
        let rec: Value = serde_json::from_str(body.trim()).expect("one json line");
        assert_eq!(rec["from"], "agent:sender");
        assert_eq!(rec["to"], "milo");
        assert_eq!(rec["outcome"]["outcome"], "refused");
        assert_eq!(rec["outcome"]["reason"], "over the rate budget");
    }
}