ranvier_runtime/

axon.rs

//! # Axon: Executable Decision Tree
//!
//! The `Axon` is the **runtime execution path** of a Typed Decision Tree.
//! It functions as a reusable Typed Decision flow (Axon<In, Out, E>).
//!
//! ## Design Philosophy
//!
//! * **Axon flows, Schematic shows**: Axon executes; Schematic describes
//! * **Builder pattern**: `Axon::start().then().then()`
//! * **Schematic extraction**: Every Axon carries its structural metadata
//!
//! "Axon is the flowing thing, Schematic is the visible thing."

use crate::persistence::{
    CompensationAutoTrigger, CompensationContext, CompensationHandle,
    CompensationIdempotencyHandle, CompensationRetryPolicy, CompletionState,
    PersistenceAutoComplete, PersistenceEnvelope, PersistenceHandle, PersistenceTraceId,
};
use ranvier_core::bus::Bus;
use ranvier_core::outcome::Outcome;
use ranvier_core::schematic::{
    BusCapabilitySchema, Edge, EdgeType, Node, NodeKind, Schematic, SourceLocation,
};
use ranvier_core::timeline::{Timeline, TimelineEvent};
use ranvier_core::transition::Transition;
use std::any::type_name;
use std::ffi::OsString;
use std::fs;
use std::future::Future;
use std::panic::Location;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{SystemTime, UNIX_EPOCH};
use tracing::Instrument;

/// Type alias for async boxed futures used in Axon execution.
pub type BoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;

/// Executor type for Axon steps.
/// Now takes an input state `In`, a resource bundle `Res`, and returns an `Outcome<Out, E>`.
/// Must be Send + Sync to be reusable across threads and clones.
pub type Executor<In, Out, E, Res> =
    Arc<dyn for<'a> Fn(In, &'a Res, &'a mut Bus) -> BoxFuture<'a, Outcome<Out, E>> + Send + Sync>;

/// Helper to extract a readable type name from a type.
fn type_name_of<T: ?Sized>() -> String {
    let full = type_name::<T>();
    full.split("::").last().unwrap_or(full).to_string()
}

/// The Axon Builder and Runtime.
///
/// `Axon` represents an executable decision tree.
/// It is reusable and thread-safe.
///
/// ## Example
///
/// ```rust,ignore
/// use ranvier_core::prelude::*;
/// // ...
/// // Start with an identity Axon (In -> In)
/// let axon = Axon::<(), (), _>::new("My Axon")
///     .then(StepA)
///     .then(StepB);
///
/// // Execute multiple times
/// let res1 = axon.execute((), &mut bus1).await;
/// let res2 = axon.execute((), &mut bus2).await;
/// ```
pub struct Axon<In, Out, E, Res = ()> {
    /// The static structure (for visualization/analysis)
    pub schematic: Schematic,
    /// The runtime executor
    executor: Executor<In, Out, E, Res>,
}

/// Schematic export request derived from command-line args/env.
#[derive(Debug, Clone)]
pub struct SchematicExportRequest {
    /// Optional output file path. If omitted, schematic is written to stdout.
    pub output: Option<PathBuf>,
}

impl<In, Out, E, Res> Clone for Axon<In, Out, E, Res> {
    fn clone(&self) -> Self {
        Self {
            schematic: self.schematic.clone(),
            executor: self.executor.clone(),
        }
    }
}

impl<In, E, Res> Axon<In, In, E, Res>
where
    In: Send + Sync + 'static,
    E: Send + 'static,
    Res: ranvier_core::transition::ResourceRequirement,
{
    /// Create a new Axon flow with the given label.
    /// This is the preferred entry point per Flat API guidelines.
    #[track_caller]
    pub fn new(label: &str) -> Self {
        let caller = Location::caller();
        Self::start_with_source(label, caller)
    }

    /// Start defining a new Axon flow.
    /// This creates an Identity Axon (In -> In) with no initial resource requirements.
    #[track_caller]
    pub fn start(label: &str) -> Self {
        let caller = Location::caller();
        Self::start_with_source(label, caller)
    }

    fn start_with_source(label: &str, caller: &'static Location<'static>) -> Self {
        let node_id = uuid::Uuid::new_v4().to_string();
        let node = Node {
            id: node_id,
            kind: NodeKind::Ingress,
            label: label.to_string(),
            description: None,
            input_type: "void".to_string(),
            output_type: type_name_of::<In>(),
            resource_type: type_name_of::<Res>(),
            metadata: Default::default(),
            bus_capability: None,
            source_location: Some(SourceLocation::new(caller.file(), caller.line())),
        };

        let mut schematic = Schematic::new(label);
        schematic.nodes.push(node);

        let executor: Executor<In, In, E, Res> =
            Arc::new(move |input, _res, _bus| Box::pin(std::future::ready(Outcome::Next(input))));

        Self {
            schematic,
            executor,
        }
    }
}

impl<In, Out, E, Res> Axon<In, Out, E, Res>
where
    In: Send + Sync + 'static,
    Out: Send + Sync + 'static,
    E: Send + 'static,
    Res: ranvier_core::transition::ResourceRequirement,
{
    /// Chain a transition to this Axon.
    ///
    /// Requires the transition to use the SAME resource bundle as the previous steps.
    #[track_caller]
    pub fn then<Next, Trans>(self, transition: Trans) -> Axon<In, Next, E, Res>
    where
        Next: Send + Sync + 'static,
        Trans: Transition<Out, Next, Resources = Res, Error = E> + Clone + Send + Sync + 'static,
    {
        let caller = Location::caller();
        // Decompose self to avoid partial move issues
        let Axon {
            mut schematic,
            executor: prev_executor,
        } = self;

        // Update Schematic
        let next_node_id = uuid::Uuid::new_v4().to_string();
        let next_node = Node {
            id: next_node_id.clone(),
            kind: NodeKind::Atom,
            label: transition.label(),
            description: transition.description(),
            input_type: type_name_of::<Out>(),
            output_type: type_name_of::<Next>(),
            resource_type: type_name_of::<Res>(),
            metadata: Default::default(),
            bus_capability: bus_capability_schema_from_policy(transition.bus_access_policy()),
            source_location: Some(SourceLocation::new(caller.file(), caller.line())),
        };

        let last_node_id = schematic
            .nodes
            .last()
            .map(|n| n.id.clone())
            .unwrap_or_default();

        schematic.nodes.push(next_node);
        schematic.edges.push(Edge {
            from: last_node_id,
            to: next_node_id.clone(),
            kind: EdgeType::Linear,
            label: Some("Next".to_string()),
        });

        // Compose Executor
        let node_id_for_exec = next_node_id.clone();
        let node_label_for_exec = transition.label();
        let bus_policy_for_exec = transition.bus_access_policy();
        let next_executor: Executor<In, Next, E, Res> = Arc::new(
            move |input: In, res: &Res, bus: &mut Bus| -> BoxFuture<'_, Outcome<Next, E>> {
                let prev = prev_executor.clone();
                let trans = transition.clone();
                let timeline_node_id = node_id_for_exec.clone();
                let timeline_node_label = node_label_for_exec.clone();
                let transition_bus_policy = bus_policy_for_exec.clone();

                Box::pin(async move {
                    // Run previous step
                    let prev_result = prev(input, res, bus).await;

                    // Unpack
                    let state = match prev_result {
                        Outcome::Next(t) => t,
                        other => return other.map(|_| unreachable!()),
                    };

                    // Run this step with automatic instrumentation
                    let label = trans.label();
                    let res_type = std::any::type_name::<Res>()
                        .split("::")
                        .last()
                        .unwrap_or("unknown");

                    let enter_ts = now_ms();
                    if let Some(timeline) = bus.read_mut::<Timeline>() {
                        timeline.push(TimelineEvent::NodeEnter {
                            node_id: timeline_node_id.clone(),
                            node_label: timeline_node_label.clone(),
                            timestamp: enter_ts,
                        });
                    }

                    let node_span = tracing::info_span!(
                        "Node",
                        ranvier.node = %label,
                        ranvier.resource_type = %res_type,
                        ranvier.outcome_kind = tracing::field::Empty,
                        ranvier.outcome_target = tracing::field::Empty
                    );
                    let started = std::time::Instant::now();
                    bus.set_access_policy(label.clone(), transition_bus_policy.clone());
                    let result = trans
                        .run(state, res, bus)
                        .instrument(node_span.clone())
                        .await;
                    bus.clear_access_policy();
                    node_span.record("ranvier.outcome_kind", outcome_kind_name(&result));
                    if let Some(target) = outcome_target(&result) {
                        node_span
                            .record("ranvier.outcome_target", tracing::field::display(&target));
                    }
                    let duration_ms = started.elapsed().as_millis() as u64;
                    let exit_ts = now_ms();

                    if let Some(timeline) = bus.read_mut::<Timeline>() {
                        timeline.push(TimelineEvent::NodeExit {
                            node_id: timeline_node_id.clone(),
                            outcome_type: outcome_type_name(&result),
                            duration_ms,
                            timestamp: exit_ts,
                        });

                        if let Outcome::Branch(branch_id, _) = &result {
                            timeline.push(TimelineEvent::Branchtaken {
                                branch_id: branch_id.clone(),
                                timestamp: exit_ts,
                            });
                        }
                    }

                    result
                })
            },
        );

        Axon {
            schematic,
            executor: next_executor,
        }
    }

    /// Add a branch point
    #[track_caller]
    pub fn branch(mut self, branch_id: impl Into<String>, label: &str) -> Self {
        let caller = Location::caller();
        let branch_id_str = branch_id.into();
        let last_node_id = self
            .schematic
            .nodes
            .last()
            .map(|n| n.id.clone())
            .unwrap_or_default();

        let branch_node = Node {
            id: uuid::Uuid::new_v4().to_string(),
            kind: NodeKind::Synapse,
            label: label.to_string(),
            description: None,
            input_type: type_name_of::<Out>(),
            output_type: type_name_of::<Out>(),
            resource_type: type_name_of::<Res>(),
            metadata: Default::default(),
            bus_capability: None,
            source_location: Some(SourceLocation::new(caller.file(), caller.line())),
        };

        self.schematic.nodes.push(branch_node);
        self.schematic.edges.push(Edge {
            from: last_node_id,
            to: branch_id_str.clone(),
            kind: EdgeType::Branch(branch_id_str),
            label: Some("Branch".to_string()),
        });

        self
    }

    /// Execute the Axon with the given input and resources.
    pub async fn execute(&self, input: In, resources: &Res, bus: &mut Bus) -> Outcome<Out, E> {
        let trace_id = persistence_trace_id(bus);
        let label = self.schematic.name.clone();
        let persistence_handle = bus.read::<PersistenceHandle>().cloned();
        let compensation_handle = bus.read::<CompensationHandle>().cloned();
        let compensation_retry_policy = compensation_retry_policy(bus);
        let compensation_idempotency = bus.read::<CompensationIdempotencyHandle>().cloned();
        let persistence_start_step = if let Some(handle) = persistence_handle.as_ref() {
            let start_step = next_persistence_step(handle, &trace_id).await;
            persist_execution_event(handle, &trace_id, &label, start_step, "Enter").await;
            Some(start_step)
        } else {
            None
        };

        let should_capture = should_attach_timeline(bus);
        let inserted_timeline = if should_capture {
            ensure_timeline(bus)
        } else {
            false
        };
        let ingress_started = std::time::Instant::now();
        let ingress_enter_ts = now_ms();
        if should_capture
            && let (Some(timeline), Some(ingress)) =
                (bus.read_mut::<Timeline>(), self.schematic.nodes.first())
        {
            timeline.push(TimelineEvent::NodeEnter {
                node_id: ingress.id.clone(),
                node_label: ingress.label.clone(),
                timestamp: ingress_enter_ts,
            });
        }

        let circuit_span = tracing::info_span!(
            "Circuit",
            ranvier.circuit = %label,
            ranvier.outcome_kind = tracing::field::Empty,
            ranvier.outcome_target = tracing::field::Empty
        );
        let outcome = (self.executor)(input, resources, bus)
            .instrument(circuit_span.clone())
            .await;
        circuit_span.record("ranvier.outcome_kind", outcome_kind_name(&outcome));
        if let Some(target) = outcome_target(&outcome) {
            circuit_span.record("ranvier.outcome_target", tracing::field::display(&target));
        }

        let ingress_exit_ts = now_ms();
        if should_capture
            && let (Some(timeline), Some(ingress)) =
                (bus.read_mut::<Timeline>(), self.schematic.nodes.first())
        {
            timeline.push(TimelineEvent::NodeExit {
                node_id: ingress.id.clone(),
                outcome_type: outcome_type_name(&outcome),
                duration_ms: ingress_started.elapsed().as_millis() as u64,
                timestamp: ingress_exit_ts,
            });
        }

        if let Some(handle) = persistence_handle.as_ref() {
            let fault_step = persistence_start_step.map(|s| s + 1).unwrap_or(1);
            persist_execution_event(
                handle,
                &trace_id,
                &label,
                fault_step,
                outcome_kind_name(&outcome),
            )
            .await;

            let mut completion = completion_from_outcome(&outcome);
            if matches!(outcome, Outcome::Fault(_))
                && let Some(compensation) = compensation_handle.as_ref()
                && compensation_auto_trigger(bus)
            {
                let context = CompensationContext {
                    trace_id: trace_id.clone(),
                    circuit: label.clone(),
                    fault_kind: outcome_kind_name(&outcome).to_string(),
                    fault_step,
                    timestamp_ms: now_ms(),
                };

                if run_compensation(
                    compensation,
                    context,
                    compensation_retry_policy,
                    compensation_idempotency.clone(),
                )
                .await
                {
                    persist_execution_event(
                        handle,
                        &trace_id,
                        &label,
                        fault_step.saturating_add(1),
                        "Compensated",
                    )
                    .await;
                    completion = CompletionState::Compensated;
                }
            }

            if persistence_auto_complete(bus) {
                persist_completion(handle, &trace_id, completion).await;
            }
        }

        if should_capture {
            maybe_export_timeline(bus, &outcome);
        }
        if inserted_timeline {
            let _ = bus.remove::<Timeline>();
        }

        outcome
    }

    /// Starts the Ranvier Inspector for this Axon on the specified port.
    /// This spawns a background task to serve the Schematic.
    pub fn serve_inspector(self, port: u16) -> Self {
        if !inspector_dev_mode_from_env() {
            tracing::info!("Inspector disabled because RANVIER_MODE is production");
            return self;
        }
        if !inspector_enabled_from_env() {
            tracing::info!("Inspector disabled by RANVIER_INSPECTOR");
            return self;
        }

        let schematic = self.schematic.clone();
        tokio::spawn(async move {
            if let Err(e) = ranvier_inspector::Inspector::new(schematic, port)
                .with_projection_files_from_env()
                .with_mode_from_env()
                .with_auth_policy_from_env()
                .serve()
                .await
            {
                tracing::error!("Inspector server failed: {}", e);
            }
        });
        self
    }

    /// Get a reference to the Schematic (structural view).
    pub fn schematic(&self) -> &Schematic {
        &self.schematic
    }

    /// Consume and return the Schematic.
    pub fn into_schematic(self) -> Schematic {
        self.schematic
    }

    /// Detect schematic export mode from runtime flags.
    ///
    /// Supported triggers:
    /// - `RANVIER_SCHEMATIC=1|true|on|yes`
    /// - `--schematic`
    ///
    /// Optional output path:
    /// - `RANVIER_SCHEMATIC_OUTPUT=<path>`
    /// - `--schematic-output <path>` / `--schematic-output=<path>`
    /// - `--output <path>` / `--output=<path>` (only relevant in schematic mode)
    pub fn schematic_export_request(&self) -> Option<SchematicExportRequest> {
        schematic_export_request_from_process()
    }

    /// Export schematic and return `true` when schematic mode is active.
    ///
    /// Use this once after circuit construction and before server/custom loops:
    ///
    /// ```rust,ignore
    /// let axon = build_axon();
    /// if axon.maybe_export_and_exit()? {
    ///     return Ok(());
    /// }
    /// // Normal runtime path...
    /// ```
    pub fn maybe_export_and_exit(&self) -> anyhow::Result<bool> {
        self.maybe_export_and_exit_with(|_| ())
    }

    /// Same as [`Self::maybe_export_and_exit`] but allows a custom hook right before export/exit.
    ///
    /// This is useful when your app has custom loop/bootstrap behavior and you want
    /// to skip or cleanup that logic in schematic mode.
    pub fn maybe_export_and_exit_with<F>(&self, on_before_exit: F) -> anyhow::Result<bool>
    where
        F: FnOnce(&SchematicExportRequest),
    {
        let Some(request) = self.schematic_export_request() else {
            return Ok(false);
        };
        on_before_exit(&request);
        self.export_schematic(&request)?;
        Ok(true)
    }

    /// Export schematic according to the provided request.
    pub fn export_schematic(&self, request: &SchematicExportRequest) -> anyhow::Result<()> {
        let json = serde_json::to_string_pretty(self.schematic())?;
        if let Some(path) = &request.output {
            if let Some(parent) = path.parent() {
                if !parent.as_os_str().is_empty() {
                    fs::create_dir_all(parent)?;
                }
            }
            fs::write(path, json.as_bytes())?;
            return Ok(());
        }
        println!("{}", json);
        Ok(())
    }
}

fn schematic_export_request_from_process() -> Option<SchematicExportRequest> {
    let args: Vec<OsString> = std::env::args_os().skip(1).collect();
    let mut enabled = env_flag_is_true("RANVIER_SCHEMATIC");
    let mut output = std::env::var_os("RANVIER_SCHEMATIC_OUTPUT").map(PathBuf::from);

    let mut i = 0;
    while i < args.len() {
        let arg = args[i].to_string_lossy();

        if arg == "--schematic" {
            enabled = true;
            i += 1;
            continue;
        }

        if arg == "--schematic-output" || arg == "--output" {
            if let Some(next) = args.get(i + 1) {
                output = Some(PathBuf::from(next));
                i += 2;
                continue;
            }
        } else if let Some(value) = arg.strip_prefix("--schematic-output=") {
            output = Some(PathBuf::from(value));
        } else if let Some(value) = arg.strip_prefix("--output=") {
            output = Some(PathBuf::from(value));
        }

        i += 1;
    }

    if enabled {
        Some(SchematicExportRequest { output })
    } else {
        None
    }
}

fn env_flag_is_true(key: &str) -> bool {
    match std::env::var(key) {
        Ok(v) => matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "on" | "yes"),
        Err(_) => false,
    }
}

fn inspector_enabled_from_env() -> bool {
    let raw = std::env::var("RANVIER_INSPECTOR").ok();
    inspector_enabled_from_value(raw.as_deref())
}

fn inspector_enabled_from_value(value: Option<&str>) -> bool {
    match value {
        Some(v) => matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "on" | "yes"),
        None => true,
    }
}

fn inspector_dev_mode_from_env() -> bool {
    let raw = std::env::var("RANVIER_MODE").ok();
    inspector_dev_mode_from_value(raw.as_deref())
}

fn inspector_dev_mode_from_value(value: Option<&str>) -> bool {
    !matches!(
        value.map(|v| v.to_ascii_lowercase()),
        Some(mode) if mode == "prod" || mode == "production"
    )
}

fn maybe_export_timeline<Out, E>(bus: &mut Bus, outcome: &Outcome<Out, E>) {
    let path = match std::env::var("RANVIER_TIMELINE_OUTPUT") {
        Ok(v) if !v.trim().is_empty() => v,
        _ => return,
    };

    let sampled = sampled_by_bus_id(bus.id, timeline_sample_rate());
    let policy = timeline_adaptive_policy();
    let forced = should_force_export(outcome, &policy);
    let should_export = sampled || forced;
    if !should_export {
        record_sampling_stats(false, sampled, forced, "none", &policy);
        return;
    }

    let mut timeline = bus.read::<Timeline>().cloned().unwrap_or_default();
    timeline.sort();

    let mode = std::env::var("RANVIER_TIMELINE_MODE")
        .unwrap_or_else(|_| "overwrite".to_string())
        .to_ascii_lowercase();

    if let Err(err) = write_timeline_with_policy(&path, &mode, timeline) {
        tracing::warn!(
            "Failed to persist timeline file {} (mode={}): {}",
            path,
            mode,
            err
        );
        record_sampling_stats(false, sampled, forced, &mode, &policy);
    } else {
        record_sampling_stats(true, sampled, forced, &mode, &policy);
    }
}

fn outcome_type_name<Out, E>(outcome: &Outcome<Out, E>) -> String {
    match outcome {
        Outcome::Next(_) => "Next".to_string(),
        Outcome::Branch(id, _) => format!("Branch:{}", id),
        Outcome::Jump(id, _) => format!("Jump:{}", id),
        Outcome::Emit(event_type, _) => format!("Emit:{}", event_type),
        Outcome::Fault(_) => "Fault".to_string(),
    }
}

fn outcome_kind_name<Out, E>(outcome: &Outcome<Out, E>) -> &'static str {
    match outcome {
        Outcome::Next(_) => "Next",
        Outcome::Branch(_, _) => "Branch",
        Outcome::Jump(_, _) => "Jump",
        Outcome::Emit(_, _) => "Emit",
        Outcome::Fault(_) => "Fault",
    }
}

fn outcome_target<Out, E>(outcome: &Outcome<Out, E>) -> Option<String> {
    match outcome {
        Outcome::Branch(branch_id, _) => Some(branch_id.clone()),
        Outcome::Jump(node_id, _) => Some(node_id.to_string()),
        Outcome::Emit(event_type, _) => Some(event_type.clone()),
        Outcome::Next(_) | Outcome::Fault(_) => None,
    }
}

fn completion_from_outcome<Out, E>(outcome: &Outcome<Out, E>) -> CompletionState {
    match outcome {
        Outcome::Fault(_) => CompletionState::Fault,
        _ => CompletionState::Success,
    }
}

fn persistence_trace_id(bus: &Bus) -> String {
    if let Some(explicit) = bus.read::<PersistenceTraceId>() {
        explicit.0.clone()
    } else {
        format!("{}:{}", bus.id, now_ms())
    }
}

fn persistence_auto_complete(bus: &Bus) -> bool {
    bus.read::<PersistenceAutoComplete>()
        .map(|v| v.0)
        .unwrap_or(true)
}

fn compensation_auto_trigger(bus: &Bus) -> bool {
    bus.read::<CompensationAutoTrigger>()
        .map(|v| v.0)
        .unwrap_or(true)
}

fn compensation_retry_policy(bus: &Bus) -> CompensationRetryPolicy {
    bus.read::<CompensationRetryPolicy>()
        .copied()
        .unwrap_or_default()
}

async fn next_persistence_step(handle: &PersistenceHandle, trace_id: &str) -> u64 {
    let store = handle.store();
    match store.load(trace_id).await {
        Ok(Some(trace)) => trace
            .events
            .last()
            .map(|event| event.step.saturating_add(1))
            .unwrap_or(0),
        Ok(None) => 0,
        Err(err) => {
            tracing::warn!(
                trace_id = %trace_id,
                error = %err,
                "Failed to load persistence trace; falling back to step=0"
            );
            0
        }
    }
}

async fn persist_execution_event(
    handle: &PersistenceHandle,
    trace_id: &str,
    circuit: &str,
    step: u64,
    outcome_kind: &str,
) {
    let store = handle.store();
    let envelope = PersistenceEnvelope {
        trace_id: trace_id.to_string(),
        circuit: circuit.to_string(),
        step,
        outcome_kind: outcome_kind.to_string(),
        timestamp_ms: now_ms(),
        payload_hash: None,
    };

    if let Err(err) = store.append(envelope).await {
        tracing::warn!(
            trace_id = %trace_id,
            circuit = %circuit,
            step,
            outcome_kind = %outcome_kind,
            error = %err,
            "Failed to append persistence envelope"
        );
    }
}

async fn persist_completion(
    handle: &PersistenceHandle,
    trace_id: &str,
    completion: CompletionState,
) {
    let store = handle.store();
    if let Err(err) = store.complete(trace_id, completion).await {
        tracing::warn!(
            trace_id = %trace_id,
            error = %err,
            "Failed to complete persistence trace"
        );
    }
}

fn compensation_idempotency_key(context: &CompensationContext) -> String {
    format!(
        "{}:{}:{}",
        context.trace_id, context.circuit, context.fault_kind
    )
}

async fn run_compensation(
    handle: &CompensationHandle,
    context: CompensationContext,
    retry_policy: CompensationRetryPolicy,
    idempotency: Option<CompensationIdempotencyHandle>,
) -> bool {
    let hook = handle.hook();
    let key = compensation_idempotency_key(&context);

    if let Some(store_handle) = idempotency.as_ref() {
        let store = store_handle.store();
        match store.was_compensated(&key).await {
            Ok(true) => {
                tracing::info!(
                    trace_id = %context.trace_id,
                    circuit = %context.circuit,
                    key = %key,
                    "Compensation already recorded; skipping duplicate hook execution"
                );
                return true;
            }
            Ok(false) => {}
            Err(err) => {
                tracing::warn!(
                    trace_id = %context.trace_id,
                    key = %key,
                    error = %err,
                    "Failed to check compensation idempotency state"
                );
            }
        }
    }

    let max_attempts = retry_policy.max_attempts.max(1);
    for attempt in 1..=max_attempts {
        match hook.compensate(context.clone()).await {
            Ok(()) => {
                if let Some(store_handle) = idempotency.as_ref() {
                    let store = store_handle.store();
                    if let Err(err) = store.mark_compensated(&key).await {
                        tracing::warn!(
                            trace_id = %context.trace_id,
                            key = %key,
                            error = %err,
                            "Failed to mark compensation idempotency state"
                        );
                    }
                }
                return true;
            }
            Err(err) => {
                let is_last = attempt == max_attempts;
                tracing::warn!(
                    trace_id = %context.trace_id,
                    circuit = %context.circuit,
                    fault_kind = %context.fault_kind,
                    fault_step = context.fault_step,
                    attempt,
                    max_attempts,
                    error = %err,
                    "Compensation hook attempt failed"
                );
                if !is_last && retry_policy.backoff_ms > 0 {
                    tokio::time::sleep(tokio::time::Duration::from_millis(retry_policy.backoff_ms))
                        .await;
                }
            }
        }
    }
    false
}

fn ensure_timeline(bus: &mut Bus) -> bool {
    if bus.has::<Timeline>() {
        false
    } else {
        bus.insert(Timeline::new());
        true
    }
}

fn should_attach_timeline(bus: &Bus) -> bool {
    // Respect explicitly provided timeline collector from caller.
    if bus.has::<Timeline>() {
        return true;
    }

    // Attach timeline when runtime export path exists.
    has_timeline_output_path()
}

fn has_timeline_output_path() -> bool {
    std::env::var("RANVIER_TIMELINE_OUTPUT")
        .ok()
        .map(|v| !v.trim().is_empty())
        .unwrap_or(false)
}

fn timeline_sample_rate() -> f64 {
    std::env::var("RANVIER_TIMELINE_SAMPLE_RATE")
        .ok()
        .and_then(|v| v.parse::<f64>().ok())
        .map(|v| v.clamp(0.0, 1.0))
        .unwrap_or(1.0)
}

fn sampled_by_bus_id(bus_id: uuid::Uuid, rate: f64) -> bool {
    if rate <= 0.0 {
        return false;
    }
    if rate >= 1.0 {
        return true;
    }
    let bucket = (bus_id.as_u128() % 10_000) as f64 / 10_000.0;
    bucket < rate
}

fn timeline_adaptive_policy() -> String {
    std::env::var("RANVIER_TIMELINE_ADAPTIVE")
        .unwrap_or_else(|_| "fault_branch".to_string())
        .to_ascii_lowercase()
}

fn should_force_export<Out, E>(outcome: &Outcome<Out, E>, policy: &str) -> bool {
    match policy {
        "off" => false,
        "fault_only" => matches!(outcome, Outcome::Fault(_)),
        "fault_branch_emit" => {
            matches!(
                outcome,
                Outcome::Fault(_) | Outcome::Branch(_, _) | Outcome::Emit(_, _)
            )
        }
        _ => matches!(outcome, Outcome::Fault(_) | Outcome::Branch(_, _)),
    }
}

#[derive(Default, Clone)]
struct SamplingStats {
    total_decisions: u64,
    exported: u64,
    skipped: u64,
    sampled_exports: u64,
    forced_exports: u64,
    last_mode: String,
    last_policy: String,
    last_updated_ms: u64,
}

static TIMELINE_SAMPLING_STATS: OnceLock<Mutex<SamplingStats>> = OnceLock::new();

fn stats_cell() -> &'static Mutex<SamplingStats> {
    TIMELINE_SAMPLING_STATS.get_or_init(|| Mutex::new(SamplingStats::default()))
}

fn record_sampling_stats(exported: bool, sampled: bool, forced: bool, mode: &str, policy: &str) {
    let snapshot = {
        let mut stats = match stats_cell().lock() {
            Ok(guard) => guard,
            Err(_) => return,
        };

        stats.total_decisions += 1;
        if exported {
            stats.exported += 1;
        } else {
            stats.skipped += 1;
        }
        if sampled && exported {
            stats.sampled_exports += 1;
        }
        if forced && exported {
            stats.forced_exports += 1;
        }
        stats.last_mode = mode.to_string();
        stats.last_policy = policy.to_string();
        stats.last_updated_ms = now_ms();
        stats.clone()
    };

    tracing::debug!(
        ranvier.timeline.total_decisions = snapshot.total_decisions,
        ranvier.timeline.exported = snapshot.exported,
        ranvier.timeline.skipped = snapshot.skipped,
        ranvier.timeline.sampled_exports = snapshot.sampled_exports,
        ranvier.timeline.forced_exports = snapshot.forced_exports,
        ranvier.timeline.mode = %snapshot.last_mode,
        ranvier.timeline.policy = %snapshot.last_policy,
        "Timeline sampling stats updated"
    );

    if let Some(path) = timeline_stats_output_path() {
        let payload = serde_json::json!({
            "total_decisions": snapshot.total_decisions,
            "exported": snapshot.exported,
            "skipped": snapshot.skipped,
            "sampled_exports": snapshot.sampled_exports,
            "forced_exports": snapshot.forced_exports,
            "last_mode": snapshot.last_mode,
            "last_policy": snapshot.last_policy,
            "last_updated_ms": snapshot.last_updated_ms
        });
        if let Some(parent) = Path::new(&path).parent() {
            let _ = fs::create_dir_all(parent);
        }
        if let Err(err) = fs::write(&path, payload.to_string()) {
            tracing::warn!("Failed to write timeline sampling stats {}: {}", path, err);
        }
    }
}

fn timeline_stats_output_path() -> Option<String> {
    std::env::var("RANVIER_TIMELINE_STATS_OUTPUT")
        .ok()
        .filter(|v| !v.trim().is_empty())
}

fn write_timeline_with_policy(
    path: &str,
    mode: &str,
    mut timeline: Timeline,
) -> Result<(), String> {
    match mode {
        "append" => {
            if Path::new(path).exists() {
                let content = fs::read_to_string(path).map_err(|e| e.to_string())?;
                match serde_json::from_str::<Timeline>(&content) {
                    Ok(mut existing) => {
                        existing.events.append(&mut timeline.events);
                        existing.sort();
                        if let Some(max_events) = max_events_limit() {
                            truncate_timeline_events(&mut existing, max_events);
                        }
                        write_timeline_json(path, &existing)
                    }
                    Err(_) => {
                        // Fallback: if existing is invalid, replace with current timeline
                        if let Some(max_events) = max_events_limit() {
                            truncate_timeline_events(&mut timeline, max_events);
                        }
                        write_timeline_json(path, &timeline)
                    }
                }
            } else {
                if let Some(max_events) = max_events_limit() {
                    truncate_timeline_events(&mut timeline, max_events);
                }
                write_timeline_json(path, &timeline)
            }
        }
        "rotate" => {
            let rotated_path = rotated_path(path, now_ms());
            write_timeline_json(rotated_path.to_string_lossy().as_ref(), &timeline)?;
            if let Some(keep) = rotate_keep_limit() {
                cleanup_rotated_files(path, keep)?;
            }
            Ok(())
        }
        _ => write_timeline_json(path, &timeline),
    }
}

fn write_timeline_json(path: &str, timeline: &Timeline) -> Result<(), String> {
    if let Some(parent) = Path::new(path).parent() {
        if !parent.as_os_str().is_empty() {
            fs::create_dir_all(parent).map_err(|e| e.to_string())?;
        }
    }
    let json = serde_json::to_string_pretty(timeline).map_err(|e| e.to_string())?;
    fs::write(path, json).map_err(|e| e.to_string())
}

fn rotated_path(path: &str, suffix: u64) -> PathBuf {
    let p = Path::new(path);
    let parent = p.parent().unwrap_or_else(|| Path::new(""));
    let stem = p.file_stem().and_then(|s| s.to_str()).unwrap_or("timeline");
    let ext = p.extension().and_then(|e| e.to_str()).unwrap_or("json");
    parent.join(format!("{}_{}.{}", stem, suffix, ext))
}

fn max_events_limit() -> Option<usize> {
    std::env::var("RANVIER_TIMELINE_MAX_EVENTS")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|v| *v > 0)
}

fn rotate_keep_limit() -> Option<usize> {
    std::env::var("RANVIER_TIMELINE_ROTATE_KEEP")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|v| *v > 0)
}

fn truncate_timeline_events(timeline: &mut Timeline, max_events: usize) {
    let len = timeline.events.len();
    if len > max_events {
        let keep_from = len - max_events;
        timeline.events = timeline.events.split_off(keep_from);
    }
}

fn cleanup_rotated_files(base_path: &str, keep: usize) -> Result<(), String> {
    let p = Path::new(base_path);
    let parent = p.parent().unwrap_or_else(|| Path::new("."));
    let stem = p.file_stem().and_then(|s| s.to_str()).unwrap_or("timeline");
    let ext = p.extension().and_then(|e| e.to_str()).unwrap_or("json");
    let prefix = format!("{}_", stem);
    let suffix = format!(".{}", ext);

    let mut files = fs::read_dir(parent)
        .map_err(|e| e.to_string())?
        .filter_map(|entry| entry.ok())
        .filter(|entry| {
            let name = entry.file_name();
            let name = name.to_string_lossy();
            name.starts_with(&prefix) && name.ends_with(&suffix)
        })
        .filter_map(|entry| {
            let modified = entry
                .metadata()
                .ok()
                .and_then(|m| m.modified().ok())
                .unwrap_or(SystemTime::UNIX_EPOCH);
            Some((entry.path(), modified))
        })
        .collect::<Vec<_>>();

    files.sort_by(|a, b| b.1.cmp(&a.1));
    for (path, _) in files.into_iter().skip(keep) {
        let _ = fs::remove_file(path);
    }
    Ok(())
}

fn bus_capability_schema_from_policy(
    policy: Option<ranvier_core::bus::BusAccessPolicy>,
) -> Option<BusCapabilitySchema> {
    let Some(policy) = policy else {
        return None;
    };

    let mut allow = policy
        .allow
        .unwrap_or_default()
        .into_iter()
        .map(|entry| entry.type_name.to_string())
        .collect::<Vec<_>>();
    let mut deny = policy
        .deny
        .into_iter()
        .map(|entry| entry.type_name.to_string())
        .collect::<Vec<_>>();
    allow.sort();
    deny.sort();

    if allow.is_empty() && deny.is_empty() {
        return None;
    }

    Some(BusCapabilitySchema { allow, deny })
}

fn now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis() as u64)
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::{
        Axon, inspector_dev_mode_from_value, inspector_enabled_from_value, sampled_by_bus_id,
        should_force_export,
    };
    use crate::persistence::{
        CompensationContext, CompensationHandle, CompensationHook, CompensationIdempotencyHandle,
        CompensationIdempotencyStore, CompensationRetryPolicy, CompletionState,
        InMemoryCompensationIdempotencyStore, InMemoryPersistenceStore, PersistenceAutoComplete,
        PersistenceHandle, PersistenceStore, PersistenceTraceId,
    };
    use anyhow::Result;
    use async_trait::async_trait;
    use ranvier_core::{Bus, BusAccessPolicy, BusTypeRef, Outcome, Transition};
    use std::sync::Arc;
    use tokio::sync::Mutex;
    use uuid::Uuid;

    #[test]
    fn inspector_enabled_flag_matrix() {
        assert!(inspector_enabled_from_value(None));
        assert!(inspector_enabled_from_value(Some("1")));
        assert!(inspector_enabled_from_value(Some("true")));
        assert!(inspector_enabled_from_value(Some("on")));
        assert!(!inspector_enabled_from_value(Some("0")));
        assert!(!inspector_enabled_from_value(Some("false")));
    }

    #[test]
    fn inspector_dev_mode_matrix() {
        assert!(inspector_dev_mode_from_value(None));
        assert!(inspector_dev_mode_from_value(Some("dev")));
        assert!(inspector_dev_mode_from_value(Some("staging")));
        assert!(!inspector_dev_mode_from_value(Some("prod")));
        assert!(!inspector_dev_mode_from_value(Some("production")));
    }

    #[test]
    fn adaptive_policy_force_export_matrix() {
        let next = Outcome::<(), &'static str>::Next(());
        let branch = Outcome::<(), &'static str>::Branch("declined".to_string(), None);
        let emit = Outcome::<(), &'static str>::Emit("audit".to_string(), None);
        let fault = Outcome::<(), &'static str>::Fault("boom");

        assert!(!should_force_export(&next, "off"));
        assert!(!should_force_export(&fault, "off"));

        assert!(!should_force_export(&branch, "fault_only"));
        assert!(should_force_export(&fault, "fault_only"));

        assert!(should_force_export(&branch, "fault_branch"));
        assert!(!should_force_export(&emit, "fault_branch"));
        assert!(should_force_export(&fault, "fault_branch"));

        assert!(should_force_export(&branch, "fault_branch_emit"));
        assert!(should_force_export(&emit, "fault_branch_emit"));
        assert!(should_force_export(&fault, "fault_branch_emit"));
    }

    #[test]
    fn sampling_and_adaptive_combination_decisions() {
        let bus_id = Uuid::nil();
        let next = Outcome::<(), &'static str>::Next(());
        let fault = Outcome::<(), &'static str>::Fault("boom");

        let sampled_never = sampled_by_bus_id(bus_id, 0.0);
        assert!(!sampled_never);
        assert!(!(sampled_never || should_force_export(&next, "off")));
        assert!(sampled_never || should_force_export(&fault, "fault_only"));

        let sampled_always = sampled_by_bus_id(bus_id, 1.0);
        assert!(sampled_always);
        assert!(sampled_always || should_force_export(&next, "off"));
        assert!(sampled_always || should_force_export(&fault, "off"));
    }

    #[derive(Clone)]
    struct AddOne;

    #[async_trait]
    impl Transition<i32, i32> for AddOne {
        type Error = std::convert::Infallible;
        type Resources = ();

        async fn run(
            &self,
            state: i32,
            _resources: &Self::Resources,
            _bus: &mut Bus,
        ) -> Outcome<i32, Self::Error> {
            Outcome::Next(state + 1)
        }
    }

    #[derive(Clone)]
    struct AlwaysFault;

    #[async_trait]
    impl Transition<i32, i32> for AlwaysFault {
        type Error = &'static str;
        type Resources = ();

        async fn run(
            &self,
            _state: i32,
            _resources: &Self::Resources,
            _bus: &mut Bus,
        ) -> Outcome<i32, Self::Error> {
            Outcome::Fault("boom")
        }
    }

    #[derive(Clone)]
    struct CapabilityGuarded;

    #[async_trait]
    impl Transition<(), ()> for CapabilityGuarded {
        type Error = String;
        type Resources = ();

        fn bus_access_policy(&self) -> Option<BusAccessPolicy> {
            Some(BusAccessPolicy::allow_only(vec![BusTypeRef::of::<i32>()]))
        }

        async fn run(
            &self,
            _state: (),
            _resources: &Self::Resources,
            bus: &mut Bus,
        ) -> Outcome<(), Self::Error> {
            match bus.get::<String>() {
                Ok(_) => Outcome::Next(()),
                Err(err) => Outcome::Fault(err.to_string()),
            }
        }
    }

    #[derive(Clone)]
    struct RecordingCompensationHook {
        calls: Arc<Mutex<Vec<CompensationContext>>>,
        should_fail: bool,
    }

    #[async_trait]
    impl CompensationHook for RecordingCompensationHook {
        async fn compensate(&self, context: CompensationContext) -> Result<()> {
            self.calls.lock().await.push(context);
            if self.should_fail {
                return Err(anyhow::anyhow!("compensation failed"));
            }
            Ok(())
        }
    }

    #[derive(Clone)]
    struct FlakyCompensationHook {
        calls: Arc<Mutex<u32>>,
        failures_remaining: Arc<Mutex<u32>>,
    }

    #[async_trait]
    impl CompensationHook for FlakyCompensationHook {
        async fn compensate(&self, _context: CompensationContext) -> Result<()> {
            {
                let mut calls = self.calls.lock().await;
                *calls += 1;
            }
            let mut failures_remaining = self.failures_remaining.lock().await;
            if *failures_remaining > 0 {
                *failures_remaining -= 1;
                return Err(anyhow::anyhow!("transient compensation failure"));
            }
            Ok(())
        }
    }

    #[derive(Clone)]
    struct FailingCompensationIdempotencyStore {
        read_calls: Arc<Mutex<u32>>,
        write_calls: Arc<Mutex<u32>>,
    }

    #[async_trait]
    impl CompensationIdempotencyStore for FailingCompensationIdempotencyStore {
        async fn was_compensated(&self, _key: &str) -> Result<bool> {
            let mut read_calls = self.read_calls.lock().await;
            *read_calls += 1;
            Err(anyhow::anyhow!("forced idempotency read failure"))
        }

        async fn mark_compensated(&self, _key: &str) -> Result<()> {
            let mut write_calls = self.write_calls.lock().await;
            *write_calls += 1;
            Err(anyhow::anyhow!("forced idempotency write failure"))
        }
    }

    #[tokio::test]
    async fn execute_persists_success_trace_when_handle_exists() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-success"));

        let axon = Axon::<i32, i32, std::convert::Infallible>::start("PersistSuccess").then(AddOne);
        let outcome = axon.execute(41, &(), &mut bus).await;
        assert!(matches!(outcome, Outcome::Next(42)));

        let persisted = store_impl.load("trace-success").await.unwrap().unwrap();
        assert_eq!(persisted.events.len(), 2);
        assert_eq!(persisted.events[0].outcome_kind, "Enter");
        assert_eq!(persisted.events[1].outcome_kind, "Next");
        assert_eq!(persisted.completion, Some(CompletionState::Success));
    }

    #[tokio::test]
    async fn execute_persists_fault_completion_state() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-fault"));

        let axon = Axon::<i32, i32, &'static str>::start("PersistFault").then(AlwaysFault);
        let outcome = axon.execute(41, &(), &mut bus).await;
        assert!(matches!(outcome, Outcome::Fault("boom")));

        let persisted = store_impl.load("trace-fault").await.unwrap().unwrap();
        assert_eq!(persisted.events.len(), 2);
        assert_eq!(persisted.events[1].outcome_kind, "Fault");
        assert_eq!(persisted.completion, Some(CompletionState::Fault));
    }

    #[tokio::test]
    async fn execute_respects_persistence_auto_complete_off() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-no-complete"));
        bus.insert(PersistenceAutoComplete(false));

        let axon =
            Axon::<i32, i32, std::convert::Infallible>::start("PersistNoComplete").then(AddOne);
        let outcome = axon.execute(1, &(), &mut bus).await;
        assert!(matches!(outcome, Outcome::Next(2)));

        let persisted = store_impl.load("trace-no-complete").await.unwrap().unwrap();
        assert_eq!(persisted.events.len(), 2);
        assert_eq!(persisted.completion, None);
    }

    #[tokio::test]
    async fn fault_triggers_compensation_and_marks_compensated() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();
        let calls = Arc::new(Mutex::new(Vec::new()));
        let compensation = RecordingCompensationHook {
            calls: calls.clone(),
            should_fail: false,
        };

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-compensated"));
        bus.insert(CompensationHandle::from_hook(compensation));

        let axon = Axon::<i32, i32, &'static str>::start("CompensatedFault").then(AlwaysFault);
        let outcome = axon.execute(7, &(), &mut bus).await;
        assert!(matches!(outcome, Outcome::Fault("boom")));

        let persisted = store_impl.load("trace-compensated").await.unwrap().unwrap();
        assert_eq!(persisted.events.len(), 3);
        assert_eq!(persisted.events[0].outcome_kind, "Enter");
        assert_eq!(persisted.events[1].outcome_kind, "Fault");
        assert_eq!(persisted.events[2].outcome_kind, "Compensated");
        assert_eq!(persisted.completion, Some(CompletionState::Compensated));

        let recorded = calls.lock().await;
        assert_eq!(recorded.len(), 1);
        assert_eq!(recorded[0].trace_id, "trace-compensated");
        assert_eq!(recorded[0].fault_kind, "Fault");
    }

    #[tokio::test]
    async fn failed_compensation_keeps_fault_completion() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();
        let calls = Arc::new(Mutex::new(Vec::new()));
        let compensation = RecordingCompensationHook {
            calls: calls.clone(),
            should_fail: true,
        };

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-compensation-failed"));
        bus.insert(CompensationHandle::from_hook(compensation));

        let axon = Axon::<i32, i32, &'static str>::start("CompensationFails").then(AlwaysFault);
        let outcome = axon.execute(7, &(), &mut bus).await;
        assert!(matches!(outcome, Outcome::Fault("boom")));

        let persisted = store_impl
            .load("trace-compensation-failed")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(persisted.events.len(), 2);
        assert_eq!(persisted.events[1].outcome_kind, "Fault");
        assert_eq!(persisted.completion, Some(CompletionState::Fault));

        let recorded = calls.lock().await;
        assert_eq!(recorded.len(), 1);
    }

    #[tokio::test]
    async fn compensation_retry_policy_succeeds_after_retries() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();
        let calls = Arc::new(Mutex::new(0u32));
        let failures_remaining = Arc::new(Mutex::new(2u32));
        let compensation = FlakyCompensationHook {
            calls: calls.clone(),
            failures_remaining,
        };

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-retry-success"));
        bus.insert(CompensationHandle::from_hook(compensation));
        bus.insert(CompensationRetryPolicy {
            max_attempts: 3,
            backoff_ms: 0,
        });

        let axon = Axon::<i32, i32, &'static str>::start("CompensationRetry").then(AlwaysFault);
        let outcome = axon.execute(7, &(), &mut bus).await;
        assert!(matches!(outcome, Outcome::Fault("boom")));

        let persisted = store_impl
            .load("trace-retry-success")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(persisted.completion, Some(CompletionState::Compensated));
        assert_eq!(
            persisted.events.last().map(|e| e.outcome_kind.as_str()),
            Some("Compensated")
        );

        let attempt_count = calls.lock().await;
        assert_eq!(*attempt_count, 3);
    }

    #[tokio::test]
    async fn compensation_idempotency_skips_duplicate_hook_execution() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();
        let calls = Arc::new(Mutex::new(Vec::new()));
        let compensation = RecordingCompensationHook {
            calls: calls.clone(),
            should_fail: false,
        };
        let idempotency = InMemoryCompensationIdempotencyStore::new();

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-idempotent"));
        bus.insert(PersistenceAutoComplete(false));
        bus.insert(CompensationHandle::from_hook(compensation));
        bus.insert(CompensationIdempotencyHandle::from_store(idempotency));

        let axon =
            Axon::<i32, i32, &'static str>::start("CompensationIdempotency").then(AlwaysFault);

        let outcome1 = axon.execute(7, &(), &mut bus).await;
        let outcome2 = axon.execute(8, &(), &mut bus).await;
        assert!(matches!(outcome1, Outcome::Fault("boom")));
        assert!(matches!(outcome2, Outcome::Fault("boom")));

        let persisted = store_impl.load("trace-idempotent").await.unwrap().unwrap();
        assert_eq!(persisted.completion, None);
        assert_eq!(persisted.events.len(), 6);
        assert_eq!(persisted.events[2].outcome_kind, "Compensated");
        assert_eq!(persisted.events[5].outcome_kind, "Compensated");

        let recorded = calls.lock().await;
        assert_eq!(recorded.len(), 1);
    }

    #[tokio::test]
    async fn compensation_idempotency_store_failure_does_not_block_compensation() {
        let store_impl = Arc::new(InMemoryPersistenceStore::new());
        let store_dyn: Arc<dyn PersistenceStore> = store_impl.clone();
        let calls = Arc::new(Mutex::new(Vec::new()));
        let read_calls = Arc::new(Mutex::new(0u32));
        let write_calls = Arc::new(Mutex::new(0u32));
        let compensation = RecordingCompensationHook {
            calls: calls.clone(),
            should_fail: false,
        };
        let idempotency = FailingCompensationIdempotencyStore {
            read_calls: read_calls.clone(),
            write_calls: write_calls.clone(),
        };

        let mut bus = Bus::new();
        bus.insert(PersistenceHandle::from_arc(store_dyn));
        bus.insert(PersistenceTraceId::new("trace-idempotency-store-failure"));
        bus.insert(CompensationHandle::from_hook(compensation));
        bus.insert(CompensationIdempotencyHandle::from_store(idempotency));

        let axon =
            Axon::<i32, i32, &'static str>::start("IdempotencyStoreFailure").then(AlwaysFault);
        let outcome = axon.execute(9, &(), &mut bus).await;
        assert!(matches!(outcome, Outcome::Fault("boom")));

        let persisted = store_impl
            .load("trace-idempotency-store-failure")
            .await
            .unwrap()
            .unwrap();
        assert_eq!(persisted.completion, Some(CompletionState::Compensated));
        assert_eq!(
            persisted.events.last().map(|e| e.outcome_kind.as_str()),
            Some("Compensated")
        );

        let recorded = calls.lock().await;
        assert_eq!(recorded.len(), 1);
        assert_eq!(*read_calls.lock().await, 1);
        assert_eq!(*write_calls.lock().await, 1);
    }

    #[tokio::test]
    async fn transition_bus_policy_blocks_unauthorized_resource_access() {
        let mut bus = Bus::new();
        bus.insert(1_i32);
        bus.insert("secret".to_string());

        let axon = Axon::<(), (), String>::start("BusPolicy").then(CapabilityGuarded);
        let outcome = axon.execute((), &(), &mut bus).await;

        match outcome {
            Outcome::Fault(msg) => {
                assert!(msg.contains("Bus access denied"), "{msg}");
                assert!(msg.contains("CapabilityGuarded"), "{msg}");
                assert!(msg.contains("alloc::string::String"), "{msg}");
            }
            other => panic!("expected fault, got {other:?}"),
        }
    }
}