scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-06 08:18:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge branch 'codex/fix-attach-terminal-authoritative-stream'

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-07 15:10:06 -04:00
commit 7d340909a4
31 changed files with 605 additions and 146 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -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",

1
lib/crates/fabro-cli/Cargo.toml
View file

@ -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 }

2
lib/crates/fabro-cli/src/commands/pr/create.rs
View file

@ -103,7 +103,7 @@ pub(super) async fn create_command(
&model,
true,
None,
&run_store,
&run_store.clone().into(),
None,
)
.await

247
lib/crates/fabro-cli/src/commands/run/runner.rs
View file

@ -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<Database> {
Arc::new(Database::new(
Arc::new(MemoryObjectStore::new()),
"",
STORE_FLUSH_INTERVAL,
))
#[derive(Clone)]
struct HttpRunStore {
run_id: RunId,
client: server_client::ServerStoreClient,
state: Arc<Mutex<RunProjection>>,
events: Arc<Mutex<Option<Vec<EventEnvelope>>>>,
}
async fn load_seed_run_store(
client: &server_client::ServerStoreClient,
run_id: &RunId,
) -> Result<RunDatabase> {
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::<Vec<_>>();
seed_run_store(run_id, &payloads).await
}
async fn seed_run_store(run_id: &RunId, events: &[EventPayload]) -> Result<RunDatabase> {
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<RunStoreHandle> {
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<T, F, Fut>(&self, operation: &'static str, mut op: F) -> Result<T>
where
F: FnMut() -> Fut,
Fut: std::future::Future<Output = Result<T>>,
{
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<RunProjection> {
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<RunProjection> {
Ok(self.state.lock().await.clone())
}
async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
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<RunBlobId> {
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<Option<bytes::Bytes>> {
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);
}
}

72
lib/crates/fabro-cli/src/server_client.rs
View file

@ -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;
@ -398,16 +399,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<u32> {
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<RunBlobId> {
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<Option<Bytes>> {
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<()> {
@ -579,6 +629,20 @@ where
}
}
fn is_not_found_error<E>(err: &progenitor_client::Error<E>) -> 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 convert_type<TInput, TOutput>(value: TInput) -> Result<TOutput>
where
TInput: serde::Serialize,

38
lib/crates/fabro-retro/src/retro_agent.rs
View file

@ -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<dyn Sandbox>,
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<RetroAgentResult> {
// 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<Mutex<Option<RetroNarrative>>> = Arc::new(Mutex::new(None));
@ -292,7 +293,8 @@ fn build_profile(provider: Provider, model: &str) -> Box<dyn AgentProfile> {
async fn upload_data_files(
sandbox: &Arc<dyn Sandbox>,
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<String> = 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<String> = 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?;

4
lib/crates/fabro-server/src/server.rs
View file

@ -2629,7 +2629,7 @@ async fn execute_run_in_process(state: Arc<AppState>, 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");
@ -2665,7 +2665,7 @@ async fn execute_run_in_process(state: Arc<AppState>, run_id: RunId) {
cancel_token: Some(Arc::clone(&cancel_token)),
emitter: Arc::clone(&emitter),
interviewer: Arc::clone(&interviewer) as Arc<dyn Interviewer>,
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,

4
lib/crates/fabro-store/src/run_state.rs
View file

@ -71,7 +71,7 @@ pub(crate) struct EventProjectionCache {
}
impl RunProjection {
pub(crate) fn apply_events(events: &[EventEnvelope]) -> Result<Self> {
pub fn apply_events(events: &[EventEnvelope]) -> Result<Self> {
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;

1
lib/crates/fabro-workflow/Cargo.toml
View file

@ -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

8
lib/crates/fabro-workflow/src/artifact.rs
View file

@ -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<String, Value>,
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();

20
lib/crates/fabro-workflow/src/event.rs
View file

@ -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<AsyncMutex<Pin<Box<dyn AsyncWrite + Send>>>>),
Callback(Arc<RunEventSinkCallback>),
Composite(Vec<Self>),
@ -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<RunStoreHandle>) -> Self {
Self {
inner: RunEventLogger::new(RunEventSink::store(run_store)),
inner: RunEventLogger::new(RunEventSink::backend(run_store.into())),
}
}

2
lib/crates/fabro-workflow/src/handler/agent.rs
View file

@ -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());

2
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -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());

2
lib/crates/fabro-workflow/src/handler/manager_loop.rs
View file

@ -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,

7
lib/crates/fabro-workflow/src/handler/mod.rs
View file

@ -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<HandlerRegistry>,
pub emitter: Arc<Emitter>,
pub sandbox: Arc<dyn Sandbox>,
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<Option<Arc<GitState>>>,
@ -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(),

4
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

@ -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());

2
lib/crates/fabro-workflow/src/handler/prompt.rs
View file

@ -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());

1
lib/crates/fabro-workflow/src/lib.rs
View file

@ -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;

6
lib/crates/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -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<Option<StageUsage>>;
/// Sub-lifecycle responsible for artifact collection, offloading, and syncing.
pub(crate) struct ArtifactLifecycle {
pub sandbox: Arc<dyn fabro_sandbox::Sandbox>,
pub run_store: RunDatabase,
pub run_store: RunStoreHandle,
pub blob_cache_dir: PathBuf,
pub emitter: Arc<Emitter>,
pub artifacts_dir: PathBuf,
@ -40,7 +40,7 @@ impl ArtifactLifecycle {
#[allow(clippy::too_many_arguments)]
pub(crate) fn new(
sandbox: Arc<dyn fabro_sandbox::Sandbox>,
run_store: RunDatabase,
run_store: RunStoreHandle,
blob_cache_dir: PathBuf,
emitter: Arc<Emitter>,
artifacts_dir: PathBuf,

4
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -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<Option<StageUsage>>;
@ -67,7 +67,7 @@ pub(crate) struct GitLifecycle {
pub emitter: Arc<Emitter>,
pub run_dir: PathBuf,
pub run_id: RunId,
pub run_store: RunDatabase,
pub run_store: RunStoreHandle,
pub run_options: Arc<RunOptions>,
pub start_node_id: Option<String>,
// Cross-lifecycle data (shared with EventLifecycle)

4
lib/crates/fabro-workflow/src/lifecycle/mod.rs
View file

@ -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<dyn Sandbox>,
graph: Arc<GvGraph>,
run_dir: &PathBuf,
run_store: &RunDatabase,
run_store: &RunStoreHandle,
run_options: &Arc<RunOptions>,
is_resume: bool,
on_node: crate::OnNodeCallback,

12
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -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<DevcontainerSpec>,
seed_context: Option<Context>,
run_store: RunDatabase,
run_store: RunStoreHandle,
event_sink: RunEventSink,
git: Option<GitCheckpointOptions>,
github_app: Option<fabro_github::GitHubAppCredentials>,
@ -70,7 +70,7 @@ pub struct StartServices {
pub cancel_token: Option<Arc<AtomicBool>>,
pub emitter: Arc<Emitter>,
pub interviewer: Arc<dyn Interviewer>,
pub run_store: RunDatabase,
pub run_store: RunStoreHandle,
pub event_sink: RunEventSink,
pub run_control: Option<Arc<RunControlState>>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
@ -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(),

6
lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -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 {

10
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -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<String>,
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,

4
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -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 {

17
lib/crates/fabro-workflow/src/pipeline/persist.rs
View file

@ -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<Persisted, FabroError> {
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(),

19
lib/crates/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -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<String, String> {
debug!("Building PR body");
@ -406,7 +407,7 @@ pub async fn maybe_open_pull_request(
model: &str,
draft: bool,
auto_merge: Option<AutoMergeOptions>,
run_store: &RunDatabase,
run_store: &RunStoreHandle,
conclusion: Option<&Conclusion>,
) -> Result<Option<PullRequestRecord>, 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"));
}

24
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -23,7 +23,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
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<Retro> {
}
})
});
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(),

18
lib/crates/fabro-workflow/src/pipeline/types.rs
View file

@ -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<Self, FabroError> {
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<Emitter>,
pub sandbox: SandboxSpec,
@ -259,7 +259,7 @@ pub struct Initialized {
pub run_options: RunOptions,
pub workflow_path: Option<PathBuf>,
pub workflow_bundle: Option<Arc<WorkflowBundle>>,
pub run_store: RunDatabase,
pub run_store: RunStoreHandle,
pub(crate) checkpoint: Option<Checkpoint>,
pub(crate) seed_context: Option<Context>,
pub emitter: Arc<Emitter>,
@ -281,7 +281,7 @@ pub struct Executed {
pub graph: Graph,
pub outcome: Result<Outcome, FabroError>,
pub run_options: RunOptions,
pub run_store: RunDatabase,
pub run_store: RunStoreHandle,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<Emitter>,
pub sandbox: Arc<dyn Sandbox>,
@ -298,7 +298,7 @@ pub struct Retroed {
pub graph: Graph,
pub outcome: Result<Outcome, FabroError>,
pub run_options: RunOptions,
pub run_store: RunDatabase,
pub run_store: RunStoreHandle,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<Emitter>,
pub sandbox: Arc<dyn Sandbox>,
@ -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<Arc<HookRunner>>,
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<PullRequestSettings>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
pub origin_url: Option<String>,

206
lib/crates/fabro-workflow/src/runtime_store.rs Normal file
View file

@ -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<RunProjection>;
async fn list_events(&self) -> Result<Vec<EventEnvelope>>;
async fn append_run_event(&self, event: &RunEvent) -> Result<()>;
async fn write_blob(&self, data: &[u8]) -> Result<RunBlobId>;
async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>>;
}
#[derive(Clone)]
pub struct RunStoreHandle {
backend: Arc<dyn RunStoreBackend>,
}
impl RunStoreHandle {
#[must_use]
pub fn new(backend: Arc<dyn RunStoreBackend>) -> 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<RunProjection> {
self.backend.load_state().await
}
pub async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
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<RunBlobId> {
self.backend.write_blob(data).await
}
pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
self.backend.read_blob(id).await
}
}
impl From<RunDatabase> 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<RunProjection> {
self.run_store.state().await.map_err(anyhow::Error::from)
}
async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
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<RunBlobId> {
self.run_store
.write_blob(data)
.await
.map_err(anyhow::Error::from)
}
async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
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}"#);
}
}

2
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -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,