From da93d9b501ba79f62241dbec64b67de7c170191e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 1 Apr 2026 21:26:44 -0400 Subject: [PATCH] Persist run creation directly into the store --- .../fabro-cli/src/commands/run/command.rs | 2 +- .../fabro-cli/src/commands/run/create.rs | 32 +- lib/crates/fabro-cli/src/commands/run/mod.rs | 2 +- lib/crates/fabro-server/src/server.rs | 29 +- .../fabro-workflow/src/operations/create.rs | 329 +++++++++++------- .../fabro-workflow/src/operations/start.rs | 55 +-- 6 files changed, 263 insertions(+), 186 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/command.rs b/lib/crates/fabro-cli/src/commands/run/command.rs index 8221f8179..b3c1f05b6 100644 --- a/lib/crates/fabro-cli/src/commands/run/command.rs +++ b/lib/crates/fabro-cli/src/commands/run/command.rs @@ -14,7 +14,7 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<( let quiet = args.detach; let prevent_idle_sleep = cli_settings.prevent_idle_sleep_enabled(); - let (run_id, run_dir) = super::create::create_run(&args, cli, styles, quiet)?; + let (run_id, run_dir) = super::create::create_run(&args, cli, styles, quiet).await?; #[cfg(feature = "sleep_inhibitor")] let _sleep_guard = crate::sleep_inhibitor::guard(prevent_idle_sleep); diff --git a/lib/crates/fabro-cli/src/commands/run/create.rs b/lib/crates/fabro-cli/src/commands/run/create.rs index d83c17940..a39997a05 100644 --- a/lib/crates/fabro-cli/src/commands/run/create.rs +++ b/lib/crates/fabro-cli/src/commands/run/create.rs @@ -1,18 +1,19 @@ use std::path::PathBuf; use crate::args::RunArgs; -use fabro_config::{ConfigLayer, FabroSettings}; +use fabro_config::{ConfigLayer, FabroSettings, FabroSettingsExt}; use fabro_types::RunId; use fabro_util::terminal::Styles; use fabro_workflow::error::FabroError; use fabro_workflow::operations::{CreateRunInput, WorkflowInput, create}; use super::output::{print_diagnostics_from_error, print_workflow_report_from_persisted}; +use crate::store; /// Create a workflow run: allocate run directory, persist RunRecord, return (run_id, run_dir). /// /// This does NOT execute the workflow — it only prepares the run directory. -pub(crate) fn create_run( +pub(crate) async fn create_run( args: &RunArgs, cli_defaults: ConfigLayer, styles: &Styles, @@ -36,16 +37,23 @@ pub(crate) fn create_run( .transpose() .map_err(|err| anyhow::anyhow!("invalid run ID: {err}"))?; - let created = match create(CreateRunInput { - workflow: WorkflowInput::Path(workflow_path.clone()), - settings, - cwd, - workflow_slug: None, - run_dir: None, - run_id, - base_branch: None, - host_repo_path: None, - }) { + let store = store::build_store(settings.storage_dir().as_path())?; + + let created = match create( + store.as_ref(), + CreateRunInput { + workflow: WorkflowInput::Path(workflow_path.clone()), + settings, + cwd, + workflow_slug: None, + run_dir: None, + run_id, + base_branch: None, + host_repo_path: None, + }, + ) + .await + { Ok(created) => created, Err(FabroError::ValidationFailed { diagnostics }) => { if !quiet { diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 7a23f2c33..33151cee6 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -47,7 +47,7 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( apply_json_defaults(&mut args, globals); let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); let cli = user_layer_with_globals(globals)?; - let (run_id, _run_dir) = create::create_run(&args, cli, styles, true)?; + let (run_id, _run_dir) = create::create_run(&args, cli, styles, true).await?; if globals.json { print_json_pretty(&serde_json::json!({ "run_id": run_id }))?; } else { diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 4dbec0d48..45c4b70ea 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -541,19 +541,24 @@ async fn start_run( 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 operations::create(CreateRunInput { - workflow: WorkflowInput::DotSource { - source: req.dot_source.clone(), - base_dir: None, + let created = match operations::create( + state.store.as_ref(), + CreateRunInput { + workflow: WorkflowInput::DotSource { + source: req.dot_source.clone(), + base_dir: None, + }, + 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, }, - 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, - }) { + ) + .await + { Ok(created) => created, Err(ref err @ FabroError::ValidationFailed { ref diagnostics }) => { let message = if diagnostics.is_empty() { diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 35677d59b..be451c72d 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -1,29 +1,24 @@ use std::collections::BTreeMap; use std::collections::HashMap; use std::path::{Path, PathBuf}; -use std::sync::Arc; -use std::time::Duration; use chrono::{Local, Utc}; use fabro_config::{FabroSettings, FabroSettingsExt}; use fabro_graphviz::graph::{AttrValue, Graph}; use fabro_model::{Catalog, Provider}; use fabro_sandbox::SandboxProvider; -use fabro_store::SlateStore; +use fabro_store::Store; use fabro_types::RunId; -use object_store::local::LocalFileSystem; use crate::error::FabroError; use crate::pipeline::types::PersistOptions; use crate::pipeline::{self, Persisted, TransformOptions, Validated}; use crate::records::RunRecord; use crate::run_lookup::default_runs_base; -use crate::run_status::{RunStatus, write_run_status}; +use crate::run_status::{RunStatus, RunStatusRecord, write_run_status}; use crate::transforms::{Transform, expand_vars}; use fabro_sandbox::daytona::detect_repo_info; -use tokio::runtime::Builder; -use super::hydrate::open_or_hydrate_run; use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow}; use crate::event::{ WorkflowRunEvent, append_progress_event, canonicalize_event_at, normalize_json_value, @@ -63,7 +58,7 @@ struct PersistCreateOptions { } /// Resolve workflow inputs, normalize settings, and persist a run directory. -pub fn create(request: CreateRunInput) -> Result { +pub async fn create(store: &dyn Store, request: CreateRunInput) -> Result { let resolved = resolve_workflow(ResolveWorkflowInput { workflow: request.workflow, settings: request.settings, @@ -131,7 +126,7 @@ pub fn create(request: CreateRunInput) -> Result { &resolved.raw_source, resolved.workflow_toml_path.as_deref(), )?; - hydrate_created_run(&persisted)?; + persist_created_run(store, &persisted, &resolved.raw_source).await?; Ok(CreatedRun { persisted, @@ -178,42 +173,79 @@ fn emit_run_created_event( .map_err(|err| FabroError::engine(err.to_string())) } -fn hydrate_created_run(persisted: &Persisted) -> Result<(), FabroError> { - let storage_dir = persisted.run_record().settings.storage_dir(); - let run_dir = persisted.run_dir().to_path_buf(); - let join = std::thread::spawn(move || -> Result<(), FabroError> { - let runtime = Builder::new_current_thread() - .enable_all() - .build() - .map_err(|err| FabroError::engine(err.to_string()))?; - runtime.block_on(async move { - let store = build_durable_store(&storage_dir)?; - open_or_hydrate_run(store.as_ref(), &run_dir) - .await - .map(|_| ()) - }) - }); +async fn persist_created_run( + store: &dyn Store, + persisted: &Persisted, + workflow_source: &str, +) -> 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 + { + Ok(run_store) => run_store, + Err(err) => store + .open_run(&record.run_id) + .await + .map_err(|open_err| FabroError::engine(open_err.to_string()))? + .ok_or_else(|| FabroError::engine(err.to_string()))?, + }; - match join.join() { - Ok(result) => result, - Err(_) => Err(FabroError::engine( - "store hydration thread panicked".to_string(), - )), + run_store.put_run(record).await.map_err(store_error)?; + if !workflow_source.is_empty() { + run_store + .put_graph(workflow_source) + .await + .map_err(store_error)?; } + run_store + .put_status(&RunStatusRecord::new(RunStatus::Submitted, None)) + .await + .map_err(store_error)?; + + let envelope = canonicalize_event_at( + &record.run_id, + &WorkflowRunEvent::RunCreated { + run_id: record.run_id, + settings: normalize_json_value( + serde_json::to_value(&record.settings) + .map_err(|err| FabroError::engine(err.to_string()))?, + ), + graph: normalize_json_value( + serde_json::to_value(&record.graph) + .map_err(|err| FabroError::engine(err.to_string()))?, + ), + workflow_source: (!workflow_source.is_empty()).then(|| workflow_source.to_string()), + workflow_config: None, + labels: record + .labels + .clone() + .into_iter() + .collect::>(), + run_dir: persisted.run_dir().display().to_string(), + working_directory: record.working_directory.display().to_string(), + host_repo_path: record.host_repo_path.clone(), + base_branch: record.base_branch.clone(), + workflow_slug: record.workflow_slug.clone(), + db_prefix: None, + }, + record.created_at, + ); + let payload = fabro_store::EventPayload::new( + serde_json::to_value(&envelope).map_err(|err| FabroError::engine(err.to_string()))?, + &record.run_id, + ) + .map_err(store_error)?; + run_store + .append_event(&payload) + .await + .map(|_| ()) + .map_err(store_error) } -fn build_durable_store(storage_dir: &Path) -> Result, FabroError> { - let store_path = storage_dir.join("store"); - std::fs::create_dir_all(&store_path)?; - let object_store = Arc::new( - LocalFileSystem::new_with_prefix(&store_path) - .map_err(|err| FabroError::engine(err.to_string()))?, - ); - Ok(Arc::new(SlateStore::new( - object_store, - "", - Duration::from_millis(5), - ))) +fn store_error(err: impl std::fmt::Display) -> FabroError { + FabroError::engine(err.to_string()) } fn validate_sandbox_provider(settings: &FabroSettings) -> Result<(), FabroError> { @@ -412,7 +444,7 @@ pub(crate) fn make_run_dir(runs_base: &Path, run_id: &str, dry_run: bool) -> Pat mod tests { use super::*; use fabro_graphviz::graph::AttrValue; - use fabro_store::{SlateStore, Store}; + use fabro_store::{InMemoryStore, SlateStore, Store}; use fabro_types::fixtures; use object_store::local::LocalFileSystem; use std::sync::Arc; @@ -421,6 +453,10 @@ mod tests { use crate::operations::{ValidateInput, validate}; use crate::run_status::RunStatusRecordExt; + fn memory_store() -> InMemoryStore { + InMemoryStore::default() + } + fn validate_dot(dot_source: &str, settings: FabroSettings) -> Validated { validate(ValidateInput { workflow: WorkflowInput::DotSource { @@ -606,26 +642,31 @@ mod tests { assert_eq!(validated.graph().goal(), "ship it"); } - #[test] - fn create_returns_validation_failed_with_diagnostics() { + #[tokio::test] + async fn create_returns_validation_failed_with_diagnostics() { let dot = r#"digraph Test { graph [goal="Test"] work [label="Work"] }"#; let dir = tempfile::tempdir().unwrap(); - let err = create(CreateRunInput { - workflow: WorkflowInput::DotSource { - source: dot.to_string(), - base_dir: None, + let store = memory_store(); + let err = create( + &store, + CreateRunInput { + workflow: WorkflowInput::DotSource { + source: dot.to_string(), + base_dir: None, + }, + settings: FabroSettings::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, }, - settings: FabroSettings::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, - }) + ) + .await .unwrap_err(); match err { @@ -636,36 +677,41 @@ mod tests { } } - #[test] - fn create_persists_normalized_config_and_initial_state() { + #[tokio::test] + async fn create_persists_normalized_config_and_initial_state() { let dir = tempfile::tempdir().unwrap(); - let created = create(CreateRunInput { - workflow: WorkflowInput::DotSource { - source: MINIMAL_DOT.to_string(), - base_dir: None, - }, - settings: FabroSettings { - llm: Some(fabro_config::run::LlmSettings { - model: Some("sonnet".to_string()), - provider: None, - fallbacks: None, - }), - pull_request: Some(fabro_config::run::PullRequestSettings { - enabled: false, + let store = memory_store(); + let created = create( + &store, + CreateRunInput { + workflow: WorkflowInput::DotSource { + source: MINIMAL_DOT.to_string(), + base_dir: None, + }, + settings: FabroSettings { + llm: Some(fabro_config::run::LlmSettings { + model: Some("sonnet".to_string()), + provider: None, + fallbacks: None, + }), + pull_request: Some(fabro_config::run::PullRequestSettings { + enabled: false, + ..Default::default() + }), + goal: Some("override goal".to_string()), + dry_run: Some(true), + labels: HashMap::from([("env".to_string(), "test".to_string())]), ..Default::default() - }), - goal: Some("override goal".to_string()), - dry_run: Some(true), - labels: HashMap::from([("env".to_string(), "test".to_string())]), - ..Default::default() + }, + 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()), }, - 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()), - }) + ) + .await .unwrap(); assert_eq!(created.run_id, fixtures::RUN_1); @@ -715,8 +761,8 @@ mod tests { assert!(!created.run_dir.join("id.txt").exists()); } - #[test] - fn create_copies_workflow_toml_snapshot() { + #[tokio::test] + async fn create_copies_workflow_toml_snapshot() { let dir = tempfile::tempdir().unwrap(); let workflow_dir = dir.path().join("workflow"); std::fs::create_dir_all(&workflow_dir).unwrap(); @@ -727,20 +773,25 @@ mod tests { ) .unwrap(); - let created = create(CreateRunInput { - workflow: WorkflowInput::Path(workflow_dir.join("workflow.toml")), - settings: FabroSettings { - storage_dir: Some(dir.path().join("storage")), - dry_run: Some(true), - ..Default::default() + let store = memory_store(); + let created = create( + &store, + CreateRunInput { + workflow: WorkflowInput::Path(workflow_dir.join("workflow.toml")), + settings: FabroSettings { + storage_dir: Some(dir.path().join("storage")), + dry_run: Some(true), + ..Default::default() + }, + cwd: dir.path().to_path_buf(), + workflow_slug: None, + run_dir: None, + run_id: None, + host_repo_path: None, + base_branch: None, }, - cwd: dir.path().to_path_buf(), - workflow_slug: None, - run_dir: None, - run_id: None, - host_repo_path: None, - base_branch: None, - }) + ) + .await .unwrap(); assert_eq!( @@ -749,29 +800,34 @@ mod tests { ); } - #[test] - fn create_resolves_working_directory_and_repo_path_from_request_cwd() { + #[tokio::test] + async fn create_resolves_working_directory_and_repo_path_from_request_cwd() { let dir = tempfile::tempdir().unwrap(); let workspace = dir.path().join("workspace"); std::fs::create_dir_all(&workspace).unwrap(); - let created = create(CreateRunInput { - workflow: WorkflowInput::DotSource { - source: MINIMAL_DOT.to_string(), - base_dir: None, + let store = memory_store(); + let created = create( + &store, + CreateRunInput { + workflow: WorkflowInput::DotSource { + source: MINIMAL_DOT.to_string(), + base_dir: None, + }, + settings: FabroSettings { + work_dir: Some("workspace".to_string()), + dry_run: Some(true), + ..Default::default() + }, + 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, }, - settings: FabroSettings { - work_dir: Some("workspace".to_string()), - dry_run: Some(true), - ..Default::default() - }, - 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, - }) + ) + .await .unwrap(); assert_eq!(created.persisted.run_record().working_directory, workspace); @@ -793,29 +849,32 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let storage_dir = dir.path().join("storage"); let run_dir = dir.path().join("run"); - let created = create(CreateRunInput { - workflow: WorkflowInput::DotSource { - source: MINIMAL_DOT.to_string(), - base_dir: None, - }, - settings: FabroSettings { - storage_dir: Some(storage_dir.clone()), - dry_run: Some(true), - ..Default::default() - }, - 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, - }) - .unwrap(); - std::fs::create_dir_all(storage_dir.join("store")).unwrap(); let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).unwrap()); let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(5))); + let created = create( + store.as_ref(), + CreateRunInput { + workflow: WorkflowInput::DotSource { + source: MINIMAL_DOT.to_string(), + base_dir: None, + }, + settings: FabroSettings { + storage_dir: Some(storage_dir.clone()), + dry_run: Some(true), + ..Default::default() + }, + 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, + }, + ) + .await + .unwrap(); let run_store = store .open_run_reader(&created.run_id) .await diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 01a53b1fd..cac4389fe 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -931,26 +931,31 @@ mod tests { start -> exit }"#; - fn persisted_workflow(dot: &str, run_dir: &Path) -> Persisted { - crate::operations::create(crate::operations::CreateRunInput { - workflow: crate::operations::WorkflowInput::DotSource { - source: dot.to_string(), - base_dir: None, + async fn persisted_workflow(dot: &str, run_dir: &Path) -> Persisted { + let store = InMemoryStore::default(); + crate::operations::create( + &store, + crate::operations::CreateRunInput { + workflow: crate::operations::WorkflowInput::DotSource { + source: dot.to_string(), + base_dir: None, + }, + settings: FabroSettings { + dry_run: Some(true), + ..Default::default() + }, + cwd: run_dir + .parent() + .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, }, - settings: FabroSettings { - dry_run: Some(true), - ..Default::default() - }, - cwd: run_dir - .parent() - .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, - }) + ) + .await .unwrap() .persisted } @@ -1007,7 +1012,7 @@ mod tests { }); } - persisted_workflow(MINIMAL_DOT, &run_dir); + persisted_workflow(MINIMAL_DOT, &run_dir).await; let started = start( &run_dir, test_start_services(&run_dir, emitter, registry).await, @@ -1030,7 +1035,7 @@ mod tests { let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1)); let registry = Arc::new(test_registry()); - persisted_workflow(MINIMAL_DOT, &run_dir); + persisted_workflow(MINIMAL_DOT, &run_dir).await; let started = start( &run_dir, @@ -1051,7 +1056,7 @@ mod tests { let registry = Arc::new(test_registry()); let visited = Arc::new(Mutex::new(Vec::new())); - persisted_workflow(MINIMAL_DOT, &run_dir); + persisted_workflow(MINIMAL_DOT, &run_dir).await; let started = start( &run_dir, @@ -1079,7 +1084,7 @@ mod tests { let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1)); let registry = Arc::new(test_registry()); - persisted_workflow(MINIMAL_DOT, &run_dir); + persisted_workflow(MINIMAL_DOT, &run_dir).await; let services = test_start_services(&run_dir, emitter, registry).await; // Write a checkpoint to the store (not disk) so start() sees it @@ -1116,7 +1121,7 @@ mod tests { let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1)); let registry = Arc::new(test_registry()); - persisted_workflow(MINIMAL_DOT, &run_dir); + persisted_workflow(MINIMAL_DOT, &run_dir).await; let result = resume( &run_dir, @@ -1138,7 +1143,7 @@ mod tests { let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1)); let registry = Arc::new(test_registry()); - persisted_workflow(MINIMAL_DOT, &run_dir); + persisted_workflow(MINIMAL_DOT, &run_dir).await; let checkpoint = Checkpoint::from_context( &Context::new(),