camel-master 0.60.0

use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::time::Duration;

use camel_api::{CamelError, InFlightGauge, MetricsCollector, PlatformService};
use camel_component_api::{
    Component, ConsumerContext, ExchangeEnvelope, InFlightClaim, NetworkRetryPolicy,
    is_retryable_camel_error,
};
use tokio::time::timeout;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};

use crate::consumer::DelegateState;
use crate::endpoint::MasterDelegateContext;

/// Property key for the leader-epoch fencing token stamped on every
/// ExchangeEnvelope emitted by a `master:` route's delegate. Downstream sinks
/// SHOULD compare against their current leader's epoch and reject older
/// envelopes (split-brain safety). See ADR-0035.
pub(crate) const LEADER_EPOCH_PROPERTY: &str = "x-camel-leader-epoch";

/// Spawn an epoch-stamping bridge between a delegate consumer and the pipeline.
///
/// Returns `(stamp_tx, JoinHandle)`. The delegate sends `ExchangeEnvelope`s
/// to `stamp_tx`; the bridge stamps each with `x-camel-leader-epoch` =
/// `my_epoch` (a snapshot taken at spawn time, NOT a live read) and forwards
/// to `real_sender`.
///
/// On sender-drop (delegate stopped), the bridge drains remaining buffered
/// envelopes — all stamped with the same constant epoch — and exits.
/// On `parent_cancel` (route shutdown), the bridge aborts immediately.
///
/// The `JoinHandle` MUST be awaited by the caller (typically `stop_delegate`)
/// within `drain_timeout` — it is NOT detached. A detached bridge could
/// continue stamping and forwarding envelopes after the leader has yielded.
pub(crate) fn spawn_epoch_bridge(
    real_sender: tokio::sync::mpsc::Sender<ExchangeEnvelope>,
    my_epoch: u64,
    parent_cancel: CancellationToken,
    in_flight: Option<Arc<InFlightGauge>>,
) -> (
    tokio::sync::mpsc::Sender<ExchangeEnvelope>,
    tokio::task::JoinHandle<()>,
) {
    let (stamp_tx, mut stamp_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(128);

    let handle = tokio::spawn(async move {
        loop {
            tokio::select! {
                _ = parent_cancel.cancelled() => break,
                env = stamp_rx.recv() => {
                    match env {
                        Some(mut env) => {
                            stamp_epoch(&mut env, my_epoch);
                            bridge_mint_claim(&mut env, &in_flight);
                            if real_sender.send(env).await.is_err() {
                                break;
                            }
                        }
                        None => {
                            // Sender dropped (delegate stopped) — drain remaining
                            // buffered envelopes with the same constant epoch.
                            while let Ok(mut env) = stamp_rx.try_recv() {
                                stamp_epoch(&mut env, my_epoch);
                                bridge_mint_claim(&mut env, &in_flight);
                                if real_sender.send(env).await.is_err() {
                                    break;
                                }
                            }
                            break;
                        }
                    }
                }
            }
        }
    });

    (stamp_tx, handle)
}

/// rc-nftni (drainclaim): the bridge mints a claim for envelopes that
/// arrive uncounted — the acceptance point into THIS route's dispatch
/// path is the `stamp_rx` dequeue. Envelopes already carrying a claim
/// (delegates using the counter installed on their synthetic context, or
/// raw-sender components minting at their own acceptance) pass through
/// untouched — never a double count. A failed forward drops the envelope
/// and rolls the claim back (RAII).
fn bridge_mint_claim(env: &mut ExchangeEnvelope, in_flight: &Option<Arc<InFlightGauge>>) {
    if env.in_flight_claim.is_none()
        && let Some(counter) = in_flight
    {
        env.in_flight_claim = Some(InFlightClaim::attach(counter));
    }
}

/// Stamp the leader-epoch fencing token on an envelope's properties.
fn stamp_epoch(env: &mut ExchangeEnvelope, epoch: u64) {
    env.exchange.properties.insert(
        LEADER_EPOCH_PROPERTY.to_string(),
        serde_json::Value::String(epoch.to_string()),
    );
}

/// Emit a leadership transition counter observation. Called by the
/// supervision loop on observed not-leading → leading (`acquired`) and
/// leading → not-leading (`lost`) state edges. See
/// openspec/specs/master-component/spec.md.
pub(crate) fn emit_leadership_transition(
    metrics: &Arc<dyn MetricsCollector>,
    lock: &str,
    route_id: &str,
    event: &str,
) {
    // allow-open-label rc-ycts rc-xl5k rc-gm6s (lock: config name; route_id; event/reason: caller literals)
    metrics.record_counter(
        "master_leadership_transitions_total",
        1.0,
        &[("lock", lock), ("route_id", route_id), ("event", event)],
    );
    // The state gauge rides the same observed edge (rc-02dx): 1 on
    // acquire, 0 on loss — steady-state readability ("who leads lock X
    // now") without changing the counters' transition-only semantics.
    metrics.set_master_leadership(lock, event == "acquired");
}

/// Emit a delegate lifecycle counter observation. All lifecycle
/// emissions go through this helper so the uniform label schema
/// (lock, route_id, event, reason) is structural.
pub(crate) fn emit_lifecycle(
    metrics: &Arc<dyn MetricsCollector>,
    lock: &str,
    route_id: &str,
    event: &str,
    reason: &str,
) {
    // allow-open-label rc-ycts rc-xl5k rc-gm6s (lock: config name; route_id; event/reason: caller literals)
    metrics.record_counter(
        "master_delegate_lifecycle_total",
        1.0,
        &[
            ("lock", lock),
            ("route_id", route_id),
            ("event", event),
            ("reason", reason),
        ],
    );
}

/// Waits on a delegate's background task until it ends or `run_token` is
/// cancelled (then aborts it). `None` means a blocking `start()` already
/// finished its work. A failure while not cancelled is returned so
/// supervision treats it like a blocking-`start()` error.
async fn supervise_background(
    bg: Option<tokio::task::JoinHandle<Result<(), CamelError>>>,
    run_token: &CancellationToken,
) -> Result<(), CamelError> {
    let Some(mut bg) = bg else {
        return Ok(());
    };
    tokio::select! {
        result = &mut bg => match result {
            Ok(Ok(())) => Ok(()),
            Ok(Err(err)) if !run_token.is_cancelled() => Err(err),
            Ok(Err(err)) => {
                debug!(error = %err, "delegate background task error during shutdown");
                Ok(())
            }
            Err(join_err) if !run_token.is_cancelled() => Err(CamelError::ProcessorError(
                format!("master delegate background task panicked: {join_err}"),
            )),
            Err(join_err) => {
                debug!(error = %join_err, "delegate background task panicked during shutdown");
                Ok(())
            }
        },
        _ = run_token.cancelled() => {
            bg.abort();
            Ok(())
        }
    }
}

pub(crate) async fn stop_delegate(
    state: &mut DelegateState,
    drain_timeout: Duration,
    lock_name: &str,
    route_id: &str,
    metrics: &Arc<dyn MetricsCollector>,
) -> Result<(), CamelError> {
    if let DelegateState::Active {
        run_token,
        mut handle,
        mut bridge_handle,
        ..
    } = std::mem::replace(state, DelegateState::Inactive)
    {
        run_token.cancel();
        let delegate_result: Result<(), CamelError> =
            match timeout(drain_timeout, &mut handle).await {
                Ok(Ok(Ok(()))) => Ok(()),
                Ok(Ok(Err(err))) => Err(err),
                Ok(Err(e)) if e.is_panic() => {
                    // log-policy: system-broken
                    error!(error = %e, "master delegate task panicked");
                    Err(CamelError::ProcessorError(format!(
                        "master delegate task panicked: {e}"
                    )))
                }
                Ok(Err(e)) => {
                    warn!(error = %e, "master delegate task cancelled");
                    Err(CamelError::ProcessorError(format!(
                        "master delegate task cancelled: {e}"
                    )))
                }
                Err(_) => {
                    warn!("master delegate shutdown timed out, aborting");
                    handle.abort();
                    let _ = handle.await;
                    Ok(())
                }
            };

        // Drain the epoch-stamping bridge. It forwards any envelopes still
        // buffered after the delegate stopped, then exits. On timeout, abort
        // to prevent a detached task from sending stale exchanges.
        if let Some(mut bridge) = bridge_handle.take() {
            match timeout(drain_timeout, &mut bridge).await {
                Ok(Ok(())) => {}
                Ok(Err(e)) => {
                    warn!(error = %e, "epoch bridge task failed");
                }
                Err(_) => {
                    warn!("epoch bridge drain timed out, aborting");
                    bridge.abort();
                    let _ = bridge.await;
                }
            }
        }

        emit_lifecycle(metrics, lock_name, route_id, "stopped", "none");

        return delegate_result;
    }
    Ok(())
}

pub(crate) struct ReconcileContext<'a> {
    pub(crate) lock_name: &'a str,
    pub(crate) delegate_component: &'a Arc<dyn Component>,
    pub(crate) delegate_uri: &'a str,
    pub(crate) route_id: String,
    pub(crate) sender: &'a tokio::sync::mpsc::Sender<ExchangeEnvelope>,
    pub(crate) parent_cancel: &'a CancellationToken,
    pub(crate) drain_timeout: Duration,
    pub(crate) metrics: &'a Arc<dyn MetricsCollector>,
    pub(crate) platform_service: &'a Arc<dyn PlatformService>,
    pub(crate) runtime: Arc<dyn camel_component_api::RuntimeObservability>,
    /// Monotonic fencing token, snapshotted by `reconcile_event` on the
    /// `StartedLeading` transition. See ADR-0035.
    pub(crate) leader_epoch: Arc<AtomicU64>,
    /// Delegate acquisition attempts within the current acquisition epoch
    /// (design §2). `AtomicU32` because the context is held across `.await`s
    /// inside the `tokio::spawn`ed supervision task (`Cell<u32>` is `!Sync`
    /// and would make the future `!Send`); access is single-task, so
    /// `Relaxed` ordering suffices.
    pub(crate) attempts: AtomicU32,
    /// Reconnect policy consulted before every delegate create attempt.
    pub(crate) reconnect: NetworkRetryPolicy,
    /// rc-nftni (drainclaim): the route's context-global accepted-not-
    /// completed gauge. Installed on the delegate's synthetic
    /// `ConsumerContext` (so `send()`/`send_and_wait()` delegates mint at
    /// `stamp_tx` entry) and consulted by the epoch bridge as a mint-if-none
    /// safety net for delegates that push raw envelopes.
    pub(crate) in_flight: Option<Arc<InFlightGauge>>,
}

pub(crate) async fn reconcile_event(
    event: camel_api::LeadershipEvent,
    state: &mut DelegateState,
    ctx: &ReconcileContext<'_>,
) -> Result<(), CamelError> {
    match event {
        camel_api::LeadershipEvent::StartedLeading => {
            // Epoch-idempotence: an identical delivery (watch re-delivery
            // without a takeover) while Active at the published epoch is a
            // no-op — no drain, no recreate, no lifecycle emission. A
            // differing epoch (term bump across a takeover) falls through to
            // the full re-reconciliation below.
            let current_epoch = ctx.leader_epoch.load(Ordering::Acquire);
            if let DelegateState::Active { epoch, .. } = state
                && *epoch == current_epoch
            {
                debug!(
                    lock = %ctx.lock_name,
                    current_epoch,
                    "dropping duplicate StartedLeading delivery at the published epoch"
                );
                return Ok(());
            }

            // A differing epoch (term bump across a takeover) detected on an
            // Active delegate begins a new acquisition epoch: the budget
            // resets below, before the consult. Captured before the drain —
            // the drain destroys the Active state and its epoch.
            let term_bump = matches!(&*state, DelegateState::Active { .. });

            stop_delegate(
                state,
                ctx.drain_timeout,
                ctx.lock_name,
                ctx.route_id.as_str(),
                ctx.metrics,
            )
            .await?;

            let delegate_ctx = MasterDelegateContext {
                delegate_component: Arc::clone(ctx.delegate_component),
                metrics: Arc::clone(ctx.metrics),
                platform_service: Arc::clone(ctx.platform_service),
            };

            // Exact acquisition budget (design §2): AFTER the drain
            // (no-zombie ordering). The consult evaluates the pre-increment
            // count, but the counter increments on EVERY delivery regardless
            // of the verdict — it counts deliveries, not successful creates.
            // A refused consult performs no create.
            if term_bump {
                ctx.attempts.swap(0, Ordering::Relaxed);
            }
            let count = ctx.attempts.load(Ordering::Relaxed);
            ctx.attempts
                .store(count.saturating_add(1), Ordering::Relaxed);
            if !ctx.reconnect.should_retry(count) {
                return Ok(());
            }

            // Logged only when the consult admits the create: a refused
            // delivery (exhausted budget or disabled policy) performs no
            // create and must not claim acquisition.
            info!(lock = %ctx.lock_name, "master leadership acquired");

            let endpoint = match ctx
                .delegate_component
                .create_endpoint(ctx.delegate_uri, &delegate_ctx)
            {
                Ok(endpoint) => endpoint,
                Err(err) => {
                    let is_retryable = is_retryable_camel_error(&err);
                    emit_lifecycle(
                        ctx.metrics,
                        ctx.lock_name,
                        ctx.route_id.as_str(),
                        "create_error",
                        if is_retryable {
                            "transient"
                        } else {
                            "permanent"
                        },
                    );
                    if is_retryable {
                        warn!(lock = %ctx.lock_name, error = %err, "transient delegate endpoint error (will retry)");
                        return Ok(()); // swallow transient — retry via next tick
                    }
                    return Err(err); // permanent — propagate to retry loop for fail-fast
                }
            };

            let mut consumer = match endpoint.create_consumer(Arc::clone(&ctx.runtime)) {
                Ok(consumer) => consumer,
                Err(err) => {
                    let is_retryable = is_retryable_camel_error(&err);
                    emit_lifecycle(
                        ctx.metrics,
                        ctx.lock_name,
                        ctx.route_id.as_str(),
                        "create_error",
                        if is_retryable {
                            "transient"
                        } else {
                            "permanent"
                        },
                    );
                    if is_retryable {
                        warn!(lock = %ctx.lock_name, error = %err, "transient delegate consumer error (will retry)");
                        return Ok(()); // swallow transient — retry via next tick
                    }
                    return Err(err); // permanent — propagate to retry loop for fail-fast
                }
            };

            // Snapshot the epoch ONCE — stale bridges carry stale terms
            // (downstream sinks reject envelopes with an older epoch).
            let my_epoch = ctx.leader_epoch.load(Ordering::Acquire);

            // Spawn the epoch-stamping bridge. The delegate sends to `stamp_tx`;
            // the bridge forwards stamped envelopes to the pipeline via
            // `ctx.sender`. Lifetime is tied to the delegate: dropping
            // `stamp_tx` (when the delegate stops) causes the bridge to
            // drain and exit.
            let (stamp_tx, bridge_handle) = spawn_epoch_bridge(
                ctx.sender.clone(),
                my_epoch,
                ctx.parent_cancel.child_token(),
                ctx.in_flight.clone(),
            );

            let run_token = ctx.parent_cancel.child_token();
            // rc-nftni: install the route's counter on the delegate's
            // synthetic context so `send()`/`send_and_wait()` delegates mint
            // at `stamp_tx` entry (full bridge-queue residency); raw-sender
            // delegates capture it via `ConsumerContext::in_flight_counter`
            // the same way. The bridge's mint-if-none covers anything that
            // still arrives uncounted.
            let mut delegate_ctx =
                ConsumerContext::new(stamp_tx, run_token.clone(), ctx.route_id.clone());
            if let Some(counter) = ctx.in_flight.clone() {
                delegate_ctx = delegate_ctx.with_in_flight_counter(counter);
            }
            let supervise_token = run_token.clone();
            let handle = tokio::spawn(async move {
                consumer.start(delegate_ctx).await?;
                // rc-w02ns: Explicit consumers (camel-jms, camel-kafka) return
                // from start() right after spawning a background task. Hold
                // the delegate alive on that task (or on cancellation) before
                // stop(), exactly as the route controller does; stopping
                // straight after start() killed the delegate and churned
                // leadership.
                let supervised =
                    supervise_background(consumer.background_task_handle(), &supervise_token).await;
                consumer.stop().await?;
                supervised
            });

            *state = DelegateState::Active {
                run_token,
                handle,
                bridge_handle: Some(bridge_handle),
                epoch: my_epoch,
            };
            emit_lifecycle(
                ctx.metrics,
                ctx.lock_name,
                ctx.route_id.as_str(),
                "started",
                "none",
            );
        }
        camel_api::LeadershipEvent::StoppedLeading => {
            info!(lock = %ctx.lock_name, "master leadership lost");
            stop_delegate(
                state,
                ctx.drain_timeout,
                ctx.lock_name,
                ctx.route_id.as_str(),
                ctx.metrics,
            )
            .await?;
        }
        // Future leadership events require no state change.
        _ => {}
    }
    Ok(())
}

// Uses camel_component_api::is_retryable_camel_error for transient/permanent
// classification (rc-7ct consolidation, was master-local in rc-i1z).

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::AtomicU64;
    use std::time::Duration;

    use camel_api::{Body, CamelError, Exchange, Message, NoOpMetrics};
    use camel_component_api::ExchangeEnvelope;
    use serde_json::Value;
    use tokio::time::timeout;
    use tokio_util::sync::CancellationToken;

    use super::{spawn_epoch_bridge, stop_delegate};
    use crate::consumer::DelegateState;

    fn make_envelope(text: &str) -> ExchangeEnvelope {
        ExchangeEnvelope {
            exchange: Exchange::new(Message::new(Body::Text(text.to_string()))),
            reply_tx: None,
            in_flight_claim: None,
        }
    }

    /// Verifies that a basic envelope sent through the bridge is stamped
    /// with the spawn-time epoch on its properties, then forwarded.
    #[tokio::test]
    async fn fence_token_stamped_on_exchange() {
        let (pipeline_tx, mut pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let cancel = CancellationToken::new();

        let (stamp_tx, _bridge) = spawn_epoch_bridge(pipeline_tx, 42, cancel, None);

        stamp_tx.send(make_envelope("hello")).await.unwrap();
        drop(stamp_tx);

        let received = timeout(Duration::from_secs(1), pipeline_rx.recv())
            .await
            .expect("bridge should forward within 1s")
            .expect("channel should yield an envelope");
        let epoch_val = received
            .exchange
            .properties
            .get(super::LEADER_EPOCH_PROPERTY)
            .expect("x-camel-leader-epoch should be stamped");
        assert_eq!(epoch_val, &Value::String("42".to_string()));
    }

    /// Verifies the drain-on-sender-drop contract: buffered envelopes are
    /// forwarded (all stamped with the same constant epoch) before the
    /// bridge exits.
    #[tokio::test]
    async fn bridge_drains_buffer_on_delegate_stop() {
        let (pipeline_tx, mut pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let cancel = CancellationToken::new();

        let (stamp_tx, bridge_handle) = spawn_epoch_bridge(pipeline_tx, 5, cancel, None);

        for i in 0..3 {
            stamp_tx
                .send(make_envelope(&format!("msg-{i}")))
                .await
                .unwrap();
        }
        drop(stamp_tx);

        // Bridge should drain all 3, then exit.
        timeout(Duration::from_secs(1), bridge_handle)
            .await
            .expect("bridge should finish within 1s")
            .expect("bridge task should not panic");

        for i in 0..3 {
            let env = timeout(Duration::from_secs(2), pipeline_rx.recv())
                .await
                .expect("envelope should arrive after drain within 2s")
                .expect("pipeline channel alive");
            let epoch = env
                .exchange
                .properties
                .get(super::LEADER_EPOCH_PROPERTY)
                .and_then(|v| v.as_str())
                .and_then(|s| s.parse::<u64>().ok());
            assert_eq!(epoch, Some(5), "envelope {i} should carry constant epoch 5");
        }
        assert!(
            pipeline_rx.try_recv().is_err(),
            "bridge should have exited — no more envelopes"
        );
    }

    /// Verifies the bridge stamps with its SPAWN-TIME snapshot, not a live
    /// read. If the live epoch changes mid-stream, envelopes still carry
    /// the snapshot value.
    #[tokio::test]
    async fn bridge_snapshot_constant_across_epoch_change() {
        let epoch = Arc::new(AtomicU64::new(7));
        let (pipeline_tx, mut pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let cancel = CancellationToken::new();

        // Snapshot epoch=7 at bridge spawn
        let my_epoch = epoch.load(std::sync::atomic::Ordering::Acquire);
        let (stamp_tx, bridge_handle) = spawn_epoch_bridge(pipeline_tx, my_epoch, cancel, None);

        // Change the live epoch — bridge should NOT pick this up
        epoch.store(99, std::sync::atomic::Ordering::Release);

        stamp_tx.send(make_envelope("stale")).await.unwrap();
        drop(stamp_tx);
        timeout(Duration::from_secs(1), bridge_handle)
            .await
            .unwrap()
            .unwrap();

        let env = timeout(Duration::from_secs(2), pipeline_rx.recv())
            .await
            .expect("envelope within 2s")
            .expect("pipeline channel alive");
        let stamped = env
            .exchange
            .properties
            .get(super::LEADER_EPOCH_PROPERTY)
            .and_then(|v| v.as_str())
            .expect("stamp should be present");
        assert_eq!(
            stamped, "7",
            "bridge must stamp with snapshot (7), not live (99)"
        );
    }

    /// Verifies that a bridge spawned with an old epoch stamps envelopes with
    /// its spawn-time snapshot even if the live epoch changes. This simulates
    /// a stale bridge from a previous leader term — downstream sinks should
    /// reject envelopes with an epoch older than the current leader's.
    #[tokio::test]
    async fn stale_bridge_carries_old_epoch_not_current() {
        let (pipeline_tx, mut pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let cancel = CancellationToken::new();

        // Bridge spawned with epoch=3 (e.g., from a previous leader term).
        let (stamp_tx, bridge_handle) = spawn_epoch_bridge(pipeline_tx, 3, cancel, None);

        stamp_tx
            .send(make_envelope("from-old-leader"))
            .await
            .unwrap();
        drop(stamp_tx);
        timeout(Duration::from_secs(1), bridge_handle)
            .await
            .expect("bridge should finish within 1s")
            .expect("bridge task should not panic");

        let env = timeout(Duration::from_secs(2), pipeline_rx.recv())
            .await
            .expect("stamped envelope should arrive within 2s")
            .expect("pipeline channel alive");
        let stamped = env
            .exchange
            .properties
            .get(super::LEADER_EPOCH_PROPERTY)
            .and_then(|v| v.as_str())
            .expect("x-camel-leader-epoch should be stamped");
        assert_eq!(
            stamped, "3",
            "old bridge stamps with its snapshot epoch (3), not the current epoch"
        );
    }

    /// Verifies the bridge aborts promptly when the parent cancel token fires.
    #[tokio::test]
    async fn bridge_aborts_on_parent_cancel() {
        let (pipeline_tx, mut _pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let cancel = CancellationToken::new();

        let (_stamp_tx, bridge_handle) = spawn_epoch_bridge(pipeline_tx, 1, cancel.clone(), None);

        cancel.cancel();

        // Bridge should exit promptly on parent cancel.
        timeout(Duration::from_millis(100), bridge_handle)
            .await
            .expect("bridge should abort within 100ms")
            .expect("bridge task should not panic");
    }

    #[tokio::test]
    async fn bridge_aborted_on_drain_timeout_when_pipeline_full() {
        // Backpressure scenario: pipeline channel capacity=1, we fill it,
        // then the bridge tries to send → blocks → drain timeout → abort.
        let (pipeline_tx, _pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(1);
        let cancel = CancellationToken::new();

        // Fill the pipeline channel so bridge.send() will block
        pipeline_tx.send(make_envelope("blocker")).await.unwrap();

        let (stamp_tx, mut bridge_handle) = spawn_epoch_bridge(pipeline_tx, 42, cancel, None);

        // Send an envelope to the bridge — it will try to forward to the
        // full pipeline channel and block.
        stamp_tx.send(make_envelope("will-block")).await.unwrap();
        drop(stamp_tx);

        // Drain with a very short timeout — bridge is stuck in send().await
        let drain_timeout = Duration::from_millis(50);
        match timeout(drain_timeout, &mut bridge_handle).await {
            Err(_) => {
                // Timeout hit — abort the bridge (mirrors stop_delegate behavior)
                bridge_handle.abort();
                let _ = bridge_handle.await;
            }
            Ok(_) => panic!("bridge should not complete while pipeline is full"),
        }

        // If we reach here, abort worked — bridge is no longer running.
    }

    /// Verifies that `stop_delegate` drains the epoch-bridge even when the
    /// delegate task returns an error. Without this fix, the early `return`
    /// on delegate error skips the bridge drain block, leaking a detached task.
    #[tokio::test]
    async fn stop_delegate_drains_bridge_on_delegate_error() {
        let delegate_handle = tokio::spawn(async {
            Err::<(), CamelError>(CamelError::ProcessorError("test delegate failure".into()))
        });

        let cancel = CancellationToken::new();
        let (pipeline_tx, _pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let (stamp_tx, bridge_handle) = spawn_epoch_bridge(pipeline_tx, 42, cancel, None);

        // Keep an AbortHandle to verify completion after stop_delegate.
        let bridge_abort = bridge_handle.abort_handle();
        assert!(
            !bridge_abort.is_finished(),
            "bridge should not be finished before stop_delegate"
        );

        // Send an envelope to give the bridge real work, then drop stamp_tx
        // so the bridge can exit after forwarding — avoids forcing the full
        // 5 s drain timeout. The drain block in stop_delegate still exercises
        // its full code path.
        stamp_tx.send(make_envelope("test")).await.unwrap();
        drop(stamp_tx);

        let mut state = DelegateState::Active {
            run_token: CancellationToken::new(),
            handle: delegate_handle,
            bridge_handle: Some(bridge_handle),
            epoch: 42,
        };

        let metrics: Arc<dyn camel_api::MetricsCollector> = Arc::new(NoOpMetrics);
        let result = stop_delegate(
            &mut state,
            Duration::from_secs(5),
            "test-lock",
            "test-route",
            &metrics,
        )
        .await;

        // (a) result should be Err matching the delegate error.
        assert!(result.is_err(), "expected Err from delegate failure");
        assert!(
            result
                .unwrap_err()
                .to_string()
                .contains("test delegate failure"),
            "error should contain delegate message"
        );

        // (b) bridge handle has exited — drain block ran.
        assert!(
            bridge_abort.is_finished(),
            "bridge should have been drained after delegate error"
        );
    }

    /// Verifies that `stop_delegate` drains the epoch-bridge even when the
    /// delegate times out (abort path). The timeout arm falls through to the
    /// drain block; this test ensures that remains true after restructuring.
    #[tokio::test]
    async fn stop_delegate_drains_bridge_on_delegate_timeout() {
        let delegate_handle = tokio::spawn(async {
            // Never completes — forces timeout in stop_delegate.
            std::future::pending::<Result<(), CamelError>>().await
        });

        let cancel = CancellationToken::new();
        let (pipeline_tx, _pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let (stamp_tx, bridge_handle) = spawn_epoch_bridge(pipeline_tx, 42, cancel, None);

        let bridge_abort = bridge_handle.abort_handle();
        assert!(
            !bridge_abort.is_finished(),
            "bridge should not be finished before stop_delegate"
        );

        stamp_tx.send(make_envelope("timeout-test")).await.unwrap();

        let mut state = DelegateState::Active {
            run_token: CancellationToken::new(),
            handle: delegate_handle,
            bridge_handle: Some(bridge_handle),
            epoch: 42,
        };

        let metrics: Arc<dyn camel_api::MetricsCollector> = Arc::new(NoOpMetrics);
        let result = stop_delegate(
            &mut state,
            Duration::from_millis(100),
            "test-lock",
            "test-route",
            &metrics,
        )
        .await;

        // (a) timeout path returns Ok(()).
        assert!(result.is_ok(), "timeout path should return Ok");

        // (b) bridge handle has exited.
        assert!(
            bridge_abort.is_finished(),
            "bridge should have been drained after delegate timeout"
        );

        drop(stamp_tx);
    }

    /// rc-nftni (drainclaim): envelopes that arrive uncounted get a claim
    /// minted by the bridge at its acceptance dequeue — mint/carry/release
    /// with exact totals.
    #[tokio::test]
    async fn bridge_mints_claim_for_uncounted_envelopes() {
        let counter = Arc::new(camel_api::InFlightGauge::new());
        let (pipeline_tx, mut pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let cancel = CancellationToken::new();

        let (stamp_tx, bridge_handle) =
            spawn_epoch_bridge(pipeline_tx, 9, cancel, Some(Arc::clone(&counter)));

        stamp_tx.send(make_envelope("uncounted")).await.unwrap();
        drop(stamp_tx);
        timeout(Duration::from_secs(1), bridge_handle)
            .await
            .expect("bridge should finish within 1s")
            .expect("bridge task should not panic");

        let mut received = timeout(Duration::from_secs(2), pipeline_rx.recv())
            .await
            .expect("envelope must arrive within 2s")
            .expect("pipeline channel alive");
        let claim = received
            .in_flight_claim
            .take()
            .expect("bridge must have minted a claim for the uncounted envelope");
        // Exact total: the minted claim is the only live claim.
        assert_eq!(counter.total(), 1);
        drop(claim);
        drop(received);
        assert_eq!(
            counter.total(),
            0,
            "dropping the envelope must release exactly once"
        );
    }

    /// rc-nftni (drainclaim): envelopes already carrying a claim (delegate
    /// minted via the counter installed on its synthetic context) pass
    /// through the bridge untouched — never a double count.
    #[tokio::test]
    async fn bridge_preserves_existing_claim_without_double_counting() {
        let counter = Arc::new(camel_api::InFlightGauge::new());
        let (pipeline_tx, mut pipeline_rx) = tokio::sync::mpsc::channel::<ExchangeEnvelope>(10);
        let cancel = CancellationToken::new();

        let (stamp_tx, bridge_handle) =
            spawn_epoch_bridge(pipeline_tx, 9, cancel, Some(Arc::clone(&counter)));

        // The delegate's own acceptance mint — exactly as ConsumerContext::send
        // would attach it on the synthetic context.
        let mut pre_claimed = make_envelope("already-counted");
        pre_claimed.in_flight_claim = Some(camel_component_api::InFlightClaim::attach(&counter));
        assert_eq!(counter.total(), 1);

        stamp_tx.send(pre_claimed).await.unwrap();
        drop(stamp_tx);
        timeout(Duration::from_secs(1), bridge_handle)
            .await
            .expect("bridge should finish within 1s")
            .expect("bridge task should not panic");

        let mut received = timeout(Duration::from_secs(2), pipeline_rx.recv())
            .await
            .expect("envelope must arrive within 2s")
            .expect("pipeline channel alive");
        let claim = received
            .in_flight_claim
            .take()
            .expect("pre-existing claim must survive the bridge");
        assert_eq!(
            counter.total(),
            1,
            "bridge must not mint a sibling for an already-counted envelope"
        );
        drop(claim);
        drop(received);
        assert_eq!(counter.total(), 0);
    }
}