//! Peer RPC protocol on top of the QUIC transport (Phase 1e).
//!
//! Every bidi stream carries one request → one response. The first byte
//! of the request is a method tag from [`Method`]; the rest is the
//! opaque per-method payload. The response is the opaque per-method
//! reply, or a single-byte error code from [`ErrorCode`] when the
//! request was malformed.
//!
//! # Wire format
//!
//! ```text
//! request  : method:u8 | payload:bytes
//! response : reply:bytes   -- or single-byte ErrorCode
//! ```
//!
//! The QUIC transport already provides message-boundary + integrity, so
//! no length-prefixing or checksums live here — `read_to_end` on a
//! finished stream returns exactly one whole message.
//!
//! # Methods
//!
//! * [`Method::Ping`] — echoes the payload back as `"pong:" || payload`.
//!   Health check / handshake smoke test.
//! * [`Method::PeerStatus`] — returns a JSON-encoded [`PeerStatusReply`]
//!   containing this node's local view of the cluster (its own
//!   name+zone, plus every peer it currently knows about via gossip).

use crate::cluster::blob::{BlobId, BlobManifest, BlobStat, BlobStore, ChunkHash};
use crate::cluster::gossip::{ClusterGossip, PeerView};
use crate::cluster::metrics::{CacheMetrics, MetricsReply};
use crate::cluster::refs::{PutOutcome, RefKey, RefStore, RefValue, StampedRef};
use crate::cluster::tags::{TagEntry, TagStore};
use anyhow::{bail, Context, Result};
use quinn::{Connection, ConnectionError};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};

/// First byte of a streaming-method reply that indicates "success — the
/// content follows". Distinct from every [`ErrorCode`] value so the
/// client can trivially route on this single byte.
pub const STREAM_STATUS_OK: u8 = 0x00;

/// Cap on a single request or response, including the method tag.
/// 16 MiB is generous enough to hold one 4 MiB blob chunk with plenty
/// of framing headroom; multi-chunk / whole-blob transfers still fit
/// well under that ceiling for anything up to a few MB. Streaming
/// (many-GB) put/get lands in Phase 2c.
pub const MAX_MESSAGE_BYTES: usize = 16 * 1024 * 1024;

/// RPC method tag byte.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum Method {
    Ping = 0x01,
    PeerStatus = 0x02,
    /// `payload`: 32-byte `BlobId`. Reply: JSON `BlobStat` or
    /// single-byte [`ErrorCode::NotFound`].
    BlobStat = 0x03,
    /// `payload`: 32-byte `BlobId`. Reply: raw blob bytes or
    /// single-byte [`ErrorCode::NotFound`].
    BlobGet = 0x04,
    /// `payload`: raw blob bytes. Reply: 32-byte `BlobId` of the stored
    /// content.
    BlobPut = 0x05,
    /// `payload`: 32-byte `BlobId`. Reply: JSON `BlobManifest` or
    /// single-byte [`ErrorCode::NotFound`].
    BlobLoadManifest = 0x06,
    /// Streaming variant of [`Method::BlobPut`] (Phase 2c). Wire:
    /// method_byte followed by arbitrarily-many bytes of blob content
    /// until the client half-closes the send stream. Reply:
    /// `STREAM_STATUS_OK` (1 byte) followed by 32-byte `BlobId`, OR a
    /// single-byte error code. Unlike `BlobPut` this is not bounded by
    /// [`MAX_MESSAGE_BYTES`]; multi-GB blobs are the target workload.
    BlobPutStream = 0x07,
    /// Streaming variant of [`Method::BlobGet`] (Phase 2c). Wire:
    /// method_byte followed by 32-byte `BlobId`. Reply:
    /// `STREAM_STATUS_OK` (1 byte) followed by the blob content
    /// streamed until the server half-closes, OR a single-byte error
    /// code (typically [`ErrorCode::NotFound`]).
    BlobGetStream = 0x08,
    /// Phase 2d: does the peer already have a specific chunk?
    /// `payload`: 32-byte `ChunkHash`. Reply: single byte —
    /// `STREAM_STATUS_OK` (present) or [`ErrorCode::NotFound`] (absent).
    /// Non-error single-byte replies make it cheap enough to fan out
    /// N of these during a partial-sync scan.
    HasChunk = 0x09,
    /// Phase 2d: upload one chunk. `payload`: 32-byte ChunkHash ||
    /// chunk bytes. Reply: `STREAM_STATUS_OK` (1 byte) on success, or
    /// single-byte error code. The server verifies the bytes hash to
    /// the claimed hash before writing (defense against poisoning).
    PutChunk = 0x0a,
    /// Phase 2d: fetch one chunk. `payload`: 32-byte ChunkHash.
    /// Reply: `STREAM_STATUS_OK` (1 byte) || chunk bytes on success,
    /// or single-byte [`ErrorCode::NotFound`].
    GetChunk = 0x0b,
    /// Phase 2d: commit a blob manifest whose chunks are already on
    /// the peer's disk. `payload`: JSON `BlobManifest`. Reply: JSON —
    /// `{"blob_id":"...","missing":[<chunk_hash>...]}` where an empty
    /// `missing` list means the manifest was written; a non-empty
    /// list tells the client which chunks to upload before retrying.
    PutManifest = 0x0c,
    /// Phase 5b: fetch a 32-byte value keyed by a 32-byte reference key.
    /// Used by the fingerprint-keyed cargo cache: key = fingerprint,
    /// value = BlobId of the cached target-dir tarball.
    /// `payload`: 32-byte `RefKey`.
    /// Reply: 32 bytes on hit, single-byte [`ErrorCode::NotFound`] on miss.
    GetRef = 0x0d,
    /// Phase 5b: set the value for a reference key. Overwrites any
    /// prior value — the CRDT/versioning semantics come later
    /// (Phase 3). `payload`: 32-byte `RefKey` || 32-byte `RefValue`.
    /// Reply: `STREAM_STATUS_OK` (1 byte) on success, or a
    /// single-byte error code.
    PutRef = 0x0e,
    /// Phase 5d: set a named tag pointing at a 32-byte value.
    /// `payload`: `key_len:u16 (LE) || key_bytes || value:32bytes`.
    /// Reply: `STREAM_STATUS_OK` on success.
    PutTag = 0x0f,
    /// Phase 5d: fetch a named tag's value.
    /// `payload`: raw key bytes (variable length, up to 4 KiB).
    /// Reply: 32-byte value on hit, `NotFound` on miss.
    GetTag = 0x10,
    /// Phase 5d: delete a named tag.
    /// `payload`: raw key bytes.
    /// Reply: `STREAM_STATUS_OK` (deleted) or `NotFound`.
    DeleteTag = 0x11,
    /// Phase 5d: list all tags on the peer. `payload`: empty.
    /// Reply: JSON `Vec<TagEntry>` sorted by key.
    ListTags = 0x12,
    /// Phase 5g: fetch a snapshot of this peer's cache metrics.
    /// `payload`: empty. Reply: JSON [`MetricsReply`].
    GetMetrics = 0x13,
    /// Ref-forwarding (2026-07-13): local-only lookup variant of
    /// [`Method::GetRef`]. Same wire shape as `GetRef` but the peer
    /// MUST NOT recurse further; used by the daemon when it forwards
    /// a local miss to peers, preventing lookup loops.
    /// `payload`: 32-byte `RefKey`. Reply: 32 bytes on hit or
    /// single-byte [`ErrorCode::NotFound`].
    GetRefLocal = 0x14,
    /// Phase 3 (2026-07-13): stamped/versioned PutRef. Merges under a
    /// Lamport-clock + node-stamp total order — concurrent writers
    /// can no longer clobber each other silently.
    /// `payload`: 32-byte `RefKey` || 32-byte `RefValue` ||
    /// 8-byte little-endian `u64` clock || 8-byte node stamp
    /// (`blake3(node_name)[0..8]`).
    /// Reply: single-byte status —
    ///   `STREAM_STATUS_OK` = accepted (Merged);
    ///   [`ErrorCode::AlreadyExists`] = rejected (older/equal).
    PutRefVersioned = 0x15,
    /// Phase 3: stamped GetRef. Returns the current
    /// (value, clock, node) triple as 48 bytes.
    /// `payload`: 32-byte `RefKey`.
    /// Reply: 48 bytes on hit, single-byte
    /// [`ErrorCode::NotFound`] on miss.
    GetRefVersioned = 0x16,
    /// Phase 3b (2026-07-13): strict local-only stamped-ref lookup.
    /// Same wire shape as [`Method::GetRefVersioned`] but the peer
    /// MUST NOT forward on miss. Used by daemons doing ref-forwarding
    /// so they never loop.
    GetRefVersionedLocal = 0x17,
    /// Phase 3c (2026-07-13): stamped/versioned PutTag. Merges under
    /// the same `(clock, node)` total order as
    /// [`Method::PutRefVersioned`].
    /// `payload`: `key_len:u16 (LE) || key_bytes || stamped_value:48`.
    /// Reply: single-byte status —
    ///   `STREAM_STATUS_OK` = accepted (Merged);
    ///   [`ErrorCode::AlreadyExists`] = rejected (older/equal).
    PutTagVersioned = 0x18,
    /// Phase 3c: fetch a stamped tag value.
    /// `payload`: raw tag key bytes (variable length, up to 4 KiB).
    /// Reply: 48 bytes on hit, single-byte
    /// [`ErrorCode::NotFound`] on miss.
    GetTagVersioned = 0x19,
    /// Phase 4b follow-on (2026-07-13): attach a TTL to a stamped
    /// tag. Sidecar semantics live in [`TagStore::set_stamped_expiry`]
    /// — `expires_at_unix == 0` clears any prior sidecar; otherwise
    /// the value is absolute wall-clock seconds.
    /// `payload`: `key_len:u16 (LE) || key_bytes || expires_at:u64 (LE)`.
    /// Reply: single-byte `STREAM_STATUS_OK`.
    SetTagExpiry = 0x1a,
    /// Phase 4b follow-on: read the TTL sidecar for a stamped tag.
    /// `payload`: raw tag key bytes.
    /// Reply: 8 bytes (`u64` LE) on hit, single-byte
    /// [`ErrorCode::NotFound`] when no sidecar is present.
    GetTagExpiry = 0x1b,
    /// Dashboard-v2 (2026-07-14): return a JSON snapshot of this
    /// node's counts (blobs / tags / refs / snapshots / ref-tracking
    /// + on-disk bytes). Feeds the fleet aggregator dashboard so an
    /// operator gets a single-pane view of the fleet from any client
    /// that can reach cluster RPC — no HTTP server on the daemon
    /// nodes required.
    ///
    /// `payload`: empty.
    /// Reply: JSON `DashboardStatusReply`.
    DashboardStatus = 0x1c,
    /// Dashboard-v2: storage-inventory payload (tags, snapshots,
    /// ref-tracking, plus sampled blob + ref lists). Fed to the
    /// aggregator's storage-browser tabs.
    ///
    /// `payload`: empty.
    /// Reply: JSON `DashboardStorageReply`.
    DashboardStorage = 0x1d,
    /// Phase 9 R1a (2026-07-15): shallow-clone a `(url, git_ref)` on
    /// this peer under a caller-provided workspace namespace. The
    /// aggregator fans this out to every fleet node; this per-peer
    /// method is deliberately narrow — one clone, one path.
    ///
    /// `payload`: JSON [`crate::cluster::repo_ensure::RepoEnsureRequest`].
    /// Reply: JSON [`crate::cluster::repo_ensure::RepoEnsureReply`].
    /// Requires [`RpcRouter::with_repo_root`]; else
    /// [`ErrorCode::NotConfigured`].
    RepoEnsure = 0x1e,
    /// Phase 9 R1a: inverse of [`Method::RepoEnsure`]. Removes the
    /// on-disk checkout for `(url, git_ref)` under the caller's
    /// workspace. Idempotent: absent checkout ⇒ `removed=false`.
    ///
    /// `payload`: JSON [`crate::cluster::repo_ensure::RepoReleaseRequest`].
    /// Reply: JSON [`crate::cluster::repo_ensure::RepoReleaseReply`].
    RepoRelease = 0x1f,
    /// Runs `safe-shutdown-prep.sh --dry-run` to completion on this
    /// node and returns the full report. Never stops anything —
    /// dry-run only, safe to call repeatedly.
    ///
    /// `payload`: JSON [`crate::cluster::shutdown_prep::ShutdownPrepCheckRequest`].
    /// Reply: JSON [`crate::cluster::shutdown_prep::ShutdownPrepCheckReply`].
    ShutdownPrepCheck = 0x20,
    /// Starts the real `safe-shutdown-prep.sh` run in a detached
    /// systemd scope and returns immediately — the script's own step
    /// stops `claw-store.service`, so this RPC connection cannot
    /// outlive full completion. See
    /// [`crate::cluster::shutdown_prep`] module docs.
    ///
    /// `payload`: JSON [`crate::cluster::shutdown_prep::ShutdownPrepExecuteRequest`].
    /// Reply: JSON [`crate::cluster::shutdown_prep::ShutdownPrepExecuteReply`].
    ShutdownPrepExecute = 0x21,
}

impl Method {
    /// Parse a byte back into a method. Unknown bytes → `None`, which
    /// the server surfaces to the caller as [`ErrorCode::UnknownMethod`].
    pub fn from_byte(b: u8) -> Option<Self> {
        match b {
            0x01 => Some(Method::Ping),
            0x02 => Some(Method::PeerStatus),
            0x03 => Some(Method::BlobStat),
            0x04 => Some(Method::BlobGet),
            0x05 => Some(Method::BlobPut),
            0x06 => Some(Method::BlobLoadManifest),
            0x07 => Some(Method::BlobPutStream),
            0x08 => Some(Method::BlobGetStream),
            0x09 => Some(Method::HasChunk),
            0x0a => Some(Method::PutChunk),
            0x0b => Some(Method::GetChunk),
            0x0c => Some(Method::PutManifest),
            0x0d => Some(Method::GetRef),
            0x0e => Some(Method::PutRef),
            0x0f => Some(Method::PutTag),
            0x10 => Some(Method::GetTag),
            0x11 => Some(Method::DeleteTag),
            0x12 => Some(Method::ListTags),
            0x13 => Some(Method::GetMetrics),
            0x14 => Some(Method::GetRefLocal),
            0x15 => Some(Method::PutRefVersioned),
            0x16 => Some(Method::GetRefVersioned),
            0x17 => Some(Method::GetRefVersionedLocal),
            0x18 => Some(Method::PutTagVersioned),
            0x19 => Some(Method::GetTagVersioned),
            0x1a => Some(Method::SetTagExpiry),
            0x1b => Some(Method::GetTagExpiry),
            0x1c => Some(Method::DashboardStatus),
            0x1d => Some(Method::DashboardStorage),
            0x1e => Some(Method::RepoEnsure),
            0x1f => Some(Method::RepoRelease),
            0x20 => Some(Method::ShutdownPrepCheck),
            0x21 => Some(Method::ShutdownPrepExecute),
            _ => None,
        }
    }

    /// Whether this method uses the streaming wire format (status byte
    /// + arbitrary-length content on the reply). Non-streaming methods
    /// use the bounded `payload | reply` shape with `read_to_end`.
    pub fn is_streaming(self) -> bool {
        matches!(self, Method::BlobPutStream | Method::BlobGetStream)
    }

    /// Byte tag as an owned u8. `as u8` also works; this exists for symmetry.
    pub fn as_byte(self) -> u8 {
        self as u8
    }
}

/// Well-known single-byte error responses the server may return in
/// place of a normal reply. The client distinguishes these by length =
/// 1 AND the byte being a known error code.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum ErrorCode {
    EmptyRequest = 0xf0,
    UnknownMethod = 0xf1,
    HandlerFailure = 0xf2,
    /// Blob (or manifest) not present in the local store.
    NotFound = 0xf3,
    /// Request payload was structurally wrong (e.g. wrong length for a
    /// 32-byte hash).
    InvalidRequest = 0xf4,
    /// Server-side subsystem required for this method wasn't
    /// configured (e.g. Blob RPCs called on a node with no local
    /// blob store).
    NotConfigured = 0xf5,
    /// Phase 3 (2026-07-13): the write was rejected because a prior
    /// value with an equal or higher `(clock, node)` already exists.
    /// The current value stayed on disk. Used by
    /// [`Method::PutRefVersioned`] to signal a "not merged" outcome.
    AlreadyExists = 0xf6,
}

impl ErrorCode {
    pub fn as_byte(self) -> u8 {
        self as u8
    }
    pub fn describe(self) -> &'static str {
        match self {
            ErrorCode::EmptyRequest => "empty request",
            ErrorCode::UnknownMethod => "unknown method",
            ErrorCode::HandlerFailure => "handler failure",
            ErrorCode::NotFound => "not found",
            ErrorCode::InvalidRequest => "invalid request",
            ErrorCode::NotConfigured => "server subsystem not configured",
            ErrorCode::AlreadyExists => {
                "rejected: existing value dominates incoming write"
            }
        }
    }
}

/// Reply payload for [`Method::PeerStatus`]. Serialised as JSON on the
/// wire — small (<< 16 KB for a 10-node fleet) and easy to inspect
/// from a shell (`jq` etc.).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct PeerStatusReply {
    /// The name this node advertises for itself.
    pub local_name: String,
    /// The zone this node is in.
    pub local_zone: String,
    /// Field finding 2026-07-12: this node's own rustc release
    /// (e.g. `1.97.0`). Absent when rustc isn't on the daemon's PATH.
    /// Runners consult this before a cold build to detect toolchain
    /// drift that would silo the produced cache.
    #[serde(default)]
    pub local_rustc_release: Option<String>,
    /// Every peer this node knows about (live + dead-in-grace-window).
    pub peers: Vec<PeerView>,
}

/// Dashboard-v2 aggregator payload. One per node. Wire: JSON.
///
/// Read by `serve_v2::AggregatorState` from any client that can
/// reach cluster RPC (typically the operator's laptop running
/// `claw-store serve --aggregator`).
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct DashboardStatusReply {
    pub node_name: String,
    pub zone: String,
    pub blob_store_root: Option<String>,
    pub blob_count: usize,
    pub tag_count: usize,
    pub ref_count: usize,
    pub snapshot_count: usize,
    pub ref_tracking_count: usize,
    pub blob_store_bytes: u64,
    /// `local_rustc_release` from PeerStatus, echoed here so the
    /// aggregator doesn't need a second round-trip.
    #[serde(default)]
    pub rustc_release: Option<String>,
    /// Human-oriented fields (added for FleetHealth landing).
    #[serde(default)]
    pub filesystem: Option<FilesystemUsage>,
    #[serde(default)]
    pub hot: Option<HotTierUsage>,
    #[serde(default)]
    pub mount: Option<MountStatus>,
    #[serde(default)]
    pub cache: Option<CacheSummary>,
    #[serde(default)]
    pub timers: Vec<TimerStatus>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct FilesystemUsage {
    pub mount_point: String,
    pub total_bytes: u64,
    pub available_bytes: u64,
    pub used_bytes: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct HotTierUsage {
    pub used_bytes: u64,
    pub max_bytes: u64,
    /// Bytes referenced by any tag or snapshot pin. `None` when
    /// we can't cheaply compute it (blob store missing).
    #[serde(default)]
    pub pinned_bytes: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct MountStatus {
    /// Configured mount point, whether or not it's currently mounted.
    pub path: String,
    pub active: bool,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct CacheSummary {
    pub hits: u64,
    pub misses: u64,
    pub bytes_served: u64,
    pub bytes_ingested: u64,
    pub hit_rate: f64,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct TimerStatus {
    pub unit: String,
    #[serde(default)]
    pub next_fire_unix: Option<u64>,
    #[serde(default)]
    pub last_result: Option<String>,
}

/// Reply payload for [`Method::PutManifest`]. When `missing` is empty
/// the manifest was persisted successfully; otherwise the client must
/// upload the listed chunks (typically via [`Method::PutChunk`]) and
/// retry the same manifest.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct PutManifestReply {
    pub blob_id: crate::cluster::blob::BlobId,
    pub missing: Vec<crate::cluster::blob::ChunkHash>,
}

/// The concrete RPC handler used by the daemon. Holds Arc references
/// to the state a request might need to read: the gossip service
/// (always), and optionally a local blob store (for Blob* methods).
///
/// Not `Clone` on its own — wrap in `Arc<RpcRouter>` so a single
/// instance backs the accept loop plus any explicit dispatch calls.
/// Dashboard-v2 storage-inventory payload. Small-and-bounded lists
/// (tags, snapshots, ref-tracking) come back in full. Large lists
/// (blobs, refs) come back as bounded samples — pagination lives
/// in a follow-on RPC when it's actually needed.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DashboardStorageReply {
    pub node_name: String,
    pub tags: Vec<DashboardTag>,
    pub snapshots: Vec<DashboardSnapshot>,
    pub ref_tracking: Vec<DashboardRefTracking>,
    /// Up to 200 blob summaries (id + size + chunk count). Ordered
    /// by blob-id hex ascending.
    pub blobs_sample: Vec<DashboardBlob>,
    pub blobs_sample_capped_at: usize,
    /// Up to 200 (fingerprint, blob-id) pairs. Ordered by fp hex.
    pub refs_sample: Vec<DashboardRef>,
    pub refs_sample_capped_at: usize,
    /// Per-repo rollup for the FleetHealth "Projects" widget.
    /// Sorted by last_seen_unix descending — hottest first.
    #[serde(default)]
    pub projects: Vec<DashboardProject>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DashboardTag {
    pub key: String,
    pub value_hex: String,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DashboardSnapshot {
    pub name: String,
    pub created_at_unix: u64,
    pub blob_count: usize,
    pub file_bytes: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DashboardRefTracking {
    pub fingerprint_hex: String,
    pub repo: String,
    pub refs: Vec<String>,
    pub first_seen_unix: u64,
    pub last_seen_unix: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DashboardBlob {
    pub blob_id_hex: String,
    pub size_bytes: u64,
    pub chunk_count: usize,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DashboardRef {
    pub fingerprint_hex: String,
    pub blob_id_hex: String,
}

/// Per-repo project rollup — the "which projects live here" view.
/// Derived from ref-tracking cross-referenced with ref-store +
/// blob-store: for each recorded (repo, git-ref, fp) triple we
/// know the blob it produced and thus its bytes.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DashboardProject {
    pub repo: String,
    /// Sum of blob sizes for every fingerprint this repo has
    /// produced on this node. 0 when we can't resolve any
    /// fp → blob mapping (e.g. eviction between record + view).
    pub cache_bytes: u64,
    pub fingerprint_count: usize,
    /// Distinct git-refs observed across all fingerprints.
    pub refs: Vec<String>,
    pub first_seen_unix: u64,
    pub last_seen_unix: u64,
    /// "active" (< 24h), "recent" (< 7d), "idle" (older).
    pub tier: String,
}

/// Group ref-tracking entries by repo. For each repo, sum the
/// bytes of every blob the recorded fingerprints resolved to
/// (via the stamped ref store). Tier is a wall-clock function
/// of `last_seen_unix` — "active" < 24h, "recent" < 7d, else
/// "idle".
async fn build_projects(
    blob_store: Option<&crate::cluster::blob::BlobStore>,
    ref_store: Option<&crate::cluster::refs::RefStore>,
    _root: &Option<std::path::PathBuf>,
    ref_tracking: &[DashboardRefTracking],
) -> Vec<DashboardProject> {
    use std::collections::HashMap;
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);

    // repo → aggregation state
    struct Agg {
        cache_bytes: u64,
        fingerprint_count: usize,
        refs: std::collections::HashSet<String>,
        first_seen_unix: u64,
        last_seen_unix: u64,
    }
    let mut by_repo: HashMap<String, Agg> = HashMap::new();

    for entry in ref_tracking {
        let fp_hex = &entry.fingerprint_hex;
        // Resolve fp → blob-id via ref store (try stamped first,
        // fall back to legacy).
        let bytes = if let (Some(rs), Some(bs)) = (ref_store, blob_store) {
            match decode_fp(fp_hex) {
                Some(key) => {
                    let val = match rs.get_stamped(&key).await {
                        Ok(Some(s)) => Some(s.value),
                        _ => rs.get(&key).await.ok().flatten(),
                    };
                    match val {
                        Some(v) => {
                            let blob_id = crate::cluster::blob::BlobId::from_bytes(v);
                            bs.load_manifest(&blob_id)
                                .await
                                .ok()
                                .flatten()
                                .map(|m| m.total_size)
                                .unwrap_or(0)
                        }
                        None => 0,
                    }
                }
                None => 0,
            }
        } else {
            0
        };

        let agg = by_repo.entry(entry.repo.clone()).or_insert(Agg {
            cache_bytes: 0,
            fingerprint_count: 0,
            refs: std::collections::HashSet::new(),
            first_seen_unix: u64::MAX,
            last_seen_unix: 0,
        });
        agg.cache_bytes = agg.cache_bytes.saturating_add(bytes);
        agg.fingerprint_count += 1;
        for r in &entry.refs {
            agg.refs.insert(r.clone());
        }
        agg.first_seen_unix = agg.first_seen_unix.min(entry.first_seen_unix);
        agg.last_seen_unix = agg.last_seen_unix.max(entry.last_seen_unix);
    }

    let mut out: Vec<DashboardProject> = by_repo
        .into_iter()
        .map(|(repo, a)| {
            let age = now.saturating_sub(a.last_seen_unix);
            let tier = if age < 86_400 {
                "active"
            } else if age < 7 * 86_400 {
                "recent"
            } else {
                "idle"
            };
            let mut refs: Vec<String> = a.refs.into_iter().collect();
            refs.sort();
            DashboardProject {
                repo,
                cache_bytes: a.cache_bytes,
                fingerprint_count: a.fingerprint_count,
                refs,
                first_seen_unix: if a.first_seen_unix == u64::MAX {
                    0
                } else {
                    a.first_seen_unix
                },
                last_seen_unix: a.last_seen_unix,
                tier: tier.to_string(),
            }
        })
        .collect();
    // Hottest first.
    out.sort_by(|a, b| b.last_seen_unix.cmp(&a.last_seen_unix));
    out
}

fn decode_fp(hex: &str) -> Option<[u8; 32]> {
    if hex.len() != 64 {
        return None;
    }
    let bytes = hex.as_bytes();
    let mut out = [0u8; 32];
    for i in 0..32 {
        let hi = decode_nibble(bytes[i * 2])?;
        let lo = decode_nibble(bytes[i * 2 + 1])?;
        out[i] = (hi << 4) | lo;
    }
    Some(out)
}

fn decode_nibble(b: u8) -> Option<u8> {
    match b {
        b'0'..=b'9' => Some(b - b'0'),
        b'a'..=b'f' => Some(b - b'a' + 10),
        b'A'..=b'F' => Some(b - b'A' + 10),
        _ => None,
    }
}

/// statvfs on the given path. Uses libc directly — cheap enough
/// that we don't need to cache. Silent on error (returns None).
fn filesystem_usage(path: &std::path::Path) -> Option<FilesystemUsage> {
    let cpath = std::ffi::CString::new(path.as_os_str().to_str()?).ok()?;
    // SAFETY: statvfs writes to a zero-initialised struct; we
    // read only when it returns 0.
    let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
    let rc = unsafe { libc::statvfs(cpath.as_ptr(), &mut stat) };
    if rc != 0 {
        return None;
    }
    let bsize = stat.f_frsize as u64;
    let total = stat.f_blocks as u64 * bsize;
    let avail = stat.f_bavail as u64 * bsize;
    let used = total.saturating_sub(avail);
    Some(FilesystemUsage {
        mount_point: path.display().to_string(),
        total_bytes: total,
        available_bytes: avail,
        used_bytes: used,
    })
}

/// Query systemd for a user-scope timer's next-fire + last result.
/// Shells out to systemctl. Silent on any failure — dashboards
/// should degrade to "unknown" rather than 500.
fn timer_status(unit: &str) -> TimerStatus {
    // NextElapseUSecRealtime returns micros since epoch, or 0.
    // The Service unit (same name minus .timer) holds the last result.
    let next = std::process::Command::new("systemctl")
        .args(["--user", "show", unit, "--no-pager", "-p", "NextElapseUSecRealtime"])
        .output()
        .ok()
        .and_then(|o| String::from_utf8(o.stdout).ok())
        .and_then(|s| {
            s.trim()
                .strip_prefix("NextElapseUSecRealtime=")
                .and_then(|v| v.parse::<u64>().ok())
        })
        .filter(|&v| v > 0)
        .map(|us| us / 1_000_000);
    let service = unit.trim_end_matches(".timer").to_string() + ".service";
    let last = std::process::Command::new("systemctl")
        .args(["--user", "show", &service, "--no-pager", "-p", "Result"])
        .output()
        .ok()
        .and_then(|o| String::from_utf8(o.stdout).ok())
        .and_then(|s| {
            s.trim()
                .strip_prefix("Result=")
                .map(|v| v.to_string())
                .filter(|v| !v.is_empty())
        });
    TimerStatus {
        unit: unit.to_string(),
        next_fire_unix: next,
        last_result: last,
    }
}

/// Is `path` currently a mount point? Cheap Linux check: read
/// /proc/mounts. On macOS returns None (aggregator doesn't run
/// mounts).
fn is_mounted(path: &std::path::Path) -> bool {
    let want = path.display().to_string();
    let mounts = match std::fs::read_to_string("/proc/mounts") {
        Ok(s) => s,
        Err(_) => return false,
    };
    for line in mounts.lines() {
        // fields: <src> <mountpoint> <fstype> ...
        let mut it = line.split_whitespace();
        it.next();
        if let Some(mp) = it.next() {
            if mp == want {
                return true;
            }
        }
    }
    false
}

/// Recursive byte count under a directory. Used by the dashboard
/// handler for reporting only; silent on read errors.
fn dir_size_bytes(root: &std::path::Path) -> u64 {
    let mut total: u64 = 0;
    let mut stack = vec![root.to_path_buf()];
    while let Some(dir) = stack.pop() {
        let entries = match std::fs::read_dir(&dir) {
            Ok(e) => e,
            Err(_) => continue,
        };
        for entry in entries.flatten() {
            let ft = match entry.file_type() {
                Ok(t) => t,
                Err(_) => continue,
            };
            if ft.is_dir() {
                stack.push(entry.path());
            } else if ft.is_file() {
                if let Ok(m) = entry.metadata() {
                    total = total.saturating_add(m.len());
                }
            }
        }
    }
    total
}

pub struct RpcRouter {
    gossip: Arc<ClusterGossip>,
    blob_store: Option<Arc<BlobStore>>,
    ref_store: Option<Arc<RefStore>>,
    tag_store: Option<Arc<TagStore>>,
    metrics: Arc<CacheMetrics>,
    local_name: String,
    local_zone: String,
    /// Ref-forwarding (2026-07-13): when set, `GetRef` misses fan out
    /// to alive peers via this client. On the first peer that has the
    /// ref, the daemon transparently pulls the blob into its local
    /// store, `PutRef`s the mapping, and returns the value — so the
    /// caller sees a plain HIT and subsequent lookups are local.
    ///
    /// `None` disables forwarding entirely; `GetRef` behaves like
    /// `GetRefLocal` (strict local-only). Set at daemon startup by
    /// `ClusterServices` when a NodeIdentity is available.
    outbound_client: Option<Arc<crate::cluster::transport::QuicClient>>,
    /// Phase 9 R1a: root directory under which `RepoEnsure` materializes
    /// checkouts. `None` disables both `RepoEnsure` and `RepoRelease`
    /// (server returns `NotConfigured`).
    repo_root: Option<std::path::PathBuf>,
}

/// Result of dispatching a request: either a real reply (`Ok`) or a
/// well-known error code the wire layer surfaces as a single byte.
/// Extracted so `handle` stays free of `Vec<u8>` shell games.
enum HandlerOutcome {
    Reply(Vec<u8>),
    Error(ErrorCode),
}

impl RpcRouter {
    pub fn new(gossip: Arc<ClusterGossip>, local_name: String, local_zone: String) -> Self {
        Self {
            gossip,
            blob_store: None,
            ref_store: None,
            tag_store: None,
            metrics: Arc::new(CacheMetrics::new()),
            local_name,
            local_zone,
            outbound_client: None,
            repo_root: None,
        }
    }

    /// Phase 9 R1a: enable `RepoEnsure` / `RepoRelease` by attaching a
    /// root directory. The directory is created on first use.
    pub fn with_repo_root(mut self, root: std::path::PathBuf) -> Self {
        self.repo_root = Some(root);
        self
    }

    /// Enable ref-forwarding on `GetRef` misses by installing the
    /// outbound QUIC client the router will use to dial peers. See
    /// the field's rustdoc for the semantics.
    pub fn with_outbound_client(
        mut self,
        client: Arc<crate::cluster::transport::QuicClient>,
    ) -> Self {
        self.outbound_client = Some(client);
        self
    }

    /// Read-only handle to the router's metrics. Used by
    /// `ClusterServices` (or tests) to sample counts without going
    /// through the RPC layer.
    pub fn metrics(&self) -> &Arc<CacheMetrics> {
        &self.metrics
    }

    /// Attach a local blob store. Enables the `Blob*` methods; nodes
    /// without a store return [`ErrorCode::NotConfigured`] for those.
    pub fn with_blob_store(mut self, store: Arc<BlobStore>) -> Self {
        self.blob_store = Some(store);
        self
    }

    /// Attach a local ref store. Enables the `GetRef` / `PutRef`
    /// methods.
    pub fn with_ref_store(mut self, store: Arc<RefStore>) -> Self {
        self.ref_store = Some(store);
        self
    }

    /// Attach a local tag store (Phase 5d). Enables the
    /// `PutTag` / `GetTag` / `DeleteTag` / `ListTags` methods.
    pub fn with_tag_store(mut self, store: Arc<TagStore>) -> Self {
        self.tag_store = Some(store);
        self
    }

    pub fn blob_store(&self) -> Option<&Arc<BlobStore>> {
        self.blob_store.as_ref()
    }

    pub fn ref_store(&self) -> Option<&Arc<RefStore>> {
        self.ref_store.as_ref()
    }

    pub fn tag_store(&self) -> Option<&Arc<TagStore>> {
        self.tag_store.as_ref()
    }

    /// Dispatch a single request. Called by [`serve_connection`] for
    /// every accepted bidi stream. Test code may call it directly to
    /// bypass the transport.
    pub async fn handle(&self, method: Method, payload: &[u8]) -> Result<Vec<u8>> {
        match self.handle_outcome(method, payload).await? {
            HandlerOutcome::Reply(bytes) => Ok(bytes),
            HandlerOutcome::Error(code) => Ok(vec![code.as_byte()]),
        }
    }

    async fn handle_outcome(&self, method: Method, payload: &[u8]) -> Result<HandlerOutcome> {
        match method {
            Method::Ping => {
                let mut reply = Vec::with_capacity(5 + payload.len());
                reply.extend_from_slice(b"pong:");
                reply.extend_from_slice(payload);
                Ok(HandlerOutcome::Reply(reply))
            }
            Method::PeerStatus => {
                let peers = self.gossip.peers().await;
                // Field finding 2026-07-12: expose our own rustc in
                // the reply so `claw-cargo build` can warn on drift
                // before wasting a full cold build.
                let local_rustc_release = self
                    .gossip
                    .self_kv(crate::cluster::gossip::keys::RUSTC_RELEASE)
                    .await;
                let reply = PeerStatusReply {
                    local_name: self.local_name.clone(),
                    local_zone: self.local_zone.clone(),
                    local_rustc_release,
                    peers,
                };
                let json = serde_json::to_vec(&reply)
                    .context("encoding PeerStatusReply as JSON")?;
                if json.len() > MAX_MESSAGE_BYTES {
                    bail!(
                        "PeerStatusReply JSON {} bytes exceeds cap {}",
                        json.len(),
                        MAX_MESSAGE_BYTES
                    );
                }
                Ok(HandlerOutcome::Reply(json))
            }
            Method::DashboardStatus => {
                // Aggregator payload for dashboard-v2. Reuses gossip
                // for name+zone+rustc; walks the on-disk stores for
                // count + byte totals. Cheap: 5 filesystem walks.
                let (blob_root, blob_count) = match &self.blob_store {
                    Some(s) => {
                        let ids = s.list_blob_ids().await.unwrap_or_default();
                        (
                            Some(s.root().display().to_string()),
                            ids.len(),
                        )
                    }
                    None => (None, 0),
                };
                let root_path = blob_root.as_ref().map(std::path::PathBuf::from);
                let tag_count = match &self.tag_store {
                    Some(t) => t.list().await.map(|v| v.len()).unwrap_or(0),
                    None => 0,
                };
                let ref_count = match &self.ref_store {
                    Some(r) => r.list().await.map(|v| v.len()).unwrap_or(0),
                    None => 0,
                };
                // Snapshot + ref-tracking aren't held on the router;
                // open by path derived from blob_store_root when
                // possible.
                let snapshot_count = match &root_path {
                    Some(p) => match crate::cluster::snapshot::SnapshotStore::open(p.clone()) {
                        Ok(s) => s.list().await.map(|v| v.len()).unwrap_or(0),
                        Err(_) => 0,
                    },
                    None => 0,
                };
                let ref_tracking_count = match &root_path {
                    Some(p) => match crate::cluster::ref_tracking::RefTracking::open(p.clone()) {
                        Ok(r) => r.list_all().await.map(|v| v.len()).unwrap_or(0),
                        Err(_) => 0,
                    },
                    None => 0,
                };
                let blob_store_bytes = match &root_path {
                    Some(p) => dir_size_bytes(p),
                    None => 0,
                };
                let rustc_release = self
                    .gossip
                    .self_kv(crate::cluster::gossip::keys::RUSTC_RELEASE)
                    .await;
                // Filesystem stat on the blob_store_root's disk.
                let filesystem = root_path.as_deref().and_then(filesystem_usage);
                // Hot tier: derive from gossip so we're not
                // duplicating disk walks.
                let hot_used = self
                    .gossip
                    .self_kv(crate::cluster::gossip::keys::HOT_USED_BYTES)
                    .await
                    .and_then(|s| s.parse::<u64>().ok())
                    .unwrap_or(0);
                let hot_max = self
                    .gossip
                    .self_kv(crate::cluster::gossip::keys::HOT_MAX_BYTES)
                    .await
                    .and_then(|s| s.parse::<u64>().ok())
                    .unwrap_or(0);
                let hot = Some(HotTierUsage {
                    used_bytes: hot_used,
                    max_bytes: hot_max,
                    pinned_bytes: None, // Phase later — needs
                    // cross-reference of pin set with blob sizes.
                });
                // Mount status: probe the conventional path. In
                // the current fleet FUSE mounts at ~/clawstor-mount
                // on Linux; we don't have a config field for this
                // yet so hardcode the convention.
                let mount = {
                    let home = std::env::var_os("HOME");
                    let path = home
                        .map(std::path::PathBuf::from)
                        .map(|h| h.join("clawstor-mount"))
                        .unwrap_or_else(|| std::path::PathBuf::from("/clawstor-mount"));
                    Some(MountStatus {
                        path: path.display().to_string(),
                        active: is_mounted(&path),
                    })
                };
                // Cache metrics — the router already tracks these
                // in-memory. Compute hit-rate here so the
                // dashboard doesn't need to divide.
                let cache = {
                    let snap = self.metrics.snapshot();
                    // The dashboard cares about "did the peer find
                    // what someone asked for". Sum the get_ref /
                    // get_tag / get_chunk counters — they're what
                    // a claw-cargo build actually queries.
                    let hits = snap.get_ref_hits
                        + snap.get_tag_hits
                        + snap.get_chunk_hits;
                    let misses = snap.get_ref_misses
                        + snap.get_tag_misses
                        + snap.get_chunk_misses;
                    let total = hits.saturating_add(misses);
                    let rate = if total > 0 {
                        hits as f64 / total as f64
                    } else {
                        0.0
                    };
                    Some(CacheSummary {
                        hits,
                        misses,
                        bytes_served: snap.blob_get_bytes,
                        bytes_ingested: snap.blob_put_bytes,
                        hit_rate: rate,
                    })
                };
                // Well-known timer set. Missing timers just get
                // next_fire_unix=None / last_result=None.
                let timers = vec![
                    timer_status("clawstor-scrub.timer"),
                    timer_status("clawstor-gc.timer"),
                    timer_status("clawstor-ref-sweep.timer"),
                    timer_status("clawstor-snapshot-rotate.timer"),
                ];
                let reply = DashboardStatusReply {
                    node_name: self.local_name.clone(),
                    zone: self.local_zone.clone(),
                    blob_store_root: blob_root,
                    blob_count,
                    tag_count,
                    ref_count,
                    snapshot_count,
                    ref_tracking_count,
                    blob_store_bytes,
                    rustc_release,
                    filesystem,
                    hot,
                    mount,
                    cache,
                    timers,
                };
                let json = serde_json::to_vec(&reply)
                    .context("encoding DashboardStatusReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::DashboardStorage => {
                const SAMPLE_CAP: usize = 200;
                let blob_root = self.blob_store.as_ref().map(|s| s.root().to_path_buf());
                let mut tags = Vec::new();
                if let Some(ts) = &self.tag_store {
                    if let Ok(list) = ts.list().await {
                        for e in list {
                            tags.push(DashboardTag {
                                key: e.key,
                                value_hex: e.value_hex,
                            });
                        }
                    }
                }
                let mut snapshots = Vec::new();
                if let Some(p) = &blob_root {
                    if let Ok(store) = crate::cluster::snapshot::SnapshotStore::open(p.clone()) {
                        if let Ok(list) = store.list().await {
                            for s in list {
                                snapshots.push(DashboardSnapshot {
                                    name: s.name,
                                    created_at_unix: s.created_at_unix,
                                    blob_count: s.blob_count,
                                    file_bytes: s.file_bytes,
                                });
                            }
                        }
                    }
                }
                let mut ref_tracking = Vec::new();
                if let Some(p) = &blob_root {
                    if let Ok(store) = crate::cluster::ref_tracking::RefTracking::open(p.clone()) {
                        if let Ok(list) = store.list_all().await {
                            for e in list {
                                let mut hex = String::with_capacity(64);
                                for b in &e.fingerprint {
                                    hex.push_str(&format!("{b:02x}"));
                                }
                                ref_tracking.push(DashboardRefTracking {
                                    fingerprint_hex: hex,
                                    repo: e.repo,
                                    refs: e.refs,
                                    first_seen_unix: e.first_seen_unix,
                                    last_seen_unix: e.last_seen_unix,
                                });
                            }
                        }
                    }
                }
                let mut blobs_sample = Vec::new();
                if let Some(store) = &self.blob_store {
                    if let Ok(mut ids) = store.list_blob_ids().await {
                        ids.sort();
                        for id in ids.into_iter().take(SAMPLE_CAP) {
                            let m = store.load_manifest(&id).await.ok().flatten();
                            blobs_sample.push(DashboardBlob {
                                blob_id_hex: id.to_hex(),
                                size_bytes: m.as_ref().map(|m| m.total_size).unwrap_or(0),
                                chunk_count: m.map(|m| m.chunks.len()).unwrap_or(0),
                            });
                        }
                    }
                }
                let mut refs_sample = Vec::new();
                if let Some(store) = &self.ref_store {
                    if let Ok(list) = store.list().await {
                        for (k, v) in list.into_iter().take(SAMPLE_CAP) {
                            let mut kh = String::with_capacity(64);
                            for b in &k {
                                kh.push_str(&format!("{b:02x}"));
                            }
                            refs_sample.push(DashboardRef {
                                fingerprint_hex: kh,
                                blob_id_hex: crate::cluster::blob::BlobId::from_bytes(v).to_hex(),
                            });
                        }
                    }
                }
                // Per-repo rollup: group ref-tracking by repo, sum
                // blob sizes for each fp via ref-store lookup.
                let projects = build_projects(
                    self.blob_store.as_deref(),
                    self.ref_store.as_deref(),
                    &blob_root,
                    &ref_tracking,
                )
                .await;

                let reply = DashboardStorageReply {
                    node_name: self.local_name.clone(),
                    tags,
                    snapshots,
                    ref_tracking,
                    blobs_sample_capped_at: SAMPLE_CAP,
                    blobs_sample,
                    refs_sample_capped_at: SAMPLE_CAP,
                    refs_sample,
                    projects,
                };
                let json = serde_json::to_vec(&reply)
                    .context("encoding DashboardStorageReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::RepoEnsure => {
                let root = match &self.repo_root {
                    Some(r) => r.clone(),
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let req: crate::cluster::repo_ensure::RepoEnsureRequest =
                    match serde_json::from_slice(payload) {
                        Ok(r) => r,
                        Err(_) => {
                            return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
                        }
                    };
                let _ = tokio::fs::create_dir_all(&root).await;
                let reply =
                    crate::cluster::repo_ensure::ensure_repo(&root, &req).await?;
                let json = serde_json::to_vec(&reply)
                    .context("encoding RepoEnsureReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::RepoRelease => {
                let root = match &self.repo_root {
                    Some(r) => r.clone(),
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let req: crate::cluster::repo_ensure::RepoReleaseRequest =
                    match serde_json::from_slice(payload) {
                        Ok(r) => r,
                        Err(_) => {
                            return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
                        }
                    };
                let reply =
                    crate::cluster::repo_ensure::release_repo(&root, &req).await?;
                let json = serde_json::to_vec(&reply)
                    .context("encoding RepoReleaseReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::ShutdownPrepCheck => {
                let reply = crate::cluster::shutdown_prep::check().await?;
                let json = serde_json::to_vec(&reply)
                    .context("encoding ShutdownPrepCheckReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::ShutdownPrepExecute => {
                let req: crate::cluster::shutdown_prep::ShutdownPrepExecuteRequest =
                    match serde_json::from_slice(payload) {
                        Ok(r) => r,
                        Err(_) => {
                            return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
                        }
                    };
                let reply =
                    crate::cluster::shutdown_prep::execute(&self.local_name, &req).await?;
                let json = serde_json::to_vec(&reply)
                    .context("encoding ShutdownPrepExecuteReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::BlobStat => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let id = match decode_blob_id(payload) {
                    Some(id) => id,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.stat(&id).await? {
                    Some(stat) => {
                        let json = serde_json::to_vec(&stat)
                            .context("encoding BlobStat as JSON")?;
                        Ok(HandlerOutcome::Reply(json))
                    }
                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
                }
            }
            Method::BlobGet => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let id = match decode_blob_id(payload) {
                    Some(id) => id,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.get_bytes(&id).await? {
                    Some(bytes) => {
                        if bytes.len() > MAX_MESSAGE_BYTES {
                            bail!(
                                "blob {} at {} bytes exceeds RPC cap {}; use streaming variant (Phase 2c)",
                                id.to_hex(),
                                bytes.len(),
                                MAX_MESSAGE_BYTES
                            );
                        }
                        self.metrics.record_blob_get_bytes(bytes.len() as u64);
                        Ok(HandlerOutcome::Reply(bytes))
                    }
                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
                }
            }
            Method::BlobPut => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                // Empty payload is a legitimate empty-blob put — falls
                // through to store.put_bytes(&[]) which yields the
                // hash of the empty byte sequence.
                let id = store.put_bytes(payload).await?;
                self.metrics.record_blob_put_bytes(payload.len() as u64);
                Ok(HandlerOutcome::Reply(id.as_bytes().to_vec()))
            }
            Method::BlobLoadManifest => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let id = match decode_blob_id(payload) {
                    Some(id) => id,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.load_manifest(&id).await? {
                    Some(manifest) => {
                        let json = serde_json::to_vec(&manifest)
                            .context("encoding BlobManifest as JSON")?;
                        Ok(HandlerOutcome::Reply(json))
                    }
                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
                }
            }
            Method::BlobPutStream | Method::BlobGetStream => {
                // Streaming methods go through a different wire path
                // (`serve_connection` peeks at the method tag and hands
                // the raw send/recv streams to the streaming handler).
                // Reaching this arm would mean the caller tried to
                // dispatch a streaming method through the bounded
                // request path — reject with UnknownMethod-shaped
                // error so it's obvious what happened.
                Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
            }
            Method::HasChunk => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let hash = match decode_chunk_hash(payload) {
                    Some(h) => h,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                if store.has_chunk(&hash).await? {
                    self.metrics.record_has_chunk_hit();
                    Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
                } else {
                    self.metrics.record_has_chunk_miss();
                    Ok(HandlerOutcome::Error(ErrorCode::NotFound))
                }
            }
            Method::PutChunk => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                if payload.len() < 32 {
                    return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest));
                }
                let mut hash_bytes = [0u8; 32];
                hash_bytes.copy_from_slice(&payload[..32]);
                let hash = ChunkHash::from_bytes(hash_bytes);
                let chunk_bytes = &payload[32..];
                match store.put_chunk(&hash, chunk_bytes).await {
                    Ok(()) => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
                    Err(e) => {
                        tracing::warn!(error = %e, "PutChunk rejected");
                        Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
                    }
                }
            }
            Method::GetChunk => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let hash = match decode_chunk_hash(payload) {
                    Some(h) => h,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.read_chunk(&hash).await? {
                    Some(bytes) => {
                        // STREAM_STATUS_OK prefix so a legitimate first
                        // content byte of 0xf3 isn't confused with
                        // NotFound. Fixed 1-byte overhead.
                        self.metrics.record_get_chunk_hit();
                        self.metrics.record_blob_get_bytes(bytes.len() as u64);
                        let mut reply = Vec::with_capacity(1 + bytes.len());
                        reply.push(STREAM_STATUS_OK);
                        reply.extend_from_slice(&bytes);
                        Ok(HandlerOutcome::Reply(reply))
                    }
                    None => {
                        self.metrics.record_get_chunk_miss();
                        Ok(HandlerOutcome::Error(ErrorCode::NotFound))
                    }
                }
            }
            Method::PutManifest => {
                let store = match &self.blob_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let manifest: BlobManifest = match serde_json::from_slice(payload) {
                    Ok(m) => m,
                    Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                let missing = store.put_manifest_verified(&manifest).await?;
                let reply = PutManifestReply {
                    blob_id: manifest.blob_id,
                    missing,
                };
                let json = serde_json::to_vec(&reply)
                    .context("encoding PutManifestReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::GetRef => {
                let store = match &self.ref_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match decode_32(payload) {
                    Some(k) => k,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                // Local first.
                if let Some(value) = store.get(&key).await? {
                    self.metrics.record_get_ref_hit();
                    return Ok(HandlerOutcome::Reply(value.to_vec()));
                }
                // Ref-forwarding: try peers via gossip. First hit wins
                // AND pulls the blob into local store so future lookups
                // (and reads) are all local.
                if let Some(value) = self.forward_get_ref(&key).await {
                    self.metrics.record_get_ref_hit();
                    return Ok(HandlerOutcome::Reply(value.to_vec()));
                }
                self.metrics.record_get_ref_miss();
                Ok(HandlerOutcome::Error(ErrorCode::NotFound))
            }
            Method::GetRefLocal => {
                // Strict local lookup — never forwards. Used by peers
                // doing ref-forwarding themselves so we don't loop.
                let store = match &self.ref_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match decode_32(payload) {
                    Some(k) => k,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.get(&key).await? {
                    Some(value) => Ok(HandlerOutcome::Reply(value.to_vec())),
                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
                }
            }
            Method::PutRef => {
                let store = match &self.ref_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                if payload.len() != 64 {
                    return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest));
                }
                let mut key = [0u8; 32];
                let mut value = [0u8; 32];
                key.copy_from_slice(&payload[..32]);
                value.copy_from_slice(&payload[32..]);
                store.put(&key, &value).await?;
                Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
            }
            Method::PutRefVersioned => {
                let store = match &self.ref_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                // Wire: 32-key || 32-value || 8-clock (LE) || 8-node = 80.
                if payload.len() != 32 + StampedRef::ENCODED_LEN {
                    return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest));
                }
                let mut key = [0u8; 32];
                key.copy_from_slice(&payload[..32]);
                let incoming = match StampedRef::from_bytes(&payload[32..]) {
                    Ok(v) => v,
                    Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.put_stamped(&key, incoming).await? {
                    PutOutcome::Merged => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
                    PutOutcome::Rejected { .. } => {
                        Ok(HandlerOutcome::Error(ErrorCode::AlreadyExists))
                    }
                }
            }
            Method::GetRefVersioned => {
                let store = match &self.ref_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match decode_32(payload) {
                    Some(k) => k,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                // Local first.
                if let Some(s) = store.get_stamped(&key).await? {
                    self.metrics.record_get_ref_hit();
                    return Ok(HandlerOutcome::Reply(s.to_bytes().to_vec()));
                }
                // Phase 3b: cross-runner sharing for stamped refs.
                // On miss, fan out to peers; first hit pulls the blob
                // locally + put_stamped so subsequent lookups are
                // pure local hits (same semantics as GetRef path).
                if let Some(s) = self.forward_get_ref_versioned(&key).await {
                    self.metrics.record_get_ref_hit();
                    return Ok(HandlerOutcome::Reply(s.to_bytes().to_vec()));
                }
                self.metrics.record_get_ref_miss();
                Ok(HandlerOutcome::Error(ErrorCode::NotFound))
            }
            Method::GetRefVersionedLocal => {
                let store = match &self.ref_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match decode_32(payload) {
                    Some(k) => k,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.get_stamped(&key).await? {
                    Some(s) => Ok(HandlerOutcome::Reply(s.to_bytes().to_vec())),
                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
                }
            }
            Method::PutTag => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let (key, value) = match crate::cluster::tags::decode_record(payload) {
                    Ok(kv) => kv,
                    Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.put(&key, &value).await {
                    Ok(()) => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
                    Err(e) => {
                        tracing::warn!(error = %e, "PutTag rejected");
                        Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
                    }
                }
            }
            Method::GetTag => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match std::str::from_utf8(payload) {
                    Ok(s) if !s.is_empty() => s,
                    _ => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.get(key).await? {
                    Some(value) => {
                        self.metrics.record_get_tag_hit();
                        Ok(HandlerOutcome::Reply(value.to_vec()))
                    }
                    None => {
                        self.metrics.record_get_tag_miss();
                        Ok(HandlerOutcome::Error(ErrorCode::NotFound))
                    }
                }
            }
            Method::PutTagVersioned => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                // Wire: key_len:u16 (LE) || key_bytes || stamped_value:48.
                let (key, stamped) =
                    match crate::cluster::tags::decode_stamped_record(payload) {
                        Ok(kv) => kv,
                        Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                    };
                match store.put_stamped(&key, stamped).await {
                    Ok(crate::cluster::tags::TagPutOutcome::Merged) => {
                        Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
                    }
                    Ok(crate::cluster::tags::TagPutOutcome::Rejected { .. }) => {
                        Ok(HandlerOutcome::Error(ErrorCode::AlreadyExists))
                    }
                    Err(e) => {
                        tracing::warn!(error = %e, "PutTagVersioned rejected");
                        Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
                    }
                }
            }
            Method::GetTagVersioned => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match std::str::from_utf8(payload) {
                    Ok(s) if !s.is_empty() => s,
                    _ => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.get_stamped(key).await? {
                    Some(s) => {
                        self.metrics.record_get_tag_hit();
                        Ok(HandlerOutcome::Reply(s.to_bytes().to_vec()))
                    }
                    None => {
                        self.metrics.record_get_tag_miss();
                        Ok(HandlerOutcome::Error(ErrorCode::NotFound))
                    }
                }
            }
            Method::DeleteTag => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match std::str::from_utf8(payload) {
                    Ok(s) if !s.is_empty() => s,
                    _ => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                if store.delete(key).await? {
                    Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK]))
                } else {
                    Ok(HandlerOutcome::Error(ErrorCode::NotFound))
                }
            }
            Method::SetTagExpiry => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let (key, expires_at) =
                    match crate::cluster::tags::decode_expiry_record(payload) {
                        Ok(kv) => kv,
                        Err(_) => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                    };
                match store.set_stamped_expiry(&key, expires_at).await {
                    Ok(()) => Ok(HandlerOutcome::Reply(vec![STREAM_STATUS_OK])),
                    Err(e) => {
                        tracing::warn!(error = %e, "SetTagExpiry failed");
                        Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest))
                    }
                }
            }
            Method::GetTagExpiry => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let key = match std::str::from_utf8(payload) {
                    Ok(s) if !s.is_empty() => s,
                    _ => return Ok(HandlerOutcome::Error(ErrorCode::InvalidRequest)),
                };
                match store.get_stamped_expiry(key).await? {
                    Some(exp) => Ok(HandlerOutcome::Reply(exp.to_le_bytes().to_vec())),
                    None => Ok(HandlerOutcome::Error(ErrorCode::NotFound)),
                }
            }
            Method::ListTags => {
                let store = match &self.tag_store {
                    Some(s) => s,
                    None => return Ok(HandlerOutcome::Error(ErrorCode::NotConfigured)),
                };
                let entries = store.list().await?;
                let json = serde_json::to_vec(&entries)
                    .context("encoding TagEntry list as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
            Method::GetMetrics => {
                let snapshot = self.metrics.snapshot();
                let json = serde_json::to_vec(&snapshot)
                    .context("encoding MetricsReply as JSON")?;
                Ok(HandlerOutcome::Reply(json))
            }
        }
    }

    /// Ref-forwarding (2026-07-13): consult live gossip peers for
    /// this ref. Returns `Some(value)` on the first hit AFTER pulling
    /// the blob into the local store; returns `None` when no peer
    /// has it, forwarding is disabled, or nothing succeeded within
    /// the timeout budget.
    ///
    /// Uses [`Method::GetRefLocal`] on peers so we never loop.
    /// Concurrent peer probes via `JoinSet`; the first successful
    /// pull wins and remaining tasks are aborted.
    async fn forward_get_ref(&self, key: &RefKey) -> Option<RefValue> {
        let client = self.outbound_client.as_ref()?.clone();
        let blob_store = self.blob_store.as_ref()?.clone();
        let ref_store = self.ref_store.as_ref()?.clone();
        let peers: Vec<PeerView> = self
            .gossip
            .peers()
            .await
            .into_iter()
            .filter(|p| p.alive && p.rpc_lan.or(p.rpc_tailscale).is_some())
            .collect();
        if peers.is_empty() {
            return None;
        }
        // Fan out — each task tries one peer. The first that pulls a
        // blob wins; concurrent tasks are aborted.
        let mut set: tokio::task::JoinSet<Option<RefValue>> = tokio::task::JoinSet::new();
        let key_owned = *key;
        for peer in peers {
            let client = client.clone();
            let blob_store = blob_store.clone();
            let ref_store = ref_store.clone();
            set.spawn(async move {
                let addr = peer.rpc_lan.or(peer.rpc_tailscale)?;
                let conn = match tokio::time::timeout(
                    std::time::Duration::from_secs(3),
                    client.connect(addr, &peer.name),
                )
                .await
                {
                    Ok(Ok(c)) => c,
                    _ => return None,
                };
                let value = match call_get_ref_local(&conn, &key_owned).await {
                    Ok(Some(v)) => v,
                    _ => return None,
                };
                let blob_id = crate::cluster::blob::BlobId::from_bytes(value);
                // Pull the blob's chunks + manifest into local store.
                if pull_blob_locally(&conn, &blob_store, &blob_id)
                    .await
                    .is_err()
                {
                    return None;
                }
                // Persist the ref locally so future GetRef calls are
                // pure local hits (no forwarding roundtrip).
                if ref_store.put(&key_owned, &value).await.is_err() {
                    return None;
                }
                Some(value)
            });
        }
        while let Some(join) = set.join_next().await {
            if let Ok(Some(value)) = join {
                set.abort_all();
                return Some(value);
            }
        }
        None
    }

    /// Phase 3b (2026-07-13): stamped-ref forwarder. Same shape as
    /// [`Self::forward_get_ref`] but uses `GetRefVersionedLocal` and
    /// `put_stamped` on the local side so the CRDT-merge semantics
    /// carry through cross-node lookups.
    async fn forward_get_ref_versioned(
        &self,
        key: &RefKey,
    ) -> Option<StampedRef> {
        let client = self.outbound_client.as_ref()?.clone();
        let blob_store = self.blob_store.as_ref()?.clone();
        let ref_store = self.ref_store.as_ref()?.clone();
        let peers: Vec<PeerView> = self
            .gossip
            .peers()
            .await
            .into_iter()
            .filter(|p| p.alive && p.rpc_lan.or(p.rpc_tailscale).is_some())
            .collect();
        if peers.is_empty() {
            return None;
        }
        let mut set: tokio::task::JoinSet<Option<StampedRef>> =
            tokio::task::JoinSet::new();
        let key_owned = *key;
        for peer in peers {
            let client = client.clone();
            let blob_store = blob_store.clone();
            let ref_store = ref_store.clone();
            set.spawn(async move {
                let addr = peer.rpc_lan.or(peer.rpc_tailscale)?;
                let conn = match tokio::time::timeout(
                    std::time::Duration::from_secs(3),
                    client.connect(addr, &peer.name),
                )
                .await
                {
                    Ok(Ok(c)) => c,
                    _ => return None,
                };
                let stamped = match call_get_ref_versioned_local(&conn, &key_owned).await {
                    Ok(Some(s)) => s,
                    _ => return None,
                };
                let blob_id = crate::cluster::blob::BlobId::from_bytes(stamped.value);
                if pull_blob_locally(&conn, &blob_store, &blob_id)
                    .await
                    .is_err()
                {
                    return None;
                }
                // `put_stamped` merges: if we happened to race a
                // concurrent local write, the higher (clock, node)
                // wins on disk. Either way return what we fetched
                // so the caller sees a hit.
                let _ = ref_store.put_stamped(&key_owned, stamped).await;
                Some(stamped)
            });
        }
        while let Some(join) = set.join_next().await {
            if let Ok(Some(s)) = join {
                set.abort_all();
                return Some(s);
            }
        }
        None
    }
}

/// Ref-forwarding helper (2026-07-13): fetch a blob's manifest and
/// missing chunks from `conn` into the local `BlobStore`. Same shape
/// as `prewarm_missing_chunks_between_parallel` but the downstream is
/// in-process rather than another peer.
async fn pull_blob_locally(
    conn: &Connection,
    local: &BlobStore,
    id: &crate::cluster::blob::BlobId,
) -> Result<()> {
    use crate::cluster::rpc::{call_blob_load_manifest, call_get_chunk};
    let manifest = call_blob_load_manifest(conn, id)
        .await?
        .with_context(|| format!("peer missing manifest for blob {}", id.to_hex()))?;
    for hash in &manifest.chunks {
        if local.has_chunk(hash).await? {
            continue;
        }
        let bytes = call_get_chunk(conn, hash).await?.with_context(|| {
            format!(
                "peer manifest referenced chunk {} but GetChunk returned NotFound",
                hash.to_hex()
            )
        })?;
        local.put_chunk(hash, &bytes).await?;
    }
    // Commit the manifest.
    let missing = local.put_manifest_verified(&manifest).await?;
    if !missing.is_empty() {
        bail!(
            "pulled blob {} but {} chunks still missing after fetch",
            id.to_hex(),
            missing.len()
        );
    }
    Ok(())
}

/// Parse a payload as a 32-byte array. Shared by `GetRef` and any
/// future single-32-byte-payload methods.
fn decode_32(payload: &[u8]) -> Option<[u8; 32]> {
    if payload.len() != 32 {
        return None;
    }
    let mut out = [0u8; 32];
    out.copy_from_slice(payload);
    Some(out)
}

/// Parse a payload as a 32-byte ChunkHash. Same shape as
/// [`decode_blob_id`] but a distinct function so the reader-visible
/// type at each call site is unambiguous.
fn decode_chunk_hash(payload: &[u8]) -> Option<ChunkHash> {
    if payload.len() != 32 {
        return None;
    }
    let mut buf = [0u8; 32];
    buf.copy_from_slice(payload);
    Some(ChunkHash::from_bytes(buf))
}

/// Parse a payload as a 32-byte BlobId. Returns `None` for any other
/// length so the caller can surface [`ErrorCode::InvalidRequest`].
fn decode_blob_id(payload: &[u8]) -> Option<BlobId> {
    if payload.len() != 32 {
        return None;
    }
    let mut buf = [0u8; 32];
    buf.copy_from_slice(payload);
    Some(BlobId::from_bytes(buf))
}

/// Server-side: loop accepting bidi streams on `conn`, dispatch to
/// `router`, write the reply. Returns cleanly when the peer closes the
/// connection.
pub async fn serve_connection(conn: Connection, router: Arc<RpcRouter>) -> Result<()> {
    loop {
        let (mut send, mut recv) = match conn.accept_bi().await {
            Ok(pair) => pair,
            Err(ConnectionError::ApplicationClosed(_))
            | Err(ConnectionError::ConnectionClosed(_))
            | Err(ConnectionError::LocallyClosed)
            | Err(ConnectionError::TimedOut) => return Ok(()),
            Err(e) => return Err(anyhow::Error::from(e)),
        };

        // Peek at the method tag byte to decide whether to hand the
        // stream off to a streaming handler or drain it into a bounded
        // request buffer.
        let mut tag = [0u8; 1];
        match recv.read_exact(&mut tag).await {
            Ok(()) => {}
            Err(_) => {
                let _ = send.write_all(&[ErrorCode::EmptyRequest.as_byte()]).await;
                let _ = send.finish();
                continue;
            }
        };

        match Method::from_byte(tag[0]) {
            Some(Method::BlobPutStream) => {
                if let Err(e) = handle_blob_put_stream(&router, recv, send).await {
                    tracing::warn!(error = %e, "BlobPutStream handler failed");
                }
            }
            Some(Method::BlobGetStream) => {
                if let Err(e) = handle_blob_get_stream(&router, recv, send).await {
                    tracing::warn!(error = %e, "BlobGetStream handler failed");
                }
            }
            Some(_) => {
                // Bounded methods: drain the remainder of the request
                // into memory and dispatch as before.
                let rest = recv
                    .read_to_end(MAX_MESSAGE_BYTES.saturating_sub(1))
                    .await
                    .context("reading bounded RPC request")?;
                let mut request = Vec::with_capacity(1 + rest.len());
                request.push(tag[0]);
                request.extend_from_slice(&rest);
                let reply = dispatch(&router, &request).await;
                send.write_all(&reply).await.context("writing RPC reply")?;
                send.finish().context("finishing RPC send stream")?;
            }
            None => {
                let _ = send.write_all(&[ErrorCode::UnknownMethod.as_byte()]).await;
                let _ = send.finish();
            }
        }
    }
}

/// Streaming handler for [`Method::BlobPutStream`]. Feeds the incoming
/// bytes straight into `BlobStore::put_stream` — memory ceiling is one
/// chunk buffer regardless of blob size.
async fn handle_blob_put_stream(
    router: &RpcRouter,
    recv: quinn::RecvStream,
    mut send: quinn::SendStream,
) -> Result<()> {
    let store = match router.blob_store() {
        Some(s) => s.clone(),
        None => {
            send.write_all(&[ErrorCode::NotConfigured.as_byte()]).await?;
            send.finish()?;
            return Ok(());
        }
    };
    match store.put_stream(recv).await {
        Ok(id) => {
            // Field finding 2026-07-12: the streaming variants had never
            // been counted, so `clawstor_cache_blob_put_bytes_total`
            // stayed at 0 even after multi-MB uploads. Look up the size
            // via `stat` — cheap (single manifest read) and gives the
            // authoritative post-store byte count.
            if let Ok(Some(stat)) = store.stat(&id).await {
                router.metrics.record_blob_put_bytes(stat.total_size);
            }
            let mut reply = Vec::with_capacity(33);
            reply.push(STREAM_STATUS_OK);
            reply.extend_from_slice(id.as_bytes());
            send.write_all(&reply).await?;
            send.finish()?;
        }
        Err(e) => {
            tracing::warn!(error = %e, "BlobPutStream put_stream failed");
            let _ = send.write_all(&[ErrorCode::HandlerFailure.as_byte()]).await;
            let _ = send.finish();
        }
    }
    Ok(())
}

/// Streaming handler for [`Method::BlobGetStream`]. Verifies the blob
/// exists (writes `NotFound` on absence), then pipes each chunk from
/// disk straight into the send stream. Callers see:
/// `STREAM_STATUS_OK` (1 byte) followed by the blob bytes.
async fn handle_blob_get_stream(
    router: &RpcRouter,
    mut recv: quinn::RecvStream,
    mut send: quinn::SendStream,
) -> Result<()> {
    let store = match router.blob_store() {
        Some(s) => s.clone(),
        None => {
            send.write_all(&[ErrorCode::NotConfigured.as_byte()]).await?;
            send.finish()?;
            return Ok(());
        }
    };
    let mut id_bytes = [0u8; 32];
    if recv.read_exact(&mut id_bytes).await.is_err() {
        send.write_all(&[ErrorCode::InvalidRequest.as_byte()]).await?;
        send.finish()?;
        return Ok(());
    }
    let id = BlobId::from_bytes(id_bytes);
    match store.load_manifest(&id).await? {
        None => {
            send.write_all(&[ErrorCode::NotFound.as_byte()]).await?;
        }
        Some(manifest) => {
            send.write_all(&[STREAM_STATUS_OK]).await?;
            store.stream_to(&id, &mut send).await?;
            // Field finding 2026-07-12: streaming GETs weren't counted,
            // leaving `blob_get_bytes_total` at 0. Record the manifest's
            // authoritative total_size — we've committed to serving the
            // whole thing by this point.
            router.metrics.record_blob_get_bytes(manifest.total_size);
        }
    }
    send.finish()?;
    Ok(())
}

/// Turn a raw wire-format request into a response — either the
/// router's real reply or a single-byte error code. Extracted so tests
/// can hit it without a QUIC connection.
async fn dispatch(router: &RpcRouter, request: &[u8]) -> Vec<u8> {
    if request.is_empty() {
        return vec![ErrorCode::EmptyRequest.as_byte()];
    }
    let method = match Method::from_byte(request[0]) {
        Some(m) => m,
        None => return vec![ErrorCode::UnknownMethod.as_byte()],
    };
    let payload = &request[1..];
    match router.handle(method, payload).await {
        Ok(reply) => reply,
        Err(e) => {
            tracing::warn!(error = %e, method = ?method, "RPC handler failed");
            vec![ErrorCode::HandlerFailure.as_byte()]
        }
    }
}

// Client helpers live in a submodule to stay under the 1300-line
// ceiling on this file. Re-exported so external code keeps writing
// `cluster::rpc::call_*`.
#[path = "rpc/client.rs"]
mod client;
pub use client::*;

#[cfg(test)]
#[path = "rpc/tests.rs"]
mod tests;

#[cfg(test)]
#[path = "rpc/tests_phase5.rs"]
mod tests_phase5;

#[cfg(test)]
#[path = "rpc/tests_forwarding.rs"]
mod tests_forwarding;

#[cfg(test)]
#[path = "rpc/tests_phase4b_ttl.rs"]
mod tests_phase4b_ttl;