udb 0.4.21

// src/runtime/saga.rs — Saga state helpers and autonomous recovery worker.
//
// NW1-3c (2026-05-29): every PG-pool reference has been replaced with
// `Arc<dyn SystemStores>` routing. The recovery worker now uses:
//
// - `SagaStore::claim_recoverable_sagas` to find indeterminate +
//   stale in-progress sagas (the old `WHERE status = 'indeterminate'
//   OR (status = 'in_progress' AND EXTRACT(EPOCH FROM …) > stale)`
//   query collapses into one trait call).
// - `SagaStore::increment_recovery_attempts` for the failure-path
//   `recovery_attempts` upsert (PG `RETURNING` was inline; the trait
//   now hides that across PG/MySQL/SQLite).
// - `SagaStore::update_saga_status` for the success / manual-review
//   transitions.
// - `CanonicalStore::try_acquire_advisory_lease` /
//   `release_advisory_lease` for the worker lease (was
//   `pg_try_advisory_lock` / `pg_advisory_unlock`).
//
// The public free functions (`list_sagas`, `get_saga`,
// `mark_saga_reviewed`, `retry_saga_compensation`) keep their public
// signature shape but now take `Arc<dyn SystemStores>` instead of
// `&PgPool` — handlers_admin and handlers_policy were updated at the
// same time (see NW1-3c.3 entry).
//
// Configuration is supplied through UdbConfig::saga. Environment
// compatibility is merged once during UdbConfig loading.

use std::sync::Arc;
use std::time::Duration;

use serde::Serialize;
use uuid::Uuid;

use crate::runtime::canonical_store::system_store::{
    CompensationStatus, SagaListFilter, SagaStatus, SagaStore,
};
use crate::runtime::canonical_store::{CanonicalStore, SystemStores};
use crate::runtime::config::SagaSettings;
use crate::runtime::system::SystemCatalogConfig;

const SAGA_WORKER_LEASE: &str = "udb-saga-worker";

/// Terminal statuses written by the recovery loop. Pinned strings
/// matching the values the admin RPC and dashboards expect; they
/// correspond 1:1 to the [`SagaStatus`] enum variants.
pub const SAGA_STATUS_COMMITTED: &str = "committed";
pub const SAGA_STATUS_COMPENSATED: &str = "compensated";
pub const SAGA_STATUS_FAILED_COMPENSATION: &str = "failed_compensation";
pub const SAGA_STATUS_MANUAL_REVIEW: &str = "manual_review";

fn saga_recompensation_not_retryable_status(saga_id: &str) -> tonic::Status {
    crate::runtime::executor_utils::policy_status(
        "retry_saga_compensation",
        "saga_not_retryable",
        format!(
            "saga {saga_id} is not in a retryable state (failed_compensation or manual_review)"
        ),
    )
}

fn saga_not_found_status(operation: &'static str, saga_id: &str) -> tonic::Status {
    crate::runtime::executor_utils::schema_status(
        tonic::Code::NotFound,
        "saga",
        operation,
        "saga_not_found",
        format!("saga {saga_id} not found"),
    )
}

fn saga_internal_status(operation: impl Into<String>, message: impl Into<String>) -> tonic::Status {
    crate::runtime::executor_utils::internal_status("saga", operation, message)
}

/// 9.12 (WorkflowService) — discriminates a durable *workflow* saga from a plain
/// data-plane saga. This is the minimal ADDITIVE seam the native `WorkflowService`
/// uses to hand a workflow to the existing saga engine; it adds NO second
/// orchestration loop. The DEFAULT kind reproduces the pre-9.12 behaviour exactly:
/// [`SagaKind::tag_operation`] returns the operation string verbatim and
/// [`SagaKind::from_operation`] classifies an untagged operation as `Default`, so
/// every existing saga path (which only ever records `Default` sagas) is unchanged
/// byte-for-byte. Because `udb_sagas` has no dedicated discriminator column, the
/// kind is carried in the existing `operation` field via an unambiguous prefix.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum SagaKind {
    /// A plain data-plane saga (XA write fan-out). Pre-9.12 behaviour, unchanged.
    #[default]
    Default,
    /// A durable workflow instance orchestrated by the native `WorkflowService`.
    Workflow,
}

impl SagaKind {
    /// Prefix marking a saga `operation` as workflow-orchestrated. Data-plane
    /// operations are bare verbs (`upsert`/`delete`/…), so this namespaced prefix
    /// can never collide with an existing operation token.
    const WORKFLOW_OPERATION_PREFIX: &'static str = "workflow:";

    /// Tag an operation string with this kind. `Default` is the identity function
    /// (verbatim), so existing data-plane sagas are recorded exactly as before.
    pub fn tag_operation(self, operation: &str) -> String {
        match self {
            SagaKind::Default => operation.to_string(),
            SagaKind::Workflow => format!("{}{operation}", Self::WORKFLOW_OPERATION_PREFIX),
        }
    }

    /// Recover the kind from a stored operation string. An untagged operation is
    /// `Default` (every pre-9.12 saga), preserving existing classification.
    pub fn from_operation(operation: &str) -> Self {
        if operation.starts_with(Self::WORKFLOW_OPERATION_PREFIX) {
            SagaKind::Workflow
        } else {
            SagaKind::Default
        }
    }

    pub fn as_str(self) -> &'static str {
        match self {
            SagaKind::Default => "default",
            SagaKind::Workflow => "workflow",
        }
    }
}

/// 9.12 seam — record a durable saga for a workflow instance, REUSING the existing
/// [`SagaStore::record_saga`] persistence and the [`SagaRecoveryWorker`]
/// reverse-order compensation machinery. This adds no new orchestration loop:
/// forward step dispatch stays in the WorkflowService tick (event-fire only, like
/// the scheduler), and compensation is the established recovery path. The saga is
/// recorded `Pending` (NOT recoverable — see [`SagaStatus::is_recoverable`]) so it
/// sits untouched until the workflow either completes (marked `Committed`) or is
/// cancelled (flipped to `Indeterminate`, which the recovery worker then
/// compensates). Returns the assigned `saga_id`.
pub async fn start_workflow_saga(
    store: &dyn SystemStores,
    kind: SagaKind,
    tenant_id: &str,
    correlation_id: &str,
    operation: &str,
    compensations: serde_json::Value,
) -> Result<Uuid, String> {
    let insert = crate::runtime::canonical_store::system_store::SagaInsert {
        tx_id: Uuid::new_v4().to_string(),
        tenant_id: tenant_id.to_string(),
        correlation_id: correlation_id.to_string(),
        backend_instance: "workflow".to_string(),
        operation: kind.tag_operation(operation),
        status: SagaStatus::Pending,
        steps: serde_json::json!([]),
        compensations,
    };
    SagaStore::record_saga(store, &insert)
        .await
        .map_err(|err| err.to_string())
}

/// A saga record as returned by admin / recovery queries.
#[derive(Debug, Clone, Serialize)]
pub struct SagaAdminRecord {
    pub saga_id: String,
    pub tx_id: String,
    pub tenant_id: String,
    pub correlation_id: String,
    pub status: String,
    pub current_step: i32,
    pub steps_json: serde_json::Value,
    pub compensations_json: serde_json::Value,
    pub last_error: String,
    pub created_at: chrono::DateTime<chrono::Utc>,
    pub updated_at: chrono::DateTime<chrono::Utc>,
}

impl From<crate::runtime::canonical_store::system_store::SagaRow> for SagaAdminRecord {
    fn from(row: crate::runtime::canonical_store::system_store::SagaRow) -> Self {
        Self {
            saga_id: row.saga_id.to_string(),
            tx_id: row.tx_id,
            tenant_id: row.tenant_id,
            correlation_id: row.correlation_id,
            status: row.status.as_str().to_string(),
            current_step: row.current_step,
            steps_json: row.steps,
            compensations_json: row.compensations,
            last_error: row.last_error,
            created_at: row.created_at,
            updated_at: row.updated_at,
        }
    }
}

/// Summary emitted after each recovery pass.
#[derive(Debug, Clone, Default, Serialize)]
pub struct SagaRecoveryReport {
    pub owner_id: String,
    pub lease_acquired: bool,
    pub scanned: usize,
    pub compensated: usize,
    pub marked_manual_review: usize,
    pub errors: Vec<String>,
}

/// Autonomous saga recovery worker.
///
/// Spawned by the runtime at startup when saga recovery is enabled.
/// NW1-3c: now holds `Arc<dyn SystemStores>` instead of a raw `PgPool`,
/// so the same worker code works against any registered canonical
/// store (PG today; MySQL / SQLite tomorrow).
pub struct SagaRecoveryWorker {
    store: Arc<dyn SystemStores>,
    interval: Duration,
    stale_threshold: Duration,
    worker_lease_ttl: Duration,
    recovery_batch_size: i64,
    owner_id: String,
    poison_threshold: i64,
    /// U19: plugin-aware backend compensator registry. Empty registry
    /// preserves the pre-U19 log-only behaviour (each step becomes a
    /// `NoCompensator` outcome and the saga lands in
    /// `failed_compensation` until an operator registers an
    /// implementation). When the runtime constructs the worker with
    /// `with_compensators`, real Qdrant/S3/etc. calls run.
    compensators: Arc<crate::runtime::saga_compensators::CompensatorRegistry>,
    /// U19 retry policy. Controls when a repeatedly-failing saga gets
    /// escalated to `manual_review` instead of retried on the next
    /// sweep.
    quarantine_policy: crate::runtime::saga_compensators::QuarantinePolicy,
    /// Metrics sink for recovery outcomes (compensated / failed-compensation).
    /// Defaults to `NoopMetrics`; the supervisor wires the live
    /// `PrometheusMetrics` via [`SagaRecoveryWorker::with_metrics`].
    metrics: std::sync::Arc<dyn crate::metrics::MetricsRecorder>,
}

impl SagaRecoveryWorker {
    /// NW1-3c: construct from a `SystemStores` trait object.
    pub fn new(store: Arc<dyn SystemStores>) -> Self {
        let mut settings = SagaSettings::default();
        settings.merge_env();
        Self::with_settings(store, &settings)
    }

    /// NW1-3c: parameterised constructor.
    pub fn with_settings(store: Arc<dyn SystemStores>, settings: &SagaSettings) -> Self {
        Self {
            store,
            interval: Duration::from_secs(settings.recovery_interval_secs.max(1)),
            stale_threshold: Duration::from_secs(settings.stale_threshold_secs.max(1)),
            worker_lease_ttl: Duration::from_secs(settings.worker_lease_ttl_secs.max(1)),
            recovery_batch_size: settings.recovery_batch_size.max(1),
            owner_id: Uuid::new_v4().to_string(),
            poison_threshold: settings.poison_threshold.max(1),
            compensators: Arc::new(
                crate::runtime::saga_compensators::CompensatorRegistry::default(),
            ),
            quarantine_policy: crate::runtime::saga_compensators::QuarantinePolicy {
                // Align the policy's max-attempts with the configured poison
                // threshold; the per-saga cooldown comes from the default (#134).
                max_attempts: settings.poison_threshold.max(1) as u32,
                ..crate::runtime::saga_compensators::QuarantinePolicy::default()
            },
            metrics: Arc::new(crate::metrics::NoopMetrics),
        }
    }

    /// Wire a live metrics recorder so recovery outcomes
    /// (`inc_saga_compensated_total` / `inc_saga_failed_compensations_total`)
    /// are emitted. Without this the worker uses `NoopMetrics`.
    pub fn with_metrics(
        mut self,
        metrics: std::sync::Arc<dyn crate::metrics::MetricsRecorder>,
    ) -> Self {
        self.metrics = metrics;
        self
    }

    /// U19: wire a registry of backend-aware compensators. The runtime
    /// supervisor builds this at startup by registering Qdrant, S3,
    /// Mongo, Neo4j, ClickHouse, and Postgres implementations and
    /// passes it here. With an empty registry the worker logs and
    /// marks `failed_compensation` as before.
    pub fn with_compensators(
        mut self,
        registry: Arc<crate::runtime::saga_compensators::CompensatorRegistry>,
    ) -> Self {
        self.compensators = registry;
        self
    }

    /// U19: override the default quarantine policy
    /// (`max_attempts=5, cooldown_secs=30`). The recovery worker uses
    /// this to escalate poisoned sagas to `manual_review`.
    pub fn with_quarantine_policy(
        mut self,
        policy: crate::runtime::saga_compensators::QuarantinePolicy,
    ) -> Self {
        self.quarantine_policy = policy;
        self
    }

    /// Return `true` when saga recovery is enabled via env.
    pub fn is_enabled() -> bool {
        let mut settings = SagaSettings::default();
        settings.merge_env();
        settings.recovery_enabled
    }

    pub fn is_enabled_with_settings(settings: &SagaSettings) -> bool {
        settings.recovery_enabled
    }

    /// Run the recovery loop forever (call from a dedicated tokio task).
    pub async fn run_forever(self) {
        let mut ticker = tokio::time::interval(self.interval);
        loop {
            ticker.tick().await;
            let report = self.run_once().await;
            if report.scanned > 0 || !report.errors.is_empty() {
                tracing::info!(
                    scanned = report.scanned,
                    compensated = report.compensated,
                    marked_manual_review = report.marked_manual_review,
                    errors = report.errors.len(),
                    "saga recovery pass completed"
                );
            }
            for err in &report.errors {
                tracing::warn!(error = err, "saga recovery error");
            }
        }
    }

    /// Execute one recovery pass. Returns a summary report.
    pub async fn run_once(&self) -> SagaRecoveryReport {
        let mut report = SagaRecoveryReport {
            owner_id: self.owner_id.clone(),
            ..SagaRecoveryReport::default()
        };
        if !self.try_acquire_worker_lease(&mut report).await {
            return report;
        }
        let mut locked_report = self.run_once_locked().await;
        locked_report.owner_id = self.owner_id.clone();
        locked_report.lease_acquired = true;
        self.release_worker_lease(&mut locked_report).await;
        locked_report
    }

    async fn refresh_saga_active_metric(&self) {
        match SagaStore::saga_summary(self.store.as_ref()).await {
            Ok(summary) => self.metrics.set_saga_active(summary.in_progress),
            Err(err) => {
                tracing::debug!(error = %err, "saga active metric refresh failed");
            }
        }
    }

    fn observe_saga_duration_since(&self, created_at: chrono::DateTime<chrono::Utc>) {
        let seconds = chrono::Utc::now()
            .signed_duration_since(created_at)
            .to_std()
            .map(|duration| duration.as_secs_f64())
            .unwrap_or(0.0);
        self.metrics.observe_saga_duration_seconds(seconds);
    }

    async fn run_once_locked(&self) -> SagaRecoveryReport {
        let mut report = SagaRecoveryReport {
            owner_id: self.owner_id.clone(),
            lease_acquired: true,
            ..SagaRecoveryReport::default()
        };
        self.refresh_saga_active_metric().await;
        // NW1-3c: claim_recoverable_sagas returns INDETERMINATE +
        // stale IN_PROGRESS in oldest-first order.
        let candidates = match SagaStore::claim_recoverable_sagas(
            self.store.as_ref(),
            self.stale_threshold,
            self.recovery_batch_size,
        )
        .await
        {
            Ok(rows) => rows,
            Err(err) => {
                report.errors.push(format!("saga scan failed: {err}"));
                self.refresh_saga_active_metric().await;
                return report;
            }
        };
        report.scanned = candidates.len();

        // Accumulate terminal outcomes and flush them grouped after the loop —
        // one status round-trip per outcome instead of one per saga (#89).
        let mut compensated_ids = Vec::new();
        let mut manual_review_ids = Vec::new();

        for row in candidates {
            let saga_id = row.saga_id;
            let saga_id_str = saga_id.to_string();
            let created_at = row.created_at;
            let compensations = row.compensations;

            // #134: drive the retry decision through the QuarantinePolicy. A saga
            // that has exhausted its attempts is escalated without re-attempting;
            // one still inside its per-saga cooldown window is skipped this sweep
            // (the claim already gates on stale_threshold; this adds the policy's
            // cooldown + max-attempts that recovery previously ignored).
            use crate::runtime::saga_compensators::QuarantineState;
            match self.quarantine_policy.evaluate(
                row.recovery_attempts.max(0) as u32,
                row.updated_at.timestamp(),
                chrono::Utc::now().timestamp(),
            ) {
                QuarantineState::Quarantine { .. } => {
                    manual_review_ids.push(saga_id);
                    self.observe_saga_duration_since(created_at);
                    report.marked_manual_review += 1;
                    continue;
                }
                QuarantineState::Cooldown { .. } => continue,
                QuarantineState::Retry { .. } => {}
            }

            // Attempt compensation
            let compensation_result = self
                .attempt_compensations(&saga_id_str, &compensations)
                .await;

            let new_status = match compensation_result {
                Ok(_) => {
                    report.compensated += 1;
                    self.metrics.inc_saga_compensated_total();
                    self.observe_saga_duration_since(created_at);
                    SagaStatus::Compensated
                }
                Err(err) => {
                    // Every failed compensation attempt is recorded (whether it
                    // will be retried next sweep or escalated to manual review).
                    self.metrics.inc_saga_failed_compensations_total();
                    self.metrics.inc_compensation_failures_total();
                    // Increment recovery_attempts atomically. The
                    // trait returns the post-increment counter; the
                    // pre-NW1-3c code did the same via PG `RETURNING`.
                    let attempts = match SagaStore::increment_recovery_attempts(
                        self.store.as_ref(),
                        saga_id,
                        &format!("compensation failed: {err}"),
                    )
                    .await
                    {
                        Ok(n) => n,
                        Err(e) => {
                            report.errors.push(format!(
                                "saga {saga_id_str} recovery_attempts increment failed: {e}"
                            ));
                            continue;
                        }
                    };

                    if attempts >= self.poison_threshold {
                        report.marked_manual_review += 1;
                        self.observe_saga_duration_since(created_at);
                        SagaStatus::ManualReview
                    } else {
                        report.errors.push(format!(
                            "saga {saga_id_str} compensation failed (attempt {attempts}): {err}"
                        ));
                        continue;
                    }
                }
            };

            match new_status {
                SagaStatus::Compensated => compensated_ids.push(saga_id),
                SagaStatus::ManualReview => manual_review_ids.push(saga_id),
                _ => {}
            }
        }

        // Flush the terminal status writes grouped by outcome — Postgres applies
        // each group in a single `WHERE saga_id = ANY(...)` round-trip (#89).
        if !compensated_ids.is_empty()
            && let Err(err) = SagaStore::update_saga_statuses_batch(
                self.store.as_ref(),
                &compensated_ids,
                SagaStatus::Compensated,
                CompensationStatus::Completed,
            )
            .await
        {
            report
                .errors
                .push(format!("batch compensated status update failed: {err}"));
        }
        if !manual_review_ids.is_empty()
            && let Err(err) = SagaStore::update_saga_statuses_batch(
                self.store.as_ref(),
                &manual_review_ids,
                SagaStatus::ManualReview,
                CompensationStatus::ManualReview,
            )
            .await
        {
            report
                .errors
                .push(format!("batch manual-review status update failed: {err}"));
        }
        self.refresh_saga_active_metric().await;
        report
    }

    async fn try_acquire_worker_lease(&self, report: &mut SagaRecoveryReport) -> bool {
        // Ensure the lease table exists before first acquire. Idempotent.
        if let Err(err) = CanonicalStore::ensure_advisory_lease_table(self.store.as_ref()).await {
            report
                .errors
                .push(format!("ensure_advisory_lease_table failed: {err}"));
            return false;
        }
        match CanonicalStore::try_acquire_advisory_lease(
            self.store.as_ref(),
            SAGA_WORKER_LEASE,
            &self.owner_id,
            self.worker_lease_ttl,
        )
        .await
        {
            Ok(true) => {
                report.lease_acquired = true;
                tracing::debug!(owner_id = self.owner_id, "saga worker lease acquired");
                true
            }
            Ok(false) => {
                tracing::debug!(owner_id = self.owner_id, "saga worker lease held elsewhere");
                false
            }
            Err(err) => {
                report
                    .errors
                    .push(format!("saga worker lease acquisition failed: {err}"));
                false
            }
        }
    }

    async fn release_worker_lease(&self, report: &mut SagaRecoveryReport) {
        if let Err(err) = CanonicalStore::release_advisory_lease(
            self.store.as_ref(),
            SAGA_WORKER_LEASE,
            &self.owner_id,
        )
        .await
        {
            report
                .errors
                .push(format!("saga worker lease release failed: {err}"));
        }
    }

    /// Attempt to run compensation actions stored in the saga record.
    ///
    /// U19: routes every step through the registered
    /// [`CompensatorRegistry`]. With no compensators registered, every
    /// step yields a `NoCompensator` outcome and this returns
    /// `Err(...)` so the worker marks the saga
    /// `failed_compensation` rather than silently advancing — matches
    /// the pre-U19 behaviour minus the misleading `Ok(())`.
    ///
    /// Compensations are stored as a JSON array of
    /// `{ "backend": "...", "operation": "...", "resource_uri": "...",
    /// "payload": {...} }` objects (the `"compensation"` key is
    /// accepted as a legacy alias for `"payload"`).
    async fn attempt_compensations(
        &self,
        saga_id: &str,
        compensations: &serde_json::Value,
    ) -> Result<(), String> {
        use crate::runtime::saga_compensators::{
            StepOutcome, all_compensated, dispatch_compensations,
        };
        let outcomes = dispatch_compensations(self.compensators.as_ref(), compensations).await;
        for (i, outcome) in outcomes.iter().enumerate() {
            match outcome {
                StepOutcome::Compensated => tracing::info!(
                    saga_id = saga_id,
                    step = i,
                    "saga compensation step succeeded"
                ),
                StepOutcome::NoCompensator { backend } => tracing::warn!(
                    saga_id = saga_id,
                    step = i,
                    backend = backend,
                    "no backend compensator registered; step skipped"
                ),
                StepOutcome::Malformed { index } => tracing::warn!(
                    saga_id = saga_id,
                    step = index,
                    "saga compensation step is malformed; skipping"
                ),
                StepOutcome::Failed { backend, reason } => tracing::error!(
                    saga_id = saga_id,
                    step = i,
                    backend = backend,
                    reason = reason,
                    "saga compensation step failed"
                ),
            }
        }
        if all_compensated(&outcomes) {
            Ok(())
        } else {
            // Aggregate the first failure / no-compensator into the
            // error message so the worker writes a useful `last_error`.
            let first_bad = outcomes
                .iter()
                .find(|o| !o.is_success())
                .map(|o| match o {
                    StepOutcome::Failed { backend, reason } => {
                        format!("compensation failed at {backend}: {reason}")
                    }
                    StepOutcome::NoCompensator { backend } => {
                        format!("no compensator registered for backend '{backend}'")
                    }
                    StepOutcome::Malformed { index } => {
                        format!("malformed compensation step at index {index}")
                    }
                    StepOutcome::Compensated => unreachable!(),
                })
                .unwrap_or_else(|| "no compensation steps recorded".to_string());
            Err(first_bad)
        }
    }
}

/// List sagas matching the given filters. Suitable for the `ListSagas` RPC.
///
/// NW1-3c: routes through `SagaStore::list_sagas`. The PG-only
/// `QueryBuilder` path is gone; arg validation (status enum membership +
/// tx_id UUID format) is now done up-front before constructing the
/// trait filter.
fn saga_field_violation(
    field: &'static str,
    description: impl Into<String>,
    message: impl Into<String>,
) -> tonic::Status {
    crate::runtime::executor_utils::invalid_argument_fields(
        message,
        [(field.to_string(), description.into())],
    )
}

fn parse_saga_uuid_field(
    field: &'static str,
    value: &str,
    message: &'static str,
) -> Result<Uuid, tonic::Status> {
    value
        .parse()
        .map_err(|_| saga_field_violation(field, "must be a valid UUID", message))
}

fn validate_tx_id_filter(tx_id_filter: &str) -> Result<(), tonic::Status> {
    if !tx_id_filter.is_empty() {
        parse_saga_uuid_field(
            "tx_id_filter",
            tx_id_filter,
            "tx_id_filter must be a valid UUID",
        )?;
    }
    Ok(())
}

fn parse_saga_status_filter(status_filter: &str) -> Result<Option<SagaStatus>, tonic::Status> {
    if status_filter.is_empty() {
        return Ok(None);
    }
    SagaStatus::parse(status_filter).map(Some).ok_or_else(|| {
        let allowed = SagaStatus::all()
            .iter()
            .map(|s| s.as_str())
            .collect::<Vec<_>>();
        saga_field_violation(
            "status_filter",
            format!("must be one of {}", allowed.join(", ")),
            format!("invalid status_filter '{status_filter}'; must be one of {allowed:?}"),
        )
    })
}

#[allow(clippy::too_many_arguments)]
pub async fn list_sagas(
    store: &dyn SystemStores,
    _config: &SystemCatalogConfig,
    tenant_id_filter: &str,
    status_filter: &str,
    tx_id_filter: &str,
    correlation_id_filter: &str,
    limit: i64,
    offset: i64,
) -> Result<Vec<SagaAdminRecord>, tonic::Status> {
    // Validate status filter against the canonical enum tokens. This
    // is the same set the PG CHECK constraint enforces.
    let status_enum = parse_saga_status_filter(status_filter)?;

    // UUID-validate tx_id_filter (defensive — keeps PG behaviour
    // identical for callers that pass UUIDs; non-UUID tx_ids are
    // technically allowed by the schema but the admin RPC rejected
    // them historically).
    validate_tx_id_filter(tx_id_filter)?;

    let filter = SagaListFilter {
        tenant_id: (!tenant_id_filter.is_empty()).then(|| tenant_id_filter.to_string()),
        status: status_enum,
        tx_id: (!tx_id_filter.is_empty()).then(|| tx_id_filter.to_string()),
        correlation_id: (!correlation_id_filter.is_empty())
            .then(|| correlation_id_filter.to_string()),
        limit,
        offset,
    };
    let rows = SagaStore::list_sagas(store, &filter).await.map_err(|err| {
        saga_internal_status("list_sagas", format!("list_sagas query failed: {err}"))
    })?;
    Ok(rows.into_iter().map(SagaAdminRecord::from).collect())
}

/// Get a single saga by ID.
///
/// NW1-3c: routes through `SagaStore::get_saga`. Returns `NotFound`
/// when the saga doesn't exist (matching the pre-NW1-3c gRPC code).
pub async fn get_saga(
    store: &dyn SystemStores,
    _config: &SystemCatalogConfig,
    saga_id: &str,
) -> Result<SagaAdminRecord, tonic::Status> {
    let id = parse_saga_uuid_field("saga_id", saga_id, "saga_id must be a UUID")?;
    let row = SagaStore::get_saga(store, id)
        .await
        .map_err(|err| saga_internal_status("get_saga", format!("get_saga failed: {err}")))?
        .ok_or_else(|| saga_not_found_status("get_saga", saga_id))?;
    Ok(SagaAdminRecord::from(row))
}

/// Mark a saga as MANUAL_REVIEW (operator-reviewed; prevents further auto-recovery).
///
/// NW1-3c: routes through `SagaStore::mark_saga_manual_review`.
/// The trait returns `InvalidInput` when the row doesn't exist; we
/// map that to `NotFound` for backward compatibility with the
/// admin RPC.
pub async fn mark_saga_reviewed(
    store: &dyn SystemStores,
    _config: &SystemCatalogConfig,
    saga_id: &str,
) -> Result<(), tonic::Status> {
    let id = parse_saga_uuid_field("saga_id", saga_id, "saga_id must be a UUID")?;
    SagaStore::mark_saga_manual_review(store, id)
        .await
        .map_err(|err| {
            let msg = err.to_string();
            if msg.contains("not found") {
                saga_not_found_status("mark_saga_reviewed", saga_id)
            } else {
                saga_internal_status(
                    "mark_saga_reviewed",
                    format!("mark_saga_reviewed failed: {msg}"),
                )
            }
        })
}

/// Trigger re-compensation for a saga (sets status back to 'indeterminate'
/// so the recovery loop picks it up on the next pass).
///
/// NW1-3c: routes through `SagaStore::request_saga_recompensation`,
/// which enforces the same state-machine guard the pre-NW1-3c SQL
/// did: refuses unless the saga is currently `failed_compensation`
/// or `manual_review`. The gRPC error code is preserved
/// (`FailedPrecondition` instead of `NotFound`).
pub async fn retry_saga_compensation(
    store: &dyn SystemStores,
    _config: &SystemCatalogConfig,
    saga_id: &str,
) -> Result<(), tonic::Status> {
    let id = parse_saga_uuid_field("saga_id", saga_id, "saga_id must be a UUID")?;
    SagaStore::request_saga_recompensation(store, id)
        .await
        .map_err(|err| {
            let msg = err.to_string();
            if msg.contains("retryable state") {
                saga_recompensation_not_retryable_status(saga_id)
            } else {
                saga_internal_status(
                    "retry_saga_compensation",
                    format!("retry_saga_compensation failed: {msg}"),
                )
            }
        })
}

#[cfg(test)]
#[allow(clippy::items_after_test_module)]
mod tests {
    use super::*;
    use crate::proto::{ErrorDetail, ErrorKind};
    #[cfg(feature = "sqlite")]
    use crate::runtime::canonical_store::sqlite::SqliteCanonicalStore;
    #[cfg(feature = "sqlite")]
    use crate::runtime::canonical_store::system_store::SagaInsert;
    use crate::runtime::executor_utils::ERROR_DETAIL_METADATA_KEY;
    #[cfg(feature = "sqlite")]
    use sqlx::sqlite::SqlitePoolOptions;

    fn decode_detail(status: &tonic::Status) -> ErrorDetail {
        let raw = status
            .metadata()
            .get_bin(ERROR_DETAIL_METADATA_KEY)
            .expect("error-detail trailer present")
            .to_bytes()
            .expect("trailer decodes to bytes");
        crate::runtime::executor_utils::decode_error_detail_from_raw(&raw)
    }

    fn assert_single_field_violation(status: &tonic::Status, field: &str, description: &str) {
        assert_eq!(status.code(), tonic::Code::InvalidArgument);
        let detail = decode_detail(status);
        assert_eq!(detail.kind, ErrorKind::Validation as i32);
        assert_eq!(detail.field_violations.len(), 1);
        assert_eq!(detail.field_violations[0].field, field);
        assert_eq!(detail.field_violations[0].description, description);
    }

    fn assert_policy_detail(
        status: &tonic::Status,
        operation: &str,
        policy_decision_id: &str,
        message: &str,
    ) {
        assert_eq!(status.code(), tonic::Code::FailedPrecondition);
        assert_eq!(status.message(), message);
        let detail = decode_detail(status);
        assert_eq!(detail.kind, ErrorKind::Policy as i32);
        assert_eq!(detail.operation, operation);
        assert_eq!(detail.policy_decision_id, policy_decision_id);
        assert!(!detail.retryable);
        assert_eq!(detail.retry_after_ms, 0);
        assert!(detail.field_violations.is_empty());
    }

    fn assert_schema_detail(
        status: &tonic::Status,
        operation: &str,
        schema_code: &str,
        message: &str,
    ) {
        assert_eq!(status.code(), tonic::Code::NotFound);
        assert_eq!(status.message(), message);
        let detail = decode_detail(status);
        assert_eq!(detail.kind, ErrorKind::Schema as i32);
        assert_eq!(detail.backend, "saga");
        assert_eq!(detail.operation, operation);
        assert_eq!(detail.capability_required, schema_code);
        assert!(!detail.retryable);
        assert_eq!(detail.retry_after_ms, 0);
        assert!(detail.field_violations.is_empty());
    }

    fn assert_internal_detail(status: &tonic::Status, operation: &str, message: &str) {
        assert_eq!(status.code(), tonic::Code::Internal);
        assert_eq!(status.message(), message);
        let detail = decode_detail(status);
        assert_eq!(detail.kind, ErrorKind::Internal as i32);
        assert_eq!(detail.backend, "saga");
        assert_eq!(detail.operation, operation);
        assert!(!detail.retryable);
        assert_eq!(detail.retry_after_ms, 0);
        assert!(detail.field_violations.is_empty());
    }

    #[test]
    fn recovery_worker_uses_env_interval() {
        // Verify the default interval values without constructing the worker.
        let interval = Duration::from_secs(60);
        let stale = Duration::from_secs(300);
        assert_eq!(interval, Duration::from_secs(60));
        assert_eq!(stale, Duration::from_secs(300));
    }

    #[test]
    fn recovery_enabled_by_default() {
        // "false", "0", "no", "FALSE" all disable recovery.
        assert!(matches!("false", "0" | "false" | "FALSE" | "no"));
        assert!(matches!("0", "0" | "false" | "FALSE" | "no"));
        // Any other value keeps recovery enabled.
        assert!(!matches!("true", "0" | "false" | "FALSE" | "no"));
        assert!(!matches!("1", "0" | "false" | "FALSE" | "no"));
    }

    #[test]
    fn saga_admin_status_filter_carries_field_violation() {
        let err = parse_saga_status_filter("bogus")
            .expect_err("unknown status_filter must fail before store access");
        assert_eq!(
            err.message(),
            "invalid status_filter 'bogus'; must be one of [\"indeterminate\", \"in_progress\", \"pending\", \"committed\", \"compensated\", \"failed\", \"in_doubt\", \"failed_compensation\", \"manual_review\"]"
        );
        assert_single_field_violation(
            &err,
            "status_filter",
            "must be one of indeterminate, in_progress, pending, committed, compensated, failed, in_doubt, failed_compensation, manual_review",
        );
    }

    #[test]
    fn saga_admin_uuid_validation_carries_field_violations() {
        let tx_id = validate_tx_id_filter("not-a-uuid")
            .expect_err("invalid tx_id_filter must fail before store access");
        assert_eq!(tx_id.message(), "tx_id_filter must be a valid UUID");
        assert_single_field_violation(&tx_id, "tx_id_filter", "must be a valid UUID");

        let saga_id = parse_saga_uuid_field("saga_id", "not-a-uuid", "saga_id must be a UUID")
            .expect_err("invalid saga_id must fail before store access");
        assert_eq!(saga_id.message(), "saga_id must be a UUID");
        assert_single_field_violation(&saga_id, "saga_id", "must be a valid UUID");
    }

    #[test]
    fn saga_recompensation_not_retryable_carries_policy_detail() {
        let saga_id = "11111111-1111-4111-8111-111111111111";
        let err = saga_recompensation_not_retryable_status(saga_id);
        assert_policy_detail(
            &err,
            "retry_saga_compensation",
            "saga_not_retryable",
            "saga 11111111-1111-4111-8111-111111111111 is not in a retryable state (failed_compensation or manual_review)",
        );
    }

    #[test]
    fn saga_not_found_statuses_carry_schema_detail() {
        let saga_id = "11111111-1111-4111-8111-111111111111";
        for operation in ["get_saga", "mark_saga_reviewed"] {
            assert_schema_detail(
                &saga_not_found_status(operation, saga_id),
                operation,
                "saga_not_found",
                "saga 11111111-1111-4111-8111-111111111111 not found",
            );
        }
    }

    #[test]
    fn saga_internal_status_carries_typed_detail() {
        let status = saga_internal_status("list_sagas", "list_sagas query failed");

        assert_internal_detail(&status, "list_sagas", "list_sagas query failed");
    }

    /// 9.12 seam — the DEFAULT kind reproduces existing saga behaviour: the
    /// operation string is recorded verbatim and classifies back to `Default`, so
    /// no pre-9.12 saga path changes. The `Workflow` kind round-trips through the
    /// `operation` field without colliding with bare data-plane verbs.
    #[test]
    fn saga_kind_default_preserves_operation_and_round_trips() {
        // Default is the identity function — every existing saga is unchanged.
        assert_eq!(SagaKind::Default.tag_operation("upsert"), "upsert");
        assert_eq!(SagaKind::from_operation("upsert"), SagaKind::Default);
        assert_eq!(SagaKind::default(), SagaKind::Default);

        // Workflow tags + recovers without colliding with a bare verb.
        let tagged = SagaKind::Workflow.tag_operation("order_fulfillment");
        assert_ne!(tagged, "order_fulfillment");
        assert_eq!(SagaKind::from_operation(&tagged), SagaKind::Workflow);
        assert_eq!(SagaKind::Workflow.as_str(), "workflow");
    }

    /// Helper: build an in-memory SQLite SystemStores with the saga
    /// + advisory_leases tables already created.
    #[cfg(feature = "sqlite")]
    async fn fresh_store() -> Arc<dyn SystemStores> {
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        let store = Arc::new(SqliteCanonicalStore::new(pool, "test", "udb_outbox_events"));
        SagaStore::ensure_saga_tables(store.as_ref()).await.unwrap();
        CanonicalStore::ensure_advisory_lease_table(store.as_ref())
            .await
            .unwrap();
        store
    }

    /// Pin: NW1-3c — list_sagas + get_saga + mark_saga_reviewed +
    /// retry_saga_compensation all work end-to-end against the
    /// trait. No PG required.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn admin_helpers_round_trip_against_sqlite_store() {
        let store = fresh_store().await;
        let cfg = SystemCatalogConfig::default();

        // Record two sagas, one indeterminate, one failed_compensation.
        let insert = |tenant: &str, status: SagaStatus| SagaInsert {
            tx_id: Uuid::new_v4().to_string(),
            tenant_id: tenant.to_string(),
            correlation_id: "corr-1".to_string(),
            backend_instance: "primary".to_string(),
            operation: "upsert".to_string(),
            status,
            steps: serde_json::json!([]),
            compensations: serde_json::json!([]),
        };
        let _alpha =
            SagaStore::record_saga(store.as_ref(), &insert("alpha", SagaStatus::Indeterminate))
                .await
                .unwrap();
        let beta = SagaStore::record_saga(
            store.as_ref(),
            &insert("beta", SagaStatus::FailedCompensation),
        )
        .await
        .unwrap();

        // list_sagas with no filter returns both.
        let all = list_sagas(store.as_ref(), &cfg, "", "", "", "", 100, 0)
            .await
            .unwrap();
        assert_eq!(all.len(), 2);

        // Filter by tenant.
        let alpha_only = list_sagas(store.as_ref(), &cfg, "alpha", "", "", "", 100, 0)
            .await
            .unwrap();
        assert_eq!(alpha_only.len(), 1);
        assert_eq!(alpha_only[0].tenant_id, "alpha");

        // Filter by status.
        let fc = list_sagas(
            store.as_ref(),
            &cfg,
            "",
            "failed_compensation",
            "",
            "",
            100,
            0,
        )
        .await
        .unwrap();
        assert_eq!(fc.len(), 1);
        assert_eq!(fc[0].tenant_id, "beta");

        // Invalid status token rejected with InvalidArgument.
        let err = list_sagas(store.as_ref(), &cfg, "", "bogus", "", "", 100, 0)
            .await
            .expect_err("must reject");
        assert_eq!(err.code(), tonic::Code::InvalidArgument);

        // get_saga round-trip.
        let row = get_saga(store.as_ref(), &cfg, &beta.to_string())
            .await
            .unwrap();
        assert_eq!(row.tenant_id, "beta");
        assert_eq!(row.status, "failed_compensation");

        // get_saga on missing → NotFound.
        let phantom = Uuid::new_v4().to_string();
        let err = get_saga(store.as_ref(), &cfg, &phantom)
            .await
            .expect_err("must error");
        assert_eq!(err.code(), tonic::Code::NotFound);
        assert_schema_detail(
            &err,
            "get_saga",
            "saga_not_found",
            &format!("saga {phantom} not found"),
        );

        // retry_saga_compensation succeeds from failed_compensation,
        // refuses from indeterminate.
        retry_saga_compensation(store.as_ref(), &cfg, &beta.to_string())
            .await
            .unwrap();
        let err = retry_saga_compensation(store.as_ref(), &cfg, &beta.to_string())
            .await
            .expect_err("now indeterminate, retry refused");
        assert_policy_detail(
            &err,
            "retry_saga_compensation",
            "saga_not_retryable",
            &format!(
                "saga {beta} is not in a retryable state (failed_compensation or manual_review)"
            ),
        );

        // mark_saga_reviewed flips status.
        mark_saga_reviewed(store.as_ref(), &cfg, &beta.to_string())
            .await
            .unwrap();
        let row = get_saga(store.as_ref(), &cfg, &beta.to_string())
            .await
            .unwrap();
        assert_eq!(row.status, "manual_review");

        // mark_saga_reviewed on missing → NotFound.
        let err = mark_saga_reviewed(store.as_ref(), &cfg, &phantom)
            .await
            .expect_err("must error");
        assert_eq!(err.code(), tonic::Code::NotFound);
        assert_schema_detail(
            &err,
            "mark_saga_reviewed",
            "saga_not_found",
            &format!("saga {phantom} not found"),
        );

        // Bad saga_id format → InvalidArgument.
        let err = get_saga(store.as_ref(), &cfg, "not-a-uuid")
            .await
            .expect_err("must reject");
        assert_eq!(err.code(), tonic::Code::InvalidArgument);
    }

    /// Pin: NW1-3c — SagaRecoveryWorker constructed from an
    /// Arc<dyn SystemStores> runs one recovery pass against an
    /// in-memory SQLite store, acquires + releases the worker lease
    /// portably (no pg_try_advisory_lock).
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn recovery_worker_runs_one_pass_against_sqlite() {
        let store = fresh_store().await;

        // Seed a recoverable saga.
        let _id = SagaStore::record_saga(
            store.as_ref(),
            &SagaInsert {
                tx_id: Uuid::new_v4().to_string(),
                tenant_id: "t1".to_string(),
                correlation_id: "c1".to_string(),
                backend_instance: "primary".to_string(),
                operation: "upsert".to_string(),
                status: SagaStatus::Indeterminate,
                steps: serde_json::json!([]),
                // Empty compensations array → all_compensated returns
                // false → recovery_attempts increments → after 3
                // attempts marked manual_review.
                compensations: serde_json::json!([]),
            },
        )
        .await
        .unwrap();

        let settings = SagaSettings {
            recovery_interval_secs: 1,
            stale_threshold_secs: 1,
            ..Default::default()
        };
        let worker = SagaRecoveryWorker::with_settings(store.clone(), &settings);

        let report = worker.run_once().await;
        assert!(
            report.lease_acquired,
            "lease must be acquired (other workers idle)"
        );
        assert_eq!(report.scanned, 1, "the one indeterminate saga was claimed");
        // No compensators registered → all_compensated == false →
        // recovery_attempts incremented but threshold (3) not yet hit.
        assert_eq!(report.compensated, 0);
    }

    /// Pin: NW1-3c — worker lease contention. Two workers race;
    /// only one wins; after release the other wins.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn worker_lease_serializes_concurrent_workers() {
        let store = fresh_store().await;

        let settings = SagaSettings {
            recovery_interval_secs: 1,
            stale_threshold_secs: 1,
            ..Default::default()
        };
        let worker_a = SagaRecoveryWorker::with_settings(store.clone(), &settings);
        let worker_b = SagaRecoveryWorker::with_settings(store.clone(), &settings);

        // Worker A acquires the lease.
        let mut report_a = SagaRecoveryReport::default();
        let a_acquired = worker_a.try_acquire_worker_lease(&mut report_a).await;
        assert!(a_acquired);

        // Worker B tries — fails.
        let mut report_b = SagaRecoveryReport::default();
        let b_acquired = worker_b.try_acquire_worker_lease(&mut report_b).await;
        assert!(!b_acquired);

        // A releases.
        worker_a.release_worker_lease(&mut report_a).await;
        assert!(report_a.errors.is_empty(), "release must not error");

        // B now wins.
        let mut report_b2 = SagaRecoveryReport::default();
        let b_now = worker_b.try_acquire_worker_lease(&mut report_b2).await;
        assert!(b_now);
    }

    /// Phase 1 HA pin: model two broker replicas against the same SystemStores
    /// lease. While replica A owns the recovery lease, replica B's full
    /// `run_once` must skip without scanning or incrementing recovery attempts.
    /// Once A releases, B may recover the saga exactly once.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn multi_node_saga_recovery_skips_peer_until_lease_release() {
        let store = fresh_store().await;
        let saga_id = SagaStore::record_saga(
            store.as_ref(),
            &SagaInsert {
                tx_id: Uuid::new_v4().to_string(),
                tenant_id: "tenant-a".to_string(),
                correlation_id: "corr-ha".to_string(),
                backend_instance: "primary".to_string(),
                operation: "upsert".to_string(),
                status: SagaStatus::Indeterminate,
                steps: serde_json::json!([]),
                compensations: serde_json::json!([]),
            },
        )
        .await
        .unwrap();

        let settings = SagaSettings {
            recovery_interval_secs: 1,
            stale_threshold_secs: 1,
            worker_lease_ttl_secs: 60,
            ..Default::default()
        };
        let worker_a = SagaRecoveryWorker::with_settings(store.clone(), &settings);
        let mut worker_b = SagaRecoveryWorker::with_settings(store.clone(), &settings);
        // This test exercises failover handoff (the next replica performs the
        // recovery attempt once the lease is released), not the #134 per-saga
        // retry cooldown. The freshly-created saga's `updated_at` is "now", so
        // the default 30s cooldown would otherwise skip it as still-cooling and
        // leave `recovery_attempts` at 0. Zero the cooldown so the handoff is
        // observed deterministically regardless of wall-clock timing.
        worker_b.quarantine_policy.cooldown_secs = 0;

        let mut held = SagaRecoveryReport::default();
        assert!(
            worker_a.try_acquire_worker_lease(&mut held).await,
            "replica A should acquire the shared recovery lease"
        );

        let skipped = worker_b.run_once().await;
        assert!(!skipped.lease_acquired);
        assert_eq!(skipped.scanned, 0);
        let row = SagaStore::get_saga(store.as_ref(), saga_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            row.recovery_attempts, 0,
            "peer replica must not touch saga state while the lease is held"
        );

        worker_a.release_worker_lease(&mut held).await;
        assert!(held.errors.is_empty(), "lease release must be clean");

        let recovered = worker_b.run_once().await;
        assert!(recovered.lease_acquired);
        assert_eq!(recovered.scanned, 1);
        let row = SagaStore::get_saga(store.as_ref(), saga_id)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(
            row.recovery_attempts, 1,
            "after failover, the next replica performs one recovery attempt"
        );
    }
}