camel-processor 0.51.0

//! Resequencer — continuation-boundary EIP.
//!
//! The resequencer is a `CompiledStep` + `StepLifecycle`. `call(input)` sends
//! the input into a bounded actor channel; an actor buffers + computes ready
//! outputs + sends them to a post-driver that drives the owned post-continuation;
//! `call()` returns a control ack. The main pipeline ends at the resequencer.
//!
//! Architecture: See ADR-0029 (resequencer continuation boundary).

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

use async_trait::async_trait;
use tokio::sync::{Mutex as TokioMutex, mpsc};
use tokio::task::JoinHandle;
use tower::Service;

pub mod batch;
pub mod stream;

use camel_api::{
    BoxProcessor, CamelError, MetricsCollector, StepLifecycle, StepShutdownReason,
    exchange::Exchange, message::Message, processor::SyncBoxProcessor,
};

/// Rate-limit window for InOut warning log emission.
const INOUT_WARN_INTERVAL: Duration = Duration::from_secs(30);

/// Configuration for the `ResequencerService`.
#[derive(Clone, Default)]
pub struct ResequencerConfig {
    /// Allow `InOut` exchanges to pass through the resequencer without
    /// emitting a warning. Defaults to `false`.
    pub allow_inout: bool,
    /// Optional metrics collector for incrementing operational counters.
    pub metrics: Option<Arc<dyn MetricsCollector>>,
    /// Optional route ID for metric labels.
    pub route_id: Option<String>,
}

impl std::fmt::Debug for ResequencerConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ResequencerConfig")
            .field("allow_inout", &self.allow_inout)
            .field("metrics", &self.metrics.as_ref().map(|_| "<metrics>"))
            .field("route_id", &self.route_id)
            .finish()
    }
}

// ── Property keys (CamelCase, matching CAMEL_AGGREGATED_COMPLETION_REASON) ──

/// Set on the ack exchange: `true` when the resequencer accepted the input.
pub const CAMEL_RESEQUENCER_ACCEPTED: &str = "CamelResequencerAccepted";

/// Set on the ack exchange: `true` when the input was dropped during shutdown.
pub const CAMEL_RESEQUENCER_DROPPED: &str = "CamelResequencerDropped";

/// Set on the ack exchange: `true` when an InOut exchange reaches the resequencer.
pub const CAMEL_RESEQUENCER_INOUT_WARN: &str = "CamelResequencerInoutWarn";

// ── Policy trait ──

/// Buffer / ordering policy for a resequencer.
///
/// Implementations (batch, stream) live in sibling modules.
#[async_trait]
pub trait ResequencePolicy: Send + Sync + 'static {
    /// Accept an input; return the list of now-ready exchanges (in emit order).
    async fn accept(&self, input: Exchange) -> Vec<Exchange>;

    /// Flush all buffered state (shutdown). Return any remaining, ordered.
    async fn flush(&self) -> Vec<Exchange>;

    /// Stable name for logging / diagnostics.
    fn name(&self) -> &'static str;

    /// Number of exchanges currently held awaiting their release condition
    /// (the awaiting-sequence buffer). Reported by the service as
    /// `camel_queue_depth{queue="resequencer:<route>"}`. Default 0 for
    /// stateless policies (passthrough).
    fn buffered(&self) -> usize {
        0
    }

    /// Set the driver channel for timeout-triggered emissions.
    /// Default is a no-op. `BatchPolicy` overrides this to receive
    /// the channel that feeds the post-driver.
    fn set_timeout_tx(&self, _tx: tokio::sync::mpsc::Sender<Exchange>) {}
}

// ── Service ──

/// Continuation-boundary resequencer.
///
/// Owns an actor task (consumes from input channel, calls `policy.accept(input)`)
/// and a post-driver task (consumes ready exchanges, drives `post_continuation`).
/// `Service::call(input)` sends into the bounded input channel and returns an ack.
#[derive(Clone)]
pub struct ResequencerService {
    policy: Arc<dyn ResequencePolicy>,
    config: ResequencerConfig,
    /// Bounded input channel sender. Wrapped in `Option` so `shutdown` can
    /// `take()` it to signal EOF to the actor (all sender clones must drop).
    input_tx: Arc<Mutex<Option<mpsc::Sender<Exchange>>>>,
    /// Post-driver channel sender. The actor task holds a clone; we hold one
    /// here for shutdown flush. Wrapped in `Option` so `shutdown` can take it
    /// to close the post-driver channel after flush.
    driver_tx: Arc<Mutex<Option<mpsc::Sender<Exchange>>>>,
    actor_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
    driver_handle: Arc<Mutex<Option<JoinHandle<()>>>>,
    shutdown_started: Arc<Mutex<bool>>,
    /// Post-step lifecycles to drain after post-driver quiesces (oracle Fix 2).
    /// Empty for Task 1a (no post-steps yet); filled by Tasks 1b/2/3.
    post_lifecycles: Arc<Mutex<Vec<Arc<dyn StepLifecycle>>>>,
    /// Metric counter for InOut exchanges that reach the resequencer.
    inout_counter: Arc<AtomicU64>,
    /// Rate-limit last-warn timestamp (TokioMutex for async safety).
    last_inout_warn: Arc<TokioMutex<Option<Instant>>>,
    /// Optional metrics collector for operational counters.
    metrics: Option<Arc<dyn MetricsCollector>>,
    /// Route ID for metric labels.
    route_id: Option<String>,
}

impl std::fmt::Debug for ResequencerService {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ResequencerService")
            .field("policy", &self.policy.name())
            .finish_non_exhaustive()
    }
}

impl ResequencerService {
    /// Create a new resequencer with the given policy, continuation, and post-step lifecycles.
    ///
    /// * `input_capacity` — bounded channel capacity (default 1024). Backpressure
    ///   propagates to the caller via `send().await`.
    /// * `post_lifecycles` — lifecycle handles for steps AFTER the resequencer
    ///   (drained in `shutdown` after post-driver quiesces; empty for Task 1a).
    ///
    /// # Panics
    ///
    /// Panics if called outside a Tokio runtime context: `new()` spawns the actor and
    /// post-driver tasks via `tokio::spawn`.
    pub fn new(
        policy: Arc<dyn ResequencePolicy>,
        post_continuation: BoxProcessor,
        input_capacity: usize,
        post_lifecycles: Vec<Arc<dyn StepLifecycle>>,
    ) -> Self {
        Self::with_config(
            policy,
            post_continuation,
            input_capacity,
            post_lifecycles,
            ResequencerConfig::default(),
        )
    }

    /// Full constructor with explicit `ResequencerConfig`.
    ///
    /// # Panics
    ///
    /// Panics if called outside a Tokio runtime context.
    pub fn with_config(
        policy: Arc<dyn ResequencePolicy>,
        post_continuation: BoxProcessor,
        input_capacity: usize,
        post_lifecycles: Vec<Arc<dyn StepLifecycle>>,
        config: ResequencerConfig,
    ) -> Self {
        // Bounded channel for input exchanges → actor
        let (input_tx, mut input_rx) = mpsc::channel::<Exchange>(input_capacity);

        // Post-driver channel: actor → post-driver (bounded, single consumer)
        let (driver_tx, mut driver_rx) = mpsc::channel::<Exchange>(input_capacity);

        // Shared (Arc<Mutex<Option<T>>>) wrappers
        let input_tx_shared: Arc<Mutex<Option<mpsc::Sender<Exchange>>>> =
            Arc::new(Mutex::new(Some(input_tx)));
        let driver_tx_shared: Arc<Mutex<Option<mpsc::Sender<Exchange>>>> =
            Arc::new(Mutex::new(Some(driver_tx.clone()))); // we keep one for flush; actor gets clone
        let actor_handle: Arc<Mutex<Option<JoinHandle<()>>>> = Arc::new(Mutex::new(None));
        let driver_handle: Arc<Mutex<Option<JoinHandle<()>>>> = Arc::new(Mutex::new(None));
        let shutdown_started = Arc::new(Mutex::new(false));

        // Wire the driver channel into the policy for timeout-triggered emissions
        policy.set_timeout_tx(driver_tx.clone());

        // Thread-safe wrapper for the post-continuation
        let sync_post = SyncBoxProcessor::new(post_continuation);

        // ── Spawn actor task ──
        {
            let policy = Arc::clone(&policy);
            let actor_h = Arc::clone(&actor_handle);
            let actor_driver_tx = driver_tx; // move the original sender into the actor
            // Queue-depth sampling: the actor loop is the resequencer's
            // maintenance pass (there is no timer-driven sweep in the
            // service). After each accept, publish the policy's
            // awaiting-sequence buffer size under `resequencer:<route>`.
            let queue_metrics = config.metrics.clone();
            let queue_label = config
                .route_id
                .as_ref()
                .map(|route| format!("resequencer:{route}"));
            let handle = tokio::spawn(async move {
                while let Some(input) = input_rx.recv().await {
                    let ready = policy.accept(input).await;
                    for ex in ready {
                        // If the post-driver channel is closed, stop
                        if actor_driver_tx.send(ex).await.is_err() {
                            // post-driver dropped → exit
                            return;
                        }
                    }
                    if let (Some(metrics), Some(label)) =
                        (queue_metrics.as_ref(), queue_label.as_ref())
                    {
                        metrics.set_queue_depth(label, policy.buffered());
                    }
                }
                // input channel closed (EOF from shutdown) → exit naturally
            });
            *actor_h.lock().expect("actor_handle lock poisoned") = Some(handle); // allow-unwrap: handle slot is None at construction time
        }

        // ── Spawn post-driver task ──
        {
            let post = sync_post.clone();
            let driver_h = Arc::clone(&driver_handle);
            let metrics = config.metrics.clone();
            let route_id = config.route_id.clone();
            let handle = tokio::spawn(async move {
                while let Some(ex) = driver_rx.recv().await {
                    // CamelStop interaction: skip continuation for stop-signaled exchanges
                    if camel_api::is_camel_stop(&ex) {
                        tracing::debug!(
                            "resequencer post-driver: skipping continuation for CamelStop exchange"
                        );
                        continue;
                    }
                    // Clone inner processor (cheap: Arc bump + Mutex briefly held)
                    let mut proc = post.clone_inner();
                    match proc.call(ex).await {
                        Ok(_) => {}
                        Err(e) => {
                            // log-policy: post-ack failure (ADR-0012 best-effort, ADR-0029 I7)
                            tracing::warn!(
                                error = %e,
                                "resequencer post-driver: continuation call failed after ack (best-effort)"
                            );
                            if let Some(ref m) = metrics {
                                m.increment_errors(
                                    route_id.as_deref().unwrap_or("unknown"),
                                    "resequencer:post_ack_failure",
                                );
                            }
                        }
                    }
                }
            });
            *driver_h.lock().expect("driver_handle lock poisoned") = Some(handle); // allow-unwrap: handle slot is None at construction time
        }

        let metrics = config.metrics.clone();
        let route_id = config.route_id.clone();
        Self {
            policy,
            config,
            input_tx: input_tx_shared,
            driver_tx: driver_tx_shared,
            actor_handle,
            driver_handle,
            shutdown_started,
            post_lifecycles: Arc::new(Mutex::new(post_lifecycles)),
            inout_counter: Arc::new(AtomicU64::new(0)),
            last_inout_warn: Arc::new(TokioMutex::new(None)),
            metrics,
            route_id,
        }
    }
}

// ── Tower Service impl ──

impl Service<Exchange> for ResequencerService {
    type Response = Exchange;
    type Error = CamelError;
    type Future =
        std::pin::Pin<Box<dyn std::future::Future<Output = Result<Exchange, CamelError>> + Send>>;

    fn poll_ready(
        &mut self,
        _cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Result<(), CamelError>> {
        // ADR-0019: always ready; backpressure via bounded send().await in call()
        std::task::Poll::Ready(Ok(()))
    }

    fn call(&mut self, input: Exchange) -> Self::Future {
        let config = self.config.clone();
        let inout_counter = Arc::clone(&self.inout_counter);
        let last_inout_warn = Arc::clone(&self.last_inout_warn);
        let tx_opt = Arc::clone(&self.input_tx);
        let metrics = self.metrics.clone();
        let route_id = self.route_id.clone();

        Box::pin(async move {
            // Build ack exchange
            let mut ack = Exchange::new(Message::default());

            // InOut guard (I6): rate-limited with metric counter and property flag
            if input.pattern == camel_api::exchange::ExchangePattern::InOut && !config.allow_inout {
                inout_counter.fetch_add(1, Ordering::Relaxed);
                ack.set_property(CAMEL_RESEQUENCER_INOUT_WARN, true);
                if let Some(ref m) = metrics {
                    m.increment_errors(
                        route_id.as_deref().unwrap_or("unknown"),
                        "resequencer:inout_warning",
                    );
                }
                let now = Instant::now();
                let mut last_guard = last_inout_warn.lock().await;
                let should_warn = last_guard
                    .map(|t| now.duration_since(t) >= INOUT_WARN_INTERVAL)
                    .unwrap_or(true);
                if should_warn {
                    let count = inout_counter.load(Ordering::Relaxed);
                    tracing::warn!(
                        inout_count = count,
                        "InOut exchange reached resequencer ({count} total); \
                         consider using InOnly pattern. \
                         Set allow_inout=true to suppress this warning."
                    );
                    *last_guard = Some(now);
                }
            }

            // Snapshot the sender (if still active). If shutdown already took it,
            // the exchange is dropped — best-effort (intake already cancelled).
            let tx = {
                let guard = tx_opt.lock().unwrap_or_else(|e| e.into_inner());
                guard.clone()
            };
            if let Some(tx) = tx {
                // Backpressure: blocks if channel is full
                match tx.send(input).await {
                    Ok(()) => {
                        ack.set_property(CAMEL_RESEQUENCER_ACCEPTED, true);
                    }
                    Err(tokio::sync::mpsc::error::SendError(input)) => {
                        tracing::warn!(
                            correlation_id = %input.correlation_id,
                            "resequencer input dropped during shutdown"
                        );
                        ack.set_property(CAMEL_RESEQUENCER_ACCEPTED, false);
                        ack.set_property(CAMEL_RESEQUENCER_DROPPED, true);
                    }
                }
            }

            Ok(ack)
        })
    }
}

// ── StepLifecycle impl ──

#[async_trait]
impl StepLifecycle for ResequencerService {
    fn name(&self) -> &'static str {
        self.policy.name()
    }

    /// Idempotent shutdown with this ordering:
    /// 1. Set shutdown flag; close/drop input_tx so actor sees EOF.
    /// 2. Await actor JoinHandle (bounded deadline).
    /// 3. policy.flush() → emit remaining in order via post-driver.
    /// 4. Close post-driver channel sender so its loop sees EOF.
    /// 5. Await post-driver JoinHandle with 5s deadline.
    /// 6. Drain post-step lifecycles (oracle Fix 2).
    async fn shutdown(&self, reason: StepShutdownReason) -> Result<(), CamelError> {
        // TODO(Task 1b): differentiate HotSwap (complete in-flight through old continuation,
        // ADR-0004) from RouteStop (flush + drain). Currently both paths run the same
        // flush-then-close sequence.
        tracing::debug!(
            reason = ?reason,
            policy = self.policy.name(),
            "ResequencerService shutdown via StepLifecycle"
        );

        // Idempotent guard (I1)
        {
            let mut started = self
                .shutdown_started
                .lock()
                .unwrap_or_else(|e| e.into_inner());
            if *started {
                tracing::debug!(
                    "ResequencerService shutdown already started (idempotent); skipping"
                );
                return Ok(());
            }
            *started = true;
        }

        // Step 1: Close/drop input_tx so actor's input_rx sees EOF.
        // Taking the sender from the Option drops it. The shared Arc<Mutex<...>>
        // means all clones see None after this.
        {
            let mut guard = self.input_tx.lock().unwrap_or_else(|e| e.into_inner());
            *guard = None; // drop the sender
        }

        // Step 2: Await actor JoinHandle (bounded deadline).
        // Extract handle first, then await outside the lock to avoid Send issue.
        let actor_handle_to_await = {
            let mut guard = self.actor_handle.lock().unwrap_or_else(|e| e.into_inner());
            guard.take()
        };
        if let Some(handle) = actor_handle_to_await {
            // 5s deadline for actor to finish processing remaining input
            let _ = tokio::time::timeout(Duration::from_secs(5), handle).await;
        }

        // Step 3: policy.flush() → emit remaining in order via post-driver.
        let flushed = self.policy.flush().await;
        if !flushed.is_empty() {
            let dt = {
                let guard = self.driver_tx.lock().unwrap_or_else(|e| e.into_inner());
                guard.clone()
            };
            if let Some(driver_tx) = dt {
                for ex in flushed {
                    if driver_tx.send(ex).await.is_err() {
                        tracing::warn!(
                            "resequencer shutdown flush: post-driver channel closed early"
                        );
                        break;
                    }
                }
            }
        }

        // Step 4: Close the post-driver channel sender so its loop sees EOF.
        {
            let mut guard = self.driver_tx.lock().unwrap_or_else(|e| e.into_inner());
            *guard = None; // drop the sender → driver_rx.recv() returns None
        }

        // Step 5: Await the post-driver JoinHandle with a 5s deadline.
        let driver_handle_to_await = {
            let mut guard = self.driver_handle.lock().unwrap_or_else(|e| e.into_inner());
            guard.take()
        };
        if let Some(handle) = driver_handle_to_await {
            let result = tokio::time::timeout(Duration::from_secs(5), handle).await;
            if result.is_err() {
                tracing::warn!(
                    "resequencer post-driver task did not finish within 5s deadline; \
                     leaking handle (best-effort)"
                );
            }
        }

        // Step 6: Drain post-step lifecycles (oracle Fix 2).
        // Must happen AFTER post-driver drains (flush emits through continuation first).
        {
            let post_lcs: Vec<Arc<dyn StepLifecycle>> = {
                let mut guard = self
                    .post_lifecycles
                    .lock()
                    .unwrap_or_else(|e| e.into_inner());
                std::mem::take(&mut *guard)
            };
            for lc in &post_lcs {
                if let Err(e) = lc.shutdown(reason).await {
                    tracing::warn!(
                        step = lc.name(),
                        error = %e,
                        "resequencer post-step lifecycle shutdown failed (best-effort)"
                    );
                }
            }
        }

        Ok(())
    }
}

// ── Passthrough policy (for testing) ──

/// Emits each input unchanged — used as a baseline policy for testing
/// the continuation-boundary mechanics.
#[derive(Debug)]
pub struct PassthroughPolicy;

#[async_trait]
impl ResequencePolicy for PassthroughPolicy {
    async fn accept(&self, input: Exchange) -> Vec<Exchange> {
        vec![input]
    }

    async fn flush(&self) -> Vec<Exchange> {
        vec![]
    }

    fn name(&self) -> &'static str {
        "passthrough"
    }
}

// ── Tests ──

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

    /// A test continuation that sends received exchanges through an mpsc channel.
    #[derive(Clone)]
    struct CapturePost {
        tx: mpsc::UnboundedSender<Exchange>,
    }

    impl Service<Exchange> for CapturePost {
        type Response = Exchange;
        type Error = CamelError;
        type Future = std::pin::Pin<
            Box<dyn std::future::Future<Output = Result<Exchange, CamelError>> + Send>,
        >;

        fn poll_ready(
            &mut self,
            _cx: &mut std::task::Context<'_>,
        ) -> std::task::Poll<Result<(), CamelError>> {
            std::task::Poll::Ready(Ok(()))
        }

        fn call(&mut self, exchange: Exchange) -> Self::Future {
            let tx = self.tx.clone();
            Box::pin(async move {
                // Fire-and-forget: drop send errors (receiver gone = test teardown)
                let _ = tx.send(exchange.clone());
                Ok(exchange)
            })
        }
    }

    #[tokio::test]
    async fn resequencer_boundary_passthrough_ack_and_continuation() {
        use camel_api::body::Body;

        // Build a resequencer with passthrough policy + channel capture continuation
        let policy: Arc<dyn ResequencePolicy> = Arc::new(PassthroughPolicy);
        let (capture_tx, mut capture_rx) = mpsc::unbounded_channel::<Exchange>();
        let capture = CapturePost { tx: capture_tx };
        let post_continuation: BoxProcessor = BoxProcessor::new(capture);

        let service = ResequencerService::new(policy, post_continuation, 1024, vec![]);

        // Send an exchange
        let mut input = Exchange::new(Message::new(Body::Text("hello".into())));
        input.set_property("seq", 1);

        let ack = service.clone().oneshot(input).await.unwrap();

        // Assert (a): ack has Body::Empty + CAMEL_RESEQUENCER_ACCEPTED=true
        assert!(
            matches!(ack.input.body, Body::Empty),
            "ack body should be Empty, got {:?}",
            ack.input.body
        );
        assert_eq!(
            ack.property(CAMEL_RESEQUENCER_ACCEPTED)
                .and_then(|v| v.as_bool()),
            Some(true),
            "CAMEL_RESEQUENCER_ACCEPTED should be true"
        );

        // Assert (b): the post-continuation receives the input payload
        let captured = tokio::time::timeout(Duration::from_millis(500), capture_rx.recv())
            .await
            .expect("post-continuation did not receive exchange within 500ms timeout")
            .expect("capture channel closed without receiving exchange");
        let body = captured.input.body.as_text();
        assert_eq!(
            body,
            Some("hello"),
            "post-continuation received body should match"
        );

        // Assert (c): shutdown is idempotent (calling twice returns Ok)
        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("first shutdown should succeed");

        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("second shutdown should succeed (idempotent)");
    }

    #[tokio::test]
    async fn resequencer_boundary_camel_stop_skipped() {
        use camel_api::body::Body;

        let policy: Arc<dyn ResequencePolicy> = Arc::new(PassthroughPolicy);
        let (capture_tx, mut capture_rx) = mpsc::unbounded_channel::<Exchange>();
        let capture = CapturePost { tx: capture_tx };
        let post_continuation: BoxProcessor = BoxProcessor::new(capture);

        let service = ResequencerService::new(policy, post_continuation, 1024, vec![]);

        // Send an exchange flagged with CamelStop
        let mut input = Exchange::new(Message::new(Body::Text(
            "should-not-reach-continuation".into(),
        )));
        input.set_property(camel_api::exchange::CAMEL_STOP, true);

        let ack = service.clone().oneshot(input).await.unwrap();

        // Assert: actor accepted the exchange (ack is returned)
        assert_eq!(
            ack.property(CAMEL_RESEQUENCER_ACCEPTED)
                .and_then(|v| v.as_bool()),
            Some(true),
            "CamelStop exchange should still be accepted by resequencer actor"
        );

        // Assert: the post-continuation does NOT receive the CamelStop exchange
        let did_receive = tokio::time::timeout(Duration::from_millis(500), capture_rx.recv()).await;
        match did_receive {
            Ok(Some(_)) => panic!("CamelStop exchange should NOT reach post-continuation"),
            Ok(None) => {}      // channel closed (expected in some teardown scenarios)
            Err(_elapsed) => {} // timeout is the expected path: nothing arrived
        }

        // Cleanup
        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("shutdown should succeed");
    }

    #[tokio::test]
    async fn inout_guard_increments_counter() {
        let policy: Arc<dyn ResequencePolicy> = Arc::new(PassthroughPolicy);
        let (tx, _rx) = mpsc::unbounded_channel::<Exchange>();
        let post: BoxProcessor = BoxProcessor::new(CapturePost { tx });
        let config = ResequencerConfig::default();
        let service = ResequencerService::with_config(policy, post, 16, vec![], config);

        // InOnly should NOT increment counter
        let ex_inonly = Exchange::new(Message::new("inonly"));
        let _ = service.clone().oneshot(ex_inonly).await.unwrap();

        // InOut SHOULD increment counter
        let ex_inout = Exchange::new_in_out(Message::new("inout"));
        let _ = service.clone().oneshot(ex_inout).await.unwrap();
        assert!(
            service.inout_counter.load(Ordering::Relaxed) > 0,
            "InOut counter should be > 0 after InOut exchange"
        );

        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("shutdown");
    }

    #[tokio::test]
    async fn inout_guard_allow_inout_suppresses() {
        let policy: Arc<dyn ResequencePolicy> = Arc::new(PassthroughPolicy);
        let (tx, _rx) = mpsc::unbounded_channel::<Exchange>();
        let post: BoxProcessor = BoxProcessor::new(CapturePost { tx });
        let config = ResequencerConfig {
            allow_inout: true,
            ..Default::default()
        };
        let service = ResequencerService::with_config(policy, post, 16, vec![], config);

        let ex_inout = Exchange::new_in_out(Message::new("inout-allowed"));
        let _ = service.clone().oneshot(ex_inout).await.unwrap();
        assert_eq!(
            service.inout_counter.load(Ordering::Relaxed),
            0,
            "InOut counter should be 0 when allow_inout=true"
        );

        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("shutdown");
    }

    // ── claimfamily (rc-hllkk): buffer residency claims ──
    //
    // The exchanges a route pipeline hands the resequencer carry an
    // `in_flight_claim` sibling (split at the pipeline drain site). These
    // tests prove the claim rides the exchange through every residency
    // path: buffered (held), emitted (released after the continuation
    // processes the exchange), shutdown flush (released), and intake
    // drop during shutdown (released immediately).

    /// Test expression that reads the "seq" property (same shape as the
    /// stream-policy test expr).
    struct SeqExpr;

    #[async_trait]
    impl camel_language_api::Expression for SeqExpr {
        async fn evaluate(
            &self,
            exchange: &Exchange,
        ) -> Result<serde_json::Value, camel_language_api::LanguageError> {
            Ok(exchange.property("seq").cloned().unwrap_or_default())
        }
    }

    fn seq_exchange(seq: u64) -> Exchange {
        let mut ex = Exchange::new(Message::new(format!("msg-{seq}")));
        ex.set_property("seq", serde_json::json!(seq));
        ex
    }

    fn stream_service(capture_tx: mpsc::UnboundedSender<Exchange>) -> ResequencerService {
        let policy: Arc<dyn ResequencePolicy> = stream::StreamPolicy::new_cyclic(
            Arc::new(SeqExpr),
            100,
            600_000, // gap timeout far beyond test duration
            camel_api::resequencer::GapPolicy::EmitPartial,
            camel_api::resequencer::CapacityPolicy::LogAndDrop,
            false,
        );
        let post: BoxProcessor = BoxProcessor::new(CapturePost { tx: capture_tx });
        ResequencerService::new(policy, post, 1024, vec![])
    }

    /// Poll until the counter reaches `want` (bounded, panics past the
    /// deadline) — same discipline as the drainclaim route tests.
    async fn await_in_flight(counter: &Arc<AtomicU64>, want: u64) {
        let deadline = Instant::now() + Duration::from_secs(2);
        while counter.load(Ordering::SeqCst) != want {
            assert!(
                Instant::now() < deadline,
                "in-flight counter did not reach {want} within 2s (now {})",
                counter.load(Ordering::SeqCst)
            );
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
    }

    #[tokio::test]
    async fn claim_held_while_buffered_released_on_completion() {
        let counter = Arc::new(AtomicU64::new(0));
        let (capture_tx, mut capture_rx) = mpsc::unbounded_channel::<Exchange>();
        let service = stream_service(capture_tx);

        // seq 2 arrives first: buffered behind the gap at seq 1. The
        // claim rides the exchange into the policy buffer — the counter
        // stays at 1 even though `call` (the ack) already resolved.
        let mut first = seq_exchange(2);
        first.in_flight_claim = Some(camel_api::InFlightClaim::attach(&counter));
        let ack = service.clone().oneshot(first).await.unwrap();
        assert_eq!(
            ack.property(CAMEL_RESEQUENCER_ACCEPTED)
                .and_then(|v| v.as_bool()),
            Some(true)
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            1,
            "buffered exchange must stay counted after the ack resolves"
        );

        // seq 1 completes the contiguous run: both emissions run the
        // post-continuation, then their claims release.
        let mut second = seq_exchange(1);
        second.in_flight_claim = Some(camel_api::InFlightClaim::attach(&counter));
        let _ = service.clone().oneshot(second).await.unwrap();
        let _ = capture_rx.recv().await;
        let _ = capture_rx.recv().await;
        await_in_flight(&counter, 0).await;

        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("shutdown");
    }

    #[tokio::test]
    async fn claim_released_on_shutdown_flush() {
        let counter = Arc::new(AtomicU64::new(0));
        let (capture_tx, mut capture_rx) = mpsc::unbounded_channel::<Exchange>();
        let service = stream_service(capture_tx);

        // seq 2 buffers (gap at 1); shutdown flush must emit it through
        // the continuation and release the claim.
        let mut buffered = seq_exchange(2);
        buffered.in_flight_claim = Some(camel_api::InFlightClaim::attach(&counter));
        let _ = service.clone().oneshot(buffered).await.unwrap();
        assert_eq!(counter.load(Ordering::SeqCst), 1);

        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("shutdown");
        let _ = capture_rx.recv().await;
        await_in_flight(&counter, 0).await;
    }

    #[tokio::test]
    async fn claim_released_when_input_dropped_after_shutdown() {
        let counter = Arc::new(AtomicU64::new(0));
        let (capture_tx, _capture_rx) = mpsc::unbounded_channel::<Exchange>();
        let service = stream_service(capture_tx);
        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("shutdown first");

        // The input sender slot is gone after shutdown: the exchange is
        // never accepted — dropped inside `call` — and the claim
        // releases immediately.
        let mut dropped = seq_exchange(1);
        dropped.in_flight_claim = Some(camel_api::InFlightClaim::attach(&counter));
        let ack = service.clone().oneshot(dropped).await.unwrap();
        assert!(
            ack.property(CAMEL_RESEQUENCER_ACCEPTED).is_none(),
            "shutdown-intake exchange must not be accepted"
        );
        assert_eq!(
            counter.load(Ordering::SeqCst),
            0,
            "dropped input releases its claim immediately"
        );
    }

    #[tokio::test]
    async fn claim_released_when_capacity_policy_drops_exchange() {
        let counter = Arc::new(AtomicU64::new(0));
        let (capture_tx, _capture_rx) = mpsc::unbounded_channel::<Exchange>();
        // Capacity 1: seq 2 occupies the queue (gap at 1); seq 3 hits
        // the cap and CapacityPolicy::LogAndDrop drops it — the dropped
        // exchange's claim releases, the held one stays counted.
        let policy: Arc<dyn ResequencePolicy> = stream::StreamPolicy::new_cyclic(
            Arc::new(SeqExpr),
            1,
            600_000,
            camel_api::resequencer::GapPolicy::EmitPartial,
            camel_api::resequencer::CapacityPolicy::LogAndDrop,
            false,
        );
        let post: BoxProcessor = BoxProcessor::new(CapturePost { tx: capture_tx });
        let service = ResequencerService::new(policy, post, 1024, vec![]);

        let mut held = seq_exchange(2);
        held.in_flight_claim = Some(camel_api::InFlightClaim::attach(&counter));
        let _ = service.clone().oneshot(held).await.unwrap();
        assert_eq!(counter.load(Ordering::SeqCst), 1);

        let mut overflow = seq_exchange(3);
        overflow.in_flight_claim = Some(camel_api::InFlightClaim::attach(&counter));
        let _ = service.clone().oneshot(overflow).await.unwrap();
        await_in_flight(&counter, 1).await;

        service
            .shutdown(StepShutdownReason::RouteStop)
            .await
            .expect("shutdown flushes the held exchange and releases it");
        await_in_flight(&counter, 0).await;
    }
}