tandem-server 0.6.9

use std::path::{Path, PathBuf};

use anyhow::{bail, Context};

use crate::stateful_runtime::backend::{
    self, params, Connection, Executor, ExecutorRaw as _, OptionalExtension, TransactionBehavior,
};
use tandem_automation::{
    validate_orchestration_spec, AutomationV2RunRecord, GoalRunLink, LongRunningGoal,
    LongRunningGoalStatus, OrchestrationSpec, OrchestrationStatus, WorkflowHandoff,
    WorkflowHandoffStatus,
};

mod definitions;
mod engine_lock;
mod goal_control;
mod goal_lifecycle;
mod migration;
pub(crate) mod protected_records;
mod runtime_records;
mod transfer;
mod transition;

pub use definitions::{DRAFT_CONCURRENCY_CONFLICT, ORCHESTRATION_DRAFT_VERSION};
pub use engine_lock::{read_engine_lock_owner, EngineLockOwner, StatefulEngineLock};
pub use goal_control::{GoalCancellationResult, GoalControlOutcome};
pub use goal_lifecycle::{GoalEventRow, GoalPauseOutcome, GoalResumeOutcome, StartGoalOutcome};
pub use migration::{
    LegacyImportContext, LegacyRuntimeMigrationPaths, LegacyRuntimeMigrationReport,
};
pub use transfer::{
    migrate_stateful_storage_backend, StatefulBackendKind, StatefulBackendMigrationReport,
    StatefulBackendMigrationRequest,
};
pub use transition::{
    GovernedTransitionRequest, GovernedTransitionResult, OrchestrationTransitionAuthority,
    WorkflowCompletionResult,
};

pub(crate) const SCHEMA_VERSION: i64 = 5;

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OrchestrationStorePaths {
    pub database_path: PathBuf,
    pub engine_lock_path: PathBuf,
}

impl OrchestrationStorePaths {
    pub fn from_runtime_events_path(runtime_events_path: &Path) -> Self {
        let root = canonical_stateful_runtime_root(runtime_events_path);
        Self {
            database_path: root.join("stateful_runtime.sqlite3"),
            engine_lock_path: root.join("stateful_runtime.engine.lock"),
        }
    }

    pub fn from_automation_runs_path(automation_runs_path: &Path) -> Self {
        let root = automation_runs_path
            .parent()
            .unwrap_or_else(|| Path::new("."));
        Self {
            database_path: root.join("stateful_runtime.sqlite3"),
            engine_lock_path: root.join("stateful_runtime.engine.lock"),
        }
    }
}

fn canonical_stateful_runtime_root(path: &Path) -> &Path {
    let parent = path.parent().unwrap_or_else(|| Path::new("."));
    if parent.file_name().is_some_and(|name| name == "runtime") {
        parent.parent().unwrap_or(parent)
    } else {
        parent
    }
}

#[derive(Debug, Clone)]
pub struct OrchestrationStateStore {
    paths: OrchestrationStorePaths,
    backend: StoreBackendSelection,
}

/// Which execution backend this store instance talks to. Selected once at
/// open time (from `TANDEM_STORAGE_BACKEND` or an explicit target) and fixed
/// for the store's lifetime; the two backends never mix within one store.
#[derive(Debug, Clone)]
enum StoreBackendSelection {
    #[cfg(feature = "storage-sqlite")]
    Sqlite,
    #[cfg(feature = "storage-postgres")]
    Postgres(backend::postgres::PostgresTarget),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AtomicHandoffCommit {
    Committed,
    AlreadyCommitted,
}

impl OrchestrationStateStore {
    pub fn from_runtime_events_path(runtime_events_path: &Path) -> anyhow::Result<Self> {
        Self::open(OrchestrationStorePaths::from_runtime_events_path(
            runtime_events_path,
        ))
    }

    pub fn from_automation_runs_path(automation_runs_path: &Path) -> anyhow::Result<Self> {
        Self::open(OrchestrationStorePaths::from_automation_runs_path(
            automation_runs_path,
        ))
    }

    pub fn open(paths: OrchestrationStorePaths) -> anyhow::Result<Self> {
        Self::open_with_config(paths, backend::StorageBackendConfig::from_env()?)
    }

    /// Acquires the process-lifetime engine lock before the store can be used
    /// by the runtime. SQLite takes its filesystem lock before the database is
    /// opened so a losing process cannot initialize or migrate the live file.
    /// PostgreSQL takes both the local lock and the advisory lock before its
    /// schema is initialized.
    pub fn acquire_engine_lock_for_runtime(
        paths: OrchestrationStorePaths,
    ) -> anyhow::Result<StatefulEngineLock> {
        Self::acquire_engine_lock_for_runtime_with_config(
            paths,
            backend::StorageBackendConfig::from_env()?,
        )
    }

    fn acquire_engine_lock_for_runtime_with_config(
        paths: OrchestrationStorePaths,
        config: backend::StorageBackendConfig,
    ) -> anyhow::Result<StatefulEngineLock> {
        match config {
            backend::StorageBackendConfig::Sqlite => {
                StatefulEngineLock::acquire(&paths.engine_lock_path)
            }
            config @ backend::StorageBackendConfig::Postgres { .. } => {
                let store = Self::open_with_config_inner(paths, config)?;
                let engine_lock = store.acquire_engine_lock()?;
                store.initialize()?;
                transfer::ensure_target_is_not_mid_transfer(&store)?;
                Ok(engine_lock)
            }
        }
    }

    /// Opens the store against an explicit backend target. Production code
    /// goes through [`Self::open`] (environment-selected); tests use this to
    /// exercise a specific backend without mutating process environment.
    pub(crate) fn open_with_config(
        paths: OrchestrationStorePaths,
        config: backend::StorageBackendConfig,
    ) -> anyhow::Result<Self> {
        let store = Self::open_with_config_inner(paths, config)?;
        store.initialize()?;
        transfer::ensure_target_is_not_mid_transfer(&store)?;
        Ok(store)
    }

    pub(super) fn open_for_backend_transfer_source(
        paths: OrchestrationStorePaths,
        config: backend::StorageBackendConfig,
    ) -> anyhow::Result<Self> {
        Self::open_with_config_inner(paths, config)
    }

    pub(super) fn open_for_backend_transfer_target(
        paths: OrchestrationStorePaths,
        config: backend::StorageBackendConfig,
    ) -> anyhow::Result<Self> {
        let store = Self::open_with_config_inner(paths, config)?;
        store.initialize()?;
        transfer::ensure_backend_transfer_schema(&store)?;
        Ok(store)
    }

    fn open_with_config_inner(
        paths: OrchestrationStorePaths,
        config: backend::StorageBackendConfig,
    ) -> anyhow::Result<Self> {
        if let Some(parent) = paths.database_path.parent() {
            std::fs::create_dir_all(parent)?;
        }
        let backend = match config {
            backend::StorageBackendConfig::Sqlite => {
                #[cfg(feature = "storage-sqlite")]
                {
                    StoreBackendSelection::Sqlite
                }
                #[cfg(not(feature = "storage-sqlite"))]
                {
                    bail!(
                        "storage backend `sqlite` requested but this build omits the \
                         `storage-sqlite` feature; set TANDEM_STORAGE_BACKEND=postgres \
                         or rebuild with SQLite support"
                    );
                }
            }
            backend::StorageBackendConfig::Postgres { url } => {
                #[cfg(feature = "storage-postgres")]
                {
                    StoreBackendSelection::Postgres(backend::postgres::PostgresTarget::for_root(
                        &url,
                        &paths.database_path,
                    )?)
                }
                #[cfg(not(feature = "storage-postgres"))]
                {
                    let _ = url;
                    bail!(
                        "storage backend `postgres` requested but this build omits the \
                         `storage-postgres` feature"
                    );
                }
            }
        };
        Ok(Self { paths, backend })
    }

    fn initialize(&self) -> anyhow::Result<()> {
        match &self.backend {
            #[cfg(feature = "storage-sqlite")]
            StoreBackendSelection::Sqlite => self.with_connection(|connection| {
                initialize_schema(
                    connection
                        .sqlite()
                        .expect("sqlite store opened a sqlite connection"),
                )
            }),
            #[cfg(feature = "storage-postgres")]
            StoreBackendSelection::Postgres(_) => {
                self.with_connection(backend::postgres::initialize_schema)
            }
        }
    }

    pub fn paths(&self) -> &OrchestrationStorePaths {
        &self.paths
    }

    pub fn acquire_engine_lock(&self) -> anyhow::Result<StatefulEngineLock> {
        let lock = StatefulEngineLock::acquire(&self.paths.engine_lock_path)?;
        match &self.backend {
            #[cfg(feature = "storage-sqlite")]
            StoreBackendSelection::Sqlite => Ok(lock),
            #[cfg(feature = "storage-postgres")]
            StoreBackendSelection::Postgres(target) => {
                // The file lock fences engines on this host; the advisory
                // lock fences engines on other hosts sharing the schema.
                Ok(lock.with_postgres_guard(target.acquire_advisory_lock()?))
            }
        }
    }

    #[cfg(feature = "storage-postgres")]
    pub(super) fn acquire_backend_transfer_target_guard(
        &self,
    ) -> anyhow::Result<Option<backend::postgres::PostgresAdvisoryLock>> {
        match &self.backend {
            #[cfg(feature = "storage-sqlite")]
            StoreBackendSelection::Sqlite => Ok(None),
            #[cfg(feature = "storage-postgres")]
            StoreBackendSelection::Postgres(target) => Ok(Some(target.acquire_advisory_lock()?)),
        }
    }

    #[cfg(not(feature = "storage-postgres"))]
    pub(super) fn acquire_backend_transfer_target_guard(&self) -> anyhow::Result<Option<()>> {
        Ok(None)
    }

    pub fn put_orchestration(&self, spec: &OrchestrationSpec) -> anyhow::Result<()> {
        let validation = validate_orchestration_spec(spec);
        if !validation.valid {
            bail!(
                "invalid orchestration {} version {}: {}",
                spec.orchestration_id,
                spec.version,
                validation
                    .issues
                    .iter()
                    .map(|issue| issue.code.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            );
        }
        let payload = protected_records::encode(
            &spec.tenant_context,
            "definition",
            &format!("{}:{}", spec.orchestration_id, spec.version),
            spec,
        )?;
        self.with_connection(|connection| {
            let existing = connection
                .query_row(
                    "SELECT status, definition_json FROM orchestration_specs
                     WHERE orchestration_id = ?1 AND version = ?2
                       AND org_id = ?3 AND workspace_id = ?4 AND deployment_key = ?5",
                    params![
                        spec.orchestration_id,
                        spec.version,
                        spec.tenant_context.org_id,
                        spec.tenant_context.workspace_id,
                        spec.tenant_context.deployment_id.as_deref().unwrap_or(""),
                    ],
                    |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
                )
                .optional()?;
            if let Some((status, existing_payload)) = existing {
                let existing_spec: OrchestrationSpec = protected_records::decode(
                    &spec.tenant_context,
                    "definition",
                    &format!("{}:{}", spec.orchestration_id, spec.version),
                    &existing_payload,
                )?;
                if status == "published" && existing_spec != *spec {
                    bail!(
                        "published orchestration {} version {} is immutable",
                        spec.orchestration_id,
                        spec.version
                    );
                }
                if status == "published" && matches!(spec.status, OrchestrationStatus::Draft) {
                    bail!("published orchestration versions cannot return to draft status");
                }
            }
            connection.execute(
                "INSERT INTO orchestration_specs (
                    orchestration_id, version, org_id, workspace_id, deployment_id, deployment_key,
                    status, definition_json, created_at_ms, updated_at_ms, published_at_ms
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
                 ON CONFLICT(org_id, workspace_id, deployment_key, orchestration_id, version)
                 DO UPDATE SET
                    status = excluded.status,
                    definition_json = excluded.definition_json,
                    updated_at_ms = excluded.updated_at_ms,
                    published_at_ms = excluded.published_at_ms",
                params![
                    spec.orchestration_id,
                    spec.version,
                    spec.tenant_context.org_id,
                    spec.tenant_context.workspace_id,
                    spec.tenant_context.deployment_id,
                    spec.tenant_context.deployment_id.as_deref().unwrap_or(""),
                    serde_json::to_value(&spec.status)?
                        .as_str()
                        .unwrap_or("draft"),
                    payload,
                    spec.created_at_ms,
                    spec.updated_at_ms,
                    spec.published_at_ms,
                ],
            )?;
            Ok(())
        })
    }

    pub fn get_orchestration(
        &self,
        orchestration_id: &str,
        version: u64,
    ) -> anyhow::Result<Option<OrchestrationSpec>> {
        self.with_connection(|connection| {
            let row = connection
                .query_row(
                    "SELECT org_id, workspace_id, deployment_id, definition_json
                     FROM orchestration_specs
                     WHERE orchestration_id = ?1 AND version = ?2",
                    params![orchestration_id, version],
                    |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, String>(1)?,
                            row.get::<_, Option<String>>(2)?,
                            row.get::<_, String>(3)?,
                        ))
                    },
                )
                .optional()?;
            row.map(|(org, workspace, deployment, payload)| {
                protected_records::decode_scoped(
                    &org,
                    &workspace,
                    deployment.as_deref(),
                    "definition",
                    &format!("{orchestration_id}:{version}"),
                    &payload,
                )
            })
            .transpose()
        })
    }

    pub fn get_orchestration_for_tenant(
        &self,
        tenant: &tandem_types::TenantContext,
        orchestration_id: &str,
        version: u64,
    ) -> anyhow::Result<Option<OrchestrationSpec>> {
        self.with_connection(|connection| {
            let payload = connection
                .query_row(
                    "SELECT definition_json FROM orchestration_specs
                     WHERE orchestration_id = ?1 AND version = ?2
                       AND org_id = ?3 AND workspace_id = ?4 AND deployment_key = ?5",
                    params![
                        orchestration_id,
                        version,
                        tenant.org_id,
                        tenant.workspace_id,
                        tenant.deployment_id.as_deref().unwrap_or(""),
                    ],
                    |row| row.get::<_, String>(0),
                )
                .optional()?;
            payload
                .map(|payload| {
                    protected_records::decode(
                        tenant,
                        "definition",
                        &format!("{orchestration_id}:{version}"),
                        &payload,
                    )
                })
                .transpose()
        })
    }

    pub fn upsert_automation_runs<'a>(
        &self,
        runs: impl IntoIterator<Item = &'a AutomationV2RunRecord>,
    ) -> anyhow::Result<usize> {
        let runs = runs.into_iter().collect::<Vec<_>>();
        self.with_connection(|connection| {
            let transaction =
                connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
            for run in &runs {
                upsert_automation_run(&transaction, run)?;
            }
            transaction.commit()?;
            Ok(runs.len())
        })
    }

    /// Replace the hot Automation V2 run index while retaining full historical
    /// rows for archive and audit reads.
    pub fn sync_hot_automation_runs<'a>(
        &self,
        runs: impl IntoIterator<Item = &'a AutomationV2RunRecord>,
    ) -> anyhow::Result<usize> {
        let runs = runs.into_iter().collect::<Vec<_>>();
        self.with_connection(|connection| {
            let transaction =
                connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
            transaction.execute(
                "UPDATE automation_runs SET is_hot = 0 WHERE is_hot != 0",
                [],
            )?;
            for run in &runs {
                upsert_automation_run(&transaction, run)?;
                transaction.execute(
                    "UPDATE automation_runs SET is_hot = 1 WHERE run_id = ?1",
                    [&run.run_id],
                )?;
            }
            transaction.commit()?;
            Ok(runs.len())
        })
    }

    pub fn load_automation_runs(&self) -> anyhow::Result<Vec<AutomationV2RunRecord>> {
        self.with_connection(|connection| {
            let mut statement = connection.prepare(
                "SELECT run_id, org_id, workspace_id, deployment_id, run_json FROM automation_runs
                 WHERE is_hot = 1 ORDER BY created_at_ms, run_id",
            )?;
            let rows = statement.query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, Option<String>>(3)?,
                    row.get::<_, String>(4)?,
                ))
            })?;
            let mut runs = Vec::new();
            for row in rows {
                let (id, org, workspace, deployment, payload) = row?;
                runs.push(protected_records::decode_scoped(
                    &org,
                    &workspace,
                    deployment.as_deref(),
                    "run",
                    &id,
                    &payload,
                )?);
            }
            Ok(runs)
        })
    }

    pub fn get_automation_run(
        &self,
        run_id: &str,
    ) -> anyhow::Result<Option<AutomationV2RunRecord>> {
        self.with_connection(|connection| {
            let row = connection
                .query_row(
                    "SELECT org_id, workspace_id, deployment_id, run_json
                     FROM automation_runs WHERE run_id = ?1",
                    [run_id],
                    |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, String>(1)?,
                            row.get::<_, Option<String>>(2)?,
                            row.get::<_, String>(3)?,
                        ))
                    },
                )
                .optional()?;
            row.map(|(org, workspace, deployment, payload)| {
                protected_records::decode_scoped(
                    &org,
                    &workspace,
                    deployment.as_deref(),
                    "run",
                    run_id,
                    &payload,
                )
            })
            .transpose()
        })
    }

    pub fn import_legacy_runs(
        &self,
        source_path: &Path,
        runs: &[AutomationV2RunRecord],
        imported_at_ms: u64,
    ) -> anyhow::Result<usize> {
        let source = source_path.to_string_lossy().to_string();
        self.with_connection(|connection| {
            let transaction =
                connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
            for run in runs {
                upsert_automation_run(&transaction, run)?;
            }
            transaction.execute(
                "INSERT INTO migration_sources
                    (source_kind, source_path, imported_at_ms, record_count)
                 VALUES ('automation_runs', ?1, ?2, ?3)
                 ON CONFLICT(source_kind, source_path) DO UPDATE SET
                    imported_at_ms = excluded.imported_at_ms,
                    record_count = excluded.record_count",
                params![source, imported_at_ms, runs.len() as u64],
            )?;
            transaction.commit()?;
            Ok(runs.len())
        })
    }

    /// Idempotently indexes legacy inbox/approved/archived JSON envelopes. The
    /// source files remain untouched as migration backups and compatibility data.
    pub fn import_legacy_handoff_directory(
        &self,
        handoff_root: &Path,
        imported_at_ms: u64,
    ) -> anyhow::Result<usize> {
        self.import_legacy_handoff_directory_with_context(
            handoff_root,
            &migration::LegacyImportContext::default(),
            imported_at_ms,
        )
    }

    pub fn import_legacy_handoff_directory_with_context(
        &self,
        handoff_root: &Path,
        context: &migration::LegacyImportContext,
        imported_at_ms: u64,
    ) -> anyhow::Result<usize> {
        // Envelopes naming automations this runtime has never known are
        // foreign and must quarantine instead of importing (they can never
        // trigger local work, only leak another root's records into this one).
        let mut known_automation_ids = self.with_connection(|connection| {
            let mut statement =
                connection.prepare("SELECT DISTINCT automation_id FROM automation_runs")?;
            let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
            rows.collect::<Result<std::collections::BTreeSet<_>, _>>()
                .map_err(Into::into)
        })?;
        known_automation_ids.extend(context.known_automation_ids.iter().cloned());
        let known = (!known_automation_ids.is_empty()).then_some(&known_automation_ids);
        let handoffs = migration::load_legacy_handoffs(Some(handoff_root), known)?;
        let source = handoff_root.to_string_lossy().to_string();
        self.with_connection(|connection| {
            let transaction =
                connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
            let mut imported = 0usize;
            for (path, handoff, status) in &handoffs.imported {
                imported += transaction.execute(
                    "INSERT INTO legacy_handoffs
                        (handoff_id, source_path, status, consumed_by_run_id, handoff_json,
                         created_at_ms, imported_at_ms)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
                     ON CONFLICT(handoff_id) DO UPDATE SET
                        source_path = excluded.source_path,
                        status = excluded.status,
                        consumed_by_run_id = excluded.consumed_by_run_id,
                        handoff_json = excluded.handoff_json,
                        imported_at_ms = excluded.imported_at_ms",
                    params![
                        handoff.handoff_id,
                        path.to_string_lossy(),
                        status,
                        handoff.consumed_by_run_id,
                        protected_records::encode(
                            &tandem_types::TenantContext::local_implicit(),
                            "legacy_handoff",
                            &handoff.handoff_id,
                            handoff,
                        )?,
                        handoff.created_at_ms,
                        imported_at_ms,
                    ],
                )?;
                transaction.execute(
                    "DELETE FROM legacy_handoff_quarantine WHERE source_path = ?1",
                    [path.to_string_lossy().as_ref()],
                )?;
            }
            for quarantine in &handoffs.quarantined {
                transaction.execute(
                    "INSERT INTO legacy_handoff_quarantine
                        (source_path, source_digest, error, quarantined_at_ms)
                     VALUES (?1, ?2, ?3, ?4)
                     ON CONFLICT(source_path) DO UPDATE SET
                        source_digest = excluded.source_digest, error = excluded.error,
                        quarantined_at_ms = excluded.quarantined_at_ms",
                    params![
                        quarantine.source_path.to_string_lossy(),
                        quarantine.source_digest,
                        quarantine.error,
                        imported_at_ms,
                    ],
                )?;
            }
            transaction.execute(
                "INSERT INTO migration_sources
                    (source_kind, source_path, imported_at_ms, record_count)
                 VALUES ('legacy_handoffs', ?1, ?2, ?3)
                 ON CONFLICT(source_kind, source_path) DO UPDATE SET
                    imported_at_ms = excluded.imported_at_ms,
                    record_count = excluded.record_count",
                params![
                    source,
                    imported_at_ms,
                    (handoffs.imported.len() + handoffs.quarantined.len()) as u64,
                ],
            )?;
            transaction.commit()?;
            Ok(imported)
        })
    }

    /// Highest snapshot sequence per run. Event retention must never prune
    /// events at or above this floor, or replay-from-latest-snapshot breaks.
    pub fn latest_stateful_snapshot_seqs(
        &self,
    ) -> anyhow::Result<std::collections::HashMap<String, u64>> {
        self.with_connection(|connection| {
            let mut statement = connection
                .prepare("SELECT run_id, MAX(seq) FROM stateful_snapshots GROUP BY run_id")?;
            let rows = statement.query_map([], |row| {
                Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
            })?;
            rows.map(|row| {
                let (run_id, seq) = row?;
                Ok((run_id, seq.max(0) as u64))
            })
            .collect()
        })
    }

    /// Deletes snapshots older than `cutoff_ms`, always retaining the newest
    /// `keep_last_per_run` snapshots of every run so replay never loses its
    /// most recent restore point. Returns the deleted snapshot IDs so callers
    /// can prune file mirrors.
    pub fn prune_stateful_runtime_snapshots(
        &self,
        cutoff_ms: u64,
        keep_last_per_run: usize,
    ) -> anyhow::Result<Vec<String>> {
        let keep = keep_last_per_run.max(1) as i64;
        self.with_connection(|connection| {
            let transaction =
                connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
            let pruned = {
                let mut statement = transaction.prepare(
                    "SELECT snapshot_id FROM stateful_snapshots
                     WHERE created_at_ms < ?1
                       AND snapshot_id NOT IN (
                          SELECT newer.snapshot_id FROM stateful_snapshots newer
                          WHERE newer.run_id = stateful_snapshots.run_id
                          ORDER BY newer.seq DESC, newer.snapshot_id DESC
                          LIMIT ?2
                       )",
                )?;
                let rows =
                    statement.query_map(params![cutoff_ms, keep], |row| row.get::<_, String>(0))?;
                rows.collect::<Result<Vec<_>, _>>()?
            };
            for snapshot_id in &pruned {
                transaction.execute(
                    "DELETE FROM stateful_snapshots WHERE snapshot_id = ?1",
                    [snapshot_id],
                )?;
            }
            transaction.commit()?;
            Ok(pruned)
        })
    }

    /// Truncates the WAL back into the main database file so long-running
    /// engines reclaim log space between retention sweeps. PostgreSQL manages
    /// its own WAL through server-side checkpoints, so this is a no-op there.
    pub fn checkpoint_wal(&self) -> anyhow::Result<()> {
        self.with_connection(|connection| {
            #[cfg(feature = "storage-sqlite")]
            if let Some(sqlite) = connection.sqlite() {
                sqlite.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
            }
            let _ = &connection;
            Ok(())
        })
    }

    pub fn put_goal(&self, goal: &LongRunningGoal) -> anyhow::Result<()> {
        self.with_connection(|connection| upsert_goal(connection, goal))
    }

    pub fn get_goal(&self, goal_id: &str) -> anyhow::Result<Option<LongRunningGoal>> {
        self.with_connection(|connection| {
            let row = connection
                .query_row(
                    "SELECT org_id, workspace_id, deployment_id, goal_json
                     FROM long_running_goals WHERE goal_id = ?1",
                    [goal_id],
                    |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, String>(1)?,
                            row.get::<_, Option<String>>(2)?,
                            row.get::<_, String>(3)?,
                        ))
                    },
                )
                .optional()?;
            row.map(|(org, workspace, deployment, payload)| {
                protected_records::decode_scoped(
                    &org,
                    &workspace,
                    deployment.as_deref(),
                    "goal",
                    goal_id,
                    &payload,
                )
            })
            .transpose()
        })
    }

    /// Atomically records a governed handoff, its one downstream run, lineage,
    /// and the goal's new active position. Replaying the same idempotency key is
    /// a no-op only when it points to the same handoff and downstream run.
    pub fn commit_handoff_transition(
        &self,
        handoff: &WorkflowHandoff,
        downstream_run: &AutomationV2RunRecord,
        link: &GoalRunLink,
        updated_goal: &LongRunningGoal,
    ) -> anyhow::Result<AtomicHandoffCommit> {
        self.commit_handoff_transition_with_event(handoff, downstream_run, link, updated_goal, None)
    }

    fn commit_handoff_transition_with_event(
        &self,
        handoff: &WorkflowHandoff,
        downstream_run: &AutomationV2RunRecord,
        link: &GoalRunLink,
        updated_goal: &LongRunningGoal,
        transition_event: Option<&crate::stateful_runtime::StatefulRunEventRecord>,
    ) -> anyhow::Result<AtomicHandoffCommit> {
        validate_atomic_transition(handoff, downstream_run, link, updated_goal)?;
        self.with_connection(|connection| {
            let transaction =
                connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
            let existing = transaction
                .query_row(
                    "SELECT handoff_id, status, consumed_by_run_id FROM workflow_handoffs
                     WHERE goal_id = ?1 AND idempotency_key = ?2",
                    params![handoff.goal_id, handoff.idempotency_key],
                    |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, String>(1)?,
                            row.get::<_, Option<String>>(2)?,
                        ))
                    },
                )
                .optional()?;
            let mut update_existing = false;
            if let Some((handoff_id, status, consumed_by_run_id)) = existing {
                if handoff_id == handoff.handoff_id
                    && consumed_by_run_id.as_deref() == Some(downstream_run.run_id.as_str())
                {
                    return Ok(AtomicHandoffCommit::AlreadyCommitted);
                }
                if handoff_id == handoff.handoff_id
                    && consumed_by_run_id.is_none()
                    && matches!(status.as_str(), "pending_approval" | "approved")
                {
                    update_existing = true;
                } else {
                    bail!(
                        "idempotency key {} is already bound to handoff {} and run {:?}",
                        handoff.idempotency_key,
                        handoff_id,
                        consumed_by_run_id
                    );
                }
            }
            ensure_current_goal_allows_handoff(&transaction, handoff)?;

            let mut consumed = handoff.clone();
            consumed.status = WorkflowHandoffStatus::Consumed;
            consumed.consumed_by_run_id = Some(downstream_run.run_id.clone());
            consumed.updated_at_ms = consumed.updated_at_ms.max(downstream_run.created_at_ms);
            let handoff_payload = protected_records::encode(
                &consumed.tenant_context,
                "handoff",
                &consumed.handoff_id,
                &consumed,
            )?;
            if update_existing {
                transaction.execute(
                    "UPDATE workflow_handoffs SET status = 'consumed', consumed_by_run_id = ?2,
                        handoff_json = ?3, updated_at_ms = ?4 WHERE handoff_id = ?1",
                    params![
                        consumed.handoff_id,
                        consumed.consumed_by_run_id,
                        handoff_payload,
                        consumed.updated_at_ms,
                    ],
                )?;
            } else {
                transaction.execute(
                    "INSERT INTO workflow_handoffs
                    (handoff_id, goal_id, idempotency_key, org_id, workspace_id, deployment_id,
                     source_run_id, target_automation_id, status, consumed_by_run_id,
                     handoff_json, created_at_ms, updated_at_ms)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'consumed', ?9, ?10, ?11, ?12)",
                    params![
                        consumed.handoff_id,
                        consumed.goal_id,
                        consumed.idempotency_key,
                        consumed.tenant_context.org_id,
                        consumed.tenant_context.workspace_id,
                        consumed.tenant_context.deployment_id,
                        consumed.source_run_id,
                        consumed.target_automation_id,
                        consumed.consumed_by_run_id,
                        handoff_payload,
                        consumed.created_at_ms,
                        consumed.updated_at_ms,
                    ],
                )?;
            }
            upsert_automation_run(&transaction, downstream_run)?;
            transaction.execute(
                "INSERT INTO goal_run_links
                    (goal_id, run_id, orchestration_node_id, orchestration_version, hop_index,
                     parent_run_id, triggering_handoff_id, link_json, created_at_ms)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
                params![
                    link.goal_id,
                    link.run_id,
                    link.orchestration_node_id,
                    link.orchestration_version,
                    link.hop_index,
                    link.parent_run_id,
                    link.triggering_handoff_id,
                    protected_records::encode(
                        &updated_goal.tenant_context,
                        "link",
                        &link.run_id,
                        link,
                    )?,
                    link.created_at_ms,
                ],
            )?;
            upsert_goal(&transaction, updated_goal)?;
            if let Some(event) = transition_event {
                let mut event = event.clone();
                event.seq = next_event_seq(&transaction, &event.run_id)?;
                let event = runtime_records::event_with_projection_snapshot(
                    &transaction,
                    &event,
                    &handoff.goal_id,
                )?;
                transaction.execute(
                    "INSERT INTO stateful_events
                        (event_id, goal_id, run_id, seq, event_json, created_at_ms,
                         org_id, workspace_id, deployment_id)
                     VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
                    params![
                        event.event_id,
                        handoff.goal_id,
                        event.run_id,
                        event.seq,
                        protected_records::encode(
                            &event.scope.tenant_context,
                            "event",
                            &event.event_id,
                            &event,
                        )?,
                        event.occurred_at_ms,
                        event.scope.tenant_context.org_id,
                        event.scope.tenant_context.workspace_id,
                        event.scope.tenant_context.deployment_id,
                    ],
                )?;
            }
            transaction.commit()?;
            Ok(AtomicHandoffCommit::Committed)
        })
    }

    fn with_connection<T>(
        &self,
        operation: impl FnOnce(&mut Connection) -> anyhow::Result<T>,
    ) -> anyhow::Result<T> {
        let mut connection = self.open_connection()?;
        operation(&mut connection)
    }

    fn open_connection(&self) -> anyhow::Result<Connection> {
        match &self.backend {
            #[cfg(feature = "storage-sqlite")]
            StoreBackendSelection::Sqlite => {
                let connection = rusqlite::Connection::open(&self.paths.database_path)
                    .with_context(|| {
                        format!(
                            "failed to open orchestration store {}",
                            self.paths.database_path.display()
                        )
                    })?;
                connection.busy_timeout(std::time::Duration::from_secs(5))?;
                connection.pragma_update(None, "foreign_keys", "ON")?;
                // `synchronous` is a per-connection pragma: without this, every
                // connection after `initialize_schema` would silently fall back to the
                // SQLite default and weaken the store's crash durability contract.
                connection.pragma_update(None, "synchronous", "FULL")?;
                Ok(Connection::from_sqlite(connection))
            }
            #[cfg(feature = "storage-postgres")]
            StoreBackendSelection::Postgres(target) => {
                Ok(Connection::from_postgres(target.connect()?))
            }
        }
    }
}

/// SQLite schema creation and version migrations. This deliberately stays on
/// the raw rusqlite connection: DDL is the one dialect-specific layer, and
/// the historical v1→v5 migration chain only ever existed on SQLite. The
/// PostgreSQL backend owns its own initialization in `backend::postgres`.
#[cfg(feature = "storage-sqlite")]
fn initialize_schema(connection: &mut rusqlite::Connection) -> anyhow::Result<()> {
    connection.pragma_update(None, "journal_mode", "WAL")?;
    connection.pragma_update(None, "synchronous", "FULL")?;
    connection.execute_batch(
        "BEGIN IMMEDIATE;
         CREATE TABLE IF NOT EXISTS schema_metadata (
            schema_version INTEGER NOT NULL
         );
         INSERT INTO schema_metadata (schema_version)
            SELECT 1 WHERE NOT EXISTS (SELECT 1 FROM schema_metadata);

         CREATE TABLE IF NOT EXISTS orchestration_specs (
            orchestration_id TEXT NOT NULL,
            version INTEGER NOT NULL,
            org_id TEXT NOT NULL,
            workspace_id TEXT NOT NULL,
            deployment_id TEXT,
            deployment_key TEXT NOT NULL DEFAULT '',
            status TEXT NOT NULL,
            definition_json TEXT NOT NULL,
            created_at_ms INTEGER NOT NULL,
            updated_at_ms INTEGER NOT NULL,
            published_at_ms INTEGER,
            PRIMARY KEY (org_id, workspace_id, deployment_key, orchestration_id, version)
         );
         CREATE INDEX IF NOT EXISTS idx_orchestration_scope_status
            ON orchestration_specs (org_id, workspace_id, status);

         CREATE TABLE IF NOT EXISTS automation_runs (
            run_id TEXT PRIMARY KEY,
            automation_id TEXT NOT NULL,
            org_id TEXT NOT NULL,
            workspace_id TEXT NOT NULL,
            deployment_id TEXT,
            status TEXT NOT NULL,
            is_hot INTEGER NOT NULL DEFAULT 1,
            run_json TEXT NOT NULL,
            created_at_ms INTEGER NOT NULL,
            updated_at_ms INTEGER NOT NULL
         );
         CREATE INDEX IF NOT EXISTS idx_automation_runs_scope_status
            ON automation_runs (org_id, workspace_id, status);

         CREATE TABLE IF NOT EXISTS long_running_goals (
            goal_id TEXT PRIMARY KEY,
            orchestration_id TEXT NOT NULL,
            orchestration_version INTEGER NOT NULL,
            org_id TEXT NOT NULL,
            workspace_id TEXT NOT NULL,
            deployment_id TEXT,
            status TEXT NOT NULL,
            active_run_id TEXT,
            goal_json TEXT NOT NULL,
            created_at_ms INTEGER NOT NULL,
            updated_at_ms INTEGER NOT NULL
         );
         CREATE INDEX IF NOT EXISTS idx_goals_scope_status
            ON long_running_goals (org_id, workspace_id, status);

         CREATE TABLE IF NOT EXISTS orchestration_tool_requests (
            org_id TEXT NOT NULL,
            workspace_id TEXT NOT NULL,
            deployment_key TEXT NOT NULL DEFAULT '',
            operation TEXT NOT NULL,
            idempotency_key TEXT NOT NULL,
            request_digest TEXT NOT NULL,
            response_json TEXT,
            created_at_ms INTEGER NOT NULL,
            completed_at_ms INTEGER,
            PRIMARY KEY (
                org_id, workspace_id, deployment_key, operation, idempotency_key
            )
         );

         CREATE TABLE IF NOT EXISTS workflow_handoffs (
            handoff_id TEXT PRIMARY KEY,
            goal_id TEXT NOT NULL,
            idempotency_key TEXT NOT NULL,
            org_id TEXT NOT NULL,
            workspace_id TEXT NOT NULL,
            deployment_id TEXT,
            source_run_id TEXT NOT NULL,
            target_automation_id TEXT NOT NULL,
            status TEXT NOT NULL,
            consumed_by_run_id TEXT UNIQUE,
            handoff_json TEXT NOT NULL,
            created_at_ms INTEGER NOT NULL,
            updated_at_ms INTEGER NOT NULL,
            UNIQUE (goal_id, idempotency_key)
         );
         CREATE INDEX IF NOT EXISTS idx_handoffs_scope_status
            ON workflow_handoffs (org_id, workspace_id, status);

         CREATE TABLE IF NOT EXISTS legacy_handoffs (
            handoff_id TEXT PRIMARY KEY,
            source_path TEXT NOT NULL,
            status TEXT NOT NULL,
            consumed_by_run_id TEXT,
            handoff_json TEXT NOT NULL,
            created_at_ms INTEGER NOT NULL,
            imported_at_ms INTEGER NOT NULL
         );
         CREATE TABLE IF NOT EXISTS legacy_handoff_quarantine (
            source_path TEXT PRIMARY KEY,
            source_digest TEXT,
            error TEXT NOT NULL,
            quarantined_at_ms INTEGER NOT NULL
         );

         CREATE TABLE IF NOT EXISTS goal_run_links (
            goal_id TEXT NOT NULL,
            run_id TEXT NOT NULL UNIQUE,
            orchestration_node_id TEXT NOT NULL,
            orchestration_version INTEGER NOT NULL,
            hop_index INTEGER NOT NULL,
            parent_run_id TEXT,
            triggering_handoff_id TEXT UNIQUE,
            link_json TEXT NOT NULL,
            created_at_ms INTEGER NOT NULL,
            PRIMARY KEY (goal_id, hop_index)
         );

         CREATE TABLE IF NOT EXISTS automation_waits (
            wait_id TEXT PRIMARY KEY, goal_id TEXT, run_id TEXT, status TEXT NOT NULL,
            wait_json TEXT NOT NULL, updated_at_ms INTEGER NOT NULL
         );
         CREATE TABLE IF NOT EXISTS wait_resolutions (
            wait_id TEXT NOT NULL, idempotency_key TEXT NOT NULL, resolution_json TEXT NOT NULL,
            resolved_at_ms INTEGER NOT NULL, PRIMARY KEY (wait_id, idempotency_key)
         );
         CREATE TABLE IF NOT EXISTS stateful_events (
            event_id TEXT PRIMARY KEY, goal_id TEXT, run_id TEXT, seq INTEGER NOT NULL,
            event_json TEXT NOT NULL, created_at_ms INTEGER NOT NULL
         );
         CREATE INDEX IF NOT EXISTS idx_stateful_events_goal
            ON stateful_events (goal_id);
         CREATE TABLE IF NOT EXISTS goal_projection_blobs (
            digest TEXT PRIMARY KEY,
            payload_json TEXT NOT NULL,
            created_at_ms INTEGER NOT NULL
         );
         CREATE TABLE IF NOT EXISTS stateful_snapshots (
            snapshot_id TEXT PRIMARY KEY, goal_id TEXT, run_id TEXT, seq INTEGER NOT NULL,
            snapshot_json TEXT NOT NULL, created_at_ms INTEGER NOT NULL
         );
         CREATE TABLE IF NOT EXISTS outbox_effects (
            effect_id TEXT PRIMARY KEY, goal_id TEXT, run_id TEXT, status TEXT NOT NULL,
            effect_json TEXT NOT NULL, updated_at_ms INTEGER NOT NULL
         );
         CREATE TABLE IF NOT EXISTS dead_letters (
            dead_letter_id TEXT PRIMARY KEY, goal_id TEXT, run_id TEXT, status TEXT NOT NULL,
            record_json TEXT NOT NULL, updated_at_ms INTEGER NOT NULL
         );
         CREATE TABLE IF NOT EXISTS compensations (
            compensation_id TEXT PRIMARY KEY, goal_id TEXT, run_id TEXT, status TEXT NOT NULL,
            record_json TEXT NOT NULL, updated_at_ms INTEGER NOT NULL
         );
         CREATE TABLE IF NOT EXISTS migration_sources (
            source_kind TEXT NOT NULL, source_path TEXT NOT NULL, imported_at_ms INTEGER NOT NULL,
            record_count INTEGER NOT NULL, PRIMARY KEY (source_kind, source_path)
         );
         COMMIT;",
    )?;
    if !table_has_column(connection, "automation_runs", "is_hot")? {
        connection.execute(
            "ALTER TABLE automation_runs ADD COLUMN is_hot INTEGER NOT NULL DEFAULT 1",
            [],
        )?;
    }
    if !table_has_column(connection, "orchestration_specs", "deployment_key")? {
        connection.execute_batch(
            "BEGIN IMMEDIATE;
             CREATE TABLE orchestration_specs_v3 (
                orchestration_id TEXT NOT NULL,
                version INTEGER NOT NULL,
                org_id TEXT NOT NULL,
                workspace_id TEXT NOT NULL,
                deployment_id TEXT,
                deployment_key TEXT NOT NULL DEFAULT '',
                status TEXT NOT NULL,
                definition_json TEXT NOT NULL,
                created_at_ms INTEGER NOT NULL,
                updated_at_ms INTEGER NOT NULL,
                published_at_ms INTEGER,
                PRIMARY KEY (org_id, workspace_id, deployment_key, orchestration_id, version)
             );
             INSERT INTO orchestration_specs_v3 (
                orchestration_id, version, org_id, workspace_id, deployment_id, deployment_key,
                status, definition_json, created_at_ms, updated_at_ms, published_at_ms
             ) SELECT orchestration_id, version, org_id, workspace_id, deployment_id,
                      COALESCE(deployment_id, ''), status, definition_json, created_at_ms,
                      updated_at_ms, published_at_ms
               FROM orchestration_specs;
             DROP TABLE orchestration_specs;
             ALTER TABLE orchestration_specs_v3 RENAME TO orchestration_specs;
             CREATE INDEX idx_orchestration_scope_status
                ON orchestration_specs (org_id, workspace_id, status);
             COMMIT;",
        )?;
    }
    let mut version: i64 = connection.query_row(
        "SELECT schema_version FROM schema_metadata LIMIT 1",
        [],
        |row| row.get(0),
    )?;
    if version == 1 {
        migrate_schema_v1_to_v2(connection)?;
        version = 2;
    }
    if version == 2 {
        migrate_schema_v2_to_v3(connection)?;
        version = 3;
    }
    if version == 3 {
        migrate_schema_v3_to_v4(connection)?;
        version = 4;
    }
    if version == 4 {
        migrate_schema_v4_to_v5(connection)?;
        version = 5;
    }
    if version != SCHEMA_VERSION {
        bail!(
            "unsupported orchestration store schema version {version}; expected {SCHEMA_VERSION}"
        );
    }
    // SQLite secondary indexes carry the hidden rowid as their tie-breaker,
    // so indexing goal_id preserves the cursor-ordered replay access path.
    connection.execute_batch(
        "CREATE INDEX IF NOT EXISTS idx_stateful_events_goal
           ON stateful_events (goal_id);",
    )?;
    Ok(())
}

#[cfg(feature = "storage-sqlite")]
fn migrate_schema_v1_to_v2(connection: &mut rusqlite::Connection) -> anyhow::Result<()> {
    connection.execute_batch(
        "BEGIN IMMEDIATE;
         CREATE TABLE IF NOT EXISTS stateful_migrations (
            migration_id TEXT PRIMARY KEY,
            status TEXT NOT NULL,
            source_fingerprint TEXT NOT NULL,
            record_count INTEGER NOT NULL,
            started_at_ms INTEGER NOT NULL,
            completed_at_ms INTEGER
         );
         CREATE TABLE IF NOT EXISTS tool_effects (
            effect_id TEXT PRIMARY KEY,
            goal_id TEXT,
            run_id TEXT,
            org_id TEXT NOT NULL,
            workspace_id TEXT NOT NULL,
            deployment_id TEXT,
            status TEXT NOT NULL,
            effect_json TEXT NOT NULL,
            updated_at_ms INTEGER NOT NULL
         );
         CREATE INDEX IF NOT EXISTS idx_tool_effects_scope_status
            ON tool_effects (org_id, workspace_id, status);
         CREATE UNIQUE INDEX IF NOT EXISTS idx_stateful_events_run_seq
            ON stateful_events (run_id, seq);
         CREATE UNIQUE INDEX IF NOT EXISTS idx_stateful_snapshots_run_seq
            ON stateful_snapshots (run_id, seq);
         COMMIT;",
    )?;
    add_scope_columns(connection, "automation_waits")?;
    add_scope_columns(connection, "stateful_events")?;
    add_scope_columns(connection, "stateful_snapshots")?;
    add_scope_columns(connection, "outbox_effects")?;
    add_scope_columns(connection, "dead_letters")?;
    add_scope_columns(connection, "compensations")?;
    connection.execute("UPDATE schema_metadata SET schema_version = 2", [])?;
    Ok(())
}

#[cfg(feature = "storage-sqlite")]
fn migrate_schema_v2_to_v3(connection: &mut rusqlite::Connection) -> anyhow::Result<()> {
    connection.execute_batch(
        "BEGIN IMMEDIATE;
         CREATE TABLE IF NOT EXISTS legacy_handoff_quarantine (
            source_path TEXT PRIMARY KEY,
            source_digest TEXT,
            error TEXT NOT NULL,
            quarantined_at_ms INTEGER NOT NULL
         );
         UPDATE schema_metadata SET schema_version = 3;
         COMMIT;",
    )?;
    Ok(())
}

#[cfg(feature = "storage-sqlite")]
fn migrate_schema_v3_to_v4(connection: &mut rusqlite::Connection) -> anyhow::Result<()> {
    // Some early development v2 stores reached v3 without the scope-column
    // backfill. Normalize them before rebuilding the scoped wait identity.
    add_scope_columns(connection, "automation_waits")?;
    connection.execute_batch(
        "BEGIN IMMEDIATE;
         CREATE TABLE automation_waits_v4 (
            wait_id TEXT NOT NULL,
            goal_id TEXT,
            run_id TEXT NOT NULL,
            org_id TEXT NOT NULL,
            workspace_id TEXT NOT NULL,
            deployment_id TEXT NOT NULL DEFAULT '',
            status TEXT NOT NULL,
            wait_json TEXT NOT NULL,
            updated_at_ms INTEGER NOT NULL,
            PRIMARY KEY (wait_id, run_id, org_id, workspace_id, deployment_id)
         );
         INSERT INTO automation_waits_v4
            (wait_id, goal_id, run_id, org_id, workspace_id, deployment_id,
             status, wait_json, updated_at_ms)
         SELECT wait_id, goal_id, COALESCE(run_id, ''), org_id, workspace_id,
                COALESCE(deployment_id, ''), status, wait_json, updated_at_ms
         FROM automation_waits;
         DROP TABLE automation_waits;
         ALTER TABLE automation_waits_v4 RENAME TO automation_waits;
         CREATE INDEX idx_automation_waits_scope_status
            ON automation_waits (org_id, workspace_id, status);
         UPDATE schema_metadata SET schema_version = 4;
         COMMIT;",
    )?;
    Ok(())
}

#[cfg(feature = "storage-sqlite")]
fn migrate_schema_v4_to_v5(connection: &mut rusqlite::Connection) -> anyhow::Result<()> {
    // Durable migration-attempt journal: rows are committed before the import
    // transaction starts, so an interrupted import leaves evidence that the
    // atomic once-only marker cannot (a rolled-back marker looks identical to
    // "never attempted").
    connection.execute_batch(
        "BEGIN IMMEDIATE;
         CREATE TABLE IF NOT EXISTS stateful_migration_attempts (
            attempt_id INTEGER PRIMARY KEY AUTOINCREMENT,
            migration_id TEXT NOT NULL,
            source_fingerprint TEXT NOT NULL,
            started_at_ms INTEGER NOT NULL,
            outcome TEXT,
            completed_at_ms INTEGER
         );
         UPDATE schema_metadata SET schema_version = 5;
         COMMIT;",
    )?;
    Ok(())
}

#[cfg(feature = "storage-sqlite")]
fn add_scope_columns(connection: &rusqlite::Connection, table: &str) -> anyhow::Result<()> {
    for (column, definition) in [
        ("org_id", "TEXT NOT NULL DEFAULT ''"),
        ("workspace_id", "TEXT NOT NULL DEFAULT ''"),
        ("deployment_id", "TEXT"),
    ] {
        if !table_has_column(connection, table, column)? {
            connection.execute(
                &format!("ALTER TABLE {table} ADD COLUMN {column} {definition}"),
                [],
            )?;
        }
    }
    connection.execute(
        &format!("CREATE INDEX IF NOT EXISTS idx_{table}_scope ON {table} (org_id, workspace_id)"),
        [],
    )?;
    Ok(())
}

fn upsert_automation_run(
    connection: &impl Executor,
    run: &AutomationV2RunRecord,
) -> anyhow::Result<()> {
    let status = serde_json::to_value(&run.status)?;
    connection.execute(
        "INSERT INTO automation_runs
            (run_id, automation_id, org_id, workspace_id, deployment_id, status,
             is_hot, run_json, created_at_ms, updated_at_ms)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, 1, ?7, ?8, ?9)
         ON CONFLICT(run_id) DO UPDATE SET
            status = excluded.status,
            is_hot = 1,
            run_json = excluded.run_json,
            updated_at_ms = excluded.updated_at_ms
         WHERE excluded.updated_at_ms >= automation_runs.updated_at_ms",
        params![
            run.run_id,
            run.automation_id,
            run.tenant_context.org_id,
            run.tenant_context.workspace_id,
            run.tenant_context.deployment_id,
            status.as_str().unwrap_or("unknown"),
            protected_records::encode(&run.tenant_context, "run", &run.run_id, run)?,
            run.created_at_ms,
            run.updated_at_ms,
        ],
    )?;
    Ok(())
}

#[cfg(feature = "storage-sqlite")]
fn table_has_column(
    connection: &rusqlite::Connection,
    table_name: &str,
    column_name: &str,
) -> anyhow::Result<bool> {
    let mut statement = connection.prepare(&format!("PRAGMA table_info({table_name})"))?;
    let columns = statement.query_map([], |row| row.get::<_, String>(1))?;
    for column in columns {
        if column? == column_name {
            return Ok(true);
        }
    }
    Ok(false)
}

fn upsert_goal(connection: &impl Executor, goal: &LongRunningGoal) -> anyhow::Result<()> {
    let status = serde_json::to_value(&goal.status)?;
    connection.execute(
        "INSERT INTO long_running_goals
            (goal_id, orchestration_id, orchestration_version, org_id, workspace_id,
             deployment_id, status, active_run_id, goal_json, created_at_ms, updated_at_ms)
         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11)
         ON CONFLICT(goal_id) DO UPDATE SET
            status = excluded.status,
            active_run_id = excluded.active_run_id,
            goal_json = excluded.goal_json,
            updated_at_ms = excluded.updated_at_ms
         WHERE excluded.updated_at_ms >= long_running_goals.updated_at_ms",
        params![
            goal.goal_id,
            goal.orchestration_id,
            goal.orchestration_version,
            goal.tenant_context.org_id,
            goal.tenant_context.workspace_id,
            goal.tenant_context.deployment_id,
            status.as_str().unwrap_or("unknown"),
            goal.active_run_id,
            protected_records::encode(&goal.tenant_context, "goal", &goal.goal_id, goal)?,
            goal.created_at_ms,
            goal.updated_at_ms,
        ],
    )?;
    Ok(())
}

fn next_event_seq(connection: &impl Executor, run_id: &str) -> anyhow::Result<u64> {
    let last: Option<u64> = connection.query_row(
        "SELECT MAX(seq) FROM stateful_events WHERE run_id = ?1",
        [run_id],
        |row| row.get(0),
    )?;
    Ok(last.unwrap_or(0).saturating_add(1))
}

fn validate_atomic_transition(
    handoff: &WorkflowHandoff,
    downstream_run: &AutomationV2RunRecord,
    link: &GoalRunLink,
    goal: &LongRunningGoal,
) -> anyhow::Result<()> {
    if handoff.goal_id != goal.goal_id || link.goal_id != goal.goal_id {
        bail!("handoff, lineage, and goal must use the same goal_id");
    }
    if handoff.tenant_context != goal.tenant_context
        || downstream_run.tenant_context != goal.tenant_context
    {
        bail!("handoff and downstream run must remain in the goal tenant scope");
    }
    if downstream_run.run_id != link.run_id
        || link.triggering_handoff_id.as_deref() != Some(handoff.handoff_id.as_str())
    {
        bail!("lineage must bind the downstream run to the triggering handoff");
    }
    if downstream_run.automation_id != handoff.target_automation_id {
        bail!("downstream run automation does not match the handoff target");
    }
    if goal.active_run_id.as_deref() != Some(downstream_run.run_id.as_str())
        || goal.hop_count != link.hop_index
    {
        bail!("updated goal must point at the linked downstream run and hop");
    }
    if handoff.orchestration_version != link.orchestration_version
        || goal.orchestration_version != link.orchestration_version
    {
        bail!("handoff, lineage, and goal must use the same orchestration version");
    }
    Ok(())
}

fn ensure_current_goal_allows_handoff(
    transaction: &impl Executor,
    handoff: &WorkflowHandoff,
) -> anyhow::Result<()> {
    let row = transaction
        .query_row(
            "SELECT org_id, workspace_id, deployment_id, goal_json
             FROM long_running_goals WHERE goal_id = ?1",
            [&handoff.goal_id],
            |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, Option<String>>(2)?,
                    row.get::<_, String>(3)?,
                ))
            },
        )
        .optional()?;
    let Some((org, workspace, deployment, payload)) = row else {
        return Ok(());
    };
    let goal: LongRunningGoal = protected_records::decode_scoped(
        &org,
        &workspace,
        deployment.as_deref(),
        "goal",
        &handoff.goal_id,
        &payload,
    )?;
    if goal.tenant_context != handoff.tenant_context
        || goal.orchestration_id != handoff.orchestration_id
        || goal.orchestration_version != handoff.orchestration_version
    {
        bail!("handoff no longer matches the persisted goal")
    }
    if goal.status.is_terminal() || goal.status == LongRunningGoalStatus::Paused {
        bail!("terminal or paused goals reject handoff consumption")
    }
    if goal.active_run_id.as_deref() != Some(handoff.source_run_id.as_str())
        || goal.current_node_id.as_deref() != Some(handoff.source_node_id.as_str())
    {
        bail!("handoff source is no longer active for the persisted goal")
    }
    Ok(())
}

fn collect_json_files(root: &Path, output: &mut Vec<PathBuf>) -> anyhow::Result<()> {
    if !root.exists() {
        return Ok(());
    }
    for entry in std::fs::read_dir(root)
        .with_context(|| format!("failed to read handoff directory {}", root.display()))?
    {
        let path = entry?.path();
        if path.is_dir() {
            collect_json_files(&path, output)?;
        } else if path.extension().and_then(|value| value.to_str()) == Some("json") {
            output.push(path);
        }
    }
    Ok(())
}

#[cfg(test)]
#[path = "orchestration_store/backend_conformance_tests.rs"]
mod backend_conformance_tests;

#[cfg(test)]
#[path = "orchestration_store/tests.rs"]
mod tests;

#[cfg(test)]
#[path = "orchestration_store/encryption_tests.rs"]
mod encryption_tests;

#[cfg(test)]
#[path = "orchestration_store/hardening_tests.rs"]
mod hardening_tests;