mod projection_cache;
mod run_catalog_index;
mod run_store;

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, OnceLock};
use std::time::Duration;

use chrono::{DateTime, Utc};
use fabro_types::{Run, RunId, SessionId};
use object_store::ObjectStore;
pub use projection_cache::CachedRunProjection;
use projection_cache::RunProjectionCache;
pub use run_catalog_index::RunCatalogIndex;
pub use run_store::RunDatabase;
use run_store::RunDatabaseInner;
use slatedb::config::{CompressionCodec, Settings};
use tokio::sync::{Mutex, OnceCell};
use tracing::warn;

use crate::{BlobStore, Error, ListRunsQuery, Result, RunProjection, RunSummaryStore, keys};

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct UnreadableRun {
    pub run_id:     RunId,
    pub created_at: DateTime<Utc>,
    pub error:      String,
}

#[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]
struct SessionRunIndexEntry {
    run_id: RunId,
}

#[derive(Clone)]
pub struct Database {
    object_store: Arc<dyn ObjectStore>,
    base_prefix: String,
    flush_interval: Duration,
    cache_path: Option<PathBuf>,
    db: Arc<OnceCell<slatedb::Db>>,
    active_runs: Arc<Mutex<HashMap<RunId, Arc<RunDatabaseInner>>>>,
    blobs: Arc<BlobStore>,
    catalog_index: Arc<OnceCell<Arc<RunCatalogIndex>>>,
    projection_cache: Arc<RunProjectionCache>,
    projection_cache_warmed: Arc<OnceCell<()>>,
    run_summary_store: Arc<OnceLock<Arc<RunSummaryStore>>>,
}

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

impl Database {
    pub fn new(
        object_store: Arc<dyn ObjectStore>,
        base_prefix: impl Into<String>,
        flush_interval: Duration,
        cache_path: Option<PathBuf>,
        blobs: Arc<BlobStore>,
    ) -> Self {
        Self {
            object_store,
            base_prefix: normalize_base_prefix(base_prefix.into()),
            flush_interval,
            cache_path,
            db: Arc::new(OnceCell::new()),
            active_runs: Arc::new(Mutex::new(HashMap::new())),
            blobs,
            catalog_index: Arc::new(OnceCell::new()),
            projection_cache: Arc::new(RunProjectionCache::default()),
            projection_cache_warmed: Arc::new(OnceCell::new()),
            run_summary_store: Arc::new(OnceLock::new()),
        }
    }

    pub fn attach_run_summary_store(&self, store: Arc<RunSummaryStore>) -> Arc<RunSummaryStore> {
        Arc::clone(self.run_summary_store.get_or_init(|| store))
    }

    fn run_summary_store(&self) -> Option<Arc<RunSummaryStore>> {
        self.run_summary_store.get().cloned()
    }

    fn shared_db_prefix(&self) -> String {
        self.base_prefix.clone()
    }

    pub(crate) async fn open_db(&self) -> Result<slatedb::Db> {
        let db = self
            .db
            .get_or_try_init(|| async {
                let mut settings = Settings {
                    flush_interval: Some(self.flush_interval),
                    compression_codec: Some(CompressionCodec::Zstd),
                    ..Settings::default()
                };
                if let Some(ref cache_path) = self.cache_path {
                    settings.object_store_cache_options.root_folder = Some(cache_path.clone());
                }
                slatedb::Db::builder(self.shared_db_prefix(), self.object_store.clone())
                    .with_settings(settings)
                    .build()
                    .await
            })
            .await?;
        Ok(db.clone())
    }

    async fn get_active_run(&self, run_id: &RunId) -> Option<RunDatabase> {
        let active_runs = self.active_runs.lock().await;
        active_run_from(&active_runs, run_id)
    }

    fn cache_active_run(
        active_runs: &mut HashMap<RunId, Arc<RunDatabaseInner>>,
        run_store: &RunDatabase,
    ) {
        active_runs.insert(run_store.run_id(), run_store.inner_arc());
    }

    async fn remove_active_run(&self, run_id: &RunId) -> Option<RunDatabase> {
        self.active_runs
            .lock()
            .await
            .remove(run_id)
            .map(RunDatabase::from_inner)
    }

    /// Builds a run handle wired to the Database-owned shared stores.
    async fn open_run_database(&self, run_id: &RunId, read_only: bool) -> Result<RunDatabase> {
        RunDatabase::build(
            *run_id,
            self.open_db().await?,
            read_only,
            self.blobs(),
            Arc::clone(&self.projection_cache),
            Arc::clone(&self.run_summary_store),
        )
        .await
    }

    pub async fn create_run(&self, run_id: &RunId) -> Result<RunDatabase> {
        self.warm_projection_cache().await?;
        self.catalog_index().await?.add(run_id).await?;
        let run_store = self.open_run_database(run_id, false).await?;
        let mut active_runs = self.active_runs.lock().await;
        Self::cache_active_run(&mut active_runs, &run_store);
        Ok(run_store)
    }

    pub async fn open_run(&self, run_id: &RunId) -> Result<RunDatabase> {
        self.warm_projection_cache().await?;
        let db = self.open_db().await?;
        // Keep the active-writer miss and insert atomic. Otherwise concurrent
        // callers can create independent writers with the same recovered seq.
        let mut active_runs = self.active_runs.lock().await;

        if let Some(active) = active_run_from(&active_runs, run_id) {
            if !active.matches_run(run_id) {
                return Err(Error::Other(format!(
                    "active run cache mismatch for run_id {run_id:?}"
                )));
            }
            return Ok(active);
        }
        if !RunDatabase::has_any_events(&db, run_id).await? {
            return Err(Error::RunNotFound(run_id.to_string()));
        }
        let run_store = self.open_run_database(run_id, false).await?;
        Self::cache_active_run(&mut active_runs, &run_store);
        Ok(run_store)
    }

    pub async fn open_run_reader(&self, run_id: &RunId) -> Result<RunDatabase> {
        let db = self.open_db().await?;
        if let Some(active) = self.get_active_run(run_id).await {
            if !active.matches_run(run_id) {
                return Err(Error::Other(format!(
                    "active run cache mismatch for run_id {run_id:?}"
                )));
            }
            return Ok(active.read_only_clone());
        }
        if !RunDatabase::has_any_events(&db, run_id).await? {
            return Err(Error::RunNotFound(run_id.to_string()));
        }
        self.open_run_database(run_id, true).await
    }

    pub async fn list_runs(&self, query: &ListRunsQuery, now: DateTime<Utc>) -> Result<Vec<Run>> {
        Ok(self
            .list_cached_runs(query, now)
            .await?
            .into_iter()
            .map(|entry| entry.summary)
            .collect())
    }

    pub async fn list_runs_with_projection(
        &self,
        query: &ListRunsQuery,
        now: DateTime<Utc>,
    ) -> Result<Vec<(Run, RunProjection)>> {
        Ok(self
            .list_cached_runs(query, now)
            .await?
            .into_iter()
            .map(|entry| (entry.summary, (*entry.projection).clone()))
            .collect())
    }

    pub async fn warm_projection_cache(&self) -> Result<()> {
        self.projection_cache_warmed
            .get_or_try_init(|| async {
                let db = self.open_db().await?;
                let run_ids = self
                    .catalog_index()
                    .await?
                    .list(&ListRunsQuery::default())
                    .await?;
                let mut entries = Vec::new();
                for run_id in run_ids {
                    match RunDatabase::build_cached_projection(&db, &run_id).await {
                        Ok(Some(entry)) => entries.push(entry),
                        Ok(None) => {}
                        Err(err) => {
                            warn!(
                                run_id = %run_id,
                                error = %err,
                                "Skipping run during projection cache warmup"
                            );
                        }
                    }
                }
                if let Some(store) = self.run_summary_store() {
                    store.reconcile(&entries).await?;
                }
                self.projection_cache.replace_all(entries).await;
                Ok::<_, Error>(())
            })
            .await?;
        Ok(())
    }

    pub async fn list_cached_runs(
        &self,
        query: &ListRunsQuery,
        now: DateTime<Utc>,
    ) -> Result<Vec<CachedRunProjection>> {
        self.warm_projection_cache().await?;
        Ok(self.projection_cache.list(query, now).await)
    }

    pub async fn list_unreadable_runs(&self) -> Result<Vec<UnreadableRun>> {
        let db = self.open_db().await?;
        let run_ids = self
            .catalog_index()
            .await?
            .list(&ListRunsQuery::default())
            .await?;
        let mut unreadable = Vec::new();
        for run_id in run_ids {
            match RunDatabase::build_cached_projection(&db, &run_id).await {
                Ok(Some(_)) => {}
                Ok(None) => unreadable.push(UnreadableRun {
                    run_id,
                    created_at: run_id.created_at(),
                    error: "run has no events".to_string(),
                }),
                Err(err) => unreadable.push(UnreadableRun {
                    run_id,
                    created_at: run_id.created_at(),
                    error: err.to_string(),
                }),
            }
        }
        unreadable.sort_by(|left, right| {
            right
                .created_at
                .cmp(&left.created_at)
                .then_with(|| right.run_id.cmp(&left.run_id))
        });
        Ok(unreadable)
    }

    #[cfg(any(test, feature = "test-support"))]
    pub(crate) async fn put_unvalidated_run_event(
        &self,
        run_id: &RunId,
        seq: u32,
        payload: &serde_json::Value,
    ) -> Result<()> {
        let db = self.open_db().await?;
        db.put(
            keys::run_event_key(run_id, seq, 0),
            serde_json::to_vec(payload)?,
        )
        .await?;
        Ok(())
    }

    pub async fn get_cached_run(&self, run_id: &RunId) -> Result<Option<CachedRunProjection>> {
        self.warm_projection_cache().await?;
        Ok(self.projection_cache.get(run_id).await)
    }

    pub async fn get_cached_projection(
        &self,
        run_id: &RunId,
    ) -> Result<Option<Arc<RunProjection>>> {
        self.warm_projection_cache().await?;
        Ok(self
            .projection_cache
            .projection_snapshot(run_id)
            .await
            .map(|(projection, _)| projection))
    }

    pub async fn get_cached_summary(
        &self,
        run_id: &RunId,
        now: DateTime<Utc>,
    ) -> Result<Option<Run>> {
        self.warm_projection_cache().await?;
        Ok(self.projection_cache.get_summary(run_id, now).await)
    }

    /// Run ids whose latest explicit pull request creation is still pending,
    /// oldest request first.
    pub async fn pending_pull_request_creation_run_ids(&self) -> Result<Vec<RunId>> {
        self.warm_projection_cache().await?;
        Ok(self.projection_cache.pending_pull_request_creations().await)
    }

    pub async fn put_session_run_index(
        &self,
        session_id: &SessionId,
        run_id: &RunId,
    ) -> Result<()> {
        let db = self.open_db().await?;
        db.put(
            keys::session_by_id_key(session_id),
            serde_json::to_vec(&SessionRunIndexEntry { run_id: *run_id })?,
        )
        .await?;
        Ok(())
    }

    pub async fn get_session_run_id(&self, session_id: &SessionId) -> Result<Option<RunId>> {
        let db = self.open_db().await?;
        if let Some(bytes) = db.get(keys::session_by_id_key(session_id)).await? {
            let entry: SessionRunIndexEntry = serde_json::from_slice(&bytes)?;
            return Ok(Some(entry.run_id));
        }
        Ok(None)
    }

    pub(crate) async fn remove_cached_run(&self, run_id: &RunId) {
        self.projection_cache.remove(run_id).await;
    }

    pub async fn delete_run(&self, run_id: &RunId) -> Result<()> {
        let active = self.remove_active_run(run_id).await;
        if let Some(active) = &active {
            active.close().await?;
        }

        let db = self.open_db().await?;
        let mut keys_to_delete = Vec::new();
        for prefix in [keys::run_data_prefix(run_id)] {
            let mut iter = db.scan_prefix(&prefix).await?;
            while let Some(entry) = iter.next().await? {
                keys_to_delete.push(String::from_utf8(entry.key.to_vec()).map_err(|err| {
                    Error::Other(format!("stored key is not valid UTF-8: {err}"))
                })?);
            }
        }
        for key in keys_to_delete {
            db.delete(key).await?;
        }
        self.delete_session_indexes_for_run(run_id).await?;
        self.catalog_index().await?.remove(run_id).await?;
        self.remove_cached_run(run_id).await;
        if let Some(store) = self.run_summary_store() {
            store.delete(run_id).await?;
        }
        Ok(())
    }

    async fn delete_session_indexes_for_run(&self, run_id: &RunId) -> Result<()> {
        let db = self.open_db().await?;
        let mut keys_to_delete = Vec::new();
        let mut iter = db.scan_prefix(keys::sessions_by_id_prefix()).await?;
        while let Some(entry) = iter.next().await? {
            let index: SessionRunIndexEntry = serde_json::from_slice(&entry.value)?;
            if index.run_id == *run_id {
                keys_to_delete.push(String::from_utf8(entry.key.to_vec()).map_err(|err| {
                    Error::Other(format!("stored key is not valid UTF-8: {err}"))
                })?);
            }
        }
        for key in keys_to_delete {
            db.delete(key).await?;
        }
        Ok(())
    }

    pub async fn catalog_index(&self) -> Result<Arc<RunCatalogIndex>> {
        let store = self
            .catalog_index
            .get_or_try_init(|| async {
                let db = Arc::new(self.open_db().await?);
                Ok::<_, Error>(Arc::new(RunCatalogIndex::new(db)))
            })
            .await?;
        Ok(Arc::clone(store))
    }

    #[must_use]
    pub fn blobs(&self) -> Arc<BlobStore> {
        Arc::clone(&self.blobs)
    }

    /// Delete every record under the retired `auth/refresh` prefix.
    ///
    /// Refresh tokens moved to SQLite without an import, so these records are
    /// unreadable -- and the reaper that used to collect them is gone, so
    /// nothing else would ever remove them. Returns the number of records
    /// deleted; a later boot finds the prefix empty and does nothing.
    pub async fn retire_refresh_token_keyspace(&self) -> Result<u64> {
        let db = self.open_db().await?;
        let mut iter = db
            .scan_prefix(keys::SlateKey::new("auth").with("refresh").into_prefix())
            .await?;
        let mut batch = slatedb::WriteBatch::new();
        let mut deletes = 0_u64;
        while let Some(entry) = iter.next().await? {
            batch.delete(entry.key);
            deletes += 1;
        }
        if deletes > 0 {
            db.write(batch).await?;
        }
        Ok(deletes)
    }

    #[must_use]
    pub fn runs(&self) -> Runs {
        Runs { db: self.clone() }
    }
}

#[derive(Clone, Debug)]
pub struct Runs {
    db: Database,
}

impl Runs {
    pub async fn get(&self, run_id: &RunId) -> Result<RunDatabase> {
        self.db.open_run(run_id).await
    }

    pub async fn find(&self, run_id: &RunId) -> Result<Option<Run>> {
        self.db.get_cached_summary(run_id, Utc::now()).await
    }

    pub async fn list(&self, query: &ListRunsQuery) -> Result<Vec<Run>> {
        self.db.list_runs(query, Utc::now()).await
    }
}

pub(crate) fn normalize_base_prefix(prefix: String) -> String {
    if prefix.is_empty() {
        return String::new();
    }
    if prefix.ends_with('/') {
        prefix
    } else {
        format!("{prefix}/")
    }
}

fn active_run_from(
    active_runs: &HashMap<RunId, Arc<RunDatabaseInner>>,
    run_id: &RunId,
) -> Option<RunDatabase> {
    active_runs
        .get(run_id)
        .cloned()
        .map(RunDatabase::from_inner)
}

#[cfg(test)]
mod tests {
    use chrono::{DateTime, Utc};
    use fabro_types::{
        AttrValue, FailureReason, Graph, RunControlAction, RunSpec, RunStatus, StageId,
        SuccessReason, WorkflowSettings, test_support,
    };
    use futures::TryStreamExt;
    use object_store::memory::InMemory;
    use object_store::path::Path;

    use super::*;
    use crate::{EventPayload, keys, test_support as store_test_support};

    fn dt(value: &str) -> DateTime<Utc> {
        value.parse().unwrap()
    }

    fn test_run_id(label: &str) -> RunId {
        let (timestamp_ms, random) = match label {
            "run-1" => (
                dt("2026-03-27T12:00:00Z")
                    .timestamp_millis()
                    .cast_unsigned(),
                1,
            ),
            "run-2" => (
                dt("2026-03-27T12:00:10Z")
                    .timestamp_millis()
                    .cast_unsigned(),
                2,
            ),
            "run-3" => (
                dt("2026-03-27T12:00:20Z")
                    .timestamp_millis()
                    .cast_unsigned(),
                3,
            ),
            "run-4" => (
                dt("2026-03-27T12:00:30Z")
                    .timestamp_millis()
                    .cast_unsigned(),
                4,
            ),
            _ => panic!("unknown test run id: {label}"),
        };
        RunId::from(ulid::Ulid::from_parts(timestamp_ms, random))
    }

    fn make_store() -> (Arc<dyn ObjectStore>, Database) {
        let object_store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
        let store = store_test_support::test_database(
            object_store.clone(),
            "runs/",
            Duration::from_millis(1),
            None,
        );
        (object_store, store)
    }

    #[tokio::test]
    async fn retire_refresh_token_keyspace_clears_the_prefix_and_is_idempotent() {
        let (_object_store, store) = make_store();
        let db = store.open_db().await.unwrap();

        let refresh_keys = ["aaa", "bbb"].map(|id| {
            keys::SlateKey::new("auth")
                .with("refresh")
                .with(id)
                .as_ref()
                .to_vec()
        });
        // "auth/code" sorts adjacent to "auth/refresh", so it is the
        // neighbour a too-wide prefix delete would take with it.
        let auth_code_key = keys::SlateKey::new("auth")
            .with("code")
            .with("keep")
            .as_ref()
            .to_vec();

        let mut batch = slatedb::WriteBatch::new();
        for key in &refresh_keys {
            batch.put(key.as_slice(), b"{}".as_slice());
        }
        batch.put(auth_code_key.as_slice(), b"{}".as_slice());
        db.write(batch).await.unwrap();

        assert_eq!(store.retire_refresh_token_keyspace().await.unwrap(), 2);
        assert_eq!(store.retire_refresh_token_keyspace().await.unwrap(), 0);
        for key in &refresh_keys {
            assert!(db.get(key.as_slice()).await.unwrap().is_none());
        }
        assert!(
            db.get(auth_code_key.as_slice()).await.unwrap().is_some(),
            "retiring refresh tokens must not touch the auth code prefix"
        );
    }

    async fn make_summary_store() -> (tempfile::TempDir, Arc<RunSummaryStore>) {
        let (directory, store) = store_test_support::sqlite_summary_store().await;
        (directory, Arc::new(store))
    }

    fn sample_run_spec(label: &str) -> RunSpec {
        let mut graph = Graph::new("night-sky");
        graph.attrs.insert(
            "goal".to_string(),
            AttrValue::String("map the constellations".to_string()),
        );
        RunSpec {
            run_id: test_run_id(label),
            settings: WorkflowSettings::default(),
            graph,
            graph_source: None,
            workflow_slug: Some("night-sky".to_string()),
            workflow_version_id: None,
            target: None,
            automation: None,
            source_directory: Some(format!("/tmp/{label}")),
            labels: std::collections::HashMap::from([("team".to_string(), "infra".to_string())]),
            provenance: test_support::test_run_provenance(),
            manifest_blob: None,
            definition_blob: None,
            spec_blob: None,
            git: Some(fabro_types::GitContext {
                origin_url: "https://github.com/fabro-sh/fabro".to_string(),
                branch:     "main".to_string(),
                sha:        None,
                dirty:      fabro_types::DirtyStatus::Clean,
            }),
            fork_source_ref: None,
        }
    }

    fn event_payload(
        run_id: &str,
        ts: &str,
        event: &str,
        properties: &serde_json::Value,
    ) -> EventPayload {
        event_payload_with_node(run_id, ts, event, properties, None)
    }

    fn event_payload_with_node(
        run_id: &str,
        ts: &str,
        event: &str,
        properties: &serde_json::Value,
        node_id: Option<&str>,
    ) -> EventPayload {
        EventPayload::new(
            serde_json::json!({
                "id": format!("evt-{run_id}-{event}"),
                "ts": ts,
                "run_id": test_run_id(run_id).to_string(),
                "event": event,
                "node_id": node_id,
                "stage_id": node_id.map(|node| format!("{node}@1")),
                "properties": properties,
            }),
            &test_run_id(run_id),
        )
        .unwrap()
    }

    async fn append_created(run: &RunDatabase, label: &str, created_at: DateTime<Utc>) {
        let run_spec = sample_run_spec(label);
        run.append_event(&event_payload(
            label,
            &created_at.to_rfc3339(),
            "run.created",
            &serde_json::json!({
                "settings": run_spec.settings,
                "graph": run_spec.graph,
                "workflow_slug": run_spec.workflow_slug,
                "source_directory": run_spec.source_directory,
                "git": run_spec.git,
                "labels": run_spec.labels,
                "provenance": run_spec.provenance,
            }),
        ))
        .await
        .unwrap();
    }

    async fn append_created_with_parent(
        run: &RunDatabase,
        label: &str,
        created_at: DateTime<Utc>,
        parent_id: RunId,
    ) {
        let run_spec = sample_run_spec(label);
        run.append_event(&event_payload(
            label,
            &created_at.to_rfc3339(),
            "run.created",
            &serde_json::json!({
                "settings": run_spec.settings,
                "graph": run_spec.graph,
                "workflow_slug": run_spec.workflow_slug,
                "source_directory": run_spec.source_directory,
                "git": run_spec.git,
                "labels": run_spec.labels,
                "parent_id": parent_id,
                "provenance": run_spec.provenance,
            }),
        ))
        .await
        .unwrap();
    }

    async fn append_runnable(run: &RunDatabase, label: &str, created_at: DateTime<Utc>) {
        append_created(run, label, created_at).await;
        run.append_event(&event_payload(
            label,
            "2026-03-27T12:00:01Z",
            "run.submitted",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            label,
            "2026-03-27T12:00:02Z",
            "run.start_requested",
            &serde_json::json!({ "resume": false }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            label,
            "2026-03-27T12:00:03Z",
            "run.runnable",
            &serde_json::json!({ "source": "start_requested" }),
        ))
        .await
        .unwrap();
    }

    fn workflow_failure_payload(label: &str) -> EventPayload {
        event_payload(
            label,
            "2026-03-27T12:00:04Z",
            "run.failed",
            &serde_json::json!({
                "failure": {
                    "reason": "workflow_error",
                    "detail": {
                        "message": "workflow failed",
                        "category": "deterministic"
                    }
                },
                "timing": {
                    "wall_time_ms": 1,
                    "inference_time_ms": 0,
                    "tool_time_ms": 0,
                    "active_time_ms": 0
                },
            }),
        )
    }

    async fn append_completed(run: &RunDatabase, label: &str, created_at: DateTime<Utc>) {
        append_running(run, label, created_at).await;
        run.append_event(&event_payload(
            label,
            "2026-03-27T12:00:03Z",
            "run.completed",
            &serde_json::json!({
                "timing": {"wall_time_ms": 3210, "inference_time_ms": 0, "tool_time_ms": 0, "active_time_ms": 0},
                "artifact_count": 1,
                "status": "succeeded",
                "reason": "completed",
                "total_cost": 1.25,
            }),
        ))
        .await
        .unwrap();
    }

    async fn append_running(run: &RunDatabase, label: &str, created_at: DateTime<Utc>) {
        append_created(run, label, created_at).await;
        run.append_event(&event_payload(
            label,
            "2026-03-27T12:00:01Z",
            "run.runnable",
            &serde_json::json!({ "source": "start_requested" }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            label,
            "2026-03-27T12:00:02Z",
            "run.starting",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            label,
            "2026-03-27T12:00:03Z",
            "run.running",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
    }

    async fn list_paths(store: Arc<dyn ObjectStore>, prefix: &str) -> Vec<String> {
        let mut items = store
            .list(Some(&Path::from(prefix.to_string())))
            .map_ok(|meta| meta.location.to_string())
            .try_collect::<Vec<_>>()
            .await
            .unwrap();
        items.sort();
        items
    }

    #[tokio::test]
    async fn create_open_list_and_delete_full_lifecycle_in_shared_db() {
        let (object_store, store) = make_store();
        let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
        let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
        append_completed(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await;
        append_created(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await;

        let summary = store
            .list_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(summary.len(), 2);
        assert_eq!(summary[0].id, test_run_id("run-2"));
        assert_eq!(summary[1].id, test_run_id("run-1"));
        assert_eq!(summary[1].workflow.name, None);
        assert_eq!(summary[1].workflow.graph_name.as_deref(), Some("night-sky"));
        assert_eq!(summary[1].goal, "map the constellations");
        assert_eq!(summary[1].lifecycle.status, RunStatus::Succeeded {
            reason: SuccessReason::Completed,
        });

        let reopened = store.open_run(&test_run_id("run-1")).await.unwrap();
        let stored = reopened.state().await.unwrap().spec;
        assert_eq!(stored.run_id, test_run_id("run-1"));

        store.delete_run(&test_run_id("run-1")).await.unwrap();
        assert!(store.open_run(&test_run_id("run-1")).await.is_err());
        let remaining = store
            .list_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(remaining.len(), 1);
        assert_eq!(remaining[0].id, test_run_id("run-2"));
        assert!(!list_paths(object_store, "runs/").await.is_empty());
    }

    #[tokio::test]
    async fn delete_run_keeps_global_cas_blobs() {
        let (_object_store, store) = make_store();
        let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
        let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
        append_created(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await;
        append_created(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await;

        let shared_blob = br#"{"summary":"shared"}"#;
        let shared_blob_hash = run_1.write_blob(shared_blob).await.unwrap();

        store.delete_run(&test_run_id("run-1")).await.unwrap();

        let reopened = store.open_run(&test_run_id("run-2")).await.unwrap();
        let read = reopened.read_blob(&shared_blob_hash).await.unwrap();
        assert_eq!(read.as_deref(), Some(shared_blob.as_slice()));
    }

    #[tokio::test]
    async fn open_run_reader_is_read_only() {
        let (_object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
        let blob = br#"{"summary":"readable"}"#;
        let blob_hash = run.write_blob(blob).await.unwrap();

        // Evict the cached writer so the reader is built through the real
        // `open_run_reader` construction path, not a clone of the writer.
        let _ = store.remove_active_run(&test_run_id("run-1")).await;

        let reader = store.open_run_reader(&test_run_id("run-1")).await.unwrap();
        assert_eq!(
            reader.read_blob(&blob_hash).await.unwrap().as_deref(),
            Some(blob.as_slice())
        );
        let err = reader.write_blob(b"blocked").await.unwrap_err();
        assert!(matches!(err, Error::ReadOnly));

        let err = reader
            .append_event(&event_payload(
                "run-1",
                "2026-03-27T12:00:01Z",
                "run.completed",
                &serde_json::json!({ "reason": "completed" }),
            ))
            .await
            .unwrap_err();
        assert!(matches!(err, Error::ReadOnly));
    }

    #[tokio::test]
    async fn append_event_if_evaluates_latest_projection_before_appending() {
        let (_object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
        let initial_title = run.state().await.unwrap().title().into_owned();

        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:01Z",
            "run.title.updated",
            &serde_json::json!({ "title": "User title" }),
        ))
        .await
        .unwrap();

        let generated_update = event_payload(
            "run-1",
            "2026-03-27T12:00:02Z",
            "run.title.updated",
            &serde_json::json!({ "title": "Generated title" }),
        );
        let appended = run
            .append_event_if(&generated_update, |projection| {
                projection.title() == initial_title
            })
            .await
            .unwrap();

        assert_eq!(appended, None);
        assert_eq!(run.state().await.unwrap().title(), "User title");
        assert_eq!(run.list_events().await.unwrap().len(), 2);
    }

    #[tokio::test]
    async fn rejected_transition_writes_nothing_and_preserves_projection_cache() {
        let (_object_store, store) = make_store();
        let run_id = test_run_id("run-1");
        let run = store.create_run(&run_id).await.unwrap();
        append_runnable(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
        let events_before = run.list_events().await.unwrap();

        let err = run
            .append_event(&workflow_failure_payload("run-1"))
            .await
            .unwrap_err();

        let Error::EventRejected { source } = err else {
            panic!("expected event rejection");
        };
        assert!(matches!(
            *source,
            Error::InvalidTransition(fabro_types::InvalidTransition {
                from: RunStatus::Runnable,
                to:   RunStatus::Failed {
                    reason: FailureReason::WorkflowError,
                },
            })
        ));
        assert_eq!(run.list_events().await.unwrap(), events_before);
        assert_eq!(run.state().await.unwrap().status, RunStatus::Runnable);
        let cached = store.get_cached_run(&run_id).await.unwrap().unwrap();
        assert_eq!(cached.last_seq, 4);
        assert_eq!(cached.projection.status, RunStatus::Runnable);
    }

    #[tokio::test]
    async fn rejected_transition_leaves_reconciled_summary_present() {
        let (_object_store, store) = make_store();
        let (_directory, summaries) = make_summary_store().await;
        store.attach_run_summary_store(Arc::clone(&summaries));
        let run_id = test_run_id("run-1");
        let run = store.create_run(&run_id).await.unwrap();
        append_runnable(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        let err = run
            .append_event(&workflow_failure_payload("run-1"))
            .await
            .unwrap_err();
        assert!(matches!(err, Error::EventRejected { .. }));

        let entries = store
            .list_cached_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        summaries.reconcile(&entries).await.unwrap();
        let summary = summaries.get(&run_id, Utc::now()).await.unwrap().unwrap();
        assert_eq!(summary.lifecycle.status, RunStatus::Runnable);
    }

    #[tokio::test]
    async fn committed_append_succeeds_when_summary_update_fails_and_is_repairable() {
        let (object_store, store) = make_store();
        let (directory, summaries) = make_summary_store().await;
        store.attach_run_summary_store(Arc::clone(&summaries));
        let run_id = test_run_id("run-1");
        let run = store.create_run(&run_id).await.unwrap();
        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
        summaries.close_pool().await;

        let result = run
            .append_event_envelope(&event_payload(
                "run-1",
                "2026-03-27T12:00:01Z",
                "run.title.updated",
                &serde_json::json!({ "title": "Committed title" }),
            ))
            .await;

        assert!(result.is_ok(), "committed append returned {result:?}");
        assert_eq!(run.list_events().await.unwrap().len(), 2);
        let cached = store.get_cached_run(&run_id).await.unwrap().unwrap();
        assert_eq!(cached.last_seq, 2);
        assert_eq!(cached.summary.title, "Committed title");
        let stored = run.get_event(2).await.unwrap().unwrap();
        assert_eq!(stored.event, result.unwrap().event);

        let repaired_summaries =
            Arc::new(store_test_support::sqlite_summary_store_at(directory.path()).await);
        let stale = repaired_summaries
            .get(&run_id, Utc::now())
            .await
            .unwrap()
            .unwrap();
        assert_ne!(stale.title, "Committed title");

        let reopened = store_test_support::test_database(
            object_store,
            "runs/",
            Duration::from_millis(1),
            None,
        );
        reopened.attach_run_summary_store(Arc::clone(&repaired_summaries));
        reopened.warm_projection_cache().await.unwrap();
        let repaired = repaired_summaries
            .get(&run_id, Utc::now())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(repaired.title, "Committed title");
    }

    #[tokio::test]
    async fn first_event_is_validated_before_write() {
        let (_object_store, store) = make_store();
        let run_id = test_run_id("run-1");
        let run = store.create_run(&run_id).await.unwrap();
        let invalid_first = event_payload(
            "run-1",
            "2026-03-27T12:00:00Z",
            "run.title.updated",
            &serde_json::json!({ "title": "Too early" }),
        );

        let err = run.append_event(&invalid_first).await.unwrap_err();

        assert!(matches!(err, Error::EventRejected { .. }));
        assert!(run.list_events().await.unwrap().is_empty());

        append_created(&run, "run-1", dt("2026-03-27T12:00:01Z")).await;
        assert_eq!(run.list_events().await.unwrap().len(), 1);
        assert!(run.state().await.is_ok());
    }

    #[tokio::test]
    async fn malformed_optional_envelope_field_is_rejected_before_write() {
        let (_object_store, store) = make_store();
        let run_id = test_run_id("run-1");
        let run = store.create_run(&run_id).await.unwrap();
        let malformed = EventPayload::new(
            serde_json::json!({
                "id": "evt-created",
                "ts": "2026-03-27T12:00:00Z",
                "run_id": run_id.to_string(),
                "event": "run.created",
                "node_id": 42,
                "properties": {
                    "settings": WorkflowSettings::default(),
                    "graph": Graph::new("test"),
                    "provenance": test_support::test_run_provenance(),
                },
            }),
            &run_id,
        )
        .unwrap();

        let err = run.append_event(&malformed).await.unwrap_err();

        assert!(matches!(err, Error::InvalidEvent(_)));
        assert!(run.list_events().await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn control_request_events_set_pending_control_without_overwriting_status() {
        let (_object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_running(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:02Z",
            "run.pause.requested",
            &serde_json::json!({ "action": "pause" }),
        ))
        .await
        .unwrap();

        let summary = store
            .list_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(summary.len(), 1);
        assert_eq!(summary[0].lifecycle.status, RunStatus::Running);
        assert_eq!(
            summary[0].lifecycle.pending_control,
            Some(RunControlAction::Pause)
        );
    }

    #[tokio::test]
    async fn parent_id_is_projected_from_created_and_parent_events() {
        let (_object_store, store) = make_store();
        let parent_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
        let parent_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
        let child = store.create_run(&test_run_id("run-3")).await.unwrap();
        append_created(&parent_1, "run-1", dt("2026-03-27T12:00:00Z")).await;
        append_created(&parent_2, "run-2", dt("2026-03-27T12:00:10Z")).await;
        append_created_with_parent(
            &child,
            "run-3",
            dt("2026-03-27T12:00:20Z"),
            test_run_id("run-1"),
        )
        .await;

        let initial = store.open_run(&test_run_id("run-3")).await.unwrap();
        assert_eq!(
            initial.state().await.unwrap().parent_id,
            Some(test_run_id("run-1"))
        );
        assert_eq!(
            store
                .get_cached_summary(&test_run_id("run-3"), Utc::now())
                .await
                .unwrap()
                .unwrap()
                .parent_id,
            Some(test_run_id("run-1"))
        );

        child
            .append_event(&event_payload(
                "run-3",
                "2026-03-27T12:00:21Z",
                "run.parent.linked",
                &serde_json::json!({
                    "previous_parent_id": test_run_id("run-1"),
                    "parent_id": test_run_id("run-2"),
                }),
            ))
            .await
            .unwrap();
        assert_eq!(
            store
                .get_cached_summary(&test_run_id("run-3"), Utc::now())
                .await
                .unwrap()
                .unwrap()
                .parent_id,
            Some(test_run_id("run-2"))
        );
        assert!(
            store
                .list_runs(
                    &ListRunsQuery {
                        parent_id: Some(test_run_id("run-1")),
                        ..ListRunsQuery::default()
                    },
                    Utc::now()
                )
                .await
                .unwrap()
                .is_empty()
        );
        assert_eq!(
            store
                .list_runs(
                    &ListRunsQuery {
                        parent_id: Some(test_run_id("run-2")),
                        ..ListRunsQuery::default()
                    },
                    Utc::now()
                )
                .await
                .unwrap()
                .into_iter()
                .map(|summary| summary.id)
                .collect::<Vec<_>>(),
            vec![test_run_id("run-3")]
        );

        child
            .append_event(&event_payload(
                "run-3",
                "2026-03-27T12:00:22Z",
                "run.parent.unlinked",
                &serde_json::json!({
                    "previous_parent_id": test_run_id("run-2"),
                }),
            ))
            .await
            .unwrap();
        assert_eq!(
            store
                .get_cached_summary(&test_run_id("run-3"), Utc::now())
                .await
                .unwrap()
                .unwrap()
                .parent_id,
            None
        );
        assert!(
            store
                .list_runs(
                    &ListRunsQuery {
                        parent_id: Some(test_run_id("run-2")),
                        ..ListRunsQuery::default()
                    },
                    Utc::now()
                )
                .await
                .unwrap()
                .is_empty()
        );
    }

    #[tokio::test]
    async fn list_runs_filters_by_parent_id() {
        let (_object_store, store) = make_store();
        let parent = store.create_run(&test_run_id("run-1")).await.unwrap();
        let child = store.create_run(&test_run_id("run-2")).await.unwrap();
        let unrelated = store.create_run(&test_run_id("run-3")).await.unwrap();
        append_created(&parent, "run-1", dt("2026-03-27T12:00:00Z")).await;
        append_created_with_parent(
            &child,
            "run-2",
            dt("2026-03-27T12:00:10Z"),
            test_run_id("run-1"),
        )
        .await;
        append_created(&unrelated, "run-3", dt("2026-03-27T12:00:20Z")).await;

        let summaries = store
            .list_runs(
                &ListRunsQuery {
                    parent_id: Some(test_run_id("run-1")),
                    ..ListRunsQuery::default()
                },
                Utc::now(),
            )
            .await
            .unwrap();

        assert_eq!(summaries.len(), 1);
        assert_eq!(summaries[0].id, test_run_id("run-2"));
        assert_eq!(summaries[0].parent_id, Some(test_run_id("run-1")));
    }

    #[tokio::test]
    async fn run_summary_includes_children_count() {
        let (_object_store, store) = make_store();
        let parent = store.create_run(&test_run_id("run-1")).await.unwrap();
        let child_a = store.create_run(&test_run_id("run-2")).await.unwrap();
        let child_b = store.create_run(&test_run_id("run-3")).await.unwrap();
        let unrelated = store.create_run(&test_run_id("run-4")).await.unwrap();
        append_created(&parent, "run-1", dt("2026-03-27T12:00:00Z")).await;
        append_created_with_parent(
            &child_a,
            "run-2",
            dt("2026-03-27T12:00:10Z"),
            test_run_id("run-1"),
        )
        .await;
        append_created_with_parent(
            &child_b,
            "run-3",
            dt("2026-03-27T12:00:20Z"),
            test_run_id("run-1"),
        )
        .await;
        append_created(&unrelated, "run-4", dt("2026-03-27T12:00:30Z")).await;

        let summaries = store
            .list_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();

        let parent_summary = summaries
            .iter()
            .find(|r| r.id == test_run_id("run-1"))
            .expect("parent summary should be present");
        assert_eq!(parent_summary.children_count, 2);

        let child_summary = summaries
            .iter()
            .find(|r| r.id == test_run_id("run-2"))
            .expect("child summary should be present");
        assert_eq!(child_summary.children_count, 0);

        let unrelated_summary = summaries
            .iter()
            .find(|r| r.id == test_run_id("run-4"))
            .expect("unrelated summary should be present");
        assert_eq!(unrelated_summary.children_count, 0);
    }

    #[tokio::test]
    async fn cached_summary_overlays_live_timing_without_mutating_cached_snapshot() {
        let (_object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:01Z",
            "run.started",
            &serde_json::json!({ "name": "Test run" }),
        ))
        .await
        .unwrap();

        let now = dt("2026-03-27T12:00:06Z");
        let expected = Some(fabro_types::RunTiming::wall_only(5_000));

        let summary = store
            .get_cached_summary(&test_run_id("run-1"), now)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(summary.timing, expected);

        let listed = store
            .list_cached_runs(&ListRunsQuery::default(), now)
            .await
            .unwrap();
        assert_eq!(listed[0].summary.timing, expected);

        let cached = store
            .get_cached_run(&test_run_id("run-1"))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(cached.summary.timing, None);
    }

    #[tokio::test]
    async fn control_effect_events_clear_pending_control_and_update_status() {
        let (_object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_running(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:02Z",
            "run.pause.requested",
            &serde_json::json!({ "action": "pause" }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:03Z",
            "run.paused",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:04Z",
            "run.unpause.requested",
            &serde_json::json!({ "action": "unpause" }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:05Z",
            "run.unpaused",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:06Z",
            "run.cancel.requested",
            &serde_json::json!({ "action": "cancel" }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:07Z",
            "run.failed",
            &serde_json::json!({
                "failure": {
                    "reason": "cancelled",
                    "detail": {
                        "message": "cancelled",
                        "category": "canceled"
                    }
                },
                "timing": {"wall_time_ms": 1, "inference_time_ms": 0, "tool_time_ms": 0, "active_time_ms": 0},
            }),
        ))
        .await
        .unwrap();

        let summary = store
            .list_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(summary.len(), 1);
        assert_eq!(summary[0].lifecycle.status, RunStatus::Failed {
            reason: FailureReason::Cancelled,
        });
        assert_eq!(summary[0].lifecycle.pending_control, None);
    }

    #[tokio::test]
    async fn reader_sees_cached_projection_and_recent_events_for_active_run() {
        let (_object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        let reader = store.open_run_reader(&test_run_id("run-1")).await.unwrap();
        let state = reader.state().await.unwrap();
        assert_eq!(state.spec.run_id, test_run_id("run-1"));

        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:01Z",
            "run.runnable",
            &serde_json::json!({ "source": "start_requested" }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:02Z",
            "run.starting",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:03Z",
            "run.running",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:04Z",
            "run.completed",
            &serde_json::json!({
                "timing": {"wall_time_ms": 3210, "inference_time_ms": 0, "tool_time_ms": 0, "active_time_ms": 0},
                "artifact_count": 1,
                "status": "succeeded",
                "reason": "completed",
                "total_cost": 1.25,
            }),
        ))
        .await
        .unwrap();

        let recent = reader.list_events_from_with_limit(4, 10).await.unwrap();
        assert_eq!(recent.len(), 2);
        assert_eq!(recent[0].seq, 4);
    }

    #[tokio::test]
    async fn reopening_store_rebuilds_from_shared_db() {
        let (object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        let reopened =
            store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
        let summary = reopened
            .list_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(summary.len(), 1);
        assert_eq!(summary[0].id, test_run_id("run-1"));
        assert_eq!(summary[0].lifecycle.status, RunStatus::Succeeded {
            reason: SuccessReason::Completed,
        });
    }

    #[tokio::test]
    async fn projection_cache_warmup_lists_newest_first_and_applies_date_filters() {
        let (object_store, store) = make_store();
        let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
        let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
        append_completed(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await;
        append_running(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await;

        let reopened =
            store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
        reopened.warm_projection_cache().await.unwrap();

        let entries = reopened
            .list_cached_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(
            entries.iter().map(|entry| entry.run_id).collect::<Vec<_>>(),
            vec![test_run_id("run-2"), test_run_id("run-1")]
        );
        assert_eq!(entries[0].summary.lifecycle.status, RunStatus::Running);
        assert_eq!(entries[0].projection.spec().run_id, test_run_id("run-2"));
        assert_eq!(entries[0].last_seq, 4);

        let filtered = reopened
            .list_cached_runs(
                &ListRunsQuery {
                    start:     Some(test_run_id("run-2").created_at()),
                    end:       Some(
                        test_run_id("run-2").created_at() + chrono::Duration::seconds(1),
                    ),
                    parent_id: None,
                },
                Utc::now(),
            )
            .await
            .unwrap();
        assert_eq!(filtered.len(), 1);
        assert_eq!(filtered[0].run_id, test_run_id("run-2"));

        let cached = reopened
            .get_cached_run(&test_run_id("run-1"))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(cached.summary.lifecycle.status, RunStatus::Succeeded {
            reason: SuccessReason::Completed,
        });
    }

    #[tokio::test]
    async fn projection_cache_warmup_skips_unreplayable_catalog_run() {
        let (object_store, store) = make_store();
        let good_run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_completed(&good_run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        let bad_run_id = test_run_id("run-2");
        store
            .catalog_index()
            .await
            .unwrap()
            .add(&bad_run_id)
            .await
            .unwrap();
        store
            .put_unvalidated_run_event(
                &bad_run_id,
                1,
                &serde_json::json!({ "not": "a valid run event" }),
            )
            .await
            .unwrap();

        let reopened =
            store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
        reopened.warm_projection_cache().await.unwrap();

        let entries = reopened
            .list_cached_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].run_id, test_run_id("run-1"));
        assert!(
            reopened
                .get_cached_run(&bad_run_id)
                .await
                .unwrap()
                .is_none()
        );
        assert!(reopened.runs().find(&bad_run_id).await.unwrap().is_none());
    }

    #[tokio::test]
    async fn list_unreadable_runs_reports_catalog_entries_that_fail_projection() {
        let (object_store, store) = make_store();
        let good_run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_completed(&good_run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        let bad_run_id = test_run_id("run-2");
        store
            .catalog_index()
            .await
            .unwrap()
            .add(&bad_run_id)
            .await
            .unwrap();
        let mut run_spec = serde_json::to_value(sample_run_spec("run-2")).unwrap();
        let run_settings = run_spec
            .get_mut("settings")
            .and_then(|settings| settings.get_mut("run"))
            .and_then(serde_json::Value::as_object_mut)
            .unwrap();
        run_settings.remove("integrations");
        store
            .put_unvalidated_run_event(
                &bad_run_id,
                1,
                &serde_json::json!({
                    "id": "evt-run-2-run.created",
                    "ts": "2026-03-27T12:00:10Z",
                    "run_id": bad_run_id,
                    "event": "run.created",
                    "properties": {
                        "settings": run_spec["settings"],
                        "graph": run_spec["graph"],
                        "workflow_slug": run_spec["workflow_slug"],
                        "source_directory": run_spec["source_directory"],
                        "git": run_spec["git"],
                        "labels": run_spec["labels"],
                    },
                }),
            )
            .await
            .unwrap();

        let reopened =
            store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
        let unreadable = reopened.list_unreadable_runs().await.unwrap();

        assert_eq!(unreadable.len(), 1);
        assert_eq!(unreadable[0].run_id, bad_run_id);
        assert_eq!(unreadable[0].created_at, bad_run_id.created_at());
        assert!(
            unreadable[0].error.contains("missing field `integrations`"),
            "expected missing integrations error, got: {}",
            unreadable[0].error
        );
    }

    #[tokio::test]
    async fn append_event_refreshes_projection_cache_and_delete_removes_it() {
        let (_object_store, store) = make_store();
        let (_directory, summaries) = make_summary_store().await;
        store.attach_run_summary_store(Arc::clone(&summaries));
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
        store.warm_projection_cache().await.unwrap();

        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:01Z",
            "run.runnable",
            &serde_json::json!({ "source": "start_requested" }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:01Z",
            "run.starting",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:02Z",
            "run.running",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload_with_node(
            "run-1",
            "2026-03-27T12:00:03Z",
            "stage.started",
            &serde_json::json!({
                "index": 0,
                "handler_type": "prompt",
                "attempt": 1,
                "max_attempts": 1,
            }),
            Some("review"),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload_with_node(
            "run-1",
            "2026-03-27T12:00:04Z",
            "interview.started",
            &serde_json::json!({
                "question_id": "q-1",
                "question": "Approve deploy?",
                "stage": "review",
                "question_type": "yes_no",
                "options": [],
                "allow_freeform": false,
                "context_display": null,
                "timeout_seconds": null,
            }),
            Some("review"),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload_with_node(
            "run-1",
            "2026-03-27T12:00:05Z",
            "checkpoint.completed",
            &serde_json::json!({
                "status": "running",
                "current_node": "review",
                "completed_nodes": [],
                "node_retries": {},
                "context_values": {},
                "node_outcomes": {},
                "next_node_id": "review",
                "git_commit_sha": "abc123",
                "loop_failure_signatures": {},
                "restart_failure_signatures": {},
                "node_visits": { "review": 1 },
            }),
            Some("review"),
        ))
        .await
        .unwrap();

        let cached = store
            .get_cached_run(&test_run_id("run-1"))
            .await
            .unwrap()
            .unwrap();
        assert_eq!(cached.summary.lifecycle.status, RunStatus::Running);
        assert_eq!(cached.last_seq, 7);
        assert_eq!(
            cached
                .projection
                .stage(&StageId::new("review", 1))
                .unwrap()
                .effective_state(),
            fabro_types::StageState::Running
        );
        assert_eq!(
            cached.projection.pending_interviews["q-1"].question.text,
            "Approve deploy?"
        );
        assert_eq!(
            cached
                .projection
                .current_checkpoint()
                .unwrap()
                .git_commit_sha
                .as_deref(),
            Some("abc123")
        );

        let cached_summaries = store
            .list_runs(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        let projected = store
            .list_runs_with_projection(&ListRunsQuery::default(), Utc::now())
            .await
            .unwrap();
        assert_eq!(cached_summaries, vec![cached.summary.clone()]);
        assert_eq!(projected[0].0, cached.summary);
        assert_eq!(
            projected[0]
                .1
                .current_checkpoint()
                .unwrap()
                .git_commit_sha
                .as_deref(),
            cached
                .projection
                .current_checkpoint()
                .unwrap()
                .git_commit_sha
                .as_deref()
        );
        let comparison_time = dt("2026-03-27T12:00:10Z");
        let cache_summary = store
            .get_cached_summary(&test_run_id("run-1"), comparison_time)
            .await
            .unwrap()
            .unwrap();
        let sql_summary = summaries
            .get(&test_run_id("run-1"), comparison_time)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(sql_summary, cache_summary);

        store.delete_run(&test_run_id("run-1")).await.unwrap();
        assert!(
            store
                .get_cached_run(&test_run_id("run-1"))
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            store
                .list_cached_runs(&ListRunsQuery::default(), Utc::now())
                .await
                .unwrap()
                .is_empty()
        );
        assert!(
            summaries
                .get(&test_run_id("run-1"), Utc::now())
                .await
                .unwrap()
                .is_none()
        );
    }

    #[tokio::test]
    async fn projection_cache_warmup_backfills_sqlite_run_summaries() {
        let (object_store, store) = make_store();
        let run = store.create_run(&test_run_id("run-1")).await.unwrap();
        append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        let reopened =
            store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
        let (_directory, summaries) = make_summary_store().await;
        reopened.attach_run_summary_store(Arc::clone(&summaries));
        reopened.warm_projection_cache().await.unwrap();

        let summary = summaries
            .get(&test_run_id("run-1"), Utc::now())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(summary.lifecycle.status, RunStatus::Succeeded {
            reason: SuccessReason::Completed,
        });
    }

    #[tokio::test]
    async fn opening_cached_run_does_not_read_older_event_history() {
        let (object_store, store) = make_store();
        let run_id = test_run_id("run-1");
        let run = store.create_run(&run_id).await.unwrap();
        append_completed(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;

        let reopened =
            store_test_support::test_database(object_store, "runs", Duration::from_millis(1), None);
        reopened.warm_projection_cache().await.unwrap();

        // If opening or projecting the run starts at the beginning, this
        // unreadable old key makes the operation fail. A hydrated run starts
        // after the shared projection's last sequence instead.
        let mut unreadable_old_key = keys::run_event_seq_prefix(&run_id, 2).as_ref().to_vec();
        unreadable_old_key.push(0xff);
        reopened
            .open_db()
            .await
            .unwrap()
            .put(unreadable_old_key, b"invalid json")
            .await
            .unwrap();

        let fresh_writer = reopened.open_run(&run_id).await.unwrap();
        assert_eq!(fresh_writer.last_event_seq().await.unwrap(), Some(5));
        let state = fresh_writer.state().await.unwrap();
        assert_eq!(state.status, RunStatus::Succeeded {
            reason: SuccessReason::Completed,
        });

        let seq = fresh_writer
            .append_event(&event_payload(
                "run-1",
                "2026-03-27T12:00:05Z",
                "run.title.updated",
                &serde_json::json!({ "title": "Renamed completed run" }),
            ))
            .await
            .unwrap();
        assert_eq!(seq, 6);
    }

    #[tokio::test]
    async fn append_event_hydrates_local_projection_cache_for_fresh_writer() {
        let (object_store, store) = make_store();
        let run_id = test_run_id("run-1");
        let run = store.create_run(&run_id).await.unwrap();
        append_created(&run, "run-1", dt("2026-03-27T12:00:00Z")).await;
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:01Z",
            "run.runnable",
            &serde_json::json!({ "source": "start_requested" }),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:02Z",
            "run.starting",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&event_payload(
            "run-1",
            "2026-03-27T12:00:03Z",
            "run.running",
            &serde_json::json!({}),
        ))
        .await
        .unwrap();
        run.append_event(&workflow_failure_payload("run-1"))
            .await
            .unwrap();

        let reopened = store_test_support::test_database(
            Arc::clone(&object_store),
            "runs/",
            Duration::from_millis(1),
            None,
        );
        let fresh_writer = reopened.open_run(&run_id).await.unwrap();
        fresh_writer
            .append_event(&event_payload(
                "run-1",
                "2026-03-27T12:00:05Z",
                "run.title.updated",
                &serde_json::json!({ "title": "Renamed failed run" }),
            ))
            .await
            .unwrap();

        let state = fresh_writer.state().await.unwrap();
        assert_eq!(state.title, "Renamed failed run");
        assert_eq!(state.status, RunStatus::Failed {
            reason: FailureReason::WorkflowError,
        });

        let cached = reopened.get_cached_run(&run_id).await.unwrap().unwrap();
        assert_eq!(cached.summary.title, "Renamed failed run");
        assert_eq!(cached.summary.lifecycle.status, RunStatus::Failed {
            reason: FailureReason::WorkflowError,
        });
    }
}
