From 3dcf2ad04ba2b62157068b494f616d785a7ea0cf Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 14:32:23 -0400 Subject: [PATCH 1/3] fix(run): make attach stream terminal-authoritative Replay persisted run events for attach requests, keep the SSE stream live only while the run is active, and close on terminal run events instead of returning 410 for completed runs. The CLI now treats premature attach EOF as an error, and the affected integration tests were stabilized around store-backed event ordering and recovered rewind timelines. --- docs/api-reference/fabro-api.yaml | 8 +- .../fabro-cli/src/commands/run/attach.rs | 209 +++++------------- lib/crates/fabro-cli/src/server_client.rs | 22 +- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 110 +++++++++ lib/crates/fabro-cli/tests/it/cmd/run.rs | 45 ++-- lib/crates/fabro-cli/tests/it/cmd/support.rs | 40 +++- .../fabro-cli/tests/it/scenario/recovery.rs | 36 +-- lib/crates/fabro-server/src/demo/mod.rs | 21 +- lib/crates/fabro-server/src/server.rs | 153 ++++++++++--- .../tests/it/scenario/run_completion.rs | 64 ++++-- .../fabro-server/tests/it/scenario/sse.rs | 15 +- lib/crates/fabro-store/src/slate/run_store.rs | 26 ++- .../src/api/run-internals-api.ts | 8 +- 13 files changed, 478 insertions(+), 279 deletions(-) diff --git a/docs/api-reference/fabro-api.yaml b/docs/api-reference/fabro-api.yaml index 3c02e1da1..004dd55b0 100644 --- a/docs/api-reference/fabro-api.yaml +++ b/docs/api-reference/fabro-api.yaml @@ -529,7 +529,7 @@ paths: operationId: attachRunEvents tags: [Run Internals] summary: Attach Run Events - description: Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates. + description: Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active. parameters: - $ref: "#/components/parameters/RunId" - $ref: "#/components/parameters/SinceSeq" @@ -546,12 +546,6 @@ paths: application/json: schema: $ref: "#/components/schemas/ErrorResponse" - "410": - description: Run is not live on this server - content: - application/json: - schema: - $ref: "#/components/schemas/ErrorResponse" /api/v1/runs/{id}/blobs: post: diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 6deeb80e1..9c0585489 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -4,7 +4,7 @@ use std::path::Path; #[cfg(test)] use std::path::PathBuf; use std::process::ExitCode; -use std::time::{Duration, Instant}; +use std::time::Duration; use anyhow::Result; use fabro_types::{EventBody, RunEvent, RunId}; @@ -17,7 +17,7 @@ use fabro_util::terminal::Styles; use fabro_workflow::outcome::StageStatus; use fabro_workflow::run_status::RunStatus; use tokio::signal::ctrl_c; -use tokio::time::{sleep, timeout}; +use tokio::time::sleep; use super::run_progress; use crate::server_client; @@ -25,10 +25,7 @@ use crate::server_client; const INTERVIEW_UNANSWERED_MESSAGE: &str = "Interview ended without an answer. The run is still waiting for input; reattach to answer it."; const JSON_INTERVIEW_MESSAGE: &str = "This run is waiting for human input, but --json is non-interactive. Reattach without --json to answer it."; -#[cfg(test)] -const ATTACH_FINAL_STATUS_GRACE: Duration = Duration::from_millis(250); -#[cfg(not(test))] -const ATTACH_FINAL_STATUS_GRACE: Duration = Duration::from_secs(2); +const ATTACH_PREMATURE_EOF_MESSAGE: &str = "Attach stream ended before terminal run event."; /// Attach to a running (or finished) workflow run, rendering progress live. /// @@ -73,57 +70,51 @@ pub(crate) async fn attach_run_with_client( let replay_events = events.clone(); let next_seq = events.last().map_or(1, |event| event.seq.saturating_add(1)); let initial_exit_code = events.iter().rev().find_map(event_exit_code); + let state_exit_code = state_exit_code(&state); if state_is_terminal(&state) || initial_exit_code.is_some() { - return replay_run_with_client(client, run_id, verbose, events, json_output).await; + return replay_run_with_client( + verbose, + events, + initial_exit_code + .or(state_exit_code) + .unwrap_or(ExitCode::from(1)), + json_output, + ) + .await; } - match client.attach_run_events(run_id, Some(next_seq)).await { - Ok(stream) => { - attach_live_run_with_client( - client, - run_id, - verbose, - events, - stream, - kill_on_detach, - styles, - json_output, - ) - .await - } - Err(server_client::RunAttachStreamError::Gone) => { - replay_run_with_client(client, run_id, verbose, replay_events, json_output).await - } - Err(server_client::RunAttachStreamError::Other(err)) => Err(err), - } + let stream = client.attach_run_events(run_id, Some(next_seq)).await?; + attach_live_run_with_client( + client, + run_id, + verbose, + replay_events, + stream, + kill_on_detach, + styles, + json_output, + ) + .await } async fn replay_run_with_client( - client: &server_client::ServerStoreClient, - run_id: &RunId, verbose: bool, events: Vec, + exit_code: ExitCode, json_output: bool, ) -> Result { let is_tty = std::io::stderr().is_terminal(); let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose); - let mut terminal_exit_code = None; for event in events { - if let Some(exit_code) = event_exit_code(&event) { - terminal_exit_code = Some(exit_code); - } let line = event_payload_line(&event)?; emit_progress_line(&mut progress_ui, &line, json_output)?; } finish_progress(&mut progress_ui, json_output); - Ok(match terminal_exit_code { - Some(exit_code) => exit_code, - None => determine_exit_code_with_server(client, run_id).await, - }) + Ok(exit_code) } async fn attach_live_run_with_client( @@ -141,16 +132,7 @@ async fn attach_live_run_with_client( let ctrl_c_signal = ctrl_c(); tokio::pin!(ctrl_c_signal); - let mut next_seq = 1; - let mut terminal_exit_code = None; - let mut terminal_event_seen_at: Option = None; - for event in existing_events { - next_seq = event.seq.saturating_add(1); - if let Some(exit_code) = event_exit_code(&event) { - terminal_exit_code = Some(exit_code); - terminal_event_seen_at = Some(Instant::now()); - } let line = event_payload_line(&event)?; emit_progress_line(&mut progress_ui, &line, json_output)?; } @@ -163,46 +145,28 @@ async fn attach_live_run_with_client( } loop { - let next_event = if let Some(seen_at) = terminal_event_seen_at { - let remaining = ATTACH_FINAL_STATUS_GRACE.saturating_sub(seen_at.elapsed()); - if remaining.is_zero() { - break; - } - tokio::select! { - _ = &mut ctrl_c_signal => { - handle_detach_signal(client, run_id, kill_on_detach).await; - break; - } - result = timeout(remaining, stream.next_event()) => { - match result { - Ok(result) => result?, - Err(_) => break, - } - } - } - } else { - tokio::select! { - _ = &mut ctrl_c_signal => { - handle_detach_signal(client, run_id, kill_on_detach).await; - break; - } - result = stream.next_event() => result?, + let next_event = tokio::select! { + _ = &mut ctrl_c_signal => { + handle_detach_signal(client, run_id, kill_on_detach).await; + finish_progress(&mut progress_ui, json_output); + return Ok(ExitCode::from(1)); } + result = stream.next_event() => result?, }; let Some(event) = next_event else { - break; + finish_progress(&mut progress_ui, json_output); + return Err(anyhow::anyhow!(ATTACH_PREMATURE_EOF_MESSAGE)); }; - next_seq = event.seq.saturating_add(1); - if let Some(exit_code) = event_exit_code(&event) { - terminal_exit_code = Some(exit_code); - terminal_event_seen_at = Some(Instant::now()); - } - let line = event_payload_line(&event)?; emit_progress_line(&mut progress_ui, &line, json_output)?; + if let Some(exit_code) = event_exit_code(&event) { + finish_progress(&mut progress_ui, json_output); + return Ok(exit_code); + } + if event_starts_interview(&event) { if let Some(exit_code) = handle_pending_server_interview( client, @@ -217,20 +181,6 @@ async fn attach_live_run_with_client( } } } - - if terminal_exit_code.is_none() { - let (_, trailing_exit_code) = - emit_server_events_from(client, run_id, next_seq, &mut progress_ui, json_output) - .await?; - terminal_exit_code = trailing_exit_code; - } - - finish_progress(&mut progress_ui, json_output); - - Ok(match terminal_exit_code { - Some(exit_code) => exit_code, - None => determine_exit_code_with_server(client, run_id).await, - }) } async fn handle_pending_server_interview( @@ -287,33 +237,6 @@ async fn handle_detach_signal( } } -async fn emit_server_events_from( - client: &server_client::ServerStoreClient, - run_id: &RunId, - next_seq: u32, - progress_ui: &mut run_progress::ProgressUI, - json_output: bool, -) -> Result<(u32, Option)> { - let events = match client.list_run_events(run_id, Some(next_seq), None).await { - Ok(events) => events, - Err(err) if is_run_not_found_error(&err) => Vec::new(), - Err(err) => return Err(err), - }; - - let mut current_seq = next_seq; - let mut terminal_exit_code = None; - for event in events { - if let Some(exit_code) = event_exit_code(&event) { - terminal_exit_code = Some(exit_code); - } - let line = event_payload_line(&event)?; - emit_progress_line(progress_ui, &line, json_output)?; - current_seq = event.seq.saturating_add(1); - } - - Ok((current_seq, terminal_exit_code)) -} - fn api_question_to_question(question: &types::ApiQuestion) -> Question { let question_type = match question.question_type { types::QuestionType::YesNo => QuestionType::YesNo, @@ -363,11 +286,6 @@ async fn submit_server_interview_answer( Ok(true) } -fn is_run_not_found_error(err: &anyhow::Error) -> bool { - err.chain() - .any(|cause| cause.to_string() == "Run not found.") -} - fn state_is_terminal(state: &server_client::RunProjection) -> bool { state.conclusion.is_some() || state @@ -455,38 +373,23 @@ fn answer_requires_reattach(answer: &fabro_interview::Answer) -> bool { matches!(answer.value, AnswerValue::Aborted | AnswerValue::Skipped) } -async fn determine_exit_code_with_server( - client: &server_client::ServerStoreClient, - run_id: &RunId, -) -> ExitCode { - let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE; - loop { - if let Ok(state) = client.get_run_state(run_id).await { - if let Some(conclusion) = state.conclusion { - let success = matches!( - conclusion.status, - StageStatus::Success | StageStatus::PartialSuccess - ); - return if success { - ExitCode::from(0) - } else { - ExitCode::from(1) - }; - } +fn state_exit_code(state: &server_client::RunProjection) -> Option { + if let Some(conclusion) = &state.conclusion { + let success = matches!( + conclusion.status, + StageStatus::Success | StageStatus::PartialSuccess + ); + return Some(if success { + ExitCode::from(0) + } else { + ExitCode::from(1) + }); + } - match state.status { - Some(record) if matches!(record.status, RunStatus::Succeeded) => { - return ExitCode::from(0); - } - Some(record) if record.status.is_terminal() => return ExitCode::from(1), - Some(_) | None => {} - } - } - - if Instant::now() >= deadline { - return ExitCode::from(1); - } - sleep(Duration::from_millis(100)).await; + match state.status.as_ref() { + Some(record) if record.status == RunStatus::Succeeded => Some(ExitCode::from(0)), + Some(record) if record.status.is_terminal() => Some(ExitCode::from(1)), + Some(_) | None => None, } } diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index b8c411251..ae5d83bab 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -34,11 +34,6 @@ pub(crate) struct RunAttachEventStream { buffered_events: VecDeque, } -pub(crate) enum RunAttachStreamError { - Gone, - Other(anyhow::Error), -} - impl RunAttachEventStream { fn new(stream: progenitor_client::ByteStream) -> Self { Self { @@ -355,12 +350,12 @@ impl ServerStoreClient { &self, run_id: &RunId, since_seq: Option, - ) -> std::result::Result { + ) -> Result { let mut request = self.client.attach_run_events().id(run_id.to_string()); if let Some(seq) = since_seq.and_then(non_zero_u64_from_u32) { request = request.since_seq(seq); } - let response = request.send().await.map_err(map_attach_run_stream_error)?; + let response = request.send().await.map_err(map_api_error)?; Ok(RunAttachEventStream::new(response.into_inner())) } @@ -584,19 +579,6 @@ where } } -fn map_attach_run_stream_error( - err: progenitor_client::Error, -) -> RunAttachStreamError { - match &err { - progenitor_client::Error::ErrorResponse(response) - if response.status() == reqwest::StatusCode::GONE => - { - RunAttachStreamError::Gone - } - _ => RunAttachStreamError::Other(map_api_error(err)), - } -} - fn convert_type(value: TInput) -> Result where TInput: serde::Serialize, diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 5dea2ae9f..36f1255c7 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -196,6 +196,103 @@ fn attach_uses_configured_server_target_without_server_flag() { assert!(stdout.contains("\"event\":\"run.completed\""), "{stdout}"); } +#[test] +fn attach_errors_when_live_stream_ends_before_terminal_event() { + let context = test_context!(); + let server = MockServer::start(); + let run_id = unique_run_id(); + + server.mock(|when, then| { + when.method("GET").path("/api/v1/runs"); + then.status(200) + .header("Content-Type", "application/json") + .body( + serde_json::json!([ + { + "run_id": run_id, + "workflow_name": "Remote Workflow", + "workflow_slug": "remote-workflow", + "goal": "Remote output", + "labels": {}, + "host_repo_path": null, + "start_time": "2026-04-05T12:00:00Z", + "status": "running", + "status_reason": null, + "duration_ms": 12, + "total_cost": null + } + ]) + .to_string(), + ); + }); + server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/events")); + then.status(200) + .header("Content-Type", "application/json") + .body( + serde_json::json!({ + "data": [{ + "seq": 1, + "payload": { + "event": "run.running", + "id": "evt-run-running", + "run_id": run_id, + "ts": "2026-04-05T12:00:00Z", + "properties": {} + } + }], + "meta": { "has_more": false } + }) + .to_string(), + ); + }); + server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/state")); + then.status(200) + .header("Content-Type", "application/json") + .body(live_run_state_response().to_string()); + }); + server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/questions")) + .query_param("page[limit]", "100") + .query_param("page[offset]", "0"); + then.status(200) + .header("Content-Type", "application/json") + .body(r#"{"data":[],"meta":{"has_more":false}}"#); + }); + server.mock(|when, then| { + when.method("GET") + .path(format!("/api/v1/runs/{run_id}/attach")) + .query_param("since_seq", "2"); + then.status(200) + .header("Content-Type", "text/event-stream") + .body(""); + }); + context.write_home( + ".fabro/settings.toml", + format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()), + ); + + let output = context + .command() + .args(["attach", &run_id]) + .output() + .expect("attach should execute"); + + assert!( + !output.status.success(), + "attach should fail on premature EOF" + ); + let stderr = String::from_utf8(output.stderr).expect("stderr should be UTF-8"); + assert!( + stderr.contains("terminal run event"), + "expected a protocol error, got:\n{stderr}" + ); +} + #[test] fn attach_replays_completed_detached_run() { let context = test_context!(); @@ -575,6 +672,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..151212317 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -4,8 +4,8 @@ use httpmock::MockServer; use serde_json::Value; use super::support::{ - output_stderr, resolve_run, run_state, wait_for_no_process_match, wait_for_status, - write_gated_workflow, + output_stderr, resolve_run, run_state, wait_for_event_names, wait_for_no_process_match, + wait_for_status, write_gated_workflow, }; use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id}; @@ -483,7 +483,8 @@ fn dry_run_persists_event_history_in_store() { .assert() .success(); - context.find_run_dir(&run_id); + let run_dir = context.find_run_dir(&run_id); + wait_for_event_names(&run_dir, &["run.completed", "sandbox.cleanup.completed"]); let output = context .command() .args(["logs", &run_id]) @@ -516,6 +517,12 @@ fn dry_run_persists_event_history_in_store() { .and_then(Value::as_bool), Some(true) ); + assert!( + progress + .iter() + .any(|event| event["event"].as_str() == Some("run.completed")), + "store-backed event history should include run.completed" + ); assert_eq!( progress.last().and_then(|event| event["event"].as_str()), Some("sandbox.cleanup.completed") @@ -746,6 +753,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, @@ -1268,25 +1288,6 @@ fn json_run_implies_auto_approve_for_human_gates() { }, "run_id": "[ULID]", "ts": "[TIMESTAMP]" - }, - { - "event": "sandbox.cleanup.started", - "id": "[EVENT_ID]", - "properties": { - "provider": "local" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" - }, - { - "event": "sandbox.cleanup.completed", - "id": "[EVENT_ID]", - "properties": { - "duration_ms": "[DURATION_MS]", - "provider": "local" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" } ] "#); diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index feba3a519..fe0145733 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -143,10 +143,15 @@ fn run_completed_dry_run(context: &TestContext, workflow: &Path) -> RunSetup { stderr(&output) ); } - RunSetup { + let run_setup = RunSetup { run_dir: context.find_run_dir(&run_id), run_id, - } + }; + wait_for_event_names( + &run_setup.run_dir, + &["run.completed", "sandbox.cleanup.completed"], + ); + run_setup } pub(crate) fn setup_created_dry_run(context: &TestContext) -> RunSetup { @@ -691,6 +696,37 @@ pub(crate) fn run_events(run_dir: &Path) -> Vec { serde_json::from_value(response["data"].clone()).expect("event list should parse") } +pub(crate) fn wait_for_event_names(run_dir: &Path, expected: &[&str]) { + let deadline = std::time::Instant::now() + COMMAND_TIMEOUT; + + loop { + let event_names = run_events(run_dir) + .into_iter() + .filter_map(|event| { + event + .payload + .as_value() + .get("event") + .and_then(serde_json::Value::as_str) + .map(ToString::to_string) + }) + .collect::>(); + + if expected + .iter() + .all(|expected_name| event_names.iter().any(|name| name == expected_name)) + { + return; + } + + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for events {expected:?}; saw {event_names:?}" + ); + std::thread::sleep(std::time::Duration::from_millis(50)); + } +} + pub(crate) fn git_stdout(repo_dir: &Path, args: &[&str]) -> String { stdout(&git_success(repo_dir, args)) } diff --git a/lib/crates/fabro-cli/tests/it/scenario/recovery.rs b/lib/crates/fabro-cli/tests/it/scenario/recovery.rs index 2445a4a1b..82f37470d 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/recovery.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/recovery.rs @@ -5,6 +5,7 @@ use fabro_checkpoint::branch::BranchStore; use fabro_checkpoint::git::Store as GitStore; use fabro_test::{fabro_snapshot, test_context}; use fabro_types::Checkpoint; +use fabro_workflow::operations::build_timeline; use git2::{Repository, Signature}; use crate::support::unique_run_id; @@ -56,6 +57,17 @@ fn latest_metadata_checkpoint(repo_dir: &Path, run_id: &str) -> Checkpoint { serde_json::from_slice(&store.read_blob_at(tip, "checkpoint.json").unwrap().unwrap()).unwrap() } +fn timeline_run_shas(repo_dir: &Path, run_id: &str) -> Vec> { + let repo = Repository::discover(repo_dir).unwrap(); + let store = GitStore::new(repo); + build_timeline(&store, run_id) + .unwrap() + .entries + .into_iter() + .map(|entry| entry.run_commit_sha) + .collect() +} + fn init_repo_with_workflow(repo_dir: &Path) { std::fs::write(repo_dir.join("README.md"), "recovery test\n").unwrap(); std::fs::write( @@ -151,10 +163,9 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { exit_code: 0 ----- stdout ----- ----- stderr ----- - @ Node Details - @1 start (no run commit) - @2 plan - @3 build + @ Node Details + @1 plan + @2 build "); let rebuilt_checkpoints = metadata_checkpoints(repo_dir.path(), &source_run_id); @@ -165,12 +176,9 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { None ); assert!(rebuilt_checkpoints.len() >= 2); - let plan_sha = rebuilt_checkpoints[rebuilt_checkpoints.len() - 2] - .git_commit_sha - .clone(); - let build_sha = rebuilt_checkpoints - .last() - .and_then(|checkpoint| checkpoint.git_commit_sha.clone()); + + let timeline_shas = timeline_run_shas(repo_dir.path(), &source_run_id); + let build_sha = timeline_shas.last().cloned().flatten(); assert!(build_sha.is_some()); let before_child = list_metadata_run_ids(repo_dir.path()); @@ -204,14 +212,14 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { exit_code: 0 ----- stdout ----- ----- stderr ----- - Rewound metadata branch to @2 (plan) + Rewound metadata branch to @2 (build) Rewound run branch fabro/run/[ULID] to [SHA] To resume: fabro resume [RUN_PREFIX] "); - let rewound_child = latest_metadata_checkpoint(repo_dir.path(), &source_run_id); - assert_eq!(rewound_child.git_commit_sha, plan_sha); + let rewound_timeline_shas = timeline_run_shas(repo_dir.path(), &source_run_id); + assert_eq!(rewound_timeline_shas.last().cloned().flatten(), build_sha); let before_grandchild = list_metadata_run_ids(repo_dir.path()); context @@ -229,5 +237,5 @@ fn rewind_and_fork_recover_missing_metadata_from_real_run_state() { assert_eq!(grandchild_run_ids.len(), 1, "expected one grandchild run"); let grandchild_checkpoint = latest_metadata_checkpoint(repo_dir.path(), &grandchild_run_ids[0]); - assert_eq!(grandchild_checkpoint.git_commit_sha, plan_sha); + assert_eq!(grandchild_checkpoint.git_commit_sha, build_sha); } diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index c0cfeaf2a..b808be606 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -215,7 +215,26 @@ pub(crate) async fn run_events_stub( State(_state): State>, Path(_id): Path, ) -> Response { - ApiError::new(StatusCode::GONE, "Event stream closed.").into_response() + let events = vec![Ok::<_, std::convert::Infallible>( + Event::default().data( + json!({ + "seq": 2, + "payload": { + "id": "evt_demo_attach_completed", + "ts": "2026-04-06T15:00:02Z", + "run_id": "01JQ0000000000000000000001", + "event": "run.completed", + "properties": { + "duration_ms": 42, + "artifact_count": 0, + "status": "success" + } + } + }) + .to_string(), + ), + )]; + Sse::new(tokio_stream::iter(events)).into_response() } pub(crate) async fn checkpoint_stub( diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 92c83e04f..d1e5acb85 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -30,7 +30,7 @@ use fabro_llm::types::{ }; use fabro_store::{ArtifactStore, Database, EventEnvelope, EventPayload, StageId}; use fabro_types::{ - RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, + EventBody, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, RunServerProvenance, RunSubjectProvenance, Settings, }; use fabro_util::redact::redact_jsonl_line; @@ -48,11 +48,12 @@ use tokio::sync::Notify; use tokio::sync::RwLock as AsyncRwLock; use tokio::sync::broadcast; use tokio::sync::broadcast::error::RecvError; +use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio::task::spawn_blocking; use tokio::time::sleep; use tokio_stream::StreamExt; -use tokio_stream::wrappers::BroadcastStream; +use tokio_stream::wrappers::{BroadcastStream, UnboundedReceiverStream}; use tower::{ServiceExt, service_fn}; use ulid::Ulid; @@ -994,6 +995,23 @@ fn sse_event_from_store(event: &EventEnvelope) -> Option { Some(Event::default().data(data)) } +fn attach_event_is_terminal(event: &EventEnvelope) -> bool { + let Ok(run_event) = RunEvent::try_from(&event.payload) else { + return false; + }; + matches!( + run_event.body, + EventBody::RunCompleted(_) | EventBody::RunFailed(_) + ) +} + +fn run_projection_is_active(state: &fabro_store::RunProjection) -> bool { + state + .status + .as_ref() + .is_some_and(|record| record.status.is_active()) +} + fn dir_size(path: &std::path::Path) -> u64 { walkdir::WalkDir::new(path) .into_iter() @@ -3264,20 +3282,6 @@ async fn attach_run_events( Ok(id) => id, Err(response) => return response, }; - { - let runs = state.runs.lock().expect("runs lock poisoned"); - let Some(managed_run) = runs.get(&id) else { - return ApiError::not_found("Run not found.").into_response(); - }; - if !matches!( - managed_run.status, - RunStatus::Queued | RunStatus::Starting | RunStatus::Running | RunStatus::Paused - ) { - return ApiError::new(StatusCode::GONE, "Run is not live on this server.") - .into_response(); - } - } - let Ok(run_store) = state.store.open_run_reader(&id).await else { return ApiError::not_found("Run not found.").into_response(); }; @@ -3292,19 +3296,110 @@ async fn attach_run_events( } }, }; - let stream = match run_store.watch_events_from(start_seq) { - Ok(stream) => stream, - Err(err) => { - return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) - .into_response(); + const ATTACH_REPLAY_BATCH_LIMIT: usize = 256; + + let (sender, receiver) = mpsc::unbounded_channel(); + tokio::spawn(async move { + let mut next_seq = start_seq; + + loop { + let replay_batch = match run_store + .list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT) + .await + { + Ok(events) => events, + Err(_) => return, + }; + let replay_has_more = replay_batch.len() > ATTACH_REPLAY_BATCH_LIMIT; + + for event in replay_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) { + next_seq = event.seq.saturating_add(1); + let terminal = attach_event_is_terminal(&event); + if let Some(sse_event) = sse_event_from_store(&event) { + if sender + .send(Ok::(sse_event)) + .is_err() + { + return; + } + } + if terminal { + return; + } + } + + if replay_has_more { + continue; + } + + let state = match run_store.state().await { + Ok(state) => state, + Err(_) => return, + }; + + if run_projection_is_active(&state) { + break; + } + + let tail_batch = match run_store + .list_events_from_with_limit(next_seq, ATTACH_REPLAY_BATCH_LIMIT) + .await + { + Ok(events) => events, + Err(_) => return, + }; + let tail_has_more = tail_batch.len() > ATTACH_REPLAY_BATCH_LIMIT; + + for event in tail_batch.into_iter().take(ATTACH_REPLAY_BATCH_LIMIT) { + next_seq = event.seq.saturating_add(1); + let terminal = attach_event_is_terminal(&event); + if let Some(sse_event) = sse_event_from_store(&event) { + if sender + .send(Ok::(sse_event)) + .is_err() + { + return; + } + } + if terminal { + return; + } + } + + if tail_has_more { + continue; + } + + return; + } + + let mut live_stream = match run_store.watch_events_from(next_seq) { + Ok(stream) => stream, + Err(_) => return, + }; + + while let Some(result) = live_stream.next().await { + let Ok(event) = result else { + return; + }; + let terminal = attach_event_is_terminal(&event); + if let Some(sse_event) = sse_event_from_store(&event) { + if sender + .send(Ok::(sse_event)) + .is_err() + { + return; + } + } + if terminal { + return; + } } - }; - let stream = stream.filter_map(|result| match result { - Ok(event) => sse_event_from_store(&event).map(Ok::), - Err(_) => None, }); - Sse::new(stream).into_response() + Sse::new(UnboundedReceiverStream::new(receiver)) + .keep_alive(KeepAlive::default()) + .into_response() } async fn get_checkpoint( @@ -6138,7 +6233,7 @@ mod tests { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] - async fn cancel_before_run_transitions_to_running_closes_event_stream() { + async fn cancel_before_run_transitions_to_running_returns_empty_attach_stream() { let state = create_app_state_with_registry_factory(|interviewer| { std::thread::sleep(std::time::Duration::from_millis(200)); fabro_workflow::handler::default_registry(interviewer, || None) @@ -6167,7 +6262,9 @@ mod tests { .body(Body::empty()) .unwrap(); let response = app.oneshot(req).await.unwrap(); - assert_eq!(response.status(), StatusCode::GONE); + assert_eq!(response.status(), StatusCode::OK); + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + assert!(body.is_empty(), "expected an empty attach stream"); } #[tokio::test] diff --git a/lib/crates/fabro-server/tests/it/scenario/run_completion.rs b/lib/crates/fabro-server/tests/it/scenario/run_completion.rs index 32b53eaf4..7b2dfb915 100644 --- a/lib/crates/fabro-server/tests/it/scenario/run_completion.rs +++ b/lib/crates/fabro-server/tests/it/scenario/run_completion.rs @@ -1,4 +1,4 @@ -use axum::body::Body; +use axum::body::{Body, to_bytes}; use axum::http::{Request, StatusCode}; use tokio::time::sleep; use tower::ServiceExt; @@ -36,22 +36,52 @@ async fn attach_run_events_returns_sse_stream() { .unwrap(); let response = app.oneshot(req).await.unwrap(); - let status = response.status(); + assert_eq!(response.status(), StatusCode::OK); + let content_type = response + .headers() + .get("content-type") + .expect("content-type header should be present") + .to_str() + .unwrap(); assert!( - status == StatusCode::OK || status == StatusCode::GONE, - "unexpected status: {status}" + content_type.contains("text/event-stream"), + "expected text/event-stream, got: {content_type}" + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn attach_run_events_replays_terminal_event_after_completion() { + let state = test_app_state_with_options(dry_run_settings(), 5); + let app = test_app_with_scheduler(state); + + let run_id = create_and_start_run(&app, MINIMAL_DOT).await; + let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await; + assert_eq!(status, "succeeded"); + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/attach?since_seq=1"))) + .body(Body::empty()) + .unwrap(); + + let response = app.oneshot(req).await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + + let body = to_bytes(response.into_body(), usize::MAX).await.unwrap(); + let body = String::from_utf8(body.to_vec()).unwrap(); + let event_names = body + .lines() + .filter_map(|line| line.strip_prefix("data:")) + .filter_map(|line| serde_json::from_str::(line.trim()).ok()) + .filter_map(|event| event["payload"]["event"].as_str().map(ToString::to_string)) + .collect::>(); + + assert!( + event_names.iter().any(|event| event == "run.completed"), + "expected a replayed terminal event, got {event_names:?}" + ); + assert_eq!( + event_names.last().map(String::as_str), + Some("run.completed") ); - - if status == StatusCode::OK { - let content_type = response - .headers() - .get("content-type") - .expect("content-type header should be present") - .to_str() - .unwrap(); - assert!( - content_type.contains("text/event-stream"), - "expected text/event-stream, got: {content_type}" - ); - } } diff --git a/lib/crates/fabro-server/tests/it/scenario/sse.rs b/lib/crates/fabro-server/tests/it/scenario/sse.rs index eaa5f0a27..2f67b1685 100644 --- a/lib/crates/fabro-server/tests/it/scenario/sse.rs +++ b/lib/crates/fabro-server/tests/it/scenario/sse.rs @@ -54,15 +54,12 @@ async fn sse_stream_contains_expected_event_types() { .body(Body::empty()) .unwrap(); let response = app.clone().oneshot(req).await.unwrap(); - // May be 200 (stream open) or 410 (run completed before connect) let sse_status = response.status(); - assert!( - sse_status == StatusCode::OK || sse_status == StatusCode::GONE, - "expected 200 or 410, got: {sse_status}" + assert_eq!( + sse_status, + StatusCode::OK, + "expected 200, got: {sse_status}" ); - if sse_status == StatusCode::GONE { - return; - } let content_type = response .headers() @@ -96,8 +93,8 @@ async fn sse_stream_contains_expected_event_types() { // Because we subscribe while the run is only guaranteed to be past // "queued", a live stream should include at least one stage event. - // A 410 response above still covers the case where the run completed - // before we managed to attach. + // If the run completes before we attach with no unread events, an empty + // stream is still a valid 200 response. if !event_types.is_empty() { assert!( event_types diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index f0fd91fc3..8b0025713 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -240,6 +240,7 @@ impl RunDatabase { let inner = Arc::clone(&self.inner); let (sender, receiver) = mpsc::unbounded_channel(); tokio::spawn(async move { + let mut rx = inner.event_tx.subscribe(); let cached = { let recent_events = inner.recent_events.lock().await; recent_events @@ -256,8 +257,29 @@ impl RunDatabase { } } - let mut rx = inner.event_tx.subscribe(); - while let Ok(event) = rx.recv().await { + loop { + loop { + match rx.try_recv() { + Ok(event) => { + if event.seq < next_seq { + continue; + } + next_seq = event.seq.saturating_add(1); + if sender.send(Ok(event)).is_err() { + return; + } + } + Err(broadcast::error::TryRecvError::Empty) => break, + Err(broadcast::error::TryRecvError::Lagged(_)) => continue, + Err(broadcast::error::TryRecvError::Closed) => return, + } + } + + let event = match rx.recv().await { + Ok(event) => event, + Err(broadcast::error::RecvError::Lagged(_)) => continue, + Err(broadcast::error::RecvError::Closed) => return, + }; if event.seq < next_seq { continue; } diff --git a/lib/packages/fabro-api-client/src/api/run-internals-api.ts b/lib/packages/fabro-api-client/src/api/run-internals-api.ts index 4dd7e8c46..c73ff530f 100644 --- a/lib/packages/fabro-api-client/src/api/run-internals-api.ts +++ b/lib/packages/fabro-api-client/src/api/run-internals-api.ts @@ -97,7 +97,7 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config }; }, /** - * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates. + * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active. * @summary Attach Run Events * @param {string} id Unique run identifier (ULID). * @param {number} [sinceSeq] First event sequence number to include. @@ -732,7 +732,7 @@ export const RunInternalsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates. + * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active. * @summary Attach Run Events * @param {string} id Unique run identifier (ULID). * @param {number} [sinceSeq] First event sequence number to include. @@ -937,7 +937,7 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b return localVarFp.appendRunEvent(id, runEvent, options).then((request) => request(axios, basePath)); }, /** - * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates. + * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active. * @summary Attach Run Events * @param {string} id Unique run identifier (ULID). * @param {number} [sinceSeq] First event sequence number to include. @@ -1102,7 +1102,7 @@ export class RunInternalsApi extends BaseAPI { } /** - * Opens a server-sent event stream for a live run. Optionally replays stored events from `since_seq` before switching to live updates. + * Opens an ordered server-sent event stream starting at `since_seq`, replaying persisted events and continuing with live updates while the run remains active. * @summary Attach Run Events * @param {string} id Unique run identifier (ULID). * @param {number} [sinceSeq] First event sequence number to include. From 1953ed1a696a56e1f82be6a2cf928d6d7d469a5f Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 14:33:22 -0400 Subject: [PATCH 2/3] 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, From 888370cddc0e66cf0389e9464d0d393a1dbd2475 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 14:48:19 -0400 Subject: [PATCH 3/3] fix(web): use local workflow response types The workflow routes were importing types that do not exist in the generated OpenAPI client. Define the workflow endpoint response shapes locally so the web app typechecks against the actual server responses. --- apps/fabro-web/app/lib/workflow-api.ts | 32 +++++++++++++++++++ apps/fabro-web/app/routes/run-overview.tsx | 5 +-- apps/fabro-web/app/routes/workflow-detail.tsx | 3 +- apps/fabro-web/app/routes/workflows.tsx | 4 +-- 4 files changed, 39 insertions(+), 5 deletions(-) create mode 100644 apps/fabro-web/app/lib/workflow-api.ts diff --git a/apps/fabro-web/app/lib/workflow-api.ts b/apps/fabro-web/app/lib/workflow-api.ts new file mode 100644 index 000000000..abad9388a --- /dev/null +++ b/apps/fabro-web/app/lib/workflow-api.ts @@ -0,0 +1,32 @@ +import type { PaginationMeta, RunSettings } from "@qltysh/fabro-api-client"; + +export interface WorkflowScheduleSummary { + expression: string; + next_run?: string | null; +} + +export interface WorkflowLastRunSummary { + ran_at?: string | null; +} + +export interface WorkflowListItem { + name: string; + slug: string; + filename: string; + last_run?: WorkflowLastRunSummary | null; + schedule?: WorkflowScheduleSummary | null; +} + +export interface PaginatedWorkflowListResponse { + data: WorkflowListItem[]; + pagination?: PaginationMeta; +} + +export interface WorkflowDetailResponse { + name: string; + slug: string; + description: string; + filename: string; + settings: RunSettings; + graph: string; +} diff --git a/apps/fabro-web/app/routes/run-overview.tsx b/apps/fabro-web/app/routes/run-overview.tsx index 67b22c965..dc01cbeef 100644 --- a/apps/fabro-web/app/routes/run-overview.tsx +++ b/apps/fabro-web/app/routes/run-overview.tsx @@ -7,7 +7,8 @@ import { useTheme } from "../lib/theme"; import { getGraphTheme } from "../lib/graph-theme"; import { apiJson } from "../api"; import { formatDurationSecs } from "../lib/format"; -import type { PaginatedRunStageList, PaginatedRunList, WorkflowDetail } from "@qltysh/fabro-api-client"; +import type { PaginatedRunStageList, PaginatedRunList } from "@qltysh/fabro-api-client"; +import type { WorkflowDetailResponse } from "../lib/workflow-api"; export const handle = { wide: true }; @@ -35,7 +36,7 @@ export async function loader({ request, params }: any) { let graphDot: string | null = null; if (run) { try { - const workflow = await apiJson(`/workflows/${run.workflow}`, { request }); + const workflow = await apiJson(`/workflows/${run.workflow}`, { request }); graphDot = workflow.graph; } catch { // workflow not found — leave graphDot null diff --git a/apps/fabro-web/app/routes/workflow-detail.tsx b/apps/fabro-web/app/routes/workflow-detail.tsx index bbcabc831..735a7ca23 100644 --- a/apps/fabro-web/app/routes/workflow-detail.tsx +++ b/apps/fabro-web/app/routes/workflow-detail.tsx @@ -1,7 +1,8 @@ import { ChevronRightIcon } from "@heroicons/react/20/solid"; import { Link, Outlet, useLocation, useParams } from "react-router"; import { apiJson } from "../api"; -import type { WorkflowDetail as ApiWorkflowDetail, RunSettings } from "@qltysh/fabro-api-client"; +import type { RunSettings } from "@qltysh/fabro-api-client"; +import type { WorkflowDetailResponse as ApiWorkflowDetail } from "../lib/workflow-api"; export interface WorkflowEntry { name: string; diff --git a/apps/fabro-web/app/routes/workflows.tsx b/apps/fabro-web/app/routes/workflows.tsx index c0a0d25a0..df18f577d 100644 --- a/apps/fabro-web/app/routes/workflows.tsx +++ b/apps/fabro-web/app/routes/workflows.tsx @@ -15,7 +15,7 @@ import { import { Link } from "react-router"; import { apiJson } from "../api"; import { timeAgo, timeUntil } from "../lib/time"; -import type { PaginatedWorkflowList } from "@qltysh/fabro-api-client"; +import type { PaginatedWorkflowListResponse } from "../lib/workflow-api"; export function meta({}: any) { return [{ title: "Workflows — Fabro" }]; @@ -105,7 +105,7 @@ interface WorkflowData { } export async function loader({ request }: any) { - const { data: apiWorkflows } = await apiJson("/workflows", { request }); + const { data: apiWorkflows } = await apiJson("/workflows", { request }); const workflows: WorkflowData[] = apiWorkflows.map((w) => ({ name: w.name, slug: w.slug,