//! Iroh blob store implementation (ADR-025)
//!
//! This module implements the `BlobStore` trait using iroh-blobs,
//! providing content-addressed blob storage with P2P mesh synchronization.
//!
//! # iroh-blobs Characteristics
//!
//! - Content-addressed storage using BLAKE3 hashes (32 bytes)
//! - Built on iroh's QUIC-based P2P networking
//! - Optimized for large file transfers with verified streaming
//! - No native metadata support (we use sidecar JSON files)
//!
//! # Relay policy (defenseunicorns/peat#833)
//!
//! `NetworkedIrohBlobStore::build_endpoint*` constructors build their
//! underlying [`iroh::Endpoint`] through a feature-gated helper that
//! respects the crate-level relay policy:
//!
//! - **Default (no `relay-n0-hosted` feature):** [`iroh::Endpoint::empty_builder`]
//!   — no relay servers and no DNS pkarr discovery via n0-hosted
//!   infrastructure. If [`IrohConfig::relay_urls`] is non-empty, those
//!   URLs are applied as `RelayMode::Custom`; otherwise relays are
//!   disabled entirely. Default builds will not transit `*.iroh.network`
//!   or `*.iroh-canary.iroh.link`.
//! - **With `--features relay-n0-hosted`:** `Endpoint::builder(iroh::endpoint::presets::N0)`
//!   — restores n0's hosted relay pool and DNS discovery. Empty
//!   `relay_urls` then falls through to n0's default relay set. Opt-in
//!   only.
//!
//! This is the same posture enforced in `peat-protocol`'s `IrohTransport`.
//! Implementers integrating this blob store should treat the default as
//! the contract: do not assume relay-backed connectivity unless the
//! consuming crate explicitly enables `relay-n0-hosted` or supplies a
//! self-hosted relay configuration via `relay_urls`.
//!
//! # Usage
//!
//! ```ignore
//! use peat_mesh::storage::{IrohBlobStore, BlobStore, BlobMetadata};
//! use std::path::Path;
//!
//! let blob_store = IrohBlobStore::new_in_memory(blob_dir).await?;
//!
//! // Create blob from file
//! let token = blob_store.create_blob(
//!     Path::new("/models/yolov8.onnx"),
//!     BlobMetadata::with_name("yolov8.onnx")
//! ).await?;
//!
//! // Token can be shared via CRDT documents
//! // Other nodes can then fetch the blob
//! ```

use super::blob_announce;
use super::blob_traits::{BlobHandle, BlobHash, BlobMetadata, BlobProgress, BlobStore, BlobToken};
use anyhow::{Context, Result};
use iroh_blobs::{store::mem::MemStore, Hash};
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::RwLock;
use std::time::{Duration, Instant};
use tokio_stream::StreamExt;
use tracing::{debug, info, warn};

/// Sidecar metadata stored alongside blobs
///
/// Since iroh-blobs doesn't support native metadata, we store
/// metadata in JSON sidecar files: `{blob_dir}/{hash}.meta.json`
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct SidecarMetadata {
    /// Original BlobMetadata
    metadata: BlobMetadata,
    /// Size in bytes
    size_bytes: u64,
    /// Creation timestamp (Unix seconds)
    created_at: u64,
}

impl SidecarMetadata {
    fn new(metadata: BlobMetadata, size_bytes: u64) -> Self {
        Self {
            metadata,
            size_bytes,
            created_at: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .map(|d| d.as_secs())
                .unwrap_or(0),
        }
    }
}

/// Iroh blob store implementing the BlobStore trait
///
/// Wraps iroh-blobs' in-memory store to provide backend-agnostic blob storage.
/// Blobs are content-addressed using BLAKE3 hashes.
///
/// # Metadata Storage
///
/// Since iroh-blobs doesn't support native metadata, we store metadata
/// in JSON sidecar files alongside the blob data.
pub struct IrohBlobStore {
    /// In-memory blob store from iroh-blobs
    store: MemStore,
    /// Cache of known blob tokens (hash -> token)
    token_cache: RwLock<HashMap<BlobHash, BlobToken>>,
    /// Directory for blob data exports and metadata sidecars
    blob_dir: PathBuf,
}

impl IrohBlobStore {
    /// Create a new Iroh blob store with in-memory storage
    ///
    /// # Arguments
    ///
    /// * `blob_dir` - Directory for exported blobs and metadata sidecars
    pub async fn new_in_memory(blob_dir: PathBuf) -> Result<Self> {
        // Create blob directory
        if let Err(e) = std::fs::create_dir_all(&blob_dir) {
            warn!("Failed to create blob directory {:?}: {}", blob_dir, e);
        }

        let store = MemStore::default();

        Ok(Self {
            store,
            token_cache: RwLock::new(HashMap::new()),
            blob_dir,
        })
    }

    /// Create with default temp directory
    pub async fn new_temp() -> Result<Self> {
        let blob_dir = std::env::temp_dir().join("peat_iroh_blobs");
        Self::new_in_memory(blob_dir).await
    }

    /// Get access to the underlying iroh-blobs store
    pub fn store(&self) -> &MemStore {
        &self.store
    }

    /// Get the blob directory path
    pub fn blob_dir(&self) -> &Path {
        &self.blob_dir
    }

    /// Convert iroh Hash to our BlobHash
    fn iroh_hash_to_blob_hash(hash: &Hash) -> BlobHash {
        BlobHash::from_hex(&hash.to_hex())
    }

    /// Convert our BlobHash to iroh Hash
    fn blob_hash_to_iroh_hash(hash: &BlobHash) -> Result<Hash> {
        Hash::from_str(hash.as_hex())
            .map_err(|e| anyhow::anyhow!("Invalid blob hash '{}': {}", hash.as_hex(), e))
    }

    /// Get the path for a blob's metadata sidecar file
    fn metadata_path(&self, hash: &BlobHash) -> PathBuf {
        self.blob_dir.join(format!("{}.meta.json", hash.as_hex()))
    }

    /// Get the local path for exported blob content
    fn local_blob_path(&self, hash: &BlobHash) -> PathBuf {
        self.blob_dir.join(hash.as_hex())
    }

    /// Save metadata sidecar file
    fn save_metadata(&self, hash: &BlobHash, metadata: &SidecarMetadata) -> Result<()> {
        let path = self.metadata_path(hash);
        let json = serde_json::to_string_pretty(metadata)?;
        std::fs::write(&path, json)
            .with_context(|| format!("Failed to write metadata to {:?}", path))?;
        Ok(())
    }

    /// Load metadata sidecar file
    fn load_metadata(&self, hash: &BlobHash) -> Option<SidecarMetadata> {
        let path = self.metadata_path(hash);
        if !path.exists() {
            return None;
        }
        std::fs::read_to_string(&path)
            .ok()
            .and_then(|json| serde_json::from_str(&json).ok())
    }

    /// Delete metadata sidecar file
    fn delete_metadata(&self, hash: &BlobHash) -> Result<()> {
        let path = self.metadata_path(hash);
        if path.exists() {
            std::fs::remove_file(&path)
                .with_context(|| format!("Failed to delete metadata at {:?}", path))?;
        }
        Ok(())
    }

    /// Cache a token for later lookup
    fn cache_token(&self, token: &BlobToken) {
        if let Ok(mut cache) = self.token_cache.write() {
            cache.insert(token.hash.clone(), token.clone());
        }
    }

    /// Export blob content to local filesystem
    async fn export_blob(&self, hash: &Hash) -> Result<PathBuf> {
        let blob_hash = Self::iroh_hash_to_blob_hash(hash);
        let local_path = self.local_blob_path(&blob_hash);

        // If already exported, return existing path
        if local_path.exists() {
            return Ok(local_path);
        }

        // Read blob content from store using get_bytes()
        let content = self
            .store
            .get_bytes(*hash)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to get blob {}: {}", hash.to_hex(), e))?;

        // Write to local file
        std::fs::write(&local_path, &content)
            .with_context(|| format!("Failed to export blob to {:?}", local_path))?;

        Ok(local_path)
    }
}

#[async_trait::async_trait]
impl BlobStore for IrohBlobStore {
    async fn create_blob(&self, path: &Path, metadata: BlobMetadata) -> Result<BlobToken> {
        info!("Creating blob from file: {:?}", path);

        // Verify file exists
        if !path.exists() {
            return Err(anyhow::anyhow!("File not found: {:?}", path));
        }

        // Read file content
        let content = std::fs::read(path).with_context(|| format!("Failed to read {:?}", path))?;
        let size_bytes = content.len() as u64;

        // Add to iroh-blobs store using add_bytes()
        let tag = self.store.add_bytes(content).await?;
        let hash = tag.hash;

        // Build our token
        let token = BlobToken {
            hash: Self::iroh_hash_to_blob_hash(&hash),
            size_bytes,
            metadata: metadata.clone(),
        };

        // Save metadata sidecar
        let sidecar = SidecarMetadata::new(metadata, size_bytes);
        self.save_metadata(&token.hash, &sidecar)?;

        // Cache for later lookups
        self.cache_token(&token);

        debug!(
            "Created blob: hash={}, size={}",
            token.hash.as_hex(),
            token.size_bytes
        );

        Ok(token)
    }

    async fn create_blob_from_stream(
        &self,
        stream: &mut (dyn tokio::io::AsyncRead + Send + Unpin),
        _expected_size: Option<u64>,
        metadata: BlobMetadata,
    ) -> Result<BlobToken> {
        use std::sync::atomic::{AtomicU64, Ordering};
        use tokio::io::AsyncWriteExt;

        static STREAM_COUNTER: AtomicU64 = AtomicU64::new(0);

        info!("Creating blob from stream");

        // Stream to temp file to limit peak memory to one copy
        let temp_path = self.blob_dir.join(format!(
            ".stream_import_{}",
            STREAM_COUNTER.fetch_add(1, Ordering::Relaxed)
        ));

        {
            let mut file = tokio::fs::File::create(&temp_path)
                .await
                .with_context(|| format!("Failed to create temp file {:?}", temp_path))?;
            tokio::io::copy(stream, &mut file)
                .await
                .map_err(|e| anyhow::anyhow!("Failed to stream blob to temp file: {}", e))?;
            file.flush().await?;
        }

        let result = self.create_blob(&temp_path, metadata).await;

        // Clean up temp file
        let _ = tokio::fs::remove_file(&temp_path).await;

        result
    }

    async fn create_blob_from_bytes(
        &self,
        data: &[u8],
        metadata: BlobMetadata,
    ) -> Result<BlobToken> {
        info!("Creating blob from {} bytes", data.len());

        let size_bytes = data.len() as u64;

        // Add to iroh-blobs store using add_bytes()
        let tag = self.store.add_bytes(data.to_vec()).await?;
        let hash = tag.hash;

        // Build our token
        let token = BlobToken {
            hash: Self::iroh_hash_to_blob_hash(&hash),
            size_bytes,
            metadata: metadata.clone(),
        };

        // Save metadata sidecar
        let sidecar = SidecarMetadata::new(metadata, size_bytes);
        self.save_metadata(&token.hash, &sidecar)?;

        // Cache for later lookups
        self.cache_token(&token);

        debug!(
            "Created blob from bytes: hash={}, size={}",
            token.hash.as_hex(),
            token.size_bytes
        );

        Ok(token)
    }

    async fn fetch_blob<F>(&self, token: &BlobToken, mut progress: F) -> Result<BlobHandle>
    where
        F: FnMut(BlobProgress) + Send + 'static,
    {
        info!("Fetching blob: hash={}", token.hash.as_hex());

        // Check if we already have it exported locally
        let local_path = self.local_blob_path(&token.hash);
        if local_path.exists() {
            debug!("Blob already exists locally at {:?}", local_path);
            progress(BlobProgress::Completed {
                local_path: local_path.clone(),
            });
            return Ok(BlobHandle::new(token.clone(), local_path));
        }

        // Send started event
        progress(BlobProgress::Started {
            total_bytes: token.size_bytes,
        });

        // Convert hash and check if in store using has()
        let iroh_hash = Self::blob_hash_to_iroh_hash(&token.hash)?;

        if self.store.has(iroh_hash).await? {
            // Blob is in our store, export it
            progress(BlobProgress::Downloading {
                downloaded_bytes: token.size_bytes / 2,
                total_bytes: token.size_bytes,
            });

            let export_path = self.export_blob(&iroh_hash).await?;

            progress(BlobProgress::Completed {
                local_path: export_path.clone(),
            });

            return Ok(BlobHandle::new(token.clone(), export_path));
        }

        // Blob not available locally
        // In Phase 1, remote fetch requires the P2P layer to be connected
        // and the blob to be announced. For now, return an error.
        progress(BlobProgress::Failed {
            error: format!(
                "Blob {} not available locally. Remote fetch requires P2P connectivity.",
                token.hash
            ),
        });

        Err(anyhow::anyhow!(
            "Blob {} not available locally. In Phase 1, ensure the blob is stored \
             on this node or connected via P2P to a node that has it.",
            token.hash.as_hex()
        ))
    }

    fn blob_exists_locally(&self, hash: &BlobHash) -> bool {
        // Check our local blob directory
        let local_path = self.local_blob_path(hash);
        if local_path.exists() {
            return true;
        }

        // Check cache
        if let Ok(cache) = self.token_cache.read() {
            if cache.contains_key(hash) {
                return true;
            }
        }

        // Check metadata sidecar (indicates we've seen this blob)
        self.metadata_path(hash).exists()
    }

    fn blob_info(&self, hash: &BlobHash) -> Option<BlobToken> {
        // Check cache first
        if let Ok(cache) = self.token_cache.read() {
            if let Some(token) = cache.get(hash) {
                return Some(token.clone());
            }
        }

        // Try loading from metadata sidecar
        if let Some(sidecar) = self.load_metadata(hash) {
            let token = BlobToken {
                hash: hash.clone(),
                size_bytes: sidecar.size_bytes,
                metadata: sidecar.metadata,
            };
            return Some(token);
        }

        None
    }

    async fn delete_blob(&self, hash: &BlobHash) -> Result<()> {
        info!("Deleting blob: hash={}", hash.as_hex());

        // Remove from local storage
        let local_path = self.local_blob_path(hash);
        if local_path.exists() {
            std::fs::remove_file(&local_path)
                .with_context(|| format!("Failed to delete local blob: {:?}", local_path))?;
        }

        // Remove metadata sidecar
        self.delete_metadata(hash)?;

        // Remove from cache
        if let Ok(mut cache) = self.token_cache.write() {
            cache.remove(hash);
        }

        // Note: We don't delete from the iroh-blobs store directly
        // as it may be shared or used for P2P sync. The in-memory store
        // will be garbage collected when no longer referenced.

        Ok(())
    }

    fn list_local_blobs(&self) -> Vec<BlobToken> {
        let mut tokens = Vec::new();

        // First, get tokens from cache
        if let Ok(cache) = self.token_cache.read() {
            tokens.extend(cache.values().cloned());
        }

        // Also scan metadata directory for any we might have missed
        if let Ok(entries) = std::fs::read_dir(&self.blob_dir) {
            for entry in entries.flatten() {
                let path = entry.path();
                if let Some(filename) = path.file_name().and_then(|n| n.to_str()) {
                    if filename.ends_with(".meta.json") {
                        let hash_hex = filename.trim_end_matches(".meta.json");
                        let hash = BlobHash::from_hex(hash_hex);

                        // Skip if already in tokens
                        if tokens.iter().any(|t| t.hash == hash) {
                            continue;
                        }

                        // Load metadata and add to tokens
                        if let Some(sidecar) = self.load_metadata(&hash) {
                            tokens.push(BlobToken {
                                hash,
                                size_bytes: sidecar.size_bytes,
                                metadata: sidecar.metadata,
                            });
                        }
                    }
                }
            }
        }

        tokens
    }

    fn local_storage_bytes(&self) -> u64 {
        // Sum up sizes from cache
        // This represents logical storage used by blobs we know about,
        // regardless of whether they've been exported to disk yet.
        if let Ok(cache) = self.token_cache.read() {
            if !cache.is_empty() {
                return cache.values().map(|t| t.size_bytes).sum();
            }
        }

        // Fallback: scan metadata sidecars for size info
        let mut total = 0u64;
        if let Ok(entries) = std::fs::read_dir(&self.blob_dir) {
            for entry in entries.flatten() {
                let path = entry.path();
                if let Some(filename) = path.file_name().and_then(|n| n.to_str()) {
                    if filename.ends_with(".meta.json") {
                        let hash_hex = filename.trim_end_matches(".meta.json");
                        let hash = BlobHash::from_hex(hash_hex);
                        if let Some(sidecar) = self.load_metadata(&hash) {
                            total += sidecar.size_bytes;
                        }
                    }
                }
            }
        }

        total
    }
}

// ============================================================================
// NetworkedIrohBlobStore - P2P blob sync with iroh endpoint
// ============================================================================

use crate::config::IrohConfig;
use iroh::address_lookup::memory::MemoryLookup;
use iroh::protocol::Router;
use iroh::{
    Endpoint, EndpointAddr, EndpointId, RelayMap, RelayMode, RelayUrl, SecretKey, TransportAddr,
};
use iroh_blobs::BlobsProtocol;
use std::collections::HashSet;
use std::sync::Arc;
use tokio::sync::RwLock as TokioRwLock;

/// No-op endpoint hooks used as the default type parameter for [`build_endpoint`].
#[derive(Debug)]
struct NoopEndpointHooks;
impl iroh::endpoint::EndpointHooks for NoopEndpointHooks {}

/// Construct a fresh iroh `Endpoint` builder respecting the relay policy
/// (defenseunicorns/peat#833).
///
/// Default builds return [`iroh::Endpoint::empty_builder`] — no relay
/// servers, no pkarr/DNS discovery via n0-hosted infrastructure. Callers
/// supply explicit relay URLs via [`IrohConfig::relay_urls`] if they need
/// them; otherwise NAT traversal relies on direct addresses or LAN
/// discovery (mDNS via `address_lookup`).
///
/// With the `relay-n0-hosted` cargo feature enabled, returns
/// `Endpoint::builder(iroh::endpoint::presets::N0)`, restoring n0's hosted
/// relay pool (`*.iroh.network`, `*.iroh-canary.iroh.link`) and DNS pkarr
/// discovery. Opt-in only.
fn relay_policy_builder() -> iroh::endpoint::Builder {
    // peat#923 / ADR-060 §5: every iroh endpoint in peat-mesh routes through
    // `crate::security::tls_provider::iroh_quic_provider()` so the FIPS-approved
    // aws-lc-rs provider is wired by construction, not by reviewer vigilance.
    // See that module's doc for the full rationale.
    let provider = crate::security::tls_provider::iroh_quic_provider();
    #[cfg(feature = "relay-n0-hosted")]
    {
        Endpoint::builder(iroh::endpoint::presets::N0).crypto_provider(provider)
    }
    #[cfg(not(feature = "relay-n0-hosted"))]
    {
        // iroh 0.98 retired `empty_builder()`; `presets::Empty` has no
        // mandatory options set, so the caller must provide the crypto
        // provider explicitly (which we do above, unconditionally).
        Endpoint::builder(iroh::endpoint::presets::Empty).crypto_provider(provider)
    }
}

/// Endpoint hooks that reject QUIC connections from peers not in the
/// [`FormationPeerSet`].
///
/// Enrollment ALPN (`peat/enroll/1`) is exempted because enrolling
/// devices are not yet formation members — enrollment has its own
/// token-based authentication.
///
/// All other ALPNs (sync, blobs, etc.) require the remote peer's
/// `EndpointId` to be present in the allowed set. Since EndpointIds
/// are derived from the formation secret via HKDF and QUIC TLS proves
/// private key possession, this is sufficient to gate access.
#[derive(Debug, Clone)]
pub struct FormationEndpointHooks {
    allowed_peers: crate::security::FormationPeerSet,
}

impl FormationEndpointHooks {
    pub fn new(allowed_peers: crate::security::FormationPeerSet) -> Self {
        Self { allowed_peers }
    }
}

#[allow(clippy::manual_async_fn)]
impl iroh::endpoint::EndpointHooks for FormationEndpointHooks {
    fn after_handshake<'a>(
        &'a self,
        conn: &'a iroh::endpoint::Connection,
    ) -> impl std::future::Future<Output = iroh::endpoint::AfterHandshakeOutcome> + Send + 'a {
        async move {
            // Enrollment ALPN is exempted — new devices need to enroll
            // before they become formation members.
            if conn.alpn() == super::enrollment_transport::CAP_ENROLLMENT_ALPN {
                return iroh::endpoint::AfterHandshakeOutcome::Accept;
            }

            let remote = conn.remote_id();
            if self.allowed_peers.contains(&remote) {
                tracing::debug!(
                    peer = %remote.fmt_short(),
                    alpn = %String::from_utf8_lossy(conn.alpn()),
                    "QUIC connection accepted (formation member)"
                );
                iroh::endpoint::AfterHandshakeOutcome::Accept
            } else {
                tracing::warn!(
                    peer = %remote.fmt_short(),
                    alpn = %String::from_utf8_lossy(conn.alpn()),
                    "QUIC connection REJECTED (not a formation member)"
                );
                iroh::endpoint::AfterHandshakeOutcome::Reject {
                    error_code: 3u32.into(),
                    reason: b"not a formation member".to_vec(),
                }
            }
        }
    }
}

// ============================================================================
// BlobPeerIndex - O(1) blob-to-peer resolution
// ============================================================================

/// Cheap well-formedness check for a blob hash received over provider gossip:
/// iroh blobs are BLAKE3, whose hex form is exactly 64 hex digits. Rejecting
/// anything else keeps a hostile peer from inflating the index with arbitrary
/// garbage keys (it is a sanity gate, not cryptographic verification — the real
/// content check happens on transfer).
fn is_plausible_blob_hash(hash: &BlobHash) -> bool {
    let hex = hash.as_hex();
    hex.len() == 64 && hex.bytes().all(|b| b.is_ascii_hexdigit())
}

/// Tracks which peers have which blobs for O(1) resolution.
///
/// Maintains bidirectional mappings between blob hashes and peer endpoints,
/// enabling efficient lookup in both directions without scanning.
///
/// # Thread Safety
///
/// This struct is intended to be wrapped in a `TokioRwLock` by its owner.
#[derive(Debug, Default)]
pub struct BlobPeerIndex {
    /// Reverse index: blob hash → set of peers that have it
    blob_to_peers: HashMap<BlobHash, HashSet<EndpointId>>,
    /// Forward index: peer → set of blob hashes they have
    peer_to_blobs: HashMap<EndpointId, HashSet<BlobHash>>,
}

impl BlobPeerIndex {
    /// Create an empty index.
    pub fn new() -> Self {
        Self::default()
    }

    /// Record that a peer has a specific blob.
    ///
    /// Returns `true` if this `(peer, hash)` pairing was newly learned (not
    /// already known). Provider gossip uses this to re-broadcast only newly
    /// learned holdings, so an epidemic announce terminates naturally once a
    /// region has converged instead of flooding indefinitely.
    pub fn advertise(&mut self, peer: EndpointId, hash: BlobHash) -> bool {
        let newly_for_blob = self
            .blob_to_peers
            .entry(hash.clone())
            .or_default()
            .insert(peer);
        self.peer_to_blobs.entry(peer).or_default().insert(hash);
        newly_for_blob
    }

    /// Record that `peer` holds each of `hashes`, returning the subset that was
    /// newly learned. Holds the index once for the whole batch instead of
    /// re-locking per hash (matters for large provider-gossip frames).
    pub fn advertise_many(&mut self, peer: EndpointId, hashes: &[BlobHash]) -> Vec<BlobHash> {
        let mut newly = Vec::new();
        for hash in hashes {
            let new_for_blob = self
                .blob_to_peers
                .entry(hash.clone())
                .or_default()
                .insert(peer);
            self.peer_to_blobs
                .entry(peer)
                .or_default()
                .insert(hash.clone());
            if new_for_blob {
                newly.push(hash.clone());
            }
        }
        newly
    }

    /// Remove all entries for a peer.
    pub fn remove_peer(&mut self, peer: &EndpointId) {
        if let Some(blobs) = self.peer_to_blobs.remove(peer) {
            for hash in &blobs {
                if let Some(peers) = self.blob_to_peers.get_mut(hash) {
                    peers.remove(peer);
                    if peers.is_empty() {
                        self.blob_to_peers.remove(hash);
                    }
                }
            }
        }
    }

    /// Get the set of peers known to have a specific blob.
    pub fn peers_with_blob(&self, hash: &BlobHash) -> Vec<EndpointId> {
        self.blob_to_peers
            .get(hash)
            .map(|peers| peers.iter().copied().collect())
            .unwrap_or_default()
    }

    /// Get the number of blobs known on a specific peer.
    pub fn peer_blob_count(&self, peer: &EndpointId) -> usize {
        self.peer_to_blobs
            .get(peer)
            .map(|blobs| blobs.len())
            .unwrap_or(0)
    }

    /// Total number of unique blob→peer mappings tracked.
    pub fn total_entries(&self) -> usize {
        self.blob_to_peers.values().map(|s| s.len()).sum()
    }
}

// ============================================================================
// PeerHealthIndex - recent per-peer fetch outcomes for peer ordering (#137)
// ============================================================================

/// Priority bucket a peer falls into when ordering fetch attempts.
///
/// Lower discriminant = tried earlier. The numeric values are the sort key
/// used by [`PeerHealthIndex::ready_peers`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum PeerHealthRank {
    /// Most recent attempt succeeded within the cooldown window.
    Healthy = 0,
    /// No record, or last record is older than the cooldown window.
    Neutral = 1,
    /// Most recent attempt stalled (no response within the threshold, or a
    /// stream-open failure) within the cooldown window. NOTE: a peer that
    /// *errored* (reachable but couldn't serve the blob) is NOT recorded
    /// here — that's a content-availability signal, not a health one.
    Unhealthy = 2,
}

/// Whether the most recent per-peer fetch attempt made it to completion.
#[derive(Debug, Clone, Copy)]
enum AttemptHealth {
    Success,
    Failure,
}

#[derive(Debug, Clone, Copy)]
struct PeerHealthRecord {
    health: AttemptHealth,
    at: Instant,
}

/// Tracks the most recent fetch outcome per peer so [`fetch_blob`] can
/// order peers by recent success rather than fixed insertion order.
///
/// Only the *latest* outcome per peer is retained — bounded by the size of
/// `known_peers`. The ordering policy (per #137) is deprioritize-not-skip:
/// a peer that recently stalled is moved behind healthy and untried peers
/// but is still attempted, preserving blob availability when the only
/// holder is a flaky peer. After `cooldown` elapses the peer reverts to
/// [`PeerHealthRank::Neutral`] and gets a fresh first-class attempt.
///
/// # Thread Safety
///
/// Intended to be wrapped in a `TokioRwLock` by its owner.
///
/// This is an internal mechanism with no public accessor (unlike
/// [`BlobPeerIndex`], which is exposed via [`NetworkedIrohBlobStore::blob_peer_index`]),
/// so it is `pub(crate)` rather than `pub`.
///
/// [`fetch_blob`]: NetworkedIrohBlobStore::fetch_blob
#[derive(Debug, Default)]
pub(crate) struct PeerHealthIndex {
    records: HashMap<EndpointId, PeerHealthRecord>,
}

impl PeerHealthIndex {
    /// Create an empty index.
    pub(crate) fn new() -> Self {
        Self::default()
    }

    /// Record the outcome of a per-peer fetch attempt, overwriting any
    /// previous record for that peer.
    fn record(&mut self, peer: EndpointId, health: AttemptHealth, at: Instant) {
        self.records.insert(peer, PeerHealthRecord { health, at });
    }

    /// Drop a peer's health record (called when a peer is removed).
    fn remove_peer(&mut self, peer: &EndpointId) {
        self.records.remove(peer);
    }

    /// Classify a single peer relative to `now` and the `cooldown` window.
    fn rank(&self, peer: &EndpointId, cooldown: Duration, now: Instant) -> PeerHealthRank {
        match self.records.get(peer) {
            // `saturating_duration_since` guards against clock weirdness
            // where `at` is somehow in the future (returns zero → in-window).
            Some(record) if now.saturating_duration_since(record.at) <= cooldown => {
                match record.health {
                    AttemptHealth::Success => PeerHealthRank::Healthy,
                    AttemptHealth::Failure => PeerHealthRank::Unhealthy,
                }
            }
            // No record, or the record has aged out of the cooldown window.
            _ => PeerHealthRank::Neutral,
        }
    }

    /// Stably reorder `peers` by recent health: healthy first, then
    /// untried/expired (neutral), then recently-failed last. The input
    /// order is preserved within each rank, so any upstream prioritization
    /// (e.g. blob-index promotion) survives as the tiebreaker.
    ///
    /// Used by tests to verify health ordering. Production code uses
    /// [`Self::ready_peers`], which additionally filters out within-cooldown
    /// Unhealthy peers to avoid the stall-reset cycle.
    #[cfg(test)]
    fn order_peers(
        &self,
        peers: &[EndpointId],
        cooldown: Duration,
        now: Instant,
    ) -> Vec<EndpointId> {
        let mut ordered = peers.to_vec();
        ordered.sort_by_key(|peer| self.rank(peer, cooldown, now) as u8);
        ordered
    }

    /// Return only the peers that are ready for an immediate fetch attempt
    /// (Healthy, Neutral, or past their cooldown window), in health-priority
    /// order. Within-cooldown Unhealthy peers are excluded.
    ///
    /// `fetch_blob` uses this instead of [`Self::order_peers`] to avoid the
    /// stall-cycle pathology: if a cooling peer were tried (demoted-last),
    /// it would stall again and reset its own cooldown timer, repeating the
    /// cycle N×stall_timeout times (e.g. 6×30 s = 180 s) for what might be
    /// a transient 5-second overload. By skipping cooling peers, `fetch_blob`
    /// returns a transient error immediately, the sweep retries in ~1 s, and
    /// once the cooldown expires the peer gets a single fresh first-class
    /// attempt. If that attempt succeeds, total latency is ~31 s instead of
    /// ~180 s; if it stalls again, the cycle continues but each miss is one
    /// stall + one cooldown rather than N stalls.
    fn ready_peers(
        &self,
        peers: &[EndpointId],
        cooldown: Duration,
        now: Instant,
    ) -> Vec<EndpointId> {
        let mut ready: Vec<EndpointId> = peers
            .iter()
            .filter(|p| self.rank(p, cooldown, now) != PeerHealthRank::Unhealthy)
            .cloned()
            .collect();
        // Stable sort: Healthy (0) before Neutral (1). An expired Unhealthy
        // peer has already reverted to Neutral via `rank()`.
        ready.sort_by_key(|p| self.rank(p, cooldown, now) as u8);
        ready
    }
}

/// Networked Iroh blob store with P2P sync capabilities
///
/// Extends `IrohBlobStore` with:
/// - iroh `Router` accepting QUIC connections for blob transfer
/// - `BlobsProtocol` for serving blobs to peers
/// - `Downloader` for fetching from remote peers
///
/// # Architecture
///
/// ```text
/// NetworkedIrohBlobStore
///     ├─ IrohBlobStore (local storage)
///     ├─ Router (QUIC accept loop + Endpoint)
///     ├─ BlobsProtocol (incoming requests)
///     └─ known_peers (for remote fetch)
/// ```
///
/// # Usage
///
/// ```ignore
/// use peat_mesh::storage::NetworkedIrohBlobStore;
///
/// // Create with networking
/// let store = NetworkedIrohBlobStore::new(blob_dir).await?;
///
/// // Add known peer for remote fetch
/// store.add_peer(peer_endpoint_id).await;
///
/// // Create blob locally (available to peers via BlobsProtocol)
/// let token = store.create_blob_from_bytes(data, metadata).await?;
///
/// // Fetch blob (tries local first, then remote peers)
/// let handle = store.fetch_blob(&token, |progress| {}).await?;
/// ```
pub struct NetworkedIrohBlobStore {
    /// Local blob store
    local_store: IrohBlobStore,
    /// Iroh protocol router — accepts incoming QUIC connections and
    /// routes blobs ALPN to BlobsProtocol. Also owns the Endpoint.
    router: Router,
    /// BlobsProtocol for serving blobs
    blobs_protocol: BlobsProtocol,
    /// Known peers that may have blobs
    known_peers: TokioRwLock<Vec<EndpointId>>,
    /// Reverse index: blob hash → peers that have it, peer → blob hashes
    blob_peer_index: TokioRwLock<BlobPeerIndex>,
    /// Most-recent per-peer fetch outcome, used to deprioritize recently
    /// stalled peers when ordering fetch attempts (peat-mesh #137).
    peer_health: TokioRwLock<PeerHealthIndex>,
    /// In-memory address lookup for adding peer addresses at runtime
    memory_lookup: MemoryLookup,
    /// Timeout for graceful shutdown
    shutdown_timeout: Duration,
    /// Connect timeout for the iroh connection pool (per peer)
    connect_timeout: Duration,
    /// Stall threshold for blob downloads — a fetch attempt is abandoned
    /// only after this duration elapses with NO meaningful activity:
    /// either an offset-advancing `Progress` event, or a discovery /
    /// part-completion event (`TryProvider`, `ProviderFailed`,
    /// `PartComplete`). A live transfer at any rate never expires; a
    /// `Progress(offset)` heartbeat with no byte advancement also does
    /// NOT reset the deadline (defensive against future iroh-blobs or
    /// peer implementations that emit progress without advancing).
    download_stall_timeout: Duration,
    /// Cooldown window for per-peer fetch-health deprioritization. A peer
    /// whose most recent attempt stalled within this window is
    /// tried after healthy and untried peers (peat-mesh #137). See
    /// [`IrohConfig::peer_health_cooldown`](crate::config::IrohConfig::peer_health_cooldown).
    peer_health_cooldown: Duration,
}

/// Outcome of a single-peer download attempt, shared between `fetch_blob`'s
/// multi-candidate loop and `fetch_blob_from_peer`'s single-target attempt.
enum PeerAttempt {
    Success(BlobHandle),
    Stalled,
    Errored(String),
    StreamOpenFailed(String),
}

impl NetworkedIrohBlobStore {
    /// Create a new networked blob store with default configuration.
    ///
    /// Binds an iroh endpoint on an ephemeral port with default relay servers.
    ///
    /// # Arguments
    ///
    /// * `blob_dir` - Directory for blob storage and metadata
    pub async fn new(blob_dir: PathBuf) -> Result<Arc<Self>> {
        Self::from_config(blob_dir, &IrohConfig::default()).await
    }

    /// Create with a specific bind address.
    pub async fn bind(blob_dir: PathBuf, bind_addr: std::net::SocketAddr) -> Result<Arc<Self>> {
        let config = IrohConfig {
            bind_addr: Some(bind_addr),
            ..Default::default()
        };
        Self::from_config(blob_dir, &config).await
    }

    /// Build an Iroh [`Endpoint`] and [`MemoryLookup`] from an [`IrohConfig`]
    /// **without** formation peer gating.
    ///
    /// **Warning:** This creates an unauthenticated endpoint. Production code
    /// must use [`build_endpoint_with_formation_peers`](Self::build_endpoint_with_formation_peers)
    /// to enforce mandatory credential authentication on all connections.
    pub async fn build_endpoint(config: &IrohConfig) -> Result<(Endpoint, MemoryLookup)> {
        Self::build_endpoint_with_hooks(config, None::<NoopEndpointHooks>).await
    }

    /// Build an Iroh [`Endpoint`] and [`MemoryLookup`] with
    /// [`FormationEndpointHooks`] that reject connections from unknown peers.
    ///
    /// This is the required entry point for production use. All QUIC
    /// connections (sync, blobs, etc.) are gated — only peers in the
    /// `formation_peers` set are accepted. Enrollment ALPN is exempted.
    pub async fn build_endpoint_with_formation_peers(
        config: &IrohConfig,
        formation_peers: crate::security::FormationPeerSet,
    ) -> Result<(Endpoint, MemoryLookup)> {
        let hooks = FormationEndpointHooks::new(formation_peers);
        Self::build_endpoint_with_hooks(config, Some(hooks)).await
    }

    /// Build an Iroh [`Endpoint`] and [`MemoryLookup`] with optional
    /// [`EndpointHooks`] for intercepting connection lifecycle events.
    ///
    /// Hooks allow callers to:
    /// - **`before_connect`**: inspect/reject outgoing connections before packets are sent
    /// - **`after_handshake`**: inspect/reject connections after TLS handshake completes
    ///
    /// [`EndpointHooks`]: iroh::endpoint::EndpointHooks
    /// [`MeshSyncTransport`]: super::mesh_sync_transport::MeshSyncTransport
    pub async fn build_endpoint_with_hooks(
        config: &IrohConfig,
        hooks: Option<impl iroh::endpoint::EndpointHooks + 'static>,
    ) -> Result<(Endpoint, MemoryLookup)> {
        let memory_lookup = MemoryLookup::new();

        // peat-mesh#226: this is the other production iroh endpoint (the one the
        // `peat-mesh-node` binary builds for its formation-gated blob/sync
        // surface). Like `AutomergeBackend::with_iroh`, it must apply the same
        // tactical QUIC keepalive + bounded idle-timeout (1s / 5s, Issue #315)
        // that every `IrohTransport` endpoint uses — without it a half-open
        // connection over this endpoint reverts to iroh's ~30s default idle
        // window before detection. Keeps the keepalive guarantee uniform across
        // every peat-mesh-built iroh endpoint.
        let mut builder = relay_policy_builder()
            .address_lookup(memory_lookup.clone())
            .transport_config(crate::network::iroh_transport::create_tactical_transport_config());

        // Install endpoint hooks if provided
        if let Some(hooks) = hooks {
            builder = builder.hooks(hooks);
        }

        // Apply deterministic secret key
        if let Some(key_bytes) = config.secret_key {
            builder = builder.secret_key(SecretKey::from_bytes(&key_bytes));
        }

        // Apply bind address. peat#890: when the caller passed an
        // unspecified address (`0.0.0.0` or `[::]`), iroh's default
        // behavior is to bind to all interfaces and publish every one
        // of them — including docker bridges, tailscale, and stale
        // leases — as candidate dial targets. Run the local-interface
        // filter and bind only to survivors; iroh's `NodeAddr` ends up
        // containing only those, so peers don't try unreachable paths.
        if let Some(bind_addr) = config.bind_addr {
            if bind_addr.ip().is_unspecified() {
                let want_v6 = bind_addr.is_ipv6();
                let sel =
                    super::interface_filter::select_advertise_interfaces(bind_addr.port(), want_v6);
                if sel.bind_addrs.is_empty() {
                    tracing::warn!(
                        "interface filter produced no bindable interfaces; falling back to unspecified bind {bind_addr}"
                    );
                    builder = builder
                        .bind_addr(bind_addr)
                        .map_err(|e| anyhow::anyhow!("Invalid bind address: {}", e))?;
                } else {
                    super::interface_filter::log_selection(&sel);
                    // Replace iroh's pre-configured `0.0.0.0` / `[::]`
                    // defaults so the survivors are the *only* sockets
                    // bound (and therefore the only addrs published).
                    // Pass each interface's actual CIDR prefix so iroh
                    // routes by longest-prefix-match (and doesn't reject
                    // the second bind with "Only a single default address
                    // can be set per IP family").
                    builder = builder.clear_ip_transports();
                    for b in &sel.bind_addrs {
                        builder = builder
                            .bind_addr_with_opts(
                                b.addr,
                                iroh::endpoint::BindOpts::default().set_prefix_len(b.prefix_len),
                            )
                            .map_err(|e| {
                                anyhow::anyhow!(
                                    "Invalid filtered bind address {}/{}: {}",
                                    b.addr,
                                    b.prefix_len,
                                    e
                                )
                            })?;
                    }
                }
            } else {
                builder = builder
                    .bind_addr(bind_addr)
                    .map_err(|e| anyhow::anyhow!("Invalid bind address: {}", e))?;
            }
        }

        // Apply relay configuration
        if !config.relay_urls.is_empty() {
            let relay_map = RelayMap::from_iter(
                config
                    .relay_urls
                    .iter()
                    .filter_map(|url| url.parse::<RelayUrl>().ok()),
            );
            builder = builder.relay_mode(RelayMode::Custom(relay_map));
        }

        let endpoint = tokio::time::timeout(config.bind_timeout, builder.bind())
            .await
            .map_err(|_| {
                anyhow::anyhow!(
                    "Iroh endpoint bind timed out after {:?}",
                    config.bind_timeout
                )
            })?
            .map_err(|e| anyhow::anyhow!("Failed to create iroh endpoint: {}", e))?;

        Ok((endpoint, memory_lookup))
    }

    /// Create a networked blob store from an [`IrohConfig`].
    ///
    /// Applies bind address and relay URL settings from the config when
    /// constructing the iroh endpoint.
    pub async fn from_config(blob_dir: PathBuf, config: &IrohConfig) -> Result<Arc<Self>> {
        let (endpoint, memory_lookup) = Self::build_endpoint(config).await?;
        Self::from_endpoint_with_protocols_with_timeouts(
            blob_dir,
            endpoint,
            memory_lookup,
            vec![],
            config.shutdown_timeout,
            config.connect_timeout,
            config.download_stall_timeout,
            config.peer_health_cooldown,
        )
        .await
    }

    /// Get the endpoint ID (public key) for this node
    pub fn endpoint_id(&self) -> EndpointId {
        self.router.endpoint().id()
    }

    /// Get a reference to the underlying endpoint
    pub fn endpoint(&self) -> &Endpoint {
        self.router.endpoint()
    }

    /// Get a reference to the blobs protocol handler
    pub fn blobs_protocol(&self) -> &BlobsProtocol {
        &self.blobs_protocol
    }

    /// The effective per-attempt download stall threshold this store uses
    /// in [`fetch_blob`](Self::fetch_blob)'s progress watchdog.
    ///
    /// Returns the exact field the watchdog reads, so a propagation test can
    /// pin that a caller-supplied `download_stall_timeout` (e.g. via
    /// `AutomergeBackend::with_iroh`) actually reaches the fetch path and is
    /// not silently re-defaulted at the wrong level (peat-mesh#227).
    pub fn download_stall_timeout(&self) -> Duration {
        self.download_stall_timeout
    }

    /// Get a reference to the local store
    pub fn local_store(&self) -> &IrohBlobStore {
        &self.local_store
    }

    /// Get a reference to the in-memory address lookup provider
    pub fn memory_lookup(&self) -> &MemoryLookup {
        &self.memory_lookup
    }

    /// Create a networked blob store from a pre-built [`Endpoint`], optionally
    /// registering extra protocol handlers alongside the blobs ALPN.
    ///
    /// Use this when you need to share the endpoint with other protocols
    /// (e.g. Automerge sync on `CAP_AUTOMERGE_ALPN`).
    ///
    /// The Router's `spawn()` automatically updates the endpoint's ALPN list
    /// from all registered protocols, so there is no need to pre-configure
    /// ALPNs on the endpoint builder.
    ///
    /// Uses default timeout values. For custom timeouts, use
    /// [`from_endpoint_with_protocols_with_timeouts`](Self::from_endpoint_with_protocols_with_timeouts).
    pub async fn from_endpoint_with_protocols(
        blob_dir: PathBuf,
        endpoint: Endpoint,
        memory_lookup: MemoryLookup,
        extra_protocols: Vec<(&'static [u8], Box<dyn iroh::protocol::DynProtocolHandler>)>,
    ) -> Result<Arc<Self>> {
        let defaults = IrohConfig::default();
        Self::from_endpoint_with_protocols_with_timeouts(
            blob_dir,
            endpoint,
            memory_lookup,
            extra_protocols,
            defaults.shutdown_timeout,
            defaults.connect_timeout,
            defaults.download_stall_timeout,
            defaults.peer_health_cooldown,
        )
        .await
    }

    /// Like [`from_endpoint_with_protocols`](Self::from_endpoint_with_protocols)
    /// but with explicit timeout configuration.
    ///
    /// `connect_timeout` controls how long the connection pool waits to
    /// establish a fresh QUIC connection to a peer. `download_stall_timeout`
    /// controls how long a fetch is allowed to make no progress before the
    /// current peer attempt is abandoned (see field docs).
    /// `peer_health_cooldown` controls how long a recently stalled
    /// peer stays deprioritized in the fetch ordering (peat-mesh #137).
    #[allow(clippy::too_many_arguments)]
    pub async fn from_endpoint_with_protocols_with_timeouts(
        blob_dir: PathBuf,
        endpoint: Endpoint,
        memory_lookup: MemoryLookup,
        extra_protocols: Vec<(&'static [u8], Box<dyn iroh::protocol::DynProtocolHandler>)>,
        shutdown_timeout: Duration,
        connect_timeout: Duration,
        download_stall_timeout: Duration,
        peer_health_cooldown: Duration,
    ) -> Result<Arc<Self>> {
        let local_store = IrohBlobStore::new_in_memory(blob_dir).await?;
        let blobs_protocol = BlobsProtocol::new(&local_store.store, None);

        let mut builder =
            Router::builder(endpoint).accept(iroh_blobs::ALPN, blobs_protocol.clone());

        for (alpn, handler) in extra_protocols {
            builder = builder.accept(alpn, handler);
        }

        let router = builder.spawn();

        Ok(Arc::new(Self {
            local_store,
            router,
            blobs_protocol,
            known_peers: TokioRwLock::new(Vec::new()),
            blob_peer_index: TokioRwLock::new(BlobPeerIndex::new()),
            peer_health: TokioRwLock::new(PeerHealthIndex::new()),
            memory_lookup,
            shutdown_timeout,
            connect_timeout,
            download_stall_timeout,
            peer_health_cooldown,
        }))
    }

    /// Gracefully shut down the protocol router and Iroh endpoint.
    ///
    /// Applies the configured `shutdown_timeout` (default 5s) to prevent
    /// hanging on stuck QUIC connections.
    pub async fn shutdown(&self) -> Result<()> {
        tokio::time::timeout(self.shutdown_timeout, self.router.shutdown())
            .await
            .map_err(|_| {
                anyhow::anyhow!(
                    "Router shutdown timed out after {:?}",
                    self.shutdown_timeout
                )
            })?
            .map_err(|e| anyhow::anyhow!("Router shutdown error: {}", e))
    }

    /// Add a peer that may have blobs we want to fetch
    pub async fn add_peer(&self, peer_id: EndpointId) {
        let mut peers = self.known_peers.write().await;
        if !peers.contains(&peer_id) {
            peers.push(peer_id);
            debug!("Added peer {} for blob fetching", peer_id.fmt_short());
        }
    }

    /// Add a known blob peer from hex-encoded endpoint ID and socket address.
    ///
    /// Convenience wrapper so callers don't need to depend on iroh types
    /// directly. Registers the address in MemoryLookup for QUIC connection
    /// resolution and adds the peer to the known-peers list for fetch_blob.
    /// Get the first bound socket address as `"ip:port"` string.
    /// Returns None if no addresses are bound.
    pub fn bound_addr_string(&self) -> Option<String> {
        let addr = self.router.endpoint().addr();
        addr.addrs.iter().find_map(|a| match a {
            TransportAddr::Ip(sock) => Some(sock.to_string()),
            _ => None,
        })
    }

    /// Parse a hex-encoded endpoint id into an `EndpointId`. Shared by
    /// `add_peer_from_hex`, `add_peer_from_hex_id`, and `fetch_blob_from_peer`.
    fn parse_endpoint_id_hex(endpoint_id_hex: &str) -> anyhow::Result<EndpointId> {
        let bytes = hex::decode(endpoint_id_hex)
            .map_err(|e| anyhow::anyhow!("invalid hex endpoint id: {}", e))?;
        let bytes: [u8; 32] = bytes
            .try_into()
            .map_err(|_| anyhow::anyhow!("endpoint id must be 32 bytes"))?;
        iroh::PublicKey::from_bytes(&bytes)
            .map_err(|e| anyhow::anyhow!("invalid public key: {}", e))
    }

    pub async fn add_peer_from_hex(&self, endpoint_id_hex: &str, addr: &str) -> anyhow::Result<()> {
        let peer_id = Self::parse_endpoint_id_hex(endpoint_id_hex)?;
        let sock_addr: std::net::SocketAddr = addr
            .parse()
            .map_err(|e| anyhow::anyhow!("invalid address: {}", e))?;

        let endpoint_addr = EndpointAddr {
            id: peer_id,
            addrs: std::collections::BTreeSet::from([TransportAddr::Ip(sock_addr)]),
        };
        self.memory_lookup().add_endpoint_info(endpoint_addr);
        self.add_peer(peer_id).await;
        Ok(())
    }

    /// Register a known blob peer by endpoint id (hex) **without** a static
    /// address, so relay/DNS discovery resolves how to reach it.
    ///
    /// Prefer this over [`add_peer_from_hex`] when the peer may be on a
    /// different network/NAT and reachable only via a relay (the
    /// `relay-n0-hosted` feature or configured `relay_urls`): pinning a single
    /// direct `TransportAddr::Ip` gives the connector one unroutable candidate
    /// and no path to fall back to the relay. With id-only registration the
    /// endpoint's discovery service supplies the reachable addresses (including
    /// the relay URL) at connect time.
    pub async fn add_peer_from_hex_id(&self, endpoint_id_hex: &str) -> anyhow::Result<()> {
        let peer_id = Self::parse_endpoint_id_hex(endpoint_id_hex)?;
        self.add_peer(peer_id).await;
        Ok(())
    }

    /// Attempt exactly one download of `token` from `peer_id` using `downloader`.
    /// Shared by `fetch_blob`'s multi-candidate loop and `fetch_blob_from_peer`'s
    /// single-target attempt. Does not decide whether to retry another peer —
    /// callers own that policy. An `Err` return means an infrastructure failure
    /// *after* a successful download (export/metadata-save) and should propagate
    /// immediately rather than be treated as "try the next peer", matching
    /// `fetch_blob`'s pre-extraction behavior.
    async fn attempt_download_from_peer<F>(
        &self,
        token: &BlobToken,
        iroh_hash: Hash,
        peer_id: EndpointId,
        downloader: &iroh_blobs::api::downloader::Downloader,
        progress: &mut F,
    ) -> Result<PeerAttempt>
    where
        F: FnMut(BlobProgress) + Send + 'static,
    {
        debug!(
            "Trying peer {} for blob {}",
            peer_id.fmt_short(),
            token.hash.as_hex()
        );

        progress(BlobProgress::Downloading {
            downloaded_bytes: 0,
            total_bytes: token.size_bytes,
        });

        let progress_stream = match downloader.download(iroh_hash, Some(peer_id)).stream().await {
            Ok(s) => s,
            Err(e) => {
                warn!(
                    "Failed to open download stream from peer {}: {}",
                    peer_id.fmt_short(),
                    e
                );
                self.peer_health.write().await.record(
                    peer_id,
                    AttemptHealth::Failure,
                    Instant::now(),
                );
                return Ok(PeerAttempt::StreamOpenFailed(e.to_string()));
            }
        };

        let outcome = drive_attempt_with_stall_watchdog(
            progress_stream,
            self.download_stall_timeout,
            |offset| {
                progress(BlobProgress::Downloading {
                    downloaded_bytes: offset,
                    total_bytes: token.size_bytes,
                });
            },
        )
        .await;

        match outcome {
            AttemptOutcome::Stalled => {
                warn!(
                    "Download from peer {} stalled — no progress in {:?}",
                    peer_id.fmt_short(),
                    self.download_stall_timeout
                );
                self.peer_health.write().await.record(
                    peer_id,
                    AttemptHealth::Failure,
                    Instant::now(),
                );
                Ok(PeerAttempt::Stalled)
            }
            AttemptOutcome::Errored(msg) => {
                warn!(
                    "Download from peer {} reported error: {}",
                    peer_id.fmt_short(),
                    msg
                );
                // Do NOT demote on an errored attempt (peat-mesh #137 follow-up):
                // `Errored` means the peer was reachable but the content wasn't
                // available yet — a transient content-availability signal, not a
                // peer-health signal.
                Ok(PeerAttempt::Errored(msg))
            }
            AttemptOutcome::Completed => {
                info!(
                    "Successfully downloaded blob {} from peer {}",
                    token.hash.as_hex(),
                    peer_id.fmt_short()
                );

                self.peer_health.write().await.record(
                    peer_id,
                    AttemptHealth::Success,
                    Instant::now(),
                );

                self.blob_peer_index
                    .write()
                    .await
                    .advertise(peer_id, token.hash.clone());

                self.announce_local_holding(token.hash.clone()).await;

                let export_path = self.local_store.export_blob(&iroh_hash).await?;

                let sidecar = SidecarMetadata::new(token.metadata.clone(), token.size_bytes);
                self.local_store.save_metadata(&token.hash, &sidecar)?;

                self.local_store.cache_token(token);

                progress(BlobProgress::Completed {
                    local_path: export_path.clone(),
                });

                Ok(PeerAttempt::Success(BlobHandle::new(
                    token.clone(),
                    export_path,
                )))
            }
        }
    }

    /// Fetch `token` from exactly `peer_id_hex`, bypassing the automatic
    /// candidate list (`known_peers` ∪ `blob_peer_index`) and the
    /// `PeerHealthIndex` readiness/cooldown filter — the caller has explicitly
    /// chosen this peer, so there is no fallback to any other known peer on
    /// failure (peat#1013 expanded scope: direct, pull-only, targeted fetch,
    /// independent of document sync).
    ///
    /// `peer_id_hex` must already be registered via [`add_peer`](Self::add_peer),
    /// [`add_peer_from_hex`](Self::add_peer_from_hex), or
    /// [`add_peer_from_hex_id`](Self::add_peer_from_hex_id) — this method does
    /// not implicitly register it.
    ///
    /// On success, still records into `peer_health` and `blob_peer_index`: a
    /// successful direct fetch is real signal worth feeding back into future
    /// automatic `fetch_blob` calls. On a stalled/stream-open failure, still
    /// records a health failure (matching `fetch_blob`'s "errored ≠ demote"
    /// policy for `Errored`).
    ///
    /// Every failure's error message contains the literal substring
    /// `"direct fetch"`, distinguishing it from `fetch_blob`'s multi-candidate
    /// exhaustion errors for callers that want to branch on it (e.g. peat-ffi).
    pub async fn fetch_blob_from_peer<F>(
        &self,
        token: &BlobToken,
        peer_id_hex: &str,
        mut progress: F,
    ) -> Result<BlobHandle>
    where
        F: FnMut(BlobProgress) + Send + 'static,
    {
        // Local-first short-circuit, mirroring fetch_blob's.
        let local_path = self.local_store.local_blob_path(&token.hash);
        if local_path.exists() {
            progress(BlobProgress::Completed {
                local_path: local_path.clone(),
            });
            return Ok(BlobHandle::new(token.clone(), local_path));
        }
        let iroh_hash = IrohBlobStore::blob_hash_to_iroh_hash(&token.hash)?;
        if self.local_store.store.has(iroh_hash).await? {
            return self.local_store.fetch_blob(token, progress).await;
        }

        let peer_id = Self::parse_endpoint_id_hex(peer_id_hex)?;

        if !self.known_peers.read().await.contains(&peer_id) {
            let msg = format!(
                "direct fetch of blob {} from peer {} failed: peer is not registered — \
                 call add_peer_from_hex or add_peer_from_hex_id first",
                token.hash.as_hex(),
                peer_id.fmt_short(),
            );
            progress(BlobProgress::Failed { error: msg.clone() });
            return Err(anyhow::anyhow!(msg));
        }

        progress(BlobProgress::Started {
            total_bytes: token.size_bytes,
        });

        let pool_opts = iroh_blobs::util::connection_pool::Options {
            connect_timeout: self.connect_timeout,
            ..Default::default()
        };
        let downloader = iroh_blobs::api::downloader::Downloader::new_with_opts(
            self.store_api(),
            self.router.endpoint(),
            pool_opts,
        );

        let attempt = self
            .attempt_download_from_peer(token, iroh_hash, peer_id, &downloader, &mut progress)
            .await;

        match attempt {
            Ok(PeerAttempt::Success(handle)) => Ok(handle),
            Ok(PeerAttempt::Stalled) => {
                let msg = format!(
                    "direct fetch of blob {} from peer {} failed: stalled (no progress within {:?})",
                    token.hash.as_hex(),
                    peer_id.fmt_short(),
                    self.download_stall_timeout,
                );
                progress(BlobProgress::Failed { error: msg.clone() });
                Err(anyhow::anyhow!(msg))
            }
            Ok(PeerAttempt::Errored(e)) => {
                let msg = format!(
                    "direct fetch of blob {} from peer {} failed: {}",
                    token.hash.as_hex(),
                    peer_id.fmt_short(),
                    e
                );
                progress(BlobProgress::Failed { error: msg.clone() });
                Err(anyhow::anyhow!(msg))
            }
            Ok(PeerAttempt::StreamOpenFailed(e)) => {
                let msg = format!(
                    "direct fetch of blob {} from peer {} failed: could not open stream: {}",
                    token.hash.as_hex(),
                    peer_id.fmt_short(),
                    e
                );
                progress(BlobProgress::Failed { error: msg.clone() });
                Err(anyhow::anyhow!(msg))
            }
            Err(e) => {
                let msg = format!(
                    "direct fetch of blob {} from peer {} failed: {}",
                    token.hash.as_hex(),
                    peer_id.fmt_short(),
                    e
                );
                progress(BlobProgress::Failed { error: msg.clone() });
                Err(anyhow::anyhow!(msg))
            }
        }
    }

    /// Reset a peer's fetch-health record without removing it from the known-peers
    /// list or the blob-peer index.
    ///
    /// Call this after successfully reconnecting to a peer that was previously
    /// demoted by the `PeerHealthIndex` (e.g. after the auto-reconnect watchdog
    /// re-dials a peer whose QUIC session had idled out). Without this call the
    /// peer's `Unhealthy` record persists for up to `peer_health_cooldown` after
    /// reconnect, causing `fetch_blob` to continue deprioritising a peer that is
    /// now reachable — the worst case is that a freshly-recovered C2 gets ignored
    /// for the full cooldown window (default 30 s) even though it is the closest
    /// holder of every blob. Resetting to `Neutral` gives the peer a fresh
    /// first-class attempt on the next `fetch_blob` call.
    ///
    /// Unlike [`remove_peer`](Self::remove_peer), this does **not** evict the peer
    /// from `known_peers` or the `BlobPeerIndex` — blob-location knowledge
    /// accumulated before the blackout is preserved.
    pub async fn reset_peer_health(&self, peer_id: &EndpointId) {
        self.peer_health.write().await.remove_peer(peer_id);
        debug!(
            peer = %peer_id.fmt_short(),
            "reset peer health record after reconnect"
        );
    }

    /// Remove a peer from the known peers list and clean up its index entries.
    pub async fn remove_peer(&self, peer_id: &EndpointId) {
        let mut peers = self.known_peers.write().await;
        peers.retain(|p| p != peer_id);
        drop(peers);

        let mut index = self.blob_peer_index.write().await;
        index.remove_peer(peer_id);
        drop(index);

        self.peer_health.write().await.remove_peer(peer_id);
    }

    /// List known peers
    pub async fn known_peers(&self) -> Vec<EndpointId> {
        self.known_peers.read().await.clone()
    }

    /// Record that a peer has a specific blob.
    ///
    /// Call this when you learn (via gossip, sync, or direct query) that
    /// a peer has a blob. Enables O(1) peer lookup in [`fetch_blob`](Self::fetch_blob).
    ///
    /// Returns `true` if this `(peer, hash)` pairing was newly learned.
    pub async fn advertise_blob(&self, peer: EndpointId, hash: BlobHash) -> bool {
        self.blob_peer_index.write().await.advertise(peer, hash)
    }

    /// Announce to our peers that *this node* holds `hash` (provider gossip,
    /// peat-node#170). Best-effort and off the hot path — the actual sends are
    /// each their own task; peers that don't speak `peat/blob-announce/1` just
    /// ignore us. Awaiting only snapshots the peer list.
    pub async fn announce_local_holding(&self, hash: BlobHash) {
        let peers = self.known_peers().await;
        if peers.is_empty() {
            return;
        }
        let announce = self.local_announce(vec![hash]);
        blob_announce::spawn_announce(self.endpoint().clone(), peers, announce);
    }

    /// Build an announce asserting this node holds `hashes`, carrying our own
    /// reachable addresses so a receiver can dial us directly.
    fn local_announce(&self, hashes: Vec<BlobHash>) -> blob_announce::BlobAnnounce {
        let endpoint = self.endpoint();
        let addrs: Vec<String> = endpoint
            .addr()
            .addrs
            .iter()
            .filter_map(|a| match a {
                TransportAddr::Ip(sock) => Some(sock.to_string()),
                _ => None,
            })
            .collect();
        blob_announce::BlobAnnounce {
            holder: endpoint.id(),
            addrs,
            hashes,
            ttl: blob_announce::DEFAULT_ANNOUNCE_TTL,
        }
    }

    /// Apply a received provider-gossip announce: record each announced blob in
    /// the provider index, learn the holder's addresses *only when we heard it
    /// from the holder itself*, and re-broadcast any *newly-learned* holdings to
    /// other peers (TTL-bounded).
    ///
    /// Trust is gated by who delivered the announce (`from`) versus the claimed
    /// `holder` — see [`blob_announce::classify_announce`]. A forged origin claim
    /// (full-TTL announce for a holder that isn't the sender) is dropped; a
    /// relayed third-party holding is recorded for discovery but its addresses
    /// are not trusted. Re-broadcasting only newly-learned holdings makes the
    /// epidemic terminate once a region converges instead of flooding.
    pub async fn apply_announce(&self, announce: blob_announce::BlobAnnounce, from: EndpointId) {
        let trust = blob_announce::classify_announce(announce.holder, from, announce.ttl);
        if trust == blob_announce::AnnounceTrust::Reject {
            warn!(
                holder = %announce.holder.fmt_short(),
                from = %from.fmt_short(),
                "rejecting forged provider announce (originated claim for a different holder)"
            );
            return;
        }

        // Learn the holder's addresses ONLY from a first-party announce — a
        // relay must not be able to inject dialing hints for a third party
        // (ADR-0006 defense-in-depth). The holder's real address arrives via its
        // own first-party announce to a direct peer.
        if trust == blob_announce::AnnounceTrust::HoldingsAndAddrs {
            let addrs: std::collections::BTreeSet<TransportAddr> = announce
                .addrs
                .iter()
                .filter_map(|s| s.parse::<std::net::SocketAddr>().ok())
                .map(TransportAddr::Ip)
                .collect();
            if !addrs.is_empty() {
                self.memory_lookup().add_endpoint_info(iroh::EndpointAddr {
                    id: announce.holder,
                    addrs,
                });
            }
        }

        // Drop hashes that aren't well-formed blob hashes so a hostile peer
        // can't inflate the index with arbitrary garbage keys.
        let valid: Vec<BlobHash> = announce
            .hashes
            .iter()
            .filter(|h| is_plausible_blob_hash(h))
            .cloned()
            .collect();
        if valid.is_empty() {
            return;
        }

        // Record holdings in one lock acquisition; keep only the newly-learned.
        let newly_learned = self.advertise_blobs(announce.holder, &valid).await;
        if newly_learned.is_empty() {
            return;
        }

        // Re-broadcast only the newly-learned holdings (TTL-bounded), excluding
        // the peer we heard it from and the holder itself. The holder is
        // preserved so dedup keys on the original (holder, hash) across hops.
        if let Some(relayed) = announce.relayed() {
            let relayed = blob_announce::BlobAnnounce {
                hashes: newly_learned,
                ..relayed
            };
            let exclude_holder = announce.holder;
            let peers: Vec<EndpointId> = self
                .known_peers()
                .await
                .into_iter()
                .filter(|p| *p != from && *p != exclude_holder)
                .collect();
            if !peers.is_empty() {
                blob_announce::spawn_announce(self.endpoint().clone(), peers, relayed);
            }
        }
    }

    /// Record that `peer` holds each of `hashes`; returns the newly-learned
    /// subset. Batches the index write (one lock for the whole frame).
    async fn advertise_blobs(&self, peer: EndpointId, hashes: &[BlobHash]) -> Vec<BlobHash> {
        self.blob_peer_index
            .write()
            .await
            .advertise_many(peer, hashes)
    }

    /// Get peers known to have a specific blob.
    ///
    /// Returns an empty vec if no peer info is known for this blob.
    pub async fn peers_with_blob(&self, hash: &BlobHash) -> Vec<EndpointId> {
        self.blob_peer_index.read().await.peers_with_blob(hash)
    }

    /// Get a reference to the blob peer index.
    pub async fn blob_peer_index(&self) -> tokio::sync::RwLockReadGuard<'_, BlobPeerIndex> {
        self.blob_peer_index.read().await
    }

    /// Get the iroh-blobs Store API for advanced operations
    pub fn store_api(&self) -> &iroh_blobs::api::Store {
        &self.local_store.store
    }
}

#[async_trait::async_trait]
impl BlobStore for NetworkedIrohBlobStore {
    async fn create_blob(&self, path: &Path, metadata: BlobMetadata) -> Result<BlobToken> {
        // Delegate to local store - blob becomes available for P2P via BlobsProtocol
        self.local_store.create_blob(path, metadata).await
    }

    async fn create_blob_from_stream(
        &self,
        stream: &mut (dyn tokio::io::AsyncRead + Send + Unpin),
        expected_size: Option<u64>,
        metadata: BlobMetadata,
    ) -> Result<BlobToken> {
        self.local_store
            .create_blob_from_stream(stream, expected_size, metadata)
            .await
    }

    async fn create_blob_from_bytes(
        &self,
        data: &[u8],
        metadata: BlobMetadata,
    ) -> Result<BlobToken> {
        // Delegate to local store
        self.local_store
            .create_blob_from_bytes(data, metadata)
            .await
    }

    async fn fetch_blob<F>(&self, token: &BlobToken, mut progress: F) -> Result<BlobHandle>
    where
        F: FnMut(BlobProgress) + Send + 'static,
    {
        info!(
            "NetworkedIrohBlobStore: Fetching blob {}",
            token.hash.as_hex()
        );

        // First, try local fetch
        let local_path = self.local_store.local_blob_path(&token.hash);
        if local_path.exists() {
            debug!("Blob exists locally at {:?}", local_path);
            progress(BlobProgress::Completed {
                local_path: local_path.clone(),
            });
            return Ok(BlobHandle::new(token.clone(), local_path));
        }

        // Check if in local store (not yet exported)
        let iroh_hash = IrohBlobStore::blob_hash_to_iroh_hash(&token.hash)?;
        if self.local_store.store.has(iroh_hash).await? {
            debug!("Blob in local store, exporting");
            return self.local_store.fetch_blob(token, progress).await;
        }

        // Not available locally - try to fetch from known peers
        progress(BlobProgress::Started {
            total_bytes: token.size_bytes,
        });

        // Build ordered peer list: indexed peers first, then remaining known peers
        let indexed_peers = self
            .blob_peer_index
            .read()
            .await
            .peers_with_blob(&token.hash);
        let all_peers = self.known_peers.read().await.clone();

        if all_peers.is_empty() {
            progress(BlobProgress::Failed {
                error: format!(
                    "Blob {} not available locally and no peers known",
                    token.hash
                ),
            });
            return Err(anyhow::anyhow!(
                "Blob {} not available locally and no peers configured for remote fetch",
                token.hash.as_hex()
            ));
        }

        // Prioritize peers from the index, then try the rest
        let mut base_order = indexed_peers.clone();
        for peer in &all_peers {
            if !base_order.contains(peer) {
                base_order.push(*peer);
            }
        }

        // Partition peers into ready (Healthy / Neutral / cooldown-expired)
        // and cooling (within-cooldown Unhealthy). Only ready peers are
        // attempted this call; cooling peers are skipped.
        //
        // Background: peat-mesh #137 introduced "demote-not-skip" so a
        // stalled peer is still tried (last) when it is the sole blob holder.
        // That invariant causes a stall-reset cycle when a peer is
        // *temporarily* overloaded: every sweep tick tries the cooling peer
        // last, it stalls for download_stall_timeout (30 s), resets its own
        // cooldown, and the 30 s cycle repeats up to N times (observed: 6 ×
        // 30 s = 180 s for a transient congestion spike). With the partition
        // approach the first stall marks the peer cooling, subsequent sweep
        // ticks skip it and return Err immediately, and once the cooldown
        // window expires the peer is Neutral again and gets a single fresh
        // attempt. If the overload resolved (common), delivery completes in
        // ~31 s (1 stall + 1 s sweep tick) instead of ~180 s.
        //
        // Single-source robustness: if the cooling peer is the only blob
        // holder, `fetch_blob` returns Err on every sweep tick until the
        // 30 s window expires, then tries once. That is one stall per 31 s
        // cycle — identical worst-case latency to the old design but without
        // the pathological cycle.
        let (ready_peers, cooling_count) = {
            let health_guard = self.peer_health.read().await;
            let ready =
                health_guard.ready_peers(&base_order, self.peer_health_cooldown, Instant::now());
            let cooling = base_order.len() - ready.len();
            (ready, cooling)
        };

        if !indexed_peers.is_empty() {
            debug!(
                "Blob {} indexed on {} peers, {} total known",
                token.hash.as_hex(),
                indexed_peers.len(),
                all_peers.len()
            );
        }

        info!(
            "Attempting to fetch blob {} from {}/{} ready peers ({} cooling)",
            token.hash.as_hex(),
            ready_peers.len(),
            base_order.len(),
            cooling_count,
        );

        // Use the downloader to fetch from peers. The pool's connect_timeout
        // bounds how long we wait to establish a fresh QUIC connection;
        // the per-peer attempt itself is bounded by a progress-stall
        // watchdog below (the wall-clock cap that used to wrap this call
        // broke transfers whose natural duration exceeded the cap on slow
        // links — see peat-mesh #121).
        let pool_opts = iroh_blobs::util::connection_pool::Options {
            connect_timeout: self.connect_timeout,
            ..Default::default()
        };
        let downloader = iroh_blobs::api::downloader::Downloader::new_with_opts(
            self.store_api(),
            self.router.endpoint(),
            pool_opts,
        );

        // Per-peer attempt logic (stream open, stall watchdog, success
        // bookkeeping) is shared with `fetch_blob_from_peer` via
        // `attempt_download_from_peer`. `Failed` progress/return-Err on
        // exhaustion is handled below, once, after the loop — NOT per
        // attempt — see that method's doc comment for why.
        for peer_id in &ready_peers {
            match self
                .attempt_download_from_peer(token, iroh_hash, *peer_id, &downloader, &mut progress)
                .await?
            {
                PeerAttempt::Success(handle) => return Ok(handle),
                PeerAttempt::Stalled
                | PeerAttempt::Errored(_)
                | PeerAttempt::StreamOpenFailed(_) => {
                    continue;
                }
            }
        }

        // If cooling peers were excluded, return a transient error so the
        // sweep retries in ~1 s. Once their cooldown window expires they
        // become Neutral and get a fresh first-class attempt.
        if cooling_count > 0 {
            progress(BlobProgress::Failed {
                error: format!(
                    "Blob {} not available from {} ready peer(s); {} peer(s) cooling off",
                    token.hash,
                    ready_peers.len(),
                    cooling_count,
                ),
            });
            return Err(anyhow::anyhow!(
                "Blob {} not fetched: {} ready peer(s) tried, {} cooling off (sweep will retry)",
                token.hash.as_hex(),
                ready_peers.len(),
                cooling_count,
            ));
        }

        // All peers failed and none are cooling.
        progress(BlobProgress::Failed {
            error: format!(
                "Failed to fetch blob {} from all {} peers",
                token.hash,
                ready_peers.len()
            ),
        });

        Err(anyhow::anyhow!(
            "Failed to fetch blob {} from any of {} known peers",
            token.hash.as_hex(),
            ready_peers.len()
        ))
    }

    fn blob_exists_locally(&self, hash: &BlobHash) -> bool {
        self.local_store.blob_exists_locally(hash)
    }

    fn blob_info(&self, hash: &BlobHash) -> Option<BlobToken> {
        self.local_store.blob_info(hash)
    }

    async fn delete_blob(&self, hash: &BlobHash) -> Result<()> {
        self.local_store.delete_blob(hash).await
    }

    fn list_local_blobs(&self) -> Vec<BlobToken> {
        self.local_store.list_local_blobs()
    }

    fn local_storage_bytes(&self) -> u64 {
        self.local_store.local_storage_bytes()
    }
}

// ============================================================================
// Stall watchdog
// ============================================================================

/// Outcome of driving one per-peer download attempt to completion.
#[derive(Debug)]
enum AttemptOutcome {
    /// Stream ended without surfacing an error item.
    Completed,
    /// No event arrived within the stall threshold.
    Stalled,
    /// Stream surfaced an `Error` or `DownloadError` item.
    Errored(String),
}

/// Drive a downloader progress stream with a stall watchdog.
///
/// Returns when:
/// - The stream ends without error (→ `Completed`)
/// - An `Error` or `DownloadError` item is observed (→ `Errored`)
/// - No *meaningful activity* arrives within `stall_timeout` (→ `Stalled`)
///
/// "Meaningful activity" is either (a) a `Progress(offset)` event whose
/// offset advanced past the previous one, or (b) a discovery /
/// part-completion event (`TryProvider`, `ProviderFailed`,
/// `PartComplete`). A `Progress(offset)` heartbeat with no byte
/// advancement is consumed from the stream but does NOT reset the
/// deadline — this defends against future iroh-blobs revisions or
/// misbehaving peer implementations that emit `Progress` events
/// without actually transferring data.
///
/// `on_progress` is invoked for each offset-advancing `Progress` event
/// with the cumulative bytes received.
async fn drive_attempt_with_stall_watchdog<S, F>(
    stream: S,
    stall_timeout: Duration,
    mut on_progress: F,
) -> AttemptOutcome
where
    S: futures::Stream<Item = iroh_blobs::api::downloader::DownloadProgressItem> + Unpin,
    F: FnMut(u64),
{
    let mut stream = stream;
    let mut last_offset: u64 = 0;
    let mut deadline = tokio::time::Instant::now() + stall_timeout;

    loop {
        let now = tokio::time::Instant::now();
        if now >= deadline {
            return AttemptOutcome::Stalled;
        }
        let remaining = deadline - now;

        match tokio::time::timeout(remaining, stream.next()).await {
            Err(_elapsed) => return AttemptOutcome::Stalled,
            Ok(None) => return AttemptOutcome::Completed,
            Ok(Some(item)) => match item {
                iroh_blobs::api::downloader::DownloadProgressItem::Error(e) => {
                    return AttemptOutcome::Errored(format!("{}", e));
                }
                iroh_blobs::api::downloader::DownloadProgressItem::DownloadError => {
                    return AttemptOutcome::Errored("DownloadError".to_string());
                }
                iroh_blobs::api::downloader::DownloadProgressItem::Progress(offset) => {
                    if offset > last_offset {
                        last_offset = offset;
                        deadline = tokio::time::Instant::now() + stall_timeout;
                        on_progress(offset);
                    }
                    // else: heartbeat with no advancement — consumed but
                    // does not reset the deadline.
                }
                // Discovery / part-completion events count as activity:
                // the peer is reachable and the downloader is engaged
                // even if no bytes have flowed yet.
                iroh_blobs::api::downloader::DownloadProgressItem::TryProvider { .. }
                | iroh_blobs::api::downloader::DownloadProgressItem::ProviderFailed { .. }
                | iroh_blobs::api::downloader::DownloadProgressItem::PartComplete { .. } => {
                    deadline = tokio::time::Instant::now() + stall_timeout;
                }
            },
        }
    }
}

#[cfg(test)]
mod stall_watchdog_tests {
    use super::*;
    use iroh_blobs::api::downloader::DownloadProgressItem;
    use std::sync::{Arc, Mutex};
    use tokio_stream::wrappers::ReceiverStream;

    /// A stream that emits `Progress(50)`, `Progress(100)`, then closes
    /// must complete and deliver both offsets to `on_progress`.
    #[tokio::test]
    async fn completes_on_clean_stream_end() {
        let (tx, rx) = tokio::sync::mpsc::channel::<DownloadProgressItem>(8);
        tx.send(DownloadProgressItem::Progress(50)).await.unwrap();
        tx.send(DownloadProgressItem::Progress(100)).await.unwrap();
        drop(tx);

        let seen = Arc::new(Mutex::new(Vec::new()));
        let seen_clone = Arc::clone(&seen);
        let outcome = drive_attempt_with_stall_watchdog(
            ReceiverStream::new(rx),
            Duration::from_secs(60),
            move |offset| seen_clone.lock().unwrap().push(offset),
        )
        .await;
        assert!(matches!(outcome, AttemptOutcome::Completed));
        assert_eq!(*seen.lock().unwrap(), vec![50, 100]);
    }

    /// A stream that never emits anything must stall after the threshold.
    #[tokio::test]
    async fn stalls_when_no_events() {
        let (_tx, rx) = tokio::sync::mpsc::channel::<DownloadProgressItem>(8);
        // Hold the sender alive so the stream stays open but emits nothing.
        let outcome = drive_attempt_with_stall_watchdog(
            ReceiverStream::new(rx),
            Duration::from_millis(50),
            |_| {},
        )
        .await;
        assert!(matches!(outcome, AttemptOutcome::Stalled));
    }

    /// THE CORE INVARIANT: a stream emitting events at a rate JUST under
    /// the stall threshold must keep going indefinitely. This is the
    /// peat-mesh #121 regression — under the old wall-clock cap, a slow
    /// transfer would be killed even while making steady progress.
    #[tokio::test(start_paused = true)]
    async fn live_transfer_never_stalls() {
        let (tx, rx) = tokio::sync::mpsc::channel::<DownloadProgressItem>(8);
        let total_progress = Arc::new(Mutex::new(0u64));
        let total_clone = Arc::clone(&total_progress);

        // Send 20 progress events at 60ms cadence (well under the 200ms
        // stall threshold), totalling > 1s of simulated wall-clock — far
        // longer than the old 30s timeout would have allowed at the
        // analogous scale.
        let producer = tokio::spawn(async move {
            for i in 1..=20 {
                tokio::time::sleep(Duration::from_millis(60)).await;
                tx.send(DownloadProgressItem::Progress(i * 100))
                    .await
                    .unwrap();
            }
            drop(tx);
        });

        let outcome = drive_attempt_with_stall_watchdog(
            ReceiverStream::new(rx),
            Duration::from_millis(200),
            move |offset| *total_clone.lock().unwrap() = offset,
        )
        .await;
        producer.await.unwrap();

        assert!(
            matches!(outcome, AttemptOutcome::Completed),
            "live transfer must complete, not stall; got {:?}",
            outcome
        );
        assert_eq!(*total_progress.lock().unwrap(), 2000);
    }

    /// A stream that emits `Error` surfaces as `Errored` immediately,
    /// regardless of how many `Progress` events preceded it.
    #[tokio::test]
    async fn error_item_terminates_with_errored() {
        let (tx, rx) = tokio::sync::mpsc::channel::<DownloadProgressItem>(8);
        tx.send(DownloadProgressItem::Progress(10)).await.unwrap();
        tx.send(DownloadProgressItem::DownloadError).await.unwrap();
        drop(tx);

        let outcome = drive_attempt_with_stall_watchdog(
            ReceiverStream::new(rx),
            Duration::from_secs(60),
            |_| {},
        )
        .await;
        assert!(matches!(outcome, AttemptOutcome::Errored(_)));
    }

    /// Discovery events (`TryProvider`, `ProviderFailed`, `PartComplete`)
    /// count as activity and reset the stall timer, even with no Progress.
    #[tokio::test(start_paused = true)]
    async fn discovery_events_count_as_activity() {
        use std::sync::Arc as StdArc;
        let (tx, rx) = tokio::sync::mpsc::channel::<DownloadProgressItem>(8);

        let producer = tokio::spawn(async move {
            // Stream of TryProvider events at 60ms cadence under 200ms
            // stall threshold. No Progress events at all — this verifies
            // the watchdog treats discovery as activity.
            for _ in 0..5 {
                tokio::time::sleep(Duration::from_millis(60)).await;
                let req: StdArc<iroh_blobs::protocol::GetRequest> = StdArc::new(
                    iroh_blobs::protocol::GetRequest::blob(iroh_blobs::Hash::EMPTY),
                );
                tx.send(DownloadProgressItem::TryProvider {
                    id: iroh::EndpointId::from_bytes(&[0u8; 32]).unwrap(),
                    request: req,
                })
                .await
                .unwrap();
            }
            drop(tx);
        });

        let outcome = drive_attempt_with_stall_watchdog(
            ReceiverStream::new(rx),
            Duration::from_millis(200),
            |_| {},
        )
        .await;
        producer.await.unwrap();
        assert!(
            matches!(outcome, AttemptOutcome::Completed),
            "discovery activity must keep the watchdog alive; got {:?}",
            outcome
        );
    }

    /// A `Progress(offset)` heartbeat that never advances the offset
    /// must NOT keep the watchdog alive. Catches the chatty-but-stuck
    /// failure mode flagged in the PR #122 review: a future iroh-blobs
    /// revision or misbehaving peer could emit progress events without
    /// transferring data, and the watchdog should still trigger.
    #[tokio::test(start_paused = true)]
    async fn chatty_but_byte_stuck_stream_stalls() {
        let (tx, rx) = tokio::sync::mpsc::channel::<DownloadProgressItem>(8);

        // Hold tx in the producer task so the stream stays open; emit
        // Progress(500) repeatedly at 60ms cadence (well under 200ms
        // stall threshold) but offset never advances.
        let producer = tokio::spawn(async move {
            for _ in 0..20 {
                tokio::time::sleep(Duration::from_millis(60)).await;
                if tx.send(DownloadProgressItem::Progress(500)).await.is_err() {
                    break;
                }
            }
            drop(tx);
        });

        let progress_calls = Arc::new(Mutex::new(0u32));
        let progress_clone = Arc::clone(&progress_calls);
        let outcome = drive_attempt_with_stall_watchdog(
            ReceiverStream::new(rx),
            Duration::from_millis(200),
            move |_| *progress_clone.lock().unwrap() += 1,
        )
        .await;
        producer.abort();
        let _ = producer.await;

        assert!(
            matches!(outcome, AttemptOutcome::Stalled),
            "non-advancing Progress heartbeats must NOT keep the watchdog alive; got {:?}",
            outcome
        );
        // The first Progress(500) advances last_offset from 0 to 500
        // (callback fires once); subsequent Progress(500)s do not
        // advance, so the callback must fire exactly once before the
        // watchdog trips.
        assert_eq!(
            *progress_calls.lock().unwrap(),
            1,
            "on_progress should fire only on offset-advancing events"
        );
    }
}

// ============================================================================
// Tests
// ============================================================================

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

    /// FIPS-posture structural pin (peat#923 / peat-mesh#158 QA Warning 3).
    ///
    /// Calls through `crate::security::tls_provider::iroh_quic_provider()`,
    /// the helper that every iroh `Endpoint::builder()` site in peat-mesh
    /// uses. The helper only compiles when peat-mesh's `rustls` feature
    /// set includes `aws-lc-rs`, so a future change that drops that feature
    /// or flips it back to `ring` stops this test from compiling — a loud
    /// build failure rather than a silent FIPS regression.
    ///
    /// Installing the provider as the process default also confirms it
    /// has the symbols rustls expects at runtime (not just at compile
    /// time). The `Arc::unwrap_or_clone` extracts the inner `CryptoProvider`
    /// because `install_default` takes the value, not the Arc. The result
    /// is intentionally ignored: another test in the same process may have
    /// already installed the same provider.
    #[test]
    fn aws_lc_rs_provider_is_linked() {
        let provider = crate::security::tls_provider::iroh_quic_provider();
        let inner = std::sync::Arc::unwrap_or_clone(provider);
        let _ = inner.install_default();
    }

    async fn create_test_store() -> (IrohBlobStore, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let store = IrohBlobStore::new_in_memory(temp_dir.path().to_path_buf())
            .await
            .unwrap();
        (store, temp_dir)
    }

    #[tokio::test]
    async fn test_create_blob_from_bytes() {
        let (store, _temp) = create_test_store().await;

        let data = b"Hello, iroh-blobs!";
        let metadata = BlobMetadata::with_name("test.txt");

        let token = store.create_blob_from_bytes(data, metadata).await.unwrap();

        assert_eq!(token.size_bytes, data.len() as u64);
        assert_eq!(token.metadata.name, Some("test.txt".to_string()));
        assert!(!token.hash.as_hex().is_empty());
    }

    #[tokio::test]
    async fn test_create_blob_from_file() {
        let (store, temp_dir) = create_test_store().await;

        // Create a test file
        let test_file = temp_dir.path().join("test_input.txt");
        std::fs::write(&test_file, "File content for testing").unwrap();

        let metadata = BlobMetadata::with_name_and_type("test_input.txt", "text/plain");
        let token = store.create_blob(&test_file, metadata).await.unwrap();

        assert_eq!(token.size_bytes, 24); // "File content for testing".len()
        assert_eq!(token.metadata.name, Some("test_input.txt".to_string()));
        assert_eq!(token.metadata.content_type, Some("text/plain".to_string()));
    }

    #[tokio::test]
    async fn test_fetch_blob() {
        let (store, _temp) = create_test_store().await;

        // Create a blob first
        let data = b"Content to fetch";
        let metadata = BlobMetadata::with_name("fetch_test.bin");
        let token = store.create_blob_from_bytes(data, metadata).await.unwrap();

        // Fetch it back
        let handle = store.fetch_blob(&token, |_progress| {}).await.unwrap();

        assert!(handle.path.exists());
        let content = std::fs::read(&handle.path).unwrap();
        assert_eq!(content, data);
    }

    #[tokio::test]
    async fn test_blob_exists_locally() {
        let (store, _temp) = create_test_store().await;

        let data = b"Test data";
        let metadata = BlobMetadata::default();
        let token = store.create_blob_from_bytes(data, metadata).await.unwrap();

        assert!(store.blob_exists_locally(&token.hash));

        let unknown_hash =
            BlobHash::from_hex("0000000000000000000000000000000000000000000000000000000000000000");
        assert!(!store.blob_exists_locally(&unknown_hash));
    }

    #[tokio::test]
    async fn test_blob_info() {
        let (store, _temp) = create_test_store().await;

        let data = b"Info test";
        let metadata = BlobMetadata::with_name("info.dat").with_custom("version", "1.0");
        let token = store.create_blob_from_bytes(data, metadata).await.unwrap();

        let info = store.blob_info(&token.hash).unwrap();
        assert_eq!(info.size_bytes, token.size_bytes);
        assert_eq!(info.metadata.name, Some("info.dat".to_string()));
        assert_eq!(
            info.metadata.custom.get("version"),
            Some(&"1.0".to_string())
        );
    }

    #[tokio::test]
    async fn test_delete_blob() {
        let (store, _temp) = create_test_store().await;

        let data = b"To be deleted";
        let metadata = BlobMetadata::default();
        let token = store.create_blob_from_bytes(data, metadata).await.unwrap();

        // Export blob to local file first
        let _ = store.fetch_blob(&token, |_| {}).await.unwrap();

        assert!(store.blob_exists_locally(&token.hash));

        store.delete_blob(&token.hash).await.unwrap();

        // Should no longer exist locally
        let local_path = store.local_blob_path(&token.hash);
        assert!(!local_path.exists());
        assert!(store.blob_info(&token.hash).is_none());
    }

    #[tokio::test]
    async fn test_list_local_blobs() {
        let (store, _temp) = create_test_store().await;

        // Create multiple blobs
        let token1 = store
            .create_blob_from_bytes(b"Blob 1", BlobMetadata::with_name("one.txt"))
            .await
            .unwrap();
        let token2 = store
            .create_blob_from_bytes(b"Blob 2", BlobMetadata::with_name("two.txt"))
            .await
            .unwrap();
        let token3 = store
            .create_blob_from_bytes(b"Blob 3", BlobMetadata::with_name("three.txt"))
            .await
            .unwrap();

        let blobs = store.list_local_blobs();
        assert_eq!(blobs.len(), 3);

        let hashes: Vec<_> = blobs.iter().map(|t| t.hash.clone()).collect();
        assert!(hashes.contains(&token1.hash));
        assert!(hashes.contains(&token2.hash));
        assert!(hashes.contains(&token3.hash));
    }

    #[tokio::test]
    async fn test_local_storage_bytes() {
        let (store, _temp) = create_test_store().await;

        // Initially zero
        assert_eq!(store.local_storage_bytes(), 0);

        // Create blobs (no need to export them - cache tracks sizes)
        let data1 = b"Small";
        let _token1 = store
            .create_blob_from_bytes(data1, BlobMetadata::default())
            .await
            .unwrap();

        let data2 = b"Larger blob content";
        let _token2 = store
            .create_blob_from_bytes(data2, BlobMetadata::default())
            .await
            .unwrap();

        let total = store.local_storage_bytes();
        assert_eq!(total, (data1.len() + data2.len()) as u64);
    }

    #[tokio::test]
    async fn test_metadata_persistence() {
        let temp_dir = TempDir::new().unwrap();
        let blob_dir = temp_dir.path().to_path_buf();

        // Create a store and add a blob
        let store1 = IrohBlobStore::new_in_memory(blob_dir.clone())
            .await
            .unwrap();

        let data = b"Persistent metadata test";
        let metadata = BlobMetadata::with_name("persist.txt").with_custom("key", "value");
        let token = store1.create_blob_from_bytes(data, metadata).await.unwrap();

        // Create a NEW store pointing to the same directory
        let store2 = IrohBlobStore::new_in_memory(blob_dir).await.unwrap();

        // Metadata should be loadable from the sidecar file
        let info = store2.blob_info(&token.hash).unwrap();
        assert_eq!(info.metadata.name, Some("persist.txt".to_string()));
        assert_eq!(info.metadata.custom.get("key"), Some(&"value".to_string()));
    }

    #[test]
    fn test_sidecar_metadata_serialization() {
        let metadata = BlobMetadata::with_name("test.onnx")
            .with_custom("version", "1.0")
            .with_custom("model_id", "yolov8");

        let sidecar = SidecarMetadata::new(metadata, 1024);

        let json = serde_json::to_string(&sidecar).unwrap();
        let parsed: SidecarMetadata = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.size_bytes, 1024);
        assert_eq!(parsed.metadata.name, Some("test.onnx".to_string()));
        assert_eq!(
            parsed.metadata.custom.get("version"),
            Some(&"1.0".to_string())
        );
    }

    #[tokio::test]
    async fn test_create_blob_from_stream() {
        let (store, _temp) = create_test_store().await;

        let data = b"Hello from a stream!";
        let mut reader: &[u8] = data;
        let metadata = BlobMetadata::with_name("streamed.txt");

        let token = store
            .create_blob_from_stream(&mut reader, Some(data.len() as u64), metadata)
            .await
            .unwrap();

        assert_eq!(token.size_bytes, data.len() as u64);
        assert_eq!(token.metadata.name, Some("streamed.txt".to_string()));

        // Verify content round-trips
        let handle = store.fetch_blob(&token, |_| {}).await.unwrap();
        let content = std::fs::read(&handle.path).unwrap();
        assert_eq!(content, data);
    }

    #[tokio::test]
    async fn test_create_blob_from_stream_unknown_size() {
        let (store, _temp) = create_test_store().await;

        let data = b"Stream with unknown size";
        let mut reader: &[u8] = data;

        let token = store
            .create_blob_from_stream(&mut reader, None, BlobMetadata::default())
            .await
            .unwrap();

        assert_eq!(token.size_bytes, data.len() as u64);
    }

    #[tokio::test]
    async fn test_open_read_stream() {
        use tokio::io::AsyncReadExt;

        let (store, _temp) = create_test_store().await;

        let data = b"Content for streaming read";
        let metadata = BlobMetadata::with_name("stream_read.bin");
        let token = store.create_blob_from_bytes(data, metadata).await.unwrap();

        // Fetch to make it available on disk
        let handle = store.fetch_blob(&token, |_| {}).await.unwrap();

        // Open as stream and read in chunks
        let mut stream = handle.open_read_stream().await.unwrap();
        let mut buf = Vec::new();
        stream.read_to_end(&mut buf).await.unwrap();
        assert_eq!(buf, data);
    }

    // ---- BlobPeerIndex unit tests ----

    fn test_peer_id(seed: u8) -> EndpointId {
        SecretKey::from_bytes(&[seed; 32]).public()
    }

    #[test]
    fn test_blob_peer_index_advertise_and_lookup() {
        let mut index = BlobPeerIndex::new();
        let peer_a = test_peer_id(1);
        let peer_b = test_peer_id(2);
        let hash = BlobHash::from_hex("abc123");

        index.advertise(peer_a, hash.clone());
        index.advertise(peer_b, hash.clone());

        let peers = index.peers_with_blob(&hash);
        assert_eq!(peers.len(), 2);
        assert!(peers.contains(&peer_a));
        assert!(peers.contains(&peer_b));
    }

    #[test]
    fn test_blob_peer_index_remove_peer() {
        let mut index = BlobPeerIndex::new();
        let peer_a = test_peer_id(1);
        let peer_b = test_peer_id(2);
        let hash1 = BlobHash::from_hex("abc");
        let hash2 = BlobHash::from_hex("def");

        index.advertise(peer_a, hash1.clone());
        index.advertise(peer_a, hash2.clone());
        index.advertise(peer_b, hash1.clone());

        assert_eq!(index.peer_blob_count(&peer_a), 2);
        assert_eq!(index.total_entries(), 3);

        index.remove_peer(&peer_a);

        assert_eq!(index.peer_blob_count(&peer_a), 0);
        assert_eq!(index.peers_with_blob(&hash1).len(), 1); // only peer_b
        assert!(index.peers_with_blob(&hash2).is_empty()); // peer_a was the only one
        assert_eq!(index.total_entries(), 1);
    }

    #[test]
    fn test_blob_peer_index_unknown_blob() {
        let index = BlobPeerIndex::new();
        let hash = BlobHash::from_hex("nonexistent");
        assert!(index.peers_with_blob(&hash).is_empty());
    }

    #[test]
    fn test_blob_peer_index_duplicate_advertise() {
        let mut index = BlobPeerIndex::new();
        let peer = test_peer_id(1);
        let hash = BlobHash::from_hex("abc");

        index.advertise(peer, hash.clone());
        index.advertise(peer, hash.clone()); // duplicate

        assert_eq!(index.peers_with_blob(&hash).len(), 1);
        assert_eq!(index.peer_blob_count(&peer), 1);
        assert_eq!(index.total_entries(), 1);
    }

    #[test]
    fn test_advertise_reports_newness() {
        // Newness drives provider-gossip re-broadcast termination: a node
        // relays a (holder, hash) only the first time it learns it.
        let mut index = BlobPeerIndex::new();
        let peer = test_peer_id(1);
        let hash = BlobHash::from_hex("abc");

        assert!(
            index.advertise(peer, hash.clone()),
            "first is newly learned"
        );
        assert!(
            !index.advertise(peer, hash.clone()),
            "second is not new for the same peer"
        );
        // A different peer holding the same blob is newly learned for that peer.
        assert!(index.advertise(test_peer_id(2), hash));
    }

    #[test]
    fn test_advertise_many_returns_only_newly_learned() {
        let mut index = BlobPeerIndex::new();
        let peer = test_peer_id(1);
        let h1 = BlobHash::from_hex("a1");
        let h2 = BlobHash::from_hex("a2");
        let h3 = BlobHash::from_hex("a3");

        // Pre-seed h1 so it isn't "new" in the batch below.
        index.advertise(peer, h1.clone());

        let newly = index.advertise_many(peer, &[h1.clone(), h2.clone(), h3.clone()]);
        assert_eq!(
            newly,
            vec![h2, h3],
            "only the not-already-known hashes are reported new"
        );
        assert_eq!(index.peer_blob_count(&peer), 3);
    }

    // ---- PeerHealthIndex unit tests (#137) ----

    const COOLDOWN: Duration = Duration::from_secs(120);

    #[test]
    fn test_peer_health_untried_peers_keep_input_order() {
        // No records → every peer is Neutral → stable sort is a no-op.
        let index = PeerHealthIndex::new();
        let peers = vec![test_peer_id(1), test_peer_id(2), test_peer_id(3)];
        let now = Instant::now();
        assert_eq!(index.order_peers(&peers, COOLDOWN, now), peers);
    }

    #[test]
    fn test_peer_health_recent_failure_demoted_last() {
        // The #137 case: the preferred peer (first in input) recently
        // stalled, so it must be tried last — but still tried (not skipped).
        let mut index = PeerHealthIndex::new();
        let dead = test_peer_id(1);
        let live = test_peer_id(2);
        let now = Instant::now();
        index.record(dead, AttemptHealth::Failure, now);

        let ordered = index.order_peers(&[dead, live], COOLDOWN, now);
        assert_eq!(ordered, vec![live, dead], "dead peer should sort last");
        assert!(ordered.contains(&dead), "demote, not skip");
    }

    #[test]
    fn test_peer_health_recent_success_promoted_first() {
        let mut index = PeerHealthIndex::new();
        let untried = test_peer_id(1);
        let good = test_peer_id(2);
        let now = Instant::now();
        index.record(good, AttemptHealth::Success, now);

        // `good` is second in input but should be promoted ahead of the
        // untried (Neutral) peer.
        let ordered = index.order_peers(&[untried, good], COOLDOWN, now);
        assert_eq!(ordered, vec![good, untried]);
    }

    #[test]
    fn test_peer_health_failure_reverts_to_neutral_after_cooldown() {
        // A failure older than the cooldown window must not keep demoting
        // the peer — it gets a fresh first-class attempt.
        let mut index = PeerHealthIndex::new();
        let recovered = test_peer_id(1);
        let other = test_peer_id(2);
        let now = Instant::now();
        let stale = now - (COOLDOWN + Duration::from_secs(1));
        index.record(recovered, AttemptHealth::Failure, stale);

        // Both are now Neutral → input order preserved (stable sort).
        let ordered = index.order_peers(&[recovered, other], COOLDOWN, now);
        assert_eq!(ordered, vec![recovered, other]);
    }

    #[test]
    fn test_peer_health_full_ranking_order() {
        // Healthy < Neutral < Unhealthy, stable within each rank.
        let mut index = PeerHealthIndex::new();
        let healthy = test_peer_id(1);
        let neutral = test_peer_id(2);
        let unhealthy = test_peer_id(3);
        let now = Instant::now();
        index.record(healthy, AttemptHealth::Success, now);
        index.record(unhealthy, AttemptHealth::Failure, now);

        // Deliberately feed worst-first; expect healthy, neutral, unhealthy.
        let ordered = index.order_peers(&[unhealthy, neutral, healthy], COOLDOWN, now);
        assert_eq!(ordered, vec![healthy, neutral, unhealthy]);
    }

    #[test]
    fn test_peer_health_stable_within_rank_preserves_promotion() {
        // Two recently-failed peers retain their input order, so any
        // upstream index-promotion ordering survives as the tiebreaker.
        let mut index = PeerHealthIndex::new();
        let first = test_peer_id(1);
        let second = test_peer_id(2);
        let now = Instant::now();
        index.record(first, AttemptHealth::Failure, now);
        index.record(second, AttemptHealth::Failure, now);

        assert_eq!(
            index.order_peers(&[first, second], COOLDOWN, now),
            vec![first, second]
        );
    }

    #[test]
    fn test_peer_health_latest_outcome_wins() {
        // A success after a failure clears the demotion.
        let mut index = PeerHealthIndex::new();
        let peer = test_peer_id(1);
        let other = test_peer_id(2);
        let now = Instant::now();
        index.record(peer, AttemptHealth::Failure, now - Duration::from_secs(5));
        index.record(peer, AttemptHealth::Success, now);

        assert_eq!(index.rank(&peer, COOLDOWN, now), PeerHealthRank::Healthy);
        let ordered = index.order_peers(&[other, peer], COOLDOWN, now);
        assert_eq!(ordered, vec![peer, other]);
    }

    #[test]
    fn test_peer_health_remove_peer() {
        let mut index = PeerHealthIndex::new();
        let peer = test_peer_id(1);
        let now = Instant::now();
        index.record(peer, AttemptHealth::Failure, now);
        assert_eq!(index.rank(&peer, COOLDOWN, now), PeerHealthRank::Unhealthy);

        index.remove_peer(&peer);
        assert_eq!(index.rank(&peer, COOLDOWN, now), PeerHealthRank::Neutral);
    }

    // ---- ready_peers unit tests ----

    #[test]
    fn test_ready_peers_excludes_cooling_unhealthy() {
        // A peer with a fresh failure is within its cooldown and must be
        // excluded from the ready set — this is the stall-cycle fix.
        let mut index = PeerHealthIndex::new();
        let cooling = test_peer_id(1);
        let ready = test_peer_id(2);
        let now = Instant::now();
        index.record(cooling, AttemptHealth::Failure, now);

        let result = index.ready_peers(&[cooling, ready], COOLDOWN, now);
        assert_eq!(result, vec![ready], "cooling peer must be excluded");
        assert!(!result.contains(&cooling), "cooling peer must not appear");
    }

    #[test]
    fn test_ready_peers_includes_expired_unhealthy() {
        // A failure older than the cooldown window has expired back to
        // Neutral and must be included in the ready set.
        let mut index = PeerHealthIndex::new();
        let expired = test_peer_id(1);
        let now = Instant::now();
        let stale = now - (COOLDOWN + Duration::from_secs(1));
        index.record(expired, AttemptHealth::Failure, stale);

        let result = index.ready_peers(&[expired], COOLDOWN, now);
        assert_eq!(result, vec![expired], "expired peer must be ready");
    }

    #[test]
    fn test_ready_peers_preserves_health_order() {
        // Within the ready set, Healthy peers precede Neutral ones.
        let mut index = PeerHealthIndex::new();
        let healthy = test_peer_id(1);
        let neutral = test_peer_id(2);
        let now = Instant::now();
        index.record(healthy, AttemptHealth::Success, now);

        let result = index.ready_peers(&[neutral, healthy], COOLDOWN, now);
        assert_eq!(result, vec![healthy, neutral]);
    }

    #[test]
    fn test_ready_peers_all_cooling_returns_empty() {
        // If every peer is in its cooldown window, the ready set is empty.
        // fetch_blob will return a transient error and the sweep retries.
        let mut index = PeerHealthIndex::new();
        let p1 = test_peer_id(1);
        let p2 = test_peer_id(2);
        let now = Instant::now();
        index.record(p1, AttemptHealth::Failure, now);
        index.record(p2, AttemptHealth::Failure, now);

        let result = index.ready_peers(&[p1, p2], COOLDOWN, now);
        assert!(result.is_empty(), "all cooling → empty ready set");
    }

    #[test]
    fn test_ready_peers_untried_always_ready() {
        // Peers with no record at all are Neutral → always ready.
        let index = PeerHealthIndex::new();
        let p1 = test_peer_id(1);
        let p2 = test_peer_id(2);
        let now = Instant::now();

        let result = index.ready_peers(&[p1, p2], COOLDOWN, now);
        assert_eq!(result, vec![p1, p2]);
    }

    // ---- Integration tests ----
    //
    // QUIC tests must run sequentially to avoid port contention under parallel load.
    static QUIC_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());

    /// Bind an iroh endpoint with a 10s timeout to prevent hangs under load.
    async fn bind_test_endpoint(
        lookup: iroh::address_lookup::memory::MemoryLookup,
        secret_key: [u8; 32],
    ) -> Endpoint {
        tokio::time::timeout(
            Duration::from_secs(10),
            Endpoint::builder(iroh::endpoint::presets::Empty)
                .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                .address_lookup(lookup)
                .secret_key(iroh::SecretKey::from_bytes(&secret_key))
                .bind(),
        )
        .await
        .expect("endpoint bind timed out")
        .unwrap()
    }

    /// Connect to an endpoint with a 10s timeout.
    async fn connect_test_endpoint(
        endpoint: &Endpoint,
        peer_id: iroh::PublicKey,
        alpn: &[u8],
    ) -> iroh::endpoint::Connection {
        tokio::time::timeout(Duration::from_secs(10), endpoint.connect(peer_id, alpn))
            .await
            .expect("connect timed out")
            .unwrap()
    }

    /// After a successful P2P fetch, the blob→peer mapping should be recorded in the index.
    #[tokio::test]
    async fn test_p2p_fetch_populates_index() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let dir_a = TempDir::new().unwrap();
        let dir_b = TempDir::new().unwrap();

        // Build endpoints directly using empty_builder to avoid external
        // DNS/relay dependencies in tests.
        let memory_lookup_a = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_a = bind_test_endpoint(memory_lookup_a.clone(), [1u8; 32]).await;

        let memory_lookup_b = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_b = bind_test_endpoint(memory_lookup_b.clone(), [2u8; 32]).await;

        let store_a = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_a.path().to_path_buf(),
            endpoint_a,
            memory_lookup_a,
            vec![],
        )
        .await
        .unwrap();

        let store_b = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_b.path().to_path_buf(),
            endpoint_b,
            memory_lookup_b.clone(),
            vec![],
        )
        .await
        .unwrap();

        // Wire up discovery using the full endpoint address
        memory_lookup_b.add_endpoint_info(store_a.endpoint().addr());
        store_b.add_peer(store_a.endpoint_id()).await;

        // Index should be empty before fetch
        let token = store_a
            .create_blob_from_bytes(b"indexed blob", BlobMetadata::with_name("idx.bin"))
            .await
            .unwrap();
        assert!(store_b.peers_with_blob(&token.hash).await.is_empty());

        // Fetch — should populate the index
        let _handle = store_b.fetch_blob(&token, |_| {}).await.unwrap();

        let indexed_peers = store_b.peers_with_blob(&token.hash).await;
        assert_eq!(indexed_peers.len(), 1);
        assert_eq!(indexed_peers[0], store_a.endpoint_id());

        // Shutdown may time out under heavy parallel test load; ignore errors
        let _ = store_a.shutdown().await;
        let _ = store_b.shutdown().await;
    }

    /// Surface-tier wiring test for the peat-mesh #137 fetch-health
    /// ordering through the actual `fetch_blob` boundary.
    ///
    /// A fetcher with an unreachable peer listed ahead of a live peer must
    /// (a) still deliver the blob (demote-not-skip preserves availability),
    /// (b) record the live peer as `Healthy` on success, and (c) order the
    /// healthy peer ahead of the other on the next fetch.
    ///
    /// Crucially, the unreachable peer here fails *fast* with an `Errored`
    /// outcome (no route → iroh-blobs "Unable to download"), and an
    /// `Errored` attempt must NOT demote the peer — that outcome is
    /// ambiguous (it's emitted identically for "peer is down" and "peer is
    /// reachable but doesn't have the blob yet", e.g. a relay still staging
    /// in a dual-C2 topology) and costs no stall penalty, so it is not a
    /// peer-health signal. Demotion is reserved for `Stalled` (no response
    /// at all), which is covered by the watchdog + `PeerHealthIndex` unit
    /// tests. So the unreachable peer stays `Neutral` here.
    #[tokio::test]
    async fn test_fetch_blob_records_peer_health() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let dir_live = TempDir::new().unwrap();
        let dir_fetch = TempDir::new().unwrap();

        let memory_lookup_live = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_live = bind_test_endpoint(memory_lookup_live.clone(), [1u8; 32]).await;

        let memory_lookup_fetch = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_fetch = bind_test_endpoint(memory_lookup_fetch.clone(), [2u8; 32]).await;

        let store_live = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_live.path().to_path_buf(),
            endpoint_live,
            memory_lookup_live,
            vec![],
        )
        .await
        .unwrap();

        // Short connect/stall timeouts so the unreachable peer fails fast;
        // the dead peer has no address wired into MemoryLookup, so the
        // attempt resolves (error or stall) within these bounds rather than
        // the 30s defaults. Cooldown left generous so the recording is
        // still "recent" when we assert.
        let store_fetch = NetworkedIrohBlobStore::from_endpoint_with_protocols_with_timeouts(
            dir_fetch.path().to_path_buf(),
            endpoint_fetch,
            memory_lookup_fetch.clone(),
            vec![],
            Duration::from_secs(5),
            Duration::from_secs(3),
            Duration::from_secs(1),
            Duration::from_secs(120),
        )
        .await
        .unwrap();

        // A dead peer that is never wired into MemoryLookup — unreachable.
        let dead_peer = SecretKey::from_bytes(&[9u8; 32]).public();
        let live_peer = store_live.endpoint_id();

        // Order matters: dead peer listed FIRST. With an empty health index
        // both are Neutral, so insertion order is preserved and the dead
        // peer is attempted before the live one.
        store_fetch.add_peer(dead_peer).await;
        memory_lookup_fetch.add_endpoint_info(store_live.endpoint().addr());
        store_fetch.add_peer(live_peer).await;

        let data = b"health-recording blob";
        let token = store_live
            .create_blob_from_bytes(data, BlobMetadata::with_name("health.bin"))
            .await
            .unwrap();

        // The blob must still arrive despite the dead peer being tried
        // first — demote-not-skip preserves availability.
        let handle = store_fetch.fetch_blob(&token, |_| {}).await.unwrap();
        assert_eq!(std::fs::read(&handle.path).unwrap(), data);

        let health = store_fetch.peer_health.read().await;
        let now = Instant::now();
        let cooldown = store_fetch.peer_health_cooldown;
        // The peer that failed *fast* with an error (no route → "Unable to
        // download") must NOT be demoted — an errored attempt is ambiguous
        // and costs no stall penalty, so it is not a health signal.
        assert_eq!(
            health.rank(&dead_peer, cooldown, now),
            PeerHealthRank::Neutral,
            "an errored (non-stalling) peer must not be demoted"
        );
        // The live peer's success is recorded.
        assert_eq!(
            health.rank(&live_peer, cooldown, now),
            PeerHealthRank::Healthy,
            "live peer should be recorded Healthy after a successful fetch"
        );

        // The recorded health still flips the ordering: the successful peer
        // (Healthy) sorts ahead of the untried/errored one (Neutral).
        assert_eq!(
            health.order_peers(&[dead_peer, live_peer], cooldown, now),
            vec![live_peer, dead_peer],
            "healthy peer should be ordered ahead of the neutral one"
        );
        drop(health);

        let _ = store_live.shutdown().await;
        let _ = store_fetch.shutdown().await;
    }

    /// Surface-tier wiring test for the cooling-skip early-return in
    /// `fetch_blob` (peat-mesh #257).
    ///
    /// Records a peer as Unhealthy via a direct health injection, then calls
    /// `fetch_blob`. With the sole peer cooling, `ready_peers()` returns an
    /// empty set → `cooling_count > 0` → the early-return fires without
    /// entering the attempt loop. Asserts:
    ///
    /// 1. The call returns `Err` immediately — elapsed << `download_stall_timeout`.
    /// 2. The error message contains "cooling off".
    ///
    /// This is the surface-tier complement to the `ready_peers` unit tests
    /// (`test_ready_peers_all_cooling_returns_empty`): those verify the
    /// `PeerHealthIndex` filter in isolation; this test verifies the wiring
    /// from the filter through `fetch_blob`'s actual execution path.
    #[tokio::test]
    async fn test_fetch_blob_returns_fast_when_all_peers_cooling() {
        let _lock = QUIC_TEST_LOCK.lock().await;

        // Build a source store to obtain a real, iroh-parseable blob token.
        // Shut it down immediately — its endpoint is no longer bound, so
        // any accidental connection attempt by the fetcher fails fast rather
        // than stalling for the full download_stall_timeout.
        let dir_src = TempDir::new().unwrap();
        let lookup_src = iroh::address_lookup::memory::MemoryLookup::new();
        let ep_src = bind_test_endpoint(lookup_src.clone(), [0xC0u8; 32]).await;
        let store_src = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_src.path().to_path_buf(),
            ep_src,
            lookup_src,
            vec![],
        )
        .await
        .unwrap();
        let token = store_src
            .create_blob_from_bytes(
                b"cooling-skip wiring test",
                BlobMetadata::with_name("cs.bin"),
            )
            .await
            .unwrap();
        let src_peer_id = store_src.endpoint_id();
        let _ = store_src.shutdown().await;

        // Build the fetcher with a short download_stall_timeout (2 s) as a
        // sentinel: if we accidentally enter the attempt loop the test fails
        // in 2 s instead of the 30 s production default.
        // Long peer_health_cooldown (120 s) ensures the injected failure stays
        // within the cooling window for the duration of the test.
        let dir_fetch = TempDir::new().unwrap();
        let lookup_fetch = iroh::address_lookup::memory::MemoryLookup::new();
        let ep_fetch = bind_test_endpoint(lookup_fetch.clone(), [0xC1u8; 32]).await;
        let store_fetch = NetworkedIrohBlobStore::from_endpoint_with_protocols_with_timeouts(
            dir_fetch.path().to_path_buf(),
            ep_fetch,
            lookup_fetch,
            vec![],
            Duration::from_secs(5),   // shutdown_timeout
            Duration::from_secs(3),   // connect_timeout
            Duration::from_secs(2),   // download_stall_timeout — sentinel
            Duration::from_secs(120), // peer_health_cooldown
        )
        .await
        .unwrap();

        // Register the source peer and inject a fresh Failure — Unhealthy/cooling.
        store_fetch.add_peer(src_peer_id).await;
        store_fetch.peer_health.write().await.record(
            src_peer_id,
            AttemptHealth::Failure,
            Instant::now(),
        );

        let t0 = Instant::now();
        let result = store_fetch.fetch_blob(&token, |_| {}).await;
        let elapsed = t0.elapsed();

        assert!(
            result.is_err(),
            "expected Err: the only known peer is cooling off"
        );
        assert!(
            elapsed < Duration::from_millis(500),
            "cooling-skip must short-circuit without opening a connection \
             (elapsed {elapsed:?}; download_stall_timeout sentinel = 2 s)"
        );
        let msg = result.unwrap_err().to_string();
        assert!(
            msg.contains("cooling off"),
            "error message must identify the cooling-off condition; got: {msg}"
        );

        let _ = store_fetch.shutdown().await;
    }

    /// `reset_peer_health` clears only the health record — `known_peers` and
    /// `BlobPeerIndex` entries for the peer survive the call unchanged.
    ///
    /// This is the load-bearing asymmetric invariant documented on the method:
    /// unlike `remove_peer`, `reset_peer_health` must not evict the peer from
    /// either list so that blob-location knowledge accumulated before a blackout
    /// is preserved after the auto-reconnect watchdog re-dials the peer.
    #[tokio::test]
    async fn test_reset_peer_health_preserves_known_peers_and_blob_index() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let dir = TempDir::new().unwrap();
        let lookup = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint = bind_test_endpoint(lookup.clone(), [0x03u8; 32]).await;

        let store = NetworkedIrohBlobStore::from_endpoint_with_protocols_with_timeouts(
            dir.path().to_path_buf(),
            endpoint,
            lookup,
            vec![],
            Duration::from_secs(5),
            Duration::from_secs(3),
            Duration::from_secs(1),
            Duration::from_secs(120),
        )
        .await
        .unwrap();

        let peer = iroh::SecretKey::from_bytes(&[0x07u8; 32]).public();
        let blob_hash = BlobHash::from_hex("deadbeef");

        // Populate both lists and mark the peer Unhealthy.
        store.add_peer(peer).await;
        store.advertise_blob(peer, blob_hash.clone()).await;
        store
            .peer_health
            .write()
            .await
            .record(peer, AttemptHealth::Failure, Instant::now());

        // Verify pre-conditions.
        {
            let now = Instant::now();
            assert_eq!(
                store
                    .peer_health
                    .read()
                    .await
                    .rank(&peer, Duration::from_secs(120), now),
                PeerHealthRank::Unhealthy,
                "pre: peer must be Unhealthy before reset"
            );
            assert!(
                store.known_peers().await.contains(&peer),
                "pre: peer must be in known_peers"
            );
            assert!(
                store
                    .blob_peer_index
                    .read()
                    .await
                    .peers_with_blob(&blob_hash)
                    .contains(&peer),
                "pre: BlobPeerIndex must know about the peer"
            );
        }

        // The call under test.
        store.reset_peer_health(&peer).await;

        // Post-conditions: health cleared, the other two lists untouched.
        {
            let now = Instant::now();
            assert_eq!(
                store
                    .peer_health
                    .read()
                    .await
                    .rank(&peer, Duration::from_secs(120), now),
                PeerHealthRank::Neutral,
                "post: peer health must be Neutral after reset"
            );
            assert!(
                store.known_peers().await.contains(&peer),
                "post: peer must still be in known_peers — reset_peer_health must not evict"
            );
            assert!(
                store
                    .blob_peer_index
                    .read()
                    .await
                    .peers_with_blob(&blob_hash)
                    .contains(&peer),
                "post: BlobPeerIndex entries must survive reset — blob-location knowledge preserved"
            );
        }

        let _ = store.shutdown().await;
    }

    /// `add_peer_from_hex_id` rejects malformed input and registers a valid
    /// endpoint id (no pinned address — discovery supplies reachable addrs).
    #[tokio::test]
    async fn test_add_peer_from_hex_id() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let dir = TempDir::new().unwrap();
        let lookup = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint = bind_test_endpoint(lookup.clone(), [0x11u8; 32]).await;

        let store = NetworkedIrohBlobStore::from_endpoint_with_protocols_with_timeouts(
            dir.path().to_path_buf(),
            endpoint,
            lookup,
            vec![],
            Duration::from_secs(5),
            Duration::from_secs(3),
            Duration::from_secs(1),
            Duration::from_secs(120),
        )
        .await
        .unwrap();

        // (a) invalid hex → "invalid hex endpoint id"
        let err = store
            .add_peer_from_hex_id("zzzz")
            .await
            .expect_err("non-hex input must error");
        assert!(
            err.to_string().contains("invalid hex endpoint id"),
            "got: {err}"
        );

        // (b) valid hex but wrong decoded length (≠32) → "endpoint id must be 32 bytes"
        let err = store
            .add_peer_from_hex_id("deadbeef")
            .await
            .expect_err("wrong-length id must error");
        assert!(
            err.to_string().contains("endpoint id must be 32 bytes"),
            "got: {err}"
        );

        // (c) happy path → the peer id lands in known_peers.
        let peer = iroh::SecretKey::from_bytes(&[0x22u8; 32]).public();
        let hex_id = hex::encode(peer.as_bytes());
        store
            .add_peer_from_hex_id(&hex_id)
            .await
            .expect("valid endpoint id must register");
        assert!(
            store.known_peers().await.contains(&peer),
            "peer must be in known_peers after id-only registration"
        );

        let _ = store.shutdown().await;
    }

    /// Two `NetworkedIrohBlobStore` instances discover each other via
    /// `MemoryLookup`, one creates a blob, the other fetches it over QUIC.
    #[tokio::test]
    async fn test_p2p_blob_transfer() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let dir_a = TempDir::new().unwrap();
        let dir_b = TempDir::new().unwrap();

        // Build endpoints directly using empty_builder to avoid external
        // DNS/relay dependencies in tests.
        let memory_lookup_a = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_a = bind_test_endpoint(memory_lookup_a.clone(), [1u8; 32]).await;

        let memory_lookup_b = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_b = bind_test_endpoint(memory_lookup_b.clone(), [2u8; 32]).await;

        let store_a = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_a.path().to_path_buf(),
            endpoint_a,
            memory_lookup_a,
            vec![],
        )
        .await
        .unwrap();

        let store_b = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_b.path().to_path_buf(),
            endpoint_b,
            memory_lookup_b.clone(),
            vec![],
        )
        .await
        .unwrap();

        // Tell store_b about store_a's full endpoint address
        memory_lookup_b.add_endpoint_info(store_a.endpoint().addr());
        store_b.add_peer(store_a.endpoint_id()).await;

        // Create a blob on store_a
        let data = b"Hello from store A via P2P!";
        let metadata = BlobMetadata::with_name("p2p_test.txt");
        let token = store_a
            .create_blob_from_bytes(data, metadata)
            .await
            .unwrap();

        // Fetch the blob from store_b (should download from store_a over QUIC)
        let handle = store_b.fetch_blob(&token, |_| {}).await.unwrap();

        let content = std::fs::read(&handle.path).unwrap();
        assert_eq!(content, data);

        // Shutdown may time out under heavy parallel test load; ignore errors
        let _ = store_a.shutdown().await;
        let _ = store_b.shutdown().await;
    }

    // ---- fetch_blob_from_peer (peat#1013 direct P2P) ----

    /// A blob already present locally is returned without ever parsing
    /// `peer_id_hex` — an obviously-invalid id must not matter.
    #[tokio::test]
    async fn test_fetch_blob_from_peer_local_short_circuit() {
        let dir = TempDir::new().unwrap();
        let store = NetworkedIrohBlobStore::new(dir.path().to_path_buf())
            .await
            .unwrap();

        let data = b"already have this one locally";
        let token = store
            .create_blob_from_bytes(data, BlobMetadata::with_name("local.txt"))
            .await
            .unwrap();

        let handle = store
            .fetch_blob_from_peer(&token, "not-a-valid-hex-id", |_| {})
            .await
            .expect("local short-circuit must bypass peer parsing entirely");

        assert_eq!(std::fs::read(&handle.path).unwrap(), data);
    }

    /// Targeting a peer that was never registered via
    /// `add_peer`/`add_peer_from_hex`/`add_peer_from_hex_id` must fail fast,
    /// without touching `known_peers`, and the error must be tagged with the
    /// "direct fetch" marker so callers (peat-ffi) can distinguish it from a
    /// mesh-sync exhaustion error.
    #[tokio::test]
    async fn test_fetch_blob_from_peer_unregistered_peer_errors() {
        let dir = TempDir::new().unwrap();
        let store = NetworkedIrohBlobStore::new(dir.path().to_path_buf())
            .await
            .unwrap();

        // A syntactically valid (64 hex char / 32 byte) but nonexistent hash —
        // `Hash::from_str` requires the correct length, and a too-short
        // placeholder would panic inside iroh_blobs' decoder before this test
        // ever reaches the peer-registration check it's exercising.
        let token = BlobToken {
            hash: BlobHash(hex::encode([0xabu8; 32])),
            size_bytes: 4,
            metadata: BlobMetadata::default(),
        };
        let unregistered = iroh::SecretKey::from_bytes(&[0x33u8; 32]).public();
        let hex_id = hex::encode(unregistered.as_bytes());

        let before = store.known_peers().await;
        assert!(before.is_empty());

        let err = store
            .fetch_blob_from_peer(&token, &hex_id, |_| {})
            .await
            .expect_err("unregistered peer must error");
        assert!(
            err.to_string().contains("direct fetch") && err.to_string().contains("not registered"),
            "got: {err}"
        );
        assert!(
            store.known_peers().await.is_empty(),
            "an unregistered-peer failure must not mutate known_peers"
        );
    }

    /// Two nodes, direct pull: node B fetches straight from node A's hex id
    /// via `fetch_blob_from_peer`, with no document sync and no automatic
    /// candidate-list involvement — proves the direct path works end to end.
    #[tokio::test]
    async fn test_fetch_blob_from_peer_p2p_direct_pull() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let dir_a = TempDir::new().unwrap();
        let dir_b = TempDir::new().unwrap();

        let memory_lookup_a = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_a = bind_test_endpoint(memory_lookup_a.clone(), [3u8; 32]).await;

        let memory_lookup_b = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_b = bind_test_endpoint(memory_lookup_b.clone(), [4u8; 32]).await;

        let store_a = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_a.path().to_path_buf(),
            endpoint_a,
            memory_lookup_a,
            vec![],
        )
        .await
        .unwrap();

        let store_b = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_b.path().to_path_buf(),
            endpoint_b,
            memory_lookup_b.clone(),
            vec![],
        )
        .await
        .unwrap();

        // Register A with B — no document, no CRDT, no sync engine anywhere.
        memory_lookup_b.add_endpoint_info(store_a.endpoint().addr());
        store_b.add_peer(store_a.endpoint_id()).await;

        let data = b"direct P2P pull, no doc sync involved";
        let token = store_a
            .create_blob_from_bytes(data, BlobMetadata::with_name("direct.txt"))
            .await
            .unwrap();

        let a_hex = hex::encode(store_a.endpoint_id().as_bytes());
        let handle = store_b
            .fetch_blob_from_peer(&token, &a_hex, |_| {})
            .await
            .unwrap();

        assert_eq!(std::fs::read(&handle.path).unwrap(), data);

        let _ = store_a.shutdown().await;
        let _ = store_b.shutdown().await;
    }

    /// The crux of "direct" semantics: targeting peer B explicitly must fail
    /// even though peer A — also known to the fetcher, and actually holding
    /// the blob — would have succeeded. `fetch_blob_from_peer` must never
    /// fall back to another known peer.
    #[tokio::test]
    async fn test_fetch_blob_from_peer_no_fallback() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let dir_a = TempDir::new().unwrap();
        let dir_b = TempDir::new().unwrap();
        let dir_c = TempDir::new().unwrap();

        let memory_lookup_a = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_a = bind_test_endpoint(memory_lookup_a.clone(), [5u8; 32]).await;
        let memory_lookup_b = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_b = bind_test_endpoint(memory_lookup_b.clone(), [6u8; 32]).await;
        let memory_lookup_c = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_c = bind_test_endpoint(memory_lookup_c.clone(), [7u8; 32]).await;

        let store_a = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_a.path().to_path_buf(),
            endpoint_a,
            memory_lookup_a,
            vec![],
        )
        .await
        .unwrap();
        let store_b = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_b.path().to_path_buf(),
            endpoint_b,
            memory_lookup_b,
            vec![],
        )
        .await
        .unwrap();
        let store_c = NetworkedIrohBlobStore::from_endpoint_with_protocols(
            dir_c.path().to_path_buf(),
            endpoint_c,
            memory_lookup_c.clone(),
            vec![],
        )
        .await
        .unwrap();

        // C knows about BOTH A (has the blob) and B (does not) — the
        // automatic candidate-list path (fetch_blob) would happily fall back
        // to A. fetch_blob_from_peer must not.
        memory_lookup_c.add_endpoint_info(store_a.endpoint().addr());
        store_c.add_peer(store_a.endpoint_id()).await;
        memory_lookup_c.add_endpoint_info(store_b.endpoint().addr());
        store_c.add_peer(store_b.endpoint_id()).await;

        let data = b"only A has this blob";
        let token = store_a
            .create_blob_from_bytes(data, BlobMetadata::with_name("no-fallback.txt"))
            .await
            .unwrap();

        let b_hex = hex::encode(store_b.endpoint_id().as_bytes());
        let err = store_c
            .fetch_blob_from_peer(&token, &b_hex, |_| {})
            .await
            .expect_err("targeting B explicitly must fail even though A (also known) has the blob");
        assert!(err.to_string().contains("direct fetch"), "got: {err}");

        let _ = store_a.shutdown().await;
        let _ = store_b.shutdown().await;
        let _ = store_c.shutdown().await;
    }

    // ---- EndpointHooks tests ----

    /// `build_endpoint_with_hooks(None)` behaves identically to `build_endpoint`.
    #[tokio::test]
    async fn test_build_endpoint_without_hooks() {
        let config = IrohConfig {
            secret_key: Some([42u8; 32]),
            ..Default::default()
        };
        let (endpoint, _lookup) = NetworkedIrohBlobStore::build_endpoint(&config)
            .await
            .unwrap();
        // Endpoint should have a valid ID derived from the secret key
        let id_str = endpoint.id().to_string();
        assert!(!id_str.is_empty());
        endpoint.close().await;
    }

    /// When `IrohConfig.bind_addr` is an unspecified IPv4 address
    /// (`0.0.0.0:0`), `build_endpoint_with_hooks` must route through
    /// `interface_filter::select_advertise_interfaces` and `clear_ip_transports`
    /// + per-interface `bind_addr_with_opts(set_prefix_len)` — the
    /// load-bearing chain that iroh rejects ("Only a single default address
    /// can be set per IP family") if `set_prefix_len` is omitted.
    /// peat-mesh#152 QA caller-surface coverage for peat#890.
    #[tokio::test]
    async fn test_build_endpoint_unspecified_bind_applies_interface_filter() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let config = IrohConfig {
            secret_key: Some([44u8; 32]),
            bind_addr: Some("0.0.0.0:0".parse().unwrap()),
            ..Default::default()
        };
        let (endpoint, _lookup) = NetworkedIrohBlobStore::build_endpoint(&config)
            .await
            .expect("multi-interface bind must succeed with set_prefix_len");

        // Successful build with multiple v4 interfaces is the load-bearing
        // assertion — if `set_prefix_len` regressed, `build_endpoint` would
        // have errored above with "Only a single default address can be set
        // per IP family". The endpoint must also report a non-empty
        // advertised set with no loopback (peat#890 contract: filter
        // always drops 127.0.0.1).
        let addr = endpoint.addr();
        assert!(
            !addr.addrs.is_empty(),
            "endpoint must advertise at least one address post-filter"
        );
        let advertises_loopback = addr.addrs.iter().any(|t| match t {
            iroh::TransportAddr::Ip(sa) => sa.ip().is_loopback(),
            _ => false,
        });
        assert!(
            !advertises_loopback,
            "filter must never publish a loopback address; got addrs={:?}",
            addr.addrs
        );
        endpoint.close().await;
    }

    /// `EndpointHooks` fire on incoming connections when installed via the builder.
    #[tokio::test]
    async fn test_build_endpoint_with_custom_hooks() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        use iroh::protocol::Router;
        use std::sync::atomic::{AtomicBool, Ordering};

        #[derive(Debug)]
        struct TrackingHooks {
            after_handshake_called: Arc<AtomicBool>,
        }
        impl iroh::endpoint::EndpointHooks for TrackingHooks {
            fn after_handshake<'a>(
                &'a self,
                _conn: &'a iroh::endpoint::Connection,
            ) -> impl std::future::Future<Output = iroh::endpoint::AfterHandshakeOutcome> + Send + 'a
            {
                self.after_handshake_called.store(true, Ordering::SeqCst);
                async { iroh::endpoint::AfterHandshakeOutcome::Accept }
            }
        }

        let called = Arc::new(AtomicBool::new(false));
        let hooks = TrackingHooks {
            after_handshake_called: called.clone(),
        };

        // Build endpoints directly using empty_builder to avoid external
        // DNS/relay dependencies in tests.
        let lookup_a = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_a = tokio::time::timeout(
            Duration::from_secs(10),
            Endpoint::builder(iroh::endpoint::presets::Empty)
                .crypto_provider(crate::security::tls_provider::iroh_quic_provider())
                .address_lookup(lookup_a.clone())
                .secret_key(iroh::SecretKey::from_bytes(&[43u8; 32]))
                .hooks(hooks)
                .bind(),
        )
        .await
        .expect("endpoint_a bind timed out")
        .unwrap();

        // Minimal handler so Router accepts on this ALPN
        #[derive(Debug)]
        struct Noop;
        impl iroh::protocol::ProtocolHandler for Noop {
            async fn accept(
                &self,
                _conn: iroh::endpoint::Connection,
            ) -> Result<(), iroh::protocol::AcceptError> {
                Ok(())
            }
        }

        let _router_a = Router::builder(endpoint_a.clone())
            .accept(b"test/hook/1", Noop)
            .spawn();

        // Build a second endpoint and connect to the first to trigger the hook
        let lookup_b = iroh::address_lookup::memory::MemoryLookup::new();
        let endpoint_b = bind_test_endpoint(lookup_b.clone(), [44u8; 32]).await;

        // Tell B about A's full endpoint address
        lookup_b.add_endpoint_info(endpoint_a.addr());

        // B connects to A — this triggers A's after_handshake hook
        let conn = connect_test_endpoint(&endpoint_b, endpoint_a.id(), b"test/hook/1").await;

        // Give the hook a moment to fire
        tokio::time::sleep(Duration::from_millis(100)).await;
        assert!(
            called.load(Ordering::SeqCst),
            "after_handshake hook should have been called"
        );

        conn.close(0u32.into(), b"done");
        endpoint_a.close().await;
        endpoint_b.close().await;
    }

    /// Issue #833: in default builds (no `relay-n0-hosted` feature), the
    /// production `build_endpoint` constructor must produce an iroh
    /// endpoint with no relay URLs configured. Guards against accidental
    /// regression — for example, re-introducing `presets::N0` outside the
    /// feature gate, or a new constructor that bypasses
    /// `relay_policy_builder()`.
    #[cfg(not(feature = "relay-n0-hosted"))]
    #[tokio::test]
    async fn build_endpoint_default_has_no_relay_urls() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let config = crate::config::IrohConfig::default();

        let (endpoint, _lookup) = NetworkedIrohBlobStore::build_endpoint(&config)
            .await
            .expect("default build must construct endpoint without errors");

        let addr = endpoint.addr();
        let relays: Vec<_> = addr.relay_urls().collect();
        assert!(
            relays.is_empty(),
            "default build must construct an iroh endpoint with no relay \
             URLs configured (issue #833); got: {relays:?}"
        );

        endpoint.close().await;
    }

    /// Issue #833: opt-in `relay-n0-hosted` builds must restore n0's
    /// hosted relay infrastructure. Guards against accidentally breaking
    /// the escape hatch — if the feature gate is removed or
    /// `relay_policy_builder()` is changed to always use `empty_builder`,
    /// this test will fail and force the fix.
    ///
    /// Marked `#[ignore]` because populating `EndpointAddr::relay_urls()`
    /// on a `presets::N0` endpoint requires an actual handshake with n0's
    /// hosted relay infrastructure, which CI runners may not be able to
    /// reach. Run locally with network connectivity:
    ///
    /// ```sh
    /// cargo test --features relay-n0-hosted -- \
    ///     --include-ignored build_endpoint_n0_hosted_has_relay_urls
    /// ```
    #[cfg(feature = "relay-n0-hosted")]
    #[tokio::test]
    #[ignore = "requires network access to n0's hosted relay infrastructure"]
    async fn build_endpoint_n0_hosted_has_relay_urls() {
        let _lock = QUIC_TEST_LOCK.lock().await;
        let config = crate::config::IrohConfig::default();

        let (endpoint, _lookup) = NetworkedIrohBlobStore::build_endpoint(&config)
            .await
            .expect("relay-n0-hosted build must construct endpoint without errors");

        // The home-relay handshake is async — wait briefly for n0's relay
        // discovery to populate the endpoint address. The exact URL is not
        // asserted (n0 may rotate hosts); the assertion is only that some
        // relay URL is present, proving the preset wired in n0's relays.
        let mut attempts = 0;
        let relays = loop {
            let addr = endpoint.addr();
            let relays: Vec<_> = addr.relay_urls().cloned().collect();
            if !relays.is_empty() || attempts >= 20 {
                break relays;
            }
            attempts += 1;
            tokio::time::sleep(Duration::from_millis(250)).await;
        };
        assert!(
            !relays.is_empty(),
            "relay-n0-hosted build must restore n0's hosted relays \
             (issue #833); none were configured after 5s"
        );

        endpoint.close().await;
    }
}
