diff --git a/Cargo.lock b/Cargo.lock index 5396eb041..4d1c2510d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1526,6 +1526,7 @@ dependencies = [ "async-trait", "axum", "base64", + "bytes", "chrono", "clap", "clap_complete", @@ -2068,6 +2069,7 @@ dependencies = [ "assert_cmd", "async-trait", "base64", + "bytes", "chrono", "dirs", "fabro-agent", 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 929563994..dc01cbeef 100644 --- a/apps/fabro-web/app/routes/run-overview.tsx +++ b/apps/fabro-web/app/routes/run-overview.tsx @@ -8,10 +8,7 @@ import { getGraphTheme } from "../lib/graph-theme"; import { apiJson } from "../api"; import { formatDurationSecs } from "../lib/format"; import type { PaginatedRunStageList, PaginatedRunList } from "@qltysh/fabro-api-client"; - -interface WorkflowGraphResponse { - graph: string; -} +import type { WorkflowDetailResponse } from "../lib/workflow-api"; export const handle = { wide: true }; @@ -39,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 46a44102d..735a7ca23 100644 --- a/apps/fabro-web/app/routes/workflow-detail.tsx +++ b/apps/fabro-web/app/routes/workflow-detail.tsx @@ -2,15 +2,7 @@ import { ChevronRightIcon } from "@heroicons/react/20/solid"; import { Link, Outlet, useLocation, useParams } from "react-router"; import { apiJson } from "../api"; import type { RunSettings } from "@qltysh/fabro-api-client"; - -interface ApiWorkflowDetail { - name: string; - slug: string; - description: string; - filename: string; - settings: RunSettings; - graph: string; -} +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 df8c34c31..df18f577d 100644 --- a/apps/fabro-web/app/routes/workflows.tsx +++ b/apps/fabro-web/app/routes/workflows.tsx @@ -15,27 +15,7 @@ import { import { Link } from "react-router"; import { apiJson } from "../api"; import { timeAgo, timeUntil } from "../lib/time"; - -interface WorkflowRunSummary { - ran_at?: string | null; -} - -interface WorkflowScheduleSummary { - expression: string; - next_run?: string | null; -} - -interface WorkflowListItem { - name: string; - slug: string; - filename: string; - last_run?: WorkflowRunSummary | null; - schedule?: WorkflowScheduleSummary | null; -} - -interface PaginatedWorkflowList { - data: WorkflowListItem[]; -} +import type { PaginatedWorkflowListResponse } from "../lib/workflow-api"; export function meta({}: any) { return [{ title: "Workflows — Fabro" }]; @@ -125,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, diff --git a/docs/api-reference/fabro-api.yaml b/docs/api-reference/fabro-api.yaml index df3d32c55..83bf6de25 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/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/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/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index f59593b68..998bd2cbd 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..f1933df7b 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; @@ -34,11 +35,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 +351,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())) } @@ -403,16 +399,65 @@ impl ServerStoreClient { Ok(()) } - pub(crate) async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result<()> { + pub(crate) async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result { 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,19 +629,20 @@ 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 +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 } - _ => RunAttachStreamError::Other(map_api_error(err)), + progenitor_client::Error::UnexpectedResponse(response) => { + response.status() == reqwest::StatusCode::NOT_FOUND + } + _ => false, } } - 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 0469f22ea..d556b6643 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!(); diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 001447b67..90e31d4fa 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}; @@ -477,7 +477,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]) @@ -510,6 +511,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") @@ -1275,25 +1282,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-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/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index c58cec75d..b67d9c08f 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 a30628c49..bb6db29ce 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -31,7 +31,7 @@ use fabro_llm::types::{ use fabro_model::{BilledModelUsage, BilledTokenCounts}; 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; @@ -49,11 +49,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; @@ -1036,6 +1037,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() @@ -1325,21 +1343,20 @@ async fn get_aggregate_billing( stages: totals.stages, }) .collect(); - let total_billing = - by_model - .iter() - .fold(BilledTokenCounts::default(), |mut acc, model| { - acc.input_tokens += model.billing.input_tokens; - acc.output_tokens += model.billing.output_tokens; - acc.reasoning_tokens += model.billing.reasoning_tokens.unwrap_or(0); - acc.cache_read_tokens += model.billing.cache_read_tokens.unwrap_or(0); - acc.cache_write_tokens += model.billing.cache_write_tokens.unwrap_or(0); - acc.total_tokens += model.billing.total_tokens; - if let Some(value) = model.billing.total_usd_micros { - *acc.total_usd_micros.get_or_insert(0) += value; - } - acc - }); + let total_billing = by_model + .iter() + .fold(BilledTokenCounts::default(), |mut acc, model| { + acc.input_tokens += model.billing.input_tokens; + acc.output_tokens += model.billing.output_tokens; + acc.reasoning_tokens += model.billing.reasoning_tokens.unwrap_or(0); + acc.cache_read_tokens += model.billing.cache_read_tokens.unwrap_or(0); + acc.cache_write_tokens += model.billing.cache_write_tokens.unwrap_or(0); + acc.total_tokens += model.billing.total_tokens; + if let Some(value) = model.billing.total_usd_micros { + *acc.total_usd_micros.get_or_insert(0) += value; + } + acc + }); let response = AggregateBilling { totals: AggregateBillingTotals { cache_read_tokens: nonzero_i64(total_billing.cache_read_tokens), @@ -2774,7 +2791,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"); @@ -2810,7 +2827,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, @@ -3427,20 +3444,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(); }; @@ -3455,19 +3458,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( @@ -6301,7 +6395,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) @@ -6330,7 +6424,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/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index abda9dce3..bfed74152 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -57,7 +57,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)?; @@ -65,7 +65,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-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/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs index b905f3fd0..eafaeffc3 100644 --- a/lib/crates/fabro-types/src/run_event/mod.rs +++ b/lib/crates/fabro-types/src/run_event/mod.rs @@ -701,7 +701,7 @@ mod tests { status: crate::StageStatus::Success, preferred_label: None, suggested_next_ids: vec!["next".to_string()], - usage: None, + billing: None, failure: None, notes: Some("done".to_string()), files_touched: vec!["src/main.rs".to_string()], @@ -770,7 +770,7 @@ mod tests { status: crate::StageStatus::Success, preferred_label: None, suggested_next_ids: vec!["next".to_string()], - usage: None, + billing: None, failure: None, notes: Some("done".to_string()), files_touched: vec!["src/main.rs".to_string()], 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 e33ae43d0..e40f2c549 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -24,6 +24,8 @@ use fabro_agent::{AgentEvent, SandboxEvent, WorktreeEvent, WorktreeEventCallback use fabro_llm::types::TokenCounts as LlmTokenCounts; 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. @@ -2372,7 +2374,7 @@ pub async fn append_event_to_sink( #[derive(Clone)] pub enum RunEventSink { - Store(RunDatabase), + Store(RunStoreHandle), JsonLines(Arc>>>), Callback(Arc), Composite(Vec), @@ -2384,6 +2386,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) } @@ -2421,12 +2428,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)?; @@ -2507,9 +2509,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 4590dd873..6c1beb854 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 a13461a32..fe0134d10 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 e791e49ac..578e7ce6c 100644 --- a/lib/crates/fabro-workflow/src/lib.rs +++ b/lib/crates/fabro-workflow/src/lib.rs @@ -140,6 +140,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 1a262f69b..8d03313d8 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::BilledModelUsage; +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 a0f4952e8..89c7e3455 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::{BilledModelUsage, Outcome, StageStatus}; 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 621b3328a..a74cee0b5 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::{BilledModelUsage, Outcome}; 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 639662776..389501107 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 4b48290aa..daa499344 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 fabro_types::BilledTokenCounts; use super::types::{Concluded, FinalizeOptions, Retroed}; @@ -64,7 +64,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, @@ -153,7 +153,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 @@ -356,7 +356,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( @@ -371,7 +371,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 c6139e672..88b0d0834 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. @@ -282,7 +283,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 @@ -300,7 +301,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"); @@ -408,7 +409,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() { @@ -1060,7 +1061,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 @@ -1130,7 +1131,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 @@ -1216,7 +1217,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 @@ -1359,7 +1360,7 @@ mod tests { "claude-sonnet-4-20250514", false, None, - &run_store, + &run_store.clone().into(), None, ) .await; @@ -1425,7 +1426,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, 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.