From 3b5bb33d73e7d5001b7a6fa487ed320fdaa30c82 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 14:33:22 -0400 Subject: [PATCH] refactor(run): remove worker-side SlateDB access Move detached workers onto an HTTP-backed runtime store so the server remains the only SlateDB owner. This replaces the worker's seeded local RunDatabase with a canonical server-backed handle for state, events, and blobs, and updates workflow runtime plumbing to use that abstraction. --- Cargo.lock | 2 + lib/crates/fabro-cli/Cargo.toml | 1 + .../fabro-cli/src/commands/pr/create.rs | 2 +- .../fabro-cli/src/commands/run/runner.rs | 247 +++++++++++++++--- lib/crates/fabro-cli/src/server_client.rs | 73 +++++- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 13 + lib/crates/fabro-cli/tests/it/cmd/run.rs | 13 + lib/crates/fabro-retro/src/retro_agent.rs | 38 ++- lib/crates/fabro-server/src/server.rs | 4 +- lib/crates/fabro-store/src/run_state.rs | 4 +- lib/crates/fabro-workflow/Cargo.toml | 1 + lib/crates/fabro-workflow/src/artifact.rs | 8 +- lib/crates/fabro-workflow/src/event.rs | 20 +- .../fabro-workflow/src/handler/agent.rs | 2 +- .../fabro-workflow/src/handler/command.rs | 2 +- .../src/handler/manager_loop.rs | 2 +- lib/crates/fabro-workflow/src/handler/mod.rs | 7 +- .../fabro-workflow/src/handler/parallel.rs | 4 +- .../fabro-workflow/src/handler/prompt.rs | 2 +- lib/crates/fabro-workflow/src/lib.rs | 1 + .../fabro-workflow/src/lifecycle/artifact.rs | 6 +- .../fabro-workflow/src/lifecycle/git.rs | 4 +- .../fabro-workflow/src/lifecycle/mod.rs | 4 +- .../fabro-workflow/src/operations/start.rs | 12 +- .../src/pipeline/execute/tests.rs | 6 +- .../fabro-workflow/src/pipeline/finalize.rs | 10 +- .../fabro-workflow/src/pipeline/initialize.rs | 4 +- .../fabro-workflow/src/pipeline/persist.rs | 17 +- .../src/pipeline/pull_request.rs | 19 +- .../fabro-workflow/src/pipeline/retro.rs | 24 +- .../fabro-workflow/src/pipeline/types.rs | 18 +- .../fabro-workflow/src/runtime_store.rs | 206 +++++++++++++++ lib/crates/fabro-workflow/src/test_support.rs | 2 +- 33 files changed, 632 insertions(+), 146 deletions(-) create mode 100644 lib/crates/fabro-workflow/src/runtime_store.rs diff --git a/Cargo.lock b/Cargo.lock index 0e695f40b..7351a9b26 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1526,6 +1526,7 @@ dependencies = [ "async-trait", "axum", "base64", + "bytes", "chrono", "clap", "clap_complete", @@ -2067,6 +2068,7 @@ dependencies = [ "assert_cmd", "async-trait", "base64", + "bytes", "chrono", "dirs", "fabro-agent", diff --git a/lib/crates/fabro-cli/Cargo.toml b/lib/crates/fabro-cli/Cargo.toml index 3062efbc9..0fa126743 100644 --- a/lib/crates/fabro-cli/Cargo.toml +++ b/lib/crates/fabro-cli/Cargo.toml @@ -82,6 +82,7 @@ sha2.workspace = true shlex = "1" walkdir.workspace = true object_store.workspace = true +bytes.workspace = true [target.'cfg(target_os = "macos")'.dependencies] core-foundation = { version = "0.9", optional = true } diff --git a/lib/crates/fabro-cli/src/commands/pr/create.rs b/lib/crates/fabro-cli/src/commands/pr/create.rs index a80c5f87c..b4b8582b3 100644 --- a/lib/crates/fabro-cli/src/commands/pr/create.rs +++ b/lib/crates/fabro-cli/src/commands/pr/create.rs @@ -103,7 +103,7 @@ pub(super) async fn create_command( &model, true, None, - &run_store, + &run_store.clone().into(), None, ) .await diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index c61c56da6..9cc6f7db0 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -4,21 +4,28 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use anyhow::{Context, Result, anyhow}; +use async_trait::async_trait; use fabro_config::RunScratch; use fabro_interview::FileInterviewer; -use fabro_store::{Database, EventPayload, RunDatabase}; -use fabro_types::{EventBody, RunEvent, RunId, Settings, StatusReason}; +use fabro_store::{EventEnvelope, EventPayload, RunProjection}; +use fabro_types::{EventBody, RunBlobId, RunEvent, RunId, Settings, StatusReason}; use fabro_workflow::event::{Emitter, RunEventSink}; use fabro_workflow::run_control::RunControlState; -use object_store::memory::InMemory as MemoryObjectStore; +use fabro_workflow::runtime_store::{RunStoreBackend, RunStoreHandle}; #[cfg(unix)] use tokio::signal::unix::{SignalKind, signal}; +use tokio::sync::Mutex; +use tokio::time::sleep; use crate::args::RunWorkerMode; use crate::server_client; use crate::shared::github::build_github_app_credentials; -const STORE_FLUSH_INTERVAL: Duration = Duration::from_millis(100); +const RUN_STORE_RETRY_DELAYS: [Duration; 3] = [ + Duration::from_millis(50), + Duration::from_millis(100), + Duration::from_millis(250), +]; #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum WorkerTitlePhase { @@ -43,7 +50,7 @@ pub(crate) async fn execute( set_worker_title(&run_id, initial_worker_title_phase(mode)); let client = server_client::connect_server_target_direct(&server).await?; - let run_store = load_seed_run_store(&client, &run_id).await?; + let run_store = HttpRunStore::connect(run_id, client.clone_for_reuse()).await?; let run_state = run_store .state() .await @@ -62,7 +69,6 @@ pub(crate) async fn execute( let cancel_token = Arc::new(AtomicBool::new(false)); install_signal_handlers(Arc::clone(&run_control), Arc::clone(&cancel_token))?; let github_app = maybe_build_github_app_credentials(&run_record.settings)?; - let event_client = client.clone_for_reuse(); let services = fabro_workflow::operations::StartServices { run_id, cancel_token: Some(Arc::clone(&cancel_token)), @@ -70,11 +76,10 @@ pub(crate) async fn execute( interviewer, run_store: run_store.clone(), event_sink: RunEventSink::fanout(vec![ - RunEventSink::store(run_store), + RunEventSink::backend(run_store), RunEventSink::callback(move |event| { update_worker_title_from_event(&event); - let client = event_client.clone_for_reuse(); - async move { client.append_run_event(&event.run_id, &event).await } + async move { Ok(()) } }), ]), run_control: Some(run_control), @@ -95,42 +100,139 @@ pub(crate) async fn execute( Ok(()) } -fn open_memory_store() -> Arc { - Arc::new(Database::new( - Arc::new(MemoryObjectStore::new()), - "", - STORE_FLUSH_INTERVAL, - )) +#[derive(Clone)] +struct HttpRunStore { + run_id: RunId, + client: server_client::ServerStoreClient, + state: Arc>, + events: Arc>>>, } -async fn load_seed_run_store( - client: &server_client::ServerStoreClient, - run_id: &RunId, -) -> Result { - let events = client - .list_run_events(run_id, None, None) - .await - .with_context(|| format!("failed to fetch run events for {run_id}"))?; - let payloads = events - .into_iter() - .map(|event| event.payload) - .collect::>(); - seed_run_store(run_id, &payloads).await -} - -async fn seed_run_store(run_id: &RunId, events: &[EventPayload]) -> Result { - let store = open_memory_store(); - let run_store = store - .create_run(run_id) - .await - .with_context(|| format!("failed to create in-memory run store for {run_id}"))?; - for payload in events { - run_store - .append_event(payload) +impl HttpRunStore { + async fn connect( + run_id: RunId, + client: server_client::ServerStoreClient, + ) -> Result { + let state = client + .get_run_state(&run_id) .await - .with_context(|| format!("failed to seed in-memory run store for {run_id}"))?; + .with_context(|| format!("failed to fetch run state for {run_id}"))?; + Ok(RunStoreHandle::new(Arc::new(Self { + run_id, + client, + state: Arc::new(Mutex::new(state)), + events: Arc::new(Mutex::new(None)), + }))) + } + + async fn with_retries(&self, operation: &'static str, mut op: F) -> Result + where + F: FnMut() -> Fut, + Fut: std::future::Future>, + { + let mut last_error = None; + for attempt in 0..=RUN_STORE_RETRY_DELAYS.len() { + match op().await { + Ok(value) => return Ok(value), + Err(err) => last_error = Some(err), + } + if let Some(delay) = RUN_STORE_RETRY_DELAYS.get(attempt) { + sleep(*delay).await; + } + } + Err(last_error + .unwrap_or_else(|| anyhow!("run store operation failed")) + .context(format!( + "worker lost canonical run store during {operation}" + ))) + } + + async fn refresh_state_from_server(&self) -> Result { + self.with_retries("refresh state", || { + let client = self.client.clone_for_reuse(); + let run_id = self.run_id; + async move { client.get_run_state(&run_id).await } + }) + .await + } + + async fn apply_acknowledged_event(&self, seq: u32, event: &RunEvent) -> Result<()> { + let payload = EventPayload::new(event.to_value()?, &self.run_id)?; + let envelope = EventEnvelope { seq, payload }; + + { + let mut state = self.state.lock().await; + if let Err(err) = state.apply_event(&envelope) { + tracing::warn!(run_id = %self.run_id, error = %err, "failed to apply acknowledged event to local run-state mirror; refreshing from server"); + drop(state); + let refreshed = self.refresh_state_from_server().await?; + *self.state.lock().await = refreshed; + } + } + + let mut events = self.events.lock().await; + if let Some(cached) = events.as_mut() { + cached.push(envelope); + } + + Ok(()) + } +} + +#[async_trait] +impl RunStoreBackend for HttpRunStore { + async fn load_state(&self) -> Result { + Ok(self.state.lock().await.clone()) + } + + async fn list_events(&self) -> Result> { + let mut cached = self.events.lock().await; + if let Some(events) = cached.as_ref() { + return Ok(events.clone()); + } + + let events = self + .with_retries("list run events", || { + let client = self.client.clone_for_reuse(); + let run_id = self.run_id; + async move { client.list_run_events(&run_id, None, None).await } + }) + .await?; + *cached = Some(events.clone()); + Ok(events) + } + + async fn append_run_event(&self, event: &RunEvent) -> Result<()> { + let seq = self + .with_retries("append run event", || { + let client = self.client.clone_for_reuse(); + let run_id = self.run_id; + let event = event.clone(); + async move { client.append_run_event(&run_id, &event).await } + }) + .await?; + self.apply_acknowledged_event(seq, event).await + } + + async fn write_blob(&self, data: &[u8]) -> Result { + self.with_retries("write run blob", || { + let client = self.client.clone_for_reuse(); + let run_id = self.run_id; + let data = data.to_vec(); + async move { client.write_run_blob(&run_id, &data).await } + }) + .await + } + + async fn read_blob(&self, id: &RunBlobId) -> Result> { + self.with_retries("read run blob", || { + let client = self.client.clone_for_reuse(); + let run_id = self.run_id; + let blob_id = *id; + async move { client.read_run_blob(&run_id, &blob_id).await } + }) + .await } - Ok(run_store) } fn set_worker_title(run_id: &RunId, phase: WorkerTitlePhase) { @@ -247,8 +349,12 @@ fn install_signal_handlers( #[cfg(test)] mod tests { + use httpmock::MockServer; + use serde_json::json; + use super::{ - WorkerTitlePhase, initial_worker_title_phase, worker_title, worker_title_phase_for_event, + WorkerTitlePhase, execute, initial_worker_title_phase, worker_title, + worker_title_phase_for_event, }; use crate::args::RunWorkerMode; use fabro_types::fixtures; @@ -342,4 +448,61 @@ mod tests { Some(WorkerTitlePhase::Failed) ); } + + #[tokio::test] + async fn worker_bootstrap_loads_run_state_without_prefetching_run_events() { + let server = MockServer::start_async().await; + let run_id = fixtures::RUN_1; + + let state_mock = server + .mock_async(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/state")); + then.status(200) + .header("Content-Type", "application/json") + .body( + json!({ + "run": null, + "graph_source": null, + "start": null, + "status": null, + "checkpoint": null, + "checkpoints": [], + "conclusion": null, + "retro": null, + "retro_prompt": null, + "retro_response": null, + "sandbox": null, + "final_patch": null, + "pull_request": null, + "nodes": {} + }) + .to_string(), + ); + }) + .await; + let events_mock = server + .mock_async(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/events")); + then.status(200) + .header("Content-Type", "application/json") + .body(json!({ "data": [], "meta": { "has_more": false } }).to_string()); + }) + .await; + + let run_dir = tempfile::tempdir().unwrap(); + let error = execute( + run_id, + format!("{}/api/v1", server.base_url()), + run_dir.path().to_path_buf(), + RunWorkerMode::Start, + ) + .await + .unwrap_err(); + + assert!(error.to_string().contains("has no run record")); + state_mock.assert_async().await; + assert_eq!(events_mock.calls_async().await, 0); + } } diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index b8c411251..e214451ca 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -4,10 +4,11 @@ use std::path::{Path, PathBuf}; use std::time::Duration; use anyhow::{Context as _, Result, anyhow, bail}; +use bytes::Bytes; use fabro_api::types; use fabro_server::bind::Bind; use fabro_store::{EventEnvelope, RunSummary, StageId}; -use fabro_types::{RunEvent, RunId, Settings}; +use fabro_types::{RunBlobId, RunEvent, RunId, Settings}; use futures::StreamExt; use serde::de::DeserializeOwned; use tokio::time::sleep; @@ -403,16 +404,65 @@ impl ServerStoreClient { Ok(()) } - pub(crate) async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result<()> { + pub(crate) async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result { let body: types::RunEvent = convert_type(event)?; - self.client + let response = self + .client .append_run_event() .id(run_id.to_string()) .body(body) .send() .await .map_err(map_api_error)?; - Ok(()) + u32::try_from(response.into_inner().seq).context("append_run_event returned invalid seq") + } + + pub(crate) async fn write_run_blob(&self, run_id: &RunId, data: &[u8]) -> Result { + let response = self + .client + .write_run_blob() + .id(run_id.to_string()) + .body(data.to_vec()) + .send() + .await + .map_err(map_api_error)?; + response + .into_inner() + .id + .parse() + .context("write_run_blob returned invalid blob id") + } + + pub(crate) async fn read_run_blob( + &self, + run_id: &RunId, + blob_id: &RunBlobId, + ) -> Result> { + let response = self + .client + .read_run_blob() + .id(run_id.to_string()) + .blob_id(blob_id.to_string()) + .send() + .await; + match response { + Ok(response) => { + let mut stream = response.into_inner(); + let mut bytes = Vec::new(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.map_err(|err| anyhow!("{err}"))?; + bytes.extend_from_slice(&chunk); + } + Ok(Some(Bytes::from(bytes))) + } + Err(err) => { + if is_not_found_error(&err) { + Ok(None) + } else { + Err(map_api_error(err)) + } + } + } } pub(crate) async fn delete_store_run(&self, run_id: &RunId) -> Result<()> { @@ -584,6 +634,21 @@ where } } +fn is_not_found_error(err: &progenitor_client::Error) -> bool +where + E: serde::Serialize + std::fmt::Debug, +{ + match err { + progenitor_client::Error::ErrorResponse(response) => { + response.status() == reqwest::StatusCode::NOT_FOUND + } + progenitor_client::Error::UnexpectedResponse(response) => { + response.status() == reqwest::StatusCode::NOT_FOUND + } + _ => false, + } +} + fn map_attach_run_stream_error( err: progenitor_client::Error, ) -> RunAttachStreamError { diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 5dea2ae9f..8d0745c87 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -575,6 +575,19 @@ fn attach_json_errors_without_prompting_for_human_input() { } }, "host_repo_path": "[TEMP_DIR]", + "provenance": { + "client": { + "name": "fabro-cli", + "user_agent": "fabro-cli/0.176.2", + "version": "0.176.2" + }, + "server": { + "version": "0.176.2" + }, + "subject": { + "auth_method": "disabled" + } + }, "run_dir": "[RUN_DIR]", "settings": { "goal": "Wait for approval", diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 2c375208e..d6610bac9 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -746,6 +746,19 @@ fn json_run_implies_auto_approve_for_human_gates() { } }, "host_repo_path": "[TEMP_DIR]", + "provenance": { + "client": { + "name": "fabro-cli", + "user_agent": "fabro-cli/0.176.2", + "version": "0.176.2" + }, + "server": { + "version": "0.176.2" + }, + "subject": { + "auth_method": "disabled" + } + }, "run_dir": "[RUN_DIR]", "settings": { "auto_approve": true, diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index 2dee2d27d..dfc27d798 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/lib/crates/fabro-retro/src/retro_agent.rs @@ -10,7 +10,7 @@ use fabro_agent::{ use fabro_llm::client::Client; use fabro_llm::provider::Provider; use fabro_llm::types::ToolDefinition; -use fabro_store::RunDatabase; +use fabro_store::{EventEnvelope, RunProjection}; use tokio::task::JoinHandle; use crate::retro::{RetroNarrative, SmoothnessRating}; @@ -135,7 +135,8 @@ pub fn build_retro_prompt(retro_data_dir: &str) -> String { /// files via tool access, then calls `submit_retro` with its analysis. pub async fn run_retro_agent( sandbox: &Arc, - run_store: &RunDatabase, + state: &RunProjection, + events: &[EventEnvelope], run_dir: &Path, llm_client: &Client, provider: Provider, @@ -144,7 +145,7 @@ pub async fn run_retro_agent( ) -> anyhow::Result { // Upload data files into sandbox (needed for Daytona; no-op effect for local // since the agent can also read from the original paths via tools). - upload_data_files(sandbox, run_store, run_dir, RETRO_DATA_DIR).await?; + upload_data_files(sandbox, state, events, run_dir, RETRO_DATA_DIR).await?; // Build provider profile with the submit_retro tool let captured: Arc>> = Arc::new(Mutex::new(None)); @@ -292,7 +293,8 @@ fn build_profile(provider: Provider, model: &str) -> Box { async fn upload_data_files( sandbox: &Arc, - run_store: &RunDatabase, + state: &RunProjection, + events: &[EventEnvelope], _run_dir: &Path, target_dir: &str, ) -> anyhow::Result<()> { @@ -302,19 +304,16 @@ async fn upload_data_files( .await .map_err(|e| anyhow::anyhow!("Failed to create retro data dir: {e}"))?; - let progress_content = match run_store.list_events().await { - Ok(envelopes) => { - let lines: Vec = envelopes - .into_iter() - .filter_map(|env| serde_json::to_string(env.payload.as_value()).ok()) - .collect(); - if lines.is_empty() { - None - } else { - Some(lines.join("\n") + "\n") - } + let progress_content = { + let lines: Vec = events + .iter() + .filter_map(|env| serde_json::to_string(env.payload.as_value()).ok()) + .collect(); + if lines.is_empty() { + None + } else { + Some(lines.join("\n") + "\n") } - Err(e) => return Err(anyhow::anyhow!("Failed to load events from store: {e}")), }; if let Some(content) = progress_content { sandbox @@ -323,24 +322,23 @@ async fn upload_data_files( .map_err(|e| anyhow::anyhow!("Failed to upload progress.jsonl: {e}"))?; } - let state = run_store - .state() - .await - .map_err(|e| anyhow::anyhow!("Failed to load run state from store: {e}"))?; let checkpoint_content = state .checkpoint + .clone() .map(|cp| serde_json::to_string_pretty(&cp)) .transpose()?; upload_file(sandbox, target_dir, "checkpoint.json", checkpoint_content).await?; let run_content = state .run + .clone() .map(|run| serde_json::to_string_pretty(&run)) .transpose()?; upload_file(sandbox, target_dir, "run.json", run_content).await?; let start_content = state .start + .clone() .map(|start| serde_json::to_string_pretty(&start)) .transpose()?; upload_file(sandbox, target_dir, "start.json", start_content).await?; diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 92c83e04f..2a867004f 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -2611,7 +2611,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { run_store.subscribe(), state.global_event_tx.clone(), )); - let persisted = match Persisted::load_from_store(&run_store, &run_dir).await { + let persisted = match Persisted::load_from_store(&run_store.clone().into(), &run_dir).await { Ok(persisted) => persisted, Err(e) => { tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run"); @@ -2647,7 +2647,7 @@ async fn execute_run_in_process(state: Arc, run_id: RunId) { cancel_token: Some(Arc::clone(&cancel_token)), emitter: Arc::clone(&emitter), interviewer: Arc::clone(&interviewer) as Arc, - run_store: run_store.clone(), + run_store: run_store.clone().into(), event_sink: workflow_event::RunEventSink::store(run_store.clone()), run_control: None, github_app, diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 060818635..476f93759 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -71,7 +71,7 @@ pub(crate) struct EventProjectionCache { } impl RunProjection { - pub(crate) fn apply_events(events: &[EventEnvelope]) -> Result { + pub fn apply_events(events: &[EventEnvelope]) -> Result { let mut state = Self::default(); for event in events { state.apply_event(event)?; @@ -79,7 +79,7 @@ impl RunProjection { Ok(state) } - pub(crate) fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> { + pub fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> { let stored = RunEvent::from_ref(event.payload.as_value()) .map_err(|err| StoreError::InvalidEvent(format!("invalid stored event: {err}")))?; let ts = stored.ts; diff --git a/lib/crates/fabro-workflow/Cargo.toml b/lib/crates/fabro-workflow/Cargo.toml index 3bef794c8..d30ca8a78 100644 --- a/lib/crates/fabro-workflow/Cargo.toml +++ b/lib/crates/fabro-workflow/Cargo.toml @@ -40,6 +40,7 @@ thiserror.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true +bytes.workspace = true object_store.workspace = true ulid.workspace = true uuid.workspace = true diff --git a/lib/crates/fabro-workflow/src/artifact.rs b/lib/crates/fabro-workflow/src/artifact.rs index 938dfff14..31eb8856e 100644 --- a/lib/crates/fabro-workflow/src/artifact.rs +++ b/lib/crates/fabro-workflow/src/artifact.rs @@ -4,9 +4,9 @@ use std::path::Path; use serde_json::Value; use fabro_agent::Sandbox; -use fabro_store::RunDatabase; use crate::error::{FabroError, Result}; +use crate::runtime_store::RunStoreHandle; /// Threshold above which values are persisted as blobs and materialized to disk (100KB). const BLOB_OFFLOAD_THRESHOLD: usize = 100 * 1024; @@ -26,7 +26,7 @@ const ARTIFACT_POINTER_PREFIX: &str = "file://"; /// Returns an error if blob persistence or cache materialization fails. pub async fn offload_large_values( updates: &mut HashMap, - run_store: &RunDatabase, + run_store: &RunStoreHandle, cache_dir: &Path, ) -> Result<()> { std::fs::create_dir_all(cache_dir)?; @@ -161,7 +161,7 @@ mod tests { let mut updates = HashMap::new(); updates.insert("response.plan".to_string(), serde_json::json!(large_string)); - offload_large_values(&mut updates, &run_store, dir.path()) + offload_large_values(&mut updates, &run_store.clone().into(), dir.path()) .await .unwrap(); @@ -196,7 +196,7 @@ mod tests { let mut updates = HashMap::new(); updates.insert("small_key".to_string(), small_value.clone()); - offload_large_values(&mut updates, &run_store, dir.path()) + offload_large_values(&mut updates, &run_store.clone().into(), dir.path()) .await .unwrap(); diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index ea603803f..366d7412a 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -22,6 +22,8 @@ use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback use fabro_llm::types::Usage as LlmUsage; use fabro_util::redact::redact_json_value; +use crate::runtime_store::RunStoreHandle; + pub use fabro_types::{EventBody, RunNoticeLevel}; /// Events emitted during workflow run execution for observability. @@ -2371,7 +2373,7 @@ pub async fn append_event_to_sink( #[derive(Clone)] pub enum RunEventSink { - Store(RunDatabase), + Store(RunStoreHandle), JsonLines(Arc>>>), Callback(Arc), Composite(Vec), @@ -2383,6 +2385,11 @@ type RunEventSinkCallback = dyn Fn(RunEvent) -> RunEventSinkFuture + Send + Sync impl RunEventSink { #[must_use] pub fn store(run_store: RunDatabase) -> Self { + Self::Store(RunStoreHandle::local(run_store)) + } + + #[must_use] + pub fn backend(run_store: RunStoreHandle) -> Self { Self::Store(run_store) } @@ -2420,12 +2427,7 @@ impl RunEventSink { while let Some(sink) = pending.pop() { match sink { Self::Store(run_store) => { - let payload = build_redacted_event_payload(event, &event.run_id)?; - run_store - .append_event(&payload) - .await - .map(|_| ()) - .map_err(anyhow::Error::from)?; + run_store.append_run_event(event).await?; } Self::JsonLines(writer) => { let line = redacted_event_json(event)?; @@ -2506,9 +2508,9 @@ pub struct StoreProgressLogger { impl StoreProgressLogger { #[must_use] - pub fn new(run_store: RunDatabase) -> Self { + pub fn new(run_store: impl Into) -> Self { Self { - inner: RunEventLogger::new(RunEventSink::store(run_store)), + inner: RunEventLogger::new(RunEventSink::backend(run_store.into())), } } diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 442abb83c..2df11f1e4 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -423,7 +423,7 @@ mod tests { let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)), - run_store: run_store.clone(), + run_store: run_store.clone().into(), ..EngineServices::test_default() }; let logger = crate::event::StoreProgressLogger::new(run_store.clone()); diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 0034aad9f..b25a717ac 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/lib/crates/fabro-workflow/src/handler/command.rs @@ -202,7 +202,7 @@ mod tests { let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)), - run_store: run_store.clone(), + run_store: run_store.clone().into(), ..EngineServices::test_default() }; let logger = crate::event::StoreProgressLogger::new(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 e3ba56f2a..9aca0778e 100644 --- a/lib/crates/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflow/src/handler/manager_loop.rs @@ -251,7 +251,7 @@ impl Handler for SubWorkflowHandler { run_options: child_run_options, workflow_path: child_workflow_path, workflow_bundle, - run_store, + run_store: run_store.into(), checkpoint: None, seed_context: Some(child_context), emitter, diff --git a/lib/crates/fabro-workflow/src/handler/mod.rs b/lib/crates/fabro-workflow/src/handler/mod.rs index 01220b71d..ec93ff93b 100644 --- a/lib/crates/fabro-workflow/src/handler/mod.rs +++ b/lib/crates/fabro-workflow/src/handler/mod.rs @@ -22,7 +22,6 @@ use async_trait::async_trait; use fabro_agent::Sandbox; #[cfg(test)] use fabro_store::Database; -use fabro_store::RunDatabase; #[cfg(test)] use object_store::memory::InMemory; @@ -30,6 +29,7 @@ use crate::context::Context; use crate::error::FabroError; use crate::event::Emitter; use crate::outcome::{Outcome, OutcomeExt}; +use crate::runtime_store::RunStoreHandle; use crate::sandbox_git::GitState; use crate::workflow_bundle::WorkflowBundle; use fabro_graphviz::graph::{Graph, Node, shape_to_handler_type}; @@ -43,7 +43,7 @@ pub struct EngineServices { pub registry: Arc, pub emitter: Arc, pub sandbox: Arc, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, /// Git state for the current run. Set via `set_git_state` at the start of /// `run_via_core` and read by parallel/fan-in handlers. pub(crate) git_state: std::sync::RwLock>>, @@ -137,7 +137,8 @@ impl EngineServices { }) }) .join() - .expect("test run store thread should join"), + .expect("test run store thread should join") + .into(), git_state: std::sync::RwLock::new(None), hook_runner: None, env: HashMap::new(), diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 6266042ce..5c7b8158a 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -634,7 +634,7 @@ mod tests { let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)), - run_store: run_store.clone(), + run_store: run_store.clone().into(), ..EngineServices::test_default() }; let logger = crate::event::StoreProgressLogger::new(run_store.clone()); @@ -686,7 +686,7 @@ mod tests { let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)), - run_store: run_store.clone(), + run_store: run_store.clone().into(), ..EngineServices::test_default() }; let logger = crate::event::StoreProgressLogger::new(run_store.clone()); diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index d6682da4d..5ca6c6a4e 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -203,7 +203,7 @@ mod tests { let run_store = store.create_run(&fixtures::RUN_1).await.unwrap(); let services = EngineServices { emitter: Arc::new(crate::event::Emitter::new(fixtures::RUN_1)), - run_store: run_store.clone(), + run_store: run_store.clone().into(), ..EngineServices::test_default() }; let logger = crate::event::StoreProgressLogger::new(run_store.clone()); diff --git a/lib/crates/fabro-workflow/src/lib.rs b/lib/crates/fabro-workflow/src/lib.rs index 4b461e44b..16c3279f2 100644 --- a/lib/crates/fabro-workflow/src/lib.rs +++ b/lib/crates/fabro-workflow/src/lib.rs @@ -138,6 +138,7 @@ pub mod run_dump; pub mod run_lookup; pub mod run_options; pub mod run_status; +pub mod runtime_store; pub mod sandbox_git; #[doc(hidden)] pub mod test_support; diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index df55cda08..4be82eaed 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -3,7 +3,6 @@ use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; use async_trait::async_trait; -use fabro_store::RunDatabase; use fabro_core::graph::NodeSpec; use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, RunLifecycle}; @@ -16,6 +15,7 @@ use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::outcome::StageUsage; +use crate::runtime_store::RunStoreHandle; use fabro_core::error::Result as CoreResult; use fabro_core::lifecycle::NodeDecision; @@ -26,7 +26,7 @@ type WfNodeDecision = NodeDecision>; /// Sub-lifecycle responsible for artifact collection, offloading, and syncing. pub(crate) struct ArtifactLifecycle { pub sandbox: Arc, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub blob_cache_dir: PathBuf, pub emitter: Arc, pub artifacts_dir: PathBuf, @@ -40,7 +40,7 @@ impl ArtifactLifecycle { #[allow(clippy::too_many_arguments)] pub(crate) fn new( sandbox: Arc, - run_store: RunDatabase, + run_store: RunStoreHandle, blob_cache_dir: PathBuf, emitter: Arc, artifacts_dir: PathBuf, diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 9ccaa4fc1..317b4aaf7 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -4,7 +4,6 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use fabro_config::RunScratch; -use fabro_store::RunDatabase; use fabro_types::RunId; use tokio::fs; @@ -21,6 +20,7 @@ use crate::graph::WorkflowNode; use crate::outcome::{Outcome, StageStatus, StageUsage}; use crate::run_dump::RunDump; use crate::run_options::RunOptions; +use crate::runtime_store::RunStoreHandle; use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host}; type WfRunState = ExecutionState>; @@ -67,7 +67,7 @@ pub(crate) struct GitLifecycle { pub emitter: Arc, pub run_dir: PathBuf, pub run_id: RunId, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub run_options: Arc, pub start_node_id: Option, // Cross-lifecycle data (shared with EventLifecycle) diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index aad7cecc3..83055b38e 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -14,7 +14,6 @@ use std::time::Instant; use async_trait::async_trait; use fabro_config::RunScratch; -use fabro_store::RunDatabase; use fabro_types::RunId; use fabro_core::error::Result as CoreResult; @@ -33,6 +32,7 @@ use crate::graph::WorkflowNode; use crate::outcome::{Outcome, StageUsage}; use crate::run_control::RunControlState; use crate::run_options::RunOptions; +use crate::runtime_store::RunStoreHandle; use fabro_graphviz::graph::types::Graph as GvGraph; use fabro_hooks::HookRunner; use fabro_sandbox::Sandbox; @@ -84,7 +84,7 @@ impl WorkflowLifecycle { sandbox: &Arc, graph: Arc, run_dir: &PathBuf, - run_store: &RunDatabase, + run_store: &RunStoreHandle, run_options: &Arc, is_resume: bool, on_node: crate::OnNodeCallback, diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 45fe0402b..7f5f1e582 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -9,7 +9,6 @@ use fabro_config::{project as project_config, run as run_config, sandbox as sand use fabro_interview::{AutoApproveInterviewer, Interviewer}; use fabro_model::{Catalog, FallbackTarget, Provider}; use fabro_sandbox::{SandboxProvider, SandboxSpec}; -use fabro_store::RunDatabase; use fabro_types::{RunId, Settings}; use crate::context::Context; @@ -29,6 +28,7 @@ use crate::records::Checkpoint; use crate::run_control::RunControlState; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; use crate::run_status::{RunStatus, StatusReason}; +use crate::runtime_store::RunStoreHandle; use crate::workflow_bundle::{StoredWorkflowBundle, WorkflowBundle}; use fabro_config::run::PullRequestSettings; use fabro_retro::retro::Retro; @@ -48,7 +48,7 @@ struct RunSession { sandbox_env: SandboxEnvSpec, devcontainer: Option, seed_context: Option, - run_store: RunDatabase, + run_store: RunStoreHandle, event_sink: RunEventSink, git: Option, github_app: Option, @@ -70,7 +70,7 @@ pub struct StartServices { pub cancel_token: Option>, pub emitter: Arc, pub interviewer: Arc, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub event_sink: RunEventSink, pub run_control: Option>, pub github_app: Option, @@ -221,7 +221,7 @@ pub(super) async fn execute_persisted_run( async fn persist_terminal_engine_failure( run_id: RunId, - run_store: &RunDatabase, + run_store: &RunStoreHandle, event_sink: &RunEventSink, _run_dir: &Path, error: &FabroError, @@ -867,7 +867,7 @@ mod tests { cancel_token: None, emitter, interviewer: Arc::new(fabro_interview::AutoApproveInterviewer), - run_store: store.open_run(&fixtures::RUN_1).await.unwrap(), + run_store: store.open_run(&fixtures::RUN_1).await.unwrap().into(), event_sink: RunEventSink::store(store.open_run(&fixtures::RUN_1).await.unwrap()), run_control: None, github_app: None, @@ -1005,7 +1005,7 @@ mod tests { node_visits: HashMap::new(), }; crate::event::append_event( - &services.run_store, + &store.open_run(&fixtures::RUN_1).await.unwrap(), &services.run_id, &Event::CheckpointCompleted { node_id: checkpoint.current_node.clone(), diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index 57b04f66e..fd2cf9e72 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs @@ -185,7 +185,7 @@ async fn execute_test_run_with_options( persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value), InitOptions { run_id: run_id_value, - run_store, + run_store: run_store.into(), dry_run: false, emitter, sandbox: SandboxSpec::Local { @@ -241,7 +241,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { persisted_workflow(graph, source, &run_dir, test_run_id("run-test")), InitOptions { run_id: test_run_id("run-test"), - run_store: test_run_store(&test_run_id("run-test")).await, + run_store: test_run_store(&test_run_id("run-test")).await.into(), dry_run: false, emitter: test_emitter_arc("run-test"), sandbox: SandboxSpec::Local { @@ -311,7 +311,7 @@ async fn run_with_lifecycle( persisted_workflow(graph.clone(), String::new(), &run_dir, run_id), InitOptions { run_id, - run_store: test_run_store(&run_id).await, + run_store: test_run_store(&run_id).await.into(), dry_run: false, emitter, sandbox: SandboxSpec::Local { diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index b5dd9b26c..4cfdf9911 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -8,9 +8,9 @@ use crate::records::{Checkpoint, Conclusion, StageSummary}; use crate::run_dump::RunDump; use crate::run_options::RunOptions; use crate::run_status::{RunStatus, StatusReason}; +use crate::runtime_store::RunStoreHandle; use crate::sandbox_git::git_push_host; use fabro_hooks::{HookContext, HookEvent, HookRunner}; -use fabro_store::RunDatabase; use super::types::{Concluded, FinalizeOptions, Retroed}; @@ -63,7 +63,7 @@ pub fn classify_engine_result( } pub(crate) async fn build_conclusion_from_store( - run_store: &RunDatabase, + run_store: &RunStoreHandle, status: StageStatus, failure_reason: Option, run_duration_ms: u64, @@ -169,7 +169,7 @@ fn build_conclusion_from_parts( /// /// This captures the last diff.patch (written after the final checkpoint) and retro.json. /// Best-effort: errors are logged as warnings. -pub async fn write_finalize_commit(run_options: &RunOptions, run_store: &RunDatabase) { +pub async fn write_finalize_commit(run_options: &RunOptions, run_store: &RunStoreHandle) { let (Some(meta_branch), Some(repo_path)) = ( run_options .git @@ -372,7 +372,7 @@ mod tests { graph: Graph::new("test"), outcome: Ok(Outcome::success()), run_options: test_run_options(&run_dir), - run_store: run_store.clone(), + run_store: run_store.clone().into(), hook_runner: None, emitter, sandbox: Arc::new(fabro_agent::LocalSandbox::new( @@ -387,7 +387,7 @@ mod tests { &FinalizeOptions { run_dir: run_dir.clone(), run_id: test_run_id(), - run_store: run_store.clone(), + run_store: run_store.clone().into(), workflow_name: "test".to_string(), hook_runner: None, preserve_sandbox: true, diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 16d3d0062..b3c999188 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -769,7 +769,7 @@ mod tests { run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); - inner + inner.into() }, dry_run: false, emitter, @@ -846,7 +846,7 @@ mod tests { persisted, InitOptions { run_id: test_run_id(), - run_store, + run_store: run_store.into(), dry_run: false, emitter, sandbox: SandboxSpec::Local { diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs index a48974114..9a7c19a30 100644 --- a/lib/crates/fabro-workflow/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs @@ -1,8 +1,7 @@ use std::path::Path; -use fabro_store::RunDatabase; - use crate::error::FabroError; +use crate::runtime_store::RunStoreHandle; use super::types::{PersistOptions, Persisted, Validated}; @@ -26,7 +25,7 @@ pub(crate) fn persist( } pub(crate) async fn load_from_store( - run_store: &RunDatabase, + run_store: &RunStoreHandle, run_dir: &Path, ) -> Result { let state = run_store @@ -231,7 +230,9 @@ mod tests { .unwrap(); let run_store = seeded_store(&run_dir, &expected, Some(&source)).await; - let loaded = load_from_store(&run_store, &run_dir).await.unwrap(); + let loaded = load_from_store(&run_store.clone().into(), &run_dir) + .await + .unwrap(); let loaded_record = loaded.run_record(); assert_eq!(loaded_record.run_id, expected.run_id); @@ -284,7 +285,9 @@ mod tests { record.graph = graph; let run_store = seeded_store(&run_dir, &record, None).await; - let loaded = load_from_store(&run_store, &run_dir).await.unwrap(); + let loaded = load_from_store(&run_store.clone().into(), &run_dir) + .await + .unwrap(); assert!(loaded.source().is_empty()); } @@ -300,7 +303,9 @@ mod tests { record.graph = graph.clone(); let run_store = seeded_store(&run_dir, &record, Some(&source)).await; - let loaded = load_from_store(&run_store, &run_dir).await.unwrap(); + let loaded = load_from_store(&run_store.clone().into(), &run_dir) + .await + .unwrap(); assert_eq!( serde_json::to_value(loaded.graph()).unwrap(), diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index 2ad40b797..a986456e8 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -1,5 +1,5 @@ use fabro_config::run::MergeStrategy; -use fabro_store::{RunDatabase, RunProjection}; +use fabro_store::RunProjection; use fabro_types::PullRequestRecord; use tracing::{debug, info}; @@ -12,6 +12,7 @@ use super::types::{Concluded, Finalized, PullRequestOptions}; use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::outcome::{StageStatus, format_cost as outcome_format_cost}; use crate::records::{Conclusion, RunRecord}; +use crate::runtime_store::RunStoreHandle; use fabro_retro::retro::Retro; /// Derive a PR title from the workflow goal. @@ -280,7 +281,7 @@ fn emit_run_notice( }); } -async fn load_pull_request_diff(run_store: &RunDatabase) -> String { +async fn load_pull_request_diff(run_store: &RunStoreHandle) -> String { run_store .state() .await @@ -298,7 +299,7 @@ pub async fn build_pr_body( diff: &str, goal: &str, model: &str, - run_store: &RunDatabase, + run_store: &RunStoreHandle, conclusion: Option<&Conclusion>, ) -> Result { debug!("Building PR body"); @@ -406,7 +407,7 @@ pub async fn maybe_open_pull_request( model: &str, draft: bool, auto_merge: Option, - run_store: &RunDatabase, + run_store: &RunStoreHandle, conclusion: Option<&Conclusion>, ) -> Result, String> { if diff.is_empty() { @@ -1064,7 +1065,7 @@ mod tests { "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n", "Implement feature", "mock-model", - &run_store, + &run_store.clone().into(), Some(&conclusion), ) .await @@ -1134,7 +1135,7 @@ mod tests { "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n", "Implement feature", "mock-model", - &run_store, + &run_store.clone().into(), Some(&conclusion), ) .await @@ -1220,7 +1221,7 @@ mod tests { "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n", "Implement feature", "mock-model", - &run_store, + &run_store.clone().into(), Some(&make_test_conclusion()), ) .await @@ -1363,7 +1364,7 @@ mod tests { "claude-sonnet-4-20250514", false, None, - &run_store, + &run_store.clone().into(), None, ) .await; @@ -1429,7 +1430,7 @@ mod tests { .await .unwrap(); - let diff = load_pull_request_diff(&run_store).await; + let diff = load_pull_request_diff(&run_store.clone().into()).await; assert!(diff.contains("from_store")); } diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index a4cb7c7d5..83886226b 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -23,7 +23,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { return None; } }; - let Some(cp) = state.checkpoint else { + let Some(ref cp) = state.checkpoint else { tracing::warn!("Could not load checkpoint, skipping retro"); if let Some(ref emitter) = options.emitter { emitter.emit(&Event::RetroFailed { @@ -80,9 +80,23 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { } }) }); + let events = match options.run_store.list_events().await { + Ok(events) => events, + Err(err) => { + tracing::warn!(error = %err, "Could not load events from store, skipping retro"); + if let Some(ref emitter) = options.emitter { + emitter.emit(&Event::RetroFailed { + error: err.to_string(), + duration_ms: 0, + }); + } + return None; + } + }; run_retro_agent( &options.sandbox, - &options.run_store, + &state, + &events, &options.run_dir, client, options.provider, @@ -319,7 +333,7 @@ mod tests { graph: Graph::new("test"), outcome: Ok(crate::outcome::Outcome::success()), run_options: test_run_options(&run_dir), - run_store: run_store.clone(), + run_store: run_store.clone().into(), hook_runner: None, emitter: Arc::clone(&emitter), sandbox: Arc::clone(&sandbox), @@ -334,7 +348,7 @@ mod tests { executed, &RetroOptions { run_id: test_run_id(), - run_store, + run_store: run_store.into(), workflow_name: "test".to_string(), goal: "Ship it".to_string(), run_dir: run_dir.clone(), @@ -371,7 +385,7 @@ mod tests { let retro = run_retro( &RetroOptions { run_id: test_run_id(), - run_store: test_run_store(&run_dir, &checkpoint).await, + run_store: test_run_store(&run_dir, &checkpoint).await.into(), workflow_name: "test".to_string(), goal: "Ship it".to_string(), run_dir: run_dir.clone(), diff --git a/lib/crates/fabro-workflow/src/pipeline/types.rs b/lib/crates/fabro-workflow/src/pipeline/types.rs index 27f22f497..00e58cd97 100644 --- a/lib/crates/fabro-workflow/src/pipeline/types.rs +++ b/lib/crates/fabro-workflow/src/pipeline/types.rs @@ -11,7 +11,6 @@ use fabro_llm::Provider; use fabro_mcp::config::McpServerSettings; use fabro_model::FallbackTarget; use fabro_sandbox::SandboxSpec; -use fabro_store::RunDatabase; use fabro_types::RunId; use fabro_validate::Diagnostic; @@ -24,6 +23,7 @@ use crate::outcome::Outcome; use crate::records::{Checkpoint, Conclusion, RunRecord}; use crate::run_control::RunControlState; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; +use crate::runtime_store::RunStoreHandle; use crate::transforms::Transform; use crate::workflow_bundle::WorkflowBundle; use fabro_config::run::PullRequestSettings; @@ -198,7 +198,7 @@ impl Persisted { } pub async fn load_from_store( - run_store: &RunDatabase, + run_store: &RunStoreHandle, run_dir: &Path, ) -> Result { super::persist::load_from_store(run_store, run_dir).await @@ -230,7 +230,7 @@ pub struct DevcontainerSpec { pub struct InitOptions { pub run_id: RunId, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub dry_run: bool, pub emitter: Arc, pub sandbox: SandboxSpec, @@ -259,7 +259,7 @@ pub struct Initialized { pub run_options: RunOptions, pub workflow_path: Option, pub workflow_bundle: Option>, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub(crate) checkpoint: Option, pub(crate) seed_context: Option, pub emitter: Arc, @@ -281,7 +281,7 @@ pub struct Executed { pub graph: Graph, pub outcome: Result, pub run_options: RunOptions, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub hook_runner: Option>, pub emitter: Arc, pub sandbox: Arc, @@ -298,7 +298,7 @@ pub struct Retroed { pub graph: Graph, pub outcome: Result, pub run_options: RunOptions, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub hook_runner: Option>, pub emitter: Arc, pub sandbox: Arc, @@ -338,7 +338,7 @@ pub struct TransformOptions { /// Options for the RETRO phase. pub struct RetroOptions { pub run_id: RunId, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub workflow_name: String, pub goal: String, pub run_dir: PathBuf, @@ -356,7 +356,7 @@ pub struct RetroOptions { pub struct FinalizeOptions { pub run_dir: PathBuf, pub run_id: RunId, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub workflow_name: String, pub hook_runner: Option>, pub preserve_sandbox: bool, @@ -366,7 +366,7 @@ pub struct FinalizeOptions { /// Options for the PULL_REQUEST phase. pub struct PullRequestOptions { pub run_dir: PathBuf, - pub run_store: RunDatabase, + pub run_store: RunStoreHandle, pub pr_config: Option, pub github_app: Option, pub origin_url: Option, diff --git a/lib/crates/fabro-workflow/src/runtime_store.rs b/lib/crates/fabro-workflow/src/runtime_store.rs new file mode 100644 index 000000000..3eec4143a --- /dev/null +++ b/lib/crates/fabro-workflow/src/runtime_store.rs @@ -0,0 +1,206 @@ +use std::sync::Arc; + +use anyhow::Result; +use async_trait::async_trait; +use bytes::Bytes; +use fabro_store::{EventEnvelope, RunDatabase, RunProjection}; +use fabro_types::{RunBlobId, RunEvent}; + +use crate::event::build_redacted_event_payload; + +#[async_trait] +pub trait RunStoreBackend: Send + Sync { + async fn load_state(&self) -> Result; + async fn list_events(&self) -> Result>; + async fn append_run_event(&self, event: &RunEvent) -> Result<()>; + async fn write_blob(&self, data: &[u8]) -> Result; + async fn read_blob(&self, id: &RunBlobId) -> Result>; +} + +#[derive(Clone)] +pub struct RunStoreHandle { + backend: Arc, +} + +impl RunStoreHandle { + #[must_use] + pub fn new(backend: Arc) -> Self { + Self { backend } + } + + #[must_use] + pub fn local(run_store: RunDatabase) -> Self { + Self::new(Arc::new(LocalRunStoreBackend { run_store })) + } + + pub async fn state(&self) -> Result { + self.backend.load_state().await + } + + pub async fn list_events(&self) -> Result> { + self.backend.list_events().await + } + + pub async fn append_run_event(&self, event: &RunEvent) -> Result<()> { + self.backend.append_run_event(event).await + } + + pub async fn write_blob(&self, data: &[u8]) -> Result { + self.backend.write_blob(data).await + } + + pub async fn read_blob(&self, id: &RunBlobId) -> Result> { + self.backend.read_blob(id).await + } +} + +impl From for RunStoreHandle { + fn from(value: RunDatabase) -> Self { + Self::local(value) + } +} + +struct LocalRunStoreBackend { + run_store: RunDatabase, +} + +#[async_trait] +impl RunStoreBackend for LocalRunStoreBackend { + async fn load_state(&self) -> Result { + self.run_store.state().await.map_err(anyhow::Error::from) + } + + async fn list_events(&self) -> Result> { + self.run_store + .list_events() + .await + .map_err(anyhow::Error::from) + } + + async fn append_run_event(&self, event: &RunEvent) -> Result<()> { + let payload = build_redacted_event_payload(event, &event.run_id)?; + self.run_store + .append_event(&payload) + .await + .map(|_| ()) + .map_err(anyhow::Error::from) + } + + async fn write_blob(&self, data: &[u8]) -> Result { + self.run_store + .write_blob(data) + .await + .map_err(anyhow::Error::from) + } + + async fn read_blob(&self, id: &RunBlobId) -> Result> { + self.run_store + .read_blob(id) + .await + .map_err(anyhow::Error::from) + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::path::PathBuf; + use std::sync::Arc; + use std::time::Duration; + + use chrono::Utc; + use fabro_graphviz::graph::Graph; + use fabro_store::Database; + use fabro_types::fixtures; + use fabro_types::run_event::RunStatusTransitionProps; + use fabro_types::{EventBody, RunEvent, Settings}; + use object_store::memory::InMemory; + + use super::RunStoreHandle; + use crate::event::{Event, append_event}; + use crate::records::RunRecord; + + async fn test_run_store() -> fabro_store::RunDatabase { + let store = Arc::new(Database::new( + Arc::new(InMemory::new()), + "", + Duration::from_millis(1), + )); + store.create_run(&fixtures::RUN_1).await.unwrap() + } + + fn test_run_record() -> RunRecord { + RunRecord { + run_id: fixtures::RUN_1, + settings: Settings::default(), + graph: Graph::new("test"), + workflow_slug: Some("test".to_string()), + working_directory: PathBuf::from("/tmp/test"), + host_repo_path: None, + repo_origin_url: None, + base_branch: None, + labels: HashMap::new(), + provenance: None, + } + } + + #[tokio::test] + async fn local_handle_loads_state_and_events() { + let run_store = test_run_store().await; + let record = test_run_record(); + append_event( + &run_store, + &fixtures::RUN_1, + &Event::RunCreated { + run_id: fixtures::RUN_1, + settings: serde_json::to_value(&record.settings).unwrap(), + graph: serde_json::to_value(&record.graph).unwrap(), + workflow_source: Some("digraph test {}".to_string()), + workflow_config: None, + labels: std::collections::BTreeMap::new(), + run_dir: "/tmp/test".to_string(), + working_directory: "/tmp/test".to_string(), + host_repo_path: None, + repo_origin_url: None, + base_branch: None, + workflow_slug: Some("test".to_string()), + db_prefix: None, + provenance: None, + }, + ) + .await + .unwrap(); + + let handle = RunStoreHandle::local(run_store); + let state = handle.state().await.unwrap(); + let events = handle.list_events().await.unwrap(); + + assert_eq!(state.run.unwrap().workflow_slug.as_deref(), Some("test")); + assert_eq!(events.len(), 1); + } + + #[tokio::test] + async fn local_handle_appends_events_and_roundtrips_blobs() { + let run_store = test_run_store().await; + let handle = RunStoreHandle::local(run_store); + + let event = RunEvent { + id: "evt-run-submitted".to_string(), + ts: Utc::now(), + run_id: fixtures::RUN_1, + node_id: None, + node_label: None, + session_id: None, + parent_session_id: None, + body: EventBody::RunSubmitted(RunStatusTransitionProps { reason: None }), + }; + handle.append_run_event(&event).await.unwrap(); + + let blob_id = handle.write_blob(br#"{"ok":true}"#).await.unwrap(); + let blob = handle.read_blob(&blob_id).await.unwrap().unwrap(); + let events = handle.list_events().await.unwrap(); + + assert_eq!(events.len(), 1); + assert_eq!(blob.as_ref(), br#"{"ok":true}"#); + } +} diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index a38f3069b..c572d9af4 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -106,7 +106,7 @@ async fn initialized( run_options: run_options.clone(), workflow_path: None, workflow_bundle: None, - run_store, + run_store: run_store.into(), checkpoint: options.checkpoint, seed_context: None, emitter,