car-server-core 0.55.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;

/// An MCP peer address is reaped after a full day without a request.
///
/// Interactive CLI sessions routinely last hours. A day leaves headroom for
/// that use while bounding abandoned HTTP sessions whose client never sends the
/// protocol DELETE request.
pub(crate) const MCP_PEER_IDLE_TTL_MS: u64 = 24 * 60 * 60 * 1000;

/// One MCP protocol session's peer state.
#[derive(Debug)]
pub(crate) struct McpPeerSession {
    pub(crate) receive_capable: bool,
    pub(crate) inbox: std::collections::VecDeque<PeerMessage>,
    pub(crate) last_seen_ms: u64,
}

fn mcp_principal(session_id: &str) -> String {
    format!("mcp:{session_id}")
}

/// Mint a protocol-session id and its unforgeable peer principal.
pub(crate) async fn open_mcp_peer_session(
    state: &ServerState,
    receive_capable: bool,
) -> (String, String) {
    prune_mcp_peer_sessions(state).await;
    let session_id = uuid::Uuid::new_v4().to_string();
    let principal = mcp_principal(&session_id);
    state.mcp_peer_sessions.lock().await.insert(
        session_id.clone(),
        McpPeerSession {
            receive_capable,
            inbox: std::collections::VecDeque::new(),
            last_seen_ms: car_peers::now_ms(),
        },
    );
    (session_id, principal)
}

/// Confirm and refresh one session, returning its peer principal.
pub(crate) async fn touch_mcp_peer_session(
    state: &ServerState,
    session_id: &str,
) -> Option<String> {
    prune_mcp_peer_sessions(state).await;
    let mut sessions = state.mcp_peer_sessions.lock().await;
    let session = sessions.get_mut(session_id)?;
    session.last_seen_ms = car_peers::now_ms();
    Some(mcp_principal(session_id))
}

/// End one MCP peer session and discard its unread inbox and channel budget.
pub(crate) async fn close_mcp_peer_session(state: &ServerState, session_id: &str) -> bool {
    let principal = mcp_principal(session_id);
    let removed = state
        .mcp_peer_sessions
        .lock()
        .await
        .remove(session_id)
        .is_some();
    if removed {
        state.peer_guards.lock().await.remove(&principal);
    }
    removed
}

async fn prune_mcp_peer_sessions(state: &ServerState) {
    let now = car_peers::now_ms();
    let expired = {
        let mut sessions = state.mcp_peer_sessions.lock().await;
        let expired: Vec<String> = sessions
            .iter()
            .filter(|(_, session)| now.saturating_sub(session.last_seen_ms) > MCP_PEER_IDLE_TTL_MS)
            .map(|(id, _)| id.clone())
            .collect();
        for id in &expired {
            sessions.remove(id);
        }
        expired
    };
    if !expired.is_empty() {
        let mut guards = state.peer_guards.lock().await;
        for id in expired {
            guards.remove(&mcp_principal(&id));
        }
    }
}

/// Snapshot receive-capable MCP sessions as addressable peers.
pub async fn snapshot_mcp_sessions(state: &ServerState) -> Vec<PeerDescriptor> {
    prune_mcp_peer_sessions(state).await;
    state
        .mcp_peer_sessions
        .lock()
        .await
        .iter()
        .filter(|(_, session)| session.receive_capable)
        .map(|(session_id, session)| {
            let principal = mcp_principal(session_id);
            PeerDescriptor {
                name: principal,
                reference: None,
                kind: PeerKind::McpSession,
                source: PeerSource::Mcp,
                address: PeerAddress::McpSession {
                    session_id: session_id.clone(),
                },
                display_name: Some("MCP session".into()),
                capability: Some("polling peer inbox".into()),
                last_seen_ms: Some(session.last_seen_ms),
                pubkey: None,
            }
        })
        .collect()
}

/// 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()),
            pubkey: None,
        })
        .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 [
        ("mcp", snapshot_mcp_sessions(state).await),
        ("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,
            // The one source that knows it: the roster is readable only with
            // the user's own bearer, so the key beside an endpoint there is one
            // they enrolled. `refresh_peer_trust` already reads the same field.
            pubkey: Some(e.pubkey).filter(|k| !k.is_empty()),
        })
        .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,
            // Deliberately absent. An advertisement can carry a key, but anyone
            // on the network can broadcast one — it is a claim, not a
            // credential, and keying a durable record on it would let a
            // stranger write into another peer's history.
            pubkey: None,
        })
        .collect()
}

/// The stable delivery preflight shared by `agents.peers` and `agents.message`.
///
/// This intentionally covers only facts on [`PeerDescriptor`]: whether the kind
/// has an inbox and whether its source is trusted. Later delivery guards depend
/// on the sender, message, standing, and point-in-time channel state, so a peer
/// that passes this snapshot can still be refused when a send is attempted.
/// Keeping these two checks here means the listing cannot advertise a peer that
/// the delivery path would reject before looking at the message.
fn peer_reachability(target: &PeerDescriptor) -> Result<(), 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()
        ));
    }

    Ok(())
}

/// Render the peer row used by `agents.peers`.
fn peer_listing_row(peer: &PeerDescriptor, standing: Option<Value>) -> Value {
    serde_json::json!({
        "standing": standing,
        "name": peer.name,
        "address": peer.address_form(),
        "reference": peer.reference,
        "kind": peer.kind.as_str(),
        "source": peer.source.as_str(),
        "can_receive": peer.kind.can_receive(),
        "reachable": peer_reachability(peer).is_ok(),
        "display_name": peer.display_name,
        "capability": peer.capability,
        "last_seen_ms": peer.last_seen_ms,
    })
}

/// `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();
    // Standing, so an operator can see WHY a peer is being throttled instead of
    // discovering it as unexplained slowness. Rendered only where a record
    // exists: absent means "nothing observed yet", which is different from a
    // clean record and should not be dressed up as one.
    let now = car_peers::now_ms();
    let standing = {
        let map = state.peer_standing.lock().await;
        peers
            .iter()
            .map(|p| {
                let key = p
                    .pubkey
                    .as_deref()
                    .map(car_a2a::peer_principal)
                    .unwrap_or_else(|| format!("agent:{}", p.name));
                map.get(&key).map(|r| {
                    serde_json::json!({
                        "state": if r.is_degraded(now) { "degraded" } else { "ok" },
                        "success": r.success_count,
                        "fail": r.fail_count,
                        "last_fail_reason": r.last_fail_reason,
                        "last_fail_via": r.last_fail_via,
                    })
                })
            })
            .collect::<Vec<_>>()
    };
    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()
            .zip(standing)
            .map(|(peer, standing)| peer_listing_row(peer, standing))
            .collect::<Vec<_>>(),
        "count": peers.len(),
    }))
}

/// `agents.message` — deliver text to one peer.
///
/// Params: `{ to, body, summary? }`. `from` is never read from params.
/// Admission for a **synchronous turn** into a live agent — `agents.chat` and
/// the A2A conversational responder.
///
/// Both reverse-call an agent through the same daemon-owned channel that
/// [`handle_agents_message`] uses, and both went through no guard and no
/// policy. So an agent set to `Deny` at the read-only tier could reach another
/// agent by chatting at it instead, and two agents could answer each other with
/// nothing to terminate the loop — the hazard [`DeliveryGuard`] exists for,
/// since neither participant is misbehaving. Tightening one method while
/// leaving its twins open just leaves the ungoverned one as the way around.
///
/// **Typed on a `&str` principal, not a `ClientSession`, and that is the
/// point.** The MCP peer-message tool already had to invent its own principal
/// because it has no session, the A2A responder has none either, and an inbound
/// remote message will be in the same position. An admission helper that can
/// only be called from a WS session is one the next caller has to route around,
/// which is how the surfaces diverged in the first place.
///
/// ## Two tiers, because callers differ in what they can prove
///
/// 1. **The channel guard applies to everyone.** Dedupe inside 10s, 20 sends
///    per sender per minute, keyed per recipient and *shared with
///    `agents.message`* — a rate budget is a property of the channel, not of
///    whichever method reached it. This is what bounds loops and floods
///    regardless of who is calling.
/// 2. **The `AgentPermissionPolicy` check applies only when the caller is a
///    bound agent** (`sender_agent`). [`admit_with`] refuses a caller that is
///    neither a bound agent nor the host, which is right for `agents.message`,
///    where every sender is a session — and wrong for the A2A responder, whose
///    caller is a remote party the operator deliberately exposed by passing
///    `share_session_runtime`. Refusing there would delete a documented
///    feature, so for that path the guard *is* the admission, and the surface's
///    own gates (loopback-only bind unless overridden) carry the rest.
///
/// **The host is exempt from both.** It is the operator's own client,
/// [`admit_with`] already passes it, and the dedupe window would otherwise
/// swallow a person legitimately retyping the same prompt inside ten seconds.
///
/// Uses [`DeliveryGuard::admit_synchronous`], which stays out of `QUEUE_CAP`
/// entirely. Attached recipients use that count for in-flight acknowledgements;
/// polling MCP recipients use it for unread inbox entries. A blocking turn
/// belongs to neither: taking a slot and releasing it immediately would let a
/// minutes-long chat cost nothing while a cheap message is counted, and holding
/// it for the turn would let fifty chats block all messaging. The
/// dedupe window and rate budget *are* shared, because those are properties of
/// the channel, and leaving them unshared is exactly what would let one method
/// be used to escape the other's limit.
///
/// The caller must report a turn that never reached the agent via
/// [`forget_synchronous_turn`], so an immediate retry is not refused as a
/// duplicate of something that was never delivered.
///
/// Every refusal is written to the peer audit journal with its outcome. That
/// matters more now that the budget is shared: an `agents.message` refused for
/// rate could have had its budget drained entirely by chat traffic, and without
/// a record there is nothing to show where it went.
pub(crate) async fn admit_turn(
    state: &ServerState,
    principal: &str,
    sender_agent: Option<String>,
    is_host: bool,
    target_agent: &str,
    body: &str,
) -> Result<(), String> {
    if is_host {
        return Ok(());
    }

    let msg = PeerMessage::new(principal, target_agent, body);
    let target = local_agent_descriptor(target_agent);

    let verdict = {
        let mut guards = state.peer_guards.lock().await;
        let guard = guards
            .entry(target_agent.to_string())
            .or_insert_with(DeliveryGuard::new);
        guard.admit_synchronous(&msg, car_peers::now_ms())
    };
    if !verdict.is_accept() {
        if matches!(
            verdict,
            car_peers::GuardVerdict::HopLimit { .. }
                | car_peers::GuardVerdict::TooLarge { .. }
                | car_peers::GuardVerdict::InvalidName { .. }
        ) {
            record_standing(state, principal, Some(&verdict.reason()), Some(&msg.via)).await;
        }
        let outcome = DeliveryOutcome::Refused {
            reason: verdict.reason(),
        };
        append_peer_audit(state, &msg, &target, &outcome);
        return Err(guard_error(&verdict));
    }

    // No bound agent means the caller cannot be graded against an agent
    // posture. See tier 2 above: that is a refusal for `agents.message` and
    // deliberately not one here.
    let Some(agent) = sender_agent else {
        return Ok(());
    };

    let outcome = admit_with(
        &crate::agent_permissions::load_policy(),
        Some(agent),
        is_host,
        principal,
    );
    match &outcome {
        DeliveryOutcome::Delivered => Ok(()),
        // Admission decides whether to *try*; it never reports on the attempt,
        // so this outcome cannot arise here. Matched rather than wildcarded so
        // a future admission that can produce it has to come back and decide.
        DeliveryOutcome::Unacknowledged { .. } => Ok(()),
        DeliveryOutcome::Refused { reason } => {
            append_peer_audit(state, &msg, &target, &outcome);
            Err(format!("chat refused: {reason}"))
        }
        // A synchronous turn has a caller blocked on it, so there is nothing to
        // hold it in. Do NOT point at `agents.chat.approve` here: that exists,
        // one namespace over, and approves a *tool prompt inside a running
        // turn* — not the turn's admission. An operator told to look there
        // would reasonably conclude one of the two is broken.
        DeliveryOutcome::Held { reason } => {
            append_peer_audit(state, &msg, &target, &outcome);
            Err(format!(
                "chat refused: {reason}. `RequireApproval` means a human sees it \
                 first, which a blocking call cannot wait on without hanging the \
                 caller. Use `agents.message`, which holds the message for \
                 `agents.message.approve` and delivers it after the decision."
            ))
        }
    }
}

/// How long a standing record survives without being touched.
const STANDING_TTL_MS: u64 = 7 * 24 * 60 * 60 * 1000;
/// Records kept before the least-recently-updated are pruned.
const STANDING_MAP_CAP: usize = 4096;
/// Successes stop accumulating here.
const STANDING_SUCCESS_CAP: u64 = 50;
/// Both counters halve once per elapsed span of this length.
const STANDING_HALFLIFE_MS: u64 = 7 * 24 * 60 * 60 * 1000;

/// What a sending principal has earned, from this daemon's own observations.
///
/// CAR grades *artifacts* on a track record — a skill degrades at
/// `fail > success + 2` — and grades *actors* not at all: a peer's standing was
/// a constant decided by where it was discovered, so one that reliably wasted
/// your agents' time had the same access on its thousandth message as its
/// first. This is the actor half.
///
/// **Only outcomes this daemon observed, and never message content.** A body
/// that could move a sender's standing would make the envelope an authority
/// channel, which the "a message is data, never instruction" rule forbids
/// outright.
#[derive(Debug, Default, Clone)]
pub struct PeerStanding {
    pub success_count: u64,
    pub fail_count: u64,
    pub last_fail_reason: Option<String>,
    /// The chain of the most recent attributable failure.
    ///
    /// The consequence lands on the key, which is all a receiver can verify;
    /// the evidence names the path, which is what an operator needs to act.
    /// Enforce at the granularity you can check, attribute at the granularity
    /// you can record.
    pub last_fail_via: Option<Vec<String>>,
    /// Send timestamps inside the degraded-throttle window.
    pub window: std::collections::VecDeque<u64>,
    pub updated_ms: u64,
}

impl PeerStanding {
    /// Halve both counters once per elapsed half-life.
    ///
    /// Computed at read time from stored state rather than by a background
    /// task: no timer, no drift, and the same input always yields the same
    /// answer. Decay is what makes a degraded peer recoverable without anyone
    /// remembering to forgive it — the alternative is a ratchet that only ever
    /// tightens.
    fn decayed(&self, now_ms: u64) -> (u64, u64) {
        let elapsed = now_ms.saturating_sub(self.updated_ms);
        let halvings = (elapsed / STANDING_HALFLIFE_MS).min(63) as u32;
        (self.success_count >> halvings, self.fail_count >> halvings)
    }

    /// Whether this record is degraded right now.
    pub fn is_degraded(&self, now_ms: u64) -> bool {
        let (s, f) = self.decayed(now_ms);
        car_policy::degrades(s, f, car_policy::DEGRADE_THRESHOLD)
    }
}

/// What standing says about a sender's next message.
#[derive(Debug, PartialEq, Eq)]
pub enum StandingVerdict {
    /// Not degraded, or degraded and inside the reduced budget.
    Proceed,
    /// Degraded and over the reduced budget.
    Throttled { reason: String },
}

/// The standing check, host-scoped and run **before** the per-recipient guard.
///
/// That ordering is what makes laundering fail. `DeliveryGuard.rate` is keyed
/// on the sender *inside* a per-recipient guard, so a degraded host that
/// renamed its agents — or simply addressed a different recipient — would
/// otherwise collect a fresh budget per name per recipient. Standing is keyed
/// on the principal this daemon actually verified.
///
/// A healthy sender is untouched: this adds no aggregate ceiling to normal
/// traffic. Only a degraded one meets [`car_peers::DEGRADED_RATE_LIMIT`].
///
/// Autonomous by design. Turning each reduction into an approval prompt would
/// rebuild the approval fatigue that drives operators to switch a gate off
/// entirely — the mitigation *is* the answer, which is the lesson
/// `skill_trust`'s own deployment gate already encodes.
pub async fn standing_gate(state: &ServerState, key: &str) -> StandingVerdict {
    let now = car_peers::now_ms();
    let mut map = state.peer_standing.lock().await;
    let Some(rec) = map.get_mut(key) else {
        return StandingVerdict::Proceed;
    };
    if !rec.is_degraded(now) {
        return StandingVerdict::Proceed;
    }
    while rec
        .window
        .front()
        .is_some_and(|t| now.saturating_sub(*t) > car_peers::RATE_WINDOW_MS)
    {
        rec.window.pop_front();
    }
    if rec.window.len() as u32 >= car_peers::DEGRADED_RATE_LIMIT {
        return StandingVerdict::Throttled {
            reason: format!(
                "`{key}` is degraded ({} failures against {} successes) and is \
                 limited to {} messages per minute until its record recovers",
                rec.fail_count,
                rec.success_count,
                car_peers::DEGRADED_RATE_LIMIT
            ),
        };
    }
    rec.window.push_back(now);
    StandingVerdict::Proceed
}

/// Record an outcome against a sender's standing.
///
/// `failure` is `Some(reason)` only for outcomes attributable to the sender's
/// **own choices** — a chain too deep, an oversized body, a malformed name or
/// lineage segment, an unresolvable recipient, an operator's explicit denial.
///
/// Deliberately excluded: a rate limit or a duplicate inside the window, both
/// of which the channel guard exists to absorb precisely because *neither party
/// is misbehaving* in a mutual loop — charging them would price correct
/// behaviour as misconduct. Also excluded: a refusal by this host's own policy
/// (our posture, not their conduct) and an unacknowledged delivery (the
/// recipient's own agent did not answer).
pub async fn record_standing(
    state: &ServerState,
    key: &str,
    failure: Option<&str>,
    via: Option<&[String]>,
) {
    let now = car_peers::now_ms();
    let mut map = state.peer_standing.lock().await;
    if map.len() >= STANDING_MAP_CAP && !map.contains_key(key) {
        // Prune the least recently updated, and anything past its TTL.
        map.retain(|_, r| now.saturating_sub(r.updated_ms) < STANDING_TTL_MS);
        if map.len() >= STANDING_MAP_CAP {
            if let Some(oldest) = map
                .iter()
                .min_by_key(|(_, r)| r.updated_ms)
                .map(|(k, _)| k.clone())
            {
                map.remove(&oldest);
            }
        }
    }
    let rec = map.entry(key.to_string()).or_default();
    // Fold the decay in before recording, so an old record does not carry its
    // full weight forward the moment it is touched again.
    let (s, f) = rec.decayed(now);
    rec.success_count = s;
    rec.fail_count = f;
    match failure {
        Some(reason) => {
            rec.fail_count = rec.fail_count.saturating_add(1);
            rec.last_fail_reason = Some(reason.to_string());
            rec.last_fail_via = via.map(|v| v.to_vec());
        }
        None => {
            // Saturating, not unbounded. Without a ceiling a peer with ten
            // thousand successes could send ten thousand attributable failures
            // before a throttle engaged — tolerable for an artifact whose
            // numerator the host controls, indefensible for an actor that
            // controls its own send rate. Headroom is the cap plus the
            // threshold, whatever the history.
            rec.success_count = rec
                .success_count
                .saturating_add(1)
                .min(STANDING_SUCCESS_CAP);
        }
    }
    rec.updated_ms = now;
}

/// The receiving half of the two-admission property.
///
/// `PeerAddress` has no variant naming a remote agent precisely so that a
/// cross-host message must land on a *daemon* and be admitted there before it
/// reaches anyone. Three doc comments asserted that happened. Nothing did it:
/// an inbound peer message was compiled into an `ActionProposal` like any other
/// A2A traffic and answered with an acknowledgement stub, so a message that had
/// passed the sender's guard and policy passed nothing on arrival.
///
/// This is that missing admission. It deliberately does **not** consult
/// `AgentPermissionPolicy`: those rows are keyed by *local* agent id, there is
/// no local sender to resolve inbound, and letting a remote-supplied name
/// select which local posture governs it would be a free pass around any
/// `Deny` — per-agent rows are sparse overrides over a permissive default, so
/// naming an unlisted id would land on "allow". The operator's per-peer control
/// is the trust set the middleware already enforced, and later an explicit
/// per-peer record; it is not this function guessing from a string the
/// counterparty wrote.
pub struct PeerInboundBroker {
    /// Weak so a stopped listener cannot keep the daemon's state alive — the
    /// same reasoning as the fleet responder wired beside it.
    pub state: std::sync::Weak<ServerState>,
}

/// Append this host's boundary marker to the chain a peer attested.
///
/// APPEND, never substitute. Everything before the marker is what the sending
/// key *claimed*; everything from it on is what a daemon *observed*. Collapsing
/// the two — by replacing the prefix, or by trusting the claim unmarked — is
/// what makes a chain unauditable, because a later reader can no longer tell
/// which segments any host actually stood behind.
///
/// Split out of `deliver` so the property has somewhere to be asserted. It was
/// previously inline, and the test named for it
/// (`the_receiver_appends_its_own_boundary_marker`) passed with the append
/// deleted — it checked that a guard entry appeared, which happens either way.
/// The end-to-end test now reads the configured state's audit row and asserts
/// this receiver-authored marker directly.
fn stamp_boundary(attested: &[String], marker: &str) -> Vec<String> {
    let mut via = attested.to_vec();
    via.push(marker.to_string());
    via
}

#[async_trait::async_trait]
impl car_a2a::PeerInbox for PeerInboundBroker {
    async fn deliver(
        &self,
        inbound: car_a2a::InboundPeerMessage,
    ) -> Result<serde_json::Value, String> {
        let state = self
            .state
            .upgrade()
            .ok_or_else(|| "daemon is shutting down".to_string())?;

        // The sender is the key the signature proved, never `carPeerFrom`. A
        // receiving broker that used the claimed name would attribute a message
        // on the strength of a string the counterparty minted — and the local
        // module doc's own rule is that a caller-supplied sender makes the
        // audit forgeable.
        let from = car_a2a::peer_principal(&inbound.peer_pubkey);

        // Standing first, host-scoped, before anything per-recipient. A
        // degraded host that renamed its agents or simply picked a different
        // recipient would otherwise collect a fresh budget each time, because
        // the channel guard's rate bucket lives inside a per-recipient guard.
        if let StandingVerdict::Throttled { reason } = standing_gate(&state, &from).await {
            return Err(reason);
        }

        if !car_peers::is_valid_peer_name(&inbound.claimed.to) {
            record_standing(&state, &from, Some("illegal recipient name"), None).await;
            return Err(format!("`{}` is not a legal peer name", inbound.claimed.to));
        }

        // Inbound lineage is remote-supplied text that this host is about to
        // show one of its own agents, and until here nothing in CAR had ever
        // validated it. Bound the shape before an agent sees it; a segment that
        // is merely *false* is not detectable at all, since everything before a
        // boundary marker is the far side's account of itself.
        if inbound.claimed.via.len() > car_peers::MAX_HOPS * 2 {
            record_standing(&state, &from, Some("oversized lineage"), None).await;
            return Err(format!(
                "chain carries {} segments, over the {} the hop cap can produce",
                inbound.claimed.via.len(),
                car_peers::MAX_HOPS * 2
            ));
        }
        if let Some(bad) = inbound
            .claimed
            .via
            .iter()
            .find(|s| !car_peers::is_valid_via_segment(s))
        {
            let bad = bad.clone();
            record_standing(&state, &from, Some("malformed lineage segment"), None).await;
            return Err(format!("`{bad}` is not a well-formed lineage segment"));
        }
        if inbound.claimed.trace.len() > 128 {
            return Err("chain id is too long".to_string());
        }

        // Attached agents only. Resolving through the full directory would let
        // this host forward to a peer it knows — turning it into an open relay
        // where host A makes host B deliver to host C under B's signature,
        // reaching hosts that never trusted A. The refusal is explicit rather
        // than a resolution failure so the sender learns the rule.
        let target = snapshot_attached(&state)
            .await
            .into_iter()
            .find(|p| p.name == inbound.claimed.to)
            .ok_or_else(|| {
                format!(
                    "`{}` is not an agent attached to this host; peer messages are \
                     delivered to local agents only and are never relayed",
                    inbound.claimed.to
                )
            });
        let target = match target {
            Ok(t) => t,
            Err(e) => {
                // Attributable: the sender chose an address this host does not
                // serve, or tried to have it relayed.
                record_standing(&state, &from, Some("unresolvable recipient"), None).await;
                return Err(e);
            }
        };

        // The id is the sender's, reused so both hosts' audit rows correlate —
        // which is exactly why it is bounded before it reaches this host's
        // journal and the `agent.peer_message` frame.
        if inbound.claimed.message_id.len() > 128 {
            return Err("message id is too long".to_string());
        }

        let mut msg = PeerMessage::new(&from, &target.name, &inbound.claimed.body);
        msg.id = inbound.claimed.message_id;
        // Stamp the chain BEFORE the guard runs. `PeerMessage::new` leaves
        // `via` empty, so admitting first would mean `hops()` was 0 on every
        // inbound message and `HopLimit` could never fire on cross-host traffic
        // — the one case it exists for.
        //
        // The boundary marker is APPENDED, never substituted for the prefix.
        // Everything before it is what this key *attested*; everything after is
        // what a daemon observed. That is what lets a later host tell the two
        // apart, and it is why the marker is stamped here — by the receiver,
        // from the key it verified — and never by the sender.
        msg.trace = if inbound.claimed.trace.is_empty() {
            msg.id.clone()
        } else {
            inbound.claimed.trace.clone()
        };
        msg.via = stamp_boundary(&inbound.claimed.via, &from);
        // `no_reply` is forced, not carried. `PeerMessage.from` is documented as
        // the address a recipient replies to by copying back — and `peer:<key>`
        // is not one: it contains `:` and base64's `+`/`/`, so
        // `is_valid_peer_name` rejects it and no provider resolves it. It is an
        // *attribution*, which is what an inbound message can honestly offer,
        // since `PeerAddress` has no variant naming a remote agent to reply to.
        // Telling the agent it may reply and handing it an unresolvable string
        // would be worse than saying so.
        msg.no_reply = true;

        // The same channel guard the local path applies, on the same
        // per-recipient budget. A remote sender that loops is bounded by the
        // recipient's channel, not by whatever the far side chose to enforce.
        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() {
            // A rate limit or an in-window duplicate is NOT charged: the guard
            // exists because in a mutual loop neither party is misbehaving, and
            // pricing that as misconduct would penalise correct behaviour. What
            // is charged is what the sender chose — a chain too deep, an
            // oversized body, an illegal name.
            let attributable = matches!(
                verdict,
                car_peers::GuardVerdict::HopLimit { .. }
                    | car_peers::GuardVerdict::TooLarge { .. }
                    | car_peers::GuardVerdict::InvalidName { .. }
            );
            if attributable {
                record_standing(&state, &from, Some(&verdict.reason()), Some(&msg.via)).await;
            }
            let outcome = DeliveryOutcome::Refused {
                reason: verdict.reason(),
            };
            append_peer_audit_dir(
                &state,
                &msg,
                &target,
                &outcome,
                PeerAuditDir::In,
                Some(&inbound.peer_pubkey),
            );
            return Err(guard_error(&verdict));
        }

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

        // Do not flatten. `deliver` reports `unacknowledged` when the frame was
        // written but the agent never answered inside the ack window, and
        // collapsing that to `delivered` would put a delivery that did not
        // happen into the inbound audit row and send the same false verdict
        // back across the hop.
        let reported = match &result {
            Ok(v) => v
                .get("outcome")
                .and_then(|o| o.as_str())
                .unwrap_or("delivered")
                .to_string(),
            Err(_) => "failed".to_string(),
        };
        // Do not squeeze `unacknowledged` through `Refused`. That variant's
        // contract is "dropped, never delivered", and the frame demonstrably
        // was written — recording it as a refusal would make the receiving
        // host's journal contradict the sending host's report about the same
        // message, with the durability claim pointing the wrong way.
        let outcome = match &result {
            Ok(_) if reported == "delivered" => DeliveryOutcome::Delivered,
            Ok(v) => DeliveryOutcome::Unacknowledged {
                detail: v
                    .get("detail")
                    .and_then(|d| d.as_str())
                    .unwrap_or(reported.as_str())
                    .to_string(),
            },
            Err(reason) => DeliveryOutcome::Refused {
                reason: reason.clone(),
            },
        };
        append_peer_audit_dir(
            &state,
            &msg,
            &target,
            &outcome,
            PeerAuditDir::In,
            Some(&inbound.peer_pubkey),
        );

        // A message that was admitted and then could not be delivered gives
        // back its dedupe record. Otherwise the sender's honest retry is
        // refused as "already delivered" — false, and it is on another host,
        // so it cannot see why.
        if result.is_err() {
            if let Some(g) = state.peer_guards.lock().await.get_mut(&target.name) {
                g.forget(&msg);
            }
        }
        // A success is an acknowledged delivery, nothing weaker. An
        // `unacknowledged` result is the RECIPIENT's agent not answering, which
        // says nothing about the sender's conduct, so it moves no counter in
        // either direction.
        if matches!(outcome, DeliveryOutcome::Delivered) {
            record_standing(&state, &from, None, None).await;
        }

        // Hand back what `deliver` actually reported, structure intact, so the
        // sending host records the same verdict this one did rather than a
        // lossy rendering of it.
        result
    }
}

/// Undo the dedupe record for a synchronous turn that was admitted and then
/// could not be delivered — the agent is not attached, or it raced a
/// disconnect. Without it, an immediate retry is refused as a duplicate and
/// told the message "was already delivered", which is false.
pub(crate) async fn forget_synchronous_turn(
    state: &ServerState,
    principal: &str,
    target_agent: &str,
    body: &str,
) {
    let msg = PeerMessage::new(principal, target_agent, body);
    if let Some(g) = state.peer_guards.lock().await.get_mut(target_agent) {
        g.forget(&msg);
    }
}

/// A [`PeerDescriptor`] for a local attached agent, so a synchronous turn's
/// refusal lands in the same audit journal, in the same shape, as a refused
/// `agents.message`. One reader, one format.
fn local_agent_descriptor(agent_id: &str) -> PeerDescriptor {
    PeerDescriptor {
        name: agent_id.to_string(),
        reference: None,
        kind: car_peers::PeerKind::CarAgent,
        source: car_peers::PeerSource::Attached,
        address: car_peers::PeerAddress::AttachedAgent {
            agent_id: agent_id.to_string(),
        },
        display_name: None,
        capability: None,
        last_seen_ms: Some(car_peers::now_ms()),
        // A local agent has no signing key; its standing keys on `agent:<id>`.
        pubkey: None,
    }
}

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())?;

    peer_reachability(&target)?;

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

    // Stage 0: standing. Host-scoped and ahead of the per-recipient guard, for
    // the same reason it runs first inbound — a degraded sender must not be
    // able to collect a fresh budget by switching recipients.
    if let StandingVerdict::Throttled { reason } = standing_gate(state, &from).await {
        return Err(reason);
    }

    // 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(state, &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(state, &msg, &target, &outcome);
    match &outcome {
        DeliveryOutcome::Refused { reason } => {
            release(state, &target.name).await;
            return Err(format!("message refused: {reason}"));
        }
        // `admit` reports on whether to try, never on the attempt, so this
        // cannot arise from it. Matched rather than wildcarded so an admission
        // that could one day produce it must come back and choose.
        DeliveryOutcome::Unacknowledged { .. } => {}
        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
            // outstanding delivery, 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(
                    state,
                    &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;
    settle_delivery_slot(state, &target, result.is_ok()).await;

    // A second row, on the ATTEMPT. The row above records the admission — that
    // this host decided to try — and until a far side could refuse, that was
    // the whole story. It is not any more: a cross-host message the remote
    // broker rejects returns `Err` here, and with only the admission row the
    // sending operator's journal would say `Delivered` about a message the
    // other host threw away. Two rows for one message is the honest shape,
    // because two decisions were made, in two places, and an incident needs
    // both.
    match &result {
        Ok(v) => {
            let reported = v.get("outcome").and_then(|o| o.as_str()).unwrap_or("");
            // `delivered` is already implied by the admission row; only record
            // an attempt that ended somewhere else.
            if reported != "delivered" {
                append_peer_audit(
                    state,
                    &msg,
                    &target,
                    &DeliveryOutcome::Unacknowledged {
                        detail: v
                            .get("detail")
                            .and_then(|d| d.as_str())
                            .unwrap_or(reported)
                            .to_string(),
                    },
                );
            }
        }
        Err(reason) => append_peer_audit(
            state,
            &msg,
            &target,
            &DeliveryOutcome::Refused {
                reason: reason.clone(),
            },
        ),
    }
    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 {
        // Ground truth. Every other failure signal is this runtime inferring
        // misconduct from shape; here a human looked at the message and said
        // no, which is the strongest evidence standing can have.
        record_standing(
            state,
            &held.message.from,
            Some("denied by the operator"),
            Some(&held.message.via),
        )
        .await;
        append_peer_audit(
            state,
            &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(
            state,
            &held.message,
            &held.target,
            &DeliveryOutcome::Refused {
                reason: verdict.reason(),
            },
        );
        return Err(guard_error(&verdict));
    }

    append_peer_audit(
        state,
        &held.message,
        &held.target,
        &DeliveryOutcome::Delivered,
    );
    let result = deliver(state, &held.target, &held.message).await;
    settle_delivery_slot(state, &held.target, result.is_ok()).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 outstanding-delivery count.
async fn release(state: &ServerState, recipient: &str) {
    if let Some(g) = state.peer_guards.lock().await.get_mut(recipient) {
        g.consumed();
    }
}

/// Release an in-flight slot unless a successful MCP enqueue now owns it.
///
/// Attached/A2A delivery slots last only for the round trip. An MCP inbox slot
/// represents an unread queued message and is released by `peer_inbox` instead.
async fn settle_delivery_slot(state: &ServerState, target: &PeerDescriptor, delivered: bool) {
    if !delivered || !matches!(target.address, PeerAddress::McpSession { .. }) {
        release(state, &target.name).await;
    }
}

/// 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. Inbound, this budget is \
                 per remote DAEMON, not per remote agent: the host key is the \
                 only principal a receiver can verify, so every agent on that \
                 host shares it.",
                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(),
        GuardVerdict::HopLimit { .. } => {
            format!(
                "{} — this chain has been forwarded far enough; act on it or \
                 answer the originator directly rather than passing it on",
                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::McpSession { session_id } => {
            return enqueue_mcp_message(state, session_id, target, msg).await;
        }
        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"
                ),
            }))
        }
    }
}

/// Queue one admitted message for a live MCP session to poll.
///
/// The caller has already taken a [`DeliveryGuard`] slot. Unlike the attached
/// WebSocket path, this slot remains occupied until `peer_inbox` drains the
/// message, so [`car_peers::QUEUE_CAP`] is a real unread-inbox bound here.
async fn enqueue_mcp_message(
    state: &ServerState,
    session_id: &str,
    target: &PeerDescriptor,
    msg: &PeerMessage,
) -> Result<Value, String> {
    let mut sessions = state.mcp_peer_sessions.lock().await;
    let session = sessions
        .get_mut(session_id)
        .ok_or_else(|| format!("MCP session `{}` disconnected before delivery", target.name))?;
    if !session.receive_capable {
        return Err(format!("MCP session `{}` is send-only", target.name));
    }
    session.inbox.push_back(msg.clone());
    Ok(serde_json::json!({
        "id": msg.id,
        "to": target.name,
        "outcome": "delivered",
        "delivery": "queued_for_poll",
    }))
}

/// Drain messages for the MCP session on the current request.
async fn drain_mcp_inbox(
    state: &ServerState,
    principal: &str,
    limit: usize,
) -> Result<Value, car_mcp::ToolError> {
    let session_id = principal
        .strip_prefix("mcp:")
        .ok_or_else(|| car_mcp::ToolError::Internal("invalid MCP peer principal".into()))?;

    let messages = {
        let mut sessions = state.mcp_peer_sessions.lock().await;
        let session = sessions.get_mut(session_id).ok_or_else(|| {
            car_mcp::ToolError::Internal("MCP peer session expired; reconnect".into())
        })?;
        if !session.receive_capable {
            return Err(car_mcp::ToolError::Internal(
                "CAR-spawned batch CLI sessions are send-only".into(),
            ));
        }
        session.last_seen_ms = car_peers::now_ms();
        let take = limit.min(session.inbox.len());
        session.inbox.drain(..take).collect::<Vec<_>>()
    };

    if !messages.is_empty() {
        let mut guards = state.peer_guards.lock().await;
        if let Some(guard) = guards.get_mut(principal) {
            for _ in 0..messages.len() {
                guard.consumed();
            }
        }
    }
    Ok(serde_json::json!({
        "self": principal,
        "messages": messages,
        "count": messages.len(),
    }))
}

/// 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 is what makes a second admission *possible* on the far side, and it is
/// why `PeerAddress` has no variant for a remote agent.
///
/// That second admission now exists: [`PeerInboundBroker`] reads `carPeerTo`,
/// rebuilds the sender from the *verified* signing key rather than the name the
/// caller claimed, applies the recipient's own channel guard, refuses to relay
/// onward, and writes an inbound audit row. So a `Delivered` outcome here means
/// the far side's broker admitted the message and its agent was reverse-called
/// — not merely that the bytes were accepted.
///
/// What it still does not mean: that the far side graded the sender against any
/// per-peer posture. There is no local sender to resolve inbound, and a
/// remote-supplied name must never select which local policy row governs it, so
/// the operator's control there remains the trust set — not a policy lookup.
///
/// 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();
    // Sent for operator-visible sender labelling. The receiving broker
    // deliberately does NOT read it for routing or attribution — it rebuilds
    // the sender from the key it verified — so this is a display string, not a
    // credential, and it is written with the constant so the two ends of the
    // wire cannot drift apart on spelling.
    metadata.insert(
        car_a2a::PEER_FROM_KEY.to_string(),
        Value::from(msg.from.clone()),
    );
    // Lineage crosses the hop here, inside the body the signature hashes, so a
    // proxy cannot alter it in flight. What the far side does with it is
    // attest-not-verify: it appends its own boundary marker rather than
    // trusting this prefix.
    if !msg.trace.is_empty() {
        metadata.insert(
            car_a2a::PEER_TRACE_KEY.to_string(),
            Value::from(msg.trace.clone()),
        );
    }
    if !msg.via.is_empty() {
        metadata.insert(
            car_a2a::PEER_VIA_KEY.to_string(),
            Value::from(msg.via.clone()),
        );
    }
    metadata.insert(
        car_a2a::PEER_TO_KEY.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 {
        // Read the far side's verdict rather than assuming one. A receiving
        // broker answers `carPeerOutcome`, and it is not always `delivered`:
        // an agent that never acknowledges inside the ack window yields
        // `unacknowledged` there, and writing `delivered` here would record a
        // delivery that did not happen — on the one journal an operator reads
        // during an incident. A peer that predates the broker sends no outcome
        // at all, so the fallback is the honest, weaker claim.
        Ok(result) => {
            // The far side's verdict, structure intact. A peer that predates
            // the broker sends none, and `accepted` is then the honest weaker
            // claim: the bytes were taken, and nothing is known about what
            // happened to them.
            let reported = match &result {
                car_a2a::types::SendMessageResult::Message(m) => {
                    m.metadata.get(car_a2a::PEER_OUTCOME_KEY).cloned()
                }
                _ => None,
            };
            let mut out = serde_json::json!({
                "id": msg.id,
                "to": target.name,
                "outcome": "accepted",
                "remote_reported": reported.is_some(),
                "transport": "a2a",
                "url": base_url,
            });
            if let Some(remote) = reported {
                if let Some(obj) = remote.as_object() {
                    for (k, v) in obj {
                        out[k.as_str()] = v.clone();
                    }
                } else if let Some(s) = remote.as_str() {
                    // A peer on the earlier flat contract.
                    out["outcome"] = serde_json::Value::from(s);
                }
            }
            Ok(out)
        }
        Err(e) => Err(format!(
            "failed to deliver to `{}` at {base_url}: {e}",
            target.name
        )),
    }
}

/// Append a peer-message record to the configured state's 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. The path was resolved when
/// `ServerState` was built; this write must never re-read process-global
/// `CAR_HOME`.
pub fn append_peer_audit(
    state: &ServerState,
    msg: &PeerMessage,
    target: &PeerDescriptor,
    outcome: &DeliveryOutcome,
) {
    append_peer_audit_dir(state, msg, target, outcome, PeerAuditDir::Out, None);
}

/// Which way a message was travelling when this row was written.
///
/// Every row CAR wrote before the receiving-side broker existed was outbound —
/// there was no inbound admission to record. A reader of
/// `~/.car/peer-messages.jsonl` therefore cannot tell "we sent this" from "a
/// peer sent us this" without the column, and the two are very different
/// questions during an incident.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PeerAuditDir {
    /// This host sent it.
    Out,
    /// This host received and admitted (or refused) it.
    In,
}

impl PeerAuditDir {
    fn as_str(self) -> &'static str {
        match self {
            PeerAuditDir::Out => "out",
            PeerAuditDir::In => "in",
        }
    }
}

/// [`append_peer_audit`] with an explicit direction and, inbound, the key that
/// was actually verified.
///
/// `attested_by` is the point of the inbound row. `msg.from` is derived from
/// that key rather than from anything the sender claimed, so recording the key
/// beside it lets an operator confirm the derivation instead of trusting it.
pub fn append_peer_audit_dir(
    state: &ServerState,
    msg: &PeerMessage,
    target: &PeerDescriptor,
    outcome: &DeliveryOutcome,
    dir: PeerAuditDir,
    attested_by: Option<&str>,
) {
    if let Some(parent) = state.peer_audit_journal.parent() {
        if std::fs::create_dir_all(parent).is_err() {
            return;
        }
    }
    append_peer_audit_at_dir(
        &state.peer_audit_journal,
        msg,
        target,
        outcome,
        dir,
        attested_by,
    );
}

/// [`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,
) {
    append_peer_audit_at_dir(path, msg, target, outcome, PeerAuditDir::Out, None);
}

/// [`append_peer_audit_at`] with an explicit direction and attestation.
pub fn append_peer_audit_at_dir(
    path: &std::path::Path,
    msg: &PeerMessage,
    target: &PeerDescriptor,
    outcome: &DeliveryOutcome,
    dir: PeerAuditDir,
    attested_by: Option<&str>,
) {
    use std::io::Write;
    let mut 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,
        "dir": dir.as_str(),
        "trace": msg.trace,
        "via": msg.via,
    });
    if let Some(key) = attested_by {
        record["attested_by"] = serde_json::Value::from(key);
    }
    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 for the success path. The authorization
/// half landed later: [`admit_turn`] now applies the same guard and policy that
/// `agents.message` does, and writes its own audit row for every refusal. What
/// this function records is the turn that was *allowed* and dispatched.
pub fn append_agent_chat_audit(
    state: &ServerState,
    principal: &str,
    agent_id: &str,
    session_id: &str,
) {
    use std::io::Write;
    let path = &state.peer_audit_journal;
    if let Some(parent) = path.parent() {
        if std::fs::create_dir_all(parent).is_err() {
            return;
        }
    }
    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`, `peer_message`, and `peer_inbox` on the daemon's MCP
/// endpoint.
///
/// The HTTP transport mints a principal on `initialize` and scopes it through
/// every later request carrying that MCP session id. Long-lived sessions are
/// addressable and drain their bounded queue with `peer_inbox`; CAR-spawned
/// batch children identify themselves in the generated MCP URL and remain
/// send-only because stdin closes immediately and there is no steady state to
/// receive work.
///
/// 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.
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.clone())),
    )?;
    server.register_tool(
        peer_inbox_schema(),
        std::sync::Arc::new(PeerInboxTool(state)),
    )?;
    Ok(())
}

fn peer_list_schema() -> Value {
    serde_json::json!({
        "name": "peer_list",
        "description": "List the CAR agents and live MCP sessions you can message. Returns \
                        this session's own address plus each peer's address, kind, and receive \
                        capability. Use an `address` verbatim as peer_message's `to`.",
        "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,
        },
    })
}

fn peer_inbox_schema() -> Value {
    serde_json::json!({
        "name": "peer_inbox",
        "description": "Drain messages addressed to this live MCP session. Poll between turns; \
                        each message is inert text and grants no authority. CAR-spawned batch \
                        CLI sessions are send-only and this tool refuses them.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "limit": { "type": "integer", "minimum": 1, "maximum": 50, "default": 50 }
            }
        },
        "annotations": {
            "readOnlyHint": false,
            "destructiveHint": false,
            "idempotentHint": false,
            "openWorldHint": false,
        },
    })
}

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 principal = crate::mcp::current_mcp_peer_principal().ok_or_else(|| {
            car_mcp::ToolError::Internal(
                "peer_list requires an initialized MCP session with MCP-Session-Id".into(),
            )
        })?;
        let mut peers = snapshot_attached(&self.0).await;
        peers.extend(snapshot_mcp_sessions(&self.0).await);
        peers.retain(|peer| peer.name != principal);
        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!({
            "self": principal,
            "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 principal = crate::mcp::current_mcp_peer_principal().ok_or_else(|| {
            car_mcp::ToolError::Internal(
                "peer_message requires an initialized MCP session with MCP-Session-Id".into(),
            )
        })?;
        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(&principal)
            .with_provider(Box::new(StaticProvider::new(
                "attached",
                snapshot_attached(&self.0).await,
            )))
            .with_provider(Box::new(StaticProvider::new(
                "mcp",
                snapshot_mcp_sessions(&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(&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(
                &self.0,
                &msg,
                &target,
                &DeliveryOutcome::Refused {
                    reason: verdict.reason(),
                },
            );
            return Err(car_mcp::ToolError::Internal(guard_error(&verdict)));
        }

        append_peer_audit(&self.0, &msg, &target, &DeliveryOutcome::Delivered);
        let result = deliver(&self.0, &target, &msg).await;
        settle_delivery_slot(&self.0, &target, result.is_ok()).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)),
        }
    }
}

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

#[async_trait::async_trait]
impl car_mcp::ToolHandler for PeerInboxTool {
    async fn call(&self, args: Value) -> Result<String, car_mcp::ToolError> {
        let limit = args.get("limit").and_then(Value::as_u64).unwrap_or(50);
        if !(1..=50).contains(&limit) {
            return Err(car_mcp::ToolError::InvalidParams(
                "`limit` must be between 1 and 50".into(),
            ));
        }
        let principal = crate::mcp::current_mcp_peer_principal().ok_or_else(|| {
            car_mcp::ToolError::Internal(
                "peer_inbox requires an initialized MCP session with MCP-Session-Id".into(),
            )
        })?;
        let value = drain_mcp_inbox(&self.0, &principal, limit as usize).await?;
        serde_json::to_string(&value).map_err(|e| car_mcp::ToolError::Internal(e.to_string()))
    }
}

#[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,
                pubkey: 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 one_mcp_session_receives_only_its_own_addressed_message() {
        let (state, _t) = test_state().await;
        let (first_id, first) = open_mcp_peer_session(&state, true).await;
        let (_second_id, second) = open_mcp_peer_session(&state, true).await;
        let peers = snapshot_mcp_sessions(&state).await;
        let target = peers
            .iter()
            .find(|peer| peer.name == first)
            .expect("first session is addressable")
            .clone();

        let msg = PeerMessage::new("agent:sender", &first, "only for first");
        let verdict = state
            .peer_guards
            .lock()
            .await
            .entry(first.clone())
            .or_insert_with(DeliveryGuard::new)
            .admit(&msg, 1_000);
        assert_eq!(verdict, GuardVerdict::Accept);
        let result = deliver(&state, &target, &msg).await;
        assert!(result.is_ok(), "queue delivery failed: {result:?}");
        settle_delivery_slot(&state, &target, result.is_ok()).await;

        let drained = drain_mcp_inbox(&state, &first, 50).await.unwrap();
        assert_eq!(drained["count"], 1);
        assert_eq!(drained["messages"][0]["body"], "only for first");
        assert_eq!(drained["messages"][0]["to"], first);
        assert_eq!(
            state.mcp_peer_sessions.lock().await[&first_id].inbox.len(),
            0
        );
        assert_eq!(
            drain_mcp_inbox(&state, &second, 50).await.unwrap()["count"],
            0,
            "a message addressed to the first session must not leak to the second"
        );
        assert_eq!(state.peer_guards.lock().await[&first].queued(), 0);
    }

    #[tokio::test]
    async fn queue_guards_are_scoped_per_mcp_session() {
        let (state, _t) = test_state().await;
        let (_, first) = open_mcp_peer_session(&state, true).await;
        let (_, second) = open_mcp_peer_session(&state, true).await;
        let mut guards = state.peer_guards.lock().await;

        for i in 0..car_peers::QUEUE_CAP {
            let msg = PeerMessage::new(format!("agent:s{i}"), &first, format!("body-{i}"));
            assert!(guards
                .entry(first.clone())
                .or_insert_with(DeliveryGuard::new)
                .admit(&msg, 1_000)
                .is_accept());
        }
        let blocked = PeerMessage::new("agent:fresh", &first, "over cap");
        assert_eq!(
            guards.get_mut(&first).unwrap().admit(&blocked, 1_000),
            GuardVerdict::QueueFull {
                cap: car_peers::QUEUE_CAP
            }
        );

        let other = PeerMessage::new("agent:fresh", &second, "over cap");
        assert!(
            guards
                .entry(second)
                .or_insert_with(DeliveryGuard::new)
                .admit(&other, 1_000)
                .is_accept(),
            "one session's full inbox must not consume another's queue budget"
        );
    }

    #[tokio::test]
    async fn batch_mcp_sessions_remain_send_only() {
        let (state, _t) = test_state().await;
        let (batch_id, batch) = open_mcp_peer_session(&state, false).await;
        assert!(snapshot_mcp_sessions(&state)
            .await
            .iter()
            .all(|peer| peer.name != batch));
        let error = drain_mcp_inbox(&state, &batch, 50).await.unwrap_err();
        assert!(error.message().contains("send-only"), "{error:?}");
        assert!(state.mcp_peer_sessions.lock().await.contains_key(&batch_id));
    }

    #[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");
    }

    #[test]
    fn listing_reachability_matches_the_delivery_preflight() {
        let mut peer = PeerDescriptor {
            name: "discovered-mac".into(),
            reference: None,
            kind: PeerKind::RemoteCar,
            source: PeerSource::Lan,
            address: PeerAddress::A2a {
                base_url: "https://peer.invalid".into(),
            },
            display_name: None,
            capability: None,
            last_seen_ms: None,
            pubkey: None,
        };

        let discovered = peer_listing_row(&peer, None);
        assert_eq!(discovered["can_receive"], true, "remote CAR has an inbox");
        assert_eq!(
            discovered["reachable"], false,
            "an untrusted LAN advertisement must not be offered for delivery"
        );
        let refusal = peer_reachability(&peer).expect_err("delivery must refuse the same peer");
        assert!(refusal.contains("not a trusted peer"), "{refusal}");

        // Trust is a pure function of the source already present on the row. A
        // trusted remote descriptor therefore flips the listing and the send
        // preflight together without changing its kind-level capability.
        peer.source = PeerSource::Parslee;
        let trusted = peer_listing_row(&peer, None);
        assert_eq!(trusted["can_receive"], true);
        assert_eq!(trusted["reachable"], true);
        assert!(peer_reachability(&peer).is_ok());

        // The other preflight guard is represented too: a trusted source does
        // not make an external batch CLI grow an inbox.
        peer.kind = PeerKind::ExternalCli;
        peer.source = PeerSource::Invocation;
        let no_inbox = peer_listing_row(&peer, None);
        assert_eq!(no_inbox["can_receive"], false);
        assert_eq!(no_inbox["reachable"], false);
        let refusal = peer_reachability(&peer).expect_err("delivery must refuse no-inbox kinds");
        assert!(refusal.contains("no inbox"), "{refusal}");
    }

    #[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,
            pubkey: 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");
    }

    /// `agents.chat` must not be the way around `agents.message`'s admission.
    /// A denied agent is denied on both, and the refusal names chat so the
    /// caller is not left guessing which surface stopped it.
    #[test]
    fn a_denied_agent_cannot_reach_another_agent_by_chatting_instead() {
        let mut policy = car_policy::AgentPermissionPolicy::default();
        policy.set_agent(
            "noisy",
            car_policy::PermissionTier::ReadOnly,
            car_policy::agent_permissions::ApprovalMode::Deny,
        );
        let outcome = admit_with(&policy, Some("noisy".into()), false, "agent:noisy");
        assert!(
            matches!(outcome, DeliveryOutcome::Refused { .. }),
            "the shared admission denies it: {outcome:?}"
        );
    }

    /// The host keeps its pass-through, or every CarHost chat turn would be
    /// graded against an agent posture the operator's own client does not have.
    #[test]
    fn the_host_is_exempt_from_chat_admission() {
        let policy = car_policy::AgentPermissionPolicy::default();
        assert!(matches!(
            admit_with(&policy, None, true, "host"),
            DeliveryOutcome::Delivered
        ));
    }

    fn inbound(to: &str, body: &str, key: &str) -> car_a2a::InboundPeerMessage {
        car_a2a::InboundPeerMessage {
            peer_pubkey: key.to_string(),
            claimed: car_a2a::ClaimedByPeer {
                message_id: format!("m-{to}"),
                to: to.to_string(),
                body: body.to_string(),
                no_reply: false,
                trace: String::new(),
                via: Vec::new(),
            },
        }
    }

    /// The relay refusal. Resolving inbound through the full directory would
    /// let host A make this host forward to host C under THIS host's signature,
    /// reaching hosts that never trusted A. A peer that is not a locally
    /// attached agent is refused, and the refusal names the rule.
    #[tokio::test]
    async fn an_inbound_message_is_never_relayed_onward() {
        let (state, _t) = test_state().await;
        let broker = PeerInboundBroker {
            state: Arc::downgrade(&state),
        };
        // Nothing attached under this name, so it cannot resolve locally — and
        // must not be looked for anywhere else.
        let err = car_a2a::PeerInbox::deliver(&broker, inbound("far-host", "fwd", "KEY1"))
            .await
            .expect_err("must refuse");
        assert!(err.contains("never relayed"), "{err}");
    }

    /// An illegal recipient name is refused before anything is resolved or a
    /// guard entry is minted for it.
    #[tokio::test]
    async fn an_illegal_recipient_name_is_refused() {
        let (state, _t) = test_state().await;
        let broker = PeerInboundBroker {
            state: Arc::downgrade(&state),
        };
        let err = car_a2a::PeerInbox::deliver(&broker, inbound(".watcher", "hi", "KEY2"))
            .await
            .expect_err("must refuse");
        assert!(err.contains("not a legal peer name"), "{err}");
        assert!(
            state.peer_guards.lock().await.is_empty(),
            "a refused name must not mint a guard entry"
        );
    }

    /// The sender an agent is shown is derived from the verified key, never
    /// from anything the calling daemon claimed, and the message is marked
    /// unreplyable because that derived form is an attribution, not an address.
    #[tokio::test]
    async fn the_sender_is_the_verified_key_and_the_message_is_unreplyable() {
        let (state, _t) = test_state().await;
        attach(&state, "milo").await;
        let broker = PeerInboundBroker {
            state: Arc::downgrade(&state),
        };
        // No session is registered for the attached client id, so delivery
        // fails after admission — which is exactly the path that exercises the
        // derivation and the dedupe rollback.
        let _ = car_a2a::PeerInbox::deliver(&broker, inbound("milo", "first", "KEYABC")).await;

        // The dedupe record was given back, so an honest retry is not refused
        // as "already delivered".
        let retry = car_a2a::PeerInbox::deliver(&broker, inbound("milo", "first", "KEYABC")).await;
        let err = retry.expect_err("delivery still fails");
        assert!(
            !err.contains("already delivered"),
            "a retry after a failed delivery must not be called a duplicate: {err}"
        );
    }

    /// A daemon that has gone away refuses rather than panicking on the Weak.
    #[tokio::test]
    async fn a_stopped_daemon_refuses_cleanly() {
        let (state, _t) = test_state().await;
        let broker = PeerInboundBroker {
            state: Arc::downgrade(&state),
        };
        drop(state);
        let err = car_a2a::PeerInbox::deliver(&broker, inbound("milo", "hi", "KEY3"))
            .await
            .expect_err("must refuse");
        assert!(err.contains("shutting down"), "{err}");
    }

    fn inbound_with(to: &str, key: &str, via: Vec<String>) -> car_a2a::InboundPeerMessage {
        car_a2a::InboundPeerMessage {
            peer_pubkey: key.to_string(),
            claimed: car_a2a::ClaimedByPeer {
                message_id: format!("m-{to}"),
                to: to.to_string(),
                body: "body".into(),
                no_reply: false,
                trace: String::new(),
                via,
            },
        }
    }

    /// The boundary marker is APPENDED by the receiver, from the key it
    /// verified — never substituted for the sender's prefix. The configured
    /// audit row is the durable end-to-end observable.
    ///
    /// This test re-enters itself in a child process so `CAR_HOME` can point at
    /// a real canary directory without racing any other test in this binary.
    /// The old write-time resolver would append there; the fixed writer uses
    /// the path captured on `ServerState` and leaves the canary untouched.
    #[tokio::test]
    async fn the_receiver_appends_its_own_boundary_marker() {
        const CHILD_STATE_DIR: &str = "CAR_PEER_AUDIT_TEST_STATE_DIR";

        if let Some(state_dir) = std::env::var_os(CHILD_STATE_DIR) {
            let state_dir = std::path::PathBuf::from(state_dir);
            let state = Arc::new(ServerState::with_config(
                crate::session::ServerStateConfig::new(state_dir.clone()),
            ));
            assert_eq!(
                state.peer_audit_journal,
                state_dir.join("peer-messages.jsonl"),
                "a custom state directory owns its peer audit journal"
            );
            attach(&state, "milo").await;
            let broker = PeerInboundBroker {
                state: Arc::downgrade(&state),
            };
            let _ = car_a2a::PeerInbox::deliver(
                &broker,
                inbound_with("milo", "KEYZ", vec!["agent:remote".into()]),
            )
            .await;
            return;
        }

        let temp = tempfile::tempdir().unwrap();
        let state_dir = temp.path().join("configured-state");
        let global_dir = temp.path().join("global-car-home");
        std::fs::create_dir_all(&global_dir).unwrap();
        let global_canary = global_dir.join("peer-messages.jsonl");
        std::fs::write(&global_canary, "operator-row\n").unwrap();

        let status = std::process::Command::new(std::env::current_exe().unwrap())
            .arg("--exact")
            .arg("peers::tests::the_receiver_appends_its_own_boundary_marker")
            .arg("--test-threads=1")
            .env("CAR_HOME", &global_dir)
            .env(CHILD_STATE_DIR, &state_dir)
            .status()
            .expect("spawn isolated peer-audit test child");
        assert!(status.success(), "peer-audit test child failed: {status}");

        let journal = state_dir.join("peer-messages.jsonl");
        let body = std::fs::read_to_string(&journal)
            .expect("inbound audit row written inside the configured test state directory");
        let row: Value = serde_json::from_str(body.trim()).expect("one audit JSON row");
        assert_eq!(row["dir"], "in");
        assert_eq!(row["attested_by"], "KEYZ");
        assert_eq!(row["trace"], "m-milo");
        assert_eq!(
            row["via"],
            serde_json::json!(["agent:remote", "peer:KEYZ"]),
            "the receiver's verified-key boundary is appended to the attested prefix"
        );
        assert_eq!(
            std::fs::read_to_string(global_canary).unwrap(),
            "operator-row\n",
            "the process-global CAR_HOME peer journal must remain untouched"
        );
    }

    /// The chain this host builds APPENDS its marker to what the peer attested.
    ///
    /// Keep the pure shape assertion beside the end-to-end journal assertion so
    /// failures distinguish stamping logic from delivery/audit plumbing.
    #[test]
    fn the_boundary_marker_is_appended_not_substituted() {
        let attested = vec!["agent:alice".to_string(), "peer:KEYA".to_string()];
        let via = stamp_boundary(&attested, "peer:KEYB");

        assert_eq!(
            via,
            vec!["agent:alice", "peer:KEYA", "peer:KEYB"],
            "the marker goes on the END, and nothing the peer attested is dropped"
        );
        assert_eq!(
            via.last().map(String::as_str),
            Some("peer:KEYB"),
            "the receiver's own marker is last, so a reader can tell where this \
             host's observation begins"
        );
        assert_eq!(
            &via[..attested.len()],
            attested.as_slice(),
            "substituting the prefix would erase which segments the sending key \
             actually stood behind"
        );
    }

    /// A root message — no prior chain — still gets exactly one marker, so
    /// `hops()` can distinguish a first hop from a forged empty chain.
    #[test]
    fn a_root_message_gets_exactly_one_marker() {
        assert_eq!(stamp_boundary(&[], "peer:KEYA"), vec!["peer:KEYA"]);
    }

    /// A malformed lineage segment is refused before an agent is shown it.
    /// Nothing in CAR validated inbound lineage before this.
    #[tokio::test]
    async fn a_malformed_lineage_segment_is_refused() {
        let (state, _t) = test_state().await;
        attach(&state, "milo").await;
        let broker = PeerInboundBroker {
            state: Arc::downgrade(&state),
        };
        let err = car_a2a::PeerInbox::deliver(
            &broker,
            inbound_with("milo", "KEYY", vec!["not-a-segment".into()]),
        )
        .await
        .expect_err("refused");
        assert!(err.contains("well-formed lineage segment"), "{err}");
    }

    /// The hop cap fires on cross-host traffic — the case it exists for. The
    /// chain must be stamped BEFORE the guard runs, or `hops()` is 0 on every
    /// inbound message and this can never trigger.
    #[tokio::test]
    async fn the_hop_cap_fires_on_a_chain_that_arrived_deep() {
        let (state, _t) = test_state().await;
        attach(&state, "milo").await;
        let broker = PeerInboundBroker {
            state: Arc::downgrade(&state),
        };
        let deep: Vec<String> = (0..=car_peers::MAX_HOPS)
            .map(|i| format!("agent:a{i}"))
            .collect();
        let err = car_a2a::PeerInbox::deliver(&broker, inbound_with("milo", "KEYX", deep))
            .await
            .expect_err("refused");
        assert!(err.contains("hop cap"), "{err}");
    }

    /// The threshold is asymmetric on purpose: a record is not condemned on its
    /// first bad day, and it takes more than a bare majority of failures.
    #[tokio::test]
    async fn standing_degrades_only_once_failures_outrun_successes() {
        let (state, _t) = test_state().await;
        for _ in 0..3 {
            record_standing(&state, "peer:K", Some("bad chain"), None).await;
        }
        let now = car_peers::now_ms();
        assert!(
            state.peer_standing.lock().await["peer:K"].is_degraded(now),
            "3 failures, 0 successes is past the threshold"
        );

        let (state2, _t2) = test_state().await;
        record_standing(&state2, "peer:K", Some("bad chain"), None).await;
        record_standing(&state2, "peer:K", Some("bad chain"), None).await;
        assert!(
            !state2.peer_standing.lock().await["peer:K"].is_degraded(now),
            "2 failures is not yet degraded"
        );
    }

    /// A rate limit and an in-window duplicate must never touch standing. The
    /// channel guard exists because in a mutual loop NEITHER party is
    /// misbehaving; charging them would price correct behaviour as misconduct.
    #[tokio::test]
    async fn the_loop_guard_verdicts_are_not_misconduct() {
        for v in [
            car_peers::GuardVerdict::RateLimited {
                sender: "peer:K".into(),
                window_ms: 60_000,
            },
            car_peers::GuardVerdict::DuplicateWithinWindow,
        ] {
            let attributable = matches!(
                v,
                car_peers::GuardVerdict::HopLimit { .. }
                    | car_peers::GuardVerdict::TooLarge { .. }
                    | car_peers::GuardVerdict::InvalidName { .. }
            );
            assert!(!attributable, "{v:?} must not be charged to the sender");
        }
    }

    /// Successes saturate. Without a ceiling a peer with a long good history
    /// could spend it on an equally long run of failures before any throttle
    /// engaged — tolerable for an artifact, not for an actor that controls its
    /// own send rate.
    #[tokio::test]
    async fn success_is_capped_so_headroom_never_grows_without_bound() {
        let (state, _t) = test_state().await;
        for _ in 0..(STANDING_SUCCESS_CAP + 25) {
            record_standing(&state, "peer:K", None, None).await;
        }
        assert_eq!(
            state.peer_standing.lock().await["peer:K"].success_count,
            STANDING_SUCCESS_CAP
        );
    }

    /// A degraded record recovers on its own. The alternative is a ratchet that
    /// only tightens and needs a human to remember to forgive it.
    #[tokio::test]
    async fn standing_decays_so_a_degraded_peer_recovers() {
        let now = car_peers::now_ms();
        let rec = PeerStanding {
            success_count: 0,
            fail_count: 8,
            updated_ms: now - STANDING_HALFLIFE_MS * 3,
            ..Default::default()
        };
        assert!(
            !rec.is_degraded(now),
            "three half-lives takes 8 failures to 1, under the threshold"
        );
        assert!(
            rec.is_degraded(rec.updated_ms),
            "and it was degraded when the failures were fresh"
        );
    }

    /// A healthy sender meets no aggregate ceiling — standing adds nothing to
    /// normal traffic.
    #[tokio::test]
    async fn a_healthy_sender_is_never_throttled() {
        let (state, _t) = test_state().await;
        for _ in 0..50 {
            record_standing(&state, "peer:K", None, None).await;
            assert_eq!(
                standing_gate(&state, "peer:K").await,
                StandingVerdict::Proceed
            );
        }
    }

    /// A degraded sender is throttled rather than severed, and the reason names
    /// the record so an operator can see why.
    #[tokio::test]
    async fn a_degraded_sender_is_throttled_not_cut_off() {
        let (state, _t) = test_state().await;
        for _ in 0..5 {
            record_standing(&state, "peer:K", Some("bad chain"), None).await;
        }
        // The reduced budget still lets some through.
        for _ in 0..car_peers::DEGRADED_RATE_LIMIT {
            assert_eq!(
                standing_gate(&state, "peer:K").await,
                StandingVerdict::Proceed
            );
        }
        match standing_gate(&state, "peer:K").await {
            StandingVerdict::Throttled { reason } => {
                assert!(reason.contains("degraded"), "{reason}");
                assert!(reason.contains("recover"), "{reason}");
            }
            v => panic!("expected a throttle, got {v:?}"),
        }
    }

    /// Blame evidence is agent-granular even though the consequence lands on
    /// the key: enforce at the granularity you can verify, attribute at the
    /// granularity you can record.
    #[tokio::test]
    async fn a_failure_records_the_chain_that_caused_it() {
        let (state, _t) = test_state().await;
        let via = vec!["agent:scraper".to_string(), "peer:K".to_string()];
        record_standing(&state, "peer:K", Some("chain too deep"), Some(&via)).await;
        let map = state.peer_standing.lock().await;
        assert_eq!(map["peer:K"].last_fail_via.as_deref(), Some(via.as_slice()));
        assert_eq!(
            map["peer:K"].last_fail_reason.as_deref(),
            Some("chain too deep")
        );
    }
}