From 0b7e284924e5d8e046c23e170d7a7dc521329f3b Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 3 Apr 2026 17:46:44 -0700 Subject: [PATCH 1/2] Derive run metadata from RunId --- Cargo.lock | 1 + lib/crates/fabro-checkpoint/src/metadata.rs | 1 - .../fabro-cli/src/commands/run/create.rs | 1 - .../fabro-cli/src/commands/store/dump.rs | 39 +- .../fabro-cli/tests/it/cmd/server_start.rs | 1 - .../fabro-cli/tests/it/cmd/server_status.rs | 1 - .../fabro-cli/tests/it/cmd/server_stop.rs | 1 - .../tests/it/scenario/server_lifecycle.rs | 1 - .../fabro-cli/tests/it/workflow/real_cli.rs | 1 - lib/crates/fabro-server/src/server.rs | 26 +- lib/crates/fabro-store/Cargo.toml | 1 + lib/crates/fabro-store/src/lib.rs | 2 +- lib/crates/fabro-store/src/run_state.rs | 12 +- lib/crates/fabro-store/src/slate/catalog.rs | 124 +++--- lib/crates/fabro-store/src/slate/mod.rs | 377 +++++++----------- lib/crates/fabro-store/src/slate/run_store.rs | 65 ++- lib/crates/fabro-store/src/types.rs | 11 - lib/crates/fabro-types/src/run.rs | 2 - lib/crates/fabro-types/src/run_id.rs | 28 ++ 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 | 69 ++-- .../fabro-workflow/src/operations/fork.rs | 1 - .../fabro-workflow/src/operations/mod.rs | 1 + .../src/operations/rebuild_meta.rs | 8 +- .../fabro-workflow/src/operations/start.rs | 1 - .../src/pipeline/execute/tests.rs | 7 +- .../fabro-workflow/src/pipeline/finalize.rs | 10 +- .../fabro-workflow/src/pipeline/initialize.rs | 20 +- .../fabro-workflow/src/pipeline/persist.rs | 16 +- .../src/pipeline/pull_request.rs | 61 +-- .../fabro-workflow/src/pipeline/retro.rs | 12 +- lib/crates/fabro-workflow/src/run_lookup.rs | 29 +- lib/crates/fabro-workflow/src/test_support.rs | 8 +- 40 files changed, 354 insertions(+), 623 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 095c67621..3cce1610f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2028,6 +2028,7 @@ dependencies = [ "tokio", "tokio-stream", "tracing", + "ulid", ] [[package]] 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 372199710..f64bb32b3 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -163,7 +163,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(), @@ -171,7 +171,6 @@ mod tests { ); RunRecord { run_id, - created_at, settings: Settings::default(), graph, workflow_slug: Some("night-sky".to_string()), @@ -283,7 +282,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(); @@ -298,7 +297,7 @@ mod tests { visit: 2, }; append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RunCreated { run_id, @@ -318,7 +317,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::WorkflowRunStarted { name: "night-sky".to_string(), @@ -333,7 +332,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RunRunning { reason: status_record.reason, @@ -343,7 +342,7 @@ mod tests { .unwrap(); for checkpoint in [&first_checkpoint, &second_checkpoint] { append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::CheckpointCompleted { node_id: checkpoint.current_node.clone(), @@ -375,7 +374,7 @@ mod tests { .unwrap(); } append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::SandboxInitialized { working_directory: sandbox.working_directory.clone(), @@ -388,7 +387,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::Prompt { stage: "code".to_string(), @@ -402,7 +401,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::PromptCompleted { node_id: "code".to_string(), @@ -415,7 +414,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::StageCompleted { node_id: "code".to_string(), @@ -446,7 +445,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::CommandStarted { node_id: "code".to_string(), @@ -458,7 +457,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::CommandCompleted { node_id: "code".to_string(), @@ -472,7 +471,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RetroStarted { prompt: Some("How did it go?".to_string()), @@ -483,7 +482,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::RetroCompleted { duration_ms: 50, @@ -494,7 +493,7 @@ mod tests { .await .unwrap(); append_workflow_event( - run.as_ref(), + &run, &run_id, &WorkflowRunEvent::WorkflowRunCompleted { duration_ms: conclusion.duration_ms, @@ -542,7 +541,7 @@ mod tests { .unwrap(); let output = tempfile::tempdir().unwrap(); - let file_count = export_run(run.as_ref(), output.path()).await.unwrap(); + let file_count = export_run(&run, output.path()).await.unwrap(); assert_eq!(file_count, 22); let exported_run: RunRecord = read_json(&output.path().join("run.json")); @@ -637,10 +636,10 @@ 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.as_ref(), + &run, &run_id, &WorkflowRunEvent::RunCreated { run_id, @@ -672,7 +671,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let output = temp.path().join("dump"); - let err = export_run(run.as_ref(), &output).await.unwrap_err(); + let err = export_run(&run, &output).await.unwrap_err(); assert!(err.to_string().contains("asset filename")); assert!(!output.exists()); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs index 67d013274..f53fd4fbf 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_status.rs b/lib/crates/fabro-cli/tests/it/cmd/server_status.rs index 13039a893..1a9785060 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_status.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_status.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs b/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs index 49a6f5c65..42cabafaf 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs index 83677825e..051c4e024 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs b/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs index 2db5868a7..f00d9a377 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/real_cli.rs @@ -32,7 +32,6 @@ async fn run_real_cli_test(provider: Provider, model: &str) { &context, None, &emitter, - workspace.path(), &env, None, ) diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 50a6f659d..b826e11f7 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -94,7 +94,6 @@ struct ManagedRun { dot_source: String, status: RunStatus, error: Option, - created_at: chrono::DateTime, // Populated when running: interviewer: Option>, event_tx: Option>, @@ -496,7 +495,7 @@ async fn list_runs( message: msg.clone(), }), queue_position: queue_positions.get(id).copied(), - created_at: managed_run.created_at, + created_at: id.created_at(), }) .collect(); let page: Vec<_> = all_items.into_iter().skip(offset).take(limit + 1).collect(); @@ -517,7 +516,7 @@ fn compute_queue_positions(runs: &HashMap) -> HashMap 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(), @@ -557,7 +555,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, @@ -589,8 +586,7 @@ 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 mut runs = state.runs.lock().expect("runs lock poisoned"); @@ -600,14 +596,13 @@ async fn start_run( dot_source: req.dot_source, status: RunStatus::Queued, error: None, - created_at, interviewer: None, event_tx: None, context: None, checkpoint: None, cancel_tx: None, cancel_token: None, - run_dir: Some(run_dir), + run_dir: Some(created.run_dir), }, ); } @@ -853,7 +848,7 @@ pub fn spawn_scheduler(state: Arc) { } runs.iter() .filter(|(_, r)| r.status == RunStatus::Queued) - .min_by_key(|(_, r)| r.created_at) + .min_by_key(|(run_id, _)| run_id.created_at()) .map(|(id, _)| *id) }; match run_to_start { @@ -894,7 +889,7 @@ async fn get_run_status( error: managed_run.error.as_ref().map(|msg| RunError { message: msg.clone(), }), - created_at: managed_run.created_at, + created_at: id.created_at(), queue_position, }), ) @@ -1132,7 +1127,6 @@ async fn cancel_run( let _ = cancel_tx.send(()); } managed_run.status = RunStatus::Cancelled; - let created_at = managed_run.created_at; ( StatusCode::OK, Json(RunStatusResponse { @@ -1140,7 +1134,7 @@ async fn cancel_run( status: RunStatus::Cancelled, error: None, queue_position: None, - created_at, + created_at: id.created_at(), }), ) .into_response() @@ -1165,7 +1159,6 @@ async fn pause_run( Some(managed_run) => match managed_run.status { RunStatus::Running => { managed_run.status = RunStatus::Paused; - let created_at = managed_run.created_at; ( StatusCode::OK, Json(RunStatusResponse { @@ -1173,7 +1166,7 @@ async fn pause_run( status: RunStatus::Paused, error: None, queue_position: None, - created_at, + created_at: id.created_at(), }), ) .into_response() @@ -1198,7 +1191,6 @@ async fn unpause_run( Some(managed_run) => match managed_run.status { RunStatus::Paused => { managed_run.status = RunStatus::Running; - let created_at = managed_run.created_at; ( StatusCode::OK, Json(RunStatusResponse { @@ -1206,7 +1198,7 @@ async fn unpause_run( status: RunStatus::Running, error: None, queue_position: None, - created_at, + created_at: id.created_at(), }), ) .into_response() 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 ec6d12845..2deb7f25b 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -13,7 +13,7 @@ pub use error::{Result, StoreError}; pub use run_state::{NodeState, RunState}; pub use runtime::RuntimeState; pub use slate::{SlateRunStore, SlateStore}; -pub use types::{CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, RunSummary}; +pub use types::{EventEnvelope, EventPayload, NodeVisitRef, RunSummary}; use fabro_types::{Outcome, StageUsage}; diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 02d50c039..7a2089f0e 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -6,9 +6,7 @@ use chrono::{DateTime, Utc}; use serde::de::DeserializeOwned; use serde_json::Value; -use crate::{ - CatalogRecord, EventEnvelope, NodeOutcomeRecord, NodeVisitRef, Result, RunSummary, StoreError, -}; +use crate::{EventEnvelope, NodeOutcomeRecord, NodeVisitRef, Result, RunSummary, StoreError}; use fabro_types::{ Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro, RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StartRecord, @@ -90,7 +88,6 @@ impl RunState { .collect::>(); self.run = Some(RunRecord { run_id, - created_at: ts, settings, graph, workflow_slug: optional_string(&properties, "workflow_slug"), @@ -296,7 +293,7 @@ impl RunState { 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() @@ -309,10 +306,7 @@ impl RunState { (!goal.is_empty()).then(|| goal.to_string()) }); RunSummary { - run_id: catalog.run_id, - created_at: catalog.created_at, - db_prefix: catalog.db_prefix.clone(), - run_dir: catalog.run_dir.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 69d1adaed..8a7e0d070 100644 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ b/lib/crates/fabro-store/src/slate/catalog.rs @@ -1,79 +1,82 @@ +use std::collections::HashSet; use std::sync::Arc; -use chrono::{DateTime, Utc}; +use bytes::Bytes; use futures::TryStreamExt; use object_store::ObjectStore; use object_store::path::Path; -use crate::{CatalogRecord, ListRunsQuery, Result}; +use crate::{ListRunsQuery, Result}; use fabro_types::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::new().into()) .await?; store - .put( - &by_start_path(base_prefix, created_at, run_id), - bytes.into(), - ) + .put(&by_start_path(base_prefix, run_id), 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(); + let mut seen = HashSet::new(); for meta in metas { - let Some(record) = read_catalog_path(store.clone(), meta.location).await? else { + let Some(run_id) = parse_run_id_from_path(&meta.location) else { continue; }; + if !seen.insert(run_id) { + 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 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}/", - created_at.format("%Y-%m-%d-%H-%M-%S-%3f") + run_id.created_at().format("%Y-%m-%d-%H-%M-%S-%3f") ) } @@ -81,28 +84,16 @@ 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)] pub(super) mod test_support { use super::*; - use std::collections::{HashMap, HashSet}; pub(crate) async fn repair_catalog( store: Arc, @@ -115,17 +106,15 @@ pub(super) mod test_support { .list(Some(&by_id_prefix)) .try_collect::>() .await?; - let mut canonical = HashMap::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); - } - } + let run_ids = by_id_metas + .iter() + .filter_map(|meta| parse_run_id_from_path(&meta.location)) + .collect::>(); - for record in canonical.values() { - let path = by_start_path(base_prefix, record.created_at, &record.run_id); + 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, serde_json::to_vec(record)?.into()).await?; + store.put(&path, Bytes::new().into()).await?; } } @@ -133,33 +122,26 @@ 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(); - let Some(record) = read_catalog_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(&record.run_id).map(|canonical_record| { - by_start_path(base_prefix, canonical_record.created_at, &record.run_id) - }); - match expected { - Some(expected) if expected == location => { - seen.insert(record.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; } + 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(record)?.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 ad9d64d2f..d635d326c 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; @@ -14,7 +13,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; pub use run_store::SlateRunStore; use run_store::SlateRunStoreInner; @@ -96,7 +95,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 { @@ -104,21 +103,25 @@ impl SlateStore { weak.upgrade().map(SlateRunStore::from_inner) } - async fn open_run_store(&self, record: &CatalogRecord) -> Result> { - if let Some(active) = self.get_active_run(&record.run_id).await { - if active.matches_record(record) { + async fn open_run_store( + &self, + run_id: &RunId, + db_prefix: &str, + ) -> Result> { + 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 {:?}", - record.run_id + run_id ))); } - if !self.db_prefix_has_objects(&record.db_prefix).await? { + if !self.db_prefix_has_objects(db_prefix).await? { return Ok(None); } - 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; @@ -129,17 +132,21 @@ 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_prefix.to_string(), db).await?; self.cache_active_run(&run_store).await; Ok(Some(run_store)) } - async fn open_run_reader_store(&self, record: &CatalogRecord) -> Result> { - if !self.db_prefix_has_objects(&record.db_prefix).await? { + 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(&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; @@ -150,7 +157,7 @@ impl SlateStore { let _ = reader.close().await; return Ok(None); } - SlateRunStore::open_reader(record.clone(), reader) + SlateRunStore::open_reader(*run_id, db_prefix.to_string(), reader) .await .map(Some) } @@ -170,140 +177,111 @@ impl SlateStore { } impl SlateStore { - pub async fn create_run( - &self, - run_id: &RunId, - created_at: DateTime, - run_dir: Option<&str>, - ) -> Result { - let locator = + pub async fn create_run(&self, run_id: &RunId) -> Result { + 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 active.created_at() != created_at - || locator - .as_ref() - .is_some_and(|existing| existing.created_at != created_at) - { + if locator_exists && !active.matches_run(run_id, &db_prefix) { 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), - }; + 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, &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_prefix.clone(), 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 exists = + catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; + 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_run_store(&locator) + .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 = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id) - .await? - .ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?; + let exists = + catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; + 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_run_reader_store(&locator) + .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, query: &ListRunsQuery) -> Result> { - 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.matches_record(&record) { + 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 {:?}", - record.run_id + run_id ))); } 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? { + 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.created_at.cmp(&a.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(locator) = - 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 catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? { delete_path( self.object_store.clone(), - &catalog::by_start_path(&self.base_prefix, locator.created_at, run_id), + &catalog::by_start_path(&self.base_prefix, run_id), ) .await?; - self.delete_db_prefix(&locator.db_prefix).await?; + self.delete_db_prefix(&db_prefix).await?; delete_path( self.object_store.clone(), &catalog::by_id_path(&self.base_prefix, run_id), @@ -312,13 +290,13 @@ impl SlateStore { return Ok(()); } - if let Some(record) = active_record { + if active.is_some() { delete_path( self.object_store.clone(), - &catalog::by_start_path(&self.base_prefix, record.created_at, run_id), + &catalog::by_start_path(&self.base_prefix, run_id), ) .await?; - self.delete_db_prefix(&record.db_prefix).await?; + self.delete_db_prefix(&db_prefix).await?; delete_path( self.object_store.clone(), &catalog::by_id_path(&self.base_prefix, run_id), @@ -338,16 +316,9 @@ impl SlateStore { if meta.location.filename() != Some(expected_name.as_str()) { continue; } - let Some(record) = - catalog::read_catalog_path(self.object_store.clone(), meta.location.clone()) - .await? - else { - delete_path(self.object_store.clone(), &meta.location).await?; - continue; - }; - self.delete_db_prefix(&record.db_prefix).await?; delete_path(self.object_store.clone(), &meta.location).await?; } + self.delete_db_prefix(&db_prefix).await?; Ok(()) } } @@ -379,11 +350,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; @@ -392,6 +363,14 @@ mod tests { use crate::{EventPayload, NodeVisitRef}; + #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)] + 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() @@ -409,16 +388,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(), @@ -426,7 +406,6 @@ mod tests { ); RunRecord { run_id: test_run_id(run_id), - created_at, settings: Settings::default(), graph, workflow_slug: Some("night-sky".to_string()), @@ -587,7 +566,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(); } @@ -598,10 +577,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( @@ -650,7 +626,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); @@ -680,7 +656,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, }; @@ -731,7 +707,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 ); @@ -741,10 +717,7 @@ mod tests { 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", @@ -804,7 +777,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, }; @@ -819,7 +792,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 @@ -838,23 +811,11 @@ mod tests { #[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"), created_at, None) - .await - .unwrap(); - store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let _created_at = dt("2026-03-27T12:00:00Z"); + 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(_)))); } @@ -862,10 +823,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", @@ -916,11 +874,8 @@ 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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + 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(); run.append_event(&event_payload( @@ -951,7 +906,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; @@ -979,7 +934,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 @@ -993,11 +948,8 @@ 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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + 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 .unwrap(); @@ -1019,24 +971,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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + 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() .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::new().into(), ) .await .unwrap(); @@ -1051,38 +999,33 @@ 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 _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 _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 { flush_interval: Some(Duration::from_millis(1)), @@ -1091,21 +1034,15 @@ 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"), 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!( @@ -1117,11 +1054,8 @@ 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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + let _created_at = dt("2026-03-27T12:00:00Z"); + let run = store.create_run(&test_run_id("run-1")).await.unwrap(); let node = NodeVisitRef { node_id: "code", visit: 2, @@ -1151,11 +1085,8 @@ 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 run = store - .create_run(&test_run_id("run-1"), created_at, None) - .await - .unwrap(); + 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 .unwrap(); @@ -1200,10 +1131,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); @@ -1450,7 +1378,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.created_at, run_record.created_at); + 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 {}")); @@ -1519,10 +1450,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"); @@ -1749,10 +1677,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", @@ -1871,10 +1796,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: EventPayload = serde_json::from_value(serde_json::json!({ "run_id": "run-1" @@ -1897,10 +1819,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( @@ -1941,10 +1860,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( @@ -1965,10 +1881,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( diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index bf9fc34d1..65e7c8465 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; @@ -14,10 +14,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; use crate::run_state::EventProjectionCache; -use crate::{ - CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, Result, RunState, RunSummary, - StoreError, -}; +use crate::{EventEnvelope, EventPayload, NodeVisitRef, Result, RunState, RunSummary, StoreError}; use fabro_types::RunId; #[derive(Clone)] @@ -29,18 +26,14 @@ 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("created_at", &self.inner.created_at) .field("db_prefix", &self.inner.db_prefix) - .field("run_dir", &self.inner.run_dir) .finish_non_exhaustive() } } pub(crate) struct SlateRunStoreInner { run_id: RunId, - created_at: DateTime, db_prefix: String, - run_dir: Option, db: SlateRunDb, event_seq: AtomicU32, close_lock: Mutex<()>, @@ -53,14 +46,16 @@ 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_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: record.run_id, - created_at: record.created_at, - db_prefix: record.db_prefix, - run_dir: record.run_dir, + run_id, + db_prefix, db: SlateRunDb::Writer(db), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -69,14 +64,16 @@ impl SlateRunStore { }) } - pub(crate) async fn open_reader(record: CatalogRecord, 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: record.run_id, - created_at: record.created_at, - db_prefix: record.db_prefix, - run_dir: record.run_dir, + run_id, + db_prefix, db: SlateRunDb::Reader(Box::new(db)), event_seq: AtomicU32::new(event_seq), close_lock: Mutex::new(()), @@ -93,24 +90,12 @@ impl SlateRunStore { Arc::downgrade(&self.inner) } - pub(crate) fn record(&self) -> CatalogRecord { - CatalogRecord { - run_id: self.inner.run_id, - created_at: self.inner.created_at, - db_prefix: self.inner.db_prefix.clone(), - run_dir: self.inner.run_dir.clone(), - } + pub(crate) fn run_id(&self) -> RunId { + self.inner.run_id } - pub(crate) fn matches_record(&self, record: &CatalogRecord) -> bool { - self.inner.run_id == record.run_id - && self.inner.created_at == record.created_at - && self.inner.db_prefix == record.db_prefix - && self.inner.run_dir == record.run_dir - } - - pub(crate) fn created_at(&self) -> DateTime { - self.inner.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<()> { @@ -129,26 +114,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 = RunState::apply_events(&events)?; - Ok(state.build_summary(catalog)) + Ok(state.build_summary(run_id)) } async fn projected_state(&self) -> Result { diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index ed06f1fa3..bcdbc4e84 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -12,20 +12,9 @@ pub struct NodeVisitRef<'a> { pub visit: u32, } -#[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 { pub run_id: RunId, - pub created_at: DateTime, - pub db_prefix: String, - pub run_dir: Option, 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..fa1fafa87 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::{TimeZone, Utc}; + use super::{RunId, fixtures}; #[test] @@ -172,4 +184,20 @@ mod tests { assert_eq!(value, fixtures::RUN_42); } + + #[test] + 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(), dt); + } + + #[test] + 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!(run_id.created_at(), dt); + } } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 778698f0a..2254afd60 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -1891,10 +1891,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 3d6924d08..64bf5ef60 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -385,10 +385,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 { run_store: run_store.clone(), ..EngineServices::test_default() diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 89bfd6d81..3388dae46 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -193,10 +193,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 2e77eef0e..f3d4a5931 100644 --- a/lib/crates/fabro-workflow/src/handler/mod.rs +++ b/lib/crates/fabro-workflow/src/handler/mod.rs @@ -88,7 +88,7 @@ impl EngineServices { )), run_store: futures::executor::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 31004d2ef..f65367a12 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -673,10 +673,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 { run_store: run_store.clone(), ..EngineServices::test_default() diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index d5c1b344d..849f37861 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -200,10 +200,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 { run_store: run_store.clone(), ..EngineServices::test_default() diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 5701984d4..05b375837 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,7 +42,6 @@ pub struct CreatedRun { struct PersistCreateOptions { settings: Settings, - run_dir: Option, run_id: Option, workflow_slug: Option, labels: HashMap, @@ -70,7 +68,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 +160,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,7 +266,6 @@ fn persist_validated( ) -> Result { let PersistCreateOptions { settings, - run_dir, run_id, workflow_slug, labels, @@ -292,12 +277,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 = default_run_dir(&run_id); let run_record = RunRecord { run_id, - created_at: Utc::now(), settings, graph: validated.graph().clone(), workflow_slug, @@ -361,25 +344,19 @@ 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(crate) fn make_run_dir(runs_base: &Path, run_id: &RunId) -> PathBuf { + 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; @@ -447,6 +424,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 { @@ -600,7 +595,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 +639,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()), @@ -697,6 +690,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()); } @@ -721,7 +715,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, @@ -748,7 +741,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()); @@ -771,7 +763,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..c8765555a 100644 --- a/lib/crates/fabro-workflow/src/operations/fork.rs +++ b/lib/crates/fabro-workflow/src/operations/fork.rs @@ -90,7 +90,6 @@ fn fork_from_entry( 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")?; diff --git a/lib/crates/fabro-workflow/src/operations/mod.rs b/lib/crates/fabro-workflow/src/operations/mod.rs index 7414107ba..e9597f3e9 100644 --- a/lib/crates/fabro-workflow/src/operations/mod.rs +++ b/lib/crates/fabro-workflow/src/operations/mod.rs @@ -10,6 +10,7 @@ mod test_support; mod validate; pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec}; +pub(crate) use create::make_run_dir; pub use create::{CreateRunInput, CreatedRun, create}; pub use fork::{ForkRunInput, fork}; pub use rebuild_meta::{ diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs index e273784bf..36cee1353 100644 --- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs +++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs @@ -369,7 +369,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 +430,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, @@ -821,10 +820,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 2e01e37f2..5d6b579d6 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; @@ -132,7 +131,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 +161,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..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}; @@ -364,14 +363,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..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; @@ -743,7 +742,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 +768,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 +830,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 d8e6d01b8..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}; @@ -120,7 +119,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), @@ -144,14 +142,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, @@ -245,8 +236,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 54c5ee42d..646aa561d 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -1058,16 +1058,8 @@ 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, - 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", @@ -1089,21 +1081,11 @@ 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, - 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()), @@ -1122,7 +1104,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(), @@ -1165,21 +1147,11 @@ 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, - 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()), @@ -1198,7 +1170,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(), @@ -1368,16 +1340,8 @@ 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, - 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,18 +1368,9 @@ 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, - 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, @@ -1434,7 +1389,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 2baa17d4d..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,19 +215,10 @@ 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(), - 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 c5c6481f5..9c10cf08a 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::{ListRunsQuery, SlateStore}; use fabro_types::RunId; use serde::Serialize; +use crate::operations::make_run_dir; use crate::run_status::{RunStatus, StatusReason}; #[derive(Debug, Clone, Serialize)] @@ -121,7 +122,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result Result Option { - let run_dir = summary.run_dir.as_deref()?; - let path = PathBuf::from(run_dir); +fn run_info_from_summary(runs_base: &Path, summary: &fabro_store::RunSummary) -> 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.created_at; + .map(|name: &std::ffi::OsStr| name.to_string_lossy().to_string())?; + let start_time_dt = summary.run_id.created_at(); let start_time = summary.start_time.unwrap_or(start_time_dt); let end_time = if summary.status.is_some_and(RunStatus::is_terminal) { summary.duration_ms.and_then(|duration_ms| { @@ -319,7 +319,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}; @@ -327,6 +326,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 { @@ -340,7 +340,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()), @@ -354,21 +353,13 @@ 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, - 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, @@ -379,7 +370,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(), 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; From 35620f08e67588357e470820d83f3d2bc3c1743f Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 3 Apr 2026 17:47:51 -0700 Subject: [PATCH 2/2] Narrow fabro-cli helper visibility --- lib/crates/fabro-cli/src/commands/doctor.rs | 30 ++++++++++++-------- lib/crates/fabro-cli/src/commands/install.rs | 2 +- 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/doctor.rs b/lib/crates/fabro-cli/src/commands/doctor.rs index 37c284826..6e8d2de61 100644 --- a/lib/crates/fabro-cli/src/commands/doctor.rs +++ b/lib/crates/fabro-cli/src/commands/doctor.rs @@ -25,7 +25,7 @@ use crate::user_config::load_user_settings; // System dependency types and parsers (server mode only) // --------------------------------------------------------------------------- -pub struct DepSpec { +pub(crate) struct DepSpec { pub name: &'static str, command: &'static [&'static str], pub required: bool, @@ -34,7 +34,7 @@ pub struct DepSpec { } #[derive(Debug, Clone, PartialEq)] -pub enum ProbeOutcome { +pub(crate) enum ProbeOutcome { NotFound, Failed, Ok { version: Option }, @@ -55,7 +55,7 @@ fn parse_version(re: &Regex, output: &str) -> Option { )) } -pub const DEP_SPECS: &[DepSpec] = &[ +pub(crate) const DEP_SPECS: &[DepSpec] = &[ DepSpec { name: "openssl", command: &["openssl", "version"], @@ -79,7 +79,7 @@ pub const DEP_SPECS: &[DepSpec] = &[ }, ]; -pub fn probe_system_deps() -> Vec { +pub(crate) fn probe_system_deps() -> Vec { DEP_SPECS .iter() .map(|spec| { @@ -113,7 +113,7 @@ fn dep_issue(name: &str, issue: &str, required: bool) -> (CheckStatus, String) { (status, format!("{name}: {issue} ({severity})")) } -pub fn check_system_deps(specs: &[DepSpec], outcomes: &[ProbeOutcome]) -> CheckResult { +pub(crate) fn check_system_deps(specs: &[DepSpec], outcomes: &[ProbeOutcome]) -> CheckResult { let mut details = Vec::new(); let mut worst_status = CheckStatus::Pass; @@ -526,7 +526,7 @@ pub(crate) fn check_github_app(status: &GithubAppStatus) -> CheckResult { } } -pub struct ApiStatus { +pub(crate) struct ApiStatus { pub base_url: String, pub authentication_strategies: Vec, } @@ -542,7 +542,10 @@ fn format_auth_strategies(strategies: &[ApiAuthStrategy]) -> String { .join(", ") } -pub fn check_api(status: &ApiStatus, live_result: Option<&Result<(), String>>) -> CheckResult { +pub(crate) fn check_api( + status: &ApiStatus, + live_result: Option<&Result<(), String>>, +) -> CheckResult { let mut details = vec![ CheckDetail::new(format!("Base URL: {}", status.base_url)), CheckDetail::new(format!( @@ -566,7 +569,7 @@ pub fn check_api(status: &ApiStatus, live_result: Option<&Result<(), String>>) - } } -pub struct WebStatus { +pub(crate) struct WebStatus { pub url: String, pub auth_provider: AuthProvider, pub allowed_usernames_count: usize, @@ -579,7 +582,10 @@ fn format_auth_provider(provider: &AuthProvider) -> &'static str { } } -pub fn check_web(status: &WebStatus, live_result: Option<&Result<(), String>>) -> CheckResult { +pub(crate) fn check_web( + status: &WebStatus, + live_result: Option<&Result<(), String>>, +) -> CheckResult { let mut details = vec![ CheckDetail::new(format!("URL: {}", status.url)), CheckDetail::new(format!( @@ -611,13 +617,13 @@ pub fn check_web(status: &WebStatus, live_result: Option<&Result<(), String>>) - // Cryptographic key validation // --------------------------------------------------------------------------- -pub struct TlsCheckInput { +pub(crate) struct TlsCheckInput { pub cert_pem: String, pub key_pem: String, pub ca_pem: String, } -pub struct CryptoInput { +pub(crate) struct CryptoInput { pub auth_strategies: Vec, pub tls_files: Option>, pub jwt_public_key: Option, @@ -732,7 +738,7 @@ impl CryptoCheckState { } } -pub fn check_crypto(input: &CryptoInput) -> CheckResult { +pub(crate) fn check_crypto(input: &CryptoInput) -> CheckResult { let has_jwt = input.auth_strategies.contains(&ApiAuthStrategy::Jwt); let has_mtls = input.auth_strategies.contains(&ApiAuthStrategy::Mtls); diff --git a/lib/crates/fabro-cli/src/commands/install.rs b/lib/crates/fabro-cli/src/commands/install.rs index 0d894dacb..68b1627df 100644 --- a/lib/crates/fabro-cli/src/commands/install.rs +++ b/lib/crates/fabro-cli/src/commands/install.rs @@ -765,7 +765,7 @@ pub(crate) async fn run_install(web_url: &str, globals: &GlobalArgs) -> Result<( // --------------------------------------------------------------------------- mod hex { - pub fn encode(bytes: &[u8]) -> String { + pub(super) fn encode(bytes: &[u8]) -> String { bytes.iter().map(|b| format!("{b:02x}")).collect() } }