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