From 6a804e9808166bf8187f6f9c12e93972a9e48213 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 12:58:37 -0400 Subject: [PATCH 1/2] refactor(cli): stream run attach over SSE Replace the attach polling loop with the existing run attach SSE endpoint. Seed from stored history once, fetch interview questions only when needed, and keep completed-run replay behavior intact. --- .../fabro-cli/src/commands/run/attach.rs | 414 ++++++++++-------- .../fabro-cli/src/commands/system/events.rs | 25 +- lib/crates/fabro-cli/src/main.rs | 1 + lib/crates/fabro-cli/src/server_client.rs | 77 ++++ lib/crates/fabro-cli/src/sse.rs | 51 +++ lib/crates/fabro-cli/tests/it/cmd/attach.rs | 117 +++-- 6 files changed, 451 insertions(+), 234 deletions(-) create mode 100644 lib/crates/fabro-cli/src/sse.rs diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 39ec63321..6deeb80e1 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -4,8 +4,6 @@ use std::path::Path; #[cfg(test)] use std::path::PathBuf; use std::process::ExitCode; -use std::sync::Arc; -use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use anyhow::Result; @@ -19,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; +use tokio::time::{sleep, timeout}; use super::run_progress; use crate::server_client; @@ -72,157 +70,52 @@ pub(crate) async fn attach_run_with_client( .as_ref() .is_some_and(|record| record.settings.verbose_enabled()); let events = client.list_run_events(run_id, None, None).await?; - let event_lines = events - .iter() - .map(event_payload_line) - .collect::>>()?; + 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); - attach_run_server( - client, - run_id, - verbose, - event_lines, - events.last().map_or(0, |event| event.seq), - initial_exit_code, - kill_on_detach, - styles, - json_output, - ) - .await + + if state_is_terminal(&state) || initial_exit_code.is_some() { + return replay_run_with_client(client, run_id, verbose, events, 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), + } } -async fn attach_run_server( +async fn replay_run_with_client( client: &server_client::ServerStoreClient, run_id: &RunId, verbose: bool, - existing_events: Vec, - last_seq: u32, - initial_exit_code: Option, - kill_on_detach: bool, - styles: &'static Styles, + events: Vec, 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; - // Install Ctrl+C handler - let cancelled = Arc::new(AtomicBool::new(false)); - { - let cancelled = Arc::clone(&cancelled); - tokio::spawn(async move { - let _ = ctrl_c().await; - cancelled.store(true, Ordering::Relaxed); - }); - } - - for line in &existing_events { - emit_progress_line(&mut progress_ui, line, json_output)?; - } - - if json_output && !client.list_run_questions(run_id).await?.is_empty() { - eprintln!("{JSON_INTERVIEW_MESSAGE}"); - return Ok(ExitCode::from(1)); - } - - let mut next_seq = if last_seq == 0 { 1 } else { last_seq + 1 }; - let mut terminal_exit_code = initial_exit_code; - let mut terminal_event_seen_at = initial_exit_code.map(|_| Instant::now()); - - loop { - if cancelled.load(Ordering::Relaxed) { - if kill_on_detach { - let _ = client.cancel_run(run_id).await; - // Wait briefly for a terminal status or conclusion - for _ in 0..20 { - if client - .get_run_state(run_id) - .await - .ok() - .is_some_and(|state| { - state.conclusion.is_some() - || state - .status - .is_some_and(|record| record.status.is_terminal()) - }) - { - break; - } - sleep(Duration::from_millis(100)).await; - } - } else { - eprintln!("Detached from run (engine continues in background)"); - } - break; - } - - let mut saw_event = false; - let events = match client.list_run_events(run_id, Some(next_seq), None).await { - Ok(events) => events, - Err(err) if terminal_event_seen_at.is_some() && is_run_not_found_error(&err) => break, - Err(err) => return Err(err), - }; - for event in events { - 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)?; - next_seq = event.seq.saturating_add(1); - saw_event = true; - } - - if let Some(seen_at) = terminal_event_seen_at { - if !saw_event && seen_at.elapsed() >= ATTACH_FINAL_STATUS_GRACE { - break; - } - if !saw_event { - sleep(Duration::from_millis(50)).await; - } - continue; - } - - // Check for server-backed interview request - if let Some(question) = client.list_run_questions(run_id).await?.into_iter().next() { - if json_output { - eprintln!("{JSON_INTERVIEW_MESSAGE}"); - return Ok(ExitCode::from(1)); - } - - hide_progress(&mut progress_ui, json_output); - let interviewer = ConsoleInterviewer::new(styles); - let answer = fabro_interview::Interviewer::ask( - &interviewer, - api_question_to_question(&question), - ) - .await; - show_progress(&mut progress_ui, json_output); - - if answer_requires_reattach(&answer) { - eprintln!("{INTERVIEW_UNANSWERED_MESSAGE}"); - return Ok(ExitCode::from(1)); - } - - submit_server_interview_answer(client, run_id, &question.id, &answer).await?; - continue; - } - - let terminal_status = client - .get_run_state(run_id) - .await - .ok() - .and_then(|state| state.status.map(|record| record.status)) - .filter(|status| status.is_terminal()); - - if terminal_status.is_some() && !saw_event { - flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output) - .await?; - break; - } - - if !saw_event { - sleep(Duration::from_millis(100)).await; + 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); @@ -233,6 +126,194 @@ async fn attach_run_server( }) } +async fn attach_live_run_with_client( + client: &server_client::ServerStoreClient, + run_id: &RunId, + verbose: bool, + existing_events: Vec, + mut stream: server_client::RunAttachEventStream, + kill_on_detach: bool, + styles: &'static Styles, + json_output: bool, +) -> Result { + let is_tty = std::io::stderr().is_terminal(); + let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose); + 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)?; + } + + if let Some(exit_code) = + handle_pending_server_interview(client, run_id, &mut progress_ui, styles, json_output) + .await? + { + return Ok(exit_code); + } + + 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 Some(event) = next_event else { + break; + }; + + 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 event_starts_interview(&event) { + if let Some(exit_code) = handle_pending_server_interview( + client, + run_id, + &mut progress_ui, + styles, + json_output, + ) + .await? + { + return Ok(exit_code); + } + } + } + + 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( + client: &server_client::ServerStoreClient, + run_id: &RunId, + progress_ui: &mut run_progress::ProgressUI, + styles: &'static Styles, + json_output: bool, +) -> Result> { + let Some(question) = client.list_run_questions(run_id).await?.into_iter().next() else { + return Ok(None); + }; + + if json_output { + eprintln!("{JSON_INTERVIEW_MESSAGE}"); + return Ok(Some(ExitCode::from(1))); + } + + hide_progress(progress_ui, json_output); + let interviewer = ConsoleInterviewer::new(styles); + let answer = + fabro_interview::Interviewer::ask(&interviewer, api_question_to_question(&question)).await; + show_progress(progress_ui, json_output); + + if answer_requires_reattach(&answer) { + eprintln!("{INTERVIEW_UNANSWERED_MESSAGE}"); + return Ok(Some(ExitCode::from(1))); + } + + submit_server_interview_answer(client, run_id, &question.id, &answer).await?; + Ok(None) +} + +async fn handle_detach_signal( + client: &server_client::ServerStoreClient, + run_id: &RunId, + kill_on_detach: bool, +) { + if kill_on_detach { + let _ = client.cancel_run(run_id).await; + for _ in 0..20 { + if client + .get_run_state(run_id) + .await + .ok() + .is_some_and(|state| state_is_terminal(&state)) + { + break; + } + sleep(Duration::from_millis(100)).await; + } + } else { + eprintln!("Detached from run (engine continues in background)"); + } +} + +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, @@ -282,45 +363,19 @@ async fn submit_server_interview_answer( Ok(true) } -async fn flush_remaining_server_events( - client: &server_client::ServerStoreClient, - run_id: &RunId, - mut next_seq: u32, - progress_ui: &mut run_progress::ProgressUI, - json_output: bool, -) -> Result<()> { - let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE; - loop { - let mut saw_new_event = false; - 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) => break, - Err(err) => return Err(err), - }; - for event in events { - let line = event_payload_line(&event)?; - emit_progress_line(progress_ui, &line, json_output)?; - next_seq = event.seq.saturating_add(1); - saw_new_event = true; - } - - if Instant::now() >= deadline { - break; - } - - if !saw_new_event { - sleep(Duration::from_millis(100)).await; - } - } - - Ok(()) -} - 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 + .status + .as_ref() + .is_some_and(|record| record.status.is_terminal()) +} + fn emit_progress_line( progress_ui: &mut run_progress::ProgressUI, line: &str, @@ -450,6 +505,13 @@ fn event_exit_code(event: &EventEnvelope) -> Option { } } +fn event_starts_interview(event: &EventEnvelope) -> bool { + let Ok(run_event) = RunEvent::try_from(&event.payload) else { + return false; + }; + matches!(run_event.body, EventBody::InterviewStarted(_)) +} + #[cfg(test)] mod tests { #![allow(clippy::absolute_paths)] diff --git a/lib/crates/fabro-cli/src/commands/system/events.rs b/lib/crates/fabro-cli/src/commands/system/events.rs index 8be8519c8..0285fc4e4 100644 --- a/lib/crates/fabro-cli/src/commands/system/events.rs +++ b/lib/crates/fabro-cli/src/commands/system/events.rs @@ -3,6 +3,7 @@ use futures::StreamExt; use crate::args::{GlobalArgs, SystemEventsArgs}; use crate::server_client; +use crate::sse; pub(super) async fn events_command(args: &SystemEventsArgs, globals: &GlobalArgs) -> Result<()> { let client = server_client::connect_server_backed_api_client_with_storage_dir( @@ -23,25 +24,15 @@ pub(super) async fn events_command(args: &SystemEventsArgs, globals: &GlobalArgs while let Some(chunk) = stream.next().await { let chunk = chunk.map_err(|err| anyhow::anyhow!("{err}"))?; pending.extend_from_slice(&chunk); - drain_sse_lines(&mut pending, globals.json)?; - } - - if !pending.is_empty() { - drain_sse_lines(&mut pending, globals.json)?; - } - - Ok(()) -} - -fn drain_sse_lines(buffer: &mut Vec, json_output: bool) -> Result<()> { - while let Some(pos) = buffer.iter().position(|byte| *byte == b'\n') { - let line = buffer.drain(..=pos).collect::>(); - let line = String::from_utf8_lossy(&line); - let line = line.trim_end_matches(['\r', '\n']); - if let Some(data) = line.strip_prefix("data:") { - render_sse_payload(data.trim(), json_output)?; + for payload in sse::drain_sse_payloads(&mut pending, false) { + render_sse_payload(&payload, globals.json)?; } } + + for payload in sse::drain_sse_payloads(&mut pending, true) { + render_sse_payload(&payload, globals.json)?; + } + Ok(()) } diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index bd09d8e5c..95f404985 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -9,6 +9,7 @@ mod server_runs; mod shared; #[cfg(feature = "sleep_inhibitor")] mod sleep_inhibitor; +mod sse; mod user_config; use anyhow::Result; diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index b1f24a593..b8c411251 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -1,3 +1,4 @@ +use std::collections::VecDeque; use std::num::NonZeroU64; use std::path::{Path, PathBuf}; use std::time::Duration; @@ -13,6 +14,7 @@ use tokio::time::sleep; use crate::args::ServerTargetArgs; use crate::commands::server::start; +use crate::sse; use crate::user_config; #[derive(Clone)] @@ -26,6 +28,55 @@ struct LocalServerRuntime { storage_dir: PathBuf, } +pub(crate) struct RunAttachEventStream { + stream: progenitor_client::ByteStream, + pending_bytes: Vec, + buffered_events: VecDeque, +} + +pub(crate) enum RunAttachStreamError { + Gone, + Other(anyhow::Error), +} + +impl RunAttachEventStream { + fn new(stream: progenitor_client::ByteStream) -> Self { + Self { + stream, + pending_bytes: Vec::new(), + buffered_events: VecDeque::new(), + } + } + + pub(crate) async fn next_event(&mut self) -> Result> { + loop { + if let Some(event) = self.buffered_events.pop_front() { + return Ok(Some(event)); + } + + match self.stream.next().await { + Some(chunk) => { + let chunk = chunk.map_err(|err| anyhow!("{err}"))?; + self.pending_bytes.extend_from_slice(&chunk); + self.buffer_sse_events(false)?; + } + None => { + self.buffer_sse_events(true)?; + return Ok(self.buffered_events.pop_front()); + } + } + } + } + + fn buffer_sse_events(&mut self, finalize: bool) -> Result<()> { + for payload in sse::drain_sse_payloads(&mut self.pending_bytes, finalize) { + let event: types::EventEnvelope = serde_json::from_str(&payload)?; + self.buffered_events.push_back(convert_type(event)?); + } + Ok(()) + } +} + pub(crate) use fabro_store::RunProjection; pub(crate) async fn connect_server(storage_dir: &Path) -> Result { @@ -300,6 +351,19 @@ impl ServerStoreClient { .collect::>>() } + pub(crate) async fn attach_run_events( + &self, + run_id: &RunId, + since_seq: Option, + ) -> std::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)?; + Ok(RunAttachEventStream::new(response.into_inner())) + } + pub(crate) async fn list_run_questions( &self, run_id: &RunId, @@ -520,6 +584,19 @@ 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/src/sse.rs b/lib/crates/fabro-cli/src/sse.rs new file mode 100644 index 000000000..7208e625d --- /dev/null +++ b/lib/crates/fabro-cli/src/sse.rs @@ -0,0 +1,51 @@ +pub(crate) fn drain_sse_payloads(buffer: &mut Vec, finalize: bool) -> Vec { + let mut payloads = Vec::new(); + + while let Some(pos) = buffer.iter().position(|byte| *byte == b'\n') { + let line = buffer.drain(..=pos).collect::>(); + if let Some(payload) = sse_data_line(&line) { + payloads.push(payload); + } + } + + if finalize && !buffer.is_empty() { + let line = std::mem::take(buffer); + if let Some(payload) = sse_data_line(&line) { + payloads.push(payload); + } + } + + payloads +} + +fn sse_data_line(line: &[u8]) -> Option { + let line = String::from_utf8_lossy(line); + let line = line.trim_end_matches(['\r', '\n']); + line.strip_prefix("data:") + .map(|data| data.trim().to_string()) +} + +#[cfg(test)] +mod tests { + use super::drain_sse_payloads; + + #[test] + fn drain_sse_payloads_handles_chunk_boundaries() { + let mut buffer = b"data: {\"a\":1}\n\ndat".to_vec(); + + assert_eq!(drain_sse_payloads(&mut buffer, false), vec![r#"{"a":1}"#]); + assert_eq!(buffer, b"dat"); + + buffer.extend_from_slice(b"a: {\"b\":2}\n\n"); + assert_eq!(drain_sse_payloads(&mut buffer, false), vec![r#"{"b":2}"#]); + assert!(buffer.is_empty()); + } + + #[test] + fn drain_sse_payloads_ignores_keepalives_and_finalizes_trailing_line() { + let mut buffer = b": keep-alive\n\n\ndata: {\"a\":1}".to_vec(); + + assert_eq!(drain_sse_payloads(&mut buffer, true), vec![r#"{"a":1}"#]); + assert!(buffer.is_empty()); + } +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 3850d190f..5dea2ae9f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -10,6 +10,48 @@ use super::support::{output_stdout, resolve_run, wait_for_status, write_gated_wo const SHARED_DAEMON_TIMEOUT: Duration = Duration::from_secs(30); +fn live_run_state_response() -> serde_json::Value { + serde_json::json!({ + "run": null, + "graph_source": null, + "start": null, + "status": { + "status": "running", + "reason": null, + "updated_at": "2026-04-05T12:00:01Z" + }, + "checkpoint": null, + "checkpoints": [], + "conclusion": null, + "retro": null, + "retro_prompt": null, + "retro_response": null, + "sandbox": null, + "final_patch": null, + "pull_request": null, + "nodes": {} + }) +} + +fn run_sse_body(run_id: &str) -> String { + let completed = serde_json::json!({ + "seq": 2, + "payload": { + "event": "run.completed", + "id": "evt-run-completed", + "run_id": run_id, + "ts": "2026-04-05T12:00:01Z", + "properties": { + "duration_ms": 12, + "artifact_count": 0, + "status": "success" + } + } + }); + + format!("data: {completed}\n\n") +} + #[test] fn help() { let context = test_context!(); @@ -76,7 +118,7 @@ fn attach_uses_configured_server_target_without_server_flag() { "labels": {}, "host_repo_path": null, "start_time": "2026-04-05T12:00:00Z", - "status": "succeeded", + "status": "running", "status_reason": null, "duration_ms": 12, "total_cost": null @@ -88,59 +130,49 @@ fn attach_uses_configured_server_target_without_server_flag() { server.mock(|when, then| { when.method("GET") .path(format!("/api/v1/runs/{run_id}/events")); - 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}/state")); then.status(200) .header("Content-Type", "application/json") .body( serde_json::json!({ - "run": null, - "graph_source": null, - "start": null, - "status": { - "status": "succeeded", - "reason": null, - "updated_at": "2026-04-05T12:00:01Z" - }, - "checkpoint": null, - "checkpoints": [], - "conclusion": { - "timestamp": "2026-04-05T12:00:01Z", - "status": "success", - "duration_ms": 12, - "stages": [], - "total_cost": null, - "total_retries": 0, - "total_input_tokens": 0, - "total_output_tokens": 0, - "total_cache_read_tokens": 0, - "total_cache_write_tokens": 0, - "total_reasoning_tokens": 0, - "has_pricing": false - }, - "retro": null, - "retro_prompt": null, - "retro_response": null, - "sandbox": null, - "final_patch": null, - "pull_request": null, - "nodes": {} + "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}/questions")); + .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}}"#); }); + let attach_mock = 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(run_sse_body(run_id.as_str())); + }); context.write_home( ".fabro/settings.toml", format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()), @@ -148,7 +180,7 @@ fn attach_uses_configured_server_target_without_server_flag() { let output = context .command() - .args(["attach", &run_id]) + .args(["--json", "attach", &run_id]) .output() .expect("attach should execute"); @@ -159,6 +191,9 @@ fn attach_uses_configured_server_target_without_server_flag() { String::from_utf8_lossy(&output.stderr) ); list_mock.assert(); + attach_mock.assert(); + let stdout = String::from_utf8(output.stdout).expect("stdout should be UTF-8"); + assert!(stdout.contains("\"event\":\"run.completed\""), "{stdout}"); } #[test] From 26e21f0a136e816e31c2f0c21f05b45d01ab40be Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Tue, 7 Apr 2026 12:59:16 -0400 Subject: [PATCH 2/2] fmt --- lib/crates/fabro-server/src/serve.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/lib/crates/fabro-server/src/serve.rs b/lib/crates/fabro-server/src/serve.rs index 1adc6bb95..55cfbc2f7 100644 --- a/lib/crates/fabro-server/src/serve.rs +++ b/lib/crates/fabro-server/src/serve.rs @@ -549,7 +549,10 @@ mod tests { let disk_store = build_object_store_with_preference(&store_path, false) .expect("disk-backed store should build"); - assert!(store_path.exists(), "disk-backed store should create store dir"); + assert!( + store_path.exists(), + "disk-backed store should create store dir" + ); drop(disk_store); let mem_path = temp.path().join("memory-store");