fleetforge_storage/

lib.rs

//! Persistent state management: Postgres schema + artifact storage contracts.

use std::collections::HashSet;
use std::convert::TryInto;
use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{anyhow, bail, Context, Result};
use chrono::{DateTime, Utc};
use serde_json::{json, Map, Value};
use sha2::{Digest, Sha256};
use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
use sqlx::types::Json;
use sqlx::{PgConnection, Pool, Postgres, QueryBuilder, Row, Transaction};
use tracing::warn;
use uuid::Uuid;

static MIGRATOR: sqlx::migrate::Migrator = sqlx::migrate!();

use crate::models::NewOutboxEvent;
mod stores;
pub use artifacts::ObjectStoreArtifactStore;
pub use fleetforge_contracts::artifacts::{ArtifactMeta, ArtifactStore};
pub use memory::{InMemoryStorage, RunStore};
use stores::{DynObjectSigner, DynObjectStore};
pub mod memory;
pub mod retention;

pub mod artifacts {
    use std::convert::TryFrom;
    use std::time::Duration;

    use anyhow::{bail, Context, Result};
    use async_trait::async_trait;
    use bytes::Bytes;
    use fleetforge_contracts::artifacts::{ArtifactMeta, ArtifactStore};
    use object_store::path::Path;
    use object_store::PutPayload;
    use reqwest::Method;
    use serde_json::Value;
    use sha2::{Digest, Sha256};
    use url::Url;

    use super::{DynObjectSigner, DynObjectStore};

    /// Default `object_store`-backed implementation using content-addressed paths.
    pub struct ObjectStoreArtifactStore {
        store: DynObjectStore,
        signer: Option<DynObjectSigner>,
    }

    impl ObjectStoreArtifactStore {
        pub fn new(store: DynObjectStore, signer: Option<DynObjectSigner>) -> Self {
            Self { store, signer }
        }

        fn path_for_digest(digest: &[u8; 32]) -> Path {
            let digest_hex = hex::encode(digest);
            let prefix_a = &digest_hex[0..2];
            let prefix_b = &digest_hex[2..4];
            Path::from(format!("sha256/{}/{}/{}", prefix_a, prefix_b, digest_hex))
        }
    }

    #[async_trait]
    impl ArtifactStore for ObjectStoreArtifactStore {
        async fn put(
            &self,
            bytes: Bytes,
            media_type: &str,
            metadata: Value,
        ) -> Result<ArtifactMeta> {
            let mut hasher = Sha256::new();
            hasher.update(&bytes);
            let digest = hasher.finalize();
            let mut sha256 = [0u8; 32];
            sha256.copy_from_slice(&digest);

            let path = Self::path_for_digest(&sha256);
            let size =
                u64::try_from(bytes.len()).context("artifact payload size does not fit in u64")?;

            self.store
                .put(&path, PutPayload::from(bytes))
                .await
                .context("failed to write artifact to object_store")?;

            Ok(ArtifactMeta {
                sha256,
                media_type: media_type.to_string(),
                size,
                metadata,
            })
        }

        async fn get(&self, sha256: [u8; 32]) -> Result<Bytes> {
            let path = Self::path_for_digest(&sha256);
            let result = self
                .store
                .get(&path)
                .await
                .context("failed to fetch artifact from object_store")?;
            result
                .bytes()
                .await
                .context("failed to collect artifact bytes from object_store")
        }

        async fn presign_get(&self, sha256: [u8; 32], ttl: Duration) -> Result<Url> {
            let path = Self::path_for_digest(&sha256);
            let Some(signer) = self.signer.as_ref() else {
                bail!("artifact backend does not support presigned URLs");
            };
            signer
                .signed_url(Method::GET, &path, ttl)
                .await
                .context("failed to create presigned artifact URL from object_store")
        }
    }
}

/// Configuration for storage backends (database + artifacts).
#[derive(Debug, Clone)]
pub struct StorageConfig {
    pub database_url: String,
    pub max_connections: u32,
    pub connect_timeout: Duration,
    pub object_store: ObjectStoreConfig,
}

impl Default for StorageConfig {
    fn default() -> Self {
        let object_store = match ObjectStoreConfig::from_env() {
            Ok(cfg) => cfg,
            Err(err) => {
                eprintln!("warning: falling back to in-memory artifact store: {err}");
                ObjectStoreConfig::InMemory
            }
        };
        Self {
            database_url: std::env::var("DATABASE_URL")
                .unwrap_or_else(|_| "postgres://localhost/fleetforge".to_string()),
            max_connections: 10,
            connect_timeout: Duration::from_secs(5),
            object_store,
        }
    }
}

/// Supported artifact store backends.
#[derive(Debug, Clone)]
pub enum ObjectStoreConfig {
    /// In-memory object store (good for tests and quickstarts).
    InMemory,
    /// Amazon S3-compatible object store (AWS, MinIO, etc.).
    S3(S3Config),
    /// Google Cloud Storage bucket.
    Gcs(GcsConfig),
    /// Azure Blob Storage container.
    Azure(AzureConfig),
    /// Generic HTTP object store.
    Http(HttpConfig),
}

/// Configuration for S3-compatible object store backends.
#[derive(Debug, Clone)]
pub struct S3Config {
    pub endpoint: Option<String>,
    pub access_key: String,
    pub secret_key: String,
    pub region: Option<String>,
    pub bucket: String,
    pub virtual_hosted_style: bool,
}

/// Configuration for Google Cloud Storage.
#[derive(Debug, Clone)]
pub struct GcsConfig {
    pub bucket: String,
    pub service_account: GcsServiceAccount,
}

#[derive(Debug, Clone)]
pub enum GcsServiceAccount {
    File(PathBuf),
    Json(String),
}

/// Configuration for Azure Blob Storage.
#[derive(Debug, Clone)]
pub struct AzureConfig {
    pub account: String,
    pub access_key: String,
    pub container: String,
}

/// Configuration for HTTP-backed object stores.
#[derive(Debug, Clone)]
pub struct HttpConfig {
    pub base_url: String,
}

impl ObjectStoreConfig {
    fn from_env() -> Result<Self> {
        match std::env::var("FLEETFORGE_OBJECT_STORE")
            .ok()
            .as_deref()
            .map(|s| s.to_lowercase())
        {
            Some(ref backend) if backend == "s3" || backend == "minio" => {
                Ok(ObjectStoreConfig::S3(S3Config::from_env()?))
            }
            Some(ref backend) if backend == "gcs" => Ok(ObjectStoreConfig::Gcs(GcsConfig::from_env()?)),
            Some(ref backend) if backend == "azure" => Ok(ObjectStoreConfig::Azure(AzureConfig::from_env()?)),
            Some(ref backend) if backend == "http" => Ok(ObjectStoreConfig::Http(HttpConfig::from_env()?)),
            Some(ref backend) if backend == "in_memory" => Ok(ObjectStoreConfig::InMemory),
            Some(other) => Err(anyhow!("unsupported object store backend '{}'; expected one of in_memory, s3, gcs, azure, http", other)),
            None => {
                if let Ok(cfg) = S3Config::from_minio_env() {
                    return Ok(ObjectStoreConfig::S3(cfg));
                }
                if let Ok(cfg) = S3Config::from_env() {
                    return Ok(ObjectStoreConfig::S3(cfg));
                }
                Ok(ObjectStoreConfig::InMemory)
            }
        }
    }
}

fn first_env(keys: &[&str]) -> Option<String> {
    keys.iter()
        .find_map(|key| std::env::var(key).ok().filter(|value| !value.is_empty()))
}

#[inline]
fn exec<'a>(tx: &'a mut Transaction<'_, Postgres>) -> &'a mut PgConnection {
    tx.as_mut()
}

impl S3Config {
    fn from_env() -> Result<Self> {
        let bucket = first_env(&["FLEETFORGE_S3_BUCKET", "S3_BUCKET", "AWS_S3_BUCKET"])
            .ok_or_else(|| {
                anyhow!("S3 bucket not configured; set FLEETFORGE_S3_BUCKET or S3_BUCKET")
            })?;
        let access_key = first_env(&[
            "FLEETFORGE_S3_ACCESS_KEY",
            "AWS_ACCESS_KEY_ID",
            "MINIO_ACCESS_KEY",
        ])
        .ok_or_else(|| {
            anyhow!(
                "S3 access key not configured; set FLEETFORGE_S3_ACCESS_KEY or AWS_ACCESS_KEY_ID"
            )
        })?;
        let secret_key = first_env(&["FLEETFORGE_S3_SECRET_KEY", "AWS_SECRET_ACCESS_KEY", "MINIO_SECRET_KEY"])
            .ok_or_else(|| anyhow!("S3 secret key not configured; set FLEETFORGE_S3_SECRET_KEY or AWS_SECRET_ACCESS_KEY"))?;
        let endpoint = first_env(&[
            "FLEETFORGE_S3_ENDPOINT",
            "S3_ENDPOINT",
            "AWS_ENDPOINT_URL_S3",
            "MINIO_ENDPOINT",
        ]);
        let region = first_env(&[
            "FLEETFORGE_S3_REGION",
            "S3_REGION",
            "AWS_REGION",
            "MINIO_REGION",
        ]);
        let virtual_hosted_style = endpoint
            .as_deref()
            .map(|ep| !ep.contains("minio"))
            .unwrap_or(true);

        Ok(Self {
            endpoint,
            access_key,
            secret_key,
            region,
            bucket,
            virtual_hosted_style,
        })
    }

    fn from_minio_env() -> Result<Self> {
        let endpoint = std::env::var("MINIO_ENDPOINT").context("MINIO_ENDPOINT not set")?;
        let access_key = std::env::var("MINIO_ACCESS_KEY").context("MINIO_ACCESS_KEY not set")?;
        let secret_key = std::env::var("MINIO_SECRET_KEY").context("MINIO_SECRET_KEY not set")?;
        let bucket = std::env::var("MINIO_BUCKET").unwrap_or_else(|_| "fleetforge".to_string());

        Ok(Self {
            endpoint: Some(endpoint),
            access_key,
            secret_key,
            region: std::env::var("MINIO_REGION").ok(),
            bucket,
            virtual_hosted_style: false,
        })
    }
}

impl GcsConfig {
    fn from_env() -> Result<Self> {
        let bucket = first_env(&["FLEETFORGE_GCS_BUCKET", "GCS_BUCKET"]).ok_or_else(|| {
            anyhow!("GCS bucket not configured; set FLEETFORGE_GCS_BUCKET or GCS_BUCKET")
        })?;
        let service_account = if let Some(json) =
            first_env(&["GCS_SERVICE_ACCOUNT_JSON", "GOOGLE_SERVICE_ACCOUNT_JSON"])
        {
            GcsServiceAccount::Json(json)
        } else if let Some(path) = first_env(&["GOOGLE_APPLICATION_CREDENTIALS"]) {
            GcsServiceAccount::File(PathBuf::from(path))
        } else {
            return Err(anyhow!(
                "GCS credentials not configured; set GCS_SERVICE_ACCOUNT_JSON or GOOGLE_APPLICATION_CREDENTIALS"
            ));
        };

        Ok(Self {
            bucket,
            service_account,
        })
    }
}

impl AzureConfig {
    fn from_env() -> Result<Self> {
        let account = first_env(&["FLEETFORGE_AZURE_STORAGE_ACCOUNT", "AZURE_STORAGE_ACCOUNT"])
            .ok_or_else(|| anyhow!("Azure storage account not configured; set FLEETFORGE_AZURE_STORAGE_ACCOUNT or AZURE_STORAGE_ACCOUNT"))?;
        let access_key = first_env(&["FLEETFORGE_AZURE_STORAGE_KEY", "AZURE_STORAGE_KEY"])
            .ok_or_else(|| anyhow!("Azure storage key not configured; set FLEETFORGE_AZURE_STORAGE_KEY or AZURE_STORAGE_KEY"))?;
        let container = first_env(&["FLEETFORGE_AZURE_CONTAINER", "AZURE_STORAGE_CONTAINER", "AZURE_CONTAINER"])
            .ok_or_else(|| anyhow!("Azure container not configured; set FLEETFORGE_AZURE_CONTAINER or AZURE_STORAGE_CONTAINER"))?;

        Ok(Self {
            account,
            access_key,
            container,
        })
    }
}

impl HttpConfig {
    fn from_env() -> Result<Self> {
        let base_url = first_env(&["FLEETFORGE_HTTP_OBJECT_STORE", "HTTP_OBJECT_STORE_BASE_URL"])
            .ok_or_else(|| anyhow!("HTTP object store base URL not configured; set FLEETFORGE_HTTP_OBJECT_STORE or HTTP_OBJECT_STORE_BASE_URL"))?;
        Ok(Self { base_url })
    }
}

impl Default for ObjectStoreConfig {
    fn default() -> Self {
        ObjectStoreConfig::InMemory
    }
}

/// High-level storage handle that encapsulates database and artifact store clients.
#[derive(Clone)]
pub struct Storage {
    pool: Pool<Postgres>,
    objects: DynObjectStore,
    artifact_store: Arc<dyn ArtifactStore>,
}

/// Outcome of attempting to consume budget prior to executing a step.
#[derive(Debug)]
pub enum BudgetResult {
    Granted(models::Budget),
    Denied,
    NoBudgetConfigured,
}

/// Summary of retention actions applied to persisted data.
#[derive(Debug, Clone, Copy, Default)]
pub struct RetentionStats {
    pub runs_scrubbed: u64,
    pub steps_scrubbed: u64,
    pub events_scrubbed: u64,
}

impl Storage {
    /// Connects to Postgres, applies migrations, and provisions the artifact store.
    pub async fn connect(config: StorageConfig) -> Result<Self> {
        let connect_options = PgConnectOptions::from_str(&config.database_url)
            .context("invalid postgres connection string")?;

        let pool = PgPoolOptions::new()
            .max_connections(config.max_connections)
            .acquire_timeout(config.connect_timeout)
            .connect_with(connect_options)
            .await?;

        MIGRATOR.run(&pool).await?;

        let (objects, signer) = stores::build(&config.object_store)?;

        let artifact_store: Arc<dyn ArtifactStore> = Arc::new(
            artifacts::ObjectStoreArtifactStore::new(objects.clone(), signer.clone()),
        );

        Ok(Self {
            pool,
            objects,
            artifact_store,
        })
    }

    /// Convenience helper for default configuration (env-driven DATABASE_URL).
    pub async fn connect_default() -> Result<Self> {
        Self::connect(StorageConfig::default()).await
    }

    /// Returns a borrowed connection pool for ad-hoc queries.
    pub fn pool(&self) -> &Pool<Postgres> {
        &self.pool
    }

    /// Returns a handle to the configured object store.
    pub fn object_store(&self) -> DynObjectStore {
        self.objects.clone()
    }

    /// Returns the artifact store abstraction.
    pub fn artifacts(&self) -> Arc<dyn ArtifactStore> {
        self.artifact_store.clone()
    }

    /// Releases a leased step back to the queue without incrementing attempt counters.
    pub async fn return_step_to_queue(&self, run_id: Uuid, step_id: Uuid) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set status = 'queued',
                leased_by = null,
                lease_expires_at = null
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "step {} for run {} not found when returning to queue",
                step_id,
                run_id
            );
        }

        Ok(())
    }

    /// Requeues expired leases back to the queued state and returns affected step IDs.
    pub async fn requeue_expired_leases(&self, limit: i64) -> Result<Vec<(Uuid, Uuid)>> {
        let rows = sqlx::query(
            r#"
            with expired as (
                select run_id, step_id
                from steps
                where status = 'leased'
                  and lease_expires_at is not null
                  and lease_expires_at < now()
                order by lease_expires_at asc
                limit $1
            )
            update steps s
            set status = 'queued',
                leased_by = null,
                lease_expires_at = null,
                not_before = null
            from expired
            where s.run_id = expired.run_id
              and s.step_id = expired.step_id
            returning s.run_id, s.step_id
            "#,
        )
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        let mut requeued = Vec::with_capacity(rows.len());
        for row in rows {
            let run_id: Uuid = row.get("run_id");
            let step_id: Uuid = row.get("step_id");
            requeued.push((run_id, step_id));
        }
        Ok(requeued)
    }

    pub async fn defer_step_to_queue(
        &self,
        run_id: Uuid,
        step_id: Uuid,
        not_before: DateTime<Utc>,
        payload: Value,
    ) -> Result<()> {
        let mut tx = self.pool.begin().await?;
        self.defer_step_to_queue_tx(&mut tx, run_id, step_id, not_before, payload)
            .await?;
        tx.commit().await?;
        Ok(())
    }

    pub async fn defer_step_to_queue_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
        not_before: DateTime<Utc>,
        payload: Value,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set status = 'queued',
                leased_by = null,
                lease_expires_at = null,
                not_before = $3
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(not_before)
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to defer step {} in run {} back to queue",
                step_id,
                run_id
            );
        }

        self.enqueue_outbox_event(
            tx,
            NewOutboxEvent {
                run_id,
                step_id: Some(step_id),
                kind: "step_deferred".to_string(),
                payload,
            },
        )
        .await?;

        Ok(())
    }

    pub async fn schedule_retry_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
        attempt: i32,
        not_before: DateTime<Utc>,
        error_json: Value,
    ) -> Result<()> {
        let error_json = Json(error_json);
        let result = sqlx::query(
            r#"
            update steps
            set status = 'queued',
                attempt = $3,
                leased_by = null,
                lease_expires_at = null,
                not_before = $4,
                error_json = $5
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(attempt)
        .bind(not_before)
        .bind(error_json)
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to schedule retry for step {} in run {}",
                step_id,
                run_id
            );
        }

        Ok(())
    }

    pub async fn insert_cost_ledger_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
        attempt: i32,
        status: &str,
        reserved_tokens: i64,
        reserved_cost: f64,
        actual_tokens: Option<i64>,
        actual_cost: Option<f64>,
    ) -> Result<()> {
        sqlx::query(
            r#"
            insert into step_cost_ledger (
                run_id, step_id, attempt, status, reserved_tokens, reserved_cost, actual_tokens, actual_cost
            ) values ($1, $2, $3, $4, $5, $6, $7, $8)
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(attempt)
        .bind(status)
        .bind(reserved_tokens)
        .bind(reserved_cost)
        .bind(actual_tokens)
        .bind(actual_cost)
        .execute(exec(tx))
        .await?;

        Ok(())
    }

    pub async fn insert_step_attempt_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        attempt: models::NewStepAttempt,
    ) -> Result<()> {
        sqlx::query(
            r#"
            insert into step_attempts (
                run_id,
                step_id,
                attempt,
                status,
                latency_ms,
                spec_snapshot,
                input_snapshot,
                output_snapshot,
                error_snapshot,
                checkpoint,
                budget_snapshot,
                guardrail_events,
                tool_events,
                artifacts
            ) values (
                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14
            )
            on conflict (run_id, step_id, attempt) do nothing
            "#,
        )
        .bind(attempt.run_id)
        .bind(attempt.step_id)
        .bind(attempt.attempt)
        .bind(attempt.status.as_str())
        .bind(attempt.latency_ms)
        .bind(Json(attempt.spec_snapshot))
        .bind(attempt.input_snapshot.map(Json))
        .bind(attempt.output_snapshot.map(Json))
        .bind(attempt.error_snapshot.map(Json))
        .bind(attempt.checkpoint.map(Json))
        .bind(attempt.budget_snapshot.map(Json))
        .bind(attempt.guardrail_events.map(Json))
        .bind(attempt.tool_events.map(Json))
        .bind(attempt.artifacts.map(Json))
        .execute(exec(tx))
        .await?;

        Ok(())
    }

    pub async fn append_run_event_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Option<Uuid>,
        kind: &str,
        payload: &Value,
    ) -> Result<()> {
        sqlx::query(
            r#"
            insert into run_event_log (run_id, step_id, kind, payload)
            values ($1, $2, $3, $4)
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(kind)
        .bind(Json(payload.clone()))
        .execute(exec(tx))
        .await?;

        Ok(())
    }

    pub async fn trigger_compensation_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        origin_step: Uuid,
        compensation_step: Uuid,
    ) -> Result<()> {
        sqlx::query(
            r#"
            update steps
            set compensation_scheduled = true
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(origin_step)
        .execute(exec(tx))
        .await?;

        let result = sqlx::query(
            r#"
            update steps
            set status = 'queued',
                pending_dependencies = 0,
                leased_by = null,
                lease_expires_at = null,
                not_before = null
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(compensation_step)
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "compensation step {} not present in run {}",
                compensation_step,
                run_id
            );
        }

        Ok(())
    }

    /// Returns unpublished outbox events up to `limit`, ordered by ID.
    pub async fn fetch_unpublished_outbox(&self, limit: i64) -> Result<Vec<models::OutboxEvent>> {
        let rows = sqlx::query_as::<_, db::OutboxRow>(
            r#"
            select *
            from outbox
            where published_at is null
            order by id asc
            limit $1
            "#,
        )
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(models::OutboxEvent::try_from)
            .collect()
    }

    /// Marks an outbox row as published by setting `published_at` to the current timestamp.
    pub async fn mark_outbox_published(&self, id: i64) -> Result<()> {
        let result = sqlx::query(
            r#"
            update outbox
            set published_at = now()
            where id = $1
              and published_at is null
            "#,
        )
        .bind(id)
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!("outbox row {} not found", id);
        }

        Ok(())
    }

    /// Returns outbox events for a run that were created after the given id.
    pub async fn fetch_outbox_for_run_since(
        &self,
        run_id: Uuid,
        after_id: i64,
        limit: i64,
    ) -> Result<Vec<models::OutboxEvent>> {
        let rows = sqlx::query_as::<_, db::OutboxRow>(
            r#"
            select *
            from outbox
            where run_id = $1
              and id > $2
            order by id asc
            limit $3
            "#,
        )
        .bind(run_id)
        .bind(after_id)
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(models::OutboxEvent::try_from)
            .collect()
    }

    /// Returns the most recent outbox event for a run, if any.
    pub async fn fetch_last_outbox_event(
        &self,
        run_id: Uuid,
    ) -> Result<Option<models::OutboxEvent>> {
        let row = sqlx::query_as::<_, db::OutboxRow>(
            r#"
            select *
            from outbox
            where run_id = $1
            order by id desc
            limit 1
            "#,
        )
        .bind(run_id)
        .fetch_optional(&self.pool)
        .await?;

        match row {
            Some(row) => Ok(Some(models::OutboxEvent::try_from(row)?)),
            None => Ok(None),
        }
    }

    /// Fetch the last acknowledged offset for a subscription cursor, if present.
    pub async fn fetch_tap_offset(&self, run_id: Uuid, cursor_id: &str) -> Result<Option<i64>> {
        let offset = sqlx::query_scalar::<_, i64>(
            r#"
            select cursor_offset
            from tap_offsets
            where run_id = $1
              and cursor_id = $2
            "#,
        )
        .bind(run_id)
        .bind(cursor_id)
        .fetch_optional(&self.pool)
        .await?;

        Ok(offset)
    }

    /// Advance the stored offset for a subscription cursor.
    pub async fn upsert_tap_offset(
        &self,
        run_id: Uuid,
        cursor_id: &str,
        offset: i64,
    ) -> Result<()> {
        sqlx::query(
            r#"
            insert into tap_offsets (run_id, cursor_id, cursor_offset)
            values ($1, $2, $3)
            on conflict (run_id, cursor_id) do update
            set cursor_offset = greatest(tap_offsets.cursor_offset, excluded.cursor_offset),
                updated_at = now()
            "#,
        )
        .bind(run_id)
        .bind(cursor_id)
        .bind(offset)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    /// Applies retention policy by scrubbing input/output payloads for stale runs.
    pub async fn apply_retention_policy(&self, cutoff: DateTime<Utc>) -> Result<RetentionStats> {
        if cutoff.timestamp_millis() <= 0 {
            return Ok(RetentionStats::default());
        }

        let mut tx = self.pool.begin().await?;

        let scrubbed_runs = sqlx::query(
            r#"
            update runs
            set input_ctx = jsonb_set(
                jsonb_set(
                    input_ctx,
                    '{inputs}',
                    '{}'::jsonb,
                    true
                ),
                '{labels}',
                '{}'::jsonb,
                true
            )
            where created_at < $1
              and (
                jsonb_typeof(input_ctx -> 'inputs') in ('object', 'array', 'string')
                or jsonb_typeof(input_ctx -> 'labels') in ('object', 'array', 'string')
              )
            "#,
        )
        .bind(cutoff)
        .execute(exec(&mut tx))
        .await?
        .rows_affected();

        let scrubbed_steps = sqlx::query(
            r#"
            update steps
            set input_snapshot = null,
                output_snapshot = null,
                output_json = null,
                error_json = null
            where created_at < $1
              and (
                input_snapshot is not null
                or output_snapshot is not null
                or output_json is not null
                or error_json is not null
            )
            "#,
        )
        .bind(cutoff)
        .execute(exec(&mut tx))
        .await?
        .rows_affected();

        let scrubbed_events = sqlx::query(
            r#"
            update outbox
            set payload = payload - 'snapshots' - 'input' - 'output'
            where run_id in (
                select run_id from runs where created_at < $1
            )
              and (
                payload ? 'snapshots'
                or payload ? 'input'
                or payload ? 'output'
              )
            "#,
        )
        .bind(cutoff)
        .execute(exec(&mut tx))
        .await?
        .rows_affected();

        tx.commit().await?;

        Ok(RetentionStats {
            runs_scrubbed: scrubbed_runs as u64,
            steps_scrubbed: scrubbed_steps as u64,
            events_scrubbed: scrubbed_events as u64,
        })
    }

    /// Attempts to consume the requested budget; returns granted totals or denial.
    pub async fn try_consume_budget(
        &self,
        run_id: Uuid,
        tokens: i64,
        cost: f64,
    ) -> Result<BudgetResult> {
        if tokens == 0 && cost == 0.0 {
            return match self.fetch_budget(run_id).await? {
                Some(budget) => Ok(BudgetResult::Granted(budget)),
                None => Ok(BudgetResult::NoBudgetConfigured),
            };
        }

        let row = sqlx::query_as::<_, db::BudgetRow>(
            r#"
            update budgets
            set tokens_used = tokens_used + $2,
                cost_used = cost_used + $3,
                updated_at = now()
            where run_id = $1
              and (tokens_limit is null or tokens_used + $2 <= tokens_limit)
              and (cost_limit is null or cost_used + $3 <= cost_limit)
            returning *
            "#,
        )
        .bind(run_id)
        .bind(tokens)
        .bind(cost)
        .fetch_optional(&self.pool)
        .await?;

        if let Some(row) = row {
            return Ok(BudgetResult::Granted(row.try_into()?));
        }

        let exists = sqlx::query_scalar::<_, i64>("select 1 from budgets where run_id = $1")
            .bind(run_id)
            .fetch_optional(&self.pool)
            .await?;

        if exists.is_some() {
            Ok(BudgetResult::Denied)
        } else {
            Ok(BudgetResult::NoBudgetConfigured)
        }
    }

    /// Fetches the current budget totals for a run, if configured.
    pub async fn fetch_budget(&self, run_id: Uuid) -> Result<Option<models::Budget>> {
        let row = sqlx::query_as::<_, db::BudgetRow>("select * from budgets where run_id = $1")
            .bind(run_id)
            .fetch_optional(&self.pool)
            .await?;

        match row {
            Some(row) => Ok(Some(row.try_into()?)),
            None => Ok(None),
        }
    }

    /// Returns all step rows for a run ordered by index.
    pub async fn fetch_steps_for_run(&self, run_id: Uuid) -> Result<Vec<models::Step>> {
        let rows = sqlx::query_as::<_, db::StepRow>(
            r#"
            select *
            from steps
            where run_id = $1
            order by idx
            "#,
        )
        .bind(run_id)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(models::Step::try_from).collect()
    }

    /// Marks a leased step as running for the provided worker.
    pub async fn mark_step_running(
        &self,
        run_id: Uuid,
        step_id: Uuid,
        worker_id: &str,
        input_snapshot: Option<&Value>,
        provider: Option<&str>,
        provider_version: Option<&str>,
    ) -> Result<()> {
        let input_snapshot = input_snapshot.map(Json);
        let result = sqlx::query(
            r#"
            update steps
            set status = 'running',
                attempt = attempt + 1,
                input_snapshot = coalesce($4, input_snapshot),
                provider = coalesce($5, provider),
                provider_version = coalesce($6, provider_version),
                not_before = null
            where run_id = $1
              and step_id = $2
              and leased_by = $3
              and status = 'leased'
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(worker_id)
        .bind(input_snapshot)
        .bind(provider)
        .bind(provider_version)
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "step {} for run {} is not leased by worker {}; cannot mark running",
                step_id,
                run_id,
                worker_id
            );
        }

        sqlx::query(
            r#"
            update runs
            set status = 'running'
            where run_id = $1
              and status = 'pending'
            "#,
        )
        .bind(run_id)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    /// Updates the stored budget estimate for a step.
    pub async fn update_step_estimate(
        &self,
        run_id: Uuid,
        step_id: Uuid,
        estimated_tokens: i64,
        estimated_cost: f64,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set estimated_tokens = $3,
                estimated_cost = $4
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(estimated_tokens)
        .bind(estimated_cost)
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to record budget estimate for step {} in run {}",
                step_id,
                run_id
            );
        }

        Ok(())
    }

    /// Updates the status, snapshot metadata, and output fields for a step.
    pub async fn update_step_status_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
        status: models::StepStatus,
        attempt: Option<i32>,
        output_json: Option<Value>,
        error_json: Option<Value>,
        input_snapshot: Option<Value>,
        output_snapshot: Option<Value>,
        checkpoint: Option<Value>,
        provider: Option<&str>,
        provider_version: Option<&str>,
    ) -> Result<()> {
        let output_json = output_json.map(Json);
        let error_json = error_json.map(Json);
        let input_snapshot = input_snapshot.map(Json);
        let output_snapshot = output_snapshot.map(Json);
        let checkpoint = checkpoint.map(Json);

        let result = sqlx::query(
            r#"
            update steps
            set status = $1,
                attempt = coalesce($2, attempt),
                output_json = $3,
                error_json = $4,
                input_snapshot = coalesce($5, input_snapshot),
                output_snapshot = coalesce($6, output_snapshot),
                checkpoint = coalesce($7, checkpoint),
                provider = coalesce($8, provider),
                provider_version = coalesce($9, provider_version),
                leased_by = case
                    when $1 in ('succeeded','failed','skipped') then null
                    else leased_by
                end,
                lease_expires_at = case
                    when $1 in ('succeeded','failed','skipped') then null
                    else lease_expires_at
                end
            where run_id = $10
              and step_id = $11
            "#,
        )
        .bind(status.as_str())
        .bind(attempt)
        .bind(output_json)
        .bind(error_json)
        .bind(input_snapshot)
        .bind(output_snapshot)
        .bind(checkpoint)
        .bind(provider)
        .bind(provider_version)
        .bind(run_id)
        .bind(step_id)
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to update status for step {} in run {}",
                step_id,
                run_id
            );
        }

        Ok(())
    }

    pub async fn fetch_step_attempts_for_run(
        &self,
        run_id: Uuid,
    ) -> Result<Vec<models::StepAttempt>> {
        let rows = sqlx::query_as::<_, db::StepAttemptRow>(
            r#"
            select
                run_id,
                step_id,
                attempt,
                status,
                recorded_at,
                latency_ms,
                spec_snapshot,
                input_snapshot,
                output_snapshot,
                error_snapshot,
                checkpoint,
                budget_snapshot,
                guardrail_events,
                tool_events,
                artifacts
            from step_attempts
            where run_id = $1
            order by step_id, attempt
            "#,
        )
        .bind(run_id)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(models::StepAttempt::try_from)
            .collect()
    }

    /// Persists the actual resource usage recorded for a step.
    pub async fn update_step_usage_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
        actual_tokens: Option<i64>,
        actual_cost: Option<f64>,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set actual_tokens = $3,
                actual_cost = $4
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(actual_tokens)
        .bind(actual_cost)
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to update usage for step {} in run {}",
                step_id,
                run_id
            );
        }

        Ok(())
    }

    /// Adjusts budget totals based on the difference between reserved and actual usage.
    pub async fn reconcile_budget_usage_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        tokens_delta: i64,
        cost_delta: f64,
    ) -> Result<()> {
        if tokens_delta == 0 && cost_delta.abs() < f64::EPSILON {
            return Ok(());
        }

        let result = sqlx::query(
            r#"
            update budgets
            set tokens_used = tokens_used + $2,
                cost_used = cost_used + $3,
                updated_at = now()
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .bind(tokens_delta)
        .bind(cost_delta)
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            warn!(run = %run_id, "reconcile_budget_usage_tx called but no budget row updated");
        }

        Ok(())
    }

    /// Inserts a run row with deterministic state materialization.
    pub async fn insert_run(&self, run: models::NewRun) -> Result<Uuid> {
        self.insert_run_with_steps(run, &[], &[]).await
    }

    /// Inserts a run and its associated steps within a single transaction.
    pub async fn insert_run_with_steps(
        &self,
        run: models::NewRun,
        steps: &[models::NewStep],
        edges: &[(Uuid, Uuid)],
    ) -> Result<Uuid> {
        let mut tx = self.pool.begin().await?;

        let models::NewRun {
            run_id,
            status,
            dag_json,
            input_ctx,
            seed,
            idempotency_key,
        } = run;

        let (actual_run_id, inserted_new) = if let Some(ref key) = idempotency_key {
            let inserted = sqlx::query_scalar::<_, Uuid>(
                r#"
                insert into runs (run_id, status, dag_json, input_ctx, seed, idempotency_key)
                values ($1, $2, $3, $4, $5, $6)
                on conflict (idempotency_key) do nothing
                returning run_id
                "#,
            )
            .bind(run_id)
            .bind(status.as_str())
            .bind(Json(dag_json.clone()))
            .bind(Json(input_ctx.clone()))
            .bind(seed)
            .bind(key)
            .fetch_optional(exec(&mut tx))
            .await?;
            if let Some(id) = inserted {
                (id, true)
            } else {
                let existing = sqlx::query_scalar::<_, Uuid>(
                    "select run_id from runs where idempotency_key = $1",
                )
                .bind(key)
                .fetch_one(exec(&mut tx))
                .await?;
                (existing, false)
            }
        } else {
            let inserted = sqlx::query_scalar::<_, Uuid>(
                r#"
                insert into runs (run_id, status, dag_json, input_ctx, seed, idempotency_key)
                values ($1, $2, $3, $4, $5, $6)
                on conflict (run_id) do nothing
                returning run_id
                "#,
            )
            .bind(run_id)
            .bind(status.as_str())
            .bind(Json(dag_json.clone()))
            .bind(Json(input_ctx.clone()))
            .bind(seed)
            .bind(idempotency_key)
            .fetch_optional(exec(&mut tx))
            .await?;
            (inserted.unwrap_or(run_id), inserted.is_some())
        };

        if !inserted_new {
            tx.rollback().await?;
            return Ok(actual_run_id);
        }

        for step in steps {
            sqlx::query(
                r#"
                insert into steps (
                    run_id,
                    step_id,
                    idx,
                    priority,
                    spec_json,
                    status,
                    attempt,
                    input_snapshot,
                    provider,
                    provider_version,
                    max_attempts,
                    retry_backoff_ms,
                    retry_backoff_factor,
                    not_before,
                    deadline_at,
                    pending_dependencies,
                    total_dependencies,
                    estimated_tokens,
                    estimated_cost,
                    actual_tokens,
                    actual_cost,
                    checkpoint,
                    compensation_step,
                    compensation_scheduled
                )
                values (
                    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
                    $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
                    $21, $22, $23, $24
                )
                on conflict (run_id, step_id) do update
                set idx = excluded.idx,
                    priority = excluded.priority,
                    spec_json = excluded.spec_json,
                    status = excluded.status,
                    attempt = excluded.attempt,
                    input_snapshot = excluded.input_snapshot,
                    provider = excluded.provider,
                    provider_version = excluded.provider_version,
                    max_attempts = excluded.max_attempts,
                    retry_backoff_ms = excluded.retry_backoff_ms,
                    retry_backoff_factor = excluded.retry_backoff_factor,
                    not_before = excluded.not_before,
                    deadline_at = excluded.deadline_at,
                    pending_dependencies = excluded.pending_dependencies,
                    total_dependencies = excluded.total_dependencies,
                    estimated_tokens = excluded.estimated_tokens,
                    estimated_cost = excluded.estimated_cost,
                    actual_tokens = excluded.actual_tokens,
                    actual_cost = excluded.actual_cost,
                    checkpoint = excluded.checkpoint,
                    compensation_step = excluded.compensation_step,
                    compensation_scheduled = excluded.compensation_scheduled
                "#,
            )
            .bind(step.run_id)
            .bind(step.step_id)
            .bind(step.idx)
            .bind(step.priority)
            .bind(Json(step.spec_json.clone()))
            .bind(step.status.as_str())
            .bind(step.attempt)
            .bind(step.input_snapshot.clone().map(Json))
            .bind(step.provider.clone())
            .bind(step.provider_version.clone())
            .bind(step.max_attempts)
            .bind(step.retry_backoff_ms)
            .bind(step.retry_backoff_factor)
            .bind(step.not_before)
            .bind(step.deadline_at)
            .bind(step.pending_dependencies)
            .bind(step.total_dependencies)
            .bind(step.estimated_tokens)
            .bind(step.estimated_cost)
            .bind(step.actual_tokens)
            .bind(step.actual_cost)
            .bind(step.checkpoint.clone().map(Json))
            .bind(step.compensation_step)
            .bind(step.compensation_scheduled)
            .execute(exec(&mut tx))
            .await?;
        }

        if !edges.is_empty() {
            let mut builder = QueryBuilder::<Postgres>::new(
                "insert into step_edges (run_id, from_step_id, to_step_id) ",
            );
            builder.push_values(edges, |mut b, (from, to)| {
                b.push_bind(actual_run_id).push_bind(from).push_bind(to);
            });
            builder.push(" on conflict do nothing");
            builder.build().execute(exec(&mut tx)).await?;
        }

        tx.commit().await?;

        Ok(actual_run_id)
    }

    /// Decrements dependency counters for successors and queues newly-ready steps.
    pub async fn unlock_successors_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
    ) -> Result<Vec<Uuid>> {
        let rows = sqlx::query(
            r#"
            with affected as (
                select to_step_id
                from step_edges
                where run_id = $1
                  and from_step_id = $2
            ),
            decremented as (
                update steps s
                set pending_dependencies = greatest(s.pending_dependencies - 1, 0)
                from affected
                where s.run_id = $1
                  and s.step_id = affected.to_step_id
                returning s.step_id, s.pending_dependencies
            ),
            queued as (
                update steps s
                set status = 'queued'
                from decremented
                where s.run_id = $1
                  and s.step_id = decremented.step_id
                  and decremented.pending_dependencies = 0
                  and s.status = 'blocked'
                returning s.step_id
            )
            select step_id from queued
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .fetch_all(exec(tx))
        .await?;

        let mut ready = Vec::with_capacity(rows.len());
        for row in rows {
            ready.push(row.get::<Uuid, _>("step_id"));
        }
        Ok(ready)
    }

    /// Marks all downstream blocked steps as skipped when a dependency fails.
    pub async fn skip_downstream_steps_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        failed_step_id: Uuid,
    ) -> Result<Vec<Uuid>> {
        let rows = sqlx::query(
            r#"
            with recursive downstream(step_id) as (
                select to_step_id
                from step_edges
                where run_id = $1
                  and from_step_id = $2
                union
                select se.to_step_id
                from step_edges se
                join downstream d on se.from_step_id = d.step_id
                where se.run_id = $1
            ),
            updated as (
                update steps s
                set status = 'skipped',
                    pending_dependencies = 0
                from downstream
                where s.run_id = $1
                  and s.step_id = downstream.step_id
                  and s.status = 'blocked'
                returning s.step_id
            )
            select step_id from updated
            "#,
        )
        .bind(run_id)
        .bind(failed_step_id)
        .fetch_all(exec(tx))
        .await?;

        let mut skipped = Vec::with_capacity(rows.len());
        for row in rows {
            skipped.push(row.get::<Uuid, _>("step_id"));
        }
        Ok(skipped)
    }

    /// Recomputes the run status based on current step outcomes.
    pub async fn refresh_run_status_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
    ) -> Result<(models::RunStatus, models::RunStatus)> {
        let prev_row = sqlx::query(
            r#"
            select status
            from runs
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .fetch_one(exec(tx))
        .await?;

        let row = sqlx::query(
            r#"
            update runs
            set status = case
                when exists (
                    select 1
                    from steps
                    where run_id = $1
                      and status = 'failed'
                ) then 'failed'
                when not exists (
                    select 1
                    from steps
                    where run_id = $1
                      and status in ('queued','blocked','leased','running')
                ) then 'succeeded'
                else status
            end
            where run_id = $1
            returning status
            "#,
        )
        .bind(run_id)
        .fetch_one(exec(tx))
        .await?;

        let previous: String = prev_row.get("status");
        let current: String = row.get("status");

        Ok((
            models::RunStatus::try_from_str(&previous)?,
            models::RunStatus::try_from_str(&current)?,
        ))
    }

    /// Inserts a step row tied to a run's DAG.
    pub async fn insert_step(&self, step: models::NewStep) -> Result<()> {
        let models::NewStep {
            run_id,
            step_id,
            idx,
            priority,
            spec_json,
            status,
            attempt,
            input_snapshot,
            provider,
            provider_version,
            max_attempts,
            retry_backoff_ms,
            retry_backoff_factor,
            not_before,
            deadline_at,
            pending_dependencies,
            total_dependencies,
            estimated_tokens,
            estimated_cost,
            actual_tokens,
            actual_cost,
            checkpoint,
            compensation_step,
            compensation_scheduled,
        } = step;

        sqlx::query(
            r#"
            insert into steps (
                run_id,
                step_id,
                idx,
                priority,
                spec_json,
                status,
                attempt,
                input_snapshot,
                provider,
                provider_version,
                max_attempts,
                retry_backoff_ms,
                retry_backoff_factor,
                not_before,
                deadline_at,
                pending_dependencies,
                total_dependencies,
                estimated_tokens,
                estimated_cost,
                actual_tokens,
                actual_cost,
                checkpoint,
                compensation_step,
                compensation_scheduled
            )
            values (
                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10,
                $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
                $21, $22, $23, $24
            )
            on conflict (run_id, step_id) do update
            set idx = excluded.idx,
                priority = excluded.priority,
                spec_json = excluded.spec_json,
                status = excluded.status,
                attempt = excluded.attempt,
                input_snapshot = excluded.input_snapshot,
                provider = excluded.provider,
                provider_version = excluded.provider_version,
                max_attempts = excluded.max_attempts,
                retry_backoff_ms = excluded.retry_backoff_ms,
                retry_backoff_factor = excluded.retry_backoff_factor,
                not_before = excluded.not_before,
                deadline_at = excluded.deadline_at,
                pending_dependencies = excluded.pending_dependencies,
                total_dependencies = excluded.total_dependencies,
                estimated_tokens = excluded.estimated_tokens,
                estimated_cost = excluded.estimated_cost,
                actual_tokens = excluded.actual_tokens,
                actual_cost = excluded.actual_cost,
                checkpoint = excluded.checkpoint,
                compensation_step = excluded.compensation_step,
                compensation_scheduled = excluded.compensation_scheduled
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(idx)
        .bind(priority)
        .bind(Json(spec_json))
        .bind(status.as_str())
        .bind(attempt)
        .bind(input_snapshot.map(Json))
        .bind(provider)
        .bind(provider_version)
        .bind(max_attempts)
        .bind(retry_backoff_ms)
        .bind(retry_backoff_factor)
        .bind(not_before)
        .bind(deadline_at)
        .bind(pending_dependencies)
        .bind(total_dependencies)
        .bind(estimated_tokens)
        .bind(estimated_cost)
        .bind(actual_tokens)
        .bind(actual_cost)
        .bind(checkpoint.map(Json))
        .bind(compensation_step)
        .bind(compensation_scheduled)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    /// Fetches a single run record by ID.
    pub async fn fetch_run(&self, run_id: Uuid) -> Result<Option<models::Run>> {
        let row = sqlx::query_as::<_, db::RunRow>(
            r#"
            select *
            from runs
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .fetch_optional(&self.pool)
        .await?;

        match row {
            Some(row) => Ok(Some(row.try_into()?)),
            None => Ok(None),
        }
    }

    pub async fn update_run_input_ctx(&self, run_id: Uuid, input_ctx: Value) -> Result<()> {
        let result = sqlx::query(
            r#"
            update runs
            set input_ctx = $2
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .bind(Json(input_ctx))
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!("run {} not found", run_id);
        }

        Ok(())
    }

    /// Updates the transparency receipt embedded in a run-level artifact entry.
    pub async fn update_run_artifact_transparency(
        &self,
        run_id: Uuid,
        artifact_key: &str,
        transparency: Value,
    ) -> Result<()> {
        let Some(run) = self.fetch_run(run_id).await? else {
            return Ok(());
        };
        let mut ctx_map = match run.input_ctx {
            Value::Object(map) => map,
            other => {
                let mut map = Map::new();
                map.insert("inputs".to_string(), other);
                map
            }
        };
        let artifacts_value = ctx_map
            .entry("artifacts".to_string())
            .or_insert_with(|| Value::Object(Map::new()));
        let Value::Object(artifacts_map) = artifacts_value else {
            return Ok(());
        };
        if let Some(entry) = artifacts_map.get_mut(artifact_key) {
            if let Value::Object(ref mut artifact_obj) = entry {
                artifact_obj.insert("transparency".to_string(), transparency);
            } else {
                let mut object = Map::new();
                object.insert("transparency".to_string(), transparency);
                artifacts_map.insert(artifact_key.to_string(), Value::Object(object));
            }
            self.update_run_input_ctx(run_id, Value::Object(ctx_map))
                .await
        } else {
            Ok(())
        }
    }

    /// Updates the transparency receipt for a step-level artifact stored in `steps.output_json`.
    pub async fn update_step_artifact_transparency(
        &self,
        run_id: Uuid,
        step_id: Uuid,
        artifact_key: &str,
        transparency: Value,
    ) -> Result<()> {
        let row = sqlx::query(
            r#"
            select output_json
            from steps
            where run_id = $1 and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .fetch_optional(&self.pool)
        .await?;

        let Some(record) = row else {
            return Ok(());
        };

        let Some(Json(mut output_value)) = record
            .try_get::<Option<Json<Value>>, _>("output_json")
            .context("failed to deserialize step output_json")?
        else {
            return Ok(());
        };

        let Some(map) = output_value.as_object_mut() else {
            return Ok(());
        };
        let artifacts_entry = map
            .entry("artifacts".to_string())
            .or_insert_with(|| Value::Object(Map::new()));
        let Value::Object(artifacts_map) = artifacts_entry else {
            return Ok(());
        };
        if let Some(entry) = artifacts_map.get_mut(artifact_key) {
            match entry {
                Value::Object(obj) => {
                    obj.insert("transparency".to_string(), transparency);
                }
                _ => {
                    let mut object = Map::new();
                    object.insert("transparency".to_string(), transparency);
                    *entry = Value::Object(object);
                }
            }
        } else {
            return Ok(());
        }

        sqlx::query(
            r#"
            update steps
            set output_json = $3
            where run_id = $1 and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(Json(output_value))
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    pub async fn update_run_input_ctx_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        input_ctx: Value,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update runs
            set input_ctx = $2
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .bind(Json(input_ctx))
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!("run {} not found", run_id);
        }

        Ok(())
    }

    pub async fn set_run_status(&self, run_id: Uuid, status: models::RunStatus) -> Result<()> {
        let result = sqlx::query(
            r#"
            update runs
            set status = $2
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .bind(status.as_str())
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!("run {} not found", run_id);
        }

        Ok(())
    }

    pub async fn set_run_status_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        status: models::RunStatus,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update runs
            set status = $2
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .bind(status.as_str())
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!("run {} not found", run_id);
        }

        Ok(())
    }

    /// Returns the tenant label for the given run, if one is set.
    pub async fn fetch_run_tenant(&self, run_id: Uuid) -> Result<Option<String>> {
        let tenant: Option<String> = sqlx::query_scalar(
            r#"
            select input_ctx -> 'labels' ->> 'tenant'
            from runs
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .fetch_optional(&self.pool)
        .await?;

        Ok(tenant.and_then(|value| if value.is_empty() { None } else { Some(value) }))
    }

    /// Ensures the provided tenant does not exceed the configured daily quota.
    pub async fn enforce_tenant_quota(&self, tenant: &str, tokens: i64, cost: f64) -> Result<()> {
        let quota: Option<(Option<i64>, Option<f64>)> = sqlx::query_as(
            r#"
            select daily_token_limit, daily_cost_limit
            from tenant_quotas
            where tenant = $1
            "#,
        )
        .bind(tenant)
        .fetch_optional(&self.pool)
        .await?;

        let Some((token_limit, cost_limit)) = quota else {
            return Ok(());
        };

        let usage: (i64, f64) = sqlx::query_as(
            r#"
            select coalesce(sum(tokens), 0), coalesce(sum(cost), 0)
            from billing_usage
            where tenant = $1
              and recorded_at >= date_trunc('day', now())
            "#,
        )
        .bind(tenant)
        .fetch_one(&self.pool)
        .await?;

        let prospective_tokens = usage.0.saturating_add(tokens.max(0));
        if let Some(limit) = token_limit {
            if prospective_tokens > limit {
                bail!("tenant '{}' exceeded daily token quota", tenant);
            }
        }

        let prospective_cost = usage.1 + cost.max(0.0);
        if let Some(limit) = cost_limit {
            if prospective_cost > limit {
                bail!("tenant '{}' exceeded daily cost quota", tenant);
            }
        }

        Ok(())
    }

    /// Records billing usage for a run inside an existing transaction.
    pub async fn record_billing_usage_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        tokens: i64,
        cost: f64,
    ) -> Result<()> {
        sqlx::query(
            r#"
            insert into billing_usage (run_id, tenant, tokens, cost)
            select $1, input_ctx -> 'labels' ->> 'tenant', $2, $3
            from runs
            where run_id = $1
            "#,
        )
        .bind(run_id)
        .bind(tokens)
        .bind(cost)
        .execute(exec(tx))
        .await?;

        Ok(())
    }

    /// Lists recent runs ordered by creation time.
    pub async fn list_recent_runs(&self, limit: i64) -> Result<Vec<models::Run>> {
        let limit = limit.max(1);
        let rows = sqlx::query_as::<_, db::RunRow>(
            r#"
            select *
            from runs
            order by created_at desc
            limit $1
            "#,
        )
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(models::Run::try_from).collect()
    }

    /// Fetches a run by idempotency key if present.
    pub async fn fetch_run_by_idempotency_key(
        &self,
        idempotency_key: &str,
    ) -> Result<Option<models::Run>> {
        let row = sqlx::query_as::<_, db::RunRow>(
            r#"
            select *
            from runs
            where idempotency_key = $1
            "#,
        )
        .bind(idempotency_key)
        .fetch_optional(&self.pool)
        .await?;

        match row {
            Some(row) => Ok(Some(row.try_into()?)),
            None => Ok(None),
        }
    }

    /// Persists artifact metadata while blob bytes live in the object store.
    pub async fn upsert_artifact(&self, artifact: models::NewArtifact) -> Result<()> {
        let models::NewArtifact {
            sha256,
            media_type,
            size_bytes,
            metadata,
        } = artifact;

        let metadata = metadata.map(Json);

        sqlx::query(
            r#"
            insert into artifacts (sha256, media_type, size_bytes, metadata)
            values ($1, $2, $3, $4)
            on conflict (sha256) do update
            set media_type = excluded.media_type,
                size_bytes = excluded.size_bytes,
                metadata = excluded.metadata
            "#,
        )
        .bind(sha256)
        .bind(media_type)
        .bind(size_bytes)
        .bind(metadata)
        .execute(&self.pool)
        .await?;

        Ok(())
    }

    /// Appends an outbox event within an existing transaction for exactly-once delivery.
    pub async fn enqueue_outbox_event(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        event: models::NewOutboxEvent,
    ) -> Result<i64> {
        let models::NewOutboxEvent {
            run_id,
            step_id,
            kind,
            mut payload,
        } = event;

        self.attach_outbox_audit(tx, run_id, step_id, &kind, &mut payload)
            .await?;

        self.append_run_event_tx(tx, run_id, step_id, &kind, &payload)
            .await?;

        let id = sqlx::query_scalar(
            r#"
            insert into outbox (run_id, step_id, kind, payload)
            values ($1, $2, $3, $4)
            returning id
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(&kind)
        .bind(Json(payload))
        .fetch_one(exec(tx))
        .await?;

        Ok(id)
    }

    fn advisory_lock_key(run_id: Uuid) -> (i64, i64) {
        let value = run_id.as_u128();
        let hi = (value >> 64) as i64;
        let lo = value as i64;
        (hi, lo)
    }

    async fn attach_outbox_audit(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Option<Uuid>,
        kind: &str,
        payload: &mut Value,
    ) -> Result<()> {
        if payload.as_object().is_none() {
            bail!(
                "outbox payload for run {} kind '{}' must be a JSON object",
                run_id,
                kind
            );
        }

        let (lock_a, lock_b) = Self::advisory_lock_key(run_id);
        sqlx::query("select pg_advisory_xact_lock($1, $2)")
            .bind(lock_a)
            .bind(lock_b)
            .fetch_optional(exec(tx))
            .await?;

        let previous_hash: Option<String> = sqlx::query_scalar(
            r#"
            select payload->'_audit'->>'chain_hash'
            from outbox
            where run_id = $1
            order by id desc
            limit 1
            for update
            "#,
        )
        .bind(run_id)
        .fetch_optional(exec(tx))
        .await?;

        let emitted_at = Utc::now();
        let canonical = serde_json::to_vec(payload)
            .context("failed to serialise outbox payload for audit hashing")?;
        let payload_hash = Sha256::digest(&canonical);
        let payload_hash_bytes: &[u8] = payload_hash.as_ref();

        let mut hasher = Sha256::new();
        if let Some(prev) = &previous_hash {
            hasher.update(prev.as_bytes());
        }
        hasher.update(run_id.as_bytes());
        if let Some(step) = step_id {
            hasher.update(step.as_bytes());
        }
        hasher.update(kind.as_bytes());
        hasher.update(payload_hash_bytes);
        let timestamp = emitted_at.timestamp_nanos_opt().unwrap_or(0);
        hasher.update(timestamp.to_be_bytes());
        let chain_hash = hex::encode(hasher.finalize());

        if let Some(map) = payload.as_object_mut() {
            map.insert(
                "_audit".to_string(),
                json!({
                    "version": 1,
                    "chain_hash": chain_hash,
                    "payload_hash": hex::encode(payload_hash_bytes),
                    "previous_hash": previous_hash,
                    "emitted_at": emitted_at,
                }),
            );
        }

        Ok(())
    }

    /// Claims queued steps using `FOR UPDATE SKIP LOCKED` for safe parallel scheduling.
    pub async fn claim_queued_steps(
        &self,
        limit: i64,
        worker_id: &str,
    ) -> Result<Vec<models::Step>> {
        let mut tx = self.pool.begin().await?;
        let rows: Vec<db::StepRow> = sqlx::query_as(
            r#"
            with cte as (
                select s.run_id, s.step_id
                from steps s
                inner join runs r on r.run_id = s.run_id
                where s.status = 'queued'
                  and s.pending_dependencies = 0
                  and r.status <> 'paused_at_step'
                  and (s.not_before is null or s.not_before <= now())
                order by s.priority desc, coalesce(s.not_before, s.created_at) asc, s.idx
                for update skip locked
                limit $1
            )
            update steps s set
                status = 'leased',
                leased_by = $2,
                lease_expires_at = now() + interval '2 minutes'
            from cte
            where s.run_id = cte.run_id and s.step_id = cte.step_id
            returning s.*
            "#,
        )
        .bind(limit)
        .bind(worker_id)
        .fetch_all(exec(&mut tx))
        .await?;

        if !rows.is_empty() {
            let run_ids_set: HashSet<Uuid> = rows.iter().map(|row| row.run_id).collect();
            let run_ids: Vec<Uuid> = run_ids_set.into_iter().collect();

            sqlx::query(
                r#"
                update runs
                set status = 'running'
                where status = 'pending'
                  and run_id = any($1)
                "#,
            )
            .bind(&run_ids)
            .execute(exec(&mut tx))
            .await?;
        }
        tx.commit().await?;

        rows.into_iter()
            .map(models::Step::try_from)
            .collect::<Result<Vec<_>>>()
    }

    pub async fn revert_step_to_queue(&self, run_id: Uuid, step_id: Uuid) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set status = 'queued',
                leased_by = null,
                lease_expires_at = null
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to revert step {} to queue for run {}",
                step_id,
                run_id
            );
        }
        Ok(())
    }

    pub async fn revert_step_to_queue_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set status = 'queued',
                leased_by = null,
                lease_expires_at = null
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to revert step {} to queue for run {}",
                step_id,
                run_id
            );
        }
        Ok(())
    }

    pub async fn update_step_spec(
        &self,
        run_id: Uuid,
        step_id: Uuid,
        spec_json: Value,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set spec_json = $3
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(Json(spec_json))
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to update spec for step {} in run {}",
                step_id,
                run_id
            );
        }
        Ok(())
    }

    pub async fn update_step_spec_tx(
        &self,
        tx: &mut Transaction<'_, Postgres>,
        run_id: Uuid,
        step_id: Uuid,
        spec_json: Value,
    ) -> Result<()> {
        let result = sqlx::query(
            r#"
            update steps
            set spec_json = $3
            where run_id = $1
              and step_id = $2
            "#,
        )
        .bind(run_id)
        .bind(step_id)
        .bind(Json(spec_json))
        .execute(exec(tx))
        .await?;

        if result.rows_affected() == 0 {
            bail!(
                "failed to update spec for step {} in run {}",
                step_id,
                run_id
            );
        }
        Ok(())
    }
}

mod db {
    use super::*;
    use serde_json::Value;

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct RunRow {
        pub run_id: Uuid,
        pub created_at: DateTime<Utc>,
        pub status: String,
        pub dag_json: Value,
        pub input_ctx: Value,
        pub seed: i64,
        pub idempotency_key: Option<String>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct StepRow {
        pub run_id: Uuid,
        pub step_id: Uuid,
        pub idx: i32,
        pub priority: i16,
        pub spec_json: Value,
        pub status: String,
        pub attempt: i32,
        pub created_at: DateTime<Utc>,
        pub max_attempts: i32,
        pub retry_backoff_ms: i64,
        pub retry_backoff_factor: f64,
        pub not_before: Option<DateTime<Utc>>,
        pub deadline_at: Option<DateTime<Utc>>,
        pub pending_dependencies: i32,
        pub total_dependencies: i32,
        pub estimated_tokens: i64,
        pub estimated_cost: f64,
        pub actual_tokens: Option<i64>,
        pub actual_cost: Option<f64>,
        pub leased_by: Option<String>,
        pub lease_expires_at: Option<DateTime<Utc>>,
        pub output_json: Option<Value>,
        pub error_json: Option<Value>,
        pub input_snapshot: Option<Value>,
        pub output_snapshot: Option<Value>,
        pub provider: Option<String>,
        pub provider_version: Option<String>,
        pub checkpoint: Option<Value>,
        pub compensation_step: Option<Uuid>,
        pub compensation_scheduled: bool,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct StepAttemptRow {
        pub run_id: Uuid,
        pub step_id: Uuid,
        pub attempt: i32,
        pub status: String,
        pub recorded_at: DateTime<Utc>,
        pub latency_ms: Option<f64>,
        pub spec_snapshot: Value,
        pub input_snapshot: Option<Value>,
        pub output_snapshot: Option<Value>,
        pub error_snapshot: Option<Value>,
        pub checkpoint: Option<Value>,
        pub budget_snapshot: Option<Value>,
        pub guardrail_events: Option<Value>,
        pub tool_events: Option<Value>,
        pub artifacts: Option<Value>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct BudgetRow {
        pub budget_id: Uuid,
        pub scope: String,
        pub run_id: Option<Uuid>,
        pub team_id: Option<Uuid>,
        pub tokens_limit: Option<i64>,
        pub tokens_used: i64,
        pub cost_limit: Option<f64>,
        pub cost_used: f64,
        pub created_at: DateTime<Utc>,
        pub updated_at: DateTime<Utc>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct ArtifactRow {
        pub sha256: Vec<u8>,
        pub media_type: String,
        pub size_bytes: i64,
        pub created_at: DateTime<Utc>,
        pub metadata: Option<Value>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct OutboxRow {
        pub id: i64,
        pub run_id: Uuid,
        pub step_id: Option<Uuid>,
        pub kind: String,
        pub payload: Value,
        pub created_at: DateTime<Utc>,
        pub published_at: Option<DateTime<Utc>>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct EvalScenarioRow {
        pub scenario_id: Uuid,
        pub name: String,
        pub description: Option<String>,
        pub spec: Value,
        pub created_at: DateTime<Utc>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct EvalResultRow {
        pub result_id: Uuid,
        pub scenario_id: Uuid,
        pub run_id: Uuid,
        pub executed_at: DateTime<Utc>,
        pub outcome: String,
        pub metrics: Value,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct ChangeGateRow {
        pub gate_id: Uuid,
        pub change_id: String,
        pub revision: Option<String>,
        pub effect: String,
        pub reasons: Value,
        pub followups: Value,
        pub scorecard: Value,
        pub decided_at: DateTime<Utc>,
        pub decided_by: Option<String>,
        pub metadata: Value,
        pub input: Value,
        pub telemetry: Value,
        pub artifact_sha256: Option<Vec<u8>>,
        pub created_at: DateTime<Utc>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct ChangeGateFollowupRow {
        pub followup_id: Uuid,
        pub gate_id: Uuid,
        pub actor: String,
        pub outcome: String,
        pub note: Option<String>,
        pub details: Value,
        pub recorded_at: DateTime<Utc>,
    }

    #[derive(Debug, Clone, sqlx::FromRow)]
    pub struct TransparencyJobRow {
        pub job_id: i64,
        pub kind: String,
        pub payload: Value,
        pub receipt: Option<Value>,
        pub error: Option<String>,
        pub attempts: i32,
        pub created_at: DateTime<Utc>,
        pub processed_at: Option<DateTime<Utc>>,
    }
}

pub mod models {
    use super::*;
    use serde::{Deserialize, Serialize};
    use serde_json::Value;

    /// Valid lifecycle states for a run.
    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
    #[serde(rename_all = "snake_case")]
    pub enum RunStatus {
        Pending,
        Running,
        Succeeded,
        Failed,
        Canceled,
        PausedAtStep,
    }

    impl RunStatus {
        pub fn as_str(self) -> &'static str {
            match self {
                RunStatus::Pending => "pending",
                RunStatus::Running => "running",
                RunStatus::Succeeded => "succeeded",
                RunStatus::Failed => "failed",
                RunStatus::Canceled => "canceled",
                RunStatus::PausedAtStep => "paused_at_step",
            }
        }

        pub fn try_from_str(value: &str) -> Result<Self> {
            match value {
                "pending" => Ok(RunStatus::Pending),
                "running" => Ok(RunStatus::Running),
                "succeeded" => Ok(RunStatus::Succeeded),
                "failed" => Ok(RunStatus::Failed),
                "canceled" => Ok(RunStatus::Canceled),
                "paused_at_step" => Ok(RunStatus::PausedAtStep),
                other => Err(anyhow!("invalid run_status '{}'", other)),
            }
        }
    }

    /// Valid lifecycle states for a step.
    #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
    #[serde(rename_all = "snake_case")]
    pub enum StepStatus {
        Queued,
        Blocked,
        Leased,
        Running,
        Succeeded,
        Failed,
        Skipped,
    }

    impl StepStatus {
        pub fn as_str(self) -> &'static str {
            match self {
                StepStatus::Queued => "queued",
                StepStatus::Blocked => "blocked",
                StepStatus::Leased => "leased",
                StepStatus::Running => "running",
                StepStatus::Succeeded => "succeeded",
                StepStatus::Failed => "failed",
                StepStatus::Skipped => "skipped",
            }
        }

        pub fn try_from_str(value: &str) -> Result<Self> {
            match value {
                "queued" => Ok(StepStatus::Queued),
                "blocked" => Ok(StepStatus::Blocked),
                "leased" => Ok(StepStatus::Leased),
                "running" => Ok(StepStatus::Running),
                "succeeded" => Ok(StepStatus::Succeeded),
                "failed" => Ok(StepStatus::Failed),
                "skipped" => Ok(StepStatus::Skipped),
                other => Err(anyhow!("invalid step_status '{}'", other)),
            }
        }
    }

    /// Run contract materialized in the database.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct Run {
        pub run_id: Uuid,
        pub created_at: DateTime<Utc>,
        pub status: RunStatus,
        pub dag_json: Value,
        pub input_ctx: Value,
        pub seed: i64,
        pub idempotency_key: Option<String>,
    }

    /// Insertable representation for a run.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewRun {
        pub run_id: Uuid,
        pub status: RunStatus,
        pub dag_json: Value,
        pub input_ctx: Value,
        pub seed: i64,
        pub idempotency_key: Option<String>,
    }

    /// Step contract spanning deterministic replay and attempt metadata.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct Step {
        pub run_id: Uuid,
        pub step_id: Uuid,
        pub idx: i32,
        pub priority: i16,
        pub spec_json: Value,
        pub status: StepStatus,
        pub attempt: i32,
        pub created_at: DateTime<Utc>,
        pub max_attempts: i32,
        pub retry_backoff_ms: i64,
        pub retry_backoff_factor: f64,
        pub not_before: Option<DateTime<Utc>>,
        pub deadline_at: Option<DateTime<Utc>>,
        pub pending_dependencies: i32,
        pub total_dependencies: i32,
        pub estimated_tokens: i64,
        pub estimated_cost: f64,
        pub actual_tokens: Option<i64>,
        pub actual_cost: Option<f64>,
        pub leased_by: Option<String>,
        pub lease_expires_at: Option<DateTime<Utc>>,
        pub output_json: Option<Value>,
        pub error_json: Option<Value>,
        pub input_snapshot: Option<Value>,
        pub output_snapshot: Option<Value>,
        pub provider: Option<String>,
        pub provider_version: Option<String>,
        pub checkpoint: Option<Value>,
        pub compensation_step: Option<Uuid>,
        pub compensation_scheduled: bool,
    }

    /// Insertable representation for a step.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewStep {
        pub run_id: Uuid,
        pub step_id: Uuid,
        pub idx: i32,
        pub priority: i16,
        pub spec_json: Value,
        pub status: StepStatus,
        pub attempt: i32,
        pub input_snapshot: Option<Value>,
        pub provider: Option<String>,
        pub provider_version: Option<String>,
        pub max_attempts: i32,
        pub retry_backoff_ms: i64,
        pub retry_backoff_factor: f64,
        pub not_before: Option<DateTime<Utc>>,
        pub deadline_at: Option<DateTime<Utc>>,
        pub pending_dependencies: i32,
        pub total_dependencies: i32,
        pub estimated_tokens: i64,
        pub estimated_cost: f64,
        pub actual_tokens: Option<i64>,
        pub actual_cost: Option<f64>,
        pub checkpoint: Option<Value>,
        pub compensation_step: Option<Uuid>,
        pub compensation_scheduled: bool,
    }

    /// Immutable snapshot of a step attempt used for deterministic replay.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct StepAttempt {
        pub run_id: Uuid,
        pub step_id: Uuid,
        pub attempt: i32,
        pub status: StepStatus,
        pub recorded_at: DateTime<Utc>,
        pub latency_ms: Option<f64>,
        pub spec_snapshot: Value,
        pub input_snapshot: Option<Value>,
        pub output_snapshot: Option<Value>,
        pub error_snapshot: Option<Value>,
        pub checkpoint: Option<Value>,
        pub budget_snapshot: Option<Value>,
        pub guardrail_events: Option<Value>,
        pub tool_events: Option<Value>,
        pub artifacts: Option<Value>,
    }

    #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
    #[serde(rename_all = "snake_case")]
    pub enum TransparencyJobKind {
        ReleaseBundle,
        RunRollup,
        ArtifactEntry,
    }

    impl TransparencyJobKind {
        pub fn as_str(self) -> &'static str {
            match self {
                TransparencyJobKind::ReleaseBundle => "release_bundle",
                TransparencyJobKind::RunRollup => "run_rollup",
                TransparencyJobKind::ArtifactEntry => "artifact_entry",
            }
        }
    }

    impl FromStr for TransparencyJobKind {
        type Err = anyhow::Error;

        fn from_str(s: &str) -> Result<Self> {
            match s {
                "release_bundle" => Ok(TransparencyJobKind::ReleaseBundle),
                "run_rollup" => Ok(TransparencyJobKind::RunRollup),
                "artifact_entry" => Ok(TransparencyJobKind::ArtifactEntry),
                other => Err(anyhow!("invalid transparency job kind '{}'", other)),
            }
        }
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct ReleaseBundlePayload {
        pub change_id: String,
        pub gate_id: Uuid,
        pub artifact_sha256: String,
        pub attestation_ids: Vec<Uuid>,
        pub metadata: Value,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct RunRollupPayload {
        pub run_id: Uuid,
        pub artifact_sha256: String,
        pub attestation_ids: Vec<Uuid>,
        pub artifact_key: String,
        pub metadata: Value,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct ArtifactEntryPayload {
        pub run_id: Option<Uuid>,
        pub step_id: Option<Uuid>,
        pub artifact_key: String,
        pub artifact_sha256: String,
        pub attestation_ids: Vec<Uuid>,
        pub metadata: Value,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    #[serde(tag = "kind", rename_all = "snake_case")]
    pub enum TransparencyJobPayload {
        ReleaseBundle(ReleaseBundlePayload),
        RunRollup(RunRollupPayload),
        ArtifactEntry(ArtifactEntryPayload),
    }

    impl TransparencyJobPayload {
        pub fn kind(&self) -> TransparencyJobKind {
            match self {
                TransparencyJobPayload::ReleaseBundle(_) => TransparencyJobKind::ReleaseBundle,
                TransparencyJobPayload::RunRollup(_) => TransparencyJobKind::RunRollup,
                TransparencyJobPayload::ArtifactEntry(_) => TransparencyJobKind::ArtifactEntry,
            }
        }
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct TransparencyJob {
        pub job_id: i64,
        pub kind: TransparencyJobKind,
        pub payload: TransparencyJobPayload,
        pub attempts: i32,
        pub last_error: Option<String>,
        pub receipt: Option<Value>,
        pub processed_at: Option<DateTime<Utc>>,
        pub created_at: DateTime<Utc>,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewTransparencyJob {
        pub payload: TransparencyJobPayload,
    }

    /// Insertable representation for a step attempt.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewStepAttempt {
        pub run_id: Uuid,
        pub step_id: Uuid,
        pub attempt: i32,
        pub status: StepStatus,
        pub latency_ms: Option<f64>,
        pub spec_snapshot: Value,
        pub input_snapshot: Option<Value>,
        pub output_snapshot: Option<Value>,
        pub error_snapshot: Option<Value>,
        pub checkpoint: Option<Value>,
        pub budget_snapshot: Option<Value>,
        pub guardrail_events: Option<Value>,
        pub tool_events: Option<Value>,
        pub artifacts: Option<Value>,
    }

    impl Default for NewStep {
        fn default() -> Self {
            Self {
                run_id: Uuid::nil(),
                step_id: Uuid::nil(),
                idx: 0,
                priority: 0,
                spec_json: Value::Null,
                status: StepStatus::Queued,
                attempt: 0,
                input_snapshot: None,
                provider: None,
                provider_version: None,
                max_attempts: 1,
                retry_backoff_ms: 0,
                retry_backoff_factor: 1.0,
                not_before: None,
                deadline_at: None,
                pending_dependencies: 0,
                total_dependencies: 0,
                estimated_tokens: 0,
                estimated_cost: 0.0,
                actual_tokens: None,
                actual_cost: None,
                checkpoint: None,
                compensation_step: None,
                compensation_scheduled: false,
            }
        }
    }

    impl TryFrom<crate::db::StepRow> for Step {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::StepRow) -> Result<Self, Self::Error> {
            Ok(Self {
                run_id: row.run_id,
                step_id: row.step_id,
                idx: row.idx,
                priority: row.priority,
                spec_json: row.spec_json,
                status: StepStatus::try_from_str(&row.status)?,
                attempt: row.attempt,
                created_at: row.created_at,
                max_attempts: row.max_attempts,
                retry_backoff_ms: row.retry_backoff_ms,
                retry_backoff_factor: row.retry_backoff_factor,
                not_before: row.not_before,
                deadline_at: row.deadline_at,
                pending_dependencies: row.pending_dependencies,
                total_dependencies: row.total_dependencies,
                estimated_tokens: row.estimated_tokens,
                estimated_cost: row.estimated_cost,
                actual_tokens: row.actual_tokens,
                actual_cost: row.actual_cost,
                leased_by: row.leased_by,
                lease_expires_at: row.lease_expires_at,
                output_json: row.output_json,
                error_json: row.error_json,
                input_snapshot: row.input_snapshot,
                output_snapshot: row.output_snapshot,
                provider: row.provider,
                provider_version: row.provider_version,
                checkpoint: row.checkpoint,
                compensation_step: row.compensation_step,
                compensation_scheduled: row.compensation_scheduled,
            })
        }
    }

    impl TryFrom<crate::db::StepAttemptRow> for StepAttempt {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::StepAttemptRow) -> Result<Self, Self::Error> {
            Ok(Self {
                run_id: row.run_id,
                step_id: row.step_id,
                attempt: row.attempt,
                status: StepStatus::try_from_str(&row.status)?,
                recorded_at: row.recorded_at,
                latency_ms: row.latency_ms,
                spec_snapshot: row.spec_snapshot,
                input_snapshot: row.input_snapshot,
                output_snapshot: row.output_snapshot,
                error_snapshot: row.error_snapshot,
                checkpoint: row.checkpoint,
                budget_snapshot: row.budget_snapshot,
                guardrail_events: row.guardrail_events,
                tool_events: row.tool_events,
                artifacts: row.artifacts,
            })
        }
    }

    /// Content-addressed artifact metadata pointing at object storage bytes.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct Artifact {
        pub sha256: Vec<u8>,
        pub media_type: String,
        pub size_bytes: i64,
        pub created_at: DateTime<Utc>,
        pub metadata: Option<Value>,
    }

    /// Insertable representation for artifact metadata.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewArtifact {
        pub sha256: Vec<u8>,
        pub media_type: String,
        pub size_bytes: i64,
        pub metadata: Option<Value>,
    }

    impl From<super::ArtifactMeta> for NewArtifact {
        fn from(meta: super::ArtifactMeta) -> Self {
            Self {
                sha256: meta.sha256.to_vec(),
                media_type: meta.media_type,
                size_bytes: std::cmp::min(meta.size, i64::MAX as u64) as i64,
                metadata: Some(meta.metadata),
            }
        }
    }

    /// Budget configuration and cumulative consumption counters.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct Budget {
        pub budget_id: Uuid,
        pub scope: String,
        pub run_id: Option<Uuid>,
        pub team_id: Option<Uuid>,
        pub tokens_limit: Option<i64>,
        pub tokens_used: i64,
        pub cost_limit: Option<f64>,
        pub cost_used: f64,
        pub created_at: DateTime<Utc>,
        pub updated_at: DateTime<Utc>,
    }

    impl TryFrom<crate::db::BudgetRow> for Budget {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::BudgetRow) -> Result<Self, Self::Error> {
            Ok(Self {
                budget_id: row.budget_id,
                scope: row.scope,
                run_id: row.run_id,
                team_id: row.team_id,
                tokens_limit: row.tokens_limit,
                tokens_used: row.tokens_used,
                cost_limit: row.cost_limit,
                cost_used: row.cost_used,
                created_at: row.created_at,
                updated_at: row.updated_at,
            })
        }
    }

    impl Budget {
        pub fn remaining_tokens(&self) -> Option<i64> {
            self.tokens_limit
                .map(|limit| limit.saturating_sub(self.tokens_used))
        }

        pub fn remaining_cost(&self) -> Option<f64> {
            self.cost_limit
                .map(|limit| (limit - self.cost_used).max(0.0))
        }
    }

    /// Outbox event persisted transactionally with state mutations.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct OutboxEvent {
        pub id: i64,
        pub run_id: Uuid,
        pub step_id: Option<Uuid>,
        pub kind: String,
        pub payload: Value,
        pub created_at: DateTime<Utc>,
        pub published_at: Option<DateTime<Utc>>,
    }

    /// Registered evaluation scenario definition.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct EvalScenario {
        pub scenario_id: Uuid,
        pub name: String,
        pub description: Option<String>,
        pub spec: Value,
        pub created_at: DateTime<Utc>,
    }

    /// Result for an evaluation scenario run.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct EvalResult {
        pub result_id: Uuid,
        pub scenario_id: Uuid,
        pub run_id: Uuid,
        pub executed_at: DateTime<Utc>,
        pub outcome: String,
        pub metrics: Value,
    }

    /// Builder for inserting a new outbox event alongside DB state changes.
    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewOutboxEvent {
        pub run_id: Uuid,
        pub step_id: Option<Uuid>,
        pub kind: String,
        pub payload: Value,
    }

    impl TryFrom<crate::db::RunRow> for Run {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::RunRow) -> Result<Self, Self::Error> {
            Ok(Self {
                run_id: row.run_id,
                created_at: row.created_at,
                status: RunStatus::try_from_str(&row.status)?,
                dag_json: row.dag_json,
                input_ctx: row.input_ctx,
                seed: row.seed,
                idempotency_key: row.idempotency_key,
            })
        }
    }

    impl TryFrom<crate::db::EvalScenarioRow> for EvalScenario {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::EvalScenarioRow) -> Result<Self, Self::Error> {
            Ok(Self {
                scenario_id: row.scenario_id,
                name: row.name,
                description: row.description,
                spec: row.spec,
                created_at: row.created_at,
            })
        }
    }

    impl TryFrom<crate::db::EvalResultRow> for EvalResult {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::EvalResultRow) -> Result<Self, Self::Error> {
            Ok(Self {
                result_id: row.result_id,
                scenario_id: row.scenario_id,
                run_id: row.run_id,
                executed_at: row.executed_at,
                outcome: row.outcome,
                metrics: row.metrics,
            })
        }
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct ChangeGate {
        pub gate_id: Uuid,
        pub change_id: String,
        pub revision: Option<String>,
        pub effect: String,
        pub reasons: Value,
        pub followups: Value,
        pub scorecard: Value,
        pub decided_at: DateTime<Utc>,
        pub decided_by: Option<String>,
        pub metadata: Value,
        pub input: Value,
        pub telemetry: Value,
        pub artifact_sha256: Option<[u8; 32]>,
        pub created_at: DateTime<Utc>,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewChangeGate {
        pub gate_id: Uuid,
        pub change_id: String,
        pub revision: Option<String>,
        pub effect: String,
        pub reasons: Value,
        pub followups: Value,
        pub scorecard: Value,
        pub decided_at: DateTime<Utc>,
        pub decided_by: Option<String>,
        pub metadata: Value,
        pub input: Value,
        pub telemetry: Value,
        pub artifact_sha256: Option<[u8; 32]>,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct ChangeGateFollowup {
        pub followup_id: Uuid,
        pub gate_id: Uuid,
        pub actor: String,
        pub outcome: String,
        pub note: Option<String>,
        pub details: Value,
        pub recorded_at: DateTime<Utc>,
    }

    #[derive(Debug, Clone, Serialize, Deserialize)]
    pub struct NewChangeGateFollowup {
        pub followup_id: Uuid,
        pub gate_id: Uuid,
        pub actor: String,
        pub outcome: String,
        pub note: Option<String>,
        pub details: Value,
    }

    impl From<crate::db::ChangeGateRow> for ChangeGate {
        fn from(row: crate::db::ChangeGateRow) -> Self {
            let artifact_sha256 = row.artifact_sha256.and_then(|vec| {
                if vec.len() == 32 {
                    let mut value = [0u8; 32];
                    value.copy_from_slice(&vec);
                    Some(value)
                } else {
                    None
                }
            });

            Self {
                gate_id: row.gate_id,
                change_id: row.change_id,
                revision: row.revision,
                effect: row.effect,
                reasons: row.reasons,
                followups: row.followups,
                scorecard: row.scorecard,
                decided_at: row.decided_at,
                decided_by: row.decided_by,
                metadata: row.metadata,
                input: row.input,
                telemetry: row.telemetry,
                artifact_sha256,
                created_at: row.created_at,
            }
        }
    }

    impl From<crate::db::ChangeGateFollowupRow> for ChangeGateFollowup {
        fn from(row: crate::db::ChangeGateFollowupRow) -> Self {
            Self {
                followup_id: row.followup_id,
                gate_id: row.gate_id,
                actor: row.actor,
                outcome: row.outcome,
                note: row.note,
                details: row.details,
                recorded_at: row.recorded_at,
            }
        }
    }

    impl TryFrom<crate::db::TransparencyJobRow> for TransparencyJob {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::TransparencyJobRow) -> Result<Self> {
            let kind = TransparencyJobKind::from_str(&row.kind)?;
            let payload: TransparencyJobPayload = serde_json::from_value(row.payload)?;
            Ok(TransparencyJob {
                job_id: row.job_id,
                kind,
                payload,
                attempts: row.attempts,
                last_error: row.error,
                receipt: row.receipt,
                processed_at: row.processed_at,
                created_at: row.created_at,
            })
        }
    }

    impl TryFrom<crate::db::ArtifactRow> for Artifact {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::ArtifactRow) -> Result<Self, Self::Error> {
            Ok(Self {
                sha256: row.sha256,
                media_type: row.media_type,
                size_bytes: row.size_bytes,
                created_at: row.created_at,
                metadata: row.metadata,
            })
        }
    }

    impl TryFrom<crate::db::OutboxRow> for OutboxEvent {
        type Error = anyhow::Error;

        fn try_from(row: crate::db::OutboxRow) -> Result<Self, Self::Error> {
            Ok(Self {
                id: row.id,
                run_id: row.run_id,
                step_id: row.step_id,
                kind: row.kind,
                payload: row.payload,
                created_at: row.created_at,
                published_at: row.published_at,
            })
        }
    }
}

impl Storage {
    /// Registers or updates an evaluation scenario definition.
    pub async fn upsert_eval_scenario(
        &self,
        name: &str,
        description: Option<&str>,
        spec: Value,
    ) -> Result<Uuid> {
        let scenario_id = sqlx::query_scalar::<_, Uuid>(
            r#"
            insert into eval_scenarios (name, description, spec)
            values ($1, $2, $3)
            on conflict (name)
            do update set description = excluded.description,
                          spec = excluded.spec
            returning scenario_id
            "#,
        )
        .bind(name)
        .bind(description)
        .bind(Json(spec))
        .fetch_one(&self.pool)
        .await?;

        Ok(scenario_id)
    }

    /// Fetches an evaluation scenario by its unique name.
    pub async fn fetch_eval_scenario_by_name(
        &self,
        name: &str,
    ) -> Result<Option<models::EvalScenario>> {
        let row = sqlx::query_as::<_, db::EvalScenarioRow>(
            r#"
            select *
            from eval_scenarios
            where name = $1
            "#,
        )
        .bind(name)
        .fetch_optional(&self.pool)
        .await?;

        row.map(models::EvalScenario::try_from).transpose()
    }

    /// Lists all evaluation scenarios ordered by name.
    pub async fn list_eval_scenarios(&self) -> Result<Vec<models::EvalScenario>> {
        let rows = sqlx::query_as::<_, db::EvalScenarioRow>(
            r#"
            select *
            from eval_scenarios
            order by name asc
            "#,
        )
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(models::EvalScenario::try_from)
            .collect()
    }

    /// Returns the latest evaluation result for a scenario, if any.
    pub async fn latest_eval_result(
        &self,
        scenario_id: Uuid,
    ) -> Result<Option<models::EvalResult>> {
        let row = sqlx::query_as::<_, db::EvalResultRow>(
            r#"
            select *
            from eval_results
            where scenario_id = $1
            order by executed_at desc
            limit 1
            "#,
        )
        .bind(scenario_id)
        .fetch_optional(&self.pool)
        .await?;

        row.map(models::EvalResult::try_from).transpose()
    }

    /// Lists evaluation results for a scenario, newest first.
    pub async fn list_eval_results(
        &self,
        scenario_id: Uuid,
        limit: i64,
    ) -> Result<Vec<models::EvalResult>> {
        let rows = sqlx::query_as::<_, db::EvalResultRow>(
            r#"
            select *
            from eval_results
            where scenario_id = $1
            order by executed_at desc
            limit $2
            "#,
        )
        .bind(scenario_id)
        .bind(limit.max(1))
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter().map(models::EvalResult::try_from).collect()
    }

    /// Inserts or replaces an evaluation result for the given scenario/run pair.
    pub async fn insert_eval_result(
        &self,
        scenario_id: Uuid,
        run_id: Uuid,
        outcome: &str,
        metrics: Value,
    ) -> Result<models::EvalResult> {
        let row = sqlx::query_as::<_, db::EvalResultRow>(
            r#"
            insert into eval_results (scenario_id, run_id, outcome, metrics)
            values ($1, $2, $3, $4)
            on conflict (scenario_id, run_id)
            do update set outcome = excluded.outcome,
                          metrics = excluded.metrics,
                          executed_at = now()
            returning *
            "#,
        )
        .bind(scenario_id)
        .bind(run_id)
        .bind(outcome)
        .bind(Json(metrics))
        .fetch_one(&self.pool)
        .await?;

        models::EvalResult::try_from(row)
    }

    /// Records a ChangeOps gate decision and returns the stored record.
    pub async fn insert_change_gate(
        &self,
        gate: models::NewChangeGate,
    ) -> Result<models::ChangeGate> {
        let models::NewChangeGate {
            gate_id,
            change_id,
            revision,
            effect,
            reasons,
            followups,
            scorecard,
            decided_at,
            decided_by,
            metadata,
            input,
            telemetry,
            artifact_sha256,
        } = gate;

        let row = sqlx::query_as::<_, db::ChangeGateRow>(
            r#"
            insert into change_gates (
                gate_id,
                change_id,
                revision,
                effect,
                reasons,
                followups,
                scorecard,
                decided_at,
                decided_by,
                metadata,
                input,
                telemetry,
                artifact_sha256
            )
            values (
                $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13
            )
            returning *
            "#,
        )
        .bind(gate_id)
        .bind(change_id)
        .bind(revision)
        .bind(effect)
        .bind(Json(reasons))
        .bind(Json(followups))
        .bind(Json(scorecard))
        .bind(decided_at)
        .bind(decided_by)
        .bind(Json(metadata))
        .bind(Json(input))
        .bind(Json(telemetry))
        .bind(artifact_sha256.map(|digest| digest.to_vec()))
        .fetch_one(&self.pool)
        .await?;

        Ok(models::ChangeGate::from(row))
    }

    /// Replaces the metadata JSON stored for a change gate decision.
    pub async fn update_change_gate_metadata(&self, gate_id: Uuid, metadata: Value) -> Result<()> {
        let result = sqlx::query(
            r#"
            update change_gates
            set metadata = $2
            where gate_id = $1
            "#,
        )
        .bind(gate_id)
        .bind(Json(metadata))
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!("change gate {} not found", gate_id);
        }

        Ok(())
    }

    /// Fetches the latest gate decision for a change id (limited by `limit`).
    pub async fn list_change_gates_for_change(
        &self,
        change_id: &str,
        limit: i64,
    ) -> Result<Vec<models::ChangeGate>> {
        let rows = sqlx::query_as::<_, db::ChangeGateRow>(
            r#"
            select *
            from change_gates
            where change_id = $1
            order by decided_at desc
            limit $2
            "#,
        )
        .bind(change_id)
        .bind(limit.max(1))
        .fetch_all(&self.pool)
        .await?;

        Ok(rows.into_iter().map(models::ChangeGate::from).collect())
    }

    /// Retrieves a specific gate decision by gate id.
    pub async fn fetch_change_gate(&self, gate_id: Uuid) -> Result<Option<models::ChangeGate>> {
        let row = sqlx::query_as::<_, db::ChangeGateRow>(
            r#"
            select *
            from change_gates
            where gate_id = $1
            "#,
        )
        .bind(gate_id)
        .fetch_optional(&self.pool)
        .await?;

        Ok(row.map(models::ChangeGate::from))
    }

    /// Enqueues a transparency job for asynchronous publishing.
    pub async fn enqueue_transparency_job(
        &self,
        payload: models::TransparencyJobPayload,
    ) -> Result<i64> {
        let kind = payload.kind().as_str().to_string();
        let job_id = sqlx::query_scalar::<_, i64>(
            r#"
            insert into transparency_jobs (kind, payload)
            values ($1, $2)
            returning job_id
            "#,
        )
        .bind(kind)
        .bind(serde_json::to_value(&payload)?)
        .fetch_one(&self.pool)
        .await?;

        Ok(job_id)
    }

    /// Returns pending transparency jobs up to the requested limit.
    pub async fn fetch_pending_transparency_jobs(
        &self,
        limit: i64,
    ) -> Result<Vec<models::TransparencyJob>> {
        let rows = sqlx::query_as::<_, db::TransparencyJobRow>(
            r#"
            select *
            from transparency_jobs
            where processed_at is null
            order by job_id asc
            limit $1
            "#,
        )
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(models::TransparencyJob::try_from)
            .collect()
    }

    /// Fetches transparency jobs by explicit identifiers.
    pub async fn fetch_transparency_jobs_by_ids(
        &self,
        job_ids: &[i64],
    ) -> Result<Vec<models::TransparencyJob>> {
        if job_ids.is_empty() {
            return Ok(Vec::new());
        }
        let rows = sqlx::query_as::<_, db::TransparencyJobRow>(
            r#"
            select *
            from transparency_jobs
            where job_id = any($1)
            order by job_id asc
            "#,
        )
        .bind(job_ids)
        .fetch_all(&self.pool)
        .await?;

        rows.into_iter()
            .map(models::TransparencyJob::try_from)
            .collect()
    }

    /// Records a successful transparency job along with the receipt payload.
    pub async fn mark_transparency_job_processed(&self, job_id: i64, receipt: Value) -> Result<()> {
        let result = sqlx::query(
            r#"
            update transparency_jobs
            set processed_at = now(),
                receipt = $2,
                error = null
            where job_id = $1
            "#,
        )
        .bind(job_id)
        .bind(Json(receipt))
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!("transparency job {} not found", job_id);
        }

        Ok(())
    }

    /// Marks a transparency job failure so the writer can retry later.
    pub async fn mark_transparency_job_failed(&self, job_id: i64, error: String) -> Result<()> {
        let result = sqlx::query(
            r#"
            update transparency_jobs
            set error = $2,
                attempts = attempts + 1
            where job_id = $1
            "#,
        )
        .bind(job_id)
        .bind(error)
        .execute(&self.pool)
        .await?;

        if result.rows_affected() == 0 {
            bail!("transparency job {} not found", job_id);
        }

        Ok(())
    }

    /// Records a follow-up acknowledgement for a gate.
    pub async fn insert_change_gate_followup(
        &self,
        followup: models::NewChangeGateFollowup,
    ) -> Result<models::ChangeGateFollowup> {
        let models::NewChangeGateFollowup {
            followup_id,
            gate_id,
            actor,
            outcome,
            note,
            details,
        } = followup;

        let row = sqlx::query_as::<_, db::ChangeGateFollowupRow>(
            r#"
            insert into change_gate_followups (
                followup_id,
                gate_id,
                actor,
                outcome,
                note,
                details
            )
            values ($1,$2,$3,$4,$5,$6)
            returning *
            "#,
        )
        .bind(followup_id)
        .bind(gate_id)
        .bind(actor)
        .bind(outcome)
        .bind(note)
        .bind(Json(details))
        .fetch_one(&self.pool)
        .await?;

        Ok(models::ChangeGateFollowup::from(row))
    }

    /// Fetches all follow-up records for a gate in chronological order.
    pub async fn fetch_change_gate_followups(
        &self,
        gate_id: Uuid,
    ) -> Result<Vec<models::ChangeGateFollowup>> {
        let rows = sqlx::query_as::<_, db::ChangeGateFollowupRow>(
            r#"
            select *
            from change_gate_followups
            where gate_id = $1
            order by recorded_at asc
            "#,
        )
        .bind(gate_id)
        .fetch_all(&self.pool)
        .await?;

        Ok(rows
            .into_iter()
            .map(models::ChangeGateFollowup::from)
            .collect())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::InMemoryStorage;
    use crate::RunStore;
    use bytes::Bytes;
    use serde_json::json;
    use std::env;
    use testcontainers::clients::Cli;
    use testcontainers::images::postgres::Postgres;
    use testcontainers::Container;
    use tokio::sync::Barrier;
    use uuid::Uuid;

    fn docker_available() -> bool {
        std::fs::metadata("/var/run/docker.sock").is_ok() || std::env::var("DOCKER_HOST").is_ok()
    }

    async fn setup_storage() -> Result<(Storage, Container<'static, Postgres>)> {
        let docker = Cli::default();
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);
        let database_url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");

        let config = StorageConfig {
            database_url,
            max_connections: 2,
            connect_timeout: Duration::from_secs(10),
            object_store: ObjectStoreConfig::InMemory,
        };

        let storage = Storage::connect(config).await?;
        let container: Container<'static, Postgres> = unsafe { std::mem::transmute(container) };
        Ok((storage, container))
    }

    struct EnvGuard {
        keys: Vec<&'static str>,
        previous: Vec<Option<String>>,
    }

    impl EnvGuard {
        fn new(keys: Vec<&'static str>) -> Self {
            let previous = keys
                .iter()
                .map(|key| env::var(key).ok())
                .collect::<Vec<_>>();
            Self { keys, previous }
        }
    }

    impl Drop for EnvGuard {
        fn drop(&mut self) {
            for (idx, key) in self.keys.iter().enumerate() {
                if let Some(value) = &self.previous[idx] {
                    env::set_var(key, value);
                } else {
                    env::remove_var(key);
                }
            }
        }
    }

    fn seed_step(run_id: Uuid, step_id: Uuid) -> models::NewStep {
        models::NewStep {
            run_id,
            step_id,
            idx: 0,
            priority: 0,
            spec_json: json!({"id": step_id, "type": "tool", "inputs": {}}),
            status: models::StepStatus::Queued,
            attempt: 0,
            input_snapshot: None,
            provider: None,
            provider_version: None,
            pending_dependencies: 0,
            total_dependencies: 0,
            estimated_tokens: 0,
            estimated_cost: 0.0,
            actual_tokens: None,
            actual_cost: None,
            ..models::NewStep::default()
        }
    }

    #[test]
    fn s3_config_reads_expected_env() -> Result<()> {
        let guard = EnvGuard::new(vec![
            "FLEETFORGE_S3_BUCKET",
            "FLEETFORGE_S3_ACCESS_KEY",
            "FLEETFORGE_S3_SECRET_KEY",
            "FLEETFORGE_S3_REGION",
            "FLEETFORGE_S3_ENDPOINT",
        ]);

        env::set_var("FLEETFORGE_S3_BUCKET", "demo-bucket");
        env::set_var("FLEETFORGE_S3_ACCESS_KEY", "demo-access");
        env::set_var("FLEETFORGE_S3_SECRET_KEY", "demo-secret");
        env::set_var("FLEETFORGE_S3_REGION", "us-west-2");
        env::set_var("FLEETFORGE_S3_ENDPOINT", "https://s3.demo");

        let cfg = S3Config::from_env()?;
        assert_eq!(cfg.bucket, "demo-bucket");
        assert_eq!(cfg.access_key, "demo-access");
        assert_eq!(cfg.secret_key, "demo-secret");
        assert_eq!(cfg.region.as_deref(), Some("us-west-2"));
        assert_eq!(cfg.endpoint.as_deref(), Some("https://s3.demo"));
        assert!(cfg.virtual_hosted_style);

        drop(guard);
        Ok(())
    }

    #[cfg_attr(not(feature = "docker-tests"), ignore = "requires Docker")]
    #[tokio::test]
    async fn migrations_apply_cleanly() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping storage::migrations test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;

        let exists: (i64,) = sqlx::query_as(
            "select count(*) from information_schema.tables where table_name = 'runs'",
        )
        .fetch_one(storage.pool())
        .await?;
        assert!(
            exists.0 >= 0,
            "runs table should be accessible after migrations"
        );

        // DAG gating: downstream steps remain blocked until predecessors succeed.
        let run_one = Uuid::new_v4();
        let step_root = Uuid::new_v4();
        let step_leaf = Uuid::new_v4();

        let root_spec = json!({"id": step_root, "type": "tool", "inputs": {}});
        let leaf_spec = json!({"id": step_leaf, "type": "tool", "inputs": {}});
        let dag_one = json!({
            "steps": [root_spec.clone(), leaf_spec.clone()],
            "edges": [[step_root, step_leaf]]
        });

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id: run_one,
                    status: models::RunStatus::Pending,
                    dag_json: dag_one,
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[
                    models::NewStep {
                        run_id: run_one,
                        step_id: step_root,
                        idx: 0,
                        priority: 0,
                        spec_json: root_spec,
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id: run_one,
                        step_id: step_leaf,
                        idx: 1,
                        priority: 0,
                        spec_json: leaf_spec,
                        status: models::StepStatus::Blocked,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 1,
                        total_dependencies: 1,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[(step_root, step_leaf)],
            )
            .await?;

        let leased = storage.claim_queued_steps(1, "worker-1").await?;
        assert_eq!(leased.len(), 1);
        assert_eq!(leased[0].status, models::StepStatus::Leased);

        let (run_status_one,): (String,) =
            sqlx::query_as("select status from runs where run_id = $1")
                .bind(run_one)
                .fetch_one(storage.pool())
                .await?;
        assert_eq!(run_status_one, "running");

        let mut tx = storage.pool().begin().await?;
        storage
            .update_step_status_tx(
                &mut tx,
                run_one,
                step_root,
                models::StepStatus::Succeeded,
                Some(1),
                None,
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .await?;
        let unlocked = storage
            .unlock_successors_tx(&mut tx, run_one, step_root)
            .await?;
        assert_eq!(unlocked, vec![step_leaf]);
        tx.commit().await?;

        let (leaf_status, leaf_pending): (String, i32) = sqlx::query_as(
            "select status, pending_dependencies from steps where run_id = $1 and step_id = $2",
        )
        .bind(run_one)
        .bind(step_leaf)
        .fetch_one(storage.pool())
        .await?;
        assert_eq!(leaf_status.as_str(), "queued");
        assert_eq!(leaf_pending, 0);

        // Completing the final step should mark the run succeeded.
        let mut tx = storage.pool().begin().await?;
        storage
            .update_step_status_tx(
                &mut tx,
                run_one,
                step_leaf,
                models::StepStatus::Succeeded,
                Some(2),
                None,
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .await?;
        let (_, new_status) = storage.refresh_run_status_tx(&mut tx, run_one).await?;
        assert_eq!(new_status, models::RunStatus::Succeeded);
        tx.commit().await?;

        // Failing a root step cascades skips and marks the run failed.
        let run_two = Uuid::new_v4();
        let step_a = Uuid::new_v4();
        let step_b = Uuid::new_v4();
        let step_c = Uuid::new_v4();

        let spec_a = json!({"id": step_a, "type": "tool", "inputs": {}});
        let spec_b = json!({"id": step_b, "type": "tool", "inputs": {}});
        let spec_c = json!({"id": step_c, "type": "tool", "inputs": {}});
        let dag_two = json!({
            "steps": [spec_a.clone(), spec_b.clone(), spec_c.clone()],
            "edges": [[step_a, step_b], [step_b, step_c]]
        });

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id: run_two,
                    status: models::RunStatus::Pending,
                    dag_json: dag_two,
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[
                    models::NewStep {
                        run_id: run_two,
                        step_id: step_a,
                        idx: 0,
                        priority: 0,
                        spec_json: spec_a,
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id: run_two,
                        step_id: step_b,
                        idx: 1,
                        priority: 0,
                        spec_json: spec_b,
                        status: models::StepStatus::Blocked,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 1,
                        total_dependencies: 1,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id: run_two,
                        step_id: step_c,
                        idx: 2,
                        priority: 0,
                        spec_json: spec_c,
                        status: models::StepStatus::Blocked,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 1,
                        total_dependencies: 1,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[(step_a, step_b), (step_b, step_c)],
            )
            .await?;

        storage.claim_queued_steps(1, "worker-2").await?;
        let (run_two_status,): (String,) =
            sqlx::query_as("select status from runs where run_id = $1")
                .bind(run_two)
                .fetch_one(storage.pool())
                .await?;
        assert_eq!(run_two_status, "running");

        let mut tx = storage.pool().begin().await?;
        storage
            .update_step_status_tx(
                &mut tx,
                run_two,
                step_a,
                models::StepStatus::Failed,
                Some(1),
                None,
                None,
                None,
                None,
                None,
                None,
                None,
            )
            .await?;
        let skipped = storage
            .skip_downstream_steps_tx(&mut tx, run_two, step_a)
            .await?;
        assert_eq!(skipped.len(), 2);
        assert!(skipped.contains(&step_b) && skipped.contains(&step_c));
        let (_, failed_status) = storage.refresh_run_status_tx(&mut tx, run_two).await?;
        assert_eq!(failed_status, models::RunStatus::Failed);
        tx.commit().await?;

        for target in [step_b, step_c] {
            let (status,): (String,) =
                sqlx::query_as("select status from steps where run_id = $1 and step_id = $2")
                    .bind(run_two)
                    .bind(target)
                    .fetch_one(storage.pool())
                    .await?;
            assert_eq!(status, "skipped");
        }

        Ok(())
    }

    #[tokio::test]
    async fn in_memory_store_supports_idempotency() -> Result<()> {
        use crate::RunStore;

        let store = InMemoryStorage::new();
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();
        let dag = json!({
            "steps": [{
                "id": step_id,
                "type": "tool",
                "inputs": {}
            }]
        });

        let first = store
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: dag.clone(),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: Some("mem-key".into()),
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 0,
                    spec_json: json!({ "id": step_id, "type": "tool", "inputs": {} }),
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await?;
        assert_eq!(first, run_id);

        let reused = store
            .insert_run_with_steps(
                models::NewRun {
                    run_id: Uuid::new_v4(),
                    status: models::RunStatus::Pending,
                    dag_json: dag,
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: Some("mem-key".into()),
                },
                &[],
                &[],
            )
            .await?;
        assert_eq!(
            reused, run_id,
            "in-memory store should reuse idempotent run id"
        );

        let fetched = store.fetch_run(run_id).await?.expect("run exists");
        assert_eq!(fetched.run_id, run_id);
        let steps = store.fetch_steps_for_run(run_id).await?;
        assert_eq!(steps.len(), 1);

        Ok(())
    }

    #[cfg_attr(not(feature = "docker-tests"), ignore = "requires Docker")]
    #[tokio::test]
    async fn idempotency_reuses_existing_run() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping idempotency test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;
        let idempotency_key = "reuse-key";
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();
        let step_spec = json!({
            "id": step_id,
            "type": "tool",
            "inputs": {}
        });
        let dag = json!({ "steps": [step_spec.clone()] });

        let inserted = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: dag.clone(),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: Some(idempotency_key.into()),
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 0,
                    spec_json: step_spec,
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await?;
        assert_eq!(inserted, run_id, "initial insert should return run id");

        let reused = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id: Uuid::new_v4(),
                    status: models::RunStatus::Pending,
                    dag_json: dag,
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: Some(idempotency_key.into()),
                },
                &[],
                &[],
            )
            .await?;
        assert_eq!(
            reused, run_id,
            "second insert with same idempotency key should reuse original run id"
        );

        drop(container);
        Ok(())
    }

    #[cfg_attr(not(feature = "docker-tests"), ignore = "requires Docker")]
    #[tokio::test]
    async fn claim_queued_orders_by_priority_then_idx() -> Result<()> {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping storage queue test because Docker is unavailable");
            return Ok(());
        }

        let docker = Cli::default();
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);
        let database_url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");

        let storage = Storage::connect(StorageConfig {
            database_url,
            max_connections: 1,
            connect_timeout: Duration::from_secs(10),
            object_store: ObjectStoreConfig::InMemory,
        })
        .await?;

        let run_id = Uuid::new_v4();
        let step_low = Uuid::new_v4();
        let step_high_a = Uuid::new_v4();
        let step_high_b = Uuid::new_v4();

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": [], "edges": []}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[
                    models::NewStep {
                        run_id,
                        step_id: step_low,
                        idx: 2,
                        priority: 1,
                        spec_json: json!({"id": step_low, "type": "tool", "inputs": {}, "policy": {"priority": 1}}),
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id,
                        step_id: step_high_a,
                        idx: 0,
                        priority: 5,
                        spec_json: json!({"id": step_high_a, "type": "tool", "inputs": {}, "policy": {"priority": 5}}),
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id,
                        step_id: step_high_b,
                        idx: 1,
                        priority: 5,
                        spec_json: json!({"id": step_high_b, "type": "tool", "inputs": {}, "policy": {"priority": 5}}),
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[],
            )
            .await?;

        let first = storage
            .claim_queued_steps(2, "worker-a")
            .await?
            .into_iter()
            .map(models::Step::try_from)
            .collect::<Result<Vec<_>>>()?;
        assert_eq!(first.len(), 2);
        assert_eq!(first[0].step_id, step_high_a);
        assert_eq!(first[1].step_id, step_high_b);
        assert!(first.iter().all(|s| s.status == models::StepStatus::Leased));

        let second = storage
            .claim_queued_steps(2, "worker-b")
            .await?
            .into_iter()
            .map(models::Step::try_from)
            .collect::<Result<Vec<_>>>()?;
        assert_eq!(second.len(), 1);
        assert_eq!(second[0].step_id, step_low);
        assert_eq!(second[0].status, models::StepStatus::Leased);

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn list_recent_runs_returns_most_recent_first() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping list_recent_runs test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;

        let run_one = Uuid::new_v4();
        let step_one = Uuid::new_v4();
        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id: run_one,
                    status: models::RunStatus::Pending,
                    dag_json: json!({ "steps": [], "edges": [] }),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id: run_one,
                    step_id: step_one,
                    idx: 0,
                    priority: 0,
                    spec_json: json!({"id": step_one, "type": "tool"}),
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await?;

        let run_two = Uuid::new_v4();
        let step_two = Uuid::new_v4();
        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id: run_two,
                    status: models::RunStatus::Pending,
                    dag_json: json!({ "steps": [], "edges": [] }),
                    input_ctx: json!({}),
                    seed: 2,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id: run_two,
                    step_id: step_two,
                    idx: 0,
                    priority: 0,
                    spec_json: json!({"id": step_two, "type": "tool"}),
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await?;

        let runs = storage.list_recent_runs(10).await?;
        assert!(
            runs.len() >= 2,
            "expected at least two runs in recent listing"
        );
        assert_eq!(
            runs.first().map(|run| run.run_id),
            Some(run_two),
            "most recent run should appear first"
        );

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn run_roundtrip_insert_and_fetch() -> Result<()> {
        if std::fs::metadata("/var/run/docker.sock").is_err()
            && std::env::var("DOCKER_HOST").is_err()
        {
            eprintln!("Skipping storage::roundtrip test because Docker is unavailable");
            return Ok(());
        }

        let docker = Cli::default();
        let container = docker.run(Postgres::default());
        let port = container.get_host_port_ipv4(5432);
        let database_url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");

        let storage = Storage::connect(StorageConfig {
            database_url,
            max_connections: 1,
            connect_timeout: Duration::from_secs(10),
            object_store: ObjectStoreConfig::InMemory,
        })
        .await?;

        let run_id = Uuid::new_v4();
        let step_root = Uuid::new_v4();
        let step_leaf = Uuid::new_v4();

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({
                        "steps": [
                            {"id": step_root, "type": "tool", "inputs": {}},
                            {"id": step_leaf, "type": "tool", "inputs": {}}
                        ],
                        "edges": [[step_root, step_leaf]]
                    }),
                    input_ctx: json!({"labels": {"example": "roundtrip"}}),
                    seed: 99,
                    idempotency_key: Some("round-trip".into()),
                },
                &[
                    models::NewStep {
                        run_id,
                        step_id: step_root,
                        idx: 0,
                        priority: 5,
                        spec_json: json!({"id": step_root, "type": "tool", "inputs": {}, "policy": {"priority": 5}}),
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 10,
                        estimated_cost: 0.1,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                    models::NewStep {
                        run_id,
                        step_id: step_leaf,
                        idx: 1,
                        priority: 1,
                        spec_json: json!({"id": step_leaf, "type": "tool", "inputs": {}, "policy": {"priority": 1}}),
                        status: models::StepStatus::Blocked,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 1,
                        total_dependencies: 1,
                        estimated_tokens: 5,
                        estimated_cost: 0.05,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    },
                ],
                &[(step_root, step_leaf)],
            )
            .await?;

        let fetched_run = storage.fetch_run(run_id).await?.expect("run should exist");
        assert_eq!(fetched_run.status, models::RunStatus::Pending);
        assert_eq!(fetched_run.seed, 99);
        assert_eq!(fetched_run.idempotency_key.as_deref(), Some("round-trip"));

        let steps = storage.fetch_steps_for_run(run_id).await?;
        assert_eq!(steps.len(), 2);
        let root = steps.iter().find(|s| s.step_id == step_root).unwrap();
        let leaf = steps.iter().find(|s| s.step_id == step_leaf).unwrap();
        assert_eq!(root.status, models::StepStatus::Queued);
        assert_eq!(leaf.status, models::StepStatus::Blocked);
        assert_eq!(root.priority, 5);
        assert_eq!(leaf.priority, 1);

        let from_key = storage
            .fetch_run_by_idempotency_key("round-trip")
            .await?
            .expect("run by key");
        assert_eq!(from_key.run_id, run_id);

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn return_step_to_queue_round_trip() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping return_step_to_queue test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;

        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();
        let step_spec = json!({"id": step_id, "type": "tool", "inputs": {}});

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": [step_spec.clone()]}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 0,
                    spec_json: step_spec,
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await?;

        let leased = storage.claim_queued_steps(1, "worker-test").await?;
        assert_eq!(leased.len(), 1);

        storage.return_step_to_queue(run_id, step_id).await?;

        let (status, leased_by): (String, Option<String>) = sqlx::query_as(
            "select status, leased_by from steps where run_id = $1 and step_id = $2",
        )
        .bind(run_id)
        .bind(step_id)
        .fetch_one(storage.pool())
        .await?;

        assert_eq!(status, "queued");
        assert!(leased_by.is_none());

        drop(container);
        Ok(())
    }

    #[cfg_attr(not(feature = "docker-tests"), ignore = "requires Docker")]
    #[tokio::test]
    async fn requeue_expired_leases_requeues_all() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping requeue_expired_leases test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;

        let run_id = Uuid::new_v4();
        let steps: Vec<(Uuid, serde_json::Value)> = (0..3)
            .map(|idx| {
                let step_id = Uuid::new_v4();
                (
                    step_id,
                    json!({
                        "id": step_id,
                        "type": "tool",
                        "inputs": {},
                        "policy": {"budget": {"tokens": idx + 1}}
                    }),
                )
            })
            .collect();

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": steps.iter().map(|(_, spec)| spec.clone()).collect::<Vec<_>>()}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &steps
                    .iter()
                    .enumerate()
                    .map(|(idx, (step_id, spec))| models::NewStep {
                        run_id,
                        step_id: *step_id,
                        idx: idx as i32,
                        priority: 0,
                        spec_json: spec.clone(),
                        status: models::StepStatus::Queued,
                        attempt: 0,
                        input_snapshot: None,
                        provider: None,
                        provider_version: None,
                        pending_dependencies: 0,
                        total_dependencies: 0,
                        estimated_tokens: 0,
                        estimated_cost: 0.0,
                        actual_tokens: None,
                        actual_cost: None,
                        ..models::NewStep::default()
                    })
                    .collect::<Vec<_>>(),
                &[],
            )
            .await?;

        let _claimed = storage.claim_queued_steps(3, "worker-requeue").await?;
        sqlx::query(
            "update steps set lease_expires_at = now() - interval '5 minutes' where run_id = $1",
        )
        .bind(run_id)
        .execute(storage.pool())
        .await?;

        let requeued = storage.requeue_expired_leases(10).await?;
        assert_eq!(requeued.len(), steps.len());
        let requeued_ids: Vec<Uuid> = requeued.iter().map(|(_, step)| *step).collect();
        for (step_id, _) in &steps {
            assert!(requeued_ids.contains(step_id));
        }

        let rows: Vec<(String, Option<String>)> =
            sqlx::query_as("select status, leased_by from steps where run_id = $1 order by idx")
                .bind(run_id)
                .fetch_all(storage.pool())
                .await?;
        for (status, leased_by) in rows {
            assert_eq!(status, "queued");
            assert!(leased_by.is_none());
        }

        drop(container);
        Ok(())
    }

    #[cfg_attr(not(feature = "docker-tests"), ignore = "requires Docker")]
    #[tokio::test]
    async fn idempotency_prevents_duplicate_runs() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping idempotency test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;
        let key = "dup-key";

        let run_one = Uuid::new_v4();
        let step_one = Uuid::new_v4();
        let step_spec = json!({"id": step_one, "type": "tool", "inputs": {}});

        let _ = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id: run_one,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": [step_spec.clone()]}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: Some(key.to_string()),
                },
                &[models::NewStep {
                    run_id: run_one,
                    step_id: step_one,
                    idx: 0,
                    priority: 0,
                    spec_json: step_spec.clone(),
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await?;

        let run_two = Uuid::new_v4();
        let duplicate_id = storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id: run_two,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": [step_spec.clone()]}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: Some(key.to_string()),
                },
                &[models::NewStep {
                    run_id: run_two,
                    step_id: Uuid::new_v4(),
                    idx: 0,
                    priority: 0,
                    spec_json: step_spec,
                    status: models::StepStatus::Queued,
                    attempt: 0,
                    input_snapshot: None,
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: None,
                    actual_cost: None,
                    ..models::NewStep::default()
                }],
                &[],
            )
            .await?;

        assert_eq!(
            duplicate_id, run_one,
            "idempotent submission should reuse run_id"
        );

        let steps = storage.fetch_steps_for_run(run_one).await?;
        assert_eq!(
            steps.len(),
            1,
            "duplicate submission should not add extra steps"
        );

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn audit_chain_attaches_metadata() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping audit chain test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": []}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[seed_step(run_id, step_id)],
                &[],
            )
            .await?;

        let mut tx = storage.pool().begin().await?;
        storage
            .enqueue_outbox_event(
                &mut tx,
                models::NewOutboxEvent {
                    run_id,
                    step_id: Some(step_id),
                    kind: "step_started".to_string(),
                    payload: json!({"status": "running"}),
                },
            )
            .await?;
        storage
            .enqueue_outbox_event(
                &mut tx,
                models::NewOutboxEvent {
                    run_id,
                    step_id: Some(step_id),
                    kind: "step_succeeded".to_string(),
                    payload: json!({"status": "succeeded"}),
                },
            )
            .await?;
        tx.commit().await?;

        let events = storage.fetch_outbox_for_run_since(run_id, 0, 10).await?;
        assert_eq!(events.len(), 2);

        let first_audit = &events[0].payload["_audit"];
        let second_audit = &events[1].payload["_audit"];
        assert_eq!(first_audit["version"], json!(1));
        assert!(first_audit["chain_hash"].as_str().is_some());
        assert!(first_audit["previous_hash"].is_null());

        let first_hash = first_audit["chain_hash"].as_str().unwrap();
        assert_eq!(second_audit["previous_hash"].as_str(), Some(first_hash));
        assert!(second_audit["chain_hash"].as_str().is_some());

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn audit_chain_serializes_concurrent_writers() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping audit concurrency test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;
        let storage = Arc::new(storage);
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": []}),
                    input_ctx: json!({}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[seed_step(run_id, step_id)],
                &[],
            )
            .await?;

        let barrier = Arc::new(Barrier::new(3));
        let mut handles = Vec::new();
        for (kind, status) in [
            ("step_running".to_string(), "running".to_string()),
            ("step_succeeded".to_string(), "succeeded".to_string()),
        ] {
            let storage = Arc::clone(&storage);
            let barrier = Arc::clone(&barrier);
            let handle = tokio::spawn(async move {
                barrier.wait().await;
                let mut tx = storage.pool().begin().await?;
                storage
                    .enqueue_outbox_event(
                        &mut tx,
                        models::NewOutboxEvent {
                            run_id,
                            step_id: Some(step_id),
                            kind,
                            payload: json!({"status": status}),
                        },
                    )
                    .await?;
                tx.commit().await?;
                Ok::<(), anyhow::Error>(())
            });
            handles.push(handle);
        }

        barrier.wait().await;
        for handle in handles {
            handle.await??;
        }

        let events = storage.fetch_outbox_for_run_since(run_id, 0, 10).await?;
        assert_eq!(events.len(), 2);
        let first = &events[0].payload["_audit"];
        let second = &events[1].payload["_audit"];
        assert_eq!(
            second["previous_hash"].as_str(),
            first["chain_hash"].as_str()
        );

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn tenant_quota_enforced() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping quota test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;
        let tenant = "tenant-alpha";
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": []}),
                    input_ctx: json!({"labels": {"tenant": tenant}}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[seed_step(run_id, step_id)],
                &[],
            )
            .await?;

        sqlx::query(
            "insert into tenant_quotas (tenant, daily_token_limit, daily_cost_limit) values ($1, $2, $3)",
        )
        .bind(tenant)
        .bind(1000_i64)
        .bind(10.0_f64)
        .execute(storage.pool())
        .await?;

        storage.enforce_tenant_quota(tenant, 100, 1.0).await?;

        let mut tx = storage.pool().begin().await?;
        storage
            .record_billing_usage_tx(&mut tx, run_id, 900, 5.0)
            .await?;
        tx.commit().await?;

        let err = storage
            .enforce_tenant_quota(tenant, 200, 1.0)
            .await
            .expect_err("quota should block overage");
        assert!(format!("{err}").contains("quota"));

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn retention_policy_scrubs_payloads() -> Result<()> {
        if !docker_available() {
            eprintln!("Skipping retention test because Docker is unavailable");
            return Ok(());
        }

        let (storage, container) = setup_storage().await?;
        let run_id = Uuid::new_v4();
        let step_id = Uuid::new_v4();

        storage
            .insert_run_with_steps(
                models::NewRun {
                    run_id,
                    status: models::RunStatus::Pending,
                    dag_json: json!({"steps": []}),
                    input_ctx: json!({"inputs": {"secret": "value"}, "labels": {"team": "core"}}),
                    seed: 1,
                    idempotency_key: None,
                },
                &[models::NewStep {
                    run_id,
                    step_id,
                    idx: 0,
                    priority: 0,
                    spec_json: json!({"id": step_id, "type": "tool", "inputs": {}}),
                    status: models::StepStatus::Succeeded,
                    attempt: 1,
                    input_snapshot: Some(json!({"secret": "value"})),
                    provider: None,
                    provider_version: None,
                    pending_dependencies: 0,
                    total_dependencies: 0,
                    estimated_tokens: 0,
                    estimated_cost: 0.0,
                    actual_tokens: Some(5),
                    actual_cost: Some(0.01),
                    leased_by: None,
                    lease_expires_at: None,
                    output_snapshot: Some(json!({"result": "ok"})),
                    output_json: Some(json!({"result": "ok"})),
                    error_json: Some(json!({"message": "none"})),
                }],
                &[],
            )
            .await?;

        {
            let mut tx = storage.pool().begin().await?;
            storage
                .enqueue_outbox_event(
                    &mut tx,
                    models::NewOutboxEvent {
                        run_id,
                        step_id: Some(step_id),
                        kind: "step_succeeded".to_string(),
                        payload: json!({
                            "status": "succeeded",
                            "input": {"secret": "value"},
                            "output": {"result": "ok"},
                            "snapshots": {"output": {"result": "ok"}}
                        }),
                    },
                )
                .await?;
            tx.commit().await?;
        }

        sqlx::query("update runs set created_at = now() - interval '10 days' where run_id = $1")
            .bind(run_id)
            .execute(storage.pool())
            .await?;
        sqlx::query("update steps set created_at = now() - interval '10 days' where run_id = $1")
            .bind(run_id)
            .execute(storage.pool())
            .await?;

        let cutoff = Utc::now() - chrono::Duration::days(7);
        let stats = storage.apply_retention_policy(cutoff).await?;
        assert!(stats.runs_scrubbed >= 1);
        assert!(stats.steps_scrubbed >= 1);
        assert!(stats.events_scrubbed >= 1);

        let run = storage.fetch_run(run_id).await?.expect("run exists");
        assert_eq!(run.input_ctx.get("inputs"), Some(&json!({})));
        assert_eq!(run.input_ctx.get("labels"), Some(&json!({})));

        let step = storage
            .fetch_steps_for_run(run_id)
            .await?
            .into_iter()
            .next()
            .unwrap();
        assert!(step.input_snapshot.is_none());
        assert!(step.output_snapshot.is_none());
        assert!(step.output_json.is_none());
        assert!(step.error_json.is_none());

        let events = storage.fetch_outbox_for_run_since(run_id, 0, 10).await?;
        assert_eq!(events.len(), 1);
        let payload = &events[0].payload;
        assert_eq!(payload.get("status"), Some(&json!("succeeded")));
        assert!(payload.get("input").is_none());
        assert!(payload.get("output").is_none());
        assert!(payload.get("snapshots").is_none());
        assert!(payload.get("_audit").is_some());

        drop(container);
        Ok(())
    }

    #[tokio::test]
    async fn artifact_store_put_get_presign() -> Result<()> {
        let store = artifacts::ObjectStoreArtifactStore::new(
            Arc::new(object_store::memory::InMemory::new()),
            None,
        );
        let payload = Bytes::from_static(b"hello world");
        let metadata = json!({"tag": "greeting"});

        let meta = store
            .put(payload.clone(), "text/plain", metadata.clone())
            .await?;
        let fetched = store.get(meta.sha256).await?;
        assert_eq!(fetched, payload);

        let url = store
            .presign_get(meta.sha256, Duration::from_secs(30))
            .await?;
        assert!(url.as_str().contains(&hex::encode(meta.sha256)));

        Ok(())
    }
}