From ebf5a0621d64f7b3b9c3ae64a1e733b8a0787dda Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 3 Apr 2026 19:15:39 -0700 Subject: [PATCH] refactor: derive durable run paths from run IDs Move durable run metadata and path derivation onto RunId, simplify the Slate catalog/index format, and carry the storage-specific run directory through workflow creation so detached and lookup flows stay aligned. Also update affected CLI snapshots and test helpers to match the new run discovery behavior. --- lib/crates/fabro-checkpoint/src/metadata.rs | 1 - .../fabro-cli/src/commands/run/create.rs | 1 - .../fabro-cli/src/commands/store/dump.rs | 7 +- lib/crates/fabro-cli/tests/it/cmd/asset_cp.rs | 2 +- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 14 +- lib/crates/fabro-server/src/server.rs | 6 +- lib/crates/fabro-store/src/lib.rs | 2 +- lib/crates/fabro-store/src/run_state.rs | 7 +- lib/crates/fabro-store/src/slate/catalog.rs | 143 +++----- lib/crates/fabro-store/src/slate/mod.rs | 312 ++++++------------ lib/crates/fabro-store/src/slate/run_store.rs | 38 +-- lib/crates/fabro-store/src/types.rs | 11 +- lib/crates/fabro-types/src/run.rs | 2 - lib/crates/fabro-types/src/run_id.rs | 31 ++ lib/crates/fabro-workflow/src/event.rs | 5 +- lib/crates/fabro-workflow/src/git.rs | 5 +- .../fabro-workflow/src/handler/agent.rs | 5 +- .../fabro-workflow/src/handler/command.rs | 5 +- .../src/handler/manager_loop.rs | 7 +- lib/crates/fabro-workflow/src/handler/mod.rs | 2 +- .../fabro-workflow/src/handler/parallel.rs | 5 +- .../fabro-workflow/src/handler/prompt.rs | 5 +- .../fabro-workflow/src/operations/create.rs | 51 +-- .../fabro-workflow/src/operations/fork.rs | 4 +- .../fabro-workflow/src/operations/mod.rs | 2 +- .../src/operations/rebuild_meta.rs | 20 +- .../fabro-workflow/src/operations/start.rs | 1 - .../src/pipeline/execute/tests.rs | 6 +- .../fabro-workflow/src/pipeline/finalize.rs | 9 +- .../fabro-workflow/src/pipeline/initialize.rs | 19 +- .../fabro-workflow/src/pipeline/persist.rs | 18 +- .../src/pipeline/pull_request.rs | 48 +-- .../fabro-workflow/src/pipeline/retro.rs | 10 +- lib/crates/fabro-workflow/src/run_lookup.rs | 28 +- lib/crates/fabro-workflow/src/test_support.rs | 8 +- 35 files changed, 274 insertions(+), 566 deletions(-) diff --git a/lib/crates/fabro-checkpoint/src/metadata.rs b/lib/crates/fabro-checkpoint/src/metadata.rs index a17913f69..c4abfa939 100644 --- a/lib/crates/fabro-checkpoint/src/metadata.rs +++ b/lib/crates/fabro-checkpoint/src/metadata.rs @@ -206,7 +206,6 @@ mod tests { fn test_run_record(run_id: fabro_types::RunId) -> RunRecord { RunRecord { run_id, - created_at: Utc.with_ymd_and_hms(2025, 1, 1, 0, 0, 0).single().unwrap(), settings: Settings::default(), graph: Graph::new("test"), workflow_slug: None, diff --git a/lib/crates/fabro-cli/src/commands/run/create.rs b/lib/crates/fabro-cli/src/commands/run/create.rs index 9d9b6cc60..e76961e3b 100644 --- a/lib/crates/fabro-cli/src/commands/run/create.rs +++ b/lib/crates/fabro-cli/src/commands/run/create.rs @@ -46,7 +46,6 @@ pub(crate) async fn create_run( settings, cwd, workflow_slug: None, - run_dir: None, run_id, base_branch: None, host_repo_path: None, diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index c72fcf765..ef96487fc 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -164,7 +164,7 @@ mod tests { )) } - fn sample_run_record(run_id: RunId, created_at: DateTime) -> RunRecord { + fn sample_run_record(run_id: RunId, _created_at: DateTime) -> RunRecord { let mut graph = Graph::new("night-sky"); graph.attrs.insert( "goal".to_string(), @@ -172,7 +172,6 @@ mod tests { ); RunRecord { run_id, - created_at, settings: Settings::default(), graph, workflow_slug: Some("night-sky".to_string()), @@ -284,7 +283,7 @@ mod tests { let store = test_store(); let created_at = dt("2026-03-27T12:00:00Z"); let run_id = test_run_id(); - let run = store.create_run(&run_id, created_at, None).await.unwrap(); + let run = store.create_run(&run_id).await.unwrap(); let run_record = sample_run_record(run_id, created_at); let start_record = sample_start_record(run_id, created_at); let status_record = sample_status(); @@ -632,7 +631,7 @@ mod tests { let store = test_store(); let created_at = dt("2026-03-27T12:00:00Z"); let run_id = test_run_id(); - let run = store.create_run(&run_id, created_at, None).await.unwrap(); + let run = store.create_run(&run_id).await.unwrap(); let run_record = sample_run_record(run_id, created_at); append_workflow_event( &run, diff --git a/lib/crates/fabro-cli/tests/it/cmd/asset_cp.rs b/lib/crates/fabro-cli/tests/it/cmd/asset_cp.rs index 6b942bcf4..1ae8d09ae 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/asset_cp.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/asset_cp.rs @@ -117,7 +117,7 @@ fn asset_cp_tree_preserves_structure() { success: true exit_code: 0 ----- stdout ----- - Copied 8 asset(s) to [TEMP_DIR]/asset-tree + Copied 6 asset(s) to [TEMP_DIR]/asset-tree ----- stderr ----- "); insta::assert_snapshot!( diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 504b68593..44355cd80 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -463,6 +463,13 @@ fn attach_json_errors_without_prompting_for_human_input() { "run_id": "[ULID]", "ts": "[TIMESTAMP]" }, + { + "event": "run.running", + "id": "[EVENT_ID]", + "properties": {}, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" + }, { "event": "sandbox.initialized", "id": "[EVENT_ID]", @@ -483,13 +490,6 @@ fn attach_json_errors_without_prompting_for_human_input() { "run_id": "[ULID]", "ts": "[TIMESTAMP]" }, - { - "event": "run.running", - "id": "[EVENT_ID]", - "properties": {}, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, { "event": "stage.started", "id": "[EVENT_ID]", diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 4fecd54b1..e7ce5b48f 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -537,7 +537,6 @@ async fn start_run( ) -> Response { let run_id = RunId::new(); info!(run_id = %run_id, "Run queued"); - let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4())); let settings = state.settings.read().unwrap().clone(); let created = match Box::pin(operations::create( state.store.as_ref(), @@ -549,7 +548,6 @@ async fn start_run( settings, cwd: std::env::current_dir().unwrap_or_else(|_| std::env::temp_dir()), workflow_slug: None, - run_dir: Some(run_dir.clone()), run_id: Some(run_id), host_repo_path: None, base_branch: None, @@ -581,8 +579,8 @@ async fn start_run( .into_response(); } }; - let persisted = created.persisted; - let created_at = persisted.run_record().created_at; + let created_at = run_id.created_at(); + let run_dir = created.run_dir; { let mut runs = state.runs.lock().expect("runs lock poisoned"); diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index df10a6034..7161fe2d0 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -14,7 +14,7 @@ pub use fabro_types::StageId; pub use run_state::{NodeState, RunProjection}; pub use runtime::RuntimeState; pub use slate::{NodeAsset, SlateRunStore, SlateStore}; -pub use types::{CatalogRecord, EventEnvelope, EventPayload, RunSummary}; +pub use types::{EventEnvelope, EventPayload, RunSummary}; pub type StoreHandle = Arc; diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 27ecc9511..5c8d86f49 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -6,7 +6,7 @@ use chrono::{DateTime, Utc}; use serde::de::DeserializeOwned; use serde_json::Value; -use crate::{CatalogRecord, EventEnvelope, Result, RunSummary, StageId, StoreError}; +use crate::{EventEnvelope, Result, RunSummary, StageId, StoreError}; use fabro_types::{ Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro, RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StageUsage, @@ -86,7 +86,6 @@ impl RunProjection { .collect::>(); self.run = Some(RunRecord { run_id, - created_at: ts, settings, graph, workflow_slug: optional_string(&properties, "workflow_slug"), @@ -281,7 +280,7 @@ impl RunProjection { visits } - pub(crate) fn build_summary(&self, catalog: &CatalogRecord) -> RunSummary { + pub(crate) fn build_summary(&self, run_id: &RunId) -> RunSummary { let workflow_name = self.run.as_ref().map(|run| { if run.graph.name.is_empty() { "unnamed".to_string() @@ -294,7 +293,7 @@ impl RunProjection { (!goal.is_empty()).then(|| goal.to_string()) }); RunSummary { - catalog: catalog.clone(), + run_id: *run_id, workflow_name, workflow_slug: self.run.as_ref().and_then(|run| run.workflow_slug.clone()), goal, diff --git a/lib/crates/fabro-store/src/slate/catalog.rs b/lib/crates/fabro-store/src/slate/catalog.rs index 4d873512a..95fafc38f 100644 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ b/lib/crates/fabro-store/src/slate/catalog.rs @@ -1,88 +1,76 @@ use std::sync::Arc; -use chrono::{DateTime, Utc}; use futures::TryStreamExt; use object_store::ObjectStore; use object_store::path::Path; -use serde::{Deserialize, Serialize}; -use crate::{CatalogRecord, ListRunsQuery, Result}; +use crate::{ListRunsQuery, Result}; use fabro_types::RunId; -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -struct StartIndexRecord { - run_id: RunId, -} - pub(crate) async fn write_catalog( store: Arc, base_prefix: &str, run_id: &RunId, - created_at: DateTime, - db_prefix: &str, - run_dir: Option<&str>, -) -> Result { - let record = CatalogRecord { - run_id: *run_id, - created_at, - db_prefix: db_prefix.to_string(), - run_dir: run_dir.map(ToOwned::to_owned), - }; - let bytes = serde_json::to_vec(&record)?; +) -> Result<()> { store - .put(&by_id_path(base_prefix, run_id), bytes.clone().into()) + .put(&by_id_path(base_prefix, run_id), bytes::Bytes::new().into()) .await?; store .put( - &by_start_path(base_prefix, created_at, run_id), - serde_json::to_vec(&StartIndexRecord { run_id: *run_id })?.into(), + &by_start_path(base_prefix, run_id), + bytes::Bytes::new().into(), ) .await?; - Ok(record) + Ok(()) } pub(crate) async fn read_locator( store: Arc, base_prefix: &str, run_id: &RunId, -) -> Result> { - read_catalog_path(store, by_id_path(base_prefix, run_id)).await +) -> Result { + match store.head(&by_id_path(base_prefix, run_id)).await { + Ok(_) => Ok(true), + Err(object_store::Error::NotFound { .. }) => Ok(false), + Err(err) => Err(err.into()), + } } -pub(crate) async fn list_catalogs( +pub(crate) async fn list_run_ids( store: Arc, base_prefix: &str, query: &ListRunsQuery, -) -> Result> { +) -> Result> { let prefix = Path::from(format!("{base_prefix}by-start")); let metas = store.list(Some(&prefix)).try_collect::>().await?; - let mut records = Vec::new(); + let mut run_ids = Vec::new(); for meta in metas { - let Some(index) = read_start_index_path(store.clone(), meta.location).await? else { + let Some(run_id) = parse_run_id_from_path(&meta.location) else { continue; }; - let Some(record) = read_locator(store.clone(), base_prefix, &index.run_id).await? else { + if !read_locator(store.clone(), base_prefix, &run_id).await? { continue; - }; + } + let created_at = run_id.created_at(); if let Some(start) = query.start { - if record.created_at < start { + if created_at < start { continue; } } if let Some(end) = query.end { - if record.created_at > end { + if created_at > end { continue; } } - records.push(record); + run_ids.push(run_id); } - Ok(records) + Ok(run_ids) } -pub(crate) fn db_prefix(base_prefix: &str, created_at: DateTime, run_id: &RunId) -> String { +pub(crate) fn db_prefix(base_prefix: &str, run_id: &RunId) -> String { format!( "{base_prefix}db/{}/{run_id}/", - created_at.format("%Y-%m-%d-%H-%M-%S-%3f") + run_id.created_at().format("%Y-%m-%d-%H-%M-%S-%3f") ) } @@ -90,44 +78,23 @@ pub(crate) fn by_id_path(base_prefix: &str, run_id: &RunId) -> Path { Path::from(format!("{base_prefix}by-id/{run_id}.json")) } -pub(crate) fn by_start_path(base_prefix: &str, created_at: DateTime, run_id: &RunId) -> Path { +pub(crate) fn by_start_path(base_prefix: &str, run_id: &RunId) -> Path { Path::from(format!( "{base_prefix}by-start/{}/{run_id}.json", - created_at.format("%Y-%m-%d-%H-%M") + run_id.created_at().format("%Y-%m-%d-%H-%M") )) } -pub(crate) async fn read_catalog_path( - store: Arc, - path: Path, -) -> Result> { - match store.get(&path).await { - Ok(result) => Ok(Some(serde_json::from_slice(&result.bytes().await?)?)), - Err(object_store::Error::NotFound { .. }) => Ok(None), - Err(err) => Err(err.into()), - } -} - -#[cfg(test)] -fn by_start_record(run_id: &RunId) -> StartIndexRecord { - StartIndexRecord { run_id: *run_id } -} - -async fn read_start_index_path( - store: Arc, - path: Path, -) -> Result> { - match store.get(&path).await { - Ok(result) => Ok(Some(serde_json::from_slice(&result.bytes().await?)?)), - Err(object_store::Error::NotFound { .. }) => Ok(None), - Err(err) => Err(err.into()), - } +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::{HashMap, HashSet}; + use std::collections::HashSet; pub(crate) async fn repair_catalog( store: Arc, @@ -140,22 +107,17 @@ pub(super) mod test_support { .list(Some(&by_id_prefix)) .try_collect::>() .await?; - let mut canonical = HashMap::new(); + let mut canonical = HashSet::new(); for meta in by_id_metas { - if let Some(record) = read_catalog_path(store.clone(), meta.location).await? { - canonical.insert(record.run_id, record); + if let Some(run_id) = parse_run_id_from_path(&meta.location) { + canonical.insert(run_id); } } - for record in canonical.values() { - let path = by_start_path(base_prefix, record.created_at, &record.run_id); + for run_id in &canonical { + let path = by_start_path(base_prefix, run_id); if !object_exists(store.clone(), &path).await? { - store - .put( - &path, - serde_json::to_vec(&by_start_record(&record.run_id))?.into(), - ) - .await?; + store.put(&path, bytes::Bytes::new().into()).await?; } } @@ -166,29 +128,28 @@ pub(super) mod test_support { let mut seen = HashSet::new(); for meta in by_start_metas { let location = meta.location.clone(); - let Some(index) = read_start_index_path(store.clone(), location.clone()).await? else { + let Some(run_id) = parse_run_id_from_path(&location) else { delete_if_exists(store.clone(), &location).await?; continue; }; - let expected = canonical.get(&index.run_id).map(|canonical_record| { - by_start_path(base_prefix, canonical_record.created_at, &index.run_id) - }); - match expected { - Some(expected) if expected == location => { - seen.insert(index.run_id); - } - _ => { - delete_if_exists(store.clone(), &location).await?; - } + if !canonical.contains(&run_id) { + delete_if_exists(store.clone(), &location).await?; + 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?; } } - for record in canonical.values() { - if !seen.contains(&record.run_id) { + for run_id in &canonical { + if !seen.contains(run_id) { store .put( - &by_start_path(base_prefix, record.created_at, &record.run_id), - serde_json::to_vec(&by_start_record(&record.run_id))?.into(), + &by_start_path(base_prefix, run_id), + bytes::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 bfa840751..33356c073 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -14,7 +14,7 @@ use slatedb::config::{DbReaderOptions, Settings}; use tokio::sync::Mutex; use crate::keys; -use crate::{CatalogRecord, ListRunsQuery, Result, RunSummary, StoreError}; +use crate::{ListRunsQuery, Result, RunSummary, StoreError}; use fabro_types::RunId; use run_store::SlateRunStoreInner; pub use run_store::{NodeAsset, SlateRunStore}; @@ -102,7 +102,7 @@ impl SlateStore { self.active_runs .lock() .await - .insert(run_store.record().run_id, run_store.downgrade()); + .insert(run_store.run_id(), run_store.downgrade()); } async fn remove_active_run(&self, run_id: &RunId) -> Option { @@ -112,27 +112,22 @@ impl SlateStore { async fn open_existing_run( &self, - record: &CatalogRecord, + run_id: RunId, mode: RunOpenMode, ) -> Result> { if matches!(mode, RunOpenMode::Writer) { - if let Some(active) = self.get_active_run(&record.run_id).await { - if active.record() == *record { - return Ok(Some(active)); - } - return Err(StoreError::Other(format!( - "active run cache mismatch for run_id {:?}", - record.run_id - ))); + if let Some(active) = self.get_active_run(&run_id).await { + return Ok(Some(active)); } } - if !self.db_prefix_has_objects(&record.db_prefix).await? { + let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id); + if !self.db_prefix_has_objects(&db_prefix).await? { return Ok(None); } match mode { RunOpenMode::Writer => { - let db = self.open_db(&record.db_prefix).await?; - let has_init = match SlateRunStore::validate_init(&db, record).await { + 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; @@ -143,13 +138,13 @@ impl SlateStore { let _ = db.close().await; return Ok(None); } - let run_store = SlateRunStore::open_writer(record.clone(), db).await?; + 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(&record.db_prefix).await?; - let has_init = match SlateRunStore::validate_init(&reader, record).await { + 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; @@ -160,9 +155,7 @@ impl SlateStore { let _ = reader.close().await; return Ok(None); } - SlateRunStore::open_reader(record.clone(), reader) - .await - .map(Some) + SlateRunStore::open_reader(run_id, reader).await.map(Some) } } } @@ -182,86 +175,51 @@ impl SlateStore { } impl SlateStore { - pub async fn create_run( - &self, - run_id: &RunId, - created_at: DateTime, - run_dir: Option<&str>, - ) -> Result { + pub async fn create_run(&self, run_id: &RunId) -> Result { let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; if let Some(active) = self.get_active_run(run_id).await { - if active.created_at() != created_at - || locator - .as_ref() - .is_some_and(|existing| existing.created_at != created_at) - { + if locator { return Err(StoreError::RunAlreadyExists(run_id.to_string())); } - let record = active.record(); - catalog::write_catalog( - self.object_store.clone(), - &self.base_prefix, - run_id, - created_at, - &record.db_prefix, - run_dir, - ) - .await?; + catalog::write_catalog(self.object_store.clone(), &self.base_prefix, run_id).await?; return Ok(active); } - let db_prefix = match locator { - Some(existing) if existing.created_at != created_at => { - return Err(StoreError::RunAlreadyExists(run_id.to_string())); - } - Some(existing) => existing.db_prefix, - None => catalog::db_prefix(&self.base_prefix, created_at, run_id), - }; - - let record = CatalogRecord { - run_id: *run_id, - created_at, - db_prefix: db_prefix.clone(), - run_dir: run_dir.map(ToOwned::to_owned), - }; + let db_prefix = catalog::db_prefix(&self.base_prefix, run_id); let db = self.open_db(&db_prefix).await?; - SlateRunStore::validate_init(&db, &record).await?; - db.put(keys::init(), serde_json::to_vec(&record)?).await?; - let run_store = SlateRunStore::open_writer(record.clone(), db).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?; self.cache_active_run(&run_store).await; - catalog::write_catalog( - self.object_store.clone(), - &self.base_prefix, - run_id, - created_at, - &db_prefix, - run_dir, - ) - .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 = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id) - .await? - .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; + let locator = + catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; + if !locator { + return Err(StoreError::RunNotFound(run_id.to_string())); + } let run_store = self - .open_existing_run(&locator, RunOpenMode::Writer) + .open_existing_run(*run_id, RunOpenMode::Writer) .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 = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id) - .await? - .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; + let locator = + catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; + if !locator { + return Err(StoreError::RunNotFound(run_id.to_string())); + } let run_store = self - .open_existing_run(&locator, RunOpenMode::Reader) + .open_existing_run(*run_id, RunOpenMode::Reader) .await? .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; Ok(run_store) @@ -277,51 +235,41 @@ impl SlateStore { end: Option>, ) -> Result> { let query = ListRunsQuery { start, end }; - let catalogs = - catalog::list_catalogs(self.object_store.clone(), &self.base_prefix, &query).await?; + let run_ids = + catalog::list_run_ids(self.object_store.clone(), &self.base_prefix, &query).await?; let mut summaries = Vec::new(); - for record in catalogs { - if let Some(active) = self.get_active_run(&record.run_id).await { - if active.record() != record { - return Err(StoreError::Other(format!( - "active run cache mismatch for run_id {:?}", - record.run_id - ))); - } + for run_id in run_ids { + if let Some(active) = self.get_active_run(&run_id).await { let snapshot = active.snapshot().await?; - summaries.push(SlateRunStore::build_summary(snapshot.as_ref(), &record).await?); + summaries.push(SlateRunStore::build_summary(snapshot.as_ref(), &run_id).await?); continue; } - if !self.db_prefix_has_objects(&record.db_prefix).await? { + let db_prefix = catalog::db_prefix(&self.base_prefix, &run_id); + if !self.db_prefix_has_objects(&db_prefix).await? { continue; } - let reader = self.open_reader(&record.db_prefix).await?; - if !SlateRunStore::validate_init(&reader, &record).await? { + let reader = self.open_reader(&db_prefix).await?; + if !SlateRunStore::validate_init(&reader, &run_id).await? { let _ = reader.close().await; continue; } - let summary = SlateRunStore::build_summary(&reader, &record).await; + let summary = SlateRunStore::build_summary(&reader, &run_id).await; let _ = reader.close().await; let summary = summary?; summaries.push(summary); } - summaries.sort_by(|a, b| b.catalog.created_at.cmp(&a.catalog.created_at)); + summaries.sort_by(|a, b| b.run_id.created_at().cmp(&a.run_id.created_at())); Ok(summaries) } pub async fn delete_run(&self, run_id: &RunId) -> Result<()> { let active = self.remove_active_run(run_id).await; - let active_record = active.as_ref().map(SlateRunStore::record); if let Some(active) = &active { active.close().await?; } - if let Some(record) = - catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id) - .await? - .or(active_record) - { - return self.delete_run_record(&record).await; + if catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? { + return self.delete_run_record(run_id).await; } self.repair_delete_run(run_id).await @@ -329,16 +277,17 @@ impl SlateStore { } impl SlateStore { - async fn delete_run_record(&self, record: &CatalogRecord) -> Result<()> { + async fn delete_run_record(&self, run_id: &RunId) -> Result<()> { delete_path( self.object_store.clone(), - &catalog::by_start_path(&self.base_prefix, record.created_at, &record.run_id), + &catalog::by_start_path(&self.base_prefix, run_id), ) .await?; - self.delete_db_prefix(&record.db_prefix).await?; + self.delete_db_prefix(&catalog::db_prefix(&self.base_prefix, run_id)) + .await?; delete_path( self.object_store.clone(), - &catalog::by_id_path(&self.base_prefix, &record.run_id), + &catalog::by_id_path(&self.base_prefix, run_id), ) .await?; Ok(()) @@ -421,6 +370,14 @@ mod tests { use crate::{EventPayload, StageId}; + #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)] + struct CatalogRecord { + run_id: RunId, + created_at: DateTime, + db_prefix: String, + run_dir: Option, + } + fn dt(rfc3339: &str) -> DateTime { DateTime::parse_from_rfc3339(rfc3339) .unwrap() @@ -455,7 +412,6 @@ mod tests { ); RunRecord { run_id: test_run_id(run_id), - created_at, settings: Settings::default(), graph, workflow_slug: Some("night-sky".to_string()), @@ -621,7 +577,7 @@ mod tests { .await .unwrap(); if include_init { - db.put(keys::init(), serde_json::to_vec(record).unwrap()) + db.put(keys::init(), serde_json::to_vec(&record.run_id).unwrap()) .await .unwrap(); } @@ -632,10 +588,7 @@ mod tests { async fn create_open_list_and_delete_full_lifecycle() { let (object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let run_record = sample_run_record("run-1", created_at); run.append_event(&event_payload( @@ -684,7 +637,7 @@ mod tests { .unwrap(); let by_id = catalog::by_id_path("runs/", &test_run_id("run-1")); - let by_start = catalog::by_start_path("runs/", created_at, &test_run_id("run-1")); + 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!( @@ -696,7 +649,7 @@ mod tests { let summary = store.list_runs().await.unwrap(); assert_eq!(summary.len(), 1); - assert_eq!(summary[0].catalog.run_id, test_run_id("run-1")); + assert_eq!(summary[0].run_id, test_run_id("run-1")); assert_eq!(summary[0].workflow_name, Some("night-sky".to_string())); assert_eq!(summary[0].goal, Some("map the constellations".to_string())); assert_eq!(summary[0].status, Some(RunStatus::Succeeded)); @@ -720,7 +673,7 @@ mod tests { let record = CatalogRecord { run_id: test_run_id("run-1"), created_at, - db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")), + db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")), run_dir: None, }; @@ -765,7 +718,7 @@ mod tests { assert!( object_exists( object_store, - &catalog::by_start_path("runs/", created_at, &test_run_id("run-1")) + &catalog::by_start_path("runs/", &test_run_id("run-1")) ) .await ); @@ -778,7 +731,7 @@ mod tests { let record = CatalogRecord { run_id: test_run_id("run-1"), created_at, - db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")), + db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")), run_dir: None, }; @@ -815,7 +768,7 @@ mod tests { .unwrap(); object_store .put( - &catalog::by_start_path("runs/", created_at, &test_run_id("run-1")), + &catalog::by_start_path("runs/", &test_run_id("run-1")), serde_json::json!({ "run_id": test_run_id("run-1").to_string(), }) @@ -827,17 +780,14 @@ mod tests { let listed = store.list_runs().await.unwrap(); assert_eq!(listed.len(), 1); - assert_eq!(listed[0].catalog, record); + assert_eq!(listed[0].run_id, record.run_id); } #[tokio::test] async fn reopen_recovers_event_sequences() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let run_record = sample_run_record("run-1", created_at); run.append_event(&event_payload( "run-1", @@ -897,7 +847,7 @@ mod tests { let record = CatalogRecord { run_id: test_run_id("run-1"), created_at, - db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")), + db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")), run_dir: None, }; @@ -912,7 +862,7 @@ mod tests { .unwrap(); object_store .put( - &catalog::by_start_path("runs/", created_at, &test_run_id("run-1")), + &catalog::by_start_path("runs/", &test_run_id("run-1")), serde_json::to_vec(&record).unwrap().into(), ) .await @@ -926,22 +876,10 @@ mod tests { 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"), created_at, None) - .await - .unwrap(); - store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + 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("run-1"), - created_at + chrono::Duration::seconds(1), - None, - ) - .await; + let conflict = store.create_run(&test_run_id("run-1")).await; assert!(matches!(conflict, Err(StoreError::RunAlreadyExists(_)))); } @@ -949,10 +887,7 @@ mod tests { async fn list_runs_and_open_run_reuse_active_handle_without_fencing() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let run_record = sample_run_record("run-1", created_at); run.append_event(&event_payload( "run-1", @@ -1004,10 +939,7 @@ mod tests { async fn watch_events_from_polls_new_events() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let mut stream = run.watch_events_from(1).unwrap(); run.append_event(&event_payload( @@ -1038,7 +970,7 @@ mod tests { let record = CatalogRecord { run_id: test_run_id("run-1"), created_at, - db_prefix: catalog::db_prefix("runs/", created_at, &test_run_id("run-1")), + db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")), run_dir: None, }; let db = seed_db(object_store.clone(), &record, true).await; @@ -1066,7 +998,7 @@ mod tests { db.close().await.unwrap(); object_store .put( - &catalog::by_start_path("runs/", created_at, &test_run_id("run-1")), + &catalog::by_start_path("runs/", &test_run_id("run-1")), serde_json::to_vec(&record).unwrap().into(), ) .await @@ -1081,10 +1013,7 @@ mod tests { async fn delete_run_closes_active_handles() { let (object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); run.put_artifact_value("summary", &serde_json::json!({"done": true})) .await .unwrap(); @@ -1108,22 +1037,18 @@ mod tests { 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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + 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")) .await - .unwrap() .unwrap(); object_store .put( - &catalog::by_start_path("runs/", wrong_time, &test_run_id("run-1")), - serde_json::to_vec(&locator).unwrap().into(), + &catalog::by_start_path("runs/", &test_run_id("run-1")), + bytes::Bytes::new().into(), ) .await .unwrap(); @@ -1138,38 +1063,32 @@ 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, - db_prefix: catalog::db_prefix("runs/", old_created_at, &test_run_id("run-1")), + db_prefix: catalog::db_prefix("runs/", &test_run_id("run-1")), run_dir: None, }; - let new_prefix = catalog::db_prefix("runs/", new_created_at, &test_run_id("run-1")); - assert_ne!(orphan.db_prefix, new_prefix); + let new_prefix = catalog::db_prefix("runs/", &test_run_id("run-1")); + assert_eq!(orphan.db_prefix, new_prefix); let db = seed_db(object_store.clone(), &orphan, true).await; db.close().await.unwrap(); - let run = store - .create_run(&test_run_id("run-1"), new_created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); assert!(run.state().await.unwrap().graph_source.is_none()); let locator = catalog::read_locator(object_store, "runs/", &test_run_id("run-1")) .await - .unwrap() .unwrap(); - assert_eq!(locator.created_at, new_created_at); - assert_eq!(locator.db_prefix, new_prefix); + assert!(locator); } #[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 db_prefix = catalog::db_prefix("runs/", created_at, &test_run_id("run-1")); + 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 { flush_interval: Some(Duration::from_millis(1)), @@ -1189,10 +1108,7 @@ mod tests { .unwrap(); db.close().await.unwrap(); - let Err(err) = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - else { + let Err(err) = store.create_run(&test_run_id("run-1")).await else { panic!("expected create_run to reject mismatched _init.json"); }; assert!(matches!( @@ -1205,10 +1121,7 @@ mod tests { 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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let node = StageId::new("code", 2); run.append_event(&event_payload( "run-1", @@ -1236,10 +1149,7 @@ mod tests { 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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); run.put_artifact_value("summary", &serde_json::json!({"done": true})) .await .unwrap(); @@ -1280,10 +1190,7 @@ mod tests { async fn create_run_state_and_node_storage_round_trip() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let run_record = sample_run_record("run-1", created_at); let start_record = sample_start_record("run-1", created_at); @@ -1527,7 +1434,6 @@ 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.created_at, run_record.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 {}")); @@ -1596,10 +1502,7 @@ mod tests { async fn state_projects_event_stream() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let run_record = sample_run_record("run-1", created_at); let retro = sample_retro("run-1"); @@ -1821,10 +1724,7 @@ mod tests { #[tokio::test] async fn state_rewind_keeps_active_projection_only() { let (_object_store, store) = make_store(); - let run = store - .create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); run.append_event(&event_payload( "run-1", @@ -1943,10 +1843,7 @@ mod tests { #[tokio::test] async fn append_event_validates_payload_shape_and_run_id() { let (_object_store, store) = make_store(); - let run = store - .create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let invalid_missing_err = serde_json::from_value::(serde_json::json!({ "run_id": "run-1" @@ -1972,10 +1869,7 @@ mod tests { #[tokio::test] async fn state_retains_checkpoint_history_by_event_sequence() { let (_object_store, store) = make_store(); - let run = store - .create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None) - .await - .unwrap(); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let checkpoint = sample_checkpoint(); let seq = run .append_event(&event_payload( @@ -2016,10 +1910,7 @@ mod tests { let early = dt("2026-03-27T10:00:00Z"); let late = dt("2026-03-27T12:00:00Z"); - let early_run = store - .create_run(&test_run_id("run-early"), early, None) - .await - .unwrap(); + let early_run = store.create_run(&test_run_id("run-early")).await.unwrap(); let early_record = sample_run_record("run-early", early); early_run .append_event(&event_payload( @@ -2040,10 +1931,7 @@ mod tests { .await .unwrap(); - let late_run = store - .create_run(&test_run_id("run-late"), late, None) - .await - .unwrap(); + let late_run = store.create_run(&test_run_id("run-late")).await.unwrap(); let late_record = sample_run_record("run-late", late); late_run .append_event(&event_payload( @@ -2095,7 +1983,7 @@ mod tests { let all = store.list_runs().await.unwrap(); assert_eq!(all.len(), 2); - assert_eq!(all[0].catalog.run_id, test_run_id("run-late")); + assert_eq!(all[0].run_id, test_run_id("run-late")); assert_eq!(all[0].workflow_name, Some("night-sky".to_string())); assert_eq!(all[0].goal, Some("map the constellations".to_string())); assert_eq!( @@ -2115,6 +2003,6 @@ mod tests { .await .unwrap(); assert_eq!(filtered.len(), 1); - assert_eq!(filtered[0].catalog.run_id, test_run_id("run-late")); + assert_eq!(filtered[0].run_id, test_run_id("run-late")); } } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 259cea462..a261954a1 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -14,10 +14,8 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; use crate::run_state::EventProjectionCache; -use crate::{ - CatalogRecord, EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId, - StoreError, -}; +use crate::{EventEnvelope, EventPayload, Result, RunProjection, RunSummary, StageId, StoreError}; +use fabro_types::RunId; #[derive(Clone)] pub struct SlateRunStore { inner: Arc, @@ -32,13 +30,13 @@ pub struct NodeAsset { impl std::fmt::Debug for SlateRunStore { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("SlateRunStore") - .field("record", &self.inner.record) + .field("run_id", &self.inner.run_id) .finish_non_exhaustive() } } pub(crate) struct SlateRunStoreInner { - record: CatalogRecord, + run_id: RunId, db: SlateRunDb, event_seq: AtomicU32, close_lock: Mutex<()>, @@ -51,11 +49,11 @@ enum SlateRunDb { } impl SlateRunStore { - pub(crate) async fn open_writer(record: CatalogRecord, db: slatedb::Db) -> Result { + pub(crate) async fn open_writer(run_id: RunId, 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 { - record, + run_id, db: SlateRunDb::Writer(db), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -64,11 +62,11 @@ impl SlateRunStore { }) } - pub(crate) async fn open_reader(record: CatalogRecord, db: DbReader) -> Result { + pub(crate) async fn open_reader(run_id: RunId, db: DbReader) -> Result { let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?; Ok(Self { inner: Arc::new(SlateRunStoreInner { - record, + run_id, db: SlateRunDb::Reader(Box::new(db)), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -85,12 +83,12 @@ impl SlateRunStore { Arc::downgrade(&self.inner) } - pub(crate) fn record(&self) -> CatalogRecord { - self.inner.record.clone() + pub fn run_id(&self) -> RunId { + self.inner.run_id } pub fn created_at(&self) -> DateTime { - self.inner.record.created_at + self.inner.run_id.created_at() } pub(crate) async fn close(&self) -> Result<()> { @@ -109,26 +107,26 @@ impl SlateRunStore { } } - pub(crate) async fn validate_init(db: &R, expected: &CatalogRecord) -> Result + pub(crate) async fn validate_init(db: &R, expected: &RunId) -> Result where R: DbRead + Sync, { - match get_json::(db, keys::init()).await? { + 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 catalog {expected:?}" + "existing _init.json {existing:?} does not match requested run id {expected:?}" ))), None => Ok(false), } } - pub(crate) async fn build_summary(db: &R, catalog: &CatalogRecord) -> Result + pub(crate) async fn build_summary(db: &R, run_id: &RunId) -> Result where R: DbRead + Sync, { let events = list_events_from(db, 1).await?; let state = RunProjection::apply_events(&events)?; - Ok(state.build_summary(catalog)) + Ok(state.build_summary(run_id)) } async fn projected_state(&self) -> Result { @@ -148,11 +146,11 @@ impl SlateRunStore { impl SlateRunStore { pub async fn append_event(&self, payload: &EventPayload) -> Result { - if payload.run_id() != self.inner.record.run_id.to_string() { + 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.record.run_id + self.inner.run_id ))); } let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst); diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index 3718470bd..780ebe7c8 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -7,18 +7,9 @@ use serde::{Deserialize, Deserializer, Serialize}; use crate::{Result, StoreError}; use fabro_types::{RunId, RunStatus, StatusReason}; -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -pub struct CatalogRecord { - pub run_id: RunId, - pub created_at: DateTime, - pub db_prefix: String, - pub run_dir: Option, -} - #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct RunSummary { - #[serde(flatten)] - pub catalog: CatalogRecord, + pub run_id: RunId, pub workflow_name: Option, pub workflow_slug: Option, pub goal: Option, diff --git a/lib/crates/fabro-types/src/run.rs b/lib/crates/fabro-types/src/run.rs index f0bbb90d1..7d270ca5f 100644 --- a/lib/crates/fabro-types/src/run.rs +++ b/lib/crates/fabro-types/src/run.rs @@ -1,7 +1,6 @@ use std::collections::HashMap; use std::path::PathBuf; -use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use crate::graph::Graph; @@ -11,7 +10,6 @@ use crate::settings::Settings; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RunRecord { pub run_id: RunId, - pub created_at: DateTime, pub settings: Settings, pub graph: Graph, #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/lib/crates/fabro-types/src/run_id.rs b/lib/crates/fabro-types/src/run_id.rs index 13f54a5aa..e0f4661b1 100644 --- a/lib/crates/fabro-types/src/run_id.rs +++ b/lib/crates/fabro-types/src/run_id.rs @@ -1,6 +1,7 @@ use std::fmt; use std::str::FromStr; +use chrono::{DateTime, Utc}; use serde::de::Error as _; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use ulid::Ulid; @@ -12,6 +13,15 @@ impl RunId { pub fn new() -> Self { Self(Ulid::new()) } + + pub fn created_at(&self) -> DateTime { + self.0.datetime().into() + } + + #[cfg(test)] + pub fn from_datetime(dt: DateTime) -> Self { + Self(Ulid::from_datetime(dt.into())) + } } impl Default for RunId { @@ -157,6 +167,8 @@ pub mod fixtures { #[cfg(test)] mod tests { + use chrono::{DateTime, Utc}; + use super::{RunId, fixtures}; #[test] @@ -172,4 +184,23 @@ mod tests { assert_eq!(value, fixtures::RUN_42); } + + #[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); + + assert_eq!(run_id.created_at(), expected); + } + + #[test] + fn from_datetime_round_trips_timestamp() { + let expected = DateTime::parse_from_rfc3339("2026-03-27T12:00:00Z") + .unwrap() + .with_timezone(&Utc); + + assert_eq!(RunId::from_datetime(expected).created_at(), expected); + } } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 1a6333d9d..f2686afea 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -1899,10 +1899,7 @@ mod tests { "", std::time::Duration::from_millis(1), ); - let run_store = store - .create_run(&fixtures::RUN_7, Utc::now(), None) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_7).await.unwrap(); let envelope = canonicalize_event( &fixtures::RUN_7, &WorkflowRunEvent::RunNotice { diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index c543ca966..deb90a884 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -511,10 +511,7 @@ mod tests { use fabro_store::EventPayload; let store = test_store(); - let run = store - .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) - .await - .unwrap(); + let run = store.create_run(&fixtures::RUN_1).await.unwrap(); let run_id_str = fixtures::RUN_1.to_string(); let event = |event_name: &str, props: serde_json::Value| -> EventPayload { diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 5b521abc6..b31ddfd81 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -509,10 +509,7 @@ mod tests { crate::event::StoreProgressLogger, ) { let store = test_store(); - let run_store = store - .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)), run_store: run_store.clone(), diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 1ccfbef58..a01534ea2 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -227,10 +227,7 @@ mod tests { crate::event::StoreProgressLogger, ) { let store = test_store(); - let run_store = store - .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)), run_store: run_store.clone(), diff --git a/lib/crates/fabro-workflow/src/handler/manager_loop.rs b/lib/crates/fabro-workflow/src/handler/manager_loop.rs index 6646467d6..9d5248832 100644 --- a/lib/crates/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflow/src/handler/manager_loop.rs @@ -15,7 +15,6 @@ use crate::pipeline::types::Initialized; use crate::run_dir::visit_from_context; use crate::run_options::RunOptions; use async_trait::async_trait; -use chrono::Utc; use fabro_graphviz::graph::{AttrValue, Graph, Node}; use fabro_store::SlateStore; use fabro_types::Settings; @@ -199,11 +198,7 @@ impl Handler for SubWorkflowHandler { Duration::from_millis(1), )); let run_store = store - .create_run( - &child_run_options.run_id, - Utc::now(), - Some(child_run_options.run_dir.to_string_lossy().as_ref()), - ) + .create_run(&child_run_options.run_id) .await .map_err(|err| FabroError::engine(err.to_string()))?; diff --git a/lib/crates/fabro-workflow/src/handler/mod.rs b/lib/crates/fabro-workflow/src/handler/mod.rs index 7b2bafdf9..9ca028451 100644 --- a/lib/crates/fabro-workflow/src/handler/mod.rs +++ b/lib/crates/fabro-workflow/src/handler/mod.rs @@ -95,7 +95,7 @@ impl EngineServices { .expect("test runtime should initialize") .block_on(async { store - .create_run(&fabro_types::RunId::new(), chrono::Utc::now(), None) + .create_run(&fabro_types::RunId::new()) .await .expect("slate-backed test run store should initialize") }) diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 1a1be405b..2a768356a 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -692,10 +692,7 @@ mod tests { #[tokio::test] async fn parallel_handler_stores_results_in_run_store() { let store = test_store(); - let run_store = store - .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)), run_store: run_store.clone(), diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 3b78b4a2d..5be3a1049 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -232,10 +232,7 @@ mod tests { crate::event::StoreProgressLogger, ) { let store = test_store(); - let run_store = store - .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)), run_store: run_store.clone(), diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 5701984d4..49e9fc960 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -1,4 +1,4 @@ -use chrono::{Local, Utc}; +use chrono::Local; use fabro_graphviz::graph::{AttrValue, Graph}; use fabro_model::{Catalog, Provider}; use fabro_sandbox::SandboxProvider; @@ -27,7 +27,6 @@ pub struct CreateRunInput { pub settings: Settings, pub cwd: PathBuf, pub workflow_slug: Option, - pub run_dir: Option, pub run_id: Option, pub host_repo_path: Option, pub base_branch: Option, @@ -43,8 +42,8 @@ pub struct CreatedRun { struct PersistCreateOptions { settings: Settings, - run_dir: Option, run_id: Option, + run_dir: Option, workflow_slug: Option, labels: HashMap, base_branch: Option, @@ -70,7 +69,6 @@ pub async fn create(store: &SlateStore, request: CreateRunInput) -> Result Result Result, ) -> Result<(), FabroError> { let record = persisted.run_record(); - let run_dir_string = persisted.run_dir().to_string_lossy().to_string(); - let run_store = match store - .create_run(&record.run_id, record.created_at, Some(&run_dir_string)) - .await - { + let run_store = match store.create_run(&record.run_id).await { Ok(run_store) => run_store, Err(err) => store .open_run(&record.run_id) @@ -174,7 +162,7 @@ async fn persist_created_run( workflow_slug: record.workflow_slug.clone(), db_prefix: None, }, - record.created_at, + record.run_id.created_at(), ); let payload = fabro_store::EventPayload::new( serde_json::to_value(&envelope).map_err(|err| FabroError::engine(err.to_string()))?, @@ -280,8 +268,8 @@ fn persist_validated( ) -> Result { let PersistCreateOptions { settings, - run_dir, run_id, + run_dir, workflow_slug, labels, base_branch, @@ -292,12 +280,10 @@ fn persist_validated( let settings = resolve_run_settings(settings, validated.graph()); let run_id = run_id.unwrap_or_else(RunId::new); - let run_dir = - run_dir.unwrap_or_else(|| default_run_dir(&run_id.to_string(), settings.dry_run_enabled())); + let run_dir = run_dir.unwrap_or_else(|| default_run_dir(&run_id)); let run_record = RunRecord { run_id, - created_at: Utc::now(), settings, graph: validated.graph().clone(), workflow_slug, @@ -361,20 +347,13 @@ pub(crate) fn resolve_run_settings(mut settings: Settings, graph: &Graph) -> Set settings } -pub(crate) fn default_run_dir(run_id: &str, dry_run: bool) -> PathBuf { - make_run_dir(&default_runs_base(), run_id, dry_run) +pub(crate) fn default_run_dir(run_id: &RunId) -> PathBuf { + make_run_dir(&default_runs_base(), run_id) } -pub(crate) fn make_run_dir(runs_base: &Path, run_id: &str, dry_run: bool) -> PathBuf { - if dry_run { - runs_base.join(format!( - "{}-dry-run-{}", - Local::now().format("%Y%m%d"), - run_id - )) - } else { - runs_base.join(format!("{}-{}", Local::now().format("%Y%m%d"), run_id)) - } +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)) } #[cfg(test)] @@ -600,7 +579,6 @@ mod tests { settings: Settings::default(), cwd: dir.path().to_path_buf(), workflow_slug: None, - run_dir: Some(dir.path().join("run")), run_id: None, host_repo_path: None, base_branch: None, @@ -645,7 +623,6 @@ mod tests { }, cwd: dir.path().to_path_buf(), workflow_slug: Some("slug".to_string()), - run_dir: Some(dir.path().join("run")), run_id: Some(fixtures::RUN_1), host_repo_path: Some(dir.path().display().to_string()), base_branch: Some("main".to_string()), @@ -721,7 +698,6 @@ mod tests { }, cwd: dir.path().to_path_buf(), workflow_slug: None, - run_dir: Some(dir.path().join("run")), run_id: Some(fixtures::RUN_2), host_repo_path: None, base_branch: None, @@ -771,7 +747,6 @@ mod tests { }, cwd: dir.path().to_path_buf(), workflow_slug: Some("slug".to_string()), - run_dir: Some(run_dir.clone()), run_id: Some(fixtures::RUN_3), host_repo_path: None, base_branch: None, diff --git a/lib/crates/fabro-workflow/src/operations/fork.rs b/lib/crates/fabro-workflow/src/operations/fork.rs index 0e47f5c30..36b173712 100644 --- a/lib/crates/fabro-workflow/src/operations/fork.rs +++ b/lib/crates/fabro-workflow/src/operations/fork.rs @@ -85,16 +85,14 @@ fn fork_from_entry( let run_record_bytes = run_record_bytes.ok_or_else(|| anyhow::anyhow!("source run has no run.json"))?; - let now = chrono::Utc::now(); - let mut run_record: RunRecord = serde_json::from_slice(&run_record_bytes).context("failed to parse source run.json")?; run_record.run_id = new_run_id; - run_record.created_at = now; let new_run_record_bytes = serde_json::to_vec_pretty(&run_record).context("failed to serialize new run.json")?; let new_start_record_bytes = if start_record_bytes.is_some() { + let now = new_run_id.created_at(); let start_record = StartRecord { run_id: new_run_id, start_time: now, diff --git a/lib/crates/fabro-workflow/src/operations/mod.rs b/lib/crates/fabro-workflow/src/operations/mod.rs index 7414107ba..557b39547 100644 --- a/lib/crates/fabro-workflow/src/operations/mod.rs +++ b/lib/crates/fabro-workflow/src/operations/mod.rs @@ -10,7 +10,7 @@ mod test_support; mod validate; pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec}; -pub use create::{CreateRunInput, CreatedRun, create}; +pub use create::{CreateRunInput, CreatedRun, create, make_run_dir}; pub use fork::{ForkRunInput, fork}; pub use rebuild_meta::{ build_timeline_or_rebuild, find_run_id_by_prefix_or_store, rebuild_metadata_branch, diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index 17024ade3..79708835c 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -183,9 +183,9 @@ 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? { - if summary.catalog.run_id.to_string() == prefix { + if summary.run_id.to_string() == prefix { if summary.host_repo_path.is_none() { - return Ok(summary.catalog.run_id); + return Ok(summary.run_id); } let Some(host_repo_path) = summary.host_repo_path.as_deref() else { @@ -198,7 +198,7 @@ pub async fn find_run_id_by_prefix_or_store( continue; }; if host_repo_root == current_repo_root { - return Ok(summary.catalog.run_id); + return Ok(summary.run_id); } continue; } @@ -212,10 +212,8 @@ pub async fn find_run_id_by_prefix_or_store( let Ok(host_repo_root) = canonical_repo_root(&host_repo) else { continue; }; - if host_repo_root == current_repo_root - && summary.catalog.run_id.to_string().starts_with(prefix) - { - matches.push(summary.catalog.run_id); + if host_repo_root == current_repo_root && summary.run_id.to_string().starts_with(prefix) { + matches.push(summary.run_id); } } @@ -369,7 +367,6 @@ mod tests { fn sample_run_record(run_id: RunId, host_repo_path: Option<&str>) -> RunRecord { RunRecord { run_id, - created_at: created_at(), settings: Settings::default(), graph: Graph::new("test"), workflow_slug: None, @@ -431,7 +428,7 @@ mod tests { run_id: RunId, host_repo_path: Option<&str>, ) -> DurableRunStore { - let run_store = store.create_run(&run_id, created_at(), None).await.unwrap(); + let run_store = store.create_run(&run_id).await.unwrap(); let run_record = sample_run_record(run_id, host_repo_path); append_workflow_event( &run_store, @@ -815,10 +812,7 @@ mod tests { async fn rebuild_metadata_branch_errors_when_run_record_is_missing() { let (_dir, git_store) = temp_repo(); let durable_store = memory_store(); - let run_store = durable_store - .create_run(&test_run_id(), created_at(), None) - .await - .unwrap(); + let run_store = durable_store.create_run(&test_run_id()).await.unwrap(); let err = rebuild_metadata_branch(&git_store, &run_store, &test_run_id()) .await diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index f22095e98..6980f2559 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -860,7 +860,6 @@ mod tests { .unwrap_or_else(|| Path::new(".")) .to_path_buf(), workflow_slug: Some("test".to_string()), - run_dir: Some(run_dir.to_path_buf()), run_id: Some(fixtures::RUN_1), host_repo_path: None, base_branch: None, diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index b4fbae7ce..8343929a2 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -132,7 +132,6 @@ fn persisted_workflow(graph: Graph, source: String, run_dir: &Path, run_id: RunI run_dir.to_path_buf(), RunRecord { run_id, - created_at: Utc::now(), settings: Settings::default(), graph, workflow_slug: Some("test".to_string()), @@ -163,10 +162,7 @@ async fn test_run_store(run_id: &RunId) -> fabro_store::SlateRunStore { "", Duration::from_millis(1), )); - store - .create_run(run_id, chrono::Utc::now(), None) - .await - .unwrap() + store.create_run(run_id).await.unwrap() } async fn execute_test_run(run_dir: &Path, graph: Graph, run_id: &str) -> Executed { diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 0d97664c8..704d881bd 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -364,14 +364,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); - let inner_store = test_store() - .create_run( - &test_run_id(), - Utc::now(), - Some(run_dir.to_string_lossy().as_ref()), - ) - .await - .unwrap(); + let inner_store = test_store().create_run(&test_run_id()).await.unwrap(); let run_store = inner_store; let emitter = Arc::new(EventEmitter::new(test_run_id())); let store_logger = StoreProgressLogger::new(run_store.clone()); diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 9cb6fc46f..08b37c785 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -743,7 +743,6 @@ mod tests { run_dir.to_path_buf(), RunRecord { run_id: test_run_id(), - created_at: Utc::now(), settings: Settings::default(), graph, workflow_slug: Some("test".to_string()), @@ -770,14 +769,7 @@ mod tests { run_id: test_run_id(), run_store: { let store = memory_store(); - let inner = store - .create_run( - &test_run_id(), - chrono::Utc::now(), - Some(run_dir.to_string_lossy().as_ref()), - ) - .await - .unwrap(); + let inner = store.create_run(&test_run_id()).await.unwrap(); inner }, dry_run: false, @@ -839,14 +831,7 @@ mod tests { let persisted = test_persisted(graph, source, &run_dir); let emitter = Arc::new(crate::event::EventEmitter::new(test_run_id())); let store = memory_store(); - let run_store = store - .create_run( - &test_run_id(), - chrono::Utc::now(), - Some(run_dir.to_string_lossy().as_ref()), - ) - .await - .unwrap(); + let run_store = store.create_run(&test_run_id()).await.unwrap(); let store_logger = StoreProgressLogger::new(run_store.clone()); let seen = Arc::new(std::sync::Mutex::new(Vec::new())); emitter.on_event({ diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs index 6f04be230..5b100e4f6 100644 --- a/lib/crates/fabro-workflow/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs @@ -33,10 +33,9 @@ pub(crate) async fn load_from_store( .state() .await .map_err(|err| FabroError::engine(err.to_string()))?; - let mut run_record = state + let run_record = state .run .ok_or_else(|| FabroError::Precondition("run record missing from store".to_string()))?; - run_record.created_at = run_store.created_at(); let graph = run_record.graph.clone(); let source = state.graph_source.unwrap_or_default(); @@ -121,7 +120,6 @@ mod tests { fn sample_record(graph: Graph) -> RunRecord { RunRecord { run_id: fixtures::RUN_1, - created_at: Utc::now(), settings: Settings { dry_run: Some(true), verbose: Some(true), @@ -145,14 +143,7 @@ mod tests { source: Option<&str>, ) -> SlateRunStore { let store = memory_store(); - let run_store = store - .create_run( - &record.run_id, - record.created_at, - Some(run_dir.to_string_lossy().as_ref()), - ) - .await - .unwrap(); + let run_store = store.create_run(&record.run_id).await.unwrap(); append_workflow_event( &run_store, &record.run_id, @@ -246,8 +237,9 @@ mod tests { let loaded_record = loaded.run_record(); assert_eq!(loaded_record.run_id, expected.run_id); assert!( - (loaded_record.created_at.timestamp_millis() - expected.created_at.timestamp_millis()) - .abs() + (loaded_record.run_id.created_at().timestamp_millis() + - expected.run_id.created_at().timestamp_millis()) + .abs() <= 1 ); assert_eq!(loaded_record.settings, expected.settings); diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index e9d69d8b0..e11813a40 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -1059,14 +1059,7 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let store = test_store(); - let run_store = store - .create_run( - &fixtures::RUN_1, - Utc::now(), - Some(&tmp.path().display().to_string()), - ) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let conclusion = make_test_conclusion(); let body = build_pr_body( "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n", @@ -1091,18 +1084,10 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let store = test_store(); let created_at = Utc::now(); - let run_store = store - .create_run( - &fixtures::RUN_1, - created_at, - Some(&tmp.path().display().to_string()), - ) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let run_record = RunRecord { run_id: fixtures::RUN_1, - created_at, settings: Settings::default(), graph: Graph::new("test"), workflow_slug: Some("test".to_string()), @@ -1167,18 +1152,10 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let store = test_store(); let created_at = Utc::now(); - let run_store = store - .create_run( - &fixtures::RUN_1, - created_at, - Some(&tmp.path().display().to_string()), - ) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let run_record = RunRecord { run_id: fixtures::RUN_1, - created_at, settings: Settings::default(), graph: Graph::new("test"), workflow_slug: Some("test".to_string()), @@ -1369,14 +1346,7 @@ mod tests { async fn empty_diff_returns_none() { let tmp = tempfile::tempdir().unwrap(); let store = test_store(); - let run_store = store - .create_run( - &fixtures::RUN_1, - Utc::now(), - Some(&tmp.path().display().to_string()), - ) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let creds = GitHubAppCredentials { app_id: "123".to_string(), private_key_pem: "unused".to_string(), @@ -1404,17 +1374,9 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let store = test_store(); let created_at = Utc::now(); - let run_store = store - .create_run( - &fixtures::RUN_1, - created_at, - Some(&tmp.path().display().to_string()), - ) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let run_record = RunRecord { run_id: fixtures::RUN_1, - created_at, settings: Settings::default(), graph: Graph::new("test"), workflow_slug: None, diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 2baa17d4d..1fea14749 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -217,18 +217,10 @@ mod tests { checkpoint: &Checkpoint, ) -> fabro_store::SlateRunStore { let created_at = Utc::now(); - let inner = test_store() - .create_run( - &test_run_id(), - created_at, - Some(run_dir.to_string_lossy().as_ref()), - ) - .await - .unwrap(); + let inner = test_store().create_run(&test_run_id()).await.unwrap(); let run_store = inner; let run_record = RunRecord { run_id: test_run_id(), - created_at, settings: Settings::default(), graph: Graph::new("test"), workflow_slug: None, diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index 5d727df63..1499bd3d6 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -7,6 +7,7 @@ use fabro_store::{RunSummary, SlateStore}; use fabro_types::RunId; use serde::Serialize; +use crate::operations::make_run_dir; use crate::run_status::{RunStatus, StatusReason}; #[derive(Debug, Clone)] @@ -48,7 +49,7 @@ impl RunInfo { pub fn run_id(&self) -> RunId { self.summary .as_ref() - .map(|summary| summary.catalog.run_id) + .map(|summary| summary.run_id) .or_else(|| parse_run_id(&self.dir_name)) .expect("RunInfo must have a run id") } @@ -82,7 +83,7 @@ impl RunInfo { pub fn start_time(&self) -> String { self.summary .as_ref() - .and_then(|summary| summary.start_time.or(Some(summary.catalog.created_at))) + .and_then(|summary| summary.start_time.or(Some(summary.run_id.created_at()))) .or(self.start_time_dt) .map(|time| time.to_rfc3339()) .unwrap_or_default() @@ -194,7 +195,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result Result Option { - let run_dir = summary.catalog.run_dir.as_deref()?; - let path = PathBuf::from(run_dir); +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; } - let dir_name = path - .file_name() - .map(|name| name.to_string_lossy().to_string())?; - let start_time_dt = summary.catalog.created_at; + let dir_name = path.file_name()?.to_string_lossy().to_string(); + let start_time_dt = summary.run_id.created_at(); let end_time = if summary.status.is_some_and(RunStatus::is_terminal) { summary.duration_ms.and_then(|duration_ms| { Some(start_time_dt + chrono::Duration::milliseconds(i64::try_from(duration_ms).ok()?)) @@ -382,7 +380,6 @@ mod tests { fn sample_run_record() -> RunRecord { RunRecord { run_id: fixtures::RUN_1, - created_at: Utc::now(), settings: Settings::default(), graph: Graph::new("test"), workflow_slug: Some("test".to_string()), @@ -403,14 +400,7 @@ mod tests { 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, - run_record.created_at, - Some(&run_dir_string), - ) - .await - .unwrap(); + let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); append_workflow_event( &run_store, &fixtures::RUN_1, diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index b122b75e8..376e27cd7 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -3,7 +3,6 @@ use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; -use chrono::Utc; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph as GvGraph; use fabro_store::SlateStore; @@ -40,18 +39,13 @@ async fn initialized( options: InitializedOptions, ) -> Initialized { std::fs::create_dir_all(&run_options.run_dir).expect("failed to create run dir"); - let created_at = Utc::now(); let store = Arc::new(SlateStore::new( Arc::new(InMemory::new()), "", Duration::from_millis(1), )); let inner_store = store - .create_run( - &run_options.run_id, - created_at, - Some(run_options.run_dir.to_string_lossy().as_ref()), - ) + .create_run(&run_options.run_id) .await .expect("failed to create slate-backed test run store"); let run_store = inner_store;