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]