wire/

relay_server.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
//
// Copyright (C) 2026  wire contributors.
//
// This file (and only this file in the wire repository) is licensed under
// the GNU Affero General Public License v3.0 or later. The protocol crate
// (signing, agent_card, trust, canonical, cli, mcp) is Apache-2.0; the CLI
// binary entry point is MIT. See LICENSE.md for the trio explanation.
//
// AGPL on the relay specifically discourages forks that operate `wire-relay`
// as a closed-source SaaS offering — those forks must publish their changes
// under AGPL too. Self-hosting your own relay for your own org or running
// the public-good relay we operate is fully permitted.
//
//! HTTP mailbox relay — minimal, persistent, bearer-authenticated.
//!
//! Design (v0.1):
//!   - One process serves N slots; each slot is a per-peer FIFO of signed events.
//!   - Slot allocation returns `(slot_id, slot_token)`. Holder of the token
//!     can post + read that slot. Tokens never expire in v0.1 (rotate by
//!     allocating a new slot).
//!   - Events are stored verbatim — the relay does NOT verify Ed25519 signatures
//!     itself. Verification happens client-side (`wire tail` + `verify_message_v31`).
//!     The relay is dumb on purpose: it is a content-addressed mailbox, not a
//!     trust authority.
//!   - 256 KiB max body per event.
//!   - Persistence: each slot's events are appended to
//!     `<state_dir>/slots/<slot_id>.jsonl` on every `POST /events`. Tokens
//!     are persisted to `<state_dir>/tokens.json` on allocation.
//!   - On startup, slots + tokens are reloaded from disk.

use anyhow::{Context, Result};
use axum::{
    Json, Router,
    extract::{Path, Query, State},
    http::{HeaderMap, StatusCode, header::AUTHORIZATION},
    response::IntoResponse,
    routing::{delete, get, post},
};
use rand::RngCore;
use serde::Deserialize;
use serde_json::{Value, json};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use tokio::sync::Mutex;
use tower_governor::{
    GovernorLayer, governor::GovernorConfigBuilder, key_extractor::GlobalKeyExtractor,
};

const MAX_EVENT_BYTES: usize = 256 * 1024;
/// Total bytes a single slot can hold before further POSTs are rejected (413).
/// Defends against an abusive bearer-holder filling relay disk (T11). At 64 MB
/// per slot, an attacker pushing the rate-limit ceiling fills their own slot
/// in ~25 seconds, then gets 413 forever — disk impact bounded.
const MAX_SLOT_BYTES: usize = 64 * 1024 * 1024;

/// Per-nick `/v1/handle/intro` rate limit (#247.3): at most this many intros to
/// a given nick within `INTRO_WINDOW_SECS`, else 429. Pair-intro is a rare,
/// human-paced action (you dial a peer once), so a tight cap is safe and stops
/// an unauthenticated flood from filling the victim's slot to MAX_SLOT_BYTES.
const INTRO_MAX_PER_WINDOW: usize = 5;
const INTRO_WINDOW_SECS: u64 = 300;

/// Backstop ceilings on the COUNT of in-memory objects (audit H1 / #291).
/// Generous — they don't bind a healthy relay — but stop an attacker from
/// allocating millions of slots/handles/pairs/invites to exhaust RAM + disk
/// (every allocation also rewrites a persistence file). `503` when reached.
const MAX_SLOTS: usize = 200_000;
const MAX_HANDLES: usize = 100_000;
const MAX_PAIR_SLOTS: usize = 50_000;
const MAX_INVITES: usize = 50_000;
/// Cap concurrent SSE subscribers per slot (audit). An authenticated slot-token
/// holder can otherwise open unbounded `GET /v1/events/:slot_id/stream`
/// connections, each a channel that every `post_event` fans out to (O(n)
/// broadcast) — memory + latency DoS. Over the cap → 503.
const MAX_STREAMS_PER_SLOT: usize = 256;
/// Soft cap on nicks tracked in `intro_times`; once exceeded, fully-aged entries
/// are swept so the map can't accumulate one stale entry per ever-touched nick
/// (#247.3 hygiene; growth is already bounded by `MAX_HANDLES`).
const MAX_INTRO_TRACKING_NICKS: usize = 10_000;

/// True iff a map at `len` entries is at/over the `max` ceiling — i.e. a new
/// allocation must be refused. Pure → unit-tested (#291 H1).
fn at_capacity(len: usize, max: usize) -> bool {
    len >= max
}

/// Drop subscriber senders whose receiver (the SSE client) has disconnected.
/// `is_closed()` becomes true once the `UnboundedReceiverStream` is dropped on
/// client disconnect. `post_event` only prunes lazily on its next broadcast, so
/// a slot that goes silent would otherwise leak dead senders forever; pruning at
/// stream admission bounds it. Pure → unit-tested.
fn retain_live_subscribers(subs: &mut Vec<tokio::sync::mpsc::UnboundedSender<Value>>) {
    subs.retain(|tx| !tx.is_closed());
}

/// Wall-clock unix seconds (best-effort; 0 on a pre-epoch clock).
fn unix_now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

/// Per-nick intro rate gate (#247.3). Prunes `times` to entries within `window`
/// of `now`, then: if `>= max` remain it's over the limit → return `false`
/// (reject); otherwise record `now` and return `true` (allow). Pure → unit-tested.
fn record_intro_within_rate(times: &mut Vec<u64>, now: u64, window: u64, max: usize) -> bool {
    times.retain(|t| now.saturating_sub(*t) < window);
    if times.len() >= max {
        return false;
    }
    times.push(now);
    true
}

/// Drop nicks from `times_by_nick` whose intro timestamps have all aged past
/// `window` relative to `now`. Bounds `intro_times` growth (#247.3 hygiene).
/// Pure → unit-tested.
fn evict_stale_intro_nicks(times_by_nick: &mut HashMap<String, Vec<u64>>, now: u64, window: u64) {
    times_by_nick.retain(|_, ts| ts.iter().any(|t| now.saturating_sub(*t) < window));
}

#[derive(Clone)]
pub struct Relay {
    inner: Arc<Mutex<Inner>>,
    state_dir: PathBuf,
    counters: Arc<RelayCounters>,
}

/// Lock-free usage counters served by GET /stats. Counter values are
/// loaded from `<state_dir>/counters.json` on startup and snapshotted back
/// to disk every 30s by `spawn_counter_persister`, so deploys + restarts
/// don't reset them. `boot_unix` is per-process — uptime is process-local.
struct RelayCounters {
    boot_unix: u64,
    handle_claims_total: AtomicU64,
    handle_first_claims_total: AtomicU64,
    slot_allocations_total: AtomicU64,
    pair_opens_total: AtomicU64,
    events_posted_total: AtomicU64,
}

#[derive(serde::Serialize, serde::Deserialize, Default)]
struct CountersSnapshot {
    handle_claims_total: u64,
    handle_first_claims_total: u64,
    slot_allocations_total: u64,
    pair_opens_total: u64,
    events_posted_total: u64,
}

/// One row in `<state_dir>/stats-history.jsonl` — written every 30s by
/// `spawn_counter_persister` so /stats.html can draw sparklines. Live-state
/// fields (`*_active`) are point-in-time; *_total fields are the cumulative
/// counters at that timestamp. Field names mirror the /stats endpoint.
#[derive(serde::Serialize, serde::Deserialize)]
struct HistoryEntry {
    ts: u64,
    handles_active: usize,
    slots_active: usize,
    pair_slots_open: usize,
    streams_active: usize,
    handle_claims_total: u64,
    handle_first_claims_total: u64,
    slot_allocations_total: u64,
    pair_opens_total: u64,
    events_posted_total: u64,
}

#[derive(Deserialize)]
pub struct StatsHistoryQuery {
    /// How many hours of history to return, default 24, max 168 (7 days).
    pub hours: Option<u64>,
}

struct Inner {
    /// slot_id -> ordered list of stored events (parsed JSON Values).
    slots: HashMap<String, Vec<Value>>,
    /// slot_id -> bearer token. Token holder may read + write that slot.
    tokens: HashMap<String, String>,
    /// slot_id -> total bytes stored. Enforced against MAX_SLOT_BYTES.
    slot_bytes: HashMap<String, usize>,
    /// slot_id -> wall-clock unix seconds of the slot owner's last `list_events`
    /// call. Used by `GET /v1/slot/:slot_id/state` so a remote sender can
    /// gauge whether the slot's owner is still polling (i.e., still attentive).
    /// `None` means the slot has never been pulled since the relay restarted.
    last_pull_at_unix: HashMap<String, u64>,
    /// slot_id -> active SSE subscribers (R1 push). Each `UnboundedSender`
    /// belongs to one open `GET /v1/events/:slot_id/stream` connection.
    /// On every successful `post_event` to a slot we walk the slot's list
    /// and broadcast the event; closed channels are pruned lazily on send-
    /// error. Auth: subscribers presented a valid slot_token at stream open.
    streams: HashMap<String, Vec<tokio::sync::mpsc::UnboundedSender<Value>>>,
    /// code_hash -> pair_id (lookup so guests find the host).
    pair_lookup: HashMap<String, String>,
    /// pair_id -> ephemeral pairing state.
    pair_slots: HashMap<String, PairSlot>,
    /// nick -> registered handle directory entry (v0.5).
    handles: HashMap<String, HandleRecord>,
    /// slot_id -> latest operator-published auto-responder health record (R3).
    responder_health: HashMap<String, ResponderHealthRecord>,
    /// token -> short-URL invite record (v0.5.10 — one-curl onboarding).
    /// Token is the path segment in `GET /i/{token}`. Record holds the
    /// underlying `wire://pair?...` URL plus TTL/uses bookkeeping.
    invites: HashMap<String, InviteRecord>,
    /// nick -> unix-second timestamps of recent `/v1/handle/intro` deliveries
    /// (#247.3). Pruned to the rate-limit window on each access; bounds the
    /// unauthenticated pair-intro flood that could otherwise fill a victim's
    /// slot to `MAX_SLOT_BYTES`.
    intro_times: HashMap<String, Vec<u64>>,
}

/// One entry in the short-URL invite map. Persisted to
/// `<state_dir>/invites.jsonl` so deploys don't drop active invites.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
struct InviteRecord {
    token: String,
    invite_url: String,
    expires_unix: u64,
    /// `None` = unlimited until TTL hits. `Some(n)` = decrement each fetch.
    uses_remaining: Option<u32>,
    created_unix: u64,
}

/// One entry in the relay's handle directory (v0.5 — agentic hotline).
/// FCFS on nick: first claimant binds the nick to their DID. Same-DID re-claims
/// are allowed (used for profile updates + slot rotation).
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
struct HandleRecord {
    pub nick: String,
    pub did: String,
    pub card: Value,
    pub slot_id: String,
    pub relay_url: Option<String>,
    pub claimed_at: String,
    /// v0.5.19 (#9.1): if false, this handle is omitted from the
    /// `/v1/handles` directory listing — operator opted out of bulk
    /// discovery. The `.well-known/wire/agent?handle=X` direct lookup
    /// still resolves so existing peers + out-of-band sharing continue
    /// to work.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub discoverable: Option<bool>,
}

impl HandleRecord {
    fn is_discoverable(&self) -> bool {
        self.discoverable.unwrap_or(true)
    }
}

#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub struct ResponderHealthRecord {
    pub status: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub last_success_at: Option<String>,
    pub set_at: String,
}

#[derive(Clone, Debug)]
struct PairSlot {
    /// SPAKE2 message from the host side.
    host_msg: Option<String>,
    /// SPAKE2 message from the guest side.
    guest_msg: Option<String>,
    /// Sealed bootstrap payload from host (after SAS confirm).
    host_bootstrap: Option<String>,
    /// Sealed bootstrap payload from guest.
    guest_bootstrap: Option<String>,
    /// Last activity time (monotonic) — used for TTL eviction.
    last_touched: std::time::Instant,
}

impl Default for PairSlot {
    fn default() -> Self {
        Self {
            host_msg: None,
            guest_msg: None,
            host_bootstrap: None,
            guest_bootstrap: None,
            last_touched: std::time::Instant::now(),
        }
    }
}

/// Pair-slot idle TTL. After this many seconds without activity, the slot
/// is evicted to free memory + bound brute-force surface (PENTEST.md §code-review #3).
const PAIR_SLOT_TTL_SECS: u64 = 300;

#[derive(Deserialize)]
pub struct AllocateRequest {
    /// Optional handle hint — purely informational, server doesn't enforce.
    #[serde(default)]
    pub handle: Option<String>,
}

#[derive(Deserialize)]
pub struct PostEventRequest {
    pub event: Value,
}

#[derive(Deserialize)]
pub struct ListEventsQuery {
    /// Resume from after this event_id (exclusive). Omit for full slot read.
    pub since: Option<String>,
    /// Max events to return. Default 100, max 1000.
    pub limit: Option<usize>,
}

impl Relay {
    pub async fn new(state_dir: PathBuf) -> Result<Self> {
        tokio::fs::create_dir_all(state_dir.join("slots")).await?;
        tokio::fs::create_dir_all(state_dir.join("handles")).await?;
        tokio::fs::create_dir_all(state_dir.join("responder-health")).await?;
        let mut inner = Inner {
            slots: HashMap::new(),
            tokens: HashMap::new(),
            slot_bytes: HashMap::new(),
            last_pull_at_unix: HashMap::new(),
            streams: HashMap::new(),
            pair_lookup: HashMap::new(),
            pair_slots: HashMap::new(),
            handles: HashMap::new(),
            responder_health: HashMap::new(),
            invites: HashMap::new(),
            intro_times: HashMap::new(),
        };
        // Reload tokens
        let token_path = state_dir.join("tokens.json");
        if token_path.exists() {
            let body = tokio::fs::read_to_string(&token_path).await?;
            inner.tokens = serde_json::from_str(&body).unwrap_or_default();
        }
        // Reload slots from JSONL
        let mut slots_dir = tokio::fs::read_dir(state_dir.join("slots")).await?;
        while let Some(entry) = slots_dir.next_entry().await? {
            let path = entry.path();
            if path.extension().map(|x| x != "jsonl").unwrap_or(true) {
                continue;
            }
            let stem = match path.file_stem().and_then(|s| s.to_str()) {
                Some(s) => s.to_string(),
                None => continue,
            };
            let body = tokio::fs::read_to_string(&path).await.unwrap_or_default();
            let mut events = Vec::new();
            for line in body.lines() {
                if let Ok(v) = serde_json::from_str::<Value>(line) {
                    events.push(v);
                }
            }
            // Recompute byte usage for the slot from its persisted events.
            let bytes: usize = events
                .iter()
                .map(|e| serde_json::to_vec(e).map(|v| v.len()).unwrap_or(0))
                .sum();
            inner.slot_bytes.insert(stem.clone(), bytes);
            inner.slots.insert(stem, events);
        }
        // Reload handle directory (v0.5).
        let handles_dir = state_dir.join("handles");
        if handles_dir.exists() {
            let mut rd = tokio::fs::read_dir(&handles_dir).await?;
            while let Some(entry) = rd.next_entry().await? {
                let path = entry.path();
                if path.extension().and_then(|x| x.to_str()) != Some("json") {
                    continue;
                }
                let body = tokio::fs::read_to_string(&path).await.unwrap_or_default();
                if let Ok(rec) = serde_json::from_str::<HandleRecord>(&body) {
                    inner.handles.insert(rec.nick.clone(), rec);
                }
            }
        }
        // Reload responder health records (R3).
        let responder_health_dir = state_dir.join("responder-health");
        if responder_health_dir.exists() {
            let mut rd = tokio::fs::read_dir(&responder_health_dir).await?;
            while let Some(entry) = rd.next_entry().await? {
                let path = entry.path();
                if path.extension().and_then(|x| x.to_str()) != Some("json") {
                    continue;
                }
                let Some(slot_id) = path.file_stem().and_then(|s| s.to_str()) else {
                    continue;
                };
                let body = tokio::fs::read_to_string(&path).await.unwrap_or_default();
                if let Ok(rec) = serde_json::from_str::<ResponderHealthRecord>(&body) {
                    inner.responder_health.insert(slot_id.to_string(), rec);
                }
            }
        }
        // Reload short-URL invites. JSONL append-only; later entries with
        // the same token overwrite earlier (won't happen — tokens are
        // unique by construction — but coded defensively).
        let invites_path = state_dir.join("invites.jsonl");
        if invites_path.exists() {
            let now_unix = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0);
            let body = tokio::fs::read_to_string(&invites_path)
                .await
                .unwrap_or_default();
            for line in body.lines() {
                if let Ok(rec) = serde_json::from_str::<InviteRecord>(line)
                    && rec.expires_unix > now_unix
                {
                    inner.invites.insert(rec.token.clone(), rec);
                }
            }
        }
        let boot_unix = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        // Reload counter snapshot. Missing/corrupt file → start at zero.
        let snap: CountersSnapshot =
            match tokio::fs::read_to_string(state_dir.join("counters.json")).await {
                Ok(body) => serde_json::from_str(&body).unwrap_or_default(),
                Err(_) => CountersSnapshot::default(),
            };
        Ok(Self {
            inner: Arc::new(Mutex::new(inner)),
            state_dir,
            counters: Arc::new(RelayCounters {
                boot_unix,
                handle_claims_total: AtomicU64::new(snap.handle_claims_total),
                handle_first_claims_total: AtomicU64::new(snap.handle_first_claims_total),
                slot_allocations_total: AtomicU64::new(snap.slot_allocations_total),
                pair_opens_total: AtomicU64::new(snap.pair_opens_total),
                events_posted_total: AtomicU64::new(snap.events_posted_total),
            }),
        })
    }

    pub fn router(self) -> Router {
        self.router_with_mode(ServerMode::default())
    }

    pub fn router_with_mode(self, mode: ServerMode) -> Router {
        // Rate limit applied to write endpoints that create new state (slots,
        // pair-sessions, bootstraps). 10 req/sec sustained, 50 req burst.
        // v0.1 uses the GLOBAL key extractor (single bucket for all callers) —
        // per-IP needs ConnectInfo middleware which axum 0.7 wires differently.
        // Per-IP keying is a v0.2 hardening; for now Cloudflare WAF + this
        // global cap shoulder DDoS protection in series.
        let governor_conf = std::sync::Arc::new(
            GovernorConfigBuilder::default()
                .per_second(10)
                .burst_size(50)
                .key_extractor(GlobalKeyExtractor)
                .finish()
                .expect("valid governor config"),
        );
        let governor_layer = GovernorLayer {
            config: governor_conf,
        };

        // Hot writes group — rate limited.
        let hot_writes = Router::new()
            .route("/v1/slot/allocate", post(allocate_slot))
            .route("/v1/pair", post(pair_open))
            .route("/v1/pair/:pair_id/bootstrap", post(pair_bootstrap))
            .route("/v1/pair/abandon", post(pair_abandon))
            .layer(governor_layer);

        // Core data-plane routes: events, pair, slots, healthz. Always
        // present, in both federation and local-only modes.
        let mut router = Router::new()
            .route("/healthz", get(healthz))
            .route("/v1/events/:slot_id", post(post_event).get(list_events))
            .route("/v1/slot/:slot_id/state", get(slot_state))
            .route(
                "/v1/slot/:slot_id/responder-health",
                post(responder_health_set),
            )
            .route("/v1/events/:slot_id/stream", get(stream_events))
            .route("/v1/pair/:pair_id", get(pair_get))
            .route("/v1/handle/intro/:nick", post(handle_intro));

        // Discovery + landing surfaces: phonebook, well-known agent cards,
        // landing page, stats, invite landing. In `--local-only` mode we
        // skip all of these — the relay becomes invisible from the outside
        // (and from other agents on the same box that might enumerate it).
        // This is the v0.5.17 within-machine privacy guarantee.
        if !mode.local_only {
            router = router
                .route("/", get(landing_index))
                .route("/favicon.svg", get(landing_favicon))
                .route("/og.png", get(landing_og))
                .route("/install", get(landing_install_sh))
                .route("/install.sh", get(landing_install_sh))
                .route("/openshell-policy.sh", get(landing_openshell_policy_sh))
                .route("/stats", get(stats_root))
                .route("/stats.json", get(stats_json))
                .route("/stats.html", get(landing_stats_html))
                .route("/stats.history", get(stats_history))
                .route("/phonebook", get(landing_phonebook_html))
                .route("/phonebook.html", get(landing_phonebook_html))
                .route("/v1/handle/claim", post(handle_claim))
                .route("/v1/handle/claim/:nick", delete(handle_unclaim))
                .route("/v1/handles", get(handles_directory))
                .route("/v1/invite/register", post(invite_register))
                .route("/i/:token", get(invite_script))
                .route("/.well-known/wire/agent", get(well_known_agent))
                .route(
                    "/.well-known/agent-card.json",
                    get(well_known_agent_card_a2a),
                );
        } else {
            // Local-only mode still needs handle_claim for in-process
            // session bootstrap (`wire session new` allocates a local
            // slot AND claims a handle on the local relay so peers can
            // resolve it). The claim is gated to loopback-only callers
            // by the bind, not by the route.
            router = router
                .route("/v1/handle/claim", post(handle_claim))
                .route("/v1/handle/claim/:nick", delete(handle_unclaim));
        }

        router.merge(hot_writes).with_state(self)
    }

    /// Evict pair-slots that have been idle past `PAIR_SLOT_TTL_SECS`.
    /// Called inline on every pair-slot mutation; a background sweeper task
    /// (see `Self::start_sweeper`) covers the long-idle case.
    async fn evict_expired_pair_slots(&self) {
        let now = std::time::Instant::now();
        let ttl = std::time::Duration::from_secs(PAIR_SLOT_TTL_SECS);
        let mut inner = self.inner.lock().await;
        let mut to_remove = Vec::new();
        for (id, slot) in inner.pair_slots.iter() {
            if now.duration_since(slot.last_touched) > ttl {
                to_remove.push(id.clone());
            }
        }
        for id in to_remove {
            inner.pair_slots.remove(&id);
            inner.pair_lookup.retain(|_, v| v != &id);
        }
    }

    /// Spawn a background tokio task that runs `evict_expired_pair_slots` every
    /// 60 seconds. Call once after `Relay::new`; the handle is leaked deliberately
    /// — process exit reaps it. Safe to skip in tests where you'd rather test
    /// eviction inline.
    /// Proactively drop fully-aged nicks from `intro_times` on the background
    /// tick, so the map rarely reaches the on-demand sweep threshold in
    /// `handle_intro` — that keeps the 10k-entry `retain` scan off the hot,
    /// globally-locked intro path (where it would block every other handler).
    /// The on-demand sweep stays as a backstop, so the bound is unchanged.
    async fn sweep_intro_times(&self) {
        let now = unix_now();
        let mut inner = self.inner.lock().await;
        evict_stale_intro_nicks(&mut inner.intro_times, now, INTRO_WINDOW_SECS);
    }

    pub fn spawn_pair_sweeper(&self) {
        let me = self.clone();
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(std::time::Duration::from_secs(60));
            loop {
                tick.tick().await;
                me.evict_expired_pair_slots().await;
                me.sweep_intro_times().await;
            }
        });
    }

    /// Snapshot the in-process counters to `<state_dir>/counters.json`. Called
    /// every 30s by `spawn_counter_persister` and once during graceful
    /// shutdown so a deploy doesn't reset the running totals.
    pub async fn persist_counters(&self) -> Result<()> {
        let snap = CountersSnapshot {
            handle_claims_total: self.counters.handle_claims_total.load(Ordering::Relaxed),
            handle_first_claims_total: self
                .counters
                .handle_first_claims_total
                .load(Ordering::Relaxed),
            slot_allocations_total: self.counters.slot_allocations_total.load(Ordering::Relaxed),
            pair_opens_total: self.counters.pair_opens_total.load(Ordering::Relaxed),
            events_posted_total: self.counters.events_posted_total.load(Ordering::Relaxed),
        };
        let body = serde_json::to_vec_pretty(&snap)?;
        let path = self.state_dir.join("counters.json");
        tokio::fs::write(path, body).await?;
        Ok(())
    }

    /// Append one row to `<state_dir>/stats-history.jsonl` mirroring the
    /// /stats endpoint at this instant. Used by /stats.html for sparklines.
    /// File grows ~250 B per call → ~720 KB/day. A future prune wave can
    /// roll old entries off once the history exceeds 90 days.
    pub async fn append_history(&self) -> Result<()> {
        use tokio::io::AsyncWriteExt;
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);
        let (handles_active, slots_active, pair_slots_open, streams_active) = {
            let inner = self.inner.lock().await;
            (
                inner.handles.len(),
                inner.slots.len(),
                inner.pair_slots.len(),
                inner.streams.values().map(Vec::len).sum::<usize>(),
            )
        };
        let entry = HistoryEntry {
            ts: now,
            handles_active,
            slots_active,
            pair_slots_open,
            streams_active,
            handle_claims_total: self.counters.handle_claims_total.load(Ordering::Relaxed),
            handle_first_claims_total: self
                .counters
                .handle_first_claims_total
                .load(Ordering::Relaxed),
            slot_allocations_total: self.counters.slot_allocations_total.load(Ordering::Relaxed),
            pair_opens_total: self.counters.pair_opens_total.load(Ordering::Relaxed),
            events_posted_total: self.counters.events_posted_total.load(Ordering::Relaxed),
        };
        let line = serde_json::to_vec(&entry)?;
        let path = self.state_dir.join("stats-history.jsonl");
        let mut f = tokio::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .await?;
        f.write_all(&line).await?;
        f.write_all(b"\n").await?;
        f.flush().await?;
        Ok(())
    }

    /// Spawn a background tokio task that calls `persist_counters` every 30s
    /// + appends a history row on the same tick. Loss bound: counters can
    ///   drift back up to 30s on crash, history can drop one row.
    pub fn spawn_counter_persister(&self) {
        let me = self.clone();
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(std::time::Duration::from_secs(30));
            // First tick fires immediately; skip it so we don't write the
            // freshly-loaded snapshot back unchanged.
            tick.tick().await;
            loop {
                tick.tick().await;
                if let Err(e) = me.persist_counters().await {
                    eprintln!("counter persist failed: {e}");
                }
                if let Err(e) = me.append_history().await {
                    eprintln!("history append failed: {e}");
                }
            }
        });
    }

    async fn persist_tokens(&self) -> Result<()> {
        let body = {
            let inner = self.inner.lock().await;
            serde_json::to_string_pretty(&inner.tokens)?
        };
        let path = self.state_dir.join("tokens.json");
        tokio::fs::write(path, body).await?;
        Ok(())
    }

    async fn append_event_to_disk(&self, slot_id: &str, event: &Value) -> Result<()> {
        // Defense in depth: only allow lowercase hex slot_ids of the exact length
        // we ever produce ourselves (16 random bytes -> 32 hex chars). Blocks
        // any future code path that might let attacker-controlled slot_ids reach
        // disk operations. allocate_slot() always meets this; this assert is
        // belt-and-suspenders against future regressions.
        if !is_valid_slot_id(slot_id) {
            return Err(anyhow::anyhow!("invalid slot_id format: {slot_id:?}"));
        }
        let path = self
            .state_dir
            .join("slots")
            .join(format!("{slot_id}.jsonl"));
        let mut line = serde_json::to_vec(event)?;
        line.push(b'\n');
        use tokio::io::AsyncWriteExt;
        let mut f = tokio::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .await
            .with_context(|| format!("opening {path:?}"))?;
        f.write_all(&line).await?;
        f.flush().await?;
        Ok(())
    }
}

async fn healthz() -> impl IntoResponse {
    (StatusCode::OK, "ok\n")
}

// Public aggregate-usage snapshot. Counter fields (`*_total`) reset on
// process restart; state fields (`handles_active`, `slots_active`) survive
// on the persistent volume. No DIDs / handles / IPs leaked — counts only.
async fn stats_history(
    State(relay): State<Relay>,
    Query(q): Query<StatsHistoryQuery>,
) -> impl IntoResponse {
    let hours = q.hours.unwrap_or(24).min(168);
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let cutoff = now.saturating_sub(hours * 3600);
    let path = relay.state_dir.join("stats-history.jsonl");
    let body = tokio::fs::read_to_string(&path).await.unwrap_or_default();
    let entries: Vec<Value> = body
        .lines()
        .filter_map(|l| serde_json::from_str::<Value>(l).ok())
        .filter(|v| {
            v.get("ts")
                .and_then(Value::as_u64)
                .map(|t| t >= cutoff)
                .unwrap_or(false)
        })
        .collect();
    (
        StatusCode::OK,
        Json(json!({
            "hours": hours,
            "now_unix": now,
            "count": entries.len(),
            "entries": entries,
        })),
    )
}

async fn landing_stats_html() -> impl IntoResponse {
    static STATS_HTML: &[u8] = include_bytes!("../landing/stats.html");
    (
        StatusCode::OK,
        [
            (axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8"),
            (axum::http::header::CACHE_CONTROL, "public, max-age=60"),
        ],
        STATS_HTML,
    )
}

async fn landing_phonebook_html() -> impl IntoResponse {
    static PHONEBOOK_HTML: &[u8] = include_bytes!("../landing/phonebook.html");
    (
        StatusCode::OK,
        [
            (axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8"),
            (axum::http::header::CACHE_CONTROL, "public, max-age=60"),
        ],
        PHONEBOOK_HTML,
    )
}

/// `/stats` dispatch: serve the pretty HTML dashboard to browsers (Accept
/// includes text/html) and JSON to everything else (curl, scripts, scrapers).
/// Keeps the JSON contract intact while letting humans land on the page at
/// the short URL.
async fn stats_root(State(relay): State<Relay>, headers: HeaderMap) -> axum::response::Response {
    let wants_html = headers
        .get(axum::http::header::ACCEPT)
        .and_then(|v| v.to_str().ok())
        .unwrap_or("")
        .contains("text/html");
    if wants_html {
        landing_stats_html().await.into_response()
    } else {
        stats_json(State(relay)).await.into_response()
    }
}

async fn stats_json(State(relay): State<Relay>) -> impl IntoResponse {
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let inner = relay.inner.lock().await;
    let streams_active: usize = inner.streams.values().map(Vec::len).sum();
    let body = json!({
        "version": env!("CARGO_PKG_VERSION"),
        "uptime_seconds": now.saturating_sub(relay.counters.boot_unix),
        "handles_active": inner.handles.len(),
        "slots_active": inner.slots.len(),
        "pair_slots_open": inner.pair_slots.len(),
        "streams_active": streams_active,
        "handle_claims_total": relay.counters.handle_claims_total.load(Ordering::Relaxed),
        "handle_first_claims_total": relay.counters.handle_first_claims_total.load(Ordering::Relaxed),
        "slot_allocations_total": relay.counters.slot_allocations_total.load(Ordering::Relaxed),
        "pair_opens_total": relay.counters.pair_opens_total.load(Ordering::Relaxed),
        "events_posted_total": relay.counters.events_posted_total.load(Ordering::Relaxed),
    });
    (StatusCode::OK, Json(body))
}

// Static landing site baked into the binary so apex (wireup.net) can flip
// straight to Fly without a separate static-host. ~37 KB total — negligible
// against the release binary size, and keeps the relay self-contained.
async fn landing_index() -> impl IntoResponse {
    static INDEX_HTML: &[u8] = include_bytes!("../landing/index.html");
    (
        StatusCode::OK,
        [(axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8")],
        INDEX_HTML,
    )
}

async fn landing_favicon() -> impl IntoResponse {
    static FAVICON_SVG: &[u8] = include_bytes!("../landing/favicon.svg");
    (
        StatusCode::OK,
        [(axum::http::header::CONTENT_TYPE, "image/svg+xml")],
        FAVICON_SVG,
    )
}

async fn landing_og() -> impl IntoResponse {
    static OG_PNG: &[u8] = include_bytes!("../landing/og.png");
    (
        StatusCode::OK,
        [
            (axum::http::header::CONTENT_TYPE, "image/png"),
            (axum::http::header::CACHE_CONTROL, "public, max-age=86400"),
        ],
        OG_PNG,
    )
}

async fn landing_install_sh() -> impl IntoResponse {
    static INSTALL_SH: &[u8] = include_bytes!("../landing/install.sh");
    (
        StatusCode::OK,
        [
            (
                axum::http::header::CONTENT_TYPE,
                "text/x-shellscript; charset=utf-8",
            ),
            (axum::http::header::CACHE_CONTROL, "public, max-age=300"),
        ],
        INSTALL_SH,
    )
}

async fn landing_openshell_policy_sh() -> impl IntoResponse {
    static POLICY_SH: &[u8] = include_bytes!("../landing/openshell-policy.sh");
    (
        StatusCode::OK,
        [
            (
                axum::http::header::CONTENT_TYPE,
                "text/x-shellscript; charset=utf-8",
            ),
            (axum::http::header::CACHE_CONTROL, "public, max-age=300"),
        ],
        POLICY_SH,
    )
}

async fn allocate_slot(
    State(relay): State<Relay>,
    Json(_req): Json<AllocateRequest>,
) -> impl IntoResponse {
    let slot_id = random_hex(16);
    let slot_token = random_hex(32);
    {
        let mut inner = relay.inner.lock().await;
        if at_capacity(inner.slots.len(), MAX_SLOTS) {
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                Json(json!({"error": "relay slot capacity reached"})),
            )
                .into_response();
        }
        inner.slots.insert(slot_id.clone(), Vec::new());
        inner.tokens.insert(slot_id.clone(), slot_token.clone());
    }
    if let Err(e) = relay.persist_tokens().await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": format!("persist failed: {e}")})),
        )
            .into_response();
    }
    relay
        .counters
        .slot_allocations_total
        .fetch_add(1, Ordering::Relaxed);
    (
        StatusCode::CREATED,
        Json(json!({"slot_id": slot_id, "slot_token": slot_token})),
    )
        .into_response()
}

async fn post_event(
    State(relay): State<Relay>,
    Path(slot_id): Path<String>,
    headers: HeaderMap,
    Json(req): Json<PostEventRequest>,
) -> impl IntoResponse {
    if let Err(resp) = check_token(&relay, &headers, &slot_id).await {
        return resp;
    }
    // Body size cap (rough; serialize and check).
    let body_bytes = match serde_json::to_vec(&req.event) {
        Ok(b) => b,
        Err(e) => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": format!("event not serializable: {e}")})),
            )
                .into_response();
        }
    };
    if body_bytes.len() > MAX_EVENT_BYTES {
        return (
            StatusCode::PAYLOAD_TOO_LARGE,
            Json(json!({"error": "event exceeds 256 KiB", "max_bytes": MAX_EVENT_BYTES})),
        )
            .into_response();
    }
    // Per-slot quota: cap accumulated bytes per slot at MAX_SLOT_BYTES.
    {
        let inner = relay.inner.lock().await;
        let used = inner.slot_bytes.get(&slot_id).copied().unwrap_or(0);
        if used + body_bytes.len() > MAX_SLOT_BYTES {
            return (
                StatusCode::PAYLOAD_TOO_LARGE,
                Json(json!({
                    "error": "slot quota exceeded",
                    "slot_bytes_used": used,
                    "slot_bytes_max": MAX_SLOT_BYTES,
                    "remediation": "operator should `wire rotate-slot` to drain old slot",
                })),
            )
                .into_response();
        }
    }
    let event_id = req
        .event
        .get("event_id")
        .and_then(Value::as_str)
        .map(str::to_string);

    // Dedupe by event_id if present.
    let dup = {
        let inner = relay.inner.lock().await;
        let slot = inner.slots.get(&slot_id);
        if let (Some(eid), Some(slot)) = (&event_id, slot) {
            slot.iter()
                .any(|e| e.get("event_id").and_then(Value::as_str) == Some(eid))
        } else {
            false
        }
    };
    if dup {
        return (
            StatusCode::OK,
            Json(json!({"event_id": event_id, "status": "duplicate"})),
        )
            .into_response();
    }

    {
        let mut inner = relay.inner.lock().await;
        let event_size = body_bytes.len();
        let slot = inner.slots.entry(slot_id.clone()).or_default();
        slot.push(req.event.clone());
        *inner.slot_bytes.entry(slot_id.clone()).or_insert(0) += event_size;
    }
    if let Err(e) = relay.append_event_to_disk(&slot_id, &req.event).await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": format!("persist failed: {e}")})),
        )
            .into_response();
    }
    relay
        .counters
        .events_posted_total
        .fetch_add(1, Ordering::Relaxed);
    // R1 push: broadcast the new event to every active SSE subscriber on
    // this slot. Dead channels are pruned in-place. The broadcast happens
    // AFTER the disk persist so subscribers and disk readers see the same
    // events; on persist failure we already returned 500 above.
    {
        let mut inner = relay.inner.lock().await;
        if let Some(subs) = inner.streams.get_mut(&slot_id) {
            subs.retain(|tx| tx.send(req.event.clone()).is_ok());
            if subs.is_empty() {
                // Don't leave an empty per-slot Vec behind — `streams` would
                // otherwise accumulate one key per ever-streamed slot.
                inner.streams.remove(&slot_id);
            }
        }
    }
    (
        StatusCode::CREATED,
        Json(json!({"event_id": event_id, "status": "stored"})),
    )
        .into_response()
}

/// R1 — server-sent-events push stream for a slot. Auth'd by slot_token
/// (same as `list_events`). The connection registers an `UnboundedSender`
/// on the slot's subscriber list; every subsequent `post_event` to the slot
/// fans out to all subscribers as `data: <event-json>\n\n` lines. The
/// connection stays open until the client disconnects.
///
/// A 30-second keepalive ping is emitted automatically so reverse proxies
/// (Cloudflare tunnel, nginx) don't time out the upstream.
///
/// Note: the subscriber sees events posted AFTER it subscribed. To catch
/// up on history first, the client should call `GET /v1/events/:slot_id`
/// with `since=` before opening the stream.
async fn stream_events(
    State(relay): State<Relay>,
    Path(slot_id): Path<String>,
    headers: HeaderMap,
) -> axum::response::Response {
    use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
    use futures::stream::StreamExt;

    if let Err(resp) = check_token(&relay, &headers, &slot_id).await {
        return resp;
    }

    let (tx, rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
    {
        let mut inner = relay.inner.lock().await;
        // Prune already-disconnected subscribers before counting toward the cap
        // (post_event only prunes lazily on a broadcast → a silent slot leaks
        // dead senders forever and over-counts against MAX_STREAMS_PER_SLOT).
        if let Some(subs) = inner.streams.get_mut(&slot_id) {
            retain_live_subscribers(subs);
            if subs.is_empty() {
                inner.streams.remove(&slot_id);
            }
        }
        // Audit: cap subscribers per slot so an authed token-holder can't open
        // unbounded streams (each fans out on every post_event → memory + O(n)
        // latency DoS). Over the ceiling → 503.
        let current = inner.streams.get(&slot_id).map_or(0, Vec::len);
        if at_capacity(current, MAX_STREAMS_PER_SLOT) {
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                Json(json!({
                    "error": format!(
                        "phyllis: too many open lines on that slot — max {MAX_STREAMS_PER_SLOT} subscribers"
                    )
                })),
            )
                .into_response();
        }
        inner.streams.entry(slot_id.clone()).or_default().push(tx);
    }

    let stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx).map(|ev| {
        SseEvent::default()
            .json_data(&ev)
            .map_err(|e| std::io::Error::other(e.to_string()))
    });

    Sse::new(stream)
        .keep_alive(
            KeepAlive::new()
                .interval(std::time::Duration::from_secs(30))
                .text("phyllis: still on the line"),
        )
        .into_response()
}

// ---------- pair-slot handlers ----------

#[derive(Deserialize)]
pub struct PairOpenRequest {
    pub code_hash: String,
    /// SPAKE2 message (base64).
    pub msg: String,
    pub role: String, // "host" or "guest"
}

#[derive(Deserialize)]
pub struct PairBootstrapRequest {
    pub role: String,
    pub sealed: String,
}

#[derive(Deserialize)]
pub struct PairAbandonRequest {
    /// SHA-256 hex digest of the code phrase. Same value the caller posts to
    /// /v1/pair as `code_hash` — knowing the code IS the auth here.
    pub code_hash: String,
}

/// Forget the pair-slot associated with this code_hash. Either side can call;
/// no auth beyond knowledge of the code (which is the shared secret of this
/// handshake anyway). Idempotent: returns 204 whether or not the slot exists.
/// Used by clients to recover after a crash mid-handshake, so the host doesn't
/// stay locked out until the 5-minute TTL.
async fn pair_abandon(
    State(relay): State<Relay>,
    Json(req): Json<PairAbandonRequest>,
) -> impl IntoResponse {
    let mut inner = relay.inner.lock().await;
    if let Some(pair_id) = inner.pair_lookup.remove(&req.code_hash) {
        inner.pair_slots.remove(&pair_id);
    }
    StatusCode::NO_CONTENT.into_response()
}

async fn pair_open(
    State(relay): State<Relay>,
    Json(req): Json<PairOpenRequest>,
) -> impl IntoResponse {
    if req.role != "host" && req.role != "guest" {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "role must be 'host' or 'guest'"})),
        )
            .into_response();
    }
    relay.evict_expired_pair_slots().await;
    let mut inner = relay.inner.lock().await;
    let pair_id = match inner.pair_lookup.get(&req.code_hash).cloned() {
        Some(id) => id,
        None => {
            if at_capacity(inner.pair_slots.len(), MAX_PAIR_SLOTS) {
                return (
                    StatusCode::SERVICE_UNAVAILABLE,
                    Json(json!({"error": "relay pair-slot capacity reached"})),
                )
                    .into_response();
            }
            let new_id = random_hex(16);
            inner
                .pair_lookup
                .insert(req.code_hash.clone(), new_id.clone());
            inner.pair_slots.insert(new_id.clone(), PairSlot::default());
            relay
                .counters
                .pair_opens_total
                .fetch_add(1, Ordering::Relaxed);
            new_id
        }
    };
    let slot = inner.pair_slots.entry(pair_id.clone()).or_default();
    slot.last_touched = std::time::Instant::now();
    if req.role == "host" {
        if slot.host_msg.is_some() {
            return (
                StatusCode::CONFLICT,
                Json(json!({"error": "host already registered for this code"})),
            )
                .into_response();
        }
        slot.host_msg = Some(req.msg);
    } else {
        if slot.guest_msg.is_some() {
            return (
                StatusCode::CONFLICT,
                Json(json!({"error": "guest already registered for this code"})),
            )
                .into_response();
        }
        slot.guest_msg = Some(req.msg);
    }
    (StatusCode::CREATED, Json(json!({"pair_id": pair_id}))).into_response()
}

#[derive(Deserialize)]
pub struct PairGetQuery {
    /// "host" or "guest" — caller's role; we return the OTHER side's data.
    pub as_role: String,
}

async fn pair_get(
    State(relay): State<Relay>,
    Path(pair_id): Path<String>,
    Query(q): Query<PairGetQuery>,
) -> impl IntoResponse {
    relay.evict_expired_pair_slots().await;
    let mut inner = relay.inner.lock().await;
    let slot = match inner.pair_slots.get_mut(&pair_id) {
        Some(s) => {
            s.last_touched = std::time::Instant::now();
            s.clone()
        }
        None => {
            return (
                StatusCode::NOT_FOUND,
                Json(json!({"error": "unknown pair_id"})),
            )
                .into_response();
        }
    };
    let (peer_msg, peer_bootstrap) = match q.as_role.as_str() {
        "host" => (slot.guest_msg, slot.guest_bootstrap),
        "guest" => (slot.host_msg, slot.host_bootstrap),
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "as_role must be 'host' or 'guest'"})),
            )
                .into_response();
        }
    };
    (
        StatusCode::OK,
        Json(json!({"peer_msg": peer_msg, "peer_bootstrap": peer_bootstrap})),
    )
        .into_response()
}

async fn pair_bootstrap(
    State(relay): State<Relay>,
    Path(pair_id): Path<String>,
    Json(req): Json<PairBootstrapRequest>,
) -> impl IntoResponse {
    relay.evict_expired_pair_slots().await;
    let mut inner = relay.inner.lock().await;
    let slot = match inner.pair_slots.get_mut(&pair_id) {
        Some(s) => s,
        None => {
            return (
                StatusCode::NOT_FOUND,
                Json(json!({"error": "unknown pair_id"})),
            )
                .into_response();
        }
    };
    slot.last_touched = std::time::Instant::now();
    match req.role.as_str() {
        "host" => slot.host_bootstrap = Some(req.sealed),
        "guest" => slot.guest_bootstrap = Some(req.sealed),
        _ => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "role must be 'host' or 'guest'"})),
            )
                .into_response();
        }
    }
    (StatusCode::CREATED, Json(json!({"ok": true}))).into_response()
}

// ---------- handle directory (v0.5) ----------

#[derive(Deserialize)]
pub struct HandleClaimRequest {
    /// Nick the claimant wants (case-folded). Domain part is implicit: the
    /// domain the relay's `.well-known` is served from.
    pub nick: String,
    /// Slot id the claimant owns on this relay (proves they allocated here).
    pub slot_id: String,
    /// Optional public-facing relay URL the relay should advertise back in
    /// `.well-known/wire/agent` responses. If omitted, callers will need to
    /// know the relay URL out-of-band.
    pub relay_url: Option<String>,
    /// Claimant's full signed agent-card (includes DID + verify_keys +
    /// optional profile).
    pub card: Value,
    /// v0.5.19 (#9.1): set false to opt out of `/v1/handles` bulk listing.
    /// Direct `.well-known/wire/agent` lookup by handle still works.
    /// Omitted on first claim defaults to discoverable=true.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub discoverable: Option<bool>,
}

/// `POST /v1/handle/claim` — claim or update a `nick@<relay-domain>` handle.
///
/// FCFS on nick. Same-DID re-claims allowed (used for profile updates +
/// slot rotation). Different-DID claims on a taken nick return 409.
/// Caller must (a) own the `slot_id` they reference (verified by token
/// being present), and (b) submit a card with a valid self-signature.
async fn handle_claim(
    State(relay): State<Relay>,
    headers: HeaderMap,
    Json(req): Json<HandleClaimRequest>,
) -> impl IntoResponse {
    // Bearer auth: claimant must hold the slot_token for the slot they
    // reference. Prevents nick-squatting from an unauthenticated POSTer.
    if let Err(resp) = check_token(&relay, &headers, &req.slot_id).await {
        return resp;
    }
    // Validate nick (same rules as the client-side parser).
    if !crate::pair_profile::is_valid_nick(&req.nick) {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({
                "error": "phyllis: that handle won't fit in the books — nicks need 2-32 chars, lowercase [a-z0-9_-], not on the reserved list",
                "nick": req.nick,
            })),
        )
            .into_response();
    }
    // Audit L3: the claimant-supplied `relay_url` is echoed verbatim into the
    // advertised A2A `endpoint` (well_known_agent_card_a2a). Reject anything
    // that isn't a clean `http(s)://host[:port]` — no userinfo (the
    // `handle@relay` bug), no `javascript:`/other schemes — so a poisoned
    // endpoint can't be planted in the directory for A2A consumers to follow.
    if let Some(url) = req.relay_url.as_deref()
        && !url.is_empty()
        && !is_valid_public_relay_url(url)
    {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({
                "error": "relay_url must be http(s)://host[:port] with no userinfo",
                "relay_url": url,
            })),
        )
            .into_response();
    }
    // Verify the card signature using the public verify_agent_card helper.
    if let Err(e) = crate::agent_card::verify_agent_card(&req.card) {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": format!("card signature invalid: {e}")})),
        )
            .into_response();
    }
    let did = match req.card.get("did").and_then(Value::as_str) {
        Some(d) => d.to_string(),
        None => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "card missing 'did' field"})),
            )
                .into_response();
        }
    };

    // ONE-NAME rule, enforced server-side. The claimed nick MUST equal
    // the card's DID-derived persona. The client coerces this before
    // POSTing, but that's courtesy — a raw HTTP claim could otherwise
    // map an arbitrary nick (e.g. a well-known handle) onto a foreign
    // DID, so `wire dial <nick>@relay` would resolve to the impostor.
    // `verify_agent_card` above already proved the DID commits to the
    // card's key, so this binds nick → key transitively.
    let canonical_nick = crate::agent_card::display_handle_from_did(&did);
    if req.nick != canonical_nick {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({
                "error": "phyllis: that nick doesn't match your DID — wire publishes one name, and it's the one your key spells out",
                "nick": req.nick,
                "expected": canonical_nick,
            })),
        )
            .into_response();
    }

    // FCFS check. Also snapshot the existing record (clone) so the
    // re-claim path below can preserve fields the client didn't include
    // in the request (notably `discoverable` from v0.5.19, so an old
    // client doing a profile-update re-claim doesn't accidentally
    // re-publish a hidden handle).
    let prior_record: Option<HandleRecord>;
    let first_claim = {
        let inner = relay.inner.lock().await;
        match inner.handles.get(&req.nick) {
            Some(existing) if existing.did != did => {
                return (
                    StatusCode::CONFLICT,
                    Json(json!({
                        "error": "phyllis: this line's already taken by a different identity (persona collision). Your handle is fixed to your key, so claim on another relay (`wire up @<other-relay>`) or mint a fresh identity (`wire nuke` then `wire up`).",
                        "nick": req.nick,
                        "claimed_by": existing.did,
                    })),
                )
                    .into_response();
            }
            Some(prev) => {
                prior_record = Some(prev.clone());
                false
            }
            None => {
                prior_record = None;
                true
            }
        }
    };

    // #291 H1: cap the directory size — but only on a NEW nick. A same-DID
    // re-claim (profile update / re-publish) must always succeed even at the
    // ceiling, since it doesn't grow the map.
    if first_claim {
        let inner = relay.inner.lock().await;
        if at_capacity(inner.handles.len(), MAX_HANDLES) {
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                Json(json!({"error": "relay handle-directory capacity reached"})),
            )
                .into_response();
        }
    }

    // v0.5.19 (#9.5): round claimed_at to whole seconds. Nanosecond
    // precision served no client purpose (display only) and acted as a
    // cross-tab fingerprint correlating one operator's multiple handles
    // claimed in the same session. Truncate before formatting.
    let now = time::OffsetDateTime::now_utc()
        .replace_nanosecond(0)
        .unwrap_or_else(|_| time::OffsetDateTime::now_utc())
        .format(&time::format_description::well_known::Rfc3339)
        .unwrap_or_default();
    // v0.5.19 (#9.1): preserve `discoverable` across re-claims. If the
    // request doesn't set it explicitly, keep whatever the existing
    // record had so a profile-update re-claim doesn't accidentally
    // re-publish a hidden handle. New first-time claims default to
    // discoverable=true explicitly.
    let discoverable = match (req.discoverable, &prior_record) {
        (Some(d), _) => Some(d),
        (None, Some(prev)) => prev.discoverable,
        (None, None) => Some(true),
    };
    let record = HandleRecord {
        nick: req.nick.clone(),
        did: did.clone(),
        card: req.card.clone(),
        slot_id: req.slot_id.clone(),
        relay_url: req.relay_url.clone(),
        claimed_at: now,
        discoverable,
    };

    // Persist to disk first (durable), then update in-memory.
    let path = relay
        .state_dir
        .join("handles")
        .join(format!("{}.json", req.nick));
    let body = match serde_json::to_vec_pretty(&record) {
        Ok(b) => b,
        Err(e) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": format!("serialize failed: {e}")})),
            )
                .into_response();
        }
    };
    if let Err(e) = tokio::fs::write(&path, &body).await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": format!("persist failed: {e}")})),
        )
            .into_response();
    }
    {
        let mut inner = relay.inner.lock().await;
        inner.handles.insert(req.nick.clone(), record);
    }
    relay
        .counters
        .handle_claims_total
        .fetch_add(1, Ordering::Relaxed);
    if first_claim {
        relay
            .counters
            .handle_first_claims_total
            .fetch_add(1, Ordering::Relaxed);
    }
    (
        StatusCode::CREATED,
        Json(json!({
            "nick": req.nick,
            "did": did,
            "status": if first_claim { "claimed" } else { "re-claimed" },
        })),
    )
        .into_response()
}

/// `DELETE /v1/handle/claim/:nick` — release a claimed handle (#247.1). Without
/// this, a claim is FCFS-permanent (no expiry, no unclaim) and an abandoned
/// handle squats the directory forever. Owner-gated: the caller must present the
/// bearer `slot_token` of the handle's slot, so only the holder can release it.
/// Removes the in-memory entry and its on-disk file. Idempotent-ish: an
/// unknown nick is 404.
async fn handle_unclaim(
    State(relay): State<Relay>,
    Path(nick): Path<String>,
    headers: HeaderMap,
) -> impl IntoResponse {
    // Validate before any filesystem use (the nick is interpolated into a path).
    if !crate::pair_profile::is_valid_nick(&nick) {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "invalid nick"})),
        )
            .into_response();
    }
    // Resolve the handle's slot so we can owner-gate the unclaim.
    let slot_id = {
        let inner = relay.inner.lock().await;
        match inner.handles.get(&nick) {
            Some(rec) => rec.slot_id.clone(),
            None => {
                return (
                    StatusCode::NOT_FOUND,
                    Json(json!({"error": format!("{nick:?} isn't claimed")})),
                )
                    .into_response();
            }
        }
    };
    // Owner-only: must hold the slot_token for the handle's slot.
    if let Err(resp) = check_token(&relay, &headers, &slot_id).await {
        return resp;
    }
    // Remove on-disk file (best-effort — absence is fine) then in-memory entry.
    let path = relay.state_dir.join("handles").join(format!("{nick}.json"));
    let _ = tokio::fs::remove_file(&path).await;
    {
        let mut inner = relay.inner.lock().await;
        inner.handles.remove(&nick);
    }
    (
        StatusCode::OK,
        Json(json!({"nick": nick, "status": "unclaimed"})),
    )
        .into_response()
}

#[derive(Deserialize)]
pub struct WellKnownAgentQuery {
    pub handle: String,
}

#[derive(Deserialize)]
pub struct HandlesDirectoryQuery {
    pub cursor: Option<String>,
    pub limit: Option<usize>,
    pub vibe: Option<String>,
}

// ─── short-URL invites (v0.5.10) ──────────────────────────────────────────
// One-curl onboarding: the invitor registers their `wire://pair?...` URL
// here, gets back a 6-hex token. Anyone who does
//   curl -fsSL https://wireup.net/i/<token> | sh
// gets wire installed (if needed) + the invite accepted, in one shot.
//
// Possession of the short URL = pair authorization (same shape as the
// underlying wire:// invite — it's just a redirector).

#[derive(Deserialize)]
pub struct InviteRegisterRequest {
    /// The wire://pair?... URL produced by `wire invite`. Required.
    pub invite_url: String,
    /// Lifetime in seconds. Default 86400 (24h). Capped at 7 days.
    #[serde(default)]
    pub ttl_seconds: Option<u64>,
    /// If `Some(n)`, the short URL can be fetched N times before 410s.
    /// `None` = unlimited until TTL hits.
    #[serde(default)]
    pub uses: Option<u32>,
}

impl Relay {
    /// Append one InviteRecord to `<state_dir>/invites.jsonl`.
    async fn persist_invite(&self, rec: &InviteRecord) -> Result<()> {
        use tokio::io::AsyncWriteExt;
        let mut line = serde_json::to_vec(rec)?;
        line.push(b'\n');
        let path = self.state_dir.join("invites.jsonl");
        let mut f = tokio::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .await?;
        f.write_all(&line).await?;
        f.flush().await?;
        Ok(())
    }
}

async fn invite_register(
    State(relay): State<Relay>,
    Json(req): Json<InviteRegisterRequest>,
) -> impl IntoResponse {
    if req.invite_url.is_empty() {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "invite_url required"})),
        )
            .into_response();
    }
    // Length cap on the embedded URL to keep persisted records bounded.
    if req.invite_url.len() > 8_192 {
        return (
            StatusCode::PAYLOAD_TOO_LARGE,
            Json(json!({"error": "invite_url > 8 KiB"})),
        )
            .into_response();
    }
    let ttl = req.ttl_seconds.unwrap_or(86_400).clamp(60, 7 * 86_400);
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    // 16-hex token → 2^64 space. Brute-force infeasible.
    let token = random_hex(8);
    let rec = InviteRecord {
        token: token.clone(),
        invite_url: req.invite_url,
        expires_unix: now + ttl,
        uses_remaining: req.uses,
        created_unix: now,
    };
    {
        let mut inner = relay.inner.lock().await;
        if at_capacity(inner.invites.len(), MAX_INVITES) {
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                Json(json!({"error": "relay invite capacity reached"})),
            )
                .into_response();
        }
        if inner.invites.contains_key(&token) {
            return (
                StatusCode::CONFLICT,
                Json(json!({"error": "token collision, retry"})),
            )
                .into_response();
        }
        inner.invites.insert(token.clone(), rec.clone());
    }
    if let Err(e) = relay.persist_invite(&rec).await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": format!("persist failed: {e}")})),
        )
            .into_response();
    }
    (
        StatusCode::CREATED,
        Json(json!({
            "token": token,
            "path": format!("/i/{token}"),
            "expires_unix": rec.expires_unix,
            "uses_remaining": rec.uses_remaining,
        })),
    )
        .into_response()
}

#[derive(Deserialize)]
pub struct InviteScriptQuery {
    /// `format=url` returns the raw `wire://pair?...` URL as text/plain
    /// (used by `wire accept https://wireup.net/i/<token>` to resolve a
    /// short URL programmatically). Default: shell-script template.
    /// Note: ?format=url does NOT decrement `uses_remaining` — it's a
    /// resolution lookup, not an acceptance. The actual accept happens
    /// when the wire:// URL is consumed by `pair_invite::accept_invite`.
    pub format: Option<String>,
}

async fn invite_script(
    State(relay): State<Relay>,
    Path(token): Path<String>,
    Query(q): Query<InviteScriptQuery>,
) -> impl IntoResponse {
    // Token shape: 6 lowercase hex. Reject anything else immediately so a
    // path-traversal try never reaches the map lookup.
    if token.len() != 6 || !token.chars().all(|c| c.is_ascii_hexdigit()) {
        return (StatusCode::NOT_FOUND, "not found\n").into_response();
    }
    let want_raw_url = q.format.as_deref() == Some("url");
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let invite_url = {
        let mut inner = relay.inner.lock().await;
        let Some(rec) = inner.invites.get_mut(&token) else {
            return (StatusCode::NOT_FOUND, "not found\n").into_response();
        };
        if rec.expires_unix <= now {
            return (StatusCode::GONE, "this invite has expired\n").into_response();
        }
        if let Some(n) = rec.uses_remaining {
            if n == 0 {
                return (StatusCode::GONE, "this invite has been used up\n").into_response();
            }
            // Only decrement on script-template fetch (the one that's
            // actually doing the pair). The raw-URL resolution path is a
            // lookup, not an accept.
            if !want_raw_url {
                rec.uses_remaining = Some(n - 1);
            }
        }
        rec.invite_url.clone()
    };
    if want_raw_url {
        return (
            StatusCode::OK,
            [
                (
                    axum::http::header::CONTENT_TYPE,
                    "text/plain; charset=utf-8",
                ),
                (
                    axum::http::header::CACHE_CONTROL,
                    "private, no-store, max-age=0",
                ),
            ],
            invite_url,
        )
            .into_response();
    }
    let escaped = invite_url.replace('\'', "'\\''");
    let script = format!(
        "#!/bin/sh\n\
         # wire — one-curl onboarding (install + pair in one shot)\n\
         # source: https://github.com/SlanchaAi/wire\n\
         set -eu\n\
         INVITE='{escaped}'\n\
         echo \"\u{2192} checking for wire CLI...\"\n\
         if ! command -v wire >/dev/null 2>&1; then\n  \
           echo \"\u{2192} wire not installed; installing first...\"\n  \
           curl -fsSL https://wireup.net/install.sh | sh\n  \
           case \":$PATH:\" in\n    \
             *:\"$HOME/.local/bin\":*) ;;\n    \
             *) export PATH=\"$HOME/.local/bin:$PATH\" ;;\n  \
           esac\n  \
           if ! command -v wire >/dev/null 2>&1; then\n    \
             echo \"\"\n    \
             echo \"wire was installed to ~/.local/bin but it's not on \\$PATH yet.\"\n    \
             echo \"Open a new shell, then run:\"\n    \
             echo \"  wire accept '$INVITE'\"\n    \
             exit 0\n  \
           fi\n\
         fi\n\
         echo \"\u{2192} accepting invite...\"\n\
         wire accept \"$INVITE\"\n"
    );
    (
        StatusCode::OK,
        [
            (
                axum::http::header::CONTENT_TYPE,
                "text/x-shellscript; charset=utf-8",
            ),
            (
                axum::http::header::CACHE_CONTROL,
                "private, no-store, max-age=0",
            ),
        ],
        script,
    )
        .into_response()
}

async fn handles_directory(
    State(relay): State<Relay>,
    Query(q): Query<HandlesDirectoryQuery>,
) -> impl IntoResponse {
    let limit = q.limit.unwrap_or(100).clamp(1, 500);
    let vibe_filter = q.vibe.as_ref().map(|v| v.to_ascii_lowercase());
    let inner = relay.inner.lock().await;
    let mut records: Vec<HandleRecord> = inner.handles.values().cloned().collect();
    drop(inner);
    records.sort_by(|a, b| a.nick.cmp(&b.nick));

    let cursor = q.cursor.as_deref();
    let mut eligible = Vec::new();
    for rec in records {
        if cursor.is_some_and(|c| rec.nick.as_str() <= c) {
            continue;
        }
        // Hygiene: hide test-shaped nicks from the public directory. Records
        // remain claimed (FCFS protection persists), they just don't surface
        // in the phone book. `demo-` is reserved for asciinema-cast handles,
        // `test-` for integration runs.
        if rec.nick.starts_with("demo-") || rec.nick.starts_with("test-") {
            continue;
        }
        // v0.5.19 (#9.1): operator opt-out — hidden handles skip the
        // bulk directory but still resolve via `.well-known/wire/agent?
        // handle=X` for out-of-band sharing.
        if !rec.is_discoverable() {
            continue;
        }
        let profile = rec.card.get("profile").cloned().unwrap_or(Value::Null);
        if profile
            .get("listed")
            .and_then(Value::as_bool)
            .is_some_and(|listed| !listed)
        {
            continue;
        }
        if let Some(want) = &vibe_filter {
            let matched = profile
                .get("vibe")
                .and_then(Value::as_array)
                .map(|arr| {
                    arr.iter().any(|v| {
                        v.as_str()
                            .map(|s| s.eq_ignore_ascii_case(want))
                            .unwrap_or(false)
                    })
                })
                .unwrap_or(false);
            if !matched {
                continue;
            }
        }
        eligible.push((rec, profile));
    }

    let has_more = eligible.len() > limit;
    let page = eligible.into_iter().take(limit).collect::<Vec<_>>();
    let next_cursor = if has_more {
        page.last().map(|(rec, _)| rec.nick.clone())
    } else {
        None
    };
    let handles: Vec<Value> = page
        .into_iter()
        .map(|(rec, profile)| {
            // v0.12.1: fall back to the DID-derived persona emoji when the
            // card carries no explicit profile emoji, so every phonebook
            // line shows a face next to the name. The persona is a
            // deterministic function of the DID, so the relay can compute it
            // without the claimant having set anything.
            let emoji = profile
                .get("emoji")
                .and_then(Value::as_str)
                .filter(|s| !s.is_empty())
                .map(str::to_string)
                .unwrap_or_else(|| crate::character::Character::from_did(&rec.did).emoji);
            json!({
                "nick": rec.nick,
                "did": rec.did,
                "profile": {
                    "emoji": emoji,
                    "motto": profile.get("motto").cloned().unwrap_or(Value::Null),
                    "vibe": profile.get("vibe").cloned().unwrap_or(Value::Null),
                    "pronouns": profile.get("pronouns").cloned().unwrap_or(Value::Null),
                    "now": profile.get("now").cloned().unwrap_or(Value::Null),
                },
                "claimed_at": rec.claimed_at,
            })
        })
        .collect();
    (
        StatusCode::OK,
        Json(json!({
            "handles": handles,
            "next_cursor": next_cursor,
        })),
    )
        .into_response()
}

/// `POST /v1/handle/intro/:nick` — drop a signed pair-introduction event
/// into a known nick's slot WITHOUT needing that slot's bearer token.
///
/// Why this exists: `.well-known/wire/agent` returns a nick's `slot_id` for
/// reachability, but NEVER its `slot_token` (that would leak read+write
/// authority to any handle-resolver). To zero-paste-pair, we need a way for
/// a stranger to deliver their signed agent-card to the nick's owner. This
/// endpoint provides exactly that, and ONLY that: the event must be `kind=1100`
/// (pair_drop / agent_card), self-signed, and the carrying agent-card embedded
/// in the body must verify-OK on its own.
///
/// SECURITY (audit / #247 finding 3): this route is UNAUTHENTICATED by design
/// (a stranger must be able to drop a pair-intro) and is NOT under the governor
/// layer — it sits on the main router, not in `hot_writes`. The only backpressure
/// is the per-slot 64MB quota, which is shared with the owner's own inbound, so a
/// flood of intros to a known nick can fill the victim's slot and 413 their
/// legitimate traffic. A per-nick / per-source intro rate-limit (and a separate
/// intro-byte budget so a flood can't evict the owner) is tracked in #247.3 —
/// it is NOT in place today. (Earlier comment here wrongly claimed governor
/// coverage.)
async fn handle_intro(
    State(relay): State<Relay>,
    Path(nick): Path<String>,
    Json(req): Json<PostEventRequest>,
) -> impl IntoResponse {
    // Look up the nick. Must already be claimed.
    let slot_id = {
        let mut inner = relay.inner.lock().await;
        let slot_id = match inner.handles.get(&nick) {
            Some(rec) => rec.slot_id.clone(),
            None => {
                return (
                    StatusCode::NOT_FOUND,
                    Json(json!({"error": format!("phyllis: that number's been disconnected — {nick:?} isn't claimed on this switchboard")})),
                )
                    .into_response();
            }
        };
        // #247.3: per-nick intro rate-limit. This endpoint is unauthenticated
        // by design (a stranger drops a pair-intro), so without a per-nick cap
        // an attacker can flood a known nick's slot to MAX_SLOT_BYTES and DoS
        // them. The global governor doesn't help (it also throttles everyone,
        // and 10/s × 256KiB still fills 64MB in ~25s). Cap intros per nick per
        // window; over the limit → 429.
        let now = unix_now();
        // Audit hygiene: before tracking this nick, opportunistically sweep
        // fully-aged nicks once the map grows past a soft cap, so it can't
        // retain one stale entry per ever-touched nick.
        if at_capacity(inner.intro_times.len(), MAX_INTRO_TRACKING_NICKS) {
            evict_stale_intro_nicks(&mut inner.intro_times, now, INTRO_WINDOW_SECS);
        }
        let times = inner.intro_times.entry(nick.clone()).or_default();
        if !record_intro_within_rate(times, now, INTRO_WINDOW_SECS, INTRO_MAX_PER_WINDOW) {
            return (
                StatusCode::TOO_MANY_REQUESTS,
                Json(json!({
                    "error": format!(
                        "too many pair-intros to {nick:?} — rate limited ({INTRO_MAX_PER_WINDOW} per {INTRO_WINDOW_SECS}s). Try again shortly."
                    ),
                })),
            )
                .into_response();
        }
        slot_id
    };

    // Only allow kind=1100 pair_drop / agent_card here. Anything else routes
    // to the standard /v1/events/:slot_id with bearer auth.
    let kind = req.event.get("kind").and_then(Value::as_u64).unwrap_or(0);
    let type_str = req.event.get("type").and_then(Value::as_str).unwrap_or("");
    if !intro_event_allowed(kind, type_str) {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({
                "error": "intro endpoint only accepts kind=1100 pair_drop / agent_card events",
                "got_kind": kind,
                "got_type": type_str,
            })),
        )
            .into_response();
    }

    // Body must embed a signed agent-card (so the receiver can pin from it).
    let embedded_card = match req.event.get("body").and_then(|b| b.get("card")) {
        Some(c) => c.clone(),
        None => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": "intro event body must embed 'card' field"})),
            )
                .into_response();
        }
    };
    if let Err(e) = crate::agent_card::verify_agent_card(&embedded_card) {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": format!("embedded card signature invalid: {e}")})),
        )
            .into_response();
    }

    // Size + quota checks (same as post_event).
    let body_bytes = match serde_json::to_vec(&req.event) {
        Ok(b) => b,
        Err(e) => {
            return (
                StatusCode::BAD_REQUEST,
                Json(json!({"error": format!("event not serializable: {e}")})),
            )
                .into_response();
        }
    };
    if body_bytes.len() > MAX_EVENT_BYTES {
        return (
            StatusCode::PAYLOAD_TOO_LARGE,
            Json(json!({"error": "intro event exceeds 256 KiB", "max_bytes": MAX_EVENT_BYTES})),
        )
            .into_response();
    }
    {
        let inner = relay.inner.lock().await;
        let used = inner.slot_bytes.get(&slot_id).copied().unwrap_or(0);
        if used + body_bytes.len() > MAX_SLOT_BYTES {
            return (
                StatusCode::PAYLOAD_TOO_LARGE,
                Json(json!({
                    "error": "target slot quota exceeded",
                    "slot_bytes_used": used,
                    "slot_bytes_max": MAX_SLOT_BYTES,
                })),
            )
                .into_response();
        }
    }

    let event_id = req
        .event
        .get("event_id")
        .and_then(Value::as_str)
        .map(str::to_string);

    // Dedupe by event_id if present.
    let dup = {
        let inner = relay.inner.lock().await;
        let slot = inner.slots.get(&slot_id);
        if let (Some(eid), Some(slot)) = (&event_id, slot) {
            slot.iter()
                .any(|e| e.get("event_id").and_then(Value::as_str) == Some(eid))
        } else {
            false
        }
    };
    if dup {
        return (
            StatusCode::OK,
            Json(json!({"event_id": event_id, "status": "duplicate"})),
        )
            .into_response();
    }

    {
        let mut inner = relay.inner.lock().await;
        let event_size = body_bytes.len();
        let slot = inner.slots.entry(slot_id.clone()).or_default();
        slot.push(req.event.clone());
        *inner.slot_bytes.entry(slot_id.clone()).or_insert(0) += event_size;
    }
    if let Err(e) = relay.append_event_to_disk(&slot_id, &req.event).await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": format!("persist failed: {e}")})),
        )
            .into_response();
    }
    (
        StatusCode::CREATED,
        Json(json!({"event_id": event_id, "status": "dropped", "to_nick": nick})),
    )
        .into_response()
}

/// `GET /.well-known/wire/agent?handle=<nick>` — WebFinger-style resolver
/// for `nick@<this-relay-domain>` handles. Returns the signed agent-card +
/// slot coords if claimed; 404 if not.
///
/// The `handle` query parameter may be just `<nick>` or `<nick>@<domain>`.
/// Domain part is ignored (the relay only serves nicks it has on file).
/// `GET /.well-known/agent-card.json?handle=<nick>` — A2A v1.0-compatible
/// AgentCard serving wire's handle directory. Same data as `well_known_agent`
/// but in the schema A2A clients (MSFT/AWS/Salesforce/SAP/ServiceNow tooling,
/// agent-card-go, agent-card-python, A2A .NET SDK) already speak.
///
/// Wire-specific fields (DID, slot_id, profile blob, raw signed card) live
/// under the standard A2A `extensions` array using the wire extension URI.
/// A2A-only clients can pair to wire agents knowing only A2A vocabulary;
/// wire-native clients get the full richer card by following the extension.
async fn well_known_agent_card_a2a(
    State(relay): State<Relay>,
    Query(q): Query<WellKnownAgentQuery>,
) -> impl IntoResponse {
    let nick = q.handle.split('@').next().unwrap_or("").to_string();
    if nick.is_empty() {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "handle missing nick"})),
        )
            .into_response();
    }
    let inner = relay.inner.lock().await;
    let rec = match inner.handles.get(&nick) {
        Some(r) => r.clone(),
        None => {
            return (
                StatusCode::NOT_FOUND,
                Json(json!({"error": format!("phyllis: that number's been disconnected — {nick:?} isn't claimed on this switchboard")})),
            )
                .into_response();
        }
    };
    drop(inner);

    let profile = rec.card.get("profile").cloned().unwrap_or(Value::Null);
    let description = profile
        .get("motto")
        .and_then(Value::as_str)
        .unwrap_or("")
        .to_string();
    let display_name = profile
        .get("display_name")
        .and_then(Value::as_str)
        .unwrap_or(&rec.nick)
        .to_string();
    let relay_url = rec.relay_url.clone().unwrap_or_default();
    // Intro endpoint = where any A2A or wire client posts a signed pair-drop.
    let endpoint = if !relay_url.is_empty() {
        format!(
            "{}/v1/handle/intro/{}",
            relay_url.trim_end_matches('/'),
            rec.nick
        )
    } else {
        format!("/v1/handle/intro/{}", rec.nick)
    };
    let card_sig = rec.card.get("signature").cloned().unwrap_or(Value::Null);

    // Build A2A v1.0 AgentCard shape with wire extension. Fields named to
    // match the A2A spec exactly so downstream tooling (agent-card-go etc.)
    // parses without custom code.
    let a2a_card = json!({
        "id": rec.did,
        "name": display_name,
        "description": description,
        "version": "wire/0.5",
        "endpoint": endpoint,
        "provider": {
            "name": "wire",
            "url": "https://github.com/SlanchaAi/wire"
        },
        "capabilities": {
            "streaming": false,
            "pushNotifications": false,
            "extendedAgentCard": true
        },
        "securitySchemes": {
            "ed25519-event-sig": {
                "type": "signature",
                "alg": "EdDSA",
                "description": "Wire-style signed events (kind=1100 pair_drop for intro; verify against embedded card pubkey)."
            }
        },
        "security": [{"ed25519-event-sig": []}],
        "skills": [],
        "extensions": [{
            // A2A extension URIs are opaque namespace identifiers, not
            // forwardable URLs. Changing this string is a coordinated
            // federation-spec bump because peers match it exactly.
            "uri": "https://slancha.ai/wire/ext/v0.5",
            "description": "Wire-native fields: full signed agent-card, profile blob, DID, slot_id, mailbox relay coords.",
            "required": false,
            "params": {
                "did": rec.did,
                "handle": rec.nick,
                "slot_id": rec.slot_id,
                "relay_url": rec.relay_url,
                "card": rec.card,
                "profile": profile,
                "claimed_at": rec.claimed_at,
            }
        }],
        "signature": card_sig,
    });
    (StatusCode::OK, Json(a2a_card)).into_response()
}

async fn well_known_agent(
    State(relay): State<Relay>,
    Query(q): Query<WellKnownAgentQuery>,
) -> impl IntoResponse {
    let nick = q.handle.split('@').next().unwrap_or("").to_string();
    if nick.is_empty() {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "handle missing nick"})),
        )
            .into_response();
    }
    let inner = relay.inner.lock().await;
    match inner.handles.get(&nick) {
        Some(rec) => (
            StatusCode::OK,
            Json(json!({
                "nick": rec.nick,
                "did": rec.did,
                "card": rec.card,
                "slot_id": rec.slot_id,
                "relay_url": rec.relay_url,
                "claimed_at": rec.claimed_at,
            })),
        )
            .into_response(),
        None => (
            StatusCode::NOT_FOUND,
            Json(json!({"error": format!("phyllis: that number's been disconnected — {nick:?} isn't claimed on this switchboard")})),
        )
            .into_response(),
    }
}

async fn list_events(
    State(relay): State<Relay>,
    Path(slot_id): Path<String>,
    Query(q): Query<ListEventsQuery>,
    headers: HeaderMap,
) -> impl IntoResponse {
    if let Err(resp) = check_token(&relay, &headers, &slot_id).await {
        return resp;
    }
    let limit = q.limit.unwrap_or(100).min(1000);
    // R4: record this pull as proof that the slot owner is still polling.
    // Anyone holding the slot_token (i.e., a paired peer) can later read
    // last_pull_at_unix via /v1/slot/:slot_id/state to gauge attentiveness.
    let now_unix = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let mut inner = relay.inner.lock().await;
    inner.last_pull_at_unix.insert(slot_id.clone(), now_unix);
    // Borrow the slot and clone ONLY the requested window. Pre-fix this
    // `.cloned()`'d the WHOLE slot Vec (bounded only by MAX_SLOT_BYTES = 64 MiB)
    // under the global mutex on every pull — a multi-MB memcpy serialized under
    // the lock that post_event and every other handler contend on, which was the
    // #342 saturation hot path. The `since` position scan stays under the lock
    // but only walks refs (no allocation); the sole clone is the <=limit slice.
    let slice: Vec<Value> = match inner.slots.get(&slot_id) {
        Some(events) => {
            let start = match q.since {
                Some(ref eid) => events
                    .iter()
                    .position(|e| e.get("event_id").and_then(Value::as_str) == Some(eid.as_str()))
                    .map(|i| i + 1)
                    .unwrap_or(0),
                None => 0,
            };
            let end = (start + limit).min(events.len());
            events[start..end].to_vec()
        }
        None => Vec::new(),
    };
    drop(inner); // release before serializing the response
    (StatusCode::OK, Json(slice)).into_response()
}

/// R4 — slot-attentiveness probe. Authenticated by slot_token (so only
/// paired peers can ask). Returns `last_pull_at_unix` (the slot owner's most
/// recent `list_events` call, in unix seconds) and `event_count` (total
/// stored). A remote sender uses this before `wire send <peer>` to warn the
/// operator if the peer hasn't polled recently.
async fn slot_state(
    State(relay): State<Relay>,
    Path(slot_id): Path<String>,
    headers: HeaderMap,
) -> impl IntoResponse {
    if let Err(resp) = check_token(&relay, &headers, &slot_id).await {
        return resp;
    }
    let inner = relay.inner.lock().await;
    let event_count = inner.slots.get(&slot_id).map(|v| v.len()).unwrap_or(0);
    let last_pull_at_unix = inner.last_pull_at_unix.get(&slot_id).copied();
    let responder_health = inner.responder_health.get(&slot_id).cloned();
    (
        StatusCode::OK,
        Json(json!({
            "slot_id": slot_id,
            "event_count": event_count,
            "last_pull_at_unix": last_pull_at_unix,
            "responder_health": responder_health,
        })),
    )
        .into_response()
}

async fn responder_health_set(
    State(relay): State<Relay>,
    Path(slot_id): Path<String>,
    headers: HeaderMap,
    Json(record): Json<ResponderHealthRecord>,
) -> impl IntoResponse {
    // Defense-in-depth (audit L1): `slot_id` is interpolated into a filesystem
    // path below, so validate its shape BEFORE use even though `check_token`
    // (which only matches `random_hex(16)` keys) would already reject a
    // traversal id. Mirrors the guard in `append_event_to_disk`.
    if !is_valid_slot_id(&slot_id) {
        return (
            StatusCode::BAD_REQUEST,
            Json(json!({"error": "invalid slot_id"})),
        )
            .into_response();
    }
    if let Err(resp) = check_token(&relay, &headers, &slot_id).await {
        return resp;
    }
    let path = relay
        .state_dir
        .join("responder-health")
        .join(format!("{slot_id}.json"));
    let body = match serde_json::to_vec_pretty(&record) {
        Ok(b) => b,
        Err(e) => {
            return (
                StatusCode::INTERNAL_SERVER_ERROR,
                Json(json!({"error": format!("serialize failed: {e}")})),
            )
                .into_response();
        }
    };
    if let Err(e) = tokio::fs::write(&path, body).await {
        return (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(json!({"error": format!("persist failed: {e}")})),
        )
            .into_response();
    }
    {
        let mut inner = relay.inner.lock().await;
        inner
            .responder_health
            .insert(slot_id.clone(), record.clone());
    }
    (StatusCode::OK, Json(record)).into_response()
}

async fn check_token(
    relay: &Relay,
    headers: &HeaderMap,
    slot_id: &str,
) -> std::result::Result<(), axum::response::Response> {
    let auth = headers
        .get(AUTHORIZATION)
        .and_then(|h| h.to_str().ok())
        .and_then(|s| s.strip_prefix("Bearer "))
        .map(str::to_string);
    let presented = match auth {
        Some(t) => t,
        None => {
            return Err((
                StatusCode::UNAUTHORIZED,
                Json(json!({"error": "missing Bearer token"})),
            )
                .into_response());
        }
    };
    let inner = relay.inner.lock().await;
    let expected = match inner.tokens.get(slot_id) {
        Some(t) => t.clone(),
        None => {
            return Err((
                StatusCode::NOT_FOUND,
                Json(json!({"error": "unknown slot"})),
            )
                .into_response());
        }
    };
    drop(inner);
    if !constant_time_eq(presented.as_bytes(), expected.as_bytes()) {
        return Err((StatusCode::FORBIDDEN, Json(json!({"error": "bad token"}))).into_response());
    }
    Ok(())
}

fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut acc = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        acc |= x ^ y;
    }
    acc == 0
}

fn is_valid_slot_id(s: &str) -> bool {
    s.len() == 32
        && s.bytes()
            .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase())
}

/// The unauthenticated `/v1/handle/intro` endpoint accepts an event ONLY when it
/// is a `kind=1100` pair-intro of type `pair_drop` or `agent_card`; everything
/// else must route through the bearer-authed `/v1/events/:slot_id`. Extracted as
/// a pure predicate so the accept/reject matrix is locked by a unit test — #334
/// fixed a boolean bug here (`&&`-chain accepted a kind=1100 event of ANY type,
/// and a wrong-kind event whose type happened to match), and this guards the
/// security property against regressing.
fn intro_event_allowed(kind: u64, type_str: &str) -> bool {
    kind == 1100 && (type_str == "pair_drop" || type_str == "agent_card")
}

/// A claimant-advertised `relay_url` is acceptable for the public directory iff
/// it is `http://` or `https://`, has a non-empty host authority, and carries NO
/// userinfo (`user@host` — the `handle@relay` bug) and no embedded
/// whitespace/control. Audit L3 — the value is echoed into the A2A `endpoint`,
/// so a non-URL / alternate-scheme value must not be planted there. Pure +
/// unit-tested. Intentionally strict: scheme + authority shape only, not a full
/// RFC-3986 parse.
fn is_valid_public_relay_url(u: &str) -> bool {
    let rest = match u
        .strip_prefix("https://")
        .or_else(|| u.strip_prefix("http://"))
    {
        Some(r) => r,
        None => return false,
    };
    // Authority is everything up to the first '/'.
    let authority = rest.split('/').next().unwrap_or("");
    if authority.is_empty() {
        return false;
    }
    // No userinfo (`user[:pass]@host`).
    if authority.contains('@') {
        return false;
    }
    // No whitespace / control chars anywhere in the URL.
    if u.chars().any(|c| c.is_whitespace() || c.is_control()) {
        return false;
    }
    // Host (strip optional :port) must be non-empty.
    let host = authority.split(':').next().unwrap_or("");
    !host.is_empty()
}

fn random_hex(n_bytes: usize) -> String {
    let mut buf = vec![0u8; n_bytes];
    rand::thread_rng().fill_bytes(&mut buf);
    hex::encode(buf)
}

/// Run the relay until SIGINT/SIGTERM.
pub async fn serve(bind: &str, state_dir: PathBuf) -> Result<()> {
    serve_with_mode(bind, state_dir, ServerMode::default()).await
}

/// v0.5.17: server-mode-aware entry point. Same as `serve` but with the
/// `--local-only` toggle exposed so the binary can refuse to publish
/// phonebook + well-known surfaces on within-machine relays.
pub async fn serve_with_mode(bind: &str, state_dir: PathBuf, mode: ServerMode) -> Result<()> {
    let relay = Relay::new(state_dir).await?;
    relay.spawn_pair_sweeper();
    relay.spawn_counter_persister();
    let app = relay.clone().router_with_mode(mode);
    let listener = tokio::net::TcpListener::bind(bind)
        .await
        .with_context(|| format!("binding {bind}"))?;
    if mode.local_only {
        eprintln!(
            "wire relay-server (LOCAL-ONLY) listening on {bind} — phonebook + well-known endpoints disabled"
        );
    } else {
        eprintln!("wire relay-server listening on {bind}");
    }
    let shutdown_relay = relay.clone();
    axum::serve(listener, app)
        .with_graceful_shutdown(async move {
            let _ = tokio::signal::ctrl_c().await;
            eprintln!("\nshutting down — final counter snapshot");
            if let Err(e) = shutdown_relay.persist_counters().await {
                eprintln!("final counter persist failed: {e}");
            }
        })
        .await?;
    Ok(())
}

/// v0.7.0-alpha.16: Unix Domain Socket entry point. Mirrors `serve_with_mode`
/// but binds to a UDS path instead of a TCP host:port. Implicitly forces
/// `mode.local_only = true` because UDS has no concept of "publish to a
/// public phonebook." Chmods the socket 0600 so only the owner uid can
/// connect — the SO_PEERCRED-equivalent trust anchor for sister sessions.
///
/// Removes any stale socket file at `path` first (otherwise bind fails
/// with EADDRINUSE). Removes the socket on graceful shutdown.
///
/// Implementation note: axum 0.7's `serve` is TcpListener-only, so we
/// run the manual hyper accept loop (per-connection
/// `hyper::server::conn::http1::Builder`). When axum 0.8 ships with
/// generic `Listener` support, this can collapse to one line.
#[cfg(unix)]
pub async fn serve_uds(socket_path: PathBuf, state_dir: PathBuf) -> Result<()> {
    use hyper::server::conn::http1;
    use hyper_util::rt::TokioIo;
    use tower_service::Service;

    // Best-effort cleanup of stale socket file.
    if socket_path.exists() {
        std::fs::remove_file(&socket_path)
            .with_context(|| format!("removing stale socket at {socket_path:?}"))?;
    }
    if let Some(parent) = socket_path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("creating socket parent {parent:?}"))?;
    }
    let relay = Relay::new(state_dir).await?;
    relay.spawn_pair_sweeper();
    relay.spawn_counter_persister();
    let app: axum::Router = relay
        .clone()
        .router_with_mode(ServerMode { local_only: true });
    let listener = tokio::net::UnixListener::bind(&socket_path)
        .with_context(|| format!("binding UDS at {socket_path:?}"))?;

    // 0600: owner-rw only. Trust anchor is the kernel-attested peer uid
    // (SO_PEERCRED equivalent); chmod is defense-in-depth.
    use std::os::unix::fs::PermissionsExt;
    if let Err(e) = std::fs::set_permissions(&socket_path, std::fs::Permissions::from_mode(0o600)) {
        eprintln!(
            "wire relay-server (UDS): chmod 0600 on {socket_path:?} failed: {e} — \
             socket may be accessible to other uids. Investigate."
        );
    }
    eprintln!(
        "wire relay-server (UDS) listening on unix://{} — same-host, owner-uid only",
        socket_path.display()
    );

    // Manual accept loop + per-conn hyper serve (axum 0.7's serve is
    // TcpListener-only). On SIGINT, persist counters + remove socket.
    let shutdown_relay = relay.clone();
    let socket_path_for_cleanup = socket_path.clone();
    let mut make_service = app.into_make_service();

    let serve_loop = async {
        loop {
            let (stream, _peer_addr) = match listener.accept().await {
                Ok(p) => p,
                Err(e) => {
                    eprintln!("wire relay-server (UDS): accept failed: {e}");
                    continue;
                }
            };
            let tower_service = match make_service.call(&stream).await {
                Ok(s) => s,
                Err(infallible) => match infallible {},
            };
            let io = TokioIo::new(stream);
            let hyper_service =
                hyper::service::service_fn(move |req: hyper::Request<hyper::body::Incoming>| {
                    let mut svc = tower_service.clone();
                    async move { Service::call(&mut svc, req).await }
                });
            tokio::task::spawn(async move {
                if let Err(e) = http1::Builder::new()
                    .serve_connection(io, hyper_service)
                    .await
                {
                    // Connection-level error; not fatal to the listener.
                    if !e.is_incomplete_message() {
                        eprintln!("wire relay-server (UDS): conn error: {e}");
                    }
                }
            });
        }
    };

    let shutdown = async {
        let _ = tokio::signal::ctrl_c().await;
        eprintln!("\nshutting down — final counter snapshot");
        if let Err(e) = shutdown_relay.persist_counters().await {
            eprintln!("final counter persist failed: {e}");
        }
        let _ = std::fs::remove_file(&socket_path_for_cleanup);
    };

    tokio::select! {
        _ = serve_loop => {},
        _ = shutdown => {},
    };
    Ok(())
}

#[cfg(not(unix))]
pub async fn serve_uds(_socket_path: PathBuf, _state_dir: PathBuf) -> Result<()> {
    Err(anyhow::anyhow!(
        "UDS transport is Unix-only; Windows falls back to loopback HTTP. \
         Use `wire relay-server --bind 127.0.0.1:8771 --local-only` on Windows."
    ))
}

/// v0.5.17: relay-server mode toggles. Default = full federation
/// (current behavior). `local_only` strips phonebook + well-known
/// surfaces so the relay is invisible from off-box and from any
/// directory-scraping agent on the same box.
#[derive(Debug, Clone, Copy, Default)]
pub struct ServerMode {
    /// When true, skip phonebook listing + `.well-known/wire/agent` +
    /// `.well-known/agent-card.json` + landing/stats pages. Pair this
    /// with a loopback bind (`--local-only` enforces this at the CLI
    /// layer) for genuinely within-machine traffic.
    pub local_only: bool,
}

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

    #[test]
    fn constant_time_eq_basic() {
        assert!(constant_time_eq(b"abc", b"abc"));
        assert!(!constant_time_eq(b"abc", b"abd"));
        assert!(!constant_time_eq(b"abc", b"abcd")); // length mismatch
    }

    #[test]
    fn at_capacity_is_inclusive_at_the_ceiling() {
        // #291 H1: a new allocation is refused once the map reaches `max`.
        assert!(!at_capacity(0, 5));
        assert!(!at_capacity(4, 5));
        assert!(at_capacity(5, 5), "at the ceiling, refuse");
        assert!(at_capacity(6, 5));
    }

    #[test]
    fn intro_rate_gate_allows_up_to_max_then_429s_then_recovers() {
        // #247.3: 5 intros in the window pass, the 6th is rejected; after the
        // window elapses the old ones prune and intros flow again.
        let mut times: Vec<u64> = Vec::new();
        let t0 = 1_000_000u64;
        // 5 allowed at the same instant.
        for i in 0..5 {
            assert!(
                record_intro_within_rate(&mut times, t0, 300, 5),
                "intro {i} within limit must pass"
            );
        }
        // 6th in-window → rejected, and NOT recorded (still 5).
        assert!(!record_intro_within_rate(&mut times, t0 + 10, 300, 5));
        assert_eq!(times.len(), 5, "rejected intro must not be recorded");
        // Still rejected just inside the window.
        assert!(!record_intro_within_rate(&mut times, t0 + 299, 300, 5));
        // Past the window → all 5 prune → allowed again.
        assert!(record_intro_within_rate(&mut times, t0 + 301, 300, 5));
        assert_eq!(
            times.len(),
            1,
            "stale entries pruned, only the new one remains"
        );
    }

    #[test]
    fn valid_public_relay_url_accepts_clean_https_and_ports() {
        assert!(is_valid_public_relay_url("https://wireup.net"));
        assert!(is_valid_public_relay_url("https://wireup.net/"));
        assert!(is_valid_public_relay_url("http://127.0.0.1:8771"));
        assert!(is_valid_public_relay_url(
            "https://relay.example.com:443/path"
        ));
    }

    #[test]
    fn valid_public_relay_url_rejects_userinfo_and_bad_schemes() {
        // Audit L3 / the `handle@relay` userinfo bug.
        assert!(!is_valid_public_relay_url(
            "https://raven-kettle@wireup.net"
        ));
        assert!(!is_valid_public_relay_url("https://user:pass@host"));
        // Non-http schemes.
        assert!(!is_valid_public_relay_url("javascript:alert(1)"));
        assert!(!is_valid_public_relay_url("file:///etc/passwd"));
        assert!(!is_valid_public_relay_url("wireup.net")); // no scheme
        // Empty host / authority.
        assert!(!is_valid_public_relay_url("https://"));
        assert!(!is_valid_public_relay_url("https:///path"));
        // Whitespace / control.
        assert!(!is_valid_public_relay_url("https://wireup.net\n"));
        assert!(!is_valid_public_relay_url("https://wire up.net"));
    }

    #[test]
    fn random_hex_length() {
        let s = random_hex(16);
        assert_eq!(s.len(), 32); // 16 bytes -> 32 hex chars
        assert!(s.chars().all(|c| c.is_ascii_hexdigit()));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn sweep_intro_times_drops_aged_nicks_off_the_hot_path() {
        let dir = std::env::temp_dir().join(format!("wire-introsweep-{}", random_hex(8)));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Relay::new(dir.clone()).await.unwrap();
        let now = unix_now();
        {
            let mut inner = relay.inner.lock().await;
            inner.intro_times.insert("fresh".into(), vec![now]); // within window
            inner
                .intro_times
                .insert("stale".into(), vec![now - INTRO_WINDOW_SECS - 1]); // aged out
            assert_eq!(inner.intro_times.len(), 2);
        }
        relay.sweep_intro_times().await;
        {
            let inner = relay.inner.lock().await;
            assert!(inner.intro_times.contains_key("fresh"), "fresh nick kept");
            assert!(!inner.intro_times.contains_key("stale"), "aged nick swept");
        }
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn pair_slot_evicts_when_idle_past_ttl() {
        let dir = std::env::temp_dir().join(format!("wire-evict-{}", random_hex(8)));
        let _ = std::fs::remove_dir_all(&dir);
        let relay = Relay::new(dir.clone()).await.unwrap();

        // Seed a pair-slot manually with a past last_touched.
        {
            let mut inner = relay.inner.lock().await;
            inner
                .pair_lookup
                .insert("hash-A".to_string(), "id-A".to_string());
            inner.pair_slots.insert(
                "id-A".to_string(),
                PairSlot {
                    last_touched: std::time::Instant::now()
                        - std::time::Duration::from_secs(PAIR_SLOT_TTL_SECS + 60),
                    ..PairSlot::default()
                },
            );

            // And a fresh one — should survive.
            inner
                .pair_lookup
                .insert("hash-B".to_string(), "id-B".to_string());
            inner
                .pair_slots
                .insert("id-B".to_string(), PairSlot::default());

            assert_eq!(inner.pair_slots.len(), 2);
            assert_eq!(inner.pair_lookup.len(), 2);
        }

        relay.evict_expired_pair_slots().await;

        let inner = relay.inner.lock().await;
        assert_eq!(
            inner.pair_slots.len(),
            1,
            "expired slot should have been evicted"
        );
        assert!(inner.pair_slots.contains_key("id-B"));
        assert_eq!(inner.pair_lookup.len(), 1);
        assert!(inner.pair_lookup.contains_key("hash-B"));
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn slot_id_validator_accepts_only_lowercase_32hex() {
        assert!(is_valid_slot_id("0123456789abcdef0123456789abcdef"));
        assert!(is_valid_slot_id(&random_hex(16)));
        // wrong length
        assert!(!is_valid_slot_id("abc"));
        assert!(!is_valid_slot_id("0123456789abcdef0123456789abcde")); // 31
        assert!(!is_valid_slot_id("0123456789abcdef0123456789abcdef0")); // 33
        // uppercase
        assert!(!is_valid_slot_id("0123456789ABCDEF0123456789abcdef"));
        // path traversal attempts
        assert!(!is_valid_slot_id("../etc/passwd0123456789abcdef0000"));
        assert!(!is_valid_slot_id("..%2Fetc%2Fpasswd00000000000000000"));
        assert!(!is_valid_slot_id("/absolute/path/that/looks/like/key"));
        // null bytes
        assert!(!is_valid_slot_id(
            "0123456789abcdef\0\x31\x32\x33456789abcdef"
        ));
    }

    #[test]
    fn intro_filter_accepts_only_kind_1100_pair_drop_or_agent_card() {
        // The two legitimate intros: kind=1100 with pair_drop or agent_card.
        assert!(intro_event_allowed(1100, "pair_drop"));
        assert!(intro_event_allowed(1100, "agent_card"));

        // #334 HIGH regression guards — the pre-fix `&&`-chain wrongly accepted
        // BOTH of these:
        // (a) a kind=1100 event of ANY other type must be rejected,
        assert!(!intro_event_allowed(1100, "decision"));
        assert!(!intro_event_allowed(1100, ""));
        assert!(!intro_event_allowed(1100, "trust_revoke_key"));
        // (b) a wrong-kind event whose type happens to match must be rejected.
        assert!(!intro_event_allowed(1, "pair_drop"));
        assert!(!intro_event_allowed(1101, "agent_card"));
        assert!(!intro_event_allowed(0, "pair_drop"));

        // Neither matching → rejected (the only case the old logic got right).
        assert!(!intro_event_allowed(1, "decision"));
    }

    #[test]
    fn stream_subscriber_ceiling_is_inclusive() {
        // Audit: a new SSE subscriber is refused once a slot reaches the cap.
        assert!(!at_capacity(0, MAX_STREAMS_PER_SLOT));
        assert!(!at_capacity(MAX_STREAMS_PER_SLOT - 1, MAX_STREAMS_PER_SLOT));
        assert!(at_capacity(MAX_STREAMS_PER_SLOT, MAX_STREAMS_PER_SLOT));
        assert!(at_capacity(MAX_STREAMS_PER_SLOT + 1, MAX_STREAMS_PER_SLOT));
    }

    #[test]
    fn retain_live_subscribers_drops_disconnected() {
        // A disconnected SSE client = a dropped receiver → sender is_closed().
        let (live_tx, _live_rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let (dead_tx, dead_rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        drop(dead_rx); // client disconnected
        let mut subs = vec![live_tx, dead_tx];
        retain_live_subscribers(&mut subs);
        assert_eq!(subs.len(), 1, "the disconnected subscriber is pruned");
        assert!(!subs[0].is_closed(), "the survivor is the live sender");
    }

    #[test]
    fn evict_stale_intro_nicks_drops_only_fully_aged_entries() {
        let now = 10_000u64;
        let window = INTRO_WINDOW_SECS;
        let mut m: HashMap<String, Vec<u64>> = HashMap::new();
        m.insert("fresh".into(), vec![now - 1]); // within window → keep
        m.insert("stale".into(), vec![now - window - 1]); // all aged → drop
        m.insert("mixed".into(), vec![now - window - 5, now - 2]); // one fresh → keep
        m.insert("empty".into(), vec![]); // no timestamps → drop
        evict_stale_intro_nicks(&mut m, now, window);
        assert!(m.contains_key("fresh"));
        assert!(m.contains_key("mixed"));
        assert!(!m.contains_key("stale"));
        assert!(!m.contains_key("empty"));
    }
}