pub(crate) mod op_ctx_task;

use freenet_stdlib::client_api::{ErrorKind, HostResponse};
use freenet_stdlib::prelude::*;
use std::collections::HashSet;
use std::fmt::Display;
use std::pin::Pin;
use std::{future::Future, time::Instant};

use crate::client_events::HostResult;
use crate::config::GlobalExecutor;
use crate::node::IsOperationCompleted;
use crate::{
    contract::{ContractHandlerEvent, StoreResponse},
    message::{InnerMessage, NetMessage, Transaction},
    node::{NetworkBridge, OpManager},
    operations::{OpInitialization, Operation},
    ring::{Location, PeerKeyLocation},
    tracing::{NetEventLog, OperationFailure, state_hash_full},
};
use either::Either;

use super::orphan_streams::{OrphanStreamError, STREAM_CLAIM_TIMEOUT};
use super::{OpEnum, OpError, OpOutcome, OperationResult, should_use_streaming};
use crate::transport::peer_connection::StreamId;

pub(crate) use self::messages::{GetMsg, GetMsgResult, GetStreamingPayload};

/// Maximum number of retries to get values.
const MAX_RETRIES: usize = 10;

/// Maximum number of peer attempts at each hop level
const DEFAULT_MAX_BREADTH: usize = 3;

/// Minimum HTL for speculative retries.
///
/// Retries use a reduced HTL (capped at current_hop) to avoid full-depth
/// traversal storms. This floor ensures retries still reach peers 2-3 hops
/// away, which is the minimum useful search depth in any topology.
const MIN_RETRY_HTL: usize = 3;

/// Create a GET operation with a specific transaction ID (for operation deduplication).
///
/// Retained for unit tests. Client-initiated GETs use
/// `op_ctx_task::start_client_get`; sub-op GETs use
/// `op_ctx_task::start_sub_op_get`. Phase 6 may delete this entry
/// alongside the legacy `PrepareRequest` state.
#[allow(dead_code)]
pub(crate) fn start_op_with_id(
    instance_id: ContractInstanceId,
    fetch_contract: bool,
    subscribe: bool,
    blocking_subscribe: bool,
    id: Transaction,
) -> GetOp {
    let contract_location = Location::from(&instance_id);
    tracing::debug!(tx = %id, "Requesting get contract {instance_id} @ loc({contract_location}) with existing transaction ID");
    // Always fetch contract code from the network (see start_op and #3757)
    let state = Some(GetState::PrepareRequest(PrepareRequestData {
        instance_id,
        id,
        fetch_contract: true,
        subscribe,
        blocking_subscribe,
    }));
    GetOp {
        id,
        state,
        result: None,
        stats: Some(Box::new(GetStats {
            contract_location,
            next_peer: None,
            transfer_time: None,
            first_response_time: None,
        })),
        upstream_addr: None, // Local operation, no upstream peer
        local_fallback: None,
        auto_fetch: false,
        client_return_code: fetch_contract,
    }
}

/// Create a GET operation pre-targeted at a specific peer, bypassing normal routing.
/// Returns the operation and the message to send. Used for auto-fetching contracts
/// from peers known to have them (e.g., UPDATE senders).
pub(crate) fn start_targeted_op(
    instance_id: ContractInstanceId,
    target: PeerKeyLocation,
    max_hops_to_live: usize,
) -> (GetOp, GetMsg) {
    let contract_location = Location::from(&instance_id);
    let id = Transaction::new::<GetMsg>();
    tracing::debug!(
        tx = %id,
        "Requesting targeted get contract {instance_id} @ loc({contract_location})"
    );

    let visited = super::VisitedPeers::new(&id);
    let mut tried_peers = HashSet::new();
    if let Some(addr) = target.socket_addr() {
        tried_peers.insert(addr);
    }

    let state = Some(GetState::AwaitingResponse(AwaitingResponseData {
        instance_id,
        retries: 0,
        fetch_contract: true,
        requester: None,
        current_hop: max_hops_to_live,
        subscribe: false,
        blocking_subscribe: false,
        next_hop: target,
        tried_peers,
        alternatives: vec![],
        attempts_at_hop: 1,
        visited: visited.clone(),
    }));

    let msg = GetMsg::Request {
        id,
        instance_id,
        fetch_contract: true,
        htl: max_hops_to_live,
        visited,
        subscribe: false, // Targeted ops (auto-fetch) don't need subscription
    };

    let op = GetOp {
        id,
        state,
        result: None,
        stats: Some(Box::new(GetStats {
            contract_location,
            next_peer: None,
            transfer_time: None,
            first_response_time: None,
        })),
        upstream_addr: None,
        local_fallback: None,
        auto_fetch: true,         // System-initiated, not a client request
        client_return_code: true, // Internal fetch, always include code
    };

    (op, msg)
}

// Note: `request_get` (legacy client-initiated GET driver) was retired
// in the #1454 sub-op GET migration. Client-initiated GETs route via
// `op_ctx_task::start_client_get`; sub-op GETs (executor and
// SUBSCRIBE-spawned fetches) route via `op_ctx_task::start_sub_op_get`.
// Legacy auto-fetch (`start_targeted_op`) keeps its parallel
// state-machine path.

// ── Type-state data structs ──────────────────────────────────────────────
//
// Each state in the GET state machine is a named struct with typed
// transition methods.  This gives compile-time guarantees:
//
//   ReceivedRequest ── (no transition methods, next state depends on message)
//   PrepareRequest ── into_awaiting_response() ──► AwaitingResponse
//   AwaitingResponse ── into_finished()         ──► Finished
//                    ── with_retry_peer()        ──► AwaitingResponse (retry at same hop)
//                    ── with_next_hop()           ──► AwaitingResponse (advance to next hop)
//
// Invalid transitions (e.g. Finished → PrepareRequest) are unrepresentable
// because the target type simply has no such method.

/// Data for the PrepareRequest state: originator preparing GET request.
#[derive(Debug)]
struct PrepareRequestData {
    instance_id: ContractInstanceId,
    id: Transaction,
    fetch_contract: bool,
    subscribe: bool,
    blocking_subscribe: bool,
}

impl PrepareRequestData {
    /// Transition to AwaitingResponse after sending the GET request.
    ///
    /// Encodes valid transition: PrepareRequest → AwaitingResponse.
    /// Carries forward instance_id, fetch_contract, subscribe, blocking_subscribe.
    #[allow(dead_code)] // Documents valid transition; see type-state pattern
    fn into_awaiting_response(
        self,
        requester: Option<PeerKeyLocation>,
        next_hop: PeerKeyLocation,
        alternatives: Vec<PeerKeyLocation>,
        visited: super::VisitedPeers,
    ) -> AwaitingResponseData {
        AwaitingResponseData {
            instance_id: self.instance_id,
            requester,
            fetch_contract: self.fetch_contract,
            retries: 0,
            current_hop: 0,
            subscribe: self.subscribe,
            blocking_subscribe: self.blocking_subscribe,
            next_hop,
            tried_peers: HashSet::new(),
            alternatives,
            attempts_at_hop: 0,
            visited,
        }
    }
}

/// Data for the AwaitingResponse state: request sent, waiting for downstream peer.
///
/// This is the most complex state, tracking retry logic across multiple hops.
#[derive(Debug)]
struct AwaitingResponseData {
    /// Contract being fetched (by instance_id since we may not have full key yet)
    instance_id: ContractInstanceId,
    /// If specified the peer waiting for the response upstream
    requester: Option<PeerKeyLocation>,
    fetch_contract: bool,
    retries: usize,
    current_hop: usize,
    subscribe: bool,
    blocking_subscribe: bool,
    /// Peer we are currently trying to reach.
    /// Note: With connection-based routing, this is only used for state tracking,
    /// not for response routing (which uses upstream_addr instead).
    #[allow(dead_code)]
    next_hop: PeerKeyLocation,
    /// Peers we've already tried at this hop level
    tried_peers: HashSet<std::net::SocketAddr>,
    /// Alternative peers we could still try at this hop
    alternatives: Vec<PeerKeyLocation>,
    /// How many peers we've tried at this hop
    attempts_at_hop: usize,
    /// Bloom filter tracking visited peers across all hops
    visited: super::VisitedPeers,
}

impl AwaitingResponseData {
    /// Transition to Finished on successful GET response.
    ///
    /// Encodes valid transition: AwaitingResponse → Finished.
    #[allow(dead_code)] // Documents valid transition; see type-state pattern
    fn into_finished(self, key: ContractKey) -> FinishedData {
        FinishedData { key }
    }
}

/// Data for the Finished state: GET operation completed successfully.
#[derive(Debug, Clone, Copy)]
struct FinishedData {
    key: ContractKey,
}

// ── State enum (wraps the typed structs) ─────────────────────────────────

#[derive(Debug)]
#[allow(clippy::large_enum_variant)]
enum GetState {
    /// A new petition for a get op received from another peer.
    /// Note: We use GetOp::upstream_addr for response routing (not PeerKeyLocation).
    ReceivedRequest,
    /// Preparing request for get op.
    PrepareRequest(PrepareRequestData),
    /// Awaiting response from petition.
    AwaitingResponse(AwaitingResponseData),
    /// Operation completed successfully
    Finished(FinishedData),
}

impl Display for GetState {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            GetState::ReceivedRequest => write!(f, "ReceivedRequest"),
            GetState::PrepareRequest(data) => {
                write!(
                    f,
                    "PrepareRequest(instance_id: {}, id: {}, fetch_contract: {}, subscribe: {})",
                    data.instance_id, data.id, data.fetch_contract, data.subscribe
                )
            }
            GetState::AwaitingResponse(data) => {
                write!(
                    f,
                    "AwaitingResponse(requester: {:?}, fetch_contract: {}, retries: {}, current_hop: {}, subscribe: {})",
                    data.requester,
                    data.fetch_contract,
                    data.retries,
                    data.current_hop,
                    data.subscribe
                )
            }
            GetState::Finished(data) => write!(f, "Finished(key: {})", data.key),
        }
    }
}

struct GetStats {
    /// Next peer in get path to be targeted
    next_peer: Option<PeerKeyLocation>,
    contract_location: Location,
    /// (start, end)
    first_response_time: Option<(Instant, Option<Instant>)>,
    /// (start, end)
    transfer_time: Option<(Instant, Option<Instant>)>,
}

impl GetStats {
    /// Start both response and transfer timers for a new attempt.
    fn start_timers(&mut self) {
        let now = Instant::now();
        self.first_response_time = Some((now, None));
        self.transfer_time = Some((now, None));
    }

    /// Record the end time for first_response_time.
    fn record_response_end(&mut self) {
        if let Some((_, ref mut end)) = self.first_response_time {
            *end = Some(Instant::now());
        }
    }

    /// Record the end time for transfer_time.
    fn record_transfer_end(&mut self) {
        if let Some((_, ref mut end)) = self.transfer_time {
            *end = Some(Instant::now());
        }
    }
}

#[derive(Clone, Debug)]
pub(crate) struct GetResult {
    key: ContractKey,
    pub state: WrappedState,
    pub contract: Option<ContractContainer>,
}

impl GetResult {
    /// Construct a `GetResult` directly from its fields. Used by the
    /// task-per-tx sub-op GET driver, which assembles the result from
    /// the wire-level terminal reply rather than a fully populated
    /// `GetOp`.
    pub(crate) fn new(
        key: ContractKey,
        state: WrappedState,
        contract: Option<ContractContainer>,
    ) -> Self {
        Self {
            key,
            state,
            contract,
        }
    }
}

impl TryFrom<GetOp> for GetResult {
    type Error = OpError;

    fn try_from(value: GetOp) -> Result<Self, Self::Error> {
        match value.result {
            Some(r) => Ok(r),
            _ => Err(OpError::UnexpectedOpState),
        }
    }
}

pub(crate) struct GetOp {
    pub id: Transaction,
    state: Option<GetState>,
    pub(super) result: Option<GetResult>,
    stats: Option<Box<GetStats>>,
    /// The address we received this operation's message from.
    /// Used for connection-based routing: responses are sent back to this address.
    upstream_addr: Option<std::net::SocketAddr>,
    /// Local cached state to fall back to if network query fails or returns NotFound.
    /// Used by: (1) original requester's `request_get` path for "network first, local
    /// fallback", (2) relay peers with stale cache (no local interest) that defer to
    /// the network, and (3) connection abort fallback.
    local_fallback: Option<(ContractKey, WrappedState, Option<ContractContainer>)>,
    /// True when this GET was spawned internally by try_auto_fetch_contract,
    /// not by a client request. Used to avoid sending spurious timeout errors
    /// to a non-existent client.
    auto_fetch: bool,
    /// Whether the client wants contract code in the response.
    /// The node always fetches WASM from the network for internal caching,
    /// but strips it from the client response when this is false.
    /// See issue #3757.
    client_return_code: bool,
}

impl GetOp {
    pub(super) fn outcome(&self) -> OpOutcome<'_> {
        if let Some((
            GetResult {
                state, contract, ..
            },
            GetStats {
                next_peer: Some(target_peer),
                contract_location,
                first_response_time: Some((response_start, Some(response_end))),
                transfer_time: Some((transfer_start, Some(transfer_end))),
                ..
            },
        )) = self.result.as_ref().zip(self.stats.as_deref())
        {
            let payload_size = state.size()
                + contract
                    .as_ref()
                    .map(|c| c.data().len())
                    .unwrap_or_default();
            OpOutcome::ContractOpSuccess {
                target_peer,
                contract_location: *contract_location,
                payload_size,
                first_response_time: *response_end - *response_start,
                payload_transfer_time: *transfer_end - *transfer_start,
            }
        } else if self.result.is_none() {
            // No result — if we have stats with a target peer, report as failure
            if let Some(GetStats {
                next_peer: Some(target_peer),
                contract_location,
                ..
            }) = self.stats.as_deref()
            {
                OpOutcome::ContractOpFailure {
                    target_peer,
                    contract_location: *contract_location,
                }
            } else {
                OpOutcome::Incomplete
            }
        } else if let Some(GetStats {
            next_peer: Some(target_peer),
            contract_location,
            ..
        }) = self.stats.as_deref()
        {
            // Result present but timing incomplete (e.g. streaming completed without
            // full handshake timing). Report as untimed success so the router still
            // gets feedback instead of silently dropping the outcome.
            OpOutcome::ContractOpSuccessUntimed {
                target_peer,
                contract_location: *contract_location,
            }
        } else {
            OpOutcome::Incomplete
        }
    }

    /// Extract the contract instance ID from the operation state.
    pub(crate) fn instance_id(&self) -> Option<ContractInstanceId> {
        match &self.state {
            Some(GetState::PrepareRequest(data)) => Some(data.instance_id),
            Some(GetState::AwaitingResponse(data)) => Some(data.instance_id),
            _ => None,
        }
    }

    /// Returns true if this GET was initiated by a local client (not forwarded from
    /// a peer and not a system-initiated auto-fetch).
    pub(crate) fn is_client_initiated(&self) -> bool {
        if self.auto_fetch {
            return false;
        }
        match &self.state {
            Some(GetState::PrepareRequest(_)) => true,
            Some(GetState::AwaitingResponse(data)) => data.requester.is_none(),
            Some(GetState::ReceivedRequest) | Some(GetState::Finished(_)) | None => false,
        }
    }

    /// Extract routing failure info for timeout reporting.
    /// Returns `(target_peer, contract_location)` if stats are available.
    pub(crate) fn failure_routing_info(&self) -> Option<(PeerKeyLocation, Location)> {
        self.stats.as_deref().and_then(|stats| {
            stats
                .next_peer
                .as_ref()
                .map(|peer| (peer.clone(), stats.contract_location))
        })
    }

    /// Handle aborted outbound connections by directly retrying with alternative peers.
    pub(crate) async fn handle_abort(mut self, op_manager: &OpManager) -> Result<(), OpError> {
        if let Some(GetState::AwaitingResponse(AwaitingResponseData {
            instance_id,
            requester,
            fetch_contract,
            retries,
            current_hop,
            subscribe,
            blocking_subscribe,
            next_hop: failed_peer,
            mut tried_peers,
            mut alternatives,
            attempts_at_hop,
            visited,
        })) = self.state.take()
        {
            // Mark the failed peer as tried
            if let Some(addr) = failed_peer.socket_addr() {
                tried_peers.insert(addr);
            }

            // Try the next alternative if available
            if !alternatives.is_empty() && attempts_at_hop < DEFAULT_MAX_BREADTH {
                let next_target = alternatives.remove(0);

                tracing::debug!(
                    tx = %self.id,
                    %instance_id,
                    next_target = %next_target,
                    "GET: Connection aborted, trying alternative peer"
                );

                if let Some(addr) = next_target.socket_addr() {
                    tried_peers.insert(addr);
                }

                let new_state = Some(GetState::AwaitingResponse(AwaitingResponseData {
                    instance_id,
                    requester,
                    fetch_contract,
                    retries,
                    current_hop,
                    subscribe,
                    blocking_subscribe,
                    next_hop: next_target,
                    tried_peers,
                    alternatives,
                    attempts_at_hop: attempts_at_hop + 1,
                    visited: visited.clone(),
                }));

                // Note: handle_abort uses current_hop directly (not the attempts-based
                // reduction from retry_with_next_alternative). Connection aborts are
                // immediate failures, not timeout-based retries, so they don't contribute
                // to retry storms the same way. The bloom filter and tried_peers already
                // prevent cycling through the same peers. (#3570)
                let msg = GetMsg::Request {
                    id: self.id,
                    instance_id,
                    fetch_contract,
                    htl: current_hop,
                    visited,
                    subscribe,
                };

                let op = GetOp {
                    id: self.id,
                    state: new_state,
                    result: None,
                    stats: self.stats,
                    upstream_addr: self.upstream_addr,
                    local_fallback: self.local_fallback,
                    auto_fetch: self.auto_fetch,
                    client_return_code: self.client_return_code,
                };

                op_manager
                    .notify_op_change(NetMessage::from(msg), OpEnum::Get(op))
                    .await?;
                return Err(OpError::StatePushed);
            }

            // No alternatives - look for new candidates
            if retries < MAX_RETRIES {
                let mut new_visited = visited.clone();
                for addr in &tried_peers {
                    new_visited.mark_visited(*addr);
                }

                let mut new_candidates = op_manager.ring.k_closest_potentially_hosting(
                    &instance_id,
                    &new_visited,
                    DEFAULT_MAX_BREADTH,
                );

                if !new_candidates.is_empty() {
                    let next_target = new_candidates.remove(0);

                    tracing::debug!(
                        tx = %self.id,
                        %instance_id,
                        next_target = %next_target,
                        "GET: Connection aborted, found new candidate"
                    );

                    let mut new_tried_peers = HashSet::new();
                    if let Some(addr) = next_target.socket_addr() {
                        new_tried_peers.insert(addr);
                    }

                    let new_state = Some(GetState::AwaitingResponse(AwaitingResponseData {
                        instance_id,
                        requester,
                        fetch_contract,
                        retries: retries + 1,
                        current_hop,
                        subscribe,
                        blocking_subscribe,
                        next_hop: next_target,
                        tried_peers: new_tried_peers,
                        alternatives: new_candidates,
                        attempts_at_hop: 1,
                        visited: new_visited.clone(),
                    }));

                    let msg = GetMsg::Request {
                        id: self.id,
                        instance_id,
                        fetch_contract,
                        htl: current_hop,
                        visited: new_visited,
                        subscribe,
                    };

                    let op = GetOp {
                        id: self.id,
                        state: new_state,
                        result: None,
                        stats: self.stats,
                        upstream_addr: self.upstream_addr,
                        local_fallback: self.local_fallback,
                        auto_fetch: self.auto_fetch,
                        client_return_code: self.client_return_code,
                    };

                    op_manager
                        .notify_op_change(NetMessage::from(msg), OpEnum::Get(op))
                        .await?;
                    return Err(OpError::StatePushed);
                }
            }

            // No alternatives and no new candidates - check for local fallback
            // If we have a local cache, use it instead of failing entirely
            if let Some((key, state, contract)) = self.local_fallback {
                tracing::info!(
                    tx = %self.id,
                    contract = %key,
                    phase = "complete",
                    "GET: Connection aborted, no peers available - returning local cache"
                );

                let completed_op = GetOp {
                    id: self.id,
                    state: Some(GetState::Finished(FinishedData { key })),
                    result: Some(GetResult {
                        key,
                        state,
                        contract,
                    }),
                    stats: self.stats,
                    upstream_addr: self.upstream_addr,
                    local_fallback: None,
                    auto_fetch: false,
                    client_return_code: self.client_return_code,
                };

                op_manager.push(self.id, OpEnum::Get(completed_op)).await?;
                return Ok(());
            }

            // No local fallback either - give up
            if let Some(upstream) = self.upstream_addr {
                // Send NotFound response back to upstream requester instead of letting them timeout
                tracing::warn!(
                    tx = %self.id,
                    %instance_id,
                    %upstream,
                    phase = "not_found",
                    "GET: Connection aborted, no peers available - sending NotFound to upstream"
                );
                // Emit telemetry: relay returning NotFound after abort
                let hop_count = Some(op_manager.ring.max_hops_to_live.saturating_sub(current_hop));
                if let Some(event) =
                    NetEventLog::get_not_found(&self.id, &op_manager.ring, instance_id, hop_count)
                {
                    op_manager.ring.register_events(Either::Left(event)).await;
                }

                let response_op = GetOp {
                    id: self.id,
                    state: None,
                    result: None,
                    stats: self.stats,
                    upstream_addr: self.upstream_addr,
                    local_fallback: None,
                    auto_fetch: false,
                    client_return_code: self.client_return_code,
                };

                op_manager
                    .notify_op_change(
                        NetMessage::from(GetMsg::Response {
                            id: self.id,
                            instance_id,
                            result: GetMsgResult::NotFound,
                        }),
                        OpEnum::Get(response_op),
                    )
                    .await?;
                return Err(OpError::StatePushed);
            } else {
                // Original requester - operation fails locally
                tracing::warn!(
                    tx = %self.id,
                    %instance_id,
                    phase = "not_found",
                    "GET: Connection aborted, no peers available - local operation fails"
                );

                // Emit failure event with hop_count
                let hop_count = Some(op_manager.ring.max_hops_to_live.saturating_sub(current_hop));
                if let Some(event) = NetEventLog::get_failure(
                    &self.id,
                    &op_manager.ring,
                    instance_id,
                    OperationFailure::NoPeersAvailable,
                    hop_count,
                ) {
                    op_manager.ring.register_events(Either::Left(event)).await;
                }

                let failed_op = GetOp {
                    id: self.id,
                    state: None,
                    result: None,
                    stats: self.stats,
                    upstream_addr: self.upstream_addr,
                    local_fallback: None,
                    auto_fetch: false,
                    client_return_code: self.client_return_code,
                };
                op_manager.push(self.id, OpEnum::Get(failed_op)).await?;
                return Ok(());
            }
        }

        // If we weren't awaiting a response, just put the op back.
        // No retry needed; another handler may pick it up later.
        op_manager.push(self.id, OpEnum::Get(self)).await?;
        Ok(())
    }

    pub(super) fn finalized(&self) -> bool {
        self.result.is_some() && matches!(self.state, Some(GetState::Finished(_)))
    }

    pub(super) fn to_host_result(&self) -> HostResult {
        match &self.result {
            Some(GetResult {
                key,
                state,
                contract,
            }) => {
                // Strip contract code from client response when the client
                // requested return_contract_code=false. The node still fetches
                // and caches WASM internally for validation/subscription/hosting.
                // See issue #3757.
                let client_contract = if self.client_return_code {
                    contract.clone()
                } else {
                    None
                };
                Ok(HostResponse::ContractResponse(
                    freenet_stdlib::client_api::ContractResponse::GetResponse {
                        key: *key,
                        contract: client_contract,
                        state: state.clone(),
                    },
                ))
            }
            None => Err(ErrorKind::OperationError {
                cause: "get didn't finish successfully".into(),
            }
            .into()),
        }
    }

    /// Get the next hop address if this operation is in a state that needs to send
    /// an outbound message. Used for hop-by-hop routing.
    pub(crate) fn get_next_hop_addr(&self) -> Option<std::net::SocketAddr> {
        match &self.state {
            Some(GetState::AwaitingResponse(data)) => data.next_hop.socket_addr(),
            _ => None,
        }
    }

    /// Get the current hop count (remaining HTL) for this operation.
    /// Returns None if the operation is not in AwaitingResponse state.
    pub(crate) fn get_current_hop(&self) -> Option<usize> {
        match &self.state {
            Some(GetState::AwaitingResponse(data)) => Some(data.current_hop),
            _ => None,
        }
    }

    /// Try to retry this GET operation by sending to the next alternative peer.
    /// If local alternatives are exhausted, `fallback_peers` provides additional
    /// candidates (e.g., all connected peers for DBF search).
    /// Returns `Ok((new_op, msg))` if a peer was available,
    /// `Err(self)` if no peers remain (caller gets ownership back).
    pub(crate) fn retry_with_next_alternative(
        mut self,
        max_hops_to_live: usize,
        fallback_peers: &[PeerKeyLocation],
    ) -> Result<(GetOp, GetMsg), Box<GetOp>> {
        let state = match self.state.take() {
            Some(s) => s,
            None => return Err(Box::new(self)),
        };
        match state {
            GetState::AwaitingResponse(mut data) => {
                // If local alternatives exhausted, inject fallback peers we haven't tried.
                // Filter through BOTH tried_peers (this hop) AND visited bloom filter
                // (all hops) to avoid retry storms cycling through the same peers (#3570).
                if data.alternatives.is_empty() && !fallback_peers.is_empty() {
                    for peer in fallback_peers {
                        if let Some(addr) = peer.socket_addr() {
                            if !data.tried_peers.contains(&addr)
                                && !data.visited.probably_visited(addr)
                            {
                                data.alternatives.push(peer.clone());
                            }
                        }
                    }
                    if !data.alternatives.is_empty() {
                        tracing::info!(
                            tx = %self.id,
                            contract = %data.instance_id,
                            new_candidates = data.alternatives.len(),
                            "GET broadening search to all connected peers (DBF fallback)"
                        );
                    }
                }

                if data.alternatives.is_empty() {
                    self.state = Some(GetState::AwaitingResponse(data));
                    return Err(Box::new(self));
                }

                let next_target = data.alternatives.remove(0);
                let instance_id = data.instance_id;
                let fetch_contract = data.fetch_contract;
                if let Some(addr) = next_target.socket_addr() {
                    data.tried_peers.insert(addr);
                    // Mark in bloom filter so downstream peers won't forward back,
                    // and future retries won't select this peer again (#3570).
                    data.visited.mark_visited(addr);
                }
                tracing::info!(
                    tx = %self.id,
                    contract = %instance_id,
                    target = %next_target,
                    remaining_alternatives = data.alternatives.len(),
                    "GET retrying with alternative peer after timeout"
                );
                // Update stats to point to the new target so timeouts/failures
                // are reported against the correct peer (#3527).
                // Reset timing for the new attempt.
                if let Some(ref mut s) = self.stats {
                    s.next_peer = Some(next_target.clone());
                    s.start_timers();
                }
                data.next_hop = next_target;
                data.attempts_at_hop += 1;
                let visited = data.visited.clone();

                // Reduce HTL on each retry to avoid full-depth traversal storms (#3570).
                // Halve the HTL for each retry attempt, floored at MIN_RETRY_HTL.
                // This limits the blast radius of retries while still allowing the
                // request to reach nearby contract holders.
                let retry_htl = (max_hops_to_live / (data.attempts_at_hop.max(1)))
                    .max(MIN_RETRY_HTL)
                    .min(max_hops_to_live);

                let subscribe = data.subscribe;
                self.state = Some(GetState::AwaitingResponse(data));

                let msg = GetMsg::Request {
                    id: self.id,
                    instance_id,
                    fetch_contract,
                    htl: retry_htl,
                    visited,
                    subscribe,
                };

                Ok((self, msg))
            }
            other @ GetState::ReceivedRequest
            | other @ GetState::PrepareRequest(_)
            | other @ GetState::Finished(_) => {
                self.state = Some(other);
                Err(Box::new(self))
            }
        }
    }
}

impl Operation for GetOp {
    type Message = GetMsg;
    type Result = GetResult;

    async fn load_or_init<'a>(
        op_manager: &'a OpManager,
        msg: &'a Self::Message,
        source_addr: Option<std::net::SocketAddr>,
    ) -> Result<OpInitialization<Self>, OpError> {
        let tx = *msg.id();
        match op_manager.pop(msg.id()) {
            Ok(Some(OpEnum::Get(get_op))) => {
                Ok(OpInitialization {
                    op: get_op,
                    source_addr,
                })
                // was an existing operation, other peer messaged back
            }
            Ok(Some(op)) => {
                if let Err(e) = op_manager.push(tx, op).await {
                    tracing::warn!(tx = %tx, error = %e, "failed to push mismatched op back");
                }
                Err(OpError::OpNotPresent(tx))
            }
            Ok(None) => {
                // Check if this is a response message - if so, the operation was likely
                // cleaned up due to timeout and we should not create a new operation
                if matches!(
                    msg,
                    GetMsg::Response { .. }
                        | GetMsg::ResponseStreaming { .. }
                        | GetMsg::ForwardingAck { .. }
                ) {
                    tracing::debug!(
                        tx = %tx,
                        phase = "load_or_init",
                        "GET response/ack arrived for non-existent operation (likely timed out)"
                    );
                    return Err(OpError::OpNotPresent(tx));
                }

                // new request to get a value for a contract, initialize the machine
                Ok(OpInitialization {
                    op: Self {
                        state: Some(GetState::ReceivedRequest),
                        id: tx,
                        result: None,
                        stats: None, // don't care about stats in target peers
                        upstream_addr: source_addr, // Connection-based routing: store who sent us this request
                        local_fallback: None,       // Remote requests don't have local fallback
                        auto_fetch: false,
                        client_return_code: true,
                    },
                    source_addr,
                })
            }
            Err(err) => Err(err.into()),
        }
    }

    fn id(&self) -> &Transaction {
        &self.id
    }

    fn process_message<'a, NB: NetworkBridge>(
        self,
        conn_manager: &'a mut NB,
        op_manager: &'a OpManager,
        input: &'a Self::Message,
        source_addr: Option<std::net::SocketAddr>,
    ) -> Pin<Box<dyn Future<Output = Result<OperationResult, OpError>> + Send + 'a>> {
        Box::pin(async move {
            // Take fields early since self will be partially moved
            let mut local_fallback = self.local_fallback;
            let auto_fetch = self.auto_fetch;
            #[allow(unused_assignments)]
            let mut return_msg = None;
            #[allow(unused_assignments)]
            let mut new_state = None;
            let mut result = None;
            let mut stats = self.stats;
            let mut stream_data: Option<(StreamId, bytes::Bytes)> = None;

            // Look up sender's PeerKeyLocation from source address for logging/routing
            // This replaces the sender field that was previously embedded in messages
            let sender_from_addr = source_addr.and_then(|addr| {
                op_manager
                    .ring
                    .connection_manager
                    .get_peer_location_by_addr(addr)
            });

            match input {
                GetMsg::Request {
                    instance_id,
                    id,
                    fetch_contract,
                    htl,
                    visited,
                    subscribe: msg_subscribe,
                } => {
                    // Handle GET request - either initial or forwarded
                    let ring_max_htl = op_manager.ring.max_hops_to_live.max(1);
                    let htl = (*htl).min(ring_max_htl);
                    let id = *id;
                    let instance_id = *instance_id;
                    let fetch_contract = *fetch_contract;
                    let msg_subscribe = *msg_subscribe;

                    // Use sender_from_addr for logging (falls back to source_addr if lookup fails)
                    let sender_display = sender_from_addr
                        .as_ref()
                        .map(|s| s.to_string())
                        .unwrap_or_else(|| {
                            source_addr
                                .map(|a| a.to_string())
                                .unwrap_or_else(|| "unknown".to_string())
                        });
                    tracing::info!(
                        tx = %id,
                        %instance_id,
                        peer_addr = %sender_display,
                        fetch_contract,
                        htl,
                        skip = ?visited,
                        phase = "request",
                        "GET Request received"
                    );

                    // Use sender_from_addr if available, but allow processing without it.
                    // The sender may not be in our ring if this is a transient connection
                    // or in simulation where direct sends bypass connection establishment.
                    // Routing uses upstream_addr (stored from source_addr), not sender.
                    let sender = sender_from_addr.clone();

                    // Check if operation is already completed
                    if matches!(self.state, Some(GetState::Finished(_))) {
                        tracing::debug!(
                            tx = %id,
                            "GET: Request received for already completed operation, ignoring duplicate request"
                        );
                        // Return the operation in its current state
                        new_state = self.state;
                        return_msg = None;
                        result = self.result;
                    } else if htl == 0 {
                        // HTL exhausted - send NotFound response
                        tracing::warn!(
                            tx = %id,
                            %instance_id,
                            peer_addr = %sender_display,
                            htl = 0,
                            phase = "not_found",
                            "GET Request exhausted HTL - sending NotFound response"
                        );
                        // Emit telemetry: relay returning NotFound due to HTL exhaustion
                        if let Some(event) = NetEventLog::get_not_found(
                            &id,
                            &op_manager.ring,
                            instance_id,
                            Some(op_manager.ring.max_hops_to_live.saturating_sub(htl)),
                        ) {
                            op_manager.ring.register_events(Either::Left(event)).await;
                        }
                        return build_op_result(GetOpResult {
                            id,
                            state: None,
                            msg: Some(GetMsg::Response {
                                id,
                                instance_id,
                                result: GetMsgResult::NotFound,
                            }),
                            result: None,
                            stats,
                            upstream_addr: self.upstream_addr,
                            stream_data: None,
                            local_fallback: None,
                            auto_fetch: false,
                            client_return_code: self.client_return_code,
                        });
                    } else {
                        // Normal case: operation should be in ReceivedRequest or AwaitingResponse state
                        debug_assert!(matches!(
                            self.state,
                            Some(GetState::ReceivedRequest) | Some(GetState::AwaitingResponse(_))
                        ));
                        tracing::debug!(
                            tx = %id,
                            %instance_id,
                            htl,
                            "GET: Request processing in state {:?}",
                            self.state
                        );

                        // Initialize/update stats for tracking the operation
                        let this_peer = op_manager.ring.connection_manager.own_location();
                        if stats.is_none() {
                            stats = Some(Box::new(GetStats {
                                contract_location: Location::from(&instance_id),
                                next_peer: Some(this_peer.clone()),
                                transfer_time: None,
                                first_response_time: None,
                            }));
                        } else if let Some(s) = stats.as_mut() {
                            s.next_peer = Some(this_peer.clone());
                        }

                        // Update skip list with current peer address and sender address
                        // Restore hash keys after deserialization (they're derived from tx)
                        let mut new_visited = visited.clone().with_transaction(&id);
                        if let Some(addr) = this_peer.socket_addr() {
                            new_visited.mark_visited(addr);
                        }
                        // Also mark the sender to prevent routing back to them
                        // This is critical for preventing request cycles, especially when
                        // a node has limited connections (e.g., only connected to gateway)
                        if let Some(addr) = source_addr {
                            new_visited.mark_visited(addr);
                        }

                        // First check if we have the contract locally before forwarding
                        let get_result = op_manager
                            .notify_contract_handler(ContractHandlerEvent::GetQuery {
                                instance_id,
                                return_contract_code: fetch_contract,
                            })
                            .await;

                        // GetResponse now returns key: Option<ContractKey>
                        let local_value = match get_result {
                            Ok(ContractHandlerEvent::GetResponse {
                                key: Some(key),
                                response:
                                    Ok(StoreResponse {
                                        state: Some(state),
                                        contract,
                                    }),
                            }) => {
                                if fetch_contract && contract.is_none() {
                                    tracing::info!(
                                        tx = %id,
                                        %instance_id,
                                        %this_peer,
                                        fetch_contract,
                                        "GET: state available locally but contract code missing; forwarding GET"
                                    );
                                    None
                                } else {
                                    Some((key, state, contract))
                                }
                            }
                            _ => None,
                        };

                        // Relay peers actively hosting a contract serve it
                        // immediately -- they receive updates via subscriptions,
                        // the interest/proximity protocol, or client connections,
                        // so their state is current. Peers with only a stale LRU
                        // cache entry (no active interest) defer to the network
                        // but keep the local value as fallback.
                        let local_value = if self.upstream_addr.is_some()
                            && matches!(self.state, Some(GetState::ReceivedRequest))
                        {
                            match &local_value {
                                Some((key, ..))
                                    if !op_manager.interest_manager.has_local_interest(key) =>
                                {
                                    // Stale cache only (no active interest) -- defer
                                    // to network but keep as fallback.
                                    if let Some(lv) = local_value {
                                        tracing::debug!(
                                            tx = %id,
                                            "Relay peer deferring stale cache (no local interest), forwarding GET"
                                        );
                                        local_fallback = Some(lv);
                                    }
                                    None
                                }
                                _ => {
                                    // Actively hosting with interest -- serve immediately.
                                    local_value
                                }
                            }
                        } else {
                            local_value
                        };

                        if let Some((key, state, contract)) = local_value {
                            // Contract found locally!
                            tracing::info!(
                                tx = %id,
                                contract = %key,
                                fetch_contract,
                                phase = "complete",
                                "GET: contract found locally"
                            );

                            // Register the GET requester as downstream subscriber when
                            // subscribe=true (piggybacked), or just basic interest otherwise.
                            if let Some(upstream_addr) = self.upstream_addr {
                                if msg_subscribe {
                                    // Full downstream subscriber registration for subscription tree
                                    super::subscribe::register_downstream_subscriber(
                                        op_manager,
                                        &key,
                                        upstream_addr,
                                        None,
                                        None,
                                        &id,
                                        " (hosting peer, piggybacked on GET)",
                                    )
                                    .await;
                                } else if let Some(pkl) = op_manager
                                    .ring
                                    .connection_manager
                                    .get_peer_by_addr(upstream_addr)
                                {
                                    let peer_key =
                                        crate::ring::interest::PeerKey::from(pkl.pub_key.clone());
                                    op_manager
                                        .interest_manager
                                        .register_peer_interest(&key, peer_key, None, false);
                                    tracing::debug!(
                                        tx = %id,
                                        contract = %key,
                                        requester = %upstream_addr,
                                        "Registered GET requester interest for update broadcasts"
                                    );
                                }
                            }

                            // Note: ResponseSent telemetry is emitted by from_outbound_msg()
                            // when the message is actually sent, avoiding duplicate events.

                            // Check if this is a forwarded request or a local request
                            // Use upstream_addr (the actual socket address) not requester (PeerKeyLocation lookup)
                            // because get_peer_location_by_addr() can fail for transient connections
                            match &self.state {
                                Some(GetState::ReceivedRequest) if self.upstream_addr.is_some() => {
                                    // This is a forwarded request - send result back to upstream
                                    tracing::debug!(tx = %id, "Returning contract {} to upstream", key);
                                    new_state = None;
                                    let store_response = StoreResponse {
                                        state: Some(state),
                                        contract,
                                    };
                                    let includes_contract = store_response.contract.is_some();
                                    let payload = GetStreamingPayload {
                                        key,
                                        value: store_response,
                                    };
                                    let payload_bytes =
                                        bincode::serialize(&payload).map_err(|e| {
                                            OpError::NotificationChannelError(format!(
                                                "Failed to serialize streaming payload: {e}"
                                            ))
                                        })?;
                                    let payload_size = payload_bytes.len();
                                    if should_use_streaming(
                                        op_manager.streaming_threshold,
                                        payload_size,
                                    ) {
                                        let sid = StreamId::next_operations();
                                        tracing::info!(
                                            tx = %id,
                                            stream_id = %sid,
                                            payload_size,
                                            "GET response using operations-level streaming"
                                        );
                                        return_msg = Some(GetMsg::ResponseStreaming {
                                            id,
                                            instance_id: *payload.key.id(),
                                            stream_id: sid,
                                            key: payload.key,
                                            total_size: payload_size as u64,
                                            includes_contract,
                                        });
                                        stream_data =
                                            Some((sid, bytes::Bytes::from(payload_bytes)));
                                    } else {
                                        return_msg = Some(GetMsg::Response {
                                            id,
                                            instance_id: *payload.key.id(),
                                            result: GetMsgResult::Found {
                                                key: payload.key,
                                                value: payload.value,
                                            },
                                        });
                                    }
                                }
                                Some(GetState::AwaitingResponse(_))
                                    if self.upstream_addr.is_some() =>
                                {
                                    // Forward contract to upstream
                                    new_state = None;
                                    tracing::debug!(tx = %id, "Returning contract {} to upstream", key);
                                    let store_response = StoreResponse {
                                        state: Some(state),
                                        contract,
                                    };
                                    let includes_contract = store_response.contract.is_some();
                                    let payload = GetStreamingPayload {
                                        key,
                                        value: store_response,
                                    };
                                    let payload_bytes =
                                        bincode::serialize(&payload).map_err(|e| {
                                            OpError::NotificationChannelError(format!(
                                                "Failed to serialize streaming payload: {e}"
                                            ))
                                        })?;
                                    let payload_size = payload_bytes.len();
                                    if should_use_streaming(
                                        op_manager.streaming_threshold,
                                        payload_size,
                                    ) {
                                        let sid = StreamId::next_operations();
                                        tracing::info!(
                                            tx = %id,
                                            stream_id = %sid,
                                            payload_size,
                                            "GET response using operations-level streaming"
                                        );
                                        return_msg = Some(GetMsg::ResponseStreaming {
                                            id,
                                            instance_id: *payload.key.id(),
                                            stream_id: sid,
                                            key: payload.key,
                                            total_size: payload_size as u64,
                                            includes_contract,
                                        });
                                        stream_data =
                                            Some((sid, bytes::Bytes::from(payload_bytes)));
                                    } else {
                                        return_msg = Some(GetMsg::Response {
                                            id,
                                            instance_id: *payload.key.id(),
                                            result: GetMsgResult::Found {
                                                key: payload.key,
                                                value: payload.value,
                                            },
                                        });
                                    }
                                }
                                Some(GetState::AwaitingResponse(AwaitingResponseData {
                                    requester: None,
                                    ..
                                })) => {
                                    // Operation completed for original requester
                                    tracing::debug!(
                                        tx = %id,
                                        "Completed operation, get response received for contract {key}"
                                    );
                                    new_state = Some(GetState::Finished(FinishedData { key }));
                                    return_msg = None;
                                    result = Some(GetResult {
                                        key,
                                        state,
                                        contract,
                                    });
                                }
                                _ => {
                                    // This is the original requester (locally initiated request)
                                    new_state = Some(GetState::Finished(FinishedData { key }));
                                    return_msg = None;
                                    result = Some(GetResult {
                                        key,
                                        state,
                                        contract,
                                    });
                                }
                            }
                        } else {
                            // Contract not found locally (or missing code), proceed with forwarding
                            tracing::debug!(
                                tx = %id,
                                %instance_id,
                                %this_peer,
                                sender = ?sender,
                                "Contract not found @ peer, forwarding to other peers"
                            );

                            // Send ForwardingAck to upstream peer before forwarding.
                            // This tells the upstream "I'm working on it" so the GC task
                            // can distinguish dead peers from slow multi-hop chains.
                            if let Some(upstream) = self.upstream_addr {
                                let ack =
                                    NetMessage::from(GetMsg::ForwardingAck { id, instance_id });
                                drop(conn_manager.send(upstream, ack).await);

                                // Emit telemetry for ForwardingAck sent (#3570 diagnostics)
                                let upstream_peer = op_manager
                                    .ring
                                    .connection_manager
                                    .get_peer_by_addr(upstream)
                                    .unwrap_or_else(|| {
                                        op_manager.ring.connection_manager.own_location()
                                    });
                                if let Some(event) = NetEventLog::get_forwarding_ack_sent(
                                    &id,
                                    &op_manager.ring,
                                    instance_id,
                                    upstream_peer,
                                ) {
                                    op_manager.ring.register_events(Either::Left(event)).await;
                                }
                            }

                            // Forward using standard routing helper
                            // Note: target is determined by routing, sender from source_addr
                            return try_forward_or_return(
                                id,
                                instance_id,
                                (htl, fetch_contract),
                                (this_peer, sender),
                                new_visited,
                                op_manager,
                                stats,
                                self.upstream_addr,
                                local_fallback,
                                msg_subscribe,
                            )
                            .await;
                        }
                    }
                }
                GetMsg::Response {
                    id,
                    instance_id,
                    result: GetMsgResult::NotFound,
                } => {
                    let id = *id;
                    let instance_id = *instance_id;

                    // Use sender_from_addr for logging (optional - sender may not be in ring)
                    let sender = sender_from_addr.clone();

                    tracing::info!(
                        tx = %id,
                        %instance_id,
                        sender = ?sender,
                        phase = "not_found",
                        "GET NotFound response received"
                    );
                    // Handle case where contract was not found
                    let this_peer = op_manager.ring.connection_manager.own_location();
                    tracing::warn!(
                        tx = %id,
                        %instance_id,
                        peer_addr = %this_peer,
                        sender = ?sender,
                        phase = "retry",
                        "Contract not found at peer, retrying with other peers"
                    );

                    match self.state {
                        Some(GetState::AwaitingResponse(AwaitingResponseData {
                            fetch_contract,
                            retries,
                            requester,
                            current_hop,
                            subscribe,
                            blocking_subscribe,
                            mut tried_peers,
                            mut alternatives,
                            attempts_at_hop,
                            next_hop: _,
                            visited,
                            instance_id: state_instance_id,
                            ..
                        })) => {
                            // Use instance_id from state for consistency
                            let instance_id = state_instance_id;

                            // Add the failed peer to tried list
                            if let Some(addr) = sender.as_ref().and_then(|s| s.socket_addr()) {
                                tried_peers.insert(addr);
                            }

                            // First, check if we have alternatives at this hop level
                            if !alternatives.is_empty() && attempts_at_hop < DEFAULT_MAX_BREADTH {
                                // Try the next alternative
                                let next_target = alternatives.remove(0);

                                tracing::info!(
                                    tx = %id,
                                    %instance_id,
                                    peer_addr = %next_target,
                                    fetch_contract,
                                    attempts_at_hop = attempts_at_hop + 1,
                                    max_attempts = DEFAULT_MAX_BREADTH,
                                    tried = ?tried_peers,
                                    remaining_alternatives = ?alternatives,
                                    phase = "retry",
                                    "Trying alternative peer at same hop level"
                                );

                                return_msg = Some(GetMsg::Request {
                                    id,
                                    instance_id,
                                    fetch_contract,
                                    htl: current_hop,
                                    visited: visited.clone(),
                                    subscribe,
                                });

                                // Update state with the new alternative being tried
                                if let Some(addr) = next_target.socket_addr() {
                                    tried_peers.insert(addr);
                                }
                                let updated_tried_peers = tried_peers.clone();
                                new_state =
                                    Some(GetState::AwaitingResponse(AwaitingResponseData {
                                        retries,
                                        fetch_contract,
                                        requester: requester.clone(),
                                        current_hop,
                                        subscribe,
                                        blocking_subscribe,
                                        tried_peers: updated_tried_peers.clone(),
                                        alternatives,
                                        attempts_at_hop: attempts_at_hop + 1,
                                        instance_id,
                                        next_hop: next_target,
                                        // Preserve the accumulated visited so future candidate
                                        // selection still avoids already-specified peers; tried_peers
                                        // tracks attempts at this hop.
                                        visited: visited.clone(),
                                    }));
                            } else if retries < MAX_RETRIES {
                                // No more alternatives at this hop, try finding new peers
                                let mut new_visited = visited.clone();
                                for addr in &tried_peers {
                                    new_visited.mark_visited(*addr);
                                }

                                // Get new candidates excluding all tried peers
                                let mut new_candidates =
                                    op_manager.ring.k_closest_potentially_hosting(
                                        &instance_id,
                                        &new_visited,
                                        DEFAULT_MAX_BREADTH,
                                    );

                                tracing::info!(
                                    tx = %id,
                                    %instance_id,
                                    new_candidates = ?new_candidates,
                                    skip = ?new_visited,
                                    htl = current_hop,
                                    retries = retries + 1,
                                    phase = "retry",
                                    "GET seeking new candidates after exhausted alternatives"
                                );

                                if !new_candidates.is_empty() {
                                    // Try with the best new peer
                                    let target = new_candidates.remove(0);
                                    return_msg = Some(GetMsg::Request {
                                        id,
                                        instance_id,
                                        fetch_contract,
                                        htl: current_hop,
                                        visited: new_visited.clone(),
                                        subscribe,
                                    });

                                    // Reset for new round of attempts
                                    let mut new_tried_peers = HashSet::new();
                                    if let Some(addr) = target.socket_addr() {
                                        new_tried_peers.insert(addr);
                                    }

                                    new_state =
                                        Some(GetState::AwaitingResponse(AwaitingResponseData {
                                            retries: retries + 1,
                                            fetch_contract,
                                            requester: requester.clone(),
                                            current_hop,
                                            subscribe,
                                            blocking_subscribe,
                                            tried_peers: new_tried_peers,
                                            alternatives: new_candidates,
                                            attempts_at_hop: 1,
                                            instance_id,
                                            next_hop: target,
                                            visited: new_visited.clone(),
                                        }));
                                } else if let Some(requester_peer) = requester.clone() {
                                    // No more peers to try — serve local fallback if available
                                    if let Some((key, state, contract)) = local_fallback.take() {
                                        tracing::info!(
                                            tx = %id,
                                            contract = %key,
                                            "Relay serving local cache as fallback (network returned NotFound)"
                                        );
                                        return_msg = Some(build_fallback_found_response(
                                            id,
                                            instance_id,
                                            key,
                                            state,
                                            contract,
                                        ));
                                    } else {
                                        tracing::warn!(
                                            tx = %id,
                                            %instance_id,
                                            peer_addr = %this_peer,
                                            target = %requester_peer,
                                            tried = ?tried_peers,
                                            skip = ?new_visited,
                                            phase = "not_found",
                                            "No other peers found while trying to get the contract, returning NotFound to requester"
                                        );
                                        return_msg = Some(GetMsg::Response {
                                            id,
                                            instance_id,
                                            result: GetMsgResult::NotFound,
                                        });
                                    }
                                } else {
                                    // Original requester - check for local fallback before failing
                                    if let Some((key, state, contract)) = local_fallback.take() {
                                        tracing::info!(
                                            tx = %id,
                                            contract = %key,
                                            phase = "complete",
                                            "GET: Network returned NotFound, returning local cache and re-hosting to network"
                                        );

                                        // Complete with local cached version
                                        new_state = Some(GetState::Finished(FinishedData { key }));
                                        return_msg = None;
                                        result = Some(GetResult {
                                            key,
                                            state: state.clone(),
                                            contract: contract.clone(),
                                        });

                                        // Re-host to the network with our local copy
                                        if let Some(contract_code) = contract {
                                            tracing::info!(
                                                tx = %id,
                                                %key,
                                                "Re-publishing to network after NotFound response"
                                            );
                                            let put_result = op_manager
                                                .notify_contract_handler(
                                                    ContractHandlerEvent::PutQuery {
                                                        key,
                                                        state,
                                                        related_contracts:
                                                            RelatedContracts::default(),
                                                        contract: Some(contract_code),
                                                    },
                                                )
                                                .await;
                                            match put_result {
                                                Ok(ContractHandlerEvent::PutResponse {
                                                    new_value: Ok(_),
                                                    ..
                                                }) => {
                                                    tracing::debug!(tx = %id, %key, "Re-hosted contract to network");
                                                    super::announce_contract_hosted(
                                                        op_manager, &key,
                                                    )
                                                    .await;
                                                }
                                                _ => {
                                                    tracing::warn!(tx = %id, %key, "Failed to re-host contract");
                                                }
                                            }
                                        }
                                    } else {
                                        // No local fallback - operation failed
                                        tracing::error!(
                                            tx = %id,
                                            %instance_id,
                                            tried = ?tried_peers,
                                            skip = ?visited,
                                            phase = "not_found",
                                            "Failed getting contract, not found after max retries"
                                        );

                                        // Emit get_not_found telemetry
                                        let hop_count = Some(
                                            op_manager
                                                .ring
                                                .max_hops_to_live
                                                .saturating_sub(current_hop),
                                        );
                                        if let Some(event) = NetEventLog::get_not_found(
                                            &id,
                                            &op_manager.ring,
                                            instance_id,
                                            hop_count,
                                        ) {
                                            op_manager
                                                .ring
                                                .register_events(Either::Left(event))
                                                .await;
                                        }

                                        notify_get_failed(
                                            op_manager,
                                            id,
                                            &format!(
                                                "contract {instance_id} not found after exhausting all peers"
                                            ),
                                        );
                                        return_msg = None;
                                        new_state = None;
                                    }
                                }
                            } else {
                                // Max retries reached
                                tracing::error!(
                                    tx = %id,
                                    %instance_id,
                                    phase = "not_found",
                                    "Failed getting contract, reached max retries"
                                );

                                if let Some(requester_peer) = requester.clone() {
                                    // No more peers to try — serve local fallback if available
                                    if let Some((key, state, contract)) = local_fallback.take() {
                                        tracing::info!(
                                            tx = %id,
                                            contract = %key,
                                            "Relay serving local cache as fallback (max retries, network returned NotFound)"
                                        );
                                        return_msg = Some(build_fallback_found_response(
                                            id,
                                            instance_id,
                                            key,
                                            state,
                                            contract,
                                        ));
                                    } else {
                                        tracing::warn!(
                                            tx = %id,
                                            %instance_id,
                                            peer_addr = %this_peer,
                                            target = %requester_peer,
                                            tried = ?tried_peers,
                                            skip = ?visited,
                                            phase = "not_found",
                                            "No other peers found, returning NotFound to requester"
                                        );
                                        return_msg = Some(GetMsg::Response {
                                            id,
                                            instance_id,
                                            result: GetMsgResult::NotFound,
                                        });
                                    }
                                    new_state = None;
                                } else {
                                    // Original requester - check for local fallback before failing
                                    if let Some((key, state, contract)) = local_fallback.take() {
                                        tracing::info!(
                                            tx = %id,
                                            contract = %key,
                                            phase = "complete",
                                            "GET: Max retries reached, returning local cache and re-hosting to network"
                                        );

                                        // Complete with local cached version
                                        new_state = Some(GetState::Finished(FinishedData { key }));
                                        return_msg = None;
                                        result = Some(GetResult {
                                            key,
                                            state: state.clone(),
                                            contract: contract.clone(),
                                        });

                                        // Re-host to the network with our local copy
                                        if let Some(contract_code) = contract {
                                            tracing::info!(
                                                tx = %id,
                                                %key,
                                                "Re-publishing to network after NotFound (max retries)"
                                            );
                                            let put_result = op_manager
                                                .notify_contract_handler(
                                                    ContractHandlerEvent::PutQuery {
                                                        key,
                                                        state,
                                                        related_contracts:
                                                            RelatedContracts::default(),
                                                        contract: Some(contract_code),
                                                    },
                                                )
                                                .await;
                                            match put_result {
                                                Ok(ContractHandlerEvent::PutResponse {
                                                    new_value: Ok(_),
                                                    ..
                                                }) => {
                                                    tracing::debug!(tx = %id, %key, "Re-hosted contract to network");
                                                    super::announce_contract_hosted(
                                                        op_manager, &key,
                                                    )
                                                    .await;
                                                }
                                                _ => {
                                                    tracing::warn!(tx = %id, %key, "Failed to re-host contract");
                                                }
                                            }
                                        }
                                    } else {
                                        // No local fallback - operation failed
                                        tracing::error!(
                                            tx = %id,
                                            %instance_id,
                                            phase = "not_found",
                                            "Failed getting contract, reached max retries"
                                        );

                                        // Emit get_not_found telemetry
                                        let hop_count = Some(
                                            op_manager
                                                .ring
                                                .max_hops_to_live
                                                .saturating_sub(current_hop),
                                        );
                                        if let Some(event) = NetEventLog::get_not_found(
                                            &id,
                                            &op_manager.ring,
                                            instance_id,
                                            hop_count,
                                        ) {
                                            op_manager
                                                .ring
                                                .register_events(Either::Left(event))
                                                .await;
                                        }

                                        notify_get_failed(
                                            op_manager,
                                            id,
                                            &format!(
                                                "contract {instance_id} not found after max retries"
                                            ),
                                        );
                                        return_msg = None;
                                        new_state = None;
                                    }
                                }
                            }
                        }
                        Some(GetState::ReceivedRequest) => {
                            // Return NotFound to sender
                            tracing::debug!(tx = %id, %instance_id, sender = ?sender, "Returning NotFound to upstream");
                            new_state = None;
                            return_msg = Some(GetMsg::Response {
                                id,
                                instance_id,
                                result: GetMsgResult::NotFound,
                            });
                        }
                        _ => return Err(OpError::invalid_transition(self.id)),
                    };
                }
                GetMsg::Response {
                    id,
                    result:
                        GetMsgResult::Found {
                            key,
                            value:
                                StoreResponse {
                                    state: Some(value),
                                    contract,
                                },
                        },
                    ..
                } => {
                    let id = *id;
                    let key = *key;

                    // Use sender_from_addr for logging (optional - sender may not be in ring)
                    let sender = sender_from_addr.clone();

                    tracing::info!(tx = %id, contract = %key, sender = ?sender, state = ?self.state, phase = "response", "GET Response received with state");

                    // Check if contract is required
                    let require_contract = matches!(
                        self.state,
                        Some(GetState::AwaitingResponse(AwaitingResponseData {
                            fetch_contract: true,
                            ..
                        }))
                    );

                    // Get requester and current_hop from current state
                    let (requester, current_hop) =
                        if let Some(GetState::AwaitingResponse(data)) = self.state.as_ref() {
                            (data.requester.clone(), Some(data.current_hop))
                        } else {
                            return Err(OpError::UnexpectedOpState);
                        };

                    // Handle case where contract is required but not provided
                    if require_contract && contract.is_none() {
                        if let Some(_requester) = requester {
                            // no contract, consider this like an error ignoring the incoming update value
                            tracing::warn!(
                                tx = %id,
                                sender = ?sender,
                                "Contract not received from peer while required"
                            );

                            tracing::warn!(
                                tx = %id,
                                %key,
                                sender = ?sender,
                                "Contract not received while required, returning response to upstream",
                            );

                            // Forward NotFound to requester (contract was required but not provided)
                            // local_fallback is only for original requester, not for forwarding
                            op_manager
                                .notify_op_change(
                                    NetMessage::from(GetMsg::Response {
                                        id,
                                        instance_id: *key.id(),
                                        result: GetMsgResult::NotFound,
                                    }),
                                    OpEnum::Get(GetOp {
                                        id,
                                        state: self.state,
                                        result: None,
                                        stats,
                                        upstream_addr: self.upstream_addr,
                                        local_fallback: None,
                                        auto_fetch: false,
                                        client_return_code: self.client_return_code,
                                    }),
                                )
                                .await?;
                            return Err(OpError::StatePushed);
                        }
                    }

                    // Check if this is the original requester (no upstream to forward to)
                    let is_original_requester = self.upstream_addr.is_none();

                    // Record response timing for the originator (non-streaming:
                    // response + payload arrive together, so both timers end now)
                    if is_original_requester {
                        if let Some(ref mut s) = stats {
                            s.record_response_end();
                            s.record_transfer_end();
                        }
                    }

                    // Check if subscription was requested
                    let (subscribe_requested, blocking_sub) =
                        if let Some(GetState::AwaitingResponse(data)) = &self.state {
                            (data.subscribe, data.blocking_subscribe)
                        } else {
                            (false, false)
                        };

                    // Always cache contracts we encounter - LRU will handle eviction
                    let should_put = true;

                    // Put contract locally if needed
                    if should_put {
                        // First check if the local state matches the incoming state
                        // to avoid triggering validation errors in contracts that implement
                        // idempotency checks in their update_state() method (issue #2018)
                        let local_state = op_manager
                            .notify_contract_handler(ContractHandlerEvent::GetQuery {
                                instance_id: *key.id(),
                                return_contract_code: false,
                            })
                            .await;

                        let state_matches = match local_state {
                            Ok(ContractHandlerEvent::GetResponse {
                                response:
                                    Ok(StoreResponse {
                                        state: Some(local), ..
                                    }),
                                ..
                            }) => {
                                // Compare the actual state bytes
                                local.as_ref() == value.as_ref()
                            }
                            _ => false, // No local state or error - we should try to cache
                        };

                        if state_matches {
                            tracing::debug!(
                                tx = %id,
                                %key,
                                "Local state matches network state, skipping redundant cache"
                            );

                            // BUG FIX (2026-01): ALWAYS refresh hosting status on GET.
                            // Previously, we only recorded access if !is_hosting_contract(),
                            // which meant re-GETs on cached contracts wouldn't refresh the
                            // hosting cache's TTL/LRU position, causing subscriptions to expire.
                            //
                            // record_get_access now returns is_new atomically, eliminating the
                            // TOCTOU race that existed when checking is_hosting_contract() first.
                            let access_result =
                                op_manager.ring.record_get_access(key, value.size() as u64);

                            // Mark locally-accessed for subscription renewal (#3769)
                            if is_original_requester {
                                op_manager.ring.mark_local_client_access(&key);
                            }

                            // Clean up interest tracking for evicted contracts (always, even if already hosting)
                            let mut removed_contracts = Vec::new();
                            for evicted_key in &access_result.evicted {
                                if op_manager
                                    .interest_manager
                                    .unregister_local_hosting(evicted_key)
                                {
                                    removed_contracts.push(*evicted_key);
                                }
                            }

                            // Only do first-time hosting setup if newly hosting
                            if access_result.is_new {
                                tracing::debug!(tx = %id, %key, "Contract newly hosted");
                                super::announce_contract_hosted(op_manager, &key).await;

                                // Register local interest for delta-based sync
                                let became_interested =
                                    op_manager.interest_manager.register_local_hosting(&key);

                                // Broadcast interest changes to peers
                                let added = if became_interested { vec![key] } else { vec![] };
                                if !added.is_empty() || !removed_contracts.is_empty() {
                                    super::broadcast_change_interests(
                                        op_manager,
                                        added,
                                        removed_contracts,
                                    )
                                    .await;
                                }
                            } else {
                                tracing::debug!(tx = %id, %key, "Refreshed hosting status for already-hosted contract");
                                // Still broadcast eviction changes if any contracts were evicted
                                if !removed_contracts.is_empty() {
                                    super::broadcast_change_interests(
                                        op_manager,
                                        vec![],
                                        removed_contracts,
                                    )
                                    .await;
                                }
                            }

                            // Auto-subscribe: piggyback or fallback
                            if crate::ring::AUTO_SUBSCRIBE_ON_GET
                                && is_original_requester
                                && (access_result.is_new || !op_manager.ring.is_subscribed(&key))
                            {
                                super::auto_subscribe_on_get_response(
                                    op_manager,
                                    &key,
                                    &id,
                                    &sender_from_addr,
                                    subscribe_requested,
                                    blocking_sub,
                                    "non-streaming",
                                )
                                .await;
                            }
                        } else {
                            // Only attempt to cache if we have the contract code.
                            // Without the code, we can't store the contract locally (issue #2306).
                            if contract.is_some() {
                                tracing::debug!(tx = %id, %key, %is_original_requester, %subscribe_requested, "Putting contract at executor - state differs from local cache");
                                let res = op_manager
                                    .notify_contract_handler(ContractHandlerEvent::PutQuery {
                                        key,
                                        state: value.clone(),
                                        related_contracts: RelatedContracts::default(), // fixme: i think we need to get the related contracts so the final put is ok
                                        contract: contract.clone(),
                                    })
                                    .await?;

                                match res {
                                    ContractHandlerEvent::PutResponse {
                                        new_value: Ok(_), ..
                                    } => {
                                        tracing::debug!(tx = %id, %key, "Contract put at executor");

                                        // BUG FIX (2026-01): ALWAYS refresh hosting status on GET.
                                        // This ensures re-GETs keep the hosting cache's TTL/LRU fresh.
                                        //
                                        // record_get_access now returns is_new atomically, eliminating the
                                        // TOCTOU race that existed when checking is_hosting_contract() first.
                                        tracing::debug!(tx = %id, %key, peer = ?op_manager.ring.connection_manager.get_own_addr(), "Recording contract access in hosting cache");
                                        let access_result = op_manager
                                            .ring
                                            .record_get_access(key, value.size() as u64);

                                        if is_original_requester {
                                            op_manager.ring.mark_local_client_access(&key);
                                        }

                                        // Clean up interest tracking for evicted contracts (always, even if already hosting)
                                        let mut removed_contracts = Vec::new();
                                        for evicted_key in &access_result.evicted {
                                            if op_manager
                                                .interest_manager
                                                .unregister_local_hosting(evicted_key)
                                            {
                                                removed_contracts.push(*evicted_key);
                                            }
                                        }

                                        // Only do first-time hosting setup if newly hosting
                                        if access_result.is_new {
                                            super::announce_contract_hosted(op_manager, &key).await;

                                            // Register local interest for delta-based sync
                                            let became_interested = op_manager
                                                .interest_manager
                                                .register_local_hosting(&key);

                                            // Broadcast interest changes to peers
                                            let added =
                                                if became_interested { vec![key] } else { vec![] };
                                            if !added.is_empty() || !removed_contracts.is_empty() {
                                                super::broadcast_change_interests(
                                                    op_manager,
                                                    added,
                                                    removed_contracts,
                                                )
                                                .await;
                                            }
                                        } else {
                                            tracing::debug!(tx = %id, %key, "Refreshed hosting status for already-hosted contract");
                                            // Still broadcast eviction changes if any contracts were evicted
                                            if !removed_contracts.is_empty() {
                                                super::broadcast_change_interests(
                                                    op_manager,
                                                    vec![],
                                                    removed_contracts,
                                                )
                                                .await;
                                            }
                                        }

                                        // Auto-subscribe: piggyback or fallback
                                        if crate::ring::AUTO_SUBSCRIBE_ON_GET
                                            && is_original_requester
                                            && (access_result.is_new
                                                || !op_manager.ring.is_subscribed(&key))
                                        {
                                            super::auto_subscribe_on_get_response(
                                                op_manager,
                                                &key,
                                                &id,
                                                &sender_from_addr,
                                                subscribe_requested,
                                                blocking_sub,
                                                "non-streaming, put-path",
                                            )
                                            .await;
                                        }
                                    }
                                    ContractHandlerEvent::PutResponse {
                                        new_value: Err(err),
                                        ..
                                    } => {
                                        // Local caching failed, but GET operation succeeded
                                        // Log warning and continue - caching is an optimization, not required
                                        tracing::warn!(
                                            tx = %id,
                                            %key,
                                            error = %err,
                                            %is_original_requester,
                                            "Failed to cache contract locally during GET - continuing with operation"
                                        );
                                        // Don't return error - the GET succeeded, caching is optional
                                        // Continue to process the GET result below
                                    }
                                    ContractHandlerEvent::DelegateRequest { .. }
                                    | ContractHandlerEvent::DelegateResponse(_)
                                    | ContractHandlerEvent::PutQuery { .. }
                                    | ContractHandlerEvent::GetQuery { .. }
                                    | ContractHandlerEvent::GetResponse { .. }
                                    | ContractHandlerEvent::UpdateQuery { .. }
                                    | ContractHandlerEvent::UpdateResponse { .. }
                                    | ContractHandlerEvent::UpdateNoChange { .. }
                                    | ContractHandlerEvent::RegisterSubscriberListener { .. }
                                    | ContractHandlerEvent::RegisterSubscriberListenerResponse
                                    | ContractHandlerEvent::QuerySubscriptions { .. }
                                    | ContractHandlerEvent::QuerySubscriptionsResponse
                                    | ContractHandlerEvent::GetSummaryQuery { .. }
                                    | ContractHandlerEvent::GetSummaryResponse { .. }
                                    | ContractHandlerEvent::GetDeltaQuery { .. }
                                    | ContractHandlerEvent::GetDeltaResponse { .. }
                                    | ContractHandlerEvent::NotifySubscriptionError { .. }
                                    | ContractHandlerEvent::NotifySubscriptionErrorResponse
                                    | ContractHandlerEvent::ClientDisconnect { .. } => {
                                        unreachable!(
                                            "PutQuery from Get operation should always return PutResponse"
                                        )
                                    }
                                }
                            } else {
                                // No contract code in GET response - can't cache or subscribe locally.
                                // Without the contract code, we can't:
                                // 1. Validate incoming state updates
                                // 2. Process delta updates from subscriptions
                                // Announcing as cached or subscribing would cause #2306 errors when
                                // updates arrive for a contract we can't process.
                                tracing::warn!(
                                    tx = %id,
                                    %key,
                                    "Cannot cache or subscribe to contract - no contract code in GET response"
                                );
                                // Don't announce cached or start subscription without contract code.
                                // The GET still succeeded - the client gets their state response.
                            }
                        }
                    }

                    // Process based on whether we're originator or forwarding
                    // Use upstream_addr for routing decision (not requester which can fail for transient connections)
                    // First validate we're in an expected state
                    match &self.state {
                        Some(GetState::AwaitingResponse(_)) | Some(GetState::ReceivedRequest) => {}
                        Some(other) => {
                            // Can't take ownership, so format the state for error reporting
                            return Err(OpError::invalid_transition_with_state(
                                self.id,
                                Box::new(format!("{}", other)),
                            ));
                        }
                        None => return Err(OpError::invalid_transition(self.id)),
                    }

                    // Now handle based on upstream_addr
                    if self.upstream_addr.is_none() {
                        // Original requester, operation completed successfully
                        tracing::info!(tx = %id, contract = %key, phase = "complete", "Get response received for contract at original requester");

                        // Emit get_success telemetry (only if sender is known)
                        let hop_count =
                            current_hop.map(|h| op_manager.ring.max_hops_to_live.saturating_sub(h));
                        let hash = Some(state_hash_full(value));
                        if let Some(sender_peer) = sender {
                            if let Some(event) = NetEventLog::get_success(
                                &id,
                                &op_manager.ring,
                                key,
                                sender_peer,
                                hop_count,
                                hash,
                            ) {
                                op_manager.ring.register_events(Either::Left(event)).await;
                            }
                        }

                        new_state = Some(GetState::Finished(FinishedData { key }));
                        return_msg = None;
                        result = Some(GetResult {
                            key,
                            state: value.clone(),
                            contract: contract.clone(),
                        });
                    } else {
                        // Forward response to upstream. Opportunistically cache contract
                        // code so this relay peer can serve future GET requests with
                        // fetch_contract=true. We don't call announce_contract_hosted()
                        // here because relay peers should not advertise contracts they
                        // are not responsible for in the ring.
                        tracing::info!(tx = %id, contract = %key, phase = "response", "Get response received for contract at hop peer");

                        // Piggyback subscription: set up forwarding at this relay node
                        // so UPDATE broadcasts propagate through the subscription tree.
                        // Only forwarding (upstream/downstream registration) -- NO
                        // ring.subscribe() or announce_contract_hosted() to avoid storms.
                        if subscribe_requested {
                            if let (Some(response_addr), Some(requester_addr)) =
                                (source_addr, self.upstream_addr)
                            {
                                super::setup_subscription_forwarding_at_relay(
                                    op_manager,
                                    &key,
                                    &id,
                                    response_addr,
                                    requester_addr,
                                )
                                .await;
                            }
                        }
                        if let Some(code) = contract {
                            if !op_manager.ring.is_hosting_contract(&key) {
                                let om = op_manager.clone();
                                let cache_key = key;
                                let v = value.clone();
                                let c = code.clone();
                                GlobalExecutor::spawn(async move {
                                    let key_str = cache_key.to_string();
                                    let cached = matches!(
                                        om.notify_contract_handler(
                                            ContractHandlerEvent::PutQuery {
                                                key: cache_key,
                                                state: v,
                                                related_contracts: RelatedContracts::default(),
                                                contract: Some(c),
                                            }
                                        )
                                        .await,
                                        Ok(ContractHandlerEvent::PutResponse {
                                            new_value: Ok(_),
                                            ..
                                        })
                                    );
                                    tracing::info!(
                                        contract = %key_str,
                                        phase = "relay_cache",
                                        cached,
                                        "Relay peer contract code cache attempt"
                                    );
                                });
                            }
                        }
                        new_state = None;
                        // Check if the response payload exceeds the streaming
                        // threshold.  When it does, convert to ResponseStreaming
                        // so that upstream relays can use pipe_stream instead of
                        // serial store-and-forward (which adds 10-50 s per hop
                        // for large contracts like the River UI at ~814 KB).
                        let store_response = StoreResponse {
                            state: Some(value.clone()),
                            contract: contract.clone(),
                        };
                        let includes_contract = store_response.contract.is_some();
                        let payload = GetStreamingPayload {
                            key,
                            value: store_response,
                        };
                        let payload_bytes = bincode::serialize(&payload).map_err(|e| {
                            OpError::NotificationChannelError(format!(
                                "Failed to serialize streaming payload: {e}"
                            ))
                        })?;
                        let payload_size = payload_bytes.len();
                        if should_use_streaming(op_manager.streaming_threshold, payload_size) {
                            let sid = StreamId::next_operations();
                            tracing::info!(
                                tx = %id,
                                stream_id = %sid,
                                payload_size,
                                phase = "relay_streaming_forward",
                                "Relay converting non-streaming GET response to streaming for upstream"
                            );
                            return_msg = Some(GetMsg::ResponseStreaming {
                                id,
                                instance_id: *payload.key.id(),
                                stream_id: sid,
                                key: payload.key,
                                total_size: payload_size as u64,
                                includes_contract,
                            });
                            stream_data = Some((sid, bytes::Bytes::from(payload_bytes)));
                        } else {
                            return_msg = Some(GetMsg::Response {
                                id,
                                instance_id: *payload.key.id(),
                                result: GetMsgResult::Found {
                                    key: payload.key,
                                    value: payload.value,
                                },
                            });
                        }
                        tracing::debug!(tx = %id, %key, upstream = ?self.upstream_addr, "Returning contract to upstream");
                        result = Some(GetResult {
                            key,
                            state: value.clone(),
                            contract: contract.clone(),
                        });
                    }
                }
                // DEFENSIVE HANDLING: Found response with empty state
                //
                // This case should not occur in normal operation because:
                // 1. Contracts are always stored with their state
                // 2. If a peer has the contract, it should have state data
                //
                // However, we handle this defensively to avoid undefined behavior if:
                // - A malformed response is received from a peer
                // - There's a race condition during contract storage
                //
                // We treat this as NotFound because a Found response without usable
                // state data is equivalent to not finding the contract.
                GetMsg::Response {
                    id,
                    instance_id,
                    result:
                        GetMsgResult::Found {
                            value: StoreResponse { state: None, .. },
                            ..
                        },
                } => {
                    let id = *id;
                    let instance_id = *instance_id;

                    // Use sender_from_addr for logging (optional - sender may not be in ring)
                    let sender = sender_from_addr.clone();

                    tracing::warn!(
                        tx = %id,
                        %instance_id,
                        sender = ?sender,
                        phase = "response",
                        "GET Found response with empty state - treating as NotFound"
                    );

                    // Forward as NotFound since Found with empty state isn't useful
                    if self.upstream_addr.is_some() {
                        return_msg = Some(GetMsg::Response {
                            id,
                            instance_id,
                            result: GetMsgResult::NotFound,
                        });
                        new_state = None;
                    } else {
                        // Original requester - operation failed
                        return_msg = None;
                        new_state = None;
                        // result remains None to indicate failure
                    }
                }

                // Streaming GET response handler
                GetMsg::ResponseStreaming {
                    id: msg_id,
                    instance_id,
                    stream_id,
                    key,
                    total_size,
                    includes_contract,
                } => {
                    let id = *msg_id;
                    let key = *key;
                    let stream_id = *stream_id;
                    let includes_contract = *includes_contract;

                    tracing::info!(
                        tx = %id,
                        %instance_id,
                        contract = %key,
                        stream_id = %stream_id,
                        total_size,
                        includes_contract,
                        phase = "streaming_response",
                        "Processing GET ResponseStreaming"
                    );

                    // Step 1: Claim the stream from orphan registry (atomic dedup)
                    let peer_addr = match source_addr {
                        Some(addr) => addr,
                        None => {
                            tracing::error!(tx = %id, "source_addr missing for streaming GET response");
                            return Err(OpError::UnexpectedOpState);
                        }
                    };
                    let stream_handle = match op_manager
                        .orphan_stream_registry()
                        .claim_or_wait(peer_addr, stream_id, STREAM_CLAIM_TIMEOUT)
                        .await
                    {
                        Ok(handle) => handle,
                        Err(OrphanStreamError::AlreadyClaimed) => {
                            tracing::debug!(
                                tx = %id,
                                stream_id = %stream_id,
                                "GET ResponseStreaming skipped — stream already claimed (dedup)"
                            );
                            // Push the operation state back since load_or_init popped it.
                            // Without this, duplicate metadata messages (from embedded fragment #1)
                            // permanently lose the operation state, preventing subsequent
                            // ResponseStreaming from being matched to the operation.
                            if self.state.is_some() {
                                if let Err(e) = op_manager
                                    .push(
                                        id,
                                        OpEnum::Get(GetOp {
                                            id,
                                            state: self.state,
                                            result: self.result,
                                            stats,
                                            upstream_addr: self.upstream_addr,
                                            local_fallback,
                                            auto_fetch: false,
                                            client_return_code: self.client_return_code,
                                        }),
                                    )
                                    .await
                                {
                                    tracing::warn!(tx = %id, error = %e, "failed to push GET op state back after dedup");
                                }
                            }
                            return Err(OpError::OpNotPresent(id));
                        }
                        Err(e) => {
                            tracing::error!(
                                tx = %id,
                                stream_id = %stream_id,
                                error = %e,
                                "Failed to claim stream from orphan registry"
                            );
                            // Push the operation state back to prevent loss
                            if self.state.is_some() {
                                if let Err(e) = op_manager
                                    .push(
                                        id,
                                        OpEnum::Get(GetOp {
                                            id,
                                            state: self.state,
                                            result: self.result,
                                            stats,
                                            upstream_addr: self.upstream_addr,
                                            local_fallback,
                                            auto_fetch: false,
                                            client_return_code: self.client_return_code,
                                        }),
                                    )
                                    .await
                                {
                                    tracing::warn!(tx = %id, error = %e, "failed to push GET op state back after orphan claim failure");
                                }
                            }
                            return Err(OpError::OrphanStreamClaimFailed);
                        }
                    };

                    // Step 2: Check if we should pipe upstream BEFORE assembly
                    // This enables low-latency forwarding of responses
                    let is_original_requester = self.upstream_addr.is_none();

                    // Record first-response timing for originator (streaming metadata arrived;
                    // transfer_time ends later after stream assembly completes)
                    if is_original_requester {
                        if let Some(ref mut s) = stats {
                            s.record_response_end();
                        }
                    }

                    // Extract subscription flags from state (needed for auto-subscribe after caching)
                    let (subscribe_requested, blocking_sub) =
                        if let Some(GetState::AwaitingResponse(data)) = &self.state {
                            (data.subscribe, data.blocking_subscribe)
                        } else {
                            (false, false)
                        };
                    let piping_started = if !is_original_requester {
                        let upstream_addr = self
                            .upstream_addr
                            .expect("non-originator must have upstream");
                        // Check if streaming should be used based on the original stream size
                        if should_use_streaming(
                            op_manager.streaming_threshold,
                            *total_size as usize,
                        ) {
                            let outbound_sid = StreamId::next_operations();
                            let forked_handle = stream_handle.fork();

                            tracing::info!(
                                tx = %id,
                                inbound_stream_id = %stream_id,
                                outbound_stream_id = %outbound_sid,
                                total_size,
                                peer_addr = %upstream_addr,
                                "Starting piped stream forwarding of GET response to upstream"
                            );

                            // Send metadata message first
                            let pipe_metadata = GetMsg::ResponseStreaming {
                                id,
                                instance_id: *instance_id,
                                stream_id: outbound_sid,
                                key,
                                total_size: *total_size,
                                includes_contract,
                            };
                            let pipe_metadata_net: NetMessage = pipe_metadata.into();
                            // Serialize metadata for embedding in fragment #1 (fix #2757)
                            let embedded_metadata = match bincode::serialize(&pipe_metadata_net) {
                                Ok(bytes) => Some(bytes::Bytes::from(bytes)),
                                Err(e) => {
                                    tracing::warn!(
                                        tx = %id,
                                        error = %e,
                                        "Failed to serialize piped stream metadata for embedding"
                                    );
                                    None
                                }
                            };
                            conn_manager.send(upstream_addr, pipe_metadata_net).await?;

                            // Start piping (runs asynchronously in background)
                            conn_manager
                                .pipe_stream(
                                    upstream_addr,
                                    outbound_sid,
                                    forked_handle,
                                    embedded_metadata,
                                )
                                .await?;

                            true
                        } else {
                            false
                        }
                    } else {
                        false
                    };

                    // Piggyback subscription: set up forwarding at this relay node
                    // (streaming path). Same logic as non-streaming relay path.
                    if subscribe_requested && !is_original_requester {
                        if let (Some(response_addr), Some(requester_addr)) =
                            (source_addr, self.upstream_addr)
                        {
                            super::setup_subscription_forwarding_at_relay(
                                op_manager,
                                &key,
                                &id,
                                response_addr,
                                requester_addr,
                            )
                            .await;
                        }
                    }

                    // Step 3: Wait for stream to complete and assemble data (for local caching)
                    let stream_data = match stream_handle.assemble().await {
                        Ok(data) => data,
                        Err(e) => {
                            if !is_original_requester {
                                // Relay node: fail immediately. The originator handles retry.
                                tracing::warn!(
                                    tx = %id,
                                    stream_id = %stream_id,
                                    error = %e,
                                    "Stream assembly failed on relay, sending Aborted upstream"
                                );
                                return Err(OpError::StreamCancelled);
                            }

                            // Originator: immediately retry with the next alternative peer
                            // instead of surfacing the error to the HTTP client.
                            let all_connected: Vec<_> = op_manager
                                .ring
                                .connection_manager
                                .get_connections_by_location()
                                .values()
                                .flat_map(|conns| conns.iter().map(|c| c.location.clone()))
                                .collect();
                            let max_htl = op_manager.ring.max_hops_to_live;

                            let retry_op = GetOp {
                                id: self.id,
                                state: self.state,
                                result: self.result,
                                stats,
                                upstream_addr: self.upstream_addr,
                                local_fallback,
                                auto_fetch,
                                client_return_code: self.client_return_code,
                            };

                            // Report routing failure for the stalled peer so the router
                            // learns to avoid it in future operations.
                            let stalled_peer_info = retry_op.failure_routing_info();

                            match retry_op.retry_with_next_alternative(max_htl, &all_connected) {
                                Ok((new_op, msg)) => {
                                    if let Some((peer, contract_location)) = stalled_peer_info {
                                        op_manager.ring.routing_finished(
                                            crate::router::RouteEvent {
                                                peer,
                                                contract_location,
                                                outcome: crate::router::RouteOutcome::Failure,
                                                op_type: Some(
                                                    crate::node::network_status::OpType::Get,
                                                ),
                                            },
                                        );
                                    }
                                    let target = match new_op.get_next_hop_addr() {
                                        Some(addr) => addr,
                                        None => {
                                            tracing::error!(
                                                tx = %id,
                                                "Retry peer has no socket address"
                                            );
                                            return Err(OpError::StreamCancelled);
                                        }
                                    };
                                    tracing::info!(
                                        tx = %id,
                                        stream_id = %stream_id,
                                        %target,
                                        error = %e,
                                        "Stream assembly failed, retrying GET with next peer"
                                    );
                                    return Ok(OperationResult::SendAndContinue {
                                        msg: NetMessage::from(msg),
                                        next_hop: Some(target),
                                        state: OpEnum::Get(new_op),
                                        stream_data: None,
                                    });
                                }
                                Err(_exhausted_op) => {
                                    tracing::warn!(
                                        tx = %id,
                                        stream_id = %stream_id,
                                        error = %e,
                                        "Stream assembly failed, no alternative peers left"
                                    );
                                    return Err(OpError::StreamCancelled);
                                }
                            }
                        }
                    };

                    // Record transfer-complete timing for originator (stream fully assembled)
                    if is_original_requester {
                        if let Some(ref mut s) = stats {
                            s.record_transfer_end();
                        }
                    }

                    tracing::debug!(
                        tx = %id,
                        stream_id = %stream_id,
                        received_size = stream_data.len(),
                        expected_size = total_size,
                        "Stream data assembled"
                    );

                    // Step 3: Deserialize the streaming payload
                    let payload: messages::GetStreamingPayload =
                        match bincode::deserialize(&stream_data) {
                            Ok(p) => p,
                            Err(e) => {
                                tracing::error!(
                                    tx = %id,
                                    stream_id = %stream_id,
                                    error = %e,
                                    "Failed to deserialize streaming payload"
                                );
                                return Err(OpError::invalid_transition(self.id));
                            }
                        };

                    // Verify key matches
                    if payload.key != key {
                        tracing::error!(
                            tx = %id,
                            expected = %key,
                            actual = %payload.key,
                            "Contract key mismatch in streaming payload"
                        );
                        return Err(OpError::invalid_transition(self.id));
                    }

                    let value = payload.value;

                    // Get current hop for telemetry
                    let current_hop = if let Some(GetState::AwaitingResponse(data)) = &self.state {
                        let current_hop = &data.current_hop;
                        Some(*current_hop)
                    } else {
                        None
                    };

                    // Step 5: Cache the contract locally, announce hosting, and auto-subscribe.
                    // This mirrors the non-streaming Response path — without it, the node
                    // never registers as subscribed after a streaming GET, causing every
                    // subsequent GET to hit the network instead of serving from local cache.
                    if let Some(state) = &value.state {
                        let contract_to_cache = if includes_contract {
                            value.contract.clone()
                        } else {
                            None
                        };

                        // Record access atomically — returns whether this is a newly-hosted contract
                        let access_result =
                            op_manager.ring.record_get_access(key, state.size() as u64);

                        if is_original_requester {
                            op_manager.ring.mark_local_client_access(&key);
                        }

                        // Clean up interest tracking for evicted contracts
                        let mut removed_contracts = Vec::new();
                        for evicted_key in &access_result.evicted {
                            if op_manager
                                .interest_manager
                                .unregister_local_hosting(evicted_key)
                            {
                                removed_contracts.push(*evicted_key);
                            }
                        }

                        if access_result.is_new {
                            // First time hosting — cache the contract, then announce only on success.
                            // Mirrors the non-streaming "state differs" path which gates announce
                            // on PutResponse { new_value: Ok(_) }.
                            let put_ok = match op_manager
                                .notify_contract_handler(ContractHandlerEvent::PutQuery {
                                    key,
                                    state: state.clone(),
                                    related_contracts: RelatedContracts::default(),
                                    contract: contract_to_cache,
                                })
                                .await
                            {
                                Ok(ContractHandlerEvent::PutResponse {
                                    new_value: Ok(_), ..
                                }) => true,
                                Ok(ContractHandlerEvent::PutResponse {
                                    new_value: Err(err),
                                    ..
                                }) => {
                                    tracing::warn!(
                                        tx = %id, contract = %key, error = %err,
                                        "Streaming GET: PutQuery rejected by executor"
                                    );
                                    false
                                }
                                Ok(other) => {
                                    tracing::warn!(
                                        tx = %id, contract = %key,
                                        response = %other,
                                        "Streaming GET: unexpected PutQuery response"
                                    );
                                    false
                                }
                                Err(e) => {
                                    tracing::warn!(
                                        contract = %key, error = %e,
                                        "failed to cache contract via PutQuery"
                                    );
                                    false
                                }
                            };

                            if put_ok {
                                tracing::debug!(tx = %id, %key, "Contract newly hosted (streaming)");
                                super::announce_contract_hosted(op_manager, &key).await;

                                let became_interested =
                                    op_manager.interest_manager.register_local_hosting(&key);
                                let added = if became_interested { vec![key] } else { vec![] };
                                if !added.is_empty() || !removed_contracts.is_empty() {
                                    super::broadcast_change_interests(
                                        op_manager,
                                        added,
                                        removed_contracts,
                                    )
                                    .await;
                                }

                                // Auto-subscribe: piggyback or fallback (streaming path)
                                if crate::ring::AUTO_SUBSCRIBE_ON_GET
                                    && is_original_requester
                                    && !op_manager.ring.is_subscribed(&key)
                                {
                                    super::auto_subscribe_on_get_response(
                                        op_manager,
                                        &key,
                                        &id,
                                        &sender_from_addr,
                                        subscribe_requested,
                                        blocking_sub,
                                        "streaming",
                                    )
                                    .await;
                                }
                            } else if !removed_contracts.is_empty() {
                                super::broadcast_change_interests(
                                    op_manager,
                                    vec![],
                                    removed_contracts,
                                )
                                .await;
                            }
                        } else {
                            // Already hosting — refresh state (fire-and-forget since we already
                            // have a valid cached copy; failure just means we keep the old state).
                            if let Err(e) = op_manager
                                .notify_contract_handler(ContractHandlerEvent::PutQuery {
                                    key,
                                    state: state.clone(),
                                    related_contracts: RelatedContracts::default(),
                                    contract: contract_to_cache,
                                })
                                .await
                            {
                                tracing::warn!(contract = %key, error = %e, "failed to refresh contract state via PutQuery");
                            }

                            tracing::debug!(tx = %id, %key, "Refreshed hosting status for already-hosted contract (streaming)");
                            if !removed_contracts.is_empty() {
                                super::broadcast_change_interests(
                                    op_manager,
                                    vec![],
                                    removed_contracts,
                                )
                                .await;
                            }

                            // Auto-subscribe: piggyback or fallback (streaming, re-subscribe)
                            if crate::ring::AUTO_SUBSCRIBE_ON_GET
                                && is_original_requester
                                && !op_manager.ring.is_subscribed(&key)
                            {
                                super::auto_subscribe_on_get_response(
                                    op_manager,
                                    &key,
                                    &id,
                                    &sender_from_addr,
                                    subscribe_requested,
                                    blocking_sub,
                                    "streaming, re-subscribe",
                                )
                                .await;
                            }
                        }
                    }

                    // Step 6: Emit telemetry
                    let hop_count =
                        current_hop.map(|h| op_manager.ring.max_hops_to_live.saturating_sub(h));
                    let hash = value.state.as_ref().map(state_hash_full);
                    if let Some(event) = NetEventLog::get_success(
                        &id,
                        &op_manager.ring,
                        key,
                        op_manager.ring.connection_manager.own_location(),
                        hop_count,
                        hash,
                    ) {
                        op_manager.ring.register_events(Either::Left(event)).await;
                    }

                    // Step 7: Build result and determine response
                    if is_original_requester {
                        // This is the original requester - operation complete
                        tracing::info!(
                            tx = %id,
                            contract = %key,
                            phase = "complete",
                            "Streaming GET complete (originator)"
                        );

                        // State must be present for a successful GET
                        let state = match value.state {
                            Some(s) => s,
                            None => {
                                tracing::error!(
                                    tx = %id,
                                    contract = %key,
                                    "Streaming GET response has no state"
                                );
                                return Err(OpError::invalid_transition(self.id));
                            }
                        };

                        result = Some(GetResult {
                            key,
                            state,
                            contract: value.contract,
                        });
                        new_state = Some(GetState::Finished(FinishedData { key }));
                        return_msg = None;
                    } else if piping_started {
                        // Piping is already underway - no need to send return_msg
                        tracing::debug!(
                            tx = %id,
                            contract = %key,
                            phase = "forward",
                            "GET response piping in progress to upstream"
                        );
                        return_msg = None;
                        new_state = None;
                    } else {
                        // Forward the response to upstream as a regular Response
                        // (streaming not used - payload below threshold)
                        tracing::debug!(
                            tx = %id,
                            contract = %key,
                            phase = "forward",
                            "Forwarding GET response as non-streaming to upstream"
                        );

                        return_msg = Some(GetMsg::Response {
                            id,
                            instance_id: *instance_id,
                            result: GetMsgResult::Found { key, value },
                        });
                        new_state = None;
                    }
                }

                // Streaming GET response acknowledgment handler
                GetMsg::ResponseStreamingAck {
                    id: msg_id,
                    stream_id,
                } => {
                    let id = *msg_id;
                    let stream_id = *stream_id;

                    // The acknowledgment confirms the stream was received.
                    // For now, we just log it and clean up.
                    tracing::info!(
                        tx = %id,
                        stream_id = %stream_id,
                        phase = "streaming_ack",
                        "Streaming GET response acknowledged"
                    );
                    new_state = None;
                    return_msg = None;
                }
                GetMsg::ForwardingAck { id, instance_id } => {
                    // A downstream relay has acknowledged forwarding our
                    // request. The GC speculative-retry block that consumed
                    // this signal was retired in #1454 Phase 5-final, but the
                    // telemetry emission stays for #3570 diagnostics. The
                    // operation's runtime state is unchanged by this message.
                    tracing::debug!(
                        tx = %id,
                        %instance_id,
                        "Received forwarding ACK from downstream relay"
                    );

                    if let Some(event) = NetEventLog::get_forwarding_ack_received(
                        &self.id,
                        &op_manager.ring,
                        *instance_id,
                    ) {
                        op_manager.ring.register_events(Either::Left(event)).await;
                    }

                    return Ok(OperationResult::ContinueOp(OpEnum::Get(GetOp {
                        id: self.id,
                        state: self.state,
                        result: self.result,
                        stats,
                        upstream_addr: self.upstream_addr,
                        local_fallback,
                        auto_fetch,
                        client_return_code: self.client_return_code,
                    })));
                }
            }

            build_op_result(GetOpResult {
                id: self.id,
                state: new_state,
                msg: return_msg,
                result,
                stats,
                upstream_addr: self.upstream_addr,
                auto_fetch,
                stream_data,
                local_fallback,
                client_return_code: self.client_return_code,
            })
        })
    }
}

/// Build a `GetMsg::Response` with `Found` result from a local fallback cache entry.
fn build_fallback_found_response(
    id: Transaction,
    instance_id: ContractInstanceId,
    key: ContractKey,
    state: WrappedState,
    contract: Option<ContractContainer>,
) -> GetMsg {
    GetMsg::Response {
        id,
        instance_id,
        result: GetMsgResult::Found {
            key,
            value: StoreResponse {
                state: Some(state),
                contract,
            },
        },
    }
}

struct GetOpResult {
    id: Transaction,
    state: Option<GetState>,
    msg: Option<GetMsg>,
    result: Option<GetResult>,
    stats: Option<Box<GetStats>>,
    upstream_addr: Option<std::net::SocketAddr>,
    auto_fetch: bool,
    stream_data: Option<(StreamId, bytes::Bytes)>,
    local_fallback: Option<(ContractKey, WrappedState, Option<ContractContainer>)>,
    client_return_code: bool,
}

fn build_op_result(params: GetOpResult) -> Result<OperationResult, OpError> {
    let GetOpResult {
        id,
        state,
        msg,
        result,
        stats,
        upstream_addr,
        auto_fetch,
        stream_data,
        local_fallback,
        client_return_code,
    } = params;
    // Determine the next hop for sending the message:
    // - For Response messages: route back to upstream_addr (who sent us the request)
    // - For Request messages being forwarded: use next_hop from state
    let next_hop = match (&msg, &state) {
        (Some(GetMsg::Response { .. }) | Some(GetMsg::ResponseStreaming { .. }), _) => {
            upstream_addr
        }
        (Some(GetMsg::Request { .. }), Some(GetState::AwaitingResponse(data))) => {
            data.next_hop.socket_addr()
        }
        _ => None,
    };

    let output_op = state.map(|state| GetOp {
        id,
        state: Some(state),
        result,
        stats,
        upstream_addr,
        local_fallback,
        auto_fetch,
        client_return_code,
    });
    let return_msg = msg.map(NetMessage::from);
    let op_state = output_op.map(OpEnum::Get);
    Ok(match (return_msg, op_state) {
        (Some(msg), Some(state)) => OperationResult::SendAndContinue {
            msg,
            next_hop,
            state,
            stream_data,
        },
        (Some(msg), None) => OperationResult::SendAndComplete {
            msg,
            next_hop,
            stream_data,
        },
        (None, Some(state)) => OperationResult::ContinueOp(state),
        (None, None) => OperationResult::Completed,
    })
}

#[allow(clippy::too_many_arguments)]
async fn try_forward_or_return(
    id: Transaction,
    instance_id: ContractInstanceId,
    (htl, fetch_contract): (usize, bool),
    (this_peer, sender): (PeerKeyLocation, Option<PeerKeyLocation>),
    visited: super::VisitedPeers,
    op_manager: &OpManager,
    stats: Option<Box<GetStats>>,
    upstream_addr: Option<std::net::SocketAddr>,
    local_fallback: Option<(ContractKey, WrappedState, Option<ContractContainer>)>,
    subscribe: bool,
) -> Result<OperationResult, OpError> {
    tracing::warn!(
        tx = %id,
        %instance_id,
        peer_addr = %this_peer,
        sender = ?sender,
        phase = "forward",
        "Contract not found while processing a get request"
    );

    let mut new_visited = visited.clone();
    if let Some(addr) = this_peer.socket_addr() {
        new_visited.mark_visited(addr);
    }

    let new_htl = htl.saturating_sub(1);

    let (new_target, alternatives) = if new_htl == 0 {
        tracing::warn!(
            tx = %id,
            sender = ?sender,
            htl = 0,
            phase = "error",
            "The maximum hops have been exceeded, sending response back to the node"
        );
        (None, vec![])
    } else {
        let mut candidates = op_manager.ring.k_closest_potentially_hosting(
            &instance_id,
            &new_visited,
            DEFAULT_MAX_BREADTH,
        );

        if candidates.is_empty() {
            tracing::warn!(
                tx = %id,
                %instance_id,
                peer_addr = %this_peer,
                phase = "error",
                "No other peers found while trying to get the contract"
            );
            (None, vec![])
        } else {
            let target = candidates.remove(0);
            (Some(target), candidates)
        }
    };

    if let Some(target) = new_target {
        tracing::debug!(
            tx = %id,
            "Forwarding get request to {}",
            target
        );

        // Emit telemetry: relay is forwarding GET to next hop
        if let Some(event) =
            NetEventLog::get_request(&id, &op_manager.ring, instance_id, target.clone(), new_htl)
        {
            op_manager.ring.register_events(Either::Left(event)).await;
        }

        let mut tried_peers = HashSet::new();
        if let Some(addr) = target.socket_addr() {
            tried_peers.insert(addr);
        }

        // Forwarding nodes always use non-blocking subscriptions:
        // blocking_subscribe is a client-side preference that only
        // applies to the originator node.
        build_op_result(GetOpResult {
            id,
            state: Some(GetState::AwaitingResponse(AwaitingResponseData {
                instance_id,
                requester: sender,
                retries: 0,
                fetch_contract,
                current_hop: new_htl,
                subscribe,
                blocking_subscribe: false,
                next_hop: target.clone(),
                tried_peers,
                alternatives,
                attempts_at_hop: 1,
                visited: new_visited.clone(),
            })),
            msg: Some(GetMsg::Request {
                id,
                instance_id,
                fetch_contract,
                htl: new_htl,
                visited: new_visited,
                subscribe,
            }),
            result: None,
            // Track the forward target so timeouts report to
            // PeerHealthTracker and the failure estimator (#3527).
            stats: stats.map(|mut s| {
                s.next_peer = Some(target.clone());
                s
            }),
            upstream_addr,
            stream_data: None,
            local_fallback,
            auto_fetch: false,
            client_return_code: true,
        })
    } else if upstream_addr.is_some() {
        // No targets found — check for local fallback before returning NotFound
        if let Some((key, state, contract)) = local_fallback {
            tracing::info!(
                tx = %id,
                contract = %key,
                "Relay serving local cache as fallback (no forwarding targets)"
            );
            let msg = build_fallback_found_response(id, instance_id, key, state, contract);
            build_op_result(GetOpResult {
                id,
                state: None,
                msg: Some(msg),
                result: None,
                stats,
                upstream_addr,
                stream_data: None,
                local_fallback: None,
                auto_fetch: false,
                client_return_code: true,
            })
        } else {
            tracing::warn!(
                tx = %id,
                %instance_id,
                phase = "not_found",
                "No peers to forward get request to, returning NotFound to upstream"
            );
            // Emit telemetry: relay returning NotFound (no forwarding targets)
            if let Some(event) =
                NetEventLog::get_not_found(&id, &op_manager.ring, instance_id, None)
            {
                op_manager.ring.register_events(Either::Left(event)).await;
            }
            build_op_result(GetOpResult {
                id,
                state: None,
                msg: Some(GetMsg::Response {
                    id,
                    instance_id,
                    result: GetMsgResult::NotFound,
                }),
                result: None,
                stats,
                upstream_addr,
                stream_data: None,
                local_fallback: None,
                auto_fetch: false,
                client_return_code: true,
            })
        }
    } else {
        // Original requester with no forwarding targets - operation fails locally
        tracing::warn!(
            tx = %id,
            %instance_id,
            phase = "not_found",
            "No peers to forward get request to and no upstream - local operation fails"
        );

        notify_get_failed(
            op_manager,
            id,
            &format!("contract {instance_id} not found, no peers available to forward"),
        );

        build_op_result(GetOpResult {
            id,
            state: None,
            msg: None,
            result: None,
            stats,
            upstream_addr,
            stream_data: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        })
    }
}

impl IsOperationCompleted for GetOp {
    fn is_completed(&self) -> bool {
        matches!(self.state, Some(GetState::Finished(_)))
    }
}

/// Notify the client that a GET operation failed with the given reason.
fn notify_get_failed(op_manager: &OpManager, tx: Transaction, reason: &str) {
    op_manager.send_client_result(
        tx,
        Err(ErrorKind::OperationError {
            cause: reason.to_string().into(),
        }
        .into()),
    );
}

mod messages {
    use std::fmt::Display;

    use serde::{Deserialize, Serialize};

    use super::*;
    use crate::transport::peer_connection::StreamId;

    /// Payload for streaming GET responses.
    /// Contains the result of a GET operation, serialized for streaming transfer.
    #[derive(Debug, Serialize, Deserialize)]
    pub(crate) struct GetStreamingPayload {
        pub key: ContractKey,
        pub value: StoreResponse,
    }

    /// Result of a GET operation - either the contract was found or it wasn't.
    ///
    /// This provides explicit semantics for "contract not found" rather than
    /// requiring interpretation of empty responses or timeouts.
    #[derive(Debug, Serialize, Deserialize, Clone)]
    pub(crate) enum GetMsgResult {
        /// Contract was found - includes full key and value
        Found {
            key: ContractKey,
            value: StoreResponse,
        },
        /// Contract was not found after exhaustive search
        NotFound,
    }

    /// GET operation messages.
    ///
    /// Uses hop-by-hop routing: each node stores `upstream_addr` from the transport layer
    /// to route responses back. No `PeerKeyLocation` is embedded in wire messages.
    #[derive(Debug, Serialize, Deserialize, Clone)]
    pub(crate) enum GetMsg {
        /// Request to retrieve a contract. Forwarded hop-by-hop toward contract location.
        /// Uses instance_id since the requester may not have the full key yet.
        Request {
            id: Transaction,
            instance_id: ContractInstanceId,
            fetch_contract: bool,
            /// Hops to live - decremented at each hop. When 0, stop forwarding.
            htl: usize,
            /// Bloom filter tracking visited peers to prevent loops.
            visited: super::super::VisitedPeers,
            /// Whether the originator wants a subscription established on the
            /// response path. When true, relay nodes set up forwarding
            /// (upstream/downstream registration) so the subscription tree is
            /// formed by the time the response reaches the originator.
            #[serde(default)]
            subscribe: bool,
        },
        /// Response for a GET operation. Routed hop-by-hop back to originator.
        /// Uses instance_id for routing (always available from the request).
        /// The full ContractKey is only present in GetMsgResult::Found.
        Response {
            id: Transaction,
            instance_id: ContractInstanceId,
            result: GetMsgResult,
        },

        /// Streaming response for large contract data. Used when the response payload
        /// exceeds streaming_threshold (default 64KB). The actual data is sent via
        /// a separate stream identified by stream_id.
        ///
        /// This variant is only used when streaming is enabled in config.
        ResponseStreaming {
            id: Transaction,
            instance_id: ContractInstanceId,
            /// Identifies the stream carrying the response data
            stream_id: StreamId,
            /// Full contract key (known since we found the contract)
            key: ContractKey,
            /// Total size of the streamed payload in bytes
            total_size: u64,
            /// Whether the response includes the contract code (not just state)
            includes_contract: bool,
        },

        /// Acknowledgment that a streaming response was received.
        /// Sent back to the responder to confirm stream completion.
        ResponseStreamingAck {
            id: Transaction,
            stream_id: StreamId,
        },

        /// Lightweight ACK sent by a relay peer back to its upstream when it forwards
        /// a GET request to the next hop. Tells the upstream "I'm working on it" so
        /// the GC task can distinguish dead peers from slow multi-hop chains.
        /// Fire-and-forget — no response expected.
        ForwardingAck {
            id: Transaction,
            instance_id: ContractInstanceId,
        },
    }

    impl InnerMessage for GetMsg {
        fn id(&self) -> &Transaction {
            match self {
                Self::Request { id, .. }
                | Self::Response { id, .. }
                | Self::ResponseStreaming { id, .. }
                | Self::ResponseStreamingAck { id, .. }
                | Self::ForwardingAck { id, .. } => id,
            }
        }

        fn requested_location(&self) -> Option<Location> {
            match self {
                Self::Request { instance_id, .. }
                | Self::Response { instance_id, .. }
                | Self::ResponseStreaming { instance_id, .. }
                | Self::ForwardingAck { instance_id, .. } => Some(Location::from(instance_id)),
                Self::ResponseStreamingAck { .. } => {
                    // Ack doesn't carry location info - routed via stream_id
                    None
                }
            }
        }
    }

    impl Display for GetMsg {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            let id = self.id();
            match self {
                Self::Request {
                    instance_id, htl, ..
                } => {
                    write!(
                        f,
                        "Get::Request(id: {id}, instance_id: {instance_id}, htl: {htl})"
                    )
                }
                Self::Response {
                    instance_id,
                    result,
                    ..
                } => {
                    let result_str = match result {
                        GetMsgResult::Found { key, .. } => format!("Found({key})"),
                        GetMsgResult::NotFound => "NotFound".to_string(),
                    };
                    write!(
                        f,
                        "Get::Response(id: {id}, instance_id: {instance_id}, result: {result_str})"
                    )
                }
                Self::ResponseStreaming {
                    instance_id,
                    stream_id,
                    key,
                    total_size,
                    ..
                } => {
                    write!(
                        f,
                        "Get::ResponseStreaming(id: {id}, instance_id: {instance_id}, key: {key}, stream: {stream_id}, size: {total_size})"
                    )
                }
                Self::ResponseStreamingAck { stream_id, .. } => {
                    write!(
                        f,
                        "Get::ResponseStreamingAck(id: {id}, stream: {stream_id})"
                    )
                }
                Self::ForwardingAck { instance_id, .. } => {
                    write!(
                        f,
                        "Get::ForwardingAck(id: {id}, instance_id: {instance_id})"
                    )
                }
            }
        }
    }
}

#[cfg(test)]
#[allow(clippy::wildcard_enum_match_arm)]
mod tests {
    use super::*;
    use crate::message::Transaction;
    use crate::operations::VisitedPeers;
    use crate::operations::test_utils::{make_contract_key, make_test_contract};

    fn make_get_op(state: Option<GetState>, result: Option<GetResult>) -> GetOp {
        GetOp {
            id: Transaction::new::<GetMsg>(),
            state,
            result,
            stats: None,
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        }
    }

    // Tests for finalized() method
    #[test]
    fn get_op_finalized_when_finished_with_result() {
        let key = make_contract_key(1);
        let result = GetResult {
            key,
            state: WrappedState::new(vec![1, 2, 3]),
            contract: None,
        };
        let op = make_get_op(Some(GetState::Finished(FinishedData { key })), Some(result));
        assert!(
            op.finalized(),
            "GetOp should be finalized when state is Finished and result is present"
        );
    }

    #[test]
    fn get_op_not_finalized_when_finished_without_result() {
        let key = make_contract_key(1);
        let op = make_get_op(Some(GetState::Finished(FinishedData { key })), None);
        assert!(
            !op.finalized(),
            "GetOp should not be finalized when state is Finished but result is None"
        );
    }

    #[test]
    fn get_op_not_finalized_when_received_request() {
        let op = make_get_op(Some(GetState::ReceivedRequest), None);
        assert!(
            !op.finalized(),
            "GetOp should not be finalized in ReceivedRequest state"
        );
    }

    #[test]
    fn get_op_not_finalized_when_state_is_none() {
        let op = make_get_op(None, None);
        assert!(
            !op.finalized(),
            "GetOp should not be finalized when state is None"
        );
    }

    // Tests for to_host_result() method
    #[test]
    fn get_op_to_host_result_success_when_result_present() {
        let key = make_contract_key(1);
        let state_data = WrappedState::new(vec![1, 2, 3]);
        let result = GetResult {
            key,
            state: state_data.clone(),
            contract: None,
        };
        let op = make_get_op(Some(GetState::Finished(FinishedData { key })), Some(result));
        let host_result = op.to_host_result();

        assert!(
            host_result.is_ok(),
            "to_host_result should return Ok when result is present"
        );

        if let Ok(HostResponse::ContractResponse(
            freenet_stdlib::client_api::ContractResponse::GetResponse {
                key: returned_key,
                state: returned_state,
                ..
            },
        )) = host_result
        {
            assert_eq!(returned_key, key, "Returned key should match");
            assert_eq!(returned_state, state_data, "Returned state should match");
        } else {
            panic!("Expected GetResponse");
        }
    }

    #[test]
    fn get_op_to_host_result_error_when_no_result() {
        let op = make_get_op(Some(GetState::ReceivedRequest), None);
        let result = op.to_host_result();
        assert!(
            result.is_err(),
            "to_host_result should return Err when result is None"
        );
    }

    /// Regression test for #3757: return_contract_code=false should strip contract
    /// from client response but still cache WASM internally.
    #[test]
    fn get_op_to_host_result_strips_contract_when_client_return_code_false() {
        let key = make_contract_key(1);
        let result = GetResult {
            key,
            state: WrappedState::new(vec![1, 2, 3]),
            contract: Some(make_test_contract(&[42u8; 100])),
        };
        let mut op = make_get_op(Some(GetState::Finished(FinishedData { key })), Some(result));

        /// Extract the contract field from a to_host_result() GetResponse.
        fn get_response_contract(op: &GetOp) -> Option<ContractContainer> {
            let Ok(HostResponse::ContractResponse(
                freenet_stdlib::client_api::ContractResponse::GetResponse { contract, .. },
            )) = op.to_host_result()
            else {
                panic!("Expected Ok(GetResponse)");
            };
            contract
        }

        op.client_return_code = true;
        assert!(
            get_response_contract(&op).is_some(),
            "Contract should be included when client_return_code=true"
        );

        op.client_return_code = false;
        assert!(
            get_response_contract(&op).is_none(),
            "Contract should be stripped when client_return_code=false"
        );
    }

    /// Regression test for #3757: start_op and start_op_with_id should always
    /// set fetch_contract=true internally, regardless of the client's preference.
    #[test]
    fn start_op_always_fetches_contract_code() {
        let key = make_contract_key(1);
        let instance_id = *key.id();

        /// Assert that the operation's internal fetch_contract is always true
        /// while client_return_code matches the caller's original request.
        fn assert_fetch_state(op: &GetOp, expected_client_return: bool) {
            let Some(GetState::PrepareRequest(data)) = &op.state else {
                panic!("Expected PrepareRequest state");
            };
            assert!(
                data.fetch_contract,
                "Internal fetch_contract should always be true (#3757)"
            );
            assert_eq!(op.client_return_code, expected_client_return);
        }

        // start_op_with_id with fetch_contract=false
        let tx_a = Transaction::new::<GetMsg>();
        assert_fetch_state(
            &start_op_with_id(instance_id, false, false, false, tx_a),
            false,
        );
        // start_op_with_id with fetch_contract=true
        let tx_b = Transaction::new::<GetMsg>();
        assert_fetch_state(
            &start_op_with_id(instance_id, true, false, false, tx_b),
            true,
        );

        // start_op_with_id explicit transaction
        let tx = Transaction::new::<GetMsg>();
        assert_fetch_state(
            &start_op_with_id(instance_id, false, false, false, tx),
            false,
        );
        let tx = Transaction::new::<GetMsg>();
        assert_fetch_state(&start_op_with_id(instance_id, true, false, false, tx), true);
    }

    // Tests for outcome() method - partial coverage since full stats require complex setup
    #[test]
    fn get_op_outcome_incomplete_without_stats() {
        let key = make_contract_key(1);
        let result = GetResult {
            key,
            state: WrappedState::new(vec![]),
            contract: None,
        };
        let op = make_get_op(Some(GetState::Finished(FinishedData { key })), Some(result));
        let outcome = op.outcome();

        assert!(matches!(outcome, OpOutcome::Incomplete));
    }

    // Tests for GetMsg helper methods
    #[test]
    fn get_msg_id_returns_transaction() {
        let tx = Transaction::new::<GetMsg>();
        let msg = GetMsg::Request {
            id: tx,
            instance_id: *make_contract_key(1).id(),
            fetch_contract: false,
            htl: 5,
            visited: VisitedPeers::new(&tx),
            subscribe: false,
        };
        assert_eq!(*msg.id(), tx, "id() should return the transaction ID");
    }

    #[test]
    fn get_msg_display_formats_correctly() {
        let tx = Transaction::new::<GetMsg>();
        let msg = GetMsg::Request {
            id: tx,
            instance_id: *make_contract_key(1).id(),
            fetch_contract: false,
            htl: 5,
            visited: VisitedPeers::new(&tx),
            subscribe: false,
        };
        let display = format!("{}", msg);
        assert!(
            display.contains("Request"),
            "Display should contain message type name"
        );
    }

    // Tests for GetState Display
    #[test]
    fn get_state_display_received_request() {
        let state = GetState::ReceivedRequest;
        let display = format!("{}", state);
        assert!(
            display.contains("ReceivedRequest"),
            "Display should contain state name"
        );
    }

    #[test]
    fn get_state_display_finished() {
        let state = GetState::Finished(FinishedData {
            key: make_contract_key(1),
        });
        let display = format!("{}", state);
        assert!(
            display.contains("Finished"),
            "Display should contain state name"
        );
    }

    #[test]
    fn get_msg_response_found_display_formats_correctly() {
        let tx = Transaction::new::<GetMsg>();
        let key = make_contract_key(1);
        let msg = GetMsg::Response {
            id: tx,
            instance_id: *key.id(),
            result: GetMsgResult::Found {
                key,
                value: StoreResponse {
                    state: Some(WrappedState::new(vec![1, 2, 3])),
                    contract: None,
                },
            },
        };
        let display = format!("{}", msg);
        assert!(
            display.contains("Response"),
            "Display should contain message type name"
        );
        assert!(
            display.contains("Found"),
            "Display should indicate Found result"
        );
    }

    #[test]
    fn get_msg_response_notfound_display_formats_correctly() {
        let tx = Transaction::new::<GetMsg>();
        let instance_id = *make_contract_key(1).id();
        let msg = GetMsg::Response {
            id: tx,
            instance_id,
            result: GetMsgResult::NotFound,
        };
        let display = format!("{}", msg);
        assert!(
            display.contains("Response"),
            "Display should contain message type name"
        );
        assert!(
            display.contains("NotFound"),
            "Display should indicate NotFound result"
        );
    }

    #[test]
    fn get_msg_result_found_contains_key_and_value() {
        let key = make_contract_key(1);
        let state = WrappedState::new(vec![1, 2, 3]);
        let result = GetMsgResult::Found {
            key,
            value: StoreResponse {
                state: Some(state.clone()),
                contract: None,
            },
        };
        if let GetMsgResult::Found {
            key: found_key,
            value,
        } = result
        {
            assert_eq!(found_key, key);
            assert_eq!(value.state, Some(state));
        } else {
            panic!("Expected Found variant");
        }
    }

    #[test]
    fn get_msg_result_notfound_is_unit_variant() {
        let result = GetMsgResult::NotFound;
        assert!(
            matches!(result, GetMsgResult::NotFound),
            "NotFound should match NotFound"
        );
    }

    #[test]
    fn get_msg_response_requested_location_uses_instance_id() {
        let tx = Transaction::new::<GetMsg>();
        let key = make_contract_key(1);
        let instance_id = *key.id();

        // Test Found variant
        let msg_found = GetMsg::Response {
            id: tx,
            instance_id,
            result: GetMsgResult::Found {
                key,
                value: StoreResponse {
                    state: Some(WrappedState::new(vec![])),
                    contract: None,
                },
            },
        };
        let location_found = msg_found.requested_location();
        assert!(
            location_found.is_some(),
            "Response should have a requested location"
        );
        assert_eq!(
            location_found.unwrap(),
            Location::from(&instance_id),
            "Location should be derived from instance_id"
        );

        // Test NotFound variant
        let msg_notfound = GetMsg::Response {
            id: tx,
            instance_id,
            result: GetMsgResult::NotFound,
        };
        let location_notfound = msg_notfound.requested_location();
        assert!(
            location_notfound.is_some(),
            "NotFound Response should have a requested location"
        );
        assert_eq!(
            location_notfound.unwrap(),
            Location::from(&instance_id),
            "Location should be derived from instance_id for NotFound too"
        );
    }

    // Tests for blocking_subscribe propagation
    #[test]
    fn start_op_propagates_blocking_subscribe_true() {
        let instance_id = ContractInstanceId::new([1u8; 32]);
        let tx = Transaction::new::<GetMsg>();
        let op = start_op_with_id(instance_id, true, true, true, tx);
        match op.state {
            Some(GetState::PrepareRequest(data)) => {
                let blocking_subscribe = data.blocking_subscribe;
                assert!(
                    blocking_subscribe,
                    "blocking_subscribe should be true in PrepareRequest"
                );
            }
            other => panic!("Expected PrepareRequest state, got {:?}", other),
        }
    }

    #[test]
    fn start_op_with_id_propagates_blocking_subscribe_true() {
        let instance_id = ContractInstanceId::new([1u8; 32]);
        let tx = Transaction::new::<GetMsg>();
        let op = start_op_with_id(instance_id, true, true, true, tx);
        match op.state {
            Some(GetState::PrepareRequest(data)) => {
                let blocking_subscribe = data.blocking_subscribe;
                assert!(
                    blocking_subscribe,
                    "blocking_subscribe should be true in PrepareRequest via start_op_with_id"
                );
            }
            other => panic!("Expected PrepareRequest state, got {:?}", other),
        }
    }

    #[test]
    fn start_op_defaults_blocking_subscribe_false() {
        let instance_id = ContractInstanceId::new([1u8; 32]);
        let tx = Transaction::new::<GetMsg>();
        let op = start_op_with_id(instance_id, true, true, false, tx);
        match op.state {
            Some(GetState::PrepareRequest(data)) => {
                let blocking_subscribe = data.blocking_subscribe;
                assert!(
                    !blocking_subscribe,
                    "blocking_subscribe should be false by default"
                );
            }
            other => panic!("Expected PrepareRequest state, got {:?}", other),
        }
    }

    #[test]
    fn test_failure_outcome_for_get() {
        use crate::ring::{Location, PeerKeyLocation};
        use std::time::Duration;

        let target_peer = PeerKeyLocation::random();
        let contract_location = Location::random();

        // GetOp with stats but no result → should return ContractOpFailure
        let op = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: None,
            result: None,
            stats: Some(Box::new(GetStats {
                next_peer: Some(target_peer.clone()),
                contract_location,
                first_response_time: None,
                transfer_time: None,
            })),
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };

        match op.outcome() {
            OpOutcome::ContractOpFailure {
                target_peer: peer,
                contract_location: loc,
            } => {
                assert_eq!(*peer, target_peer);
                assert_eq!(loc, contract_location);
            }
            OpOutcome::ContractOpSuccess { .. }
            | OpOutcome::ContractOpSuccessUntimed { .. }
            | OpOutcome::Incomplete
            | OpOutcome::Irrelevant => panic!("Expected ContractOpFailure"),
        }

        // GetOp with no stats and no result → should return Incomplete
        let op_no_stats = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: None,
            result: None,
            stats: None,
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };
        assert!(
            matches!(op_no_stats.outcome(), OpOutcome::Incomplete),
            "GetOp with no stats should return Incomplete"
        );

        // GetOp with result + complete stats → should return ContractOpSuccess
        let now = Instant::now();
        let later = now + Duration::from_millis(50);
        let op_success = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: None,
            result: Some(GetResult {
                key: make_contract_key(1),
                state: WrappedState::new(vec![1, 2, 3]),
                contract: None,
            }),
            stats: Some(Box::new(GetStats {
                next_peer: Some(target_peer.clone()),
                contract_location,
                first_response_time: Some((now, Some(later))),
                transfer_time: Some((now, Some(later))),
            })),
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };
        assert!(
            matches!(op_success.outcome(), OpOutcome::ContractOpSuccess { .. }),
            "GetOp with result and complete stats should return ContractOpSuccess"
        );
    }

    // ============ Partial timing outcome tests (validates GET partial timing fix) ============

    /// Result + target_peer but incomplete timing → ContractOpSuccessUntimed.
    /// Before the fix, this returned Incomplete, so the router never got feedback.
    #[test]
    fn test_get_outcome_success_untimed_partial_timing() {
        use crate::ring::{Location, PeerKeyLocation};

        let target_peer = PeerKeyLocation::random();
        let contract_location = Location::random();
        let now = Instant::now();

        // Result present, target_peer set, first_response_time has start but no end
        let op = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: Some(GetState::Finished(FinishedData {
                key: make_contract_key(1),
            })),
            result: Some(GetResult {
                key: make_contract_key(1),
                state: WrappedState::new(vec![1, 2, 3]),
                contract: None,
            }),
            stats: Some(Box::new(GetStats {
                next_peer: Some(target_peer.clone()),
                contract_location,
                first_response_time: Some((now, None)), // Started but not completed
                transfer_time: Some((now, None)),       // Started but not completed
            })),
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };

        match op.outcome() {
            OpOutcome::ContractOpSuccessUntimed {
                target_peer: peer,
                contract_location: loc,
            } => {
                assert_eq!(*peer, target_peer);
                assert_eq!(loc, contract_location);
            }
            other @ OpOutcome::ContractOpSuccess { .. }
            | other @ OpOutcome::ContractOpFailure { .. }
            | other @ OpOutcome::Incomplete
            | other @ OpOutcome::Irrelevant => {
                panic!("Expected ContractOpSuccessUntimed, got {other:?}")
            }
        }
    }

    /// Result + target_peer but no transfer_time at all → ContractOpSuccessUntimed.
    #[test]
    fn test_get_outcome_success_untimed_no_transfer_time() {
        use crate::ring::{Location, PeerKeyLocation};

        let target_peer = PeerKeyLocation::random();
        let contract_location = Location::random();

        let op = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: Some(GetState::Finished(FinishedData {
                key: make_contract_key(1),
            })),
            result: Some(GetResult {
                key: make_contract_key(1),
                state: WrappedState::new(vec![1, 2, 3]),
                contract: None,
            }),
            stats: Some(Box::new(GetStats {
                next_peer: Some(target_peer.clone()),
                contract_location,
                first_response_time: None,
                transfer_time: None,
            })),
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };

        match op.outcome() {
            OpOutcome::ContractOpSuccessUntimed {
                target_peer: peer,
                contract_location: loc,
            } => {
                assert_eq!(*peer, target_peer);
                assert_eq!(loc, contract_location);
            }
            other @ OpOutcome::ContractOpSuccess { .. }
            | other @ OpOutcome::ContractOpFailure { .. }
            | other @ OpOutcome::Incomplete
            | other @ OpOutcome::Irrelevant => {
                panic!("Expected ContractOpSuccessUntimed, got {other:?}")
            }
        }
    }

    /// Result but no target_peer → Incomplete (can't attribute to a peer).
    #[test]
    fn test_get_outcome_incomplete_result_no_peer() {
        let op = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: Some(GetState::Finished(FinishedData {
                key: make_contract_key(1),
            })),
            result: Some(GetResult {
                key: make_contract_key(1),
                state: WrappedState::new(vec![1, 2, 3]),
                contract: None,
            }),
            stats: Some(Box::new(GetStats {
                next_peer: None,
                contract_location: Location::random(),
                first_response_time: None,
                transfer_time: None,
            })),
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };

        assert!(
            matches!(op.outcome(), OpOutcome::Incomplete),
            "Result with no target_peer should return Incomplete"
        );
    }

    /// No result + partial timing → still ContractOpFailure (regression guard).
    #[test]
    fn test_get_outcome_failure_no_result_partial_timing() {
        use crate::ring::{Location, PeerKeyLocation};

        let target_peer = PeerKeyLocation::random();
        let contract_location = Location::random();
        let now = Instant::now();

        let op = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: None,
            result: None,
            stats: Some(Box::new(GetStats {
                next_peer: Some(target_peer.clone()),
                contract_location,
                first_response_time: Some((now, None)),
                transfer_time: None,
            })),
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };

        match op.outcome() {
            OpOutcome::ContractOpFailure {
                target_peer: peer,
                contract_location: loc,
            } => {
                assert_eq!(*peer, target_peer);
                assert_eq!(loc, contract_location);
            }
            other @ OpOutcome::ContractOpSuccess { .. }
            | other @ OpOutcome::ContractOpSuccessUntimed { .. }
            | other @ OpOutcome::Incomplete
            | other @ OpOutcome::Irrelevant => panic!("Expected ContractOpFailure, got {other:?}"),
        }
    }

    #[test]
    fn test_build_fallback_found_response() {
        let id = Transaction::new::<GetMsg>();
        let key = make_contract_key(1);
        let instance_id = *key.id();
        let state = WrappedState::new(vec![1, 2, 3]);
        let msg = build_fallback_found_response(id, instance_id, key, state.clone(), None);
        match msg {
            GetMsg::Response {
                id: resp_id,
                instance_id: resp_iid,
                result: GetMsgResult::Found { key: k, value },
            } => {
                assert_eq!(resp_id, id);
                assert_eq!(resp_iid, instance_id);
                assert_eq!(k, key);
                assert_eq!(value.state, Some(state));
                assert!(value.contract.is_none());
            }
            other @ GetMsg::Request { .. }
            | other @ GetMsg::Response { .. }
            | other @ GetMsg::ResponseStreaming { .. }
            | other @ GetMsg::ResponseStreamingAck { .. }
            | other @ GetMsg::ForwardingAck { .. } => {
                panic!("Expected Found response, got {other:?}")
            }
        }
    }

    // === Tests for retry_with_next_alternative ===

    use crate::operations::test_utils::make_peer;

    fn make_awaiting_op(alternatives: Vec<PeerKeyLocation>, tried: &[PeerKeyLocation]) -> GetOp {
        let id = Transaction::new::<GetMsg>();
        let instance_id = ContractInstanceId::new([42u8; 32]);
        let visited = VisitedPeers::new(&id);
        let next_hop = make_peer(9000);
        let mut tried_peers = HashSet::new();
        for p in tried {
            if let Some(addr) = p.socket_addr() {
                tried_peers.insert(addr);
            }
        }
        if let Some(addr) = next_hop.socket_addr() {
            tried_peers.insert(addr);
        }
        GetOp {
            id,
            state: Some(GetState::AwaitingResponse(AwaitingResponseData {
                instance_id,
                retries: 0,
                fetch_contract: true,
                requester: None,
                current_hop: 7,
                subscribe: false,
                blocking_subscribe: false,
                next_hop,
                tried_peers,
                alternatives,
                attempts_at_hop: 1,
                visited,
            })),
            result: None,
            stats: None,
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        }
    }

    #[test]
    fn retry_with_local_alternative_succeeds() {
        let alt1 = make_peer(1001);
        let alt2 = make_peer(1002);
        let op = make_awaiting_op(vec![alt1.clone(), alt2], &[]);

        let result = op.retry_with_next_alternative(7, &[]);
        assert!(result.is_ok(), "Should succeed with local alternatives");

        let (new_op, msg) = result.unwrap_or_else(|_| panic!("expected Ok"));
        // The message should be a Request preserving original fetch_contract
        match &msg {
            GetMsg::Request { fetch_contract, .. } => {
                assert!(
                    *fetch_contract,
                    "retry must preserve original fetch_contract"
                );
            }
            GetMsg::Response { .. }
            | GetMsg::ResponseStreaming { .. }
            | GetMsg::ResponseStreamingAck { .. }
            | GetMsg::ForwardingAck { .. } => panic!("Expected GetMsg::Request"),
        }
        // Should have 1 alternative remaining
        if let Some(GetState::AwaitingResponse(data)) = &new_op.state {
            assert_eq!(data.alternatives.len(), 1);
            assert!(data.tried_peers.contains(&alt1.socket_addr().unwrap()));
        } else {
            panic!("Expected AwaitingResponse state");
        }
    }

    #[test]
    fn retry_exhausted_returns_err() {
        let op = make_awaiting_op(vec![], &[]);

        let result = op.retry_with_next_alternative(7, &[]);
        assert!(result.is_err(), "Should fail with no alternatives");

        // Verify op is returned intact
        let returned_op = match result {
            Err(op) => *op,
            Ok(_) => panic!("expected Err"),
        };
        assert!(
            matches!(&returned_op.state, Some(GetState::AwaitingResponse(_))),
            "State should be preserved"
        );
    }

    #[test]
    fn retry_dbf_fallback_injects_new_peers() {
        // No local alternatives, but provide fallback peers
        let fallback1 = make_peer(2001);
        let fallback2 = make_peer(2002);
        let op = make_awaiting_op(vec![], &[]);

        let result = op.retry_with_next_alternative(7, &[fallback1.clone(), fallback2]);
        assert!(result.is_ok(), "Should succeed with DBF fallback peers");

        let (new_op, _msg) = result.unwrap_or_else(|_| panic!("expected Ok"));
        if let Some(GetState::AwaitingResponse(data)) = &new_op.state {
            // One fallback used as next_hop, one remaining
            assert_eq!(data.alternatives.len(), 1);
            assert!(data.tried_peers.contains(&fallback1.socket_addr().unwrap()));
        } else {
            panic!("Expected AwaitingResponse state");
        }
    }

    #[test]
    fn retry_dbf_fallback_skips_already_tried() {
        let already_tried = make_peer(3001);
        let fresh_peer = make_peer(3002);
        let op = make_awaiting_op(vec![], std::slice::from_ref(&already_tried));

        let result = op.retry_with_next_alternative(7, &[already_tried, fresh_peer.clone()]);
        assert!(result.is_ok(), "Should find fresh_peer");

        let (new_op, _msg) = result.unwrap_or_else(|_| panic!("expected Ok"));
        if let Some(GetState::AwaitingResponse(data)) = &new_op.state {
            assert_eq!(data.alternatives.len(), 0, "Only one fresh peer injected");
            assert!(
                data.tried_peers
                    .contains(&fresh_peer.socket_addr().unwrap())
            );
        } else {
            panic!("Expected AwaitingResponse state");
        }
    }

    #[test]
    fn retry_dbf_all_tried_returns_err() {
        let tried1 = make_peer(4001);
        let tried2 = make_peer(4002);
        let op = make_awaiting_op(vec![], &[tried1.clone(), tried2.clone()]);

        let result = op.retry_with_next_alternative(7, &[tried1, tried2]);
        assert!(result.is_err(), "All fallback peers already tried");
    }

    /// Verify that retry HTL decreases with each attempt (#3570).
    ///
    /// At the originator, current_hop == max_hops_to_live, so the old code
    /// (htl: max_hops_to_live) always sent full-depth retries. The fix divides
    /// HTL by attempts_at_hop, creating progressively shorter retry chains.
    #[test]
    fn retry_htl_decreases_with_attempts() {
        let alt1 = make_peer(6001);
        let alt2 = make_peer(6002);
        let alt3 = make_peer(6003);
        let op = make_awaiting_op(vec![alt1, alt2, alt3], &[]);

        // make_awaiting_op sets attempts_at_hop=1, then retry increments to 2 before computing HTL.
        // First retry: attempts_at_hop becomes 2 → 10/2 = 5
        let (op, msg) = op
            .retry_with_next_alternative(10, &[])
            .unwrap_or_else(|_| panic!("retry 1 failed"));
        match &msg {
            GetMsg::Request { htl, .. } => {
                assert_eq!(
                    *htl, 5,
                    "First retry: 10/2=5 (attempts_at_hop incremented to 2)"
                );
            }
            other @ GetMsg::Response { .. }
            | other @ GetMsg::ResponseStreaming { .. }
            | other @ GetMsg::ResponseStreamingAck { .. }
            | other @ GetMsg::ForwardingAck { .. } => {
                panic!("Expected Request, got {other}")
            }
        }

        // Second retry: attempts_at_hop becomes 3 → 10/3 = 3
        let (op, msg) = op
            .retry_with_next_alternative(10, &[])
            .unwrap_or_else(|_| panic!("retry 2 failed"));
        match &msg {
            GetMsg::Request { htl, .. } => {
                assert_eq!(
                    *htl, MIN_RETRY_HTL,
                    "Second retry: 10/3=3 (clamped to MIN_RETRY_HTL)"
                );
            }
            other @ GetMsg::Response { .. }
            | other @ GetMsg::ResponseStreaming { .. }
            | other @ GetMsg::ResponseStreamingAck { .. }
            | other @ GetMsg::ForwardingAck { .. } => {
                panic!("Expected Request, got {other}")
            }
        }

        // Third retry: attempts_at_hop becomes 4 → 10/4 = 2 → clamped to MIN_RETRY_HTL=3
        let (_op, msg) = op
            .retry_with_next_alternative(10, &[])
            .unwrap_or_else(|_| panic!("retry 3 failed"));
        match &msg {
            GetMsg::Request { htl, .. } => {
                assert_eq!(
                    *htl, MIN_RETRY_HTL,
                    "Third retry: 10/4=2 → clamped to MIN_RETRY_HTL"
                );
            }
            other @ GetMsg::Response { .. }
            | other @ GetMsg::ResponseStreaming { .. }
            | other @ GetMsg::ResponseStreamingAck { .. }
            | other @ GetMsg::ForwardingAck { .. } => {
                panic!("Expected Request, got {other}")
            }
        }
    }

    /// Verify that MIN_RETRY_HTL floor is applied even when max_htl is low.
    #[test]
    fn retry_htl_floor_at_min() {
        let alt = make_peer(7001);
        let mut op = make_awaiting_op(vec![alt], &[]);
        // Simulate many attempts to push HTL below floor
        if let Some(GetState::AwaitingResponse(ref mut data)) = op.state {
            data.attempts_at_hop = 20;
        }

        let (_op, msg) = op
            .retry_with_next_alternative(10, &[])
            .unwrap_or_else(|_| panic!("retry failed"));
        match &msg {
            GetMsg::Request { htl, .. } => {
                assert!(
                    *htl >= MIN_RETRY_HTL,
                    "HTL {} should not fall below MIN_RETRY_HTL {}",
                    htl,
                    MIN_RETRY_HTL
                );
            }
            other @ GetMsg::Response { .. }
            | other @ GetMsg::ResponseStreaming { .. }
            | other @ GetMsg::ResponseStreamingAck { .. }
            | other @ GetMsg::ForwardingAck { .. } => {
                panic!("Expected Request, got {other}")
            }
        }
    }

    /// Verify that bloom-filter-visited peers are excluded from fallback injection (#3570).
    #[test]
    fn retry_dbf_fallback_skips_bloom_visited() {
        let visited_peer = make_peer(8001);
        let fresh_peer = make_peer(8002);

        // Create op with the visited peer marked in the bloom filter
        let mut op = make_awaiting_op(vec![], &[]);
        if let Some(GetState::AwaitingResponse(ref mut data)) = op.state {
            if let Some(addr) = visited_peer.socket_addr() {
                data.visited.mark_visited(addr);
            }
            // Remove visited_peer from tried_peers so only bloom filter catches it
            if let Some(addr) = visited_peer.socket_addr() {
                data.tried_peers.remove(&addr);
            }
        }

        let result = op.retry_with_next_alternative(7, &[visited_peer.clone(), fresh_peer.clone()]);
        assert!(
            result.is_ok(),
            "Should find fresh_peer despite visited_peer in bloom"
        );

        let (new_op, _msg) = result.unwrap_or_else(|_| panic!("retry failed"));
        if let Some(GetState::AwaitingResponse(data)) = &new_op.state {
            assert_eq!(
                data.alternatives.len(),
                0,
                "Only fresh_peer should be injected (visited_peer filtered by bloom)"
            );
        } else {
            panic!("Expected AwaitingResponse state");
        }
    }

    /// Verify that retry targets are marked in the bloom filter (#3570).
    #[test]
    fn retry_marks_target_in_bloom_filter() {
        let alt = make_peer(9001);
        let alt_addr = alt.socket_addr().unwrap();
        let op = make_awaiting_op(vec![alt], &[]);

        let (new_op, _msg) = op
            .retry_with_next_alternative(7, &[])
            .unwrap_or_else(|_| panic!("retry failed"));
        if let Some(GetState::AwaitingResponse(data)) = &new_op.state {
            assert!(
                data.visited.probably_visited(alt_addr),
                "Retry target should be marked in bloom filter"
            );
        } else {
            panic!("Expected AwaitingResponse state");
        }
    }

    #[test]
    fn retry_wrong_state_returns_err() {
        // PrepareRequest state
        let op = make_get_op(
            Some(GetState::PrepareRequest(PrepareRequestData {
                instance_id: ContractInstanceId::new([1u8; 32]),
                id: Transaction::new::<GetMsg>(),
                fetch_contract: true,
                subscribe: false,
                blocking_subscribe: false,
            })),
            None,
        );
        assert!(op.retry_with_next_alternative(7, &[]).is_err());

        // ReceivedRequest state
        let op = make_get_op(Some(GetState::ReceivedRequest), None);
        assert!(op.retry_with_next_alternative(7, &[]).is_err());

        // None state
        let op = make_get_op(None, None);
        assert!(op.retry_with_next_alternative(7, &[]).is_err());
    }

    // === Tests for start_targeted_op ===

    #[test]
    fn start_targeted_op_creates_correct_state() {
        let target = make_peer(5001);
        let instance_id = ContractInstanceId::new([99u8; 32]);

        let (op, msg) = start_targeted_op(instance_id, target.clone(), 10);

        // Should be auto_fetch
        assert!(op.auto_fetch, "Targeted op should be marked as auto_fetch");

        // Should be in AwaitingResponse state targeting the peer
        if let Some(GetState::AwaitingResponse(data)) = &op.state {
            assert_eq!(data.instance_id, instance_id);
            assert!(data.fetch_contract);
            assert!(data.requester.is_none());
            assert!(data.tried_peers.contains(&target.socket_addr().unwrap()));
            assert!(data.alternatives.is_empty());
        } else {
            panic!("Expected AwaitingResponse state");
        }

        // Message should be a Request with correct fields
        match msg {
            GetMsg::Request {
                instance_id: msg_id,
                fetch_contract,
                htl,
                ..
            } => {
                assert_eq!(msg_id, instance_id);
                assert!(fetch_contract);
                assert_eq!(htl, 10);
            }
            GetMsg::Response { .. }
            | GetMsg::ResponseStreaming { .. }
            | GetMsg::ResponseStreamingAck { .. }
            | GetMsg::ForwardingAck { .. } => {
                panic!("Expected Request message")
            }
        }
    }

    // === Tests for is_client_initiated ===

    #[test]
    fn is_client_initiated_true_for_prepare_request() {
        let op = make_get_op(
            Some(GetState::PrepareRequest(PrepareRequestData {
                instance_id: ContractInstanceId::new([1u8; 32]),
                id: Transaction::new::<GetMsg>(),
                fetch_contract: true,
                subscribe: false,
                blocking_subscribe: false,
            })),
            None,
        );
        assert!(op.is_client_initiated());
    }

    #[test]
    fn is_client_initiated_true_for_awaiting_no_requester() {
        let op = make_awaiting_op(vec![], &[]);
        assert!(op.is_client_initiated());
    }

    #[test]
    fn is_client_initiated_false_for_awaiting_with_requester() {
        let id = Transaction::new::<GetMsg>();
        let instance_id = ContractInstanceId::new([42u8; 32]);
        let visited = VisitedPeers::new(&id);
        let requester = make_peer(6001);
        let op = GetOp {
            id,
            state: Some(GetState::AwaitingResponse(AwaitingResponseData {
                instance_id,
                retries: 0,
                fetch_contract: true,
                requester: Some(requester),
                current_hop: 7,
                subscribe: false,
                blocking_subscribe: false,
                next_hop: make_peer(6002),
                tried_peers: HashSet::new(),
                alternatives: vec![],
                attempts_at_hop: 1,
                visited,
            })),
            result: None,
            stats: None,
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };
        assert!(!op.is_client_initiated());
    }

    #[test]
    fn is_client_initiated_false_for_auto_fetch() {
        let target = make_peer(7001);
        let instance_id = ContractInstanceId::new([77u8; 32]);
        let (op, _msg) = start_targeted_op(instance_id, target, 10);
        // auto_fetch ops should NOT be client-initiated even though requester is None
        assert!(
            !op.is_client_initiated(),
            "Auto-fetch GET should not be client-initiated"
        );
    }

    #[test]
    fn is_client_initiated_false_for_other_states() {
        let op = make_get_op(Some(GetState::ReceivedRequest), None);
        assert!(!op.is_client_initiated());

        let key = make_contract_key(1);
        let op = make_get_op(Some(GetState::Finished(FinishedData { key })), None);
        assert!(!op.is_client_initiated());

        let op = make_get_op(None, None);
        assert!(!op.is_client_initiated());
    }

    // ── ForwardingAck wire-format pin (#3570 telemetry hook survives) ──────

    /// Pin test: `GetMsg::ForwardingAck` survives Phase 5-final as a wire
    /// variant + telemetry-only handler (no production reader after the
    /// retry block was retired). This serde round-trip guards against
    /// accidental bincode-discriminant shifts that would break
    /// cross-version compatibility for any deployed peer still emitting
    /// the variant.
    #[test]
    fn forwarding_ack_serde_roundtrip() {
        let id = Transaction::new::<GetMsg>();
        let instance_id = ContractInstanceId::new([42; 32]);
        let msg = GetMsg::ForwardingAck { id, instance_id };

        let serialized = bincode::serialize(&msg).expect("serialize");
        let deserialized: GetMsg = bincode::deserialize(&serialized).expect("deserialize");

        match deserialized {
            GetMsg::ForwardingAck {
                id: deser_id,
                instance_id: deser_iid,
            } => {
                assert_eq!(deser_id, id);
                assert_eq!(deser_iid, instance_id);
            }
            other @ GetMsg::Request { .. }
            | other @ GetMsg::Response { .. }
            | other @ GetMsg::ResponseStreaming { .. }
            | other @ GetMsg::ResponseStreamingAck { .. } => {
                panic!("Expected ForwardingAck, got {other}")
            }
        }
    }

    // ── Stream failure retry tests (#3756) ──────────────────────────────────

    /// Regression test: when a streaming GET response fails on the originator
    /// (stream assembly timeout), the operation should retry with the next
    /// alternative peer via retry_with_next_alternative, producing a new
    /// GetMsg::Request. Previously, StreamCancelled was returned as a fatal
    /// error, bypassing retry entirely and surfacing the error to the user.
    #[test]
    fn stream_failure_on_originator_retries_with_next_peer() {
        let alt1 = make_peer(7001);
        let alt2 = make_peer(7002);
        // upstream_addr: None = originator node
        let op = make_awaiting_op(vec![alt1.clone(), alt2], &[]);
        assert!(op.upstream_addr.is_none(), "Must be originator");

        // Simulate stream failure: reconstruct the op as the inline retry does,
        // then call retry_with_next_alternative (same code path as the fix).
        let retry_op = GetOp {
            id: op.id,
            state: op.state,
            result: op.result,
            stats: op.stats,
            upstream_addr: op.upstream_addr,
            local_fallback: op.local_fallback,
            auto_fetch: op.auto_fetch,
            client_return_code: true,
        };

        let result = retry_op.retry_with_next_alternative(7, &[]);
        assert!(
            result.is_ok(),
            "Originator should retry with alternative peer"
        );

        let (new_op, msg) = result.map_err(|_| "retry failed").unwrap();
        // Verify the retry targets the first alternative
        assert!(
            new_op.get_next_hop_addr().is_some(),
            "Retry op must have a valid next hop"
        );
        assert_eq!(
            new_op.get_next_hop_addr().unwrap(),
            alt1.socket_addr().unwrap(),
            "Should retry with the first available alternative"
        );
        // Verify the message is a Request (not a Response or ACK)
        assert!(
            matches!(msg, GetMsg::Request { .. }),
            "Retry should produce a Request message, got {:?}",
            std::mem::discriminant(&msg)
        );
    }

    /// When a streaming GET fails on the originator and all alternatives are
    /// exhausted, retry_with_next_alternative returns Err and the operation
    /// should propagate StreamCancelled to the client.
    #[test]
    fn stream_failure_with_no_alternatives_returns_error() {
        // No alternatives, no fallback peers
        let op = make_awaiting_op(vec![], &[]);
        assert!(op.upstream_addr.is_none(), "Must be originator");

        let retry_op = GetOp {
            id: op.id,
            state: op.state,
            result: op.result,
            stats: op.stats,
            upstream_addr: op.upstream_addr,
            local_fallback: op.local_fallback,
            auto_fetch: op.auto_fetch,
            client_return_code: true,
        };

        let result = retry_op.retry_with_next_alternative(7, &[]);
        assert!(
            result.is_err(),
            "Should fail when no alternatives are available"
        );
    }

    /// On a relay node (upstream_addr is Some), stream failure should NOT
    /// attempt retry -- only the originator retries.
    #[test]
    fn stream_failure_on_relay_does_not_retry() {
        let alt1 = make_peer(8001);
        let mut op = make_awaiting_op(vec![alt1], &[]);
        // Set upstream_addr to make this a relay node
        op.upstream_addr = Some("127.0.0.1:9999".parse().unwrap());

        // The relay code path returns Err(StreamCancelled) immediately,
        // never reaching retry_with_next_alternative. Verify the relay
        // has alternatives available (would retry if it were an originator)
        // but the upstream_addr presence prevents retry.
        assert!(op.upstream_addr.is_some(), "Must be relay node");
        // The actual code checks is_original_requester = upstream_addr.is_none()
        // and only retries when true. This test documents that contract.
    }

    // ── Intermediate node stats tracking tests (#3527) ─────────────────────

    use crate::ring::Location;

    fn make_get_op_with_stats(stats: Option<Box<GetStats>>) -> GetOp {
        GetOp {
            id: Transaction::new::<GetMsg>(),
            state: Some(GetState::ReceivedRequest),
            result: None,
            stats,
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        }
    }

    fn make_get_stats(
        target: PeerKeyLocation,
        contract_location: Location,
    ) -> Option<Box<GetStats>> {
        Some(Box::new(GetStats {
            next_peer: Some(target),
            contract_location,
            first_response_time: None,
            transfer_time: None,
        }))
    }

    /// Non-finalized GET with stats reports ContractOpFailure on timeout.
    #[test]
    fn test_get_failure_outcome_with_stats() {
        let target = make_peer(9001);
        let contract_location = Location::random();
        let op = make_get_op_with_stats(make_get_stats(target.clone(), contract_location));

        match op.outcome() {
            OpOutcome::ContractOpFailure {
                target_peer,
                contract_location: loc,
            } => {
                assert_eq!(target_peer, &target);
                assert_eq!(loc, contract_location);
            }
            other => panic!("Expected ContractOpFailure, got {other:?}"),
        }
    }

    /// Non-finalized GET without stats reports Incomplete.
    #[test]
    fn test_get_failure_outcome_without_stats() {
        let op = make_get_op_with_stats(None);
        assert!(
            matches!(op.outcome(), OpOutcome::Incomplete),
            "GET without stats should return Incomplete"
        );
    }

    /// failure_routing_info() returns correct peer and location from stats.
    #[test]
    fn test_get_failure_routing_info() {
        let target = make_peer(9002);
        let contract_location = Location::random();
        let op = make_get_op_with_stats(make_get_stats(target.clone(), contract_location));

        let (peer, loc) = op.failure_routing_info().expect("should have routing info");
        assert_eq!(peer, target);
        assert_eq!(loc, contract_location);
    }

    /// Round-trip serialization: subscribe field is preserved through bincode.
    /// Note: bincode uses positional encoding, so #[serde(default)] does NOT
    /// provide backward compat with older binaries missing the field. Wire
    /// compat is handled by MIN_COMPATIBLE_VERSION + auto-update at handshake.
    #[test]
    fn test_get_msg_subscribe_roundtrip() {
        use freenet_stdlib::prelude::ContractInstanceId;

        // subscribe=true round-trips correctly
        let msg = GetMsg::Request {
            id: Transaction::new::<GetMsg>(),
            instance_id: ContractInstanceId::new([1; 32]),
            fetch_contract: true,
            htl: 10,
            visited: VisitedPeers::default(),
            subscribe: true,
        };
        let bytes = bincode::serialize(&msg).unwrap();
        let restored: GetMsg = bincode::deserialize(&bytes).unwrap();
        match restored {
            GetMsg::Request { subscribe, .. } => assert!(subscribe),
            _ => panic!("expected Request"),
        }

        // subscribe=false round-trips correctly
        let msg_false = GetMsg::Request {
            id: Transaction::new::<GetMsg>(),
            instance_id: ContractInstanceId::new([2; 32]),
            fetch_contract: true,
            htl: 10,
            visited: VisitedPeers::default(),
            subscribe: false,
        };
        let bytes_false = bincode::serialize(&msg_false).unwrap();
        let restored_false: GetMsg = bincode::deserialize(&bytes_false).unwrap();
        match restored_false {
            GetMsg::Request { subscribe, .. } => assert!(!subscribe),
            _ => panic!("expected Request"),
        }
    }

    // ============ Timer methods regression tests ============

    /// Verify start_timers() sets both timing fields and record_*_end() completes them,
    /// producing a ContractOpSuccess with non-zero durations.
    #[test]
    fn test_get_stats_timer_methods_produce_valid_timing() {
        use crate::ring::{Location, PeerKeyLocation};
        use std::time::Duration;

        let target_peer = PeerKeyLocation::random();
        let contract_location = Location::random();

        let mut stats = GetStats {
            next_peer: Some(target_peer.clone()),
            contract_location,
            first_response_time: None,
            transfer_time: None,
        };

        // start_timers sets both to (now, None)
        stats.start_timers();
        assert!(stats.first_response_time.is_some());
        assert!(stats.transfer_time.is_some());
        assert!(stats.first_response_time.unwrap().1.is_none());
        assert!(stats.transfer_time.unwrap().1.is_none());

        // Simulate some work
        std::thread::sleep(Duration::from_millis(1));

        // record_response_end completes first_response_time
        stats.record_response_end();
        assert!(stats.first_response_time.unwrap().1.is_some());
        assert!(stats.transfer_time.unwrap().1.is_none()); // transfer still open

        std::thread::sleep(Duration::from_millis(1));

        // record_transfer_end completes transfer_time
        stats.record_transfer_end();
        assert!(stats.transfer_time.unwrap().1.is_some());

        // Verify the outcome produces ContractOpSuccess with non-zero durations
        let op = GetOp {
            id: Transaction::new::<GetMsg>(),
            state: None,
            result: Some(GetResult {
                key: make_contract_key(99),
                state: WrappedState::new(vec![1, 2, 3]),
                contract: None,
            }),
            stats: Some(Box::new(stats)),
            upstream_addr: None,
            local_fallback: None,
            auto_fetch: false,
            client_return_code: true,
        };

        match op.outcome() {
            OpOutcome::ContractOpSuccess {
                first_response_time,
                payload_transfer_time,
                ..
            } => {
                assert!(
                    first_response_time > Duration::ZERO,
                    "first_response_time should be positive, got {first_response_time:?}"
                );
                assert!(
                    payload_transfer_time > Duration::ZERO,
                    "payload_transfer_time should be positive, got {payload_transfer_time:?}"
                );
                assert!(
                    payload_transfer_time >= first_response_time,
                    "transfer should take at least as long as response"
                );
            }
            other => panic!("Expected ContractOpSuccess, got {other:?}"),
        }
    }

    /// Verify start_timers resets previous timing (as happens on retry).
    #[test]
    fn test_get_stats_start_timers_resets_on_retry() {
        let mut stats = GetStats {
            next_peer: None,
            contract_location: Location::random(),
            first_response_time: None,
            transfer_time: None,
        };

        stats.start_timers();
        let first_start = stats.first_response_time.unwrap().0;

        std::thread::sleep(std::time::Duration::from_millis(1));

        // Simulate retry: start_timers again
        stats.start_timers();
        let second_start = stats.first_response_time.unwrap().0;

        assert!(
            second_start > first_start,
            "Retry should reset timers to a later instant"
        );
        // End timestamps should be cleared on retry
        assert!(stats.first_response_time.unwrap().1.is_none());
        assert!(stats.transfer_time.unwrap().1.is_none());
    }

    /// Simulates the relay peer local-cache decision from process_message.
    ///
    /// When a relay peer finds a contract locally, it checks whether it has
    /// active local interest (hosting, subscription, or client connections).
    /// If yes, state is current -- serve immediately. If no (stale LRU cache
    /// only), defer to the network with local fallback.
    ///
    /// Returns (local_value, local_fallback) after applying the decision.
    type LocalValue = Option<(ContractKey, WrappedState, Option<ContractContainer>)>;

    fn apply_relay_cache_decision(
        is_relay: bool,
        has_local_interest: bool,
        local_value: LocalValue,
    ) -> (LocalValue, LocalValue) {
        // This mirrors the logic at get.rs ~line 1404.
        let mut local_fallback = None;
        let local_value = if is_relay {
            match &local_value {
                Some(_) if !has_local_interest => {
                    // Stale cache -- defer to network, keep as fallback
                    local_fallback = local_value;
                    None
                }
                _ => {
                    // Active interest -- serve immediately
                    local_value
                }
            }
        } else {
            local_value
        };
        (local_value, local_fallback)
    }

    /// Regression test: relay peers actively hosting a contract (with local
    /// interest via subscription, proximity, or client) must serve it
    /// immediately. Previously, ALL relay peers deferred to the network
    /// regardless of hosting status, causing GET timeouts.
    #[test]
    fn relay_peer_with_local_interest_serves_immediately() {
        let key = make_contract_key(1);
        let state = WrappedState::new(vec![1, 2, 3]);
        let local_value = Some((key, state.clone(), None));

        // Relay peer WITH active local interest: serve immediately
        let (value, fallback) = apply_relay_cache_decision(true, true, local_value.clone());
        assert!(
            value.is_some(),
            "Relay peer with local interest must serve immediately"
        );
        assert!(
            fallback.is_none(),
            "Should not defer to fallback when actively hosting"
        );
    }

    /// Relay peer with stale cache (no local interest) should defer to network
    /// but keep local value as fallback.
    #[test]
    fn relay_peer_with_stale_cache_defers_with_fallback() {
        let key = make_contract_key(1);
        let state = WrappedState::new(vec![1, 2, 3]);
        let local_value = Some((key, state.clone(), None));

        // Relay peer WITHOUT local interest: defer to network
        let (value, fallback) = apply_relay_cache_decision(true, false, local_value.clone());
        assert!(
            value.is_none(),
            "Relay peer without local interest should defer to network"
        );
        assert!(fallback.is_some(), "Stale cache should be kept as fallback");
    }

    /// Non-relay (original requester) always serves local cache regardless
    /// of interest status.
    #[test]
    fn original_requester_always_serves_local_cache() {
        let key = make_contract_key(1);
        let state = WrappedState::new(vec![1, 2, 3]);
        let local_value = Some((key, state.clone(), None));

        let (value, fallback) = apply_relay_cache_decision(false, false, local_value);
        assert!(value.is_some(), "Original requester always serves locally");
        assert!(fallback.is_none());
    }

    /// Relay peer without local cache forwards regardless.
    #[test]
    fn relay_peer_without_cache_forwards_to_network() {
        let (value, fallback) = apply_relay_cache_decision(true, true, None);
        assert!(value.is_none(), "Nothing to serve without cache");
        assert!(fallback.is_none(), "Nothing to fall back to");
    }
}
