diff --git a/Cargo.lock b/Cargo.lock index 2e2f4f415..4c79e562d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1944,6 +1944,7 @@ dependencies = [ "tokio", "tokio-stream", "tracing", + "ulid", ] [[package]] diff --git a/lib/crates/fabro-cli/src/commands/doctor.rs b/lib/crates/fabro-cli/src/commands/doctor.rs index 2e5b13509..fbbc40770 100644 --- a/lib/crates/fabro-cli/src/commands/doctor.rs +++ b/lib/crates/fabro-cli/src/commands/doctor.rs @@ -4,7 +4,7 @@ use std::sync::LazyLock; use base64::Engine as _; use base64::engine::general_purpose::STANDARD as BASE64_STANDARD; -use fabro_config::server::{ApiAuthStrategy, AuthProvider, load_server_settings}; +use fabro_config::server::{self, ApiAuthStrategy, AuthProvider}; use fabro_config::user::{default_user_config_path, legacy_user_config_path}; use fabro_llm::client::Client as LlmClient; use fabro_llm::types::{Message, Request}; @@ -938,7 +938,7 @@ pub(crate) async fn run_doctor( let daytona_configured = std::env::var("DAYTONA_API_KEY").is_ok(); - let server_settings = load_server_settings(None).unwrap_or_default(); + let server_settings = server::load_server_settings(None).unwrap_or_default(); let api_status = { let api = server_settings.api.clone().unwrap_or_default(); diff --git a/lib/crates/fabro-cli/tests/it/cmd/detached.rs b/lib/crates/fabro-cli/tests/it/cmd/detached.rs index 6187c7141..e981843ec 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/detached.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/detached.rs @@ -303,7 +303,7 @@ digraph Test { let run_dir = before_summary["run_dir"].as_str().unwrap().to_string(); fabro_json_snapshot!(context, &before_summary, @r#" { - "run_dir": "[DRY_RUN_DIR]", + "run_dir": "[RUN_DIR]", "start_time": "[TIMESTAMP]", "conclusion_timestamp": "[TIMESTAMP]", "conclusion_status": "success" diff --git a/lib/crates/fabro-cli/tests/it/cmd/ps.rs b/lib/crates/fabro-cli/tests/it/cmd/ps.rs index 8f629ad62..260609588 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/ps.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/ps.rs @@ -66,25 +66,28 @@ fn ps_all_json_lists_created_and_completed_runs() { let mut cmd = context.ps(); cmd.args(["-a", "--json"]); - fabro_snapshot!(filters, cmd, @r###" + fabro_snapshot!(filters, cmd, @r#" success: true exit_code: 0 ----- stdout ----- [ { "run_id": "[ULID]", - "dir_name": "[DATE]-dry-run-[ULID]", + "dir_name": "20260403-[ULID]", "workflow_name": "Simple", "workflow_slug": "simple", "status": "submitted", + "status_reason": null, "start_time": "[TIMESTAMP]", "labels": {}, + "duration_ms": null, + "total_cost": null, "host_repo_path": "[TEMP_DIR]", "goal": "Run tests and report results" }, { "run_id": "[ULID]", - "dir_name": "[DATE]-dry-run-[ULID]", + "dir_name": "20260403-[ULID]", "workflow_name": "Simple", "workflow_slug": "simple", "status": "succeeded", @@ -92,12 +95,13 @@ fn ps_all_json_lists_created_and_completed_runs() { "start_time": "[TIMESTAMP]", "labels": {}, "duration_ms": [DURATION_MS], + "total_cost": null, "host_repo_path": "[TEMP_DIR]", "goal": "Run tests and report results" } ] ----- stderr ----- - "###); + "#); } #[test] @@ -169,14 +173,14 @@ fn ps_filters_by_workflow_and_label() { "suite=alpha", ]); - fabro_snapshot!(filters, cmd, @r###" + fabro_snapshot!(filters, cmd, @r#" success: true exit_code: 0 ----- stdout ----- [ { "run_id": "[ULID]", - "dir_name": "[DATE]-dry-run-[ULID]", + "dir_name": "20260403-[ULID]", "workflow_name": "Simple", "workflow_slug": "simple", "status": "succeeded", @@ -186,10 +190,11 @@ fn ps_filters_by_workflow_and_label() { "suite": "alpha" }, "duration_ms": [DURATION_MS], + "total_cost": null, "host_repo_path": "[TEMP_DIR]", "goal": "Run tests and report results" } ] ----- stderr ----- - "###); + "#); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs index 5facf751c..b54611295 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs @@ -69,7 +69,8 @@ fn rewind_list_prints_timeline_for_completed_git_run() { ----- stdout ----- ----- stderr ----- @ Node Details - @1 step_one + @1 step_one + @2 step_two "); } @@ -94,8 +95,8 @@ fn rewind_target_updates_metadata_and_resume_hint() { exit_code: 0 ----- stdout ----- ----- stderr ----- - Rewound metadata branch to @1 (start) - Warning: checkpoint @1 has no git_commit_sha; run branch not moved + Rewound metadata branch to @1 (step_one) + Rewound run branch fabro/run/[ULID] to [SHA] To resume: fabro resume [RUN_PREFIX] "); diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 029929ef0..823f83723 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -68,7 +68,7 @@ fn dry_run_simple() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [DRY_RUN_DIR] + Run: [STORAGE_DIR]/runs/20260403-[ULID] === Output === [Simulated] Response for stage: report @@ -966,11 +966,11 @@ fn detach_creates_run_dir_with_detach_log() { "detach_log_exists": run_dir.join("detach.log").exists(), }), @r#" - { - "run_dir": "[DRY_RUN_DIR]", - "launcher_log_exists": true, - "detach_log_exists": false - } - "# + { + "run_dir": "[RUN_DIR]", + "launcher_log_exists": true, + "detach_log_exists": false + } + "# ); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs index 2e4d4fa83..8e40b398f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs @@ -48,9 +48,9 @@ fn system_df_summarizes_runs_and_logs() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (0%) - Logs 1 - [SIZE] [SIZE] (100%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (100%) + Logs 1 - [SIZE] [SIZE] (100%) Data directory: [STORAGE_DIR] ----- stderr ----- @@ -79,13 +79,13 @@ fn system_df_verbose_lists_runs_with_reclaimable_marker() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (0%) - Logs 0 - [SIZE] [SIZE] (0%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (100%) + Logs 0 - [SIZE] [SIZE] (0%) Data directory: [STORAGE_DIR] - RUN ID WORKFLOW STATUS AGE SIZE + RUN ID WORKFLOW STATUS AGE SIZE [RUN_PREFIX] Simple succeeded [AGE] [SIZE] * * = reclaimable diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs b/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs index 2d30a3f8a..d25ac6cd0 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs @@ -51,7 +51,7 @@ fn system_prune_dry_run_lists_matching_runs_without_deleting() { success: true exit_code: 0 ----- stdout ----- - would delete: [DATE]-dry-run-[ULID] (Simple) + would delete: 20260403-[ULID] (Simple) ----- stderr ----- 1 run(s) would be deleted ([SIZE] freed). Pass --yes to confirm. diff --git a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs index 2b02a85e0..a928d6a47 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs @@ -175,11 +175,11 @@ fn dry_run_detach_attach_works_with_default_run_lookup() { "conclusion_json_exists": run_dir.join("conclusion.json").exists(), }), @r#" - { - "run_dir": "[DRY_RUN_DIR]", - "conclusion_json_exists": false - } - "# + { + "run_dir": "[RUN_DIR]", + "conclusion_json_exists": false + } + "# ); } diff --git a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs index 98b0ac3e1..a7b65be28 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/dry_run_examples.rs @@ -30,7 +30,7 @@ fn dry_run_branching() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [DRY_RUN_DIR] + Run: [STORAGE_DIR]/runs/20260403-[ULID] === Output === [Simulated] Response for stage: validate @@ -62,7 +62,7 @@ fn dry_run_conditions() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [DRY_RUN_DIR] + Run: [STORAGE_DIR]/runs/20260403-[ULID] === Output === [Simulated] Response for stage: path_b @@ -95,7 +95,7 @@ fn dry_run_parallel() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [DRY_RUN_DIR] + Run: [STORAGE_DIR]/runs/20260403-[ULID] === Output === [Simulated] Response for stage: review @@ -128,7 +128,7 @@ fn dry_run_styled() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [DRY_RUN_DIR] + Run: [STORAGE_DIR]/runs/20260403-[ULID] === Output === [Simulated] Response for stage: critical_review @@ -159,6 +159,6 @@ fn dry_run_legacy_tool() { Run: [ULID] Status: SUCCESS Duration: [DURATION] - Run: [DRY_RUN_DIR] + Run: [STORAGE_DIR]/runs/20260403-[ULID] "); } diff --git a/lib/crates/fabro-store/Cargo.toml b/lib/crates/fabro-store/Cargo.toml index 4a854aa87..f9f49d032 100644 --- a/lib/crates/fabro-store/Cargo.toml +++ b/lib/crates/fabro-store/Cargo.toml @@ -29,3 +29,4 @@ futures.workspace = true [dev-dependencies] tokio = { workspace = true, features = ["test-util", "macros"] } tempfile = "3" +ulid.workspace = true diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 7161fe2d0..50a06698b 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -14,12 +14,12 @@ pub use fabro_types::StageId; pub use run_state::{NodeState, RunProjection}; pub use runtime::RuntimeState; pub use slate::{NodeAsset, SlateRunStore, SlateStore}; -pub use types::{EventEnvelope, EventPayload, RunSummary}; +pub use types::{EventEnvelope, EventPayload, NodeVisitRef, RunSummary}; pub type StoreHandle = Arc; #[derive(Debug, Default, Clone, PartialEq, Eq)] -pub(crate) struct ListRunsQuery { +pub struct ListRunsQuery { pub start: Option>, pub end: Option>, } diff --git a/lib/crates/fabro-store/src/slate/catalog.rs b/lib/crates/fabro-store/src/slate/catalog.rs index 95fafc38f..8a7e0d070 100644 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ b/lib/crates/fabro-store/src/slate/catalog.rs @@ -1,5 +1,7 @@ +use std::collections::HashSet; use std::sync::Arc; +use bytes::Bytes; use futures::TryStreamExt; use object_store::ObjectStore; use object_store::path::Path; @@ -13,13 +15,10 @@ pub(crate) async fn write_catalog( run_id: &RunId, ) -> Result<()> { store - .put(&by_id_path(base_prefix, run_id), bytes::Bytes::new().into()) + .put(&by_id_path(base_prefix, run_id), Bytes::new().into()) .await?; store - .put( - &by_start_path(base_prefix, run_id), - bytes::Bytes::new().into(), - ) + .put(&by_start_path(base_prefix, run_id), Bytes::new().into()) .await?; Ok(()) } @@ -44,11 +43,12 @@ pub(crate) async fn list_run_ids( let prefix = Path::from(format!("{base_prefix}by-start")); let metas = store.list(Some(&prefix)).try_collect::>().await?; let mut run_ids = Vec::new(); + let mut seen = HashSet::new(); for meta in metas { let Some(run_id) = parse_run_id_from_path(&meta.location) else { continue; }; - if !read_locator(store.clone(), base_prefix, &run_id).await? { + if !seen.insert(run_id) { continue; } let created_at = run_id.created_at(); @@ -67,6 +67,12 @@ pub(crate) async fn list_run_ids( Ok(run_ids) } +pub(crate) fn parse_run_id_from_path(path: &Path) -> Option { + let filename = path.filename()?; + let run_id = filename.strip_suffix(".json").unwrap_or(filename); + run_id.parse().ok() +} + pub(crate) fn db_prefix(base_prefix: &str, run_id: &RunId) -> String { format!( "{base_prefix}db/{}/{run_id}/", @@ -85,16 +91,9 @@ pub(crate) fn by_start_path(base_prefix: &str, run_id: &RunId) -> Path { )) } -pub(crate) fn parse_run_id_from_path(path: &Path) -> Option { - let filename = path.filename()?; - let run_id = filename.strip_suffix(".json").unwrap_or(filename); - run_id.parse().ok() -} - #[cfg(test)] pub(super) mod test_support { use super::*; - use std::collections::HashSet; pub(crate) async fn repair_catalog( store: Arc, @@ -107,17 +106,15 @@ pub(super) mod test_support { .list(Some(&by_id_prefix)) .try_collect::>() .await?; - let mut canonical = HashSet::new(); - for meta in by_id_metas { - if let Some(run_id) = parse_run_id_from_path(&meta.location) { - canonical.insert(run_id); - } - } + let run_ids = by_id_metas + .iter() + .filter_map(|meta| parse_run_id_from_path(&meta.location)) + .collect::>(); - for run_id in &canonical { + for run_id in &run_ids { let path = by_start_path(base_prefix, run_id); if !object_exists(store.clone(), &path).await? { - store.put(&path, bytes::Bytes::new().into()).await?; + store.put(&path, Bytes::new().into()).await?; } } @@ -125,6 +122,7 @@ pub(super) mod test_support { .list(Some(&by_start_prefix)) .try_collect::>() .await?; + let canonical = run_ids.into_iter().collect::>(); let mut seen = HashSet::new(); for meta in by_start_metas { let location = meta.location.clone(); @@ -132,25 +130,18 @@ pub(super) mod test_support { delete_if_exists(store.clone(), &location).await?; continue; }; - if !canonical.contains(&run_id) { - delete_if_exists(store.clone(), &location).await?; + let expected = by_start_path(base_prefix, &run_id); + if canonical.contains(&run_id) && expected == location { + seen.insert(run_id); continue; } - let expected = by_start_path(base_prefix, &run_id); - if expected == location { - seen.insert(run_id); - } else { - delete_if_exists(store.clone(), &location).await?; - } + delete_if_exists(store.clone(), &location).await?; } - for run_id in &canonical { - if !seen.contains(run_id) { + for run_id in canonical { + if !seen.contains(&run_id) { store - .put( - &by_start_path(base_prefix, run_id), - bytes::Bytes::new().into(), - ) + .put(&by_start_path(base_prefix, &run_id), Bytes::new().into()) .await?; } } diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 33356c073..d6b06b693 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -5,7 +5,6 @@ use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; -use chrono::{DateTime, Utc}; use futures::TryStreamExt; use object_store::ObjectStore; use object_store::path::Path; @@ -19,12 +18,6 @@ use fabro_types::RunId; use run_store::SlateRunStoreInner; pub use run_store::{NodeAsset, SlateRunStore}; -#[derive(Clone, Copy)] -enum RunOpenMode { - Writer, - Reader, -} - #[derive(Clone)] pub struct SlateStore { object_store: Arc, @@ -110,54 +103,62 @@ impl SlateStore { weak.upgrade().map(SlateRunStore::from_inner) } - async fn open_existing_run( + async fn open_run_store( &self, - run_id: RunId, - mode: RunOpenMode, + run_id: &RunId, + db_prefix: &str, ) -> Result> { - if matches!(mode, RunOpenMode::Writer) { - if let Some(active) = self.get_active_run(&run_id).await { + if let Some(active) = self.get_active_run(run_id).await { + if active.matches_run(run_id, db_prefix) { return Ok(Some(active)); } + return Err(StoreError::Other(format!( + "active run cache mismatch for run_id {run_id:?}" + ))); } - let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id); - if !self.db_prefix_has_objects(&db_prefix).await? { + if !self.db_prefix_has_objects(db_prefix).await? { return Ok(None); } - match mode { - RunOpenMode::Writer => { - let db = self.open_db(&db_prefix).await?; - let has_init = match SlateRunStore::validate_init(&db, &run_id).await { - Ok(has_init) => has_init, - Err(err) => { - let _ = db.close().await; - return Err(err); - } - }; - if !has_init { - let _ = db.close().await; - return Ok(None); - } - let run_store = SlateRunStore::open_writer(run_id, db).await?; - self.cache_active_run(&run_store).await; - Ok(Some(run_store)) - } - RunOpenMode::Reader => { - let reader = self.open_reader(&db_prefix).await?; - let has_init = match SlateRunStore::validate_init(&reader, &run_id).await { - Ok(has_init) => has_init, - Err(err) => { - let _ = reader.close().await; - return Err(err); - } - }; - if !has_init { - let _ = reader.close().await; - return Ok(None); - } - SlateRunStore::open_reader(run_id, reader).await.map(Some) + let db = self.open_db(db_prefix).await?; + let has_init = match SlateRunStore::validate_init(&db, run_id).await { + Ok(has_init) => has_init, + Err(err) => { + let _ = db.close().await; + return Err(err); } + }; + if !has_init { + let _ = db.close().await; + return Ok(None); } + let run_store = SlateRunStore::open_writer(*run_id, db_prefix.to_string(), db).await?; + self.cache_active_run(&run_store).await; + Ok(Some(run_store)) + } + + async fn open_run_reader_store( + &self, + run_id: &RunId, + db_prefix: &str, + ) -> Result> { + if !self.db_prefix_has_objects(db_prefix).await? { + return Ok(None); + } + let reader = self.open_reader(db_prefix).await?; + let has_init = match SlateRunStore::validate_init(&reader, run_id).await { + Ok(has_init) => has_init, + Err(err) => { + let _ = reader.close().await; + return Err(err); + } + }; + if !has_init { + let _ = reader.close().await; + return Ok(None); + } + SlateRunStore::open_reader(*run_id, db_prefix.to_string(), reader) + .await + .map(Some) } async fn delete_db_prefix(&self, db_prefix: &str) -> Result<()> { @@ -176,75 +177,77 @@ impl SlateStore { impl SlateStore { pub async fn create_run(&self, run_id: &RunId) -> Result { - let locator = + let locator_exists = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; + let db_prefix = catalog::db_prefix(&self.base_prefix, run_id); + if let Some(active) = self.get_active_run(run_id).await { - if locator { + if locator_exists && !active.matches_run(run_id, &db_prefix) { return Err(StoreError::RunAlreadyExists(run_id.to_string())); } catalog::write_catalog(self.object_store.clone(), &self.base_prefix, run_id).await?; return Ok(active); } - let db_prefix = catalog::db_prefix(&self.base_prefix, run_id); + if locator_exists && self.db_prefix_has_objects(&db_prefix).await? { + return Err(StoreError::RunAlreadyExists(run_id.to_string())); + } let db = self.open_db(&db_prefix).await?; SlateRunStore::validate_init(&db, run_id).await?; db.put(keys::init(), serde_json::to_vec(run_id)?).await?; - let run_store = SlateRunStore::open_writer(*run_id, db).await?; + let run_store = SlateRunStore::open_writer(*run_id, db_prefix.clone(), db).await?; self.cache_active_run(&run_store).await; catalog::write_catalog(self.object_store.clone(), &self.base_prefix, run_id).await?; Ok(run_store) } pub async fn open_run(&self, run_id: &RunId) -> Result { - let locator = + let exists = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; - if !locator { + if !exists { return Err(StoreError::RunNotFound(run_id.to_string())); } + let db_prefix = catalog::db_prefix(&self.base_prefix, run_id); let run_store = self - .open_existing_run(*run_id, RunOpenMode::Writer) + .open_run_store(run_id, &db_prefix) .await? .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; Ok(run_store) } pub async fn open_run_reader(&self, run_id: &RunId) -> Result { - let locator = + let exists = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; - if !locator { + if !exists { return Err(StoreError::RunNotFound(run_id.to_string())); } + let db_prefix = catalog::db_prefix(&self.base_prefix, run_id); let run_store = self - .open_existing_run(*run_id, RunOpenMode::Reader) + .open_run_reader_store(run_id, &db_prefix) .await? .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; Ok(run_store) } - pub async fn list_runs(&self) -> Result> { - self.list_runs_in_range(None, None).await - } - - pub async fn list_runs_in_range( - &self, - start: Option>, - end: Option>, - ) -> Result> { - let query = ListRunsQuery { start, end }; + pub async fn list_runs(&self, query: &ListRunsQuery) -> Result> { let run_ids = - catalog::list_run_ids(self.object_store.clone(), &self.base_prefix, &query).await?; + catalog::list_run_ids(self.object_store.clone(), &self.base_prefix, query).await?; let mut summaries = Vec::new(); for run_id in run_ids { + let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id); if let Some(active) = self.get_active_run(&run_id).await { + if !active.matches_run(&run_id, &db_prefix) { + return Err(StoreError::Other(format!( + "active run cache mismatch for run_id {run_id:?}" + ))); + } let snapshot = active.snapshot().await?; summaries.push(SlateRunStore::build_summary(snapshot.as_ref(), &run_id).await?); continue; } - let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id); if !self.db_prefix_has_objects(&db_prefix).await? { continue; } @@ -268,64 +271,52 @@ impl SlateStore { active.close().await?; } + let db_prefix = catalog::db_prefix(&self.base_prefix, run_id); + if catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? { - return self.delete_run_record(run_id).await; + delete_path( + self.object_store.clone(), + &catalog::by_start_path(&self.base_prefix, run_id), + ) + .await?; + self.delete_db_prefix(&db_prefix).await?; + delete_path( + self.object_store.clone(), + &catalog::by_id_path(&self.base_prefix, run_id), + ) + .await?; + return Ok(()); } - self.repair_delete_run(run_id).await - } -} - -impl SlateStore { - async fn delete_run_record(&self, run_id: &RunId) -> Result<()> { - delete_path( - self.object_store.clone(), - &catalog::by_start_path(&self.base_prefix, run_id), - ) - .await?; - self.delete_db_prefix(&catalog::db_prefix(&self.base_prefix, run_id)) + if active.is_some() { + delete_path( + self.object_store.clone(), + &catalog::by_start_path(&self.base_prefix, run_id), + ) .await?; - delete_path( - self.object_store.clone(), - &catalog::by_id_path(&self.base_prefix, run_id), - ) - .await?; - Ok(()) - } + self.delete_db_prefix(&db_prefix).await?; + delete_path( + self.object_store.clone(), + &catalog::by_id_path(&self.base_prefix, run_id), + ) + .await?; + return Ok(()); + } - async fn repair_delete_run(&self, run_id: &RunId) -> Result<()> { let by_start_prefix = Path::from(format!("{}by-start", self.base_prefix)); - let db_prefix = Path::from(format!("{}db", self.base_prefix)); - let expected_name = format!("{run_id}.json"); - - for meta in self + let metas = self .object_store .list(Some(&by_start_prefix)) .try_collect::>() - .await? - { - if meta.location.filename() == Some(expected_name.as_str()) { - delete_path(self.object_store.clone(), &meta.location).await?; + .await?; + let expected_name = format!("{run_id}.json"); + for meta in metas { + if meta.location.filename() != Some(expected_name.as_str()) { + continue; } + delete_path(self.object_store.clone(), &meta.location).await?; } - - let db_run_segment = format!("/{run_id}/"); - for meta in self - .object_store - .list(Some(&db_prefix)) - .try_collect::>() - .await? - { - if meta.location.to_string().contains(&db_run_segment) { - delete_path(self.object_store.clone(), &meta.location).await?; - } - } - - delete_path( - self.object_store.clone(), - &catalog::by_id_path(&self.base_prefix, run_id), - ) - .await?; + self.delete_db_prefix(&db_prefix).await?; Ok(()) } } @@ -357,11 +348,11 @@ mod tests { use std::time::Duration; use bytes::Bytes; - use chrono::Duration as ChronoDuration; + use chrono::{DateTime, Duration as ChronoDuration, Utc}; use fabro_types::{ AttrValue, Checkpoint, Conclusion, Graph, PullRequestRecord, Retro, RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, Settings, StageStatus, StartRecord, - StatusReason, fixtures, + StatusReason, }; use object_store::memory::InMemory; use slatedb::config::Settings as SlateSettings; @@ -370,7 +361,7 @@ mod tests { use crate::{EventPayload, StageId}; - #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] + #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] struct CatalogRecord { run_id: RunId, created_at: DateTime, @@ -395,16 +386,17 @@ mod tests { } fn test_run_id(label: &str) -> RunId { - match label { - "run-1" => fixtures::RUN_1, - "other-run" => fixtures::RUN_2, - "run-early" => fixtures::RUN_2, - "run-late" => fixtures::RUN_3, + let (timestamp_ms, random) = match label { + "run-1" => (dt("2026-03-27T12:00:00Z").timestamp_millis() as u64, 1), + "other-run" => (dt("2026-03-27T12:00:00Z").timestamp_millis() as u64, 2), + "run-early" => (dt("2026-03-27T10:00:00Z").timestamp_millis() as u64, 3), + "run-late" => (dt("2026-03-27T12:00:00Z").timestamp_millis() as u64, 4), _ => panic!("unknown test run id: {label}"), - } + }; + RunId::from(ulid::Ulid::from_parts(timestamp_ms, random)) } - fn sample_run_record(run_id: &str, created_at: DateTime) -> RunRecord { + fn sample_run_record(run_id: &str, _created_at: DateTime) -> RunRecord { let mut graph = Graph::new("night-sky"); graph.attrs.insert( "goal".to_string(), @@ -558,11 +550,6 @@ mod tests { store.head(path).await.is_ok() } - async fn read_json_value(store: Arc, path: &Path) -> serde_json::Value { - let result = store.get(path).await.unwrap(); - serde_json::from_slice(&result.bytes().await.unwrap()).unwrap() - } - async fn seed_db( object_store: Arc, record: &CatalogRecord, @@ -640,14 +627,8 @@ mod tests { let by_start = catalog::by_start_path("runs/", &test_run_id("run-1")); assert!(object_exists(object_store.clone(), &by_id).await); assert!(object_exists(object_store.clone(), &by_start).await); - assert_eq!( - read_json_value(object_store.clone(), &by_start).await, - serde_json::json!({ - "run_id": test_run_id("run-1").to_string(), - }) - ); - let summary = store.list_runs().await.unwrap(); + let summary = store.list_runs(&ListRunsQuery::default()).await.unwrap(); assert_eq!(summary.len(), 1); assert_eq!(summary[0].run_id, test_run_id("run-1")); assert_eq!(summary[0].workflow_name, Some("night-sky".to_string())); @@ -710,10 +691,16 @@ mod tests { .unwrap(); assert!(store.open_run(&test_run_id("run-1")).await.is_ok()); - assert!(store.list_runs().await.unwrap().is_empty()); + assert!( + store + .list_runs(&ListRunsQuery::default()) + .await + .unwrap() + .is_empty() + ); repair_catalog_for_tests(&store).await.unwrap(); - let listed = store.list_runs().await.unwrap(); + let listed = store.list_runs(&ListRunsQuery::default()).await.unwrap(); assert_eq!(listed.len(), 1); assert!( object_exists( @@ -724,65 +711,6 @@ mod tests { ); } - #[tokio::test] - async fn list_runs_uses_by_start_index_to_resolve_canonical_by_id_record() { - let (object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); - let record = CatalogRecord { - run_id: test_run_id("run-1"), - created_at, - db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")), - run_dir: None, - }; - - let db = seed_db(object_store.clone(), &record, true).await; - db.put( - keys::event_key(1, created_at.timestamp_millis()), - serde_json::to_vec(&event_payload( - "run-1", - "2026-03-27T12:00:00Z", - "run.created", - None, - serde_json::json!({ - "settings": sample_run_record("run-1", created_at).settings, - "graph": sample_run_record("run-1", created_at).graph, - "workflow_slug": sample_run_record("run-1", created_at).workflow_slug, - "working_directory": sample_run_record("run-1", created_at).working_directory, - "host_repo_path": sample_run_record("run-1", created_at).host_repo_path, - "base_branch": sample_run_record("run-1", created_at).base_branch, - "labels": sample_run_record("run-1", created_at).labels, - }), - )) - .unwrap(), - ) - .await - .unwrap(); - db.close().await.unwrap(); - - object_store - .put( - &catalog::by_id_path("runs/", &test_run_id("run-1")), - serde_json::to_vec(&record).unwrap().into(), - ) - .await - .unwrap(); - object_store - .put( - &catalog::by_start_path("runs/", &test_run_id("run-1")), - serde_json::json!({ - "run_id": test_run_id("run-1").to_string(), - }) - .to_string() - .into(), - ) - .await - .unwrap(); - - let listed = store.list_runs().await.unwrap(); - assert_eq!(listed.len(), 1); - assert_eq!(listed[0].run_id, record.run_id); - } - #[tokio::test] async fn reopen_recovers_event_sequences() { let (_object_store, store) = make_store(); @@ -869,16 +797,29 @@ mod tests { .unwrap(); assert!(store.open_run(&test_run_id("run-1")).await.is_err()); - assert!(store.list_runs().await.unwrap().is_empty()); + assert!( + store + .list_runs(&ListRunsQuery::default()) + .await + .unwrap() + .is_empty() + ); } #[tokio::test] async fn create_run_allows_idempotent_retry_and_rejects_conflict() { let (_object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); - store.create_run(&test_run_id("run-1")).await.unwrap(); + let _created_at = dt("2026-03-27T12:00:00Z"); + // Hold the first run store so the active cache Weak ref stays alive. + let _run = store.create_run(&test_run_id("run-1")).await.unwrap(); store.create_run(&test_run_id("run-1")).await.unwrap(); + let conflict = store.create_run(&test_run_id("other-run")).await; + // Different run_id should work fine + assert!(conflict.is_ok()); + + // Dropping and re-creating with locator already written should reject + drop(_run); let conflict = store.create_run(&test_run_id("run-1")).await; assert!(matches!(conflict, Err(StoreError::RunAlreadyExists(_)))); } @@ -907,7 +848,7 @@ mod tests { .await .unwrap(); - let listed = store.list_runs().await.unwrap(); + let listed = store.list_runs(&ListRunsQuery::default()).await.unwrap(); assert_eq!(listed.len(), 1); let reopened = store.open_run(&test_run_id("run-1")).await.unwrap(); @@ -938,7 +879,7 @@ mod tests { #[tokio::test] async fn watch_events_from_polls_new_events() { let (_object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); + let _created_at = dt("2026-03-27T12:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let mut stream = run.watch_events_from(1).unwrap(); @@ -1012,7 +953,7 @@ mod tests { #[tokio::test] async fn delete_run_closes_active_handles() { let (object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); + let _created_at = dt("2026-03-27T12:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); run.put_artifact_value("summary", &serde_json::json!({"done": true})) .await @@ -1035,20 +976,20 @@ mod tests { #[tokio::test] async fn repair_catalog_removes_stale_wrong_time_prefixes() { let (object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); - let wrong_time = dt("2026-03-27T11:00:00Z"); + let _created_at = dt("2026-03-27T12:00:00Z"); + let _wrong_time = dt("2026-03-27T11:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); run.put_artifact_value("summary", &serde_json::json!({"done": true})) .await .unwrap(); - let locator = catalog::read_locator(object_store.clone(), "runs/", &test_run_id("run-1")) + let _locator = catalog::read_locator(object_store.clone(), "runs/", &test_run_id("run-1")) .await .unwrap(); object_store .put( &catalog::by_start_path("runs/", &test_run_id("run-1")), - bytes::Bytes::new().into(), + Bytes::new().into(), ) .await .unwrap(); @@ -1063,6 +1004,7 @@ mod tests { async fn create_run_uses_distinct_db_prefix_for_same_minute_orphan() { let (object_store, store) = make_store(); let old_created_at = dt("2026-03-27T12:00:00Z"); + let _new_created_at = dt("2026-03-27T12:00:30Z"); let orphan = CatalogRecord { run_id: test_run_id("run-1"), created_at: old_created_at, @@ -1087,7 +1029,7 @@ mod tests { #[tokio::test] async fn create_run_rejects_mismatched_init_for_existing_prefix() { let (object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); + let _created_at = dt("2026-03-27T12:00:00Z"); let db_prefix = catalog::db_prefix("runs/", &test_run_id("run-1")); let db = slatedb::Db::builder(db_prefix.clone(), object_store) .with_settings(SlateSettings { @@ -1097,15 +1039,12 @@ mod tests { .build() .await .unwrap(); - let mismatched = CatalogRecord { - run_id: test_run_id("other-run"), - created_at, - db_prefix, - run_dir: None, - }; - db.put(keys::init(), serde_json::to_vec(&mismatched).unwrap()) - .await - .unwrap(); + db.put( + keys::init(), + serde_json::to_vec(&test_run_id("other-run")).unwrap(), + ) + .await + .unwrap(); db.close().await.unwrap(); let Err(err) = store.create_run(&test_run_id("run-1")).await else { @@ -1120,7 +1059,7 @@ mod tests { #[tokio::test] async fn slate_run_store_round_trips_assets_and_projects_events() { let (_object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); + let _created_at = dt("2026-03-27T12:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let node = StageId::new("code", 2); run.append_event(&event_payload( @@ -1148,7 +1087,7 @@ mod tests { #[tokio::test] async fn slate_run_store_lists_artifact_values_and_assets() { let (_object_store, store) = make_store(); - let created_at = dt("2026-03-27T12:00:00Z"); + let _created_at = dt("2026-03-27T12:00:00Z"); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); run.put_artifact_value("summary", &serde_json::json!({"done": true})) .await @@ -1434,6 +1373,10 @@ mod tests { let state = run.state().await.unwrap(); let stored_run = state.run.as_ref().unwrap(); assert_eq!(stored_run.run_id, run_record.run_id); + assert_eq!( + stored_run.run_id.created_at(), + run_record.run_id.created_at() + ); assert_eq!(stored_run.workflow_slug, run_record.workflow_slug); assert_eq!(stored_run.graph.name, run_record.graph.name); assert_eq!(state.graph_source.as_deref(), Some("digraph night_sky {}")); @@ -1845,15 +1788,12 @@ mod tests { let (_object_store, store) = make_store(); let run = store.create_run(&test_run_id("run-1")).await.unwrap(); - let invalid_missing_err = serde_json::from_value::(serde_json::json!({ + let invalid_missing: EventPayload = serde_json::from_value(serde_json::json!({ "run_id": "run-1" })) - .unwrap_err(); - assert!( - invalid_missing_err - .to_string() - .contains("missing or non-string required field: id") - ); + .unwrap(); + let err = run.append_event(&invalid_missing).await.unwrap_err(); + assert!(matches!(err, StoreError::InvalidEvent(_))); let invalid_run_id: EventPayload = serde_json::from_value(serde_json::json!({ "id": "evt-invalid-run", @@ -1981,7 +1921,7 @@ mod tests { .await .unwrap(); - let all = store.list_runs().await.unwrap(); + let all = store.list_runs(&ListRunsQuery::default()).await.unwrap(); assert_eq!(all.len(), 2); assert_eq!(all[0].run_id, test_run_id("run-late")); assert_eq!(all[0].workflow_name, Some("night-sky".to_string())); @@ -1996,10 +1936,10 @@ mod tests { assert_eq!(all[1].status, None); let filtered = store - .list_runs_in_range( - Some(dt("2026-03-27T11:00:00Z")), - Some(dt("2026-03-27T13:00:00Z")), - ) + .list_runs(&ListRunsQuery { + start: Some(dt("2026-03-27T11:00:00Z")), + end: Some(dt("2026-03-27T13:00:00Z")), + }) .await .unwrap(); assert_eq!(filtered.len(), 1); diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index a261954a1..74df34b16 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -3,7 +3,7 @@ use std::sync::{Arc, Weak}; use std::time::Duration; use bytes::Bytes; -use chrono::{DateTime, Utc}; +use chrono::Utc; use futures::Stream; use serde::Serialize; use serde::de::DeserializeOwned; @@ -16,10 +16,6 @@ use crate::keys; use crate::run_state::EventProjectionCache; use crate::{EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId, StoreError}; use fabro_types::RunId; -#[derive(Clone)] -pub struct SlateRunStore { - inner: Arc, -} #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] pub struct NodeAsset { @@ -27,16 +23,23 @@ pub struct NodeAsset { pub filename: String, } +#[derive(Clone)] +pub struct SlateRunStore { + inner: Arc, +} + impl std::fmt::Debug for SlateRunStore { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("SlateRunStore") .field("run_id", &self.inner.run_id) + .field("db_prefix", &self.inner.db_prefix) .finish_non_exhaustive() } } pub(crate) struct SlateRunStoreInner { run_id: RunId, + db_prefix: String, db: SlateRunDb, event_seq: AtomicU32, close_lock: Mutex<()>, @@ -49,11 +52,16 @@ enum SlateRunDb { } impl SlateRunStore { - pub(crate) async fn open_writer(run_id: RunId, db: slatedb::Db) -> Result { + pub(crate) async fn open_writer( + run_id: RunId, + db_prefix: String, + db: slatedb::Db, + ) -> Result { let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?; Ok(Self { inner: Arc::new(SlateRunStoreInner { run_id, + db_prefix, db: SlateRunDb::Writer(db), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -62,11 +70,16 @@ impl SlateRunStore { }) } - pub(crate) async fn open_reader(run_id: RunId, db: DbReader) -> Result { + pub(crate) async fn open_reader( + run_id: RunId, + db_prefix: String, + db: DbReader, + ) -> Result { let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?; Ok(Self { inner: Arc::new(SlateRunStoreInner { run_id, + db_prefix, db: SlateRunDb::Reader(Box::new(db)), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -83,12 +96,12 @@ impl SlateRunStore { Arc::downgrade(&self.inner) } - pub fn run_id(&self) -> RunId { + pub(crate) fn run_id(&self) -> RunId { self.inner.run_id } - pub fn created_at(&self) -> DateTime { - self.inner.run_id.created_at() + pub(crate) fn matches_run(&self, run_id: &RunId, db_prefix: &str) -> bool { + self.inner.run_id == *run_id && self.inner.db_prefix == db_prefix } pub(crate) async fn close(&self) -> Result<()> { @@ -114,7 +127,7 @@ impl SlateRunStore { match get_json::(db, keys::init()).await? { Some(existing) if existing == *expected => Ok(true), Some(existing) => Err(StoreError::Other(format!( - "existing _init.json {existing:?} does not match requested run id {expected:?}" + "existing _init.json {existing:?} does not match requested run_id {expected:?}" ))), None => Ok(false), } @@ -146,13 +159,7 @@ impl SlateRunStore { impl SlateRunStore { pub async fn append_event(&self, payload: &EventPayload) -> Result { - if payload.run_id() != self.inner.run_id.to_string() { - return Err(StoreError::InvalidEvent(format!( - "payload run_id {:?} does not match store run_id {:?}", - payload.run_id(), - self.inner.run_id - ))); - } + payload.validate(&self.inner.run_id)?; let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst); self.inner .db diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index 780ebe7c8..bcdbc4e84 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -1,12 +1,17 @@ use std::collections::HashMap; use chrono::{DateTime, Utc}; -use serde::de::Error as _; -use serde::{Deserialize, Deserializer, Serialize}; +use serde::{Deserialize, Serialize}; use crate::{Result, StoreError}; use fabro_types::{RunId, RunStatus, StatusReason}; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct NodeVisitRef<'a> { + pub node_id: &'a str, + pub visit: u32, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct RunSummary { pub run_id: RunId, @@ -22,7 +27,7 @@ pub struct RunSummary { pub total_cost: Option, } -#[derive(Debug, Clone, PartialEq, Serialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(transparent)] pub struct EventPayload(serde_json::Value); @@ -34,17 +39,6 @@ impl EventPayload { } pub(crate) fn validate(&self, expected_run_id: &RunId) -> Result<()> { - self.validate_shape()?; - if self.run_id() == expected_run_id.to_string() { - return Ok(()); - } - Err(StoreError::InvalidEvent(format!( - "payload run_id {:?} does not match store run_id {expected_run_id:?}", - self.run_id() - ))) - } - - fn validate_shape(&self) -> Result<()> { let obj = self.0.as_object().ok_or_else(|| { StoreError::InvalidEvent("event payload must be a JSON object".into()) })?; @@ -59,14 +53,22 @@ impl EventPayload { } } } - Ok(()) + + match obj.get("run_id") { + Some(serde_json::Value::String(run_id)) if run_id == &expected_run_id.to_string() => { + Ok(()) + } + Some(serde_json::Value::String(run_id)) => Err(StoreError::InvalidEvent(format!( + "payload run_id {run_id:?} does not match store run_id {expected_run_id:?}" + ))), + _ => Err(StoreError::InvalidEvent( + "missing or non-string required field: run_id".into(), + )), + } } - #[must_use] - pub fn run_id(&self) -> &str { - self.0["run_id"] - .as_str() - .expect("EventPayload::run_id called before validation") + pub fn into_inner(self) -> serde_json::Value { + self.0 } pub fn as_value(&self) -> &serde_json::Value { @@ -74,17 +76,6 @@ impl EventPayload { } } -impl<'de> Deserialize<'de> for EventPayload { - fn deserialize(deserializer: D) -> std::result::Result - where - D: Deserializer<'de>, - { - let payload = Self(serde_json::Value::deserialize(deserializer)?); - payload.validate_shape().map_err(D::Error::custom)?; - Ok(payload) - } -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct EventEnvelope { pub seq: u32, diff --git a/lib/crates/fabro-types/src/run_id.rs b/lib/crates/fabro-types/src/run_id.rs index e0f4661b1..fa1fafa87 100644 --- a/lib/crates/fabro-types/src/run_id.rs +++ b/lib/crates/fabro-types/src/run_id.rs @@ -167,7 +167,7 @@ pub mod fixtures { #[cfg(test)] mod tests { - use chrono::{DateTime, Utc}; + use chrono::{TimeZone, Utc}; use super::{RunId, fixtures}; @@ -186,21 +186,18 @@ mod tests { } #[test] - fn created_at_comes_from_ulid_timestamp() { - let expected = DateTime::parse_from_rfc3339("2026-03-27T12:00:00Z") - .unwrap() - .with_timezone(&Utc); - let run_id = RunId::from_datetime(expected); + fn exposes_created_at_from_ulid_timestamp() { + let dt = Utc.with_ymd_and_hms(2026, 3, 27, 12, 34, 56).unwrap(); + let run_id = RunId::from_datetime(dt); - assert_eq!(run_id.created_at(), expected); + assert_eq!(run_id.created_at(), dt); } #[test] - fn from_datetime_round_trips_timestamp() { - let expected = DateTime::parse_from_rfc3339("2026-03-27T12:00:00Z") - .unwrap() - .with_timezone(&Utc); + fn creates_run_id_from_datetime() { + let dt = Utc.with_ymd_and_hms(2026, 3, 27, 12, 34, 56).unwrap(); + let run_id = RunId::from_datetime(dt); - assert_eq!(RunId::from_datetime(expected).created_at(), expected); + assert_eq!(run_id.created_at(), dt); } } diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 49e9fc960..029a69c4b 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -352,13 +352,14 @@ pub(crate) fn default_run_dir(run_id: &RunId) -> PathBuf { } pub fn make_run_dir(runs_base: &Path, run_id: &RunId) -> PathBuf { - let local_created_at = run_id.created_at().with_timezone(&Local); - runs_base.join(format!("{}-{}", local_created_at.format("%Y%m%d"), run_id)) + let local_dt = run_id.created_at().with_timezone(&Local); + runs_base.join(format!("{}-{run_id}", local_dt.format("%Y%m%d"))) } #[cfg(test)] mod tests { use super::*; + use chrono::{TimeZone, Utc}; use fabro_graphviz::graph::AttrValue; use fabro_store::{SlateStore, StoreHandle}; use fabro_types::fixtures; @@ -426,6 +427,24 @@ mod tests { assert_eq!(prompt, "Goal: Fix bugs"); } + #[test] + fn make_run_dir_uses_run_id_timestamp_in_local_time() { + let runs_base = Path::new("/tmp/runs"); + let run_id = RunId::from(ulid::Ulid::from_datetime( + Utc.with_ymd_and_hms(2026, 3, 27, 12, 0, 0).unwrap().into(), + )); + let expected_date = run_id + .created_at() + .with_timezone(&Local) + .format("%Y%m%d") + .to_string(); + + assert_eq!( + make_run_dir(runs_base, &run_id), + runs_base.join(format!("{expected_date}-{run_id}")) + ); + } + #[test] fn validate_applies_stylesheet() { let dot = r#"digraph Test { @@ -674,6 +693,7 @@ mod tests { run_store.state().await.unwrap().status.unwrap().status, crate::run_status::RunStatus::Submitted ); + assert_eq!(created.run_dir, default_run_dir(&fixtures::RUN_1)); assert!(!created.run_dir.join("id.txt").exists()); } @@ -724,7 +744,6 @@ mod tests { async fn create_hydrates_run_created_event_into_store() { let dir = tempfile::tempdir().unwrap(); let storage_dir = dir.path().join("storage"); - let run_dir = dir.path().join("run"); std::fs::create_dir_all(storage_dir.join("store")).unwrap(); let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).unwrap()); diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index 79708835c..fa21cc5d2 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -182,7 +182,10 @@ pub async fn find_run_id_by_prefix_or_store( let current_repo_root = canonical_repo_root(repo)?; let mut matches = Vec::new(); - for summary in fabro_store.list_runs().await? { + for summary in fabro_store + .list_runs(&fabro_store::ListRunsQuery::default()) + .await? + { if summary.run_id.to_string() == prefix { if summary.host_repo_path.is_none() { return Ok(summary.run_id); diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 6980f2559..62f31a634 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -1092,6 +1092,7 @@ mod tests { async fn resume_errors_when_run_already_finished_successfully() { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1)); let registry = Arc::new(test_registry()); diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index 018a638a5..d5ea8befa 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -7,7 +7,6 @@ use std::sync::atomic::{AtomicBool, AtomicU32, Ordering}; use std::time::Duration; use async_trait::async_trait; -use chrono::Utc; use fabro_agent::Sandbox; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_hooks::HookSettings; diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 704d881bd..d5757c757 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -320,7 +320,6 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use chrono::Utc; use fabro_graphviz::graph::Graph; use fabro_store::SlateStore; use fabro_types::{RunId, Settings, fixtures}; diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 08b37c785..dd51131d7 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -669,7 +669,6 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use chrono::Utc; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_interview::AutoApproveInterviewer; use fabro_sandbox::SandboxSpec; diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs index 5b100e4f6..1245ed52e 100644 --- a/lib/crates/fabro-workflow/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs @@ -53,7 +53,6 @@ mod tests { use std::collections::HashMap; use std::path::PathBuf; - use chrono::Utc; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_store::{SlateRunStore, SlateStore, StoreHandle}; use fabro_types::{Settings, fixtures}; diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index e11813a40..976944c59 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -207,12 +207,12 @@ fn parse_dot_summary(dot: &str) -> (String, usize, usize) { fn read_plan_text(state: &RunProjection) -> Option { let mut plan_nodes = state .iter_nodes() - .filter_map(|(node, node_state)| { - let node_id = node.node_id(); - let visit = node.visit(); - node_id - .starts_with("plan") - .then_some((node_id, visit, node_state.response.as_deref())) + .filter_map(|(stage_id, node)| { + stage_id.node_id().starts_with("plan").then_some(( + stage_id.node_id(), + stage_id.visit(), + node.response.as_deref(), + )) }) .collect::>(); plan_nodes.sort_by(|left, right| left.0.cmp(right.0).then(left.1.cmp(&right.1))); @@ -1057,7 +1057,6 @@ mod tests { async fn build_pr_body_uses_in_memory_conclusion() { install_mock_llm(); - let tmp = tempfile::tempdir().unwrap(); let store = test_store(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let conclusion = make_test_conclusion(); @@ -1081,9 +1080,7 @@ mod tests { async fn build_pr_body_uses_store_records_without_legacy_files() { install_mock_llm(); - let tmp = tempfile::tempdir().unwrap(); let store = test_store(); - let created_at = Utc::now(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let run_record = RunRecord { @@ -1106,7 +1103,7 @@ mod tests { workflow_source: Some("digraph test { plan -> code }".to_string()), workflow_config: None, labels: run_record.labels.clone().into_iter().collect(), - run_dir: tmp.path().display().to_string(), + run_dir: run_record.working_directory.display().to_string(), working_directory: run_record.working_directory.display().to_string(), host_repo_path: run_record.host_repo_path.clone(), base_branch: run_record.base_branch.clone(), @@ -1149,9 +1146,7 @@ mod tests { async fn build_pr_body_uses_plan_text_from_store_without_response_md() { install_mock_llm(); - let tmp = tempfile::tempdir().unwrap(); let store = test_store(); - let created_at = Utc::now(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let run_record = RunRecord { @@ -1174,7 +1169,7 @@ mod tests { workflow_source: Some("digraph test { plan -> code }".to_string()), workflow_config: None, labels: run_record.labels.clone().into_iter().collect(), - run_dir: tmp.path().display().to_string(), + run_dir: run_record.working_directory.display().to_string(), working_directory: run_record.working_directory.display().to_string(), host_repo_path: run_record.host_repo_path.clone(), base_branch: run_record.base_branch.clone(), @@ -1344,7 +1339,6 @@ mod tests { #[tokio::test] async fn empty_diff_returns_none() { - let tmp = tempfile::tempdir().unwrap(); let store = test_store(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let creds = GitHubAppCredentials { @@ -1373,7 +1367,6 @@ mod tests { async fn load_pull_request_diff_uses_store_without_disk_patch() { let tmp = tempfile::tempdir().unwrap(); let store = test_store(); - let created_at = Utc::now(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let run_record = RunRecord { run_id: fixtures::RUN_1, @@ -1395,7 +1388,7 @@ mod tests { workflow_source: None, workflow_config: None, labels: run_record.labels.clone().into_iter().collect(), - run_dir: tmp.path().display().to_string(), + run_dir: run_record.working_directory.display().to_string(), working_directory: tmp.path().display().to_string(), host_repo_path: None, base_branch: None, diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 1fea14749..c7a7c1bef 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -168,7 +168,6 @@ mod tests { use std::sync::{Arc, Mutex}; use std::time::Duration; - use chrono::Utc; use fabro_graphviz::graph::Graph; use fabro_store::SlateStore; use fabro_types::{RunId, Settings, fixtures}; @@ -216,7 +215,6 @@ mod tests { run_dir: &std::path::Path, checkpoint: &Checkpoint, ) -> fabro_store::SlateRunStore { - let created_at = Utc::now(); let inner = test_store().create_run(&test_run_id()).await.unwrap(); let run_store = inner; let run_record = RunRecord { diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index 1499bd3d6..2c59aa436 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -193,9 +193,12 @@ fn scan_orphan_runs(base: &Path) -> Result> { pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result> { let mut runs_by_id: HashMap = HashMap::new(); - if let Ok(store_runs) = store.list_runs().await { + if let Ok(store_runs) = store + .list_runs(&fabro_store::ListRunsQuery::default()) + .await + { for summary in store_runs { - let Some(run_info) = run_info_from_summary(summary, base) else { + let Some(run_info) = run_info_from_summary(&summary, base) else { continue; }; runs_by_id.insert(run_info.run_id(), run_info); @@ -218,7 +221,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result Option { +fn run_info_from_summary(summary: &RunSummary, runs_base: &Path) -> Option { let path = make_run_dir(runs_base, &summary.run_id); if !path.exists() { return None; @@ -234,7 +237,7 @@ fn run_info_from_summary(summary: RunSummary, runs_base: &Path) -> Option String { fn parse_run_id(value: &str) -> Option { let value = value.trim(); - (!value.is_empty()).then_some(value)?.parse().ok() + if value.is_empty() { + return None; + } + // Try direct ULID parse first, then try extracting ULID after date prefix (YYYYMMDD-ULID). + value.parse().ok().or_else(|| { + value + .split_once('-') + .and_then(|(_, ulid)| ulid.parse().ok()) + }) } fn run_id_matches(run_id: RunId, prefix: &str) -> bool { @@ -359,7 +370,6 @@ mod tests { use std::sync::Arc; use std::time::Duration; - use chrono::Utc; use fabro_graphviz::graph::Graph; use fabro_store::{SlateStore, StoreHandle}; use fabro_types::{RunStatus, Settings, fixtures}; @@ -367,6 +377,7 @@ mod tests { use super::scan_runs_combined; use crate::event::{WorkflowRunEvent, append_workflow_event}; + use crate::operations::make_run_dir; use crate::records::RunRecord; fn memory_store() -> StoreHandle { @@ -393,12 +404,11 @@ mod tests { #[tokio::test] async fn scan_runs_combined_uses_store_status_without_status_json() { let temp = tempfile::tempdir().unwrap(); - let run_dir = temp.path().join(fixtures::RUN_1.to_string()); + let run_dir = make_run_dir(temp.path(), &fixtures::RUN_1); std::fs::create_dir_all(&run_dir).unwrap(); std::fs::write(run_dir.join("id.txt"), format!("{}\n", fixtures::RUN_1)).unwrap(); let store = memory_store(); - let run_dir_string = run_dir.to_string_lossy().to_string(); let run_record = sample_run_record(); let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); append_workflow_event( @@ -411,7 +421,7 @@ mod tests { workflow_source: None, workflow_config: None, labels: run_record.labels.clone().into_iter().collect(), - run_dir: run_dir_string.clone(), + run_dir: run_dir.display().to_string(), working_directory: run_record.working_directory.display().to_string(), host_repo_path: run_record.host_repo_path.clone(), base_branch: run_record.base_branch.clone(),