diff --git a/lib/crates/fabro-cli/src/args.rs b/lib/crates/fabro-cli/src/args.rs index d74593658..d010d4846 100644 --- a/lib/crates/fabro-cli/src/args.rs +++ b/lib/crates/fabro-cli/src/args.rs @@ -635,7 +635,7 @@ pub(crate) enum RunCommands { Run(RunArgs), /// Create a workflow run (allocate run dir, persist spec) Create(RunArgs), - /// Start a created workflow run (spawn engine process) + /// Start a created workflow run on the server Start { /// Run ID prefix or workflow name run: String, @@ -645,18 +645,12 @@ pub(crate) enum RunCommands { /// Run ID prefix or workflow name run: String, }, - /// Internal: run the engine process - #[command(name = "__detached", hide = true)] - Detached { + /// Internal: queue or resume a workflow run via the server + #[command(name = "__runner", hide = true)] + Runner { /// Run ID #[arg(long)] run_id: fabro_types::RunId, - /// Run directory - #[arg(long)] - run_dir: PathBuf, - /// Launcher metadata path - #[arg(long)] - launcher_path: PathBuf, /// Resume from checkpoint instead of fresh start #[arg(long)] resume: bool, @@ -683,7 +677,7 @@ impl RunCommands { Self::Create(_) => "create", Self::Start { .. } => "start", Self::Attach { .. } => "attach", - Self::Detached { .. } => "__detached", + Self::Runner { .. } => "__runner", Self::Diff(_) => "diff", Self::Logs(_) => "logs", Self::Resume(_) => "resume", diff --git a/lib/crates/fabro-cli/src/commands/artifact/cp.rs b/lib/crates/fabro-cli/src/commands/artifact/cp.rs index da17dc5ec..b1288b132 100644 --- a/lib/crates/fabro-cli/src/commands/artifact/cp.rs +++ b/lib/crates/fabro-cli/src/commands/artifact/cp.rs @@ -3,20 +3,17 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result, bail}; use fabro_store::RuntimeState; use fabro_workflow::artifacts::{ArtifactEntry, scan_artifacts}; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use crate::args::{ArtifactCpArgs, GlobalArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{print_json_pretty, split_run_path}; use crate::user_config::load_user_settings_with_globals; pub(super) async fn cp_command(args: &ArtifactCpArgs, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); let (run_id, asset_path) = parse_source(&args.source); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, run_id)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(run_id)?; let runtime_state = RuntimeState::new(&run.path); let entries = scan_artifacts( &runtime_state.artifacts_dir(), diff --git a/lib/crates/fabro-cli/src/commands/artifact/list.rs b/lib/crates/fabro-cli/src/commands/artifact/list.rs index e364634b6..a7715df5d 100644 --- a/lib/crates/fabro-cli/src/commands/artifact/list.rs +++ b/lib/crates/fabro-cli/src/commands/artifact/list.rs @@ -1,19 +1,16 @@ use anyhow::Result; use fabro_store::RuntimeState; use fabro_workflow::artifacts::scan_artifacts; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use crate::args::{ArtifactListArgs, GlobalArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::format_size; use crate::user_config::load_user_settings_with_globals; pub(super) async fn list_command(args: &ArtifactListArgs, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run_id)?; let runtime_state = RuntimeState::new(&run.path); let entries = scan_artifacts( &runtime_state.artifacts_dir(), diff --git a/lib/crates/fabro-cli/src/commands/pr/create.rs b/lib/crates/fabro-cli/src/commands/pr/create.rs index f81bddfc9..d12cd6fb5 100644 --- a/lib/crates/fabro-cli/src/commands/pr/create.rs +++ b/lib/crates/fabro-cli/src/commands/pr/create.rs @@ -5,12 +5,12 @@ use fabro_model::Catalog; use fabro_sandbox::daytona::detect_repo_info; use fabro_workflow::outcome::StageStatus; use fabro_workflow::pull_request::maybe_open_pull_request; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; +use fabro_workflow::run_lookup::runs_base; use tracing::info; use crate::args::{GlobalArgs, PrCreateArgs}; use crate::commands::store::rebuild::rebuild_run_store; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; @@ -30,12 +30,10 @@ async fn create_from( github_app: Option, globals: &GlobalArgs, ) -> Result<()> { - let storage_dir = base.parent().unwrap_or(base); - let client = server_client::connect_server(storage_dir).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, base, &args.run_id)?; + let lookup = ServerRunLookup::connect_from_runs_base(base).await?; + let run = lookup.resolve(&args.run_id)?; let run_id = run.run_id(); - let events = client.list_run_events(&run_id, None, None).await?; + let events = lookup.client().list_run_events(&run_id, None, None).await?; let run_store = rebuild_run_store(&run_id, &events).await?; let state = run_store.state().await?; diff --git a/lib/crates/fabro-cli/src/commands/pr/list.rs b/lib/crates/fabro-cli/src/commands/pr/list.rs index 6e2e7fc8f..85e3b7789 100644 --- a/lib/crates/fabro-cli/src/commands/pr/list.rs +++ b/lib/crates/fabro-cli/src/commands/pr/list.rs @@ -9,6 +9,7 @@ use tracing::info; use crate::args::{GlobalArgs, PrListArgs}; use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; @@ -28,9 +29,16 @@ pub(super) async fn list_command( ) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - list_from(&client, &summaries, &base, args, github_app, globals).await + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + list_from( + lookup.client(), + lookup.summaries(), + &base, + args, + github_app, + globals, + ) + .await } async fn list_from( diff --git a/lib/crates/fabro-cli/src/commands/pr/mod.rs b/lib/crates/fabro-cli/src/commands/pr/mod.rs index 3bbff5516..fce7abd2a 100644 --- a/lib/crates/fabro-cli/src/commands/pr/mod.rs +++ b/lib/crates/fabro-cli/src/commands/pr/mod.rs @@ -9,10 +9,9 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; use fabro_types::PullRequestRecord; -use fabro_workflow::run_lookup::resolve_run_from_summaries; use crate::args::{GlobalArgs, PrCommand, PrNamespace}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::github::build_github_app_credentials; use crate::user_config::load_user_settings_with_globals; @@ -35,13 +34,11 @@ pub(crate) async fn load_pr_record( base: &Path, run_id: &str, ) -> Result<(PullRequestRecord, PathBuf)> { - let storage_dir = base.parent().unwrap_or(base); - let client = server_client::connect_server(storage_dir).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, base, run_id)?; + let lookup = ServerRunLookup::connect_from_runs_base(base).await?; + let run = lookup.resolve(run_id)?; let run_id = run.run_id(); let run_dir = run.path; - let state = client.get_run_state(&run_id).await?; + let state = lookup.client().get_run_state(&run_id).await?; let record = state.pull_request.with_context(|| { format!("No pull request found in store. Create one first with: fabro pr create {run_id}") })?; diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 7f3849729..175a7f42b 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -21,10 +21,6 @@ use tokio::time::sleep; use super::run_progress; use crate::server_client; -#[cfg(test)] -const ATTACH_STARTUP_GRACE: Duration = Duration::from_millis(200); -#[cfg(not(test))] -const ATTACH_STARTUP_GRACE: Duration = Duration::from_secs(3); 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."; @@ -42,7 +38,6 @@ pub(crate) async fn attach_run( run_id: Option<&RunId>, kill_on_detach: bool, styles: &'static Styles, - engine_child: Option, json_output: bool, ) -> Result { let inferred_storage_dir = infer_storage_dir(run_dir); @@ -64,7 +59,6 @@ pub(crate) async fn attach_run( .collect::>>()?; let initial_exit_code = events.iter().rev().find_map(event_exit_code); return attach_run_server( - run_dir, &client, run_id, verbose, @@ -73,7 +67,6 @@ pub(crate) async fn attach_run( initial_exit_code, kill_on_detach, styles, - engine_child, json_output, ) .await; @@ -85,7 +78,6 @@ pub(crate) async fn attach_run( } async fn attach_run_server( - run_dir: &Path, client: &server_client::ServerStoreClient, run_id: &RunId, verbose: bool, @@ -94,11 +86,8 @@ async fn attach_run_server( initial_exit_code: Option, kill_on_detach: bool, styles: &'static Styles, - engine_child: Option, json_output: bool, ) -> Result { - let mut engine_guard = engine_child.map(EngineChildGuard::new); - let is_tty = std::io::stderr().is_terminal(); let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose); @@ -117,30 +106,18 @@ async fn attach_run_server( } if json_output && !client.list_run_questions(run_id).await?.is_empty() { - defuse_engine_child(&mut engine_guard); eprintln!("{JSON_INTERVIEW_MESSAGE}"); return Ok(ExitCode::from(1)); } let mut next_seq = if last_seq == 0 { 1 } else { last_seq + 1 }; - let mut cached_pid: Option = None; - let attach_started = Instant::now(); let mut terminal_exit_code = initial_exit_code; let mut terminal_event_seen_at = initial_exit_code.map(|_| Instant::now()); loop { - let server_owned = engine_guard.is_none() && read_launcher_pid(run_dir).is_none(); if cancelled.load(Ordering::Relaxed) { if kill_on_detach { - if let Some(guard) = engine_guard.as_mut() { - if let Some(child) = guard.inner() { - let _ = child.kill(); - } - } else if server_owned { - let _ = client.cancel_run(run_id).await; - } else { - kill_engine(run_dir); - } + let _ = client.cancel_run(run_id).await; // Wait briefly for a terminal status or conclusion for _ in 0..20 { if client @@ -159,9 +136,6 @@ async fn attach_run_server( sleep(Duration::from_millis(100)).await; } } else { - if let Some(guard) = engine_guard.as_mut() { - guard.defuse(); - } eprintln!("Detached from run (engine continues in background)"); } break; @@ -197,7 +171,6 @@ async fn attach_run_server( // Check for server-backed interview request if let Some(question) = client.list_run_questions(run_id).await?.into_iter().next() { if json_output { - defuse_engine_child(&mut engine_guard); eprintln!("{JSON_INTERVIEW_MESSAGE}"); return Ok(ExitCode::from(1)); } @@ -212,9 +185,6 @@ async fn attach_run_server( show_progress(&mut progress_ui, json_output); if answer_requires_reattach(&answer) { - if let Some(guard) = engine_guard.as_mut() { - guard.defuse(); - } eprintln!("{INTERVIEW_UNANSWERED_MESSAGE}"); return Ok(ExitCode::from(1)); } @@ -230,64 +200,10 @@ async fn attach_run_server( .and_then(|state| state.status.map(|record| record.status)) .filter(|status| status.is_terminal()); - let child_alive_via_handle = engine_guard.as_mut().and_then(|guard| { - guard.inner().map(|child| match child.try_wait() { - Ok(None) => true, // still running - Ok(Some(_)) | Err(_) => false, // exited or error - }) - }); - - if let Some(child_alive) = child_alive_via_handle { - if !child_alive && !saw_event { - flush_remaining_server_events( - client, - run_id, - next_seq, - &mut progress_ui, - json_output, - ) + if terminal_status.is_some() && !saw_event { + flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output) .await?; - break; - } - } else { - if terminal_status.is_some() && !saw_event { - flush_remaining_server_events( - client, - run_id, - next_seq, - &mut progress_ui, - json_output, - ) - .await?; - break; - } - - let engine_alive = if server_owned { - true - } else { - match cached_pid { - Some(pid) => process_alive(pid), - None => { - if let Some(pid) = read_launcher_pid(run_dir) { - cached_pid = Some(pid); - process_alive(pid) - } else { - attach_started.elapsed() < ATTACH_STARTUP_GRACE - } - } - } - }; - if !engine_alive { - flush_remaining_server_events( - client, - run_id, - next_seq, - &mut progress_ui, - json_output, - ) - .await?; - break; - } + break; } if !saw_event { @@ -450,10 +366,6 @@ fn restore_empty_run_properties(value: &mut serde_json::Value) { } } -fn read_launcher_pid(run_dir: &Path) -> Option { - super::launcher::active_launcher_record_for_run(run_dir).map(|record| record.pid) -} - fn infer_storage_dir(run_dir: &Path) -> Option { let runs_dir = run_dir.parent()?; let storage_dir = runs_dir.parent()?; @@ -461,124 +373,17 @@ fn infer_storage_dir(run_dir: &Path) -> Option { } fn infer_run_id(run_dir: &Path) -> Option { - super::launcher::launcher_record_for_run(run_dir) - .map(|record| record.run_id) - .or_else(|| { - std::fs::read_to_string(run_dir.join("id.txt")) - .ok() - .map(|run_id| run_id.trim().to_string()) - .filter(|run_id| !run_id.is_empty()) - .and_then(|run_id| run_id.parse().ok()) - }) -} - -#[cfg(test)] -struct InterviewClaimGuard { - claim_path: PathBuf, -} - -#[cfg(test)] -impl InterviewClaimGuard { - fn acquire(claim_path: &Path) -> Option { - if try_claim_interview_request(claim_path) { - Some(Self { - claim_path: claim_path.to_path_buf(), - }) - } else { - None - } - } -} - -#[cfg(test)] -impl Drop for InterviewClaimGuard { - fn drop(&mut self) { - let _ = std::fs::remove_file(&self.claim_path); - } -} - -struct EngineChildGuard { - child: Option, -} - -impl EngineChildGuard { - fn new(child: std::process::Child) -> Self { - Self { child: Some(child) } - } - - fn inner(&mut self) -> Option<&mut std::process::Child> { - self.child.as_mut() - } - - fn defuse(&mut self) { - self.child.take(); - } -} - -fn defuse_engine_child(engine_guard: &mut Option) { - if let Some(guard) = engine_guard.as_mut() { - guard.defuse(); - } -} - -impl Drop for EngineChildGuard { - fn drop(&mut self) { - if let Some(mut child) = self.child.take() { - let _ = child.kill(); - let _ = child.wait(); - } - } -} - -#[cfg(test)] -fn try_claim_interview_request(claim_path: &Path) -> bool { - if let Some(parent) = claim_path.parent() { - if std::fs::create_dir_all(parent).is_err() { - return false; - } - } - - if let Ok(existing) = std::fs::read_to_string(claim_path) { - if let Ok(pid) = existing.trim().parse::() { - if process_alive(pid) { - return pid == std::process::id(); - } - } - let _ = std::fs::remove_file(claim_path); - } - - match std::fs::OpenOptions::new() - .create_new(true) - .write(true) - .open(claim_path) - { - Ok(mut file) => { - let _ = writeln!(file, "{}", std::process::id()); - true - } - Err(_) => false, - } + std::fs::read_to_string(run_dir.join("id.txt")) + .ok() + .map(|run_id| run_id.trim().to_string()) + .filter(|run_id| !run_id.is_empty()) + .and_then(|run_id| run_id.parse().ok()) } fn answer_requires_reattach(answer: &fabro_interview::Answer) -> bool { matches!(answer.value, AnswerValue::Aborted | AnswerValue::Skipped) } -#[cfg(test)] -fn write_interview_response_atomically( - response_path: &Path, - answer: &fabro_interview::Answer, -) -> Result<()> { - let response_json = serde_json::to_string_pretty(answer)?; - if let Some(parent) = response_path.parent() { - std::fs::create_dir_all(parent)?; - } - let temp_path = response_path.with_extension("json.tmp"); - std::fs::write(&temp_path, response_json)?; - std::fs::rename(temp_path, response_path)?; - Ok(()) -} - async fn determine_exit_code_with_server( client: &server_client::ServerStoreClient, run_id: &RunId, @@ -614,16 +419,6 @@ async fn determine_exit_code_with_server( } } -fn kill_engine(run_dir: &Path) { - if let Some(pid) = read_launcher_pid(run_dir) { - fabro_proc::sigterm(pid); - } -} - -fn process_alive(pid: u32) -> bool { - fabro_proc::process_alive(pid) -} - fn event_exit_code(event: &EventEnvelope) -> Option { let run_event = RunEvent::try_from(&event.payload).ok()?; match run_event.body { @@ -642,8 +437,6 @@ fn event_exit_code(event: &EventEnvelope) -> Option { #[cfg(test)] mod tests { use super::*; - use crate::commands::run::launcher; - use chrono::Utc; use fabro_interview::{Answer, AnswerValue}; use fabro_util::terminal::Styles; @@ -655,17 +448,9 @@ mod tests { async fn attach_errors_without_store_context() { let dir = tempfile::tempdir().unwrap(); - let err = attach_run( - dir.path(), - None, - None, - false, - no_color_styles(), - None, - false, - ) - .await - .unwrap_err(); + let err = attach_run(dir.path(), None, None, false, no_color_styles(), false) + .await + .unwrap_err(); assert!( err.to_string() @@ -690,55 +475,20 @@ mod tests { } #[test] - fn infer_run_id_uses_launcher_record_without_run_json() { + fn infer_run_id_reads_id_txt() { let dir = tempfile::tempdir().unwrap(); let storage_dir = dir.path().join("storage"); let run_dir = storage_dir.join("runs").join("20260401-test"); std::fs::create_dir_all(&run_dir).unwrap(); - - launcher::write_launcher_record( - &launcher::launcher_record_path(&storage_dir, &fabro_types::fixtures::RUN_1), - &launcher::LauncherRecord { - run_id: fabro_types::fixtures::RUN_1, - run_dir: run_dir.clone(), - pid: u32::MAX, - resume: false, - log_path: dir.path().join("launcher.log"), - started_at: Utc::now(), - }, + std::fs::write( + run_dir.join("id.txt"), + format!("{}\n", fabro_types::fixtures::RUN_1), ) .unwrap(); assert_eq!(infer_run_id(&run_dir), Some(fabro_types::fixtures::RUN_1)); } - #[test] - fn try_claim_interview_request_reclaims_stale_claim() { - let dir = tempfile::tempdir().unwrap(); - let claim_path = dir.path().join("runtime").join("interview_request.claim"); - std::fs::create_dir_all(claim_path.parent().unwrap()).unwrap(); - std::fs::write(&claim_path, "999999\n").unwrap(); - - assert!(try_claim_interview_request(&claim_path)); - assert_eq!( - std::fs::read_to_string(claim_path).unwrap(), - format!("{}\n", std::process::id()) - ); - } - - #[test] - fn interview_claim_guard_releases_claim_on_drop() { - let dir = tempfile::tempdir().unwrap(); - let claim_path = dir.path().join("runtime").join("interview_request.claim"); - - { - let _guard = InterviewClaimGuard::acquire(&claim_path).unwrap(); - assert!(claim_path.exists()); - } - - assert!(!claim_path.exists()); - } - #[test] fn answer_requires_reattach_for_aborted_and_skipped_answers() { let aborted = Answer { @@ -757,65 +507,4 @@ mod tests { assert!(answer_requires_reattach(&skipped)); assert!(!answer_requires_reattach(&answered)); } - - #[test] - fn engine_child_guard_kills_on_drop() { - let child = std::process::Command::new("sleep") - .arg("60") - .spawn() - .unwrap(); - let pid = child.id(); - - { - let _guard = EngineChildGuard::new(child); - } - - // Process should be dead after guard is dropped - assert!( - !process_alive(pid), - "process should be dead after guard drop" - ); - } - - #[test] - fn engine_child_guard_defuse_keeps_alive() { - let child = std::process::Command::new("sleep") - .arg("60") - .spawn() - .unwrap(); - let pid = child.id(); - - { - let mut guard = EngineChildGuard::new(child); - guard.defuse(); - } - - // Process should still be alive after defused guard is dropped - assert!( - process_alive(pid), - "process should still be alive after defused guard drop" - ); - - // Clean up - #[cfg(unix)] - fabro_proc::sigkill(pid); - } - - #[test] - fn write_interview_response_atomically_persists_answer() { - let dir = tempfile::tempdir().unwrap(); - let response_path = dir.path().join("interview_response.json"); - let answer = Answer { - value: AnswerValue::Text("ship it".to_string()), - selected_option: None, - text: Some("ship it".to_string()), - }; - - write_interview_response_atomically(&response_path, &answer).unwrap(); - - let saved: Answer = - serde_json::from_str(&std::fs::read_to_string(&response_path).unwrap()).unwrap(); - assert_eq!(saved.text.as_deref(), Some("ship it")); - assert!(!response_path.with_extension("json.tmp").exists()); - } } diff --git a/lib/crates/fabro-cli/src/commands/run/command.rs b/lib/crates/fabro-cli/src/commands/run/command.rs index 12cc31818..dc97a1223 100644 --- a/lib/crates/fabro-cli/src/commands/run/command.rs +++ b/lib/crates/fabro-cli/src/commands/run/command.rs @@ -36,7 +36,6 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<( Some(&run_id), true, styles, - None, globals.json, ) .await?; diff --git a/lib/crates/fabro-cli/src/commands/run/cp.rs b/lib/crates/fabro-cli/src/commands/run/cp.rs index eb3990052..fe6db44e1 100644 --- a/lib/crates/fabro-cli/src/commands/run/cp.rs +++ b/lib/crates/fabro-cli/src/commands/run/cp.rs @@ -3,12 +3,11 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result, bail}; use fabro_agent::sandbox::Sandbox; use fabro_sandbox::reconnect::reconnect; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use tokio::fs; use tracing::{debug, info}; use crate::args::{CpArgs, GlobalArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{print_json_pretty, split_run_path}; use crate::user_config::load_user_settings_with_globals; @@ -28,7 +27,6 @@ enum CopyDirection { pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()> { let direction = parse_direction(&args.src, &args.dst)?; let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); match direction { CopyDirection::Download { @@ -36,7 +34,7 @@ pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()> remote_path, local_path, } => { - let sandbox = load_sandbox(&cli_settings.storage_dir(), &base, &run_prefix).await?; + let sandbox = load_sandbox(&cli_settings.storage_dir(), &run_prefix).await?; let file_count = if args.recursive { Some(download_recursive(&*sandbox, &remote_path, &local_path).await?) @@ -69,7 +67,7 @@ pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()> run_prefix, remote_path, } => { - let sandbox = load_sandbox(&cli_settings.storage_dir(), &base, &run_prefix).await?; + let sandbox = load_sandbox(&cli_settings.storage_dir(), &run_prefix).await?; let file_count = if args.recursive { Some(upload_recursive(&*sandbox, &local_path, &remote_path).await?) @@ -123,15 +121,11 @@ fn parse_direction(src: &str, dst: &str) -> Result { } } -async fn load_sandbox( - storage_dir: &Path, - base: &Path, - run_prefix: &str, -) -> Result> { - let client = server_client::connect_server(storage_dir).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, base, run_prefix)?; - let record = client +async fn load_sandbox(storage_dir: &Path, run_prefix: &str) -> Result> { + let lookup = ServerRunLookup::connect(storage_dir).await?; + let run = lookup.resolve(run_prefix)?; + let record = lookup + .client() .get_run_state(&run.run_id()) .await? .sandbox diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index c52b02473..acb148259 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -3,25 +3,22 @@ use std::path::Path; use anyhow::{Context, Result, bail}; use fabro_sandbox::reconnect::reconnect; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use fabro_workflow::sandbox_git::GIT_REMOTE; use tracing::{debug, info}; use crate::args::{DiffArgs, GlobalArgs}; -use crate::server_client; use crate::server_client::RunProjection; +use crate::server_runs::ServerRunLookup; use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> { info!(run_id = %args.run, "Showing diff"); let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run)?; let run_id = run.run_id(); - let state = client.get_run_state(&run_id).await?; + let state = lookup.client().get_run_state(&run_id).await?; let patch = resolve_diff(&run.path, &state, &args).await?; diff --git a/lib/crates/fabro-cli/src/commands/run/fork.rs b/lib/crates/fabro-cli/src/commands/run/fork.rs index 9fd7f4a98..2c9a3023a 100644 --- a/lib/crates/fabro-cli/src/commands/run/fork.rs +++ b/lib/crates/fabro-cli/src/commands/run/fork.rs @@ -3,25 +3,22 @@ use anyhow::Result; use fabro_checkpoint::git::Store; use fabro_util::terminal::Styles; use fabro_workflow::operations::{ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork}; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use git2::Repository; use crate::args::{ForkArgs, GlobalArgs}; use crate::commands::store::rebuild::rebuild_run_store; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn run(args: &ForkArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> { let repo = Repository::discover(".").context("not in a git repository")?; let cli_settings = load_user_settings_with_globals(globals)?; - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let base = runs_base(&cli_settings.storage_dir()); - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run_id)?; let run_id = run.run_id(); let store = Store::new(repo); - let events = client.list_run_events(&run_id, None, None).await?; + let events = lookup.client().list_run_events(&run_id, None, None).await?; let run_store = rebuild_run_store(&run_id, &events).await?; let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?; diff --git a/lib/crates/fabro-cli/src/commands/run/launcher.rs b/lib/crates/fabro-cli/src/commands/run/launcher.rs deleted file mode 100644 index b40a39be1..000000000 --- a/lib/crates/fabro-cli/src/commands/run/launcher.rs +++ /dev/null @@ -1,217 +0,0 @@ -use std::path::{Path, PathBuf}; - -#[cfg(test)] -use anyhow::{Context, Result}; -use chrono::{DateTime, Utc}; -use fabro_types::RunId; -use serde::{Deserialize, Serialize}; - -#[cfg(test)] -use crate::commands::run::short_run_id; - -#[derive(Debug, Clone, Serialize, Deserialize)] -pub(crate) struct LauncherRecord { - pub run_id: RunId, - pub run_dir: PathBuf, - pub pid: u32, - pub resume: bool, - pub log_path: PathBuf, - pub started_at: DateTime, -} - -pub(crate) fn launcher_dir(storage_dir: &Path) -> PathBuf { - storage_dir.join("launchers") -} - -pub(crate) fn launcher_record_path(storage_dir: &Path, run_id: &RunId) -> PathBuf { - launcher_dir(storage_dir).join(format!("{run_id}.json")) -} - -#[cfg(test)] -pub(crate) fn write_launcher_record(path: &Path, record: &LauncherRecord) -> Result<()> { - if let Some(parent) = path.parent() { - std::fs::create_dir_all(parent)?; - } - std::fs::write(path, serde_json::to_string_pretty(record)?) - .with_context(|| format!("Failed to write launcher metadata to {}", path.display())) -} - -pub(crate) fn read_launcher_record(path: &Path) -> Option { - let content = std::fs::read_to_string(path).ok()?; - serde_json::from_str(&content).ok() -} - -pub(crate) fn remove_launcher_record(path: &Path) { - let _ = std::fs::remove_file(path); -} - -pub(crate) fn active_launcher_record_for_run(run_dir: &Path) -> Option { - let launcher = launcher_record_for_run(run_dir)?; - let storage_dir = run_dir.parent()?.parent()?; - let path = launcher_record_path(storage_dir, &launcher.run_id); - if launcher_record_is_running(&launcher) { - Some(launcher) - } else { - remove_launcher_record(&path); - None - } -} - -pub(crate) fn launcher_record_for_run(run_dir: &Path) -> Option { - let storage_dir = run_dir.parent()?.parent()?; - let launchers_dir = launcher_dir(storage_dir); - let entries = std::fs::read_dir(&launchers_dir).ok()?; - for entry in entries.filter_map(std::result::Result::ok) { - let path = entry.path(); - let Some(launcher) = read_launcher_record(&path) else { - continue; - }; - if launcher.run_dir == run_dir { - return Some(launcher); - } - } - - None -} - -#[cfg(test)] -pub(crate) fn active_launcher_record(storage_dir: &Path, run_id: &RunId) -> Option { - let path = launcher_record_path(storage_dir, run_id); - let launcher = read_launcher_record(&path)?; - if launcher_record_is_running(&launcher) { - Some(launcher) - } else { - remove_launcher_record(&path); - None - } -} - -pub(crate) fn launcher_record_is_running(record: &LauncherRecord) -> bool { - fabro_proc::process_alive(record.pid) && launcher_process_matches(record) -} - -#[cfg(unix)] -fn launcher_process_matches(record: &LauncherRecord) -> bool { - let output = match std::process::Command::new("ps") - .args(["-ww", "-o", "command=", "-p", &record.pid.to_string()]) - .output() - { - Ok(output) if output.status.success() => output, - _ => return false, - }; - - let command = String::from_utf8_lossy(&output.stdout); - command_matches_launcher(record, &command) -} - -#[cfg(unix)] -fn command_matches_launcher(record: &LauncherRecord, command: &str) -> bool { - let run_dir = record.run_dir.to_string_lossy(); - let old_match = command.contains("__detached") && command.contains(run_dir.as_ref()); - let run_id = record.run_id.to_string(); - let new_match = command.contains(&format!("fabro: {}", super::short_run_id(&run_id))); - old_match || new_match -} - -#[cfg(not(unix))] -fn launcher_process_matches(_record: &LauncherRecord) -> bool { - true -} - -#[cfg(test)] -mod tests { - use super::*; - use chrono::Utc; - use fabro_types::fixtures; - - #[test] - fn active_launcher_record_for_run_removes_stale_record() { - let dir = tempfile::tempdir().unwrap(); - let storage_dir = dir.path().join("storage"); - let run_dir = storage_dir.join("runs").join("run"); - std::fs::create_dir_all(&run_dir).unwrap(); - - let launcher_path = launcher_record_path(&storage_dir, &fixtures::RUN_1); - write_launcher_record( - &launcher_path, - &LauncherRecord { - run_id: fixtures::RUN_1, - run_dir: run_dir.clone(), - pid: u32::MAX, - resume: false, - log_path: dir.path().join("launcher.log"), - started_at: Utc::now(), - }, - ) - .unwrap(); - - assert!(active_launcher_record_for_run(&run_dir).is_none()); - assert!(active_launcher_record(&storage_dir, &fixtures::RUN_1).is_none()); - assert!(!launcher_path.exists()); - } - - #[test] - fn active_launcher_record_for_run_falls_back_without_run_json() { - let dir = tempfile::tempdir().unwrap(); - let storage_dir = dir.path().join("storage"); - let run_dir = storage_dir.join("runs").join("20260401-test"); - std::fs::create_dir_all(&run_dir).unwrap(); - - let launcher_path = launcher_record_path(&storage_dir, &fixtures::RUN_1); - write_launcher_record( - &launcher_path, - &LauncherRecord { - run_id: fixtures::RUN_1, - run_dir: run_dir.clone(), - pid: u32::MAX, - resume: false, - log_path: dir.path().join("launcher.log"), - started_at: Utc::now(), - }, - ) - .unwrap(); - - assert!(active_launcher_record_for_run(&run_dir).is_none()); - assert!(!launcher_path.exists()); - } - - #[cfg(unix)] - #[test] - fn command_matches_launcher_accepts_old_detached_format() { - let dir = tempfile::tempdir().unwrap(); - let record = LauncherRecord { - run_id: fixtures::RUN_2, - run_dir: dir.path().join("run"), - pid: 42, - resume: false, - log_path: dir.path().join("launcher.log"), - started_at: Utc::now(), - }; - - let command = format!( - "/usr/local/bin/fabro __detached --run-dir {} --launcher-path /tmp/launcher.json", - record.run_dir.display() - ); - - assert!(command_matches_launcher(&record, &command)); - } - - #[cfg(unix)] - #[test] - fn command_matches_launcher_accepts_new_title_format() { - let dir = tempfile::tempdir().unwrap(); - let record = LauncherRecord { - run_id: fixtures::RUN_3, - run_dir: dir.path().join("run"), - pid: 42, - resume: false, - log_path: dir.path().join("launcher.log"), - started_at: Utc::now(), - }; - - assert!(command_matches_launcher( - &record, - &format!("fabro: {} plan", short_run_id(&record.run_id.to_string())) - )); - } -} diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 7697e60aa..43b0c045b 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -7,20 +7,19 @@ use chrono::{DateTime, Utc}; use fabro_util::json::normalize_json_value; use fabro_util::redact::redact_jsonl_line; use fabro_util::terminal::Styles; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use tokio::time; use tracing::{debug, info}; use crate::args::{GlobalArgs, LogsArgs}; use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run)?; + let client = lookup.client(); let run_id = run.run_id(); info!(run_id = %run_id, "Showing logs"); @@ -64,7 +63,7 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) return Ok(()); } follow_store_logs( - &client, + client, &run_id, if last_seq == 0 { 1 } else { last_seq + 1 }, pretty, diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 37ca2333b..817aae37b 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -1,9 +1,8 @@ use anyhow::Result; use fabro_util::terminal::Styles; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use crate::args::{GlobalArgs, RunArgs, RunCommands}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::print_json_pretty; use crate::user_config::{load_user_settings_with_globals, user_layer_with_globals}; @@ -11,10 +10,8 @@ pub(crate) mod attach; pub(crate) mod command; pub(crate) mod cp; pub(crate) mod create; -pub(crate) mod detached; pub(crate) mod diff; pub(crate) mod fork; -pub(crate) mod launcher; pub(crate) mod logs; pub(crate) mod output; pub(crate) mod overrides; @@ -22,14 +19,11 @@ pub(crate) mod preview; pub(crate) mod resume; pub(crate) mod rewind; pub(crate) mod run_progress; +pub(crate) mod runner; pub(crate) mod ssh; pub(crate) mod start; pub(crate) mod wait; -pub(super) fn short_run_id(id: &str) -> &str { - if id.len() > 12 { &id[..12] } else { id } -} - fn apply_json_defaults(args: &mut RunArgs, globals: &GlobalArgs) { if globals.json { args.auto_approve = true; @@ -56,10 +50,8 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( } RunCommands::Start { run } => { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run_info = resolve_run_from_summaries(&summaries, &base, &run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run_info = lookup.resolve(&run)?; let run_id = run_info.run_id(); start::start_run(&run_id, &cli_settings.storage_dir(), false).await?; if globals.json { @@ -70,10 +62,8 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( RunCommands::Attach { run } => { let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run_info = resolve_run_from_summaries(&summaries, &base, &run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run_info = lookup.resolve(&run)?; let run_id = run_info.run_id(); let exit_code = attach::attach_run( &run_info.path, @@ -81,7 +71,6 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( Some(&run_id), false, styles, - None, globals.json, ) .await?; @@ -90,20 +79,8 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( } Ok(()) } - RunCommands::Detached { - run_id, - run_dir, - launcher_path, - resume, - } => { - detached::execute( - run_id, - run_dir, - globals.storage_dir.clone(), - launcher_path, - resume, - ) - .await + RunCommands::Runner { run_id, resume } => { + runner::execute(run_id, globals.storage_dir.clone(), resume).await } RunCommands::Diff(args) => diff::run(args, globals).await, RunCommands::Logs(args) => { diff --git a/lib/crates/fabro-cli/src/commands/run/preview.rs b/lib/crates/fabro-cli/src/commands/run/preview.rs index cc245686d..702198a66 100644 --- a/lib/crates/fabro-cli/src/commands/run/preview.rs +++ b/lib/crates/fabro-cli/src/commands/run/preview.rs @@ -1,20 +1,18 @@ use anyhow::{Context, Result}; use fabro_sandbox::daytona::DaytonaSandbox; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use tracing::info; use crate::args::{GlobalArgs, PreviewArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{print_json_pretty, validate_daytona_provider}; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn run(args: PreviewArgs, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run)?; - let record = client + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run)?; + let record = lookup + .client() .get_run_state(&run.run_id()) .await? .sandbox diff --git a/lib/crates/fabro-cli/src/commands/run/resume.rs b/lib/crates/fabro-cli/src/commands/run/resume.rs index 59b76230e..bc339e833 100644 --- a/lib/crates/fabro-cli/src/commands/run/resume.rs +++ b/lib/crates/fabro-cli/src/commands/run/resume.rs @@ -1,15 +1,14 @@ use fabro_util::terminal::Styles; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use crate::args::{GlobalArgs, ResumeArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; /// Resume an interrupted workflow run. /// /// Looks up the run by ID prefix, validates a checkpoint exists, cleans stale -/// artifacts from the previous execution, then spawns an engine subprocess +/// artifacts from the previous execution, then asks the server to resume it /// (identical to `fabro run`'s create→start→attach flow). pub(crate) async fn resume_command( args: ResumeArgs, @@ -17,10 +16,8 @@ pub(crate) async fn resume_command( globals: &GlobalArgs, ) -> anyhow::Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run)?; let run_id = run.run_id(); let run_dir = run.path; @@ -39,7 +36,6 @@ pub(crate) async fn resume_command( Some(&run_id), true, styles, - None, globals.json, ) .await?; diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index 407870da2..3e4d7d0dc 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -10,14 +10,13 @@ use fabro_workflow::git::MetadataStore; use fabro_workflow::operations::{ RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, rewind, }; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use git2::Repository; use serde::Serialize; use crate::args::{GlobalArgs, RewindArgs}; use crate::commands::store::rebuild::rebuild_run_store; -use crate::server_client; use crate::server_client::ServerStoreClient; +use crate::server_runs::ServerRunLookup; use crate::shared::{color_if, print_json_pretty}; use crate::user_config::load_user_settings_with_globals; @@ -32,13 +31,11 @@ pub(crate) struct TimelineEntryJson { pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> { let repo = Repository::discover(".").context("not in a git repository")?; let cli_settings = load_user_settings_with_globals(globals)?; - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let base = runs_base(&cli_settings.storage_dir()); - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run_id)?; let run_id = run.run_id(); let store = Store::new(repo); - let events = client.list_run_events(&run_id, None, None).await?; + let events = lookup.client().list_run_events(&run_id, None, None).await?; let run_store = rebuild_run_store(&run_id, &events).await?; let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?; @@ -63,7 +60,7 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs }, )?; let entry = timeline.resolve(&target)?; - reset_rewound_run_state(&client, &store, &run_id, &run.path, entry).await?; + reset_rewound_run_state(lookup.client(), &store, &run_id, &run.path, entry).await?; let run_id_string = run_id.to_string(); diff --git a/lib/crates/fabro-cli/src/commands/run/detached.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs similarity index 87% rename from lib/crates/fabro-cli/src/commands/run/detached.rs rename to lib/crates/fabro-cli/src/commands/run/runner.rs index 9306fe076..d2c72bcdc 100644 --- a/lib/crates/fabro-cli/src/commands/run/detached.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -10,17 +10,11 @@ use crate::user_config::load_user_settings; pub(crate) async fn execute( run_id: RunId, - _run_dir: PathBuf, storage_dir: Option, - launcher_path: PathBuf, resume: bool, ) -> Result<()> { let _ = fabro_proc::title_init(); - let _launcher_guard = scopeguard::guard(launcher_path.clone(), |path| { - super::launcher::remove_launcher_record(&path); - }); - let storage_dir = match storage_dir { Some(storage_dir) => storage_dir, None => load_user_settings()?.storage_dir(), diff --git a/lib/crates/fabro-cli/src/commands/run/ssh.rs b/lib/crates/fabro-cli/src/commands/run/ssh.rs index cedcff36d..772931f51 100644 --- a/lib/crates/fabro-cli/src/commands/run/ssh.rs +++ b/lib/crates/fabro-cli/src/commands/run/ssh.rs @@ -1,10 +1,9 @@ use anyhow::{Context, Result, bail}; use fabro_sandbox::daytona::DaytonaSandbox; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use tracing::info; use crate::args::{GlobalArgs, SshArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{print_json_pretty, validate_daytona_provider}; use crate::user_config::load_user_settings_with_globals; @@ -14,12 +13,11 @@ pub(crate) async fn run(args: SshArgs, globals: &GlobalArgs) -> Result<()> { } let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run)?; let run_id = run.run_id(); - let record = client + let record = lookup + .client() .get_run_state(&run_id) .await? .sandbox diff --git a/lib/crates/fabro-cli/src/commands/run/wait.rs b/lib/crates/fabro-cli/src/commands/run/wait.rs index b0ddbd7eb..7cd327ce4 100644 --- a/lib/crates/fabro-cli/src/commands/run/wait.rs +++ b/lib/crates/fabro-cli/src/commands/run/wait.rs @@ -4,12 +4,11 @@ use anyhow::{Result, bail}; use fabro_types::RunId; use fabro_util::terminal::Styles; use fabro_workflow::records::Conclusion; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use fabro_workflow::run_status::RunStatus; use tracing::info; use crate::args::{GlobalArgs, WaitArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::format_duration_ms; use crate::user_config::load_user_settings_with_globals; @@ -20,10 +19,9 @@ const WAIT_STARTUP_GRACE: std::time::Duration = std::time::Duration::from_secs(3 pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run_info = resolve_run_from_summaries(&summaries, &base, &args.run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run_info = lookup.resolve(&args.run)?; + let client = lookup.client(); let run_id = run_info.run_id(); info!(run_id = %run_id, "Waiting for run to complete"); diff --git a/lib/crates/fabro-cli/src/commands/runs/inspect.rs b/lib/crates/fabro-cli/src/commands/runs/inspect.rs index 2e436558d..c132e1b52 100644 --- a/lib/crates/fabro-cli/src/commands/runs/inspect.rs +++ b/lib/crates/fabro-cli/src/commands/runs/inspect.rs @@ -4,12 +4,11 @@ use anyhow::Result; use fabro_types::RunId; use serde::Serialize; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; use fabro_workflow::run_status::RunStatus; use crate::args::{GlobalArgs, InspectArgs}; -use crate::server_client; use crate::server_client::RunProjection; +use crate::server_runs::ServerRunLookup; use crate::user_config::load_user_settings_with_globals; #[derive(Debug, Serialize)] @@ -26,12 +25,10 @@ pub(crate) struct InspectOutput { pub(crate) async fn run(args: &InspectArgs, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run)?; let run_id = run.run_id(); - let state = client.get_run_state(&run_id).await?; + let state = lookup.client().get_run_state(&run_id).await?; let output = inspect_run_state(&run_id, &run.path, run.status(), state); let json = serde_json::to_string_pretty(&[output])?; println!("{json}"); diff --git a/lib/crates/fabro-cli/src/commands/runs/list.rs b/lib/crates/fabro-cli/src/commands/runs/list.rs index 2979edb5e..7e8eb73a6 100644 --- a/lib/crates/fabro-cli/src/commands/runs/list.rs +++ b/lib/crates/fabro-cli/src/commands/runs/list.rs @@ -11,7 +11,7 @@ use fabro_workflow::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs use fabro_workflow::run_status::RunStatus; use crate::args::{GlobalArgs, RunsListArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{color_if, format_duration_ms, tilde_path}; use crate::user_config::load_user_settings_with_globals; @@ -25,9 +25,8 @@ pub(crate) async fn list_command( ) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let runs = scan_runs_with_summaries(&summaries, &base)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let runs = scan_runs_with_summaries(lookup.summaries(), &base)?; let label_filters = parse_label_filters(&args.filter.label); let filtered = filter_runs( &runs, diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index 6b76bc038..214c8b0c3 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -4,12 +4,13 @@ use anyhow::{Context, Result, bail}; use fabro_sandbox::reconnect::reconnect as reconnect_sandbox; use fabro_workflow::event::{Event, to_run_event}; use fabro_workflow::run_lookup::RunInfo; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; +use fabro_workflow::run_lookup::resolve_run_from_summaries; use tracing::warn; use crate::args::{GlobalArgs, RunsRemoveArgs}; use crate::server_client; use crate::server_client::RunProjection; +use crate::server_runs::ServerRunLookup; use crate::shared::print_json_pretty; use crate::user_config::load_user_settings_with_globals; @@ -17,10 +18,15 @@ use super::short_run_id; pub(crate) async fn remove_command(args: &RunsRemoveArgs, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - remove_from(args, &client, &summaries, &base, globals).await + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + remove_from( + args, + lookup.client(), + lookup.summaries(), + lookup.runs_base(), + globals, + ) + .await } async fn remove_from( diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 58b6f30b4..8f02a67a1 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -3,7 +3,6 @@ use anyhow::{Context, Result}; use fabro_store::StageId; use fabro_store::{RunProjection, SlateRunStore}; use fabro_workflow::run_dump::RunDump; -use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base}; #[cfg(test)] use serde::de::DeserializeOwned; use std::io::ErrorKind; @@ -11,18 +10,16 @@ use std::path::Path; use crate::args::{GlobalArgs, StoreDumpArgs}; use crate::commands::store::rebuild::rebuild_run_store; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{absolute_or_current, print_json_pretty}; use crate::user_config::load_user_settings_with_globals; pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; - let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - let run = resolve_run_from_summaries(&summaries, &base, &args.run)?; + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + let run = lookup.resolve(&args.run)?; let run_id = run.run_id(); - let events = client.list_run_events(&run_id, None, None).await?; + let events = lookup.client().list_run_events(&run_id, None, None).await?; let run_store = rebuild_run_store(&run_id, &events).await?; let file_count = export_run(&run_store, &args.output).await?; diff --git a/lib/crates/fabro-cli/src/commands/system/df.rs b/lib/crates/fabro-cli/src/commands/system/df.rs index 8e3cb840d..8b5038b46 100644 --- a/lib/crates/fabro-cli/src/commands/system/df.rs +++ b/lib/crates/fabro-cli/src/commands/system/df.rs @@ -10,7 +10,7 @@ use fabro_workflow::run_lookup::{logs_base, runs_base, scan_runs_with_summaries} use fabro_workflow::run_status::RunStatus; use crate::args::{DfArgs, GlobalArgs}; -use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{format_size, print_json_pretty}; use crate::user_config::load_user_settings_with_globals; @@ -49,11 +49,10 @@ pub(super) async fn df_command(args: &DfArgs, globals: &GlobalArgs) -> Result<() let data_dir = cli_settings.storage_dir(); let runs_base_dir = runs_base(&data_dir); let logs_base_dir = logs_base(&data_dir); - let client = server_client::connect_server(&data_dir).await?; - let summaries = client.list_store_runs().await?; + let lookup = ServerRunLookup::connect(&data_dir).await?; df_from( args, - &summaries, + lookup.summaries(), &data_dir, &runs_base_dir, &logs_base_dir, diff --git a/lib/crates/fabro-cli/src/commands/system/prune.rs b/lib/crates/fabro-cli/src/commands/system/prune.rs index 7a948aa76..df14820ce 100644 --- a/lib/crates/fabro-cli/src/commands/system/prune.rs +++ b/lib/crates/fabro-cli/src/commands/system/prune.rs @@ -10,6 +10,7 @@ use fabro_workflow::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs use crate::args::{GlobalArgs, RunsPruneArgs}; use crate::commands::runs::rm::remove_run_with_cleanup; use crate::server_client; +use crate::server_runs::ServerRunLookup; use crate::shared::{format_size, print_json_pretty}; use crate::user_config::load_user_settings_with_globals; @@ -24,9 +25,8 @@ struct PruneRunRow { pub(super) async fn prune_command(args: &RunsPruneArgs, globals: &GlobalArgs) -> Result<()> { let cli_settings = load_user_settings_with_globals(globals)?; let base = runs_base(&cli_settings.storage_dir()); - let client = server_client::connect_server(&cli_settings.storage_dir()).await?; - let summaries = client.list_store_runs().await?; - prune_from(args, &client, &summaries, &base, globals).await + let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?; + prune_from(args, lookup.client(), lookup.summaries(), &base, globals).await } pub(crate) fn parse_duration(s: &str) -> Result { diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 051ef6fdd..8e969b878 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -4,6 +4,7 @@ mod args; mod commands; mod logging; mod server_client; +mod server_runs; mod shared; #[cfg(feature = "sleep_inhibitor")] mod sleep_inhibitor; @@ -421,31 +422,17 @@ mod tests { } #[test] - fn parse_detached_command() { + fn parse_runner_command() { let cli = Cli::try_parse_from([ "fabro", - "__detached", + "__runner", "--run-id", "01ARZ3NDEKTSV4RRFFQ69G5FAV", - "--run-dir", - "/tmp/fabro/runs/01ABC", - "--launcher-path", - "/tmp/fabro/launchers/01ABC.json", ]) .expect("should parse"); match *cli.command { - Commands::RunCmd(RunCommands::Detached { - run_id, - run_dir, - launcher_path, - resume, - }) => { + Commands::RunCmd(RunCommands::Runner { run_id, resume }) => { assert_eq!(run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap()); - assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC")); - assert_eq!( - launcher_path, - std::path::PathBuf::from("/tmp/fabro/launchers/01ABC.json") - ); assert!(!resume); } _ => panic!("unexpected command variant"), @@ -453,32 +440,18 @@ mod tests { } #[test] - fn parse_detached_with_resume() { + fn parse_runner_with_resume() { let cli = Cli::try_parse_from([ "fabro", - "__detached", + "__runner", "--run-id", "01ARZ3NDEKTSV4RRFFQ69G5FAV", - "--run-dir", - "/tmp/fabro/runs/01ABC", - "--launcher-path", - "/tmp/fabro/launchers/01ABC.json", "--resume", ]) .expect("should parse"); match *cli.command { - Commands::RunCmd(RunCommands::Detached { - run_id, - run_dir, - launcher_path, - resume, - }) => { + Commands::RunCmd(RunCommands::Runner { run_id, resume }) => { assert_eq!(run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap()); - assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC")); - assert_eq!( - launcher_path, - std::path::PathBuf::from("/tmp/fabro/launchers/01ABC.json") - ); assert!(resume); } _ => panic!("unexpected command variant"), diff --git a/lib/crates/fabro-cli/src/server_runs.rs b/lib/crates/fabro-cli/src/server_runs.rs new file mode 100644 index 000000000..e3e24ecfa --- /dev/null +++ b/lib/crates/fabro-cli/src/server_runs.rs @@ -0,0 +1,46 @@ +use std::path::{Path, PathBuf}; + +use anyhow::Result; +use fabro_store::RunSummary; +use fabro_workflow::run_lookup::{RunInfo, resolve_run_from_summaries, runs_base}; + +use crate::server_client::{self, ServerStoreClient}; + +pub(crate) struct ServerRunLookup { + client: ServerStoreClient, + runs_base: PathBuf, + summaries: Vec, +} + +impl ServerRunLookup { + pub(crate) async fn connect(storage_dir: &Path) -> Result { + Self::connect_from_runs_base(&runs_base(storage_dir)).await + } + + pub(crate) async fn connect_from_runs_base(runs_base: &Path) -> Result { + let storage_dir = runs_base.parent().unwrap_or(runs_base); + let client = server_client::connect_server(storage_dir).await?; + let summaries = client.list_store_runs().await?; + Ok(Self { + client, + runs_base: runs_base.to_path_buf(), + summaries, + }) + } + + pub(crate) fn client(&self) -> &ServerStoreClient { + &self.client + } + + pub(crate) fn runs_base(&self) -> &Path { + &self.runs_base + } + + pub(crate) fn summaries(&self) -> &[RunSummary] { + &self.summaries + } + + pub(crate) fn resolve(&self, selector: &str) -> Result { + resolve_run_from_summaries(&self.summaries, &self.runs_base, selector) + } +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/fabro.rs b/lib/crates/fabro-cli/tests/it/cmd/fabro.rs index 6fb94e38b..371aba9a6 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/fabro.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/fabro.rs @@ -14,7 +14,7 @@ fn help() { Commands: run Launch a workflow run create Create a workflow run (allocate run dir, persist spec) - start Start a created workflow run (spawn engine process) + start Start a created workflow run on the server attach Attach to a running or finished workflow run logs View the event log of a workflow run resume Resume an interrupted workflow run diff --git a/lib/crates/fabro-cli/tests/it/cmd/logs.rs b/lib/crates/fabro-cli/tests/it/cmd/logs.rs index 5664d8d50..9961f26bb 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/logs.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/logs.rs @@ -8,7 +8,9 @@ fn parse_ndjson(stdout: &[u8]) -> Vec { .expect("stdout should be valid UTF-8") .lines() .filter(|line| !line.trim().is_empty()) - .map(|line| serde_json::from_str::(line).expect("logs output should be valid NDJSON")) + .map(|line| { + serde_json::from_str::(line).expect("logs output should be valid NDJSON") + }) .collect() } @@ -22,7 +24,8 @@ fn assert_event_sequence_contains(events: &[Value], expected: &[&str]) { for expected_name in expected { let Some(found_at) = event_names[cursor..] .iter() - .position(|name| name == expected_name) else { + .position(|name| name == expected_name) + else { panic!("missing event {expected_name} in sequence: {event_names:?}"); }; cursor += found_at + 1; diff --git a/lib/crates/fabro-cli/tests/it/cmd/mod.rs b/lib/crates/fabro-cli/tests/it/cmd/mod.rs index 453a90406..84875457a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mod.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mod.rs @@ -5,7 +5,6 @@ mod attach; mod completion; mod config; mod create; -mod detached; mod diff; mod discord; mod docs; @@ -41,6 +40,7 @@ mod resume; mod rewind; mod rm; mod run; +mod runner; mod sandbox_cp; mod sandbox_preview; mod sandbox_ssh; diff --git a/lib/crates/fabro-cli/tests/it/cmd/resume.rs b/lib/crates/fabro-cli/tests/it/cmd/resume.rs index 5096e0aa5..033b2074c 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/resume.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/resume.rs @@ -2,6 +2,8 @@ use fabro_test::{fabro_snapshot, test_context}; use super::support::{git_stdout, output_stderr, setup_git_backed_changed_run}; +const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + #[test] fn help() { let context = test_context!(); @@ -115,3 +117,44 @@ fn resume_rewound_run_succeeds() { ); assert_ne!(resumed_head.trim(), rewound_head.trim()); } + +#[test] +fn resume_detached_does_not_create_launcher_record() { + let context = test_context!(); + let setup = setup_git_backed_changed_run(&context); + + context + .command() + .current_dir(&setup.repo_dir) + .args(["rewind", &setup.run.run_id, "@1", "--no-push"]) + .assert() + .success(); + + let mut resume_cmd = context.command(); + resume_cmd.current_dir(&setup.repo_dir); + resume_cmd.env("OPENAI_API_KEY", "test"); + resume_cmd.args(["resume", "--detach", &setup.run.run_id]); + let resume_output = resume_cmd.output().expect("resume should execute"); + assert!( + resume_output.status.success(), + "resume should succeed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&resume_output.stdout), + output_stderr(&resume_output) + ); + + assert!( + !context + .storage_dir + .join("launchers") + .join(format!("{}.json", setup.run.run_id)) + .exists(), + "server-owned resume should not create a launcher record" + ); + + context + .command() + .args(["wait", &setup.run.run_id]) + .timeout(SHARED_DAEMON_TIMEOUT) + .assert() + .success(); +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index b9c45e283..95755543d 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -1,8 +1,15 @@ use fabro_test::{fabro_snapshot, test_context}; +use fabro_types::StatusReason; use serde_json::Value; +use super::support::{ + only_run, output_stderr, run_count_for_test_case, run_state, wait_for_status, + write_gated_workflow, +}; use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id}; +const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + #[test] fn help() { let context = test_context!(); @@ -242,7 +249,10 @@ fn json_run_implies_auto_approve_for_human_gates() { let Some(llm) = llm.as_object_mut() else { continue; }; - llm.insert("model".to_string(), Value::String("[LLM_MODEL]".to_string())); + llm.insert( + "model".to_string(), + Value::String("[LLM_MODEL]".to_string()), + ); llm.insert( "provider".to_string(), Value::String("[LLM_PROVIDER]".to_string()), @@ -961,3 +971,69 @@ fn detach_creates_run_dir_with_detach_log() { "# ); } + +#[test] +fn ctrl_c_cancels_active_run_via_server() { + let context = test_context!(); + let _gate = write_gated_workflow(&context.temp_dir.join("slow.fabro"), "slow", "Run slowly"); + + let mut run_cmd = std::process::Command::new(env!("CARGO_BIN_EXE_fabro")); + run_cmd.current_dir(&context.temp_dir); + run_cmd.env("NO_COLOR", "1"); + run_cmd.env("HOME", &context.home_dir); + run_cmd.env("FABRO_NO_UPGRADE_CHECK", "true"); + run_cmd.env("FABRO_STORAGE_DIR", &context.storage_dir); + run_cmd.env("FABRO_SERVER_MAX_CONCURRENT_RUNS", "64"); + run_cmd.env("OPENAI_API_KEY", "test"); + run_cmd.args([ + "run", + "--label", + &context.test_run_label(), + "--label", + &context.test_case_label(), + "--provider", + "openai", + "--sandbox", + "local", + "--no-retro", + "slow.fabro", + ]); + let child = run_cmd.spawn().expect("run should spawn"); + + let deadline = std::time::Instant::now() + SHARED_DAEMON_TIMEOUT; + while run_count_for_test_case(&context) == 0 { + assert!( + std::time::Instant::now() < deadline, + "timed out waiting for run directory" + ); + std::thread::sleep(std::time::Duration::from_millis(50)); + } + + let run = only_run(&context); + wait_for_status(&run.run_dir, &["running"]); + + let kill_status = std::process::Command::new("kill") + .args(["-INT", &child.id().to_string()]) + .status() + .expect("kill should execute"); + assert!(kill_status.success(), "kill -INT should succeed"); + + let output = child + .wait_with_output() + .expect("run should exit after SIGINT"); + assert!( + !output.status.success(), + "run should exit non-zero after cancellation\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + output_stderr(&output) + ); + + let final_status = wait_for_status(&run.run_dir, &["failed"]); + assert_eq!(final_status, "failed"); + assert_eq!( + run_state(&run.run_dir) + .status + .and_then(|record| record.reason), + Some(StatusReason::Cancelled) + ); +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/detached.rs b/lib/crates/fabro-cli/tests/it/cmd/runner.rs similarity index 70% rename from lib/crates/fabro-cli/tests/it/cmd/detached.rs rename to lib/crates/fabro-cli/tests/it/cmd/runner.rs index c83d573ac..ae9d7bc08 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/detached.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/runner.rs @@ -9,41 +9,32 @@ const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_sec fn help() { let context = test_context!(); let mut cmd = context.command(); - cmd.args(["__detached", "--help"]); + cmd.args(["__runner", "--help"]); fabro_snapshot!(context.filters(), cmd, @" success: true exit_code: 0 ----- stdout ----- - Internal: run the engine process + Internal: queue or resume a workflow run via the server - Usage: fabro __detached [OPTIONS] --run-id --run-dir --launcher-path + Usage: fabro __runner [OPTIONS] --run-id Options: - --json Output as JSON [env: FABRO_JSON=] - --run-id Run ID - --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] - --run-dir Run directory - --launcher-path Launcher metadata path - --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] - --quiet Suppress non-essential output [env: FABRO_QUIET=] - --resume Resume from checkpoint instead of fresh start - --verbose Enable verbose output [env: FABRO_VERBOSE=] - --storage-dir Storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]] - --server-url Server URL (overrides server.base_url from user.toml) [env: FABRO_SERVER_URL=] - -h, --help Print help + --json Output as JSON [env: FABRO_JSON=] + --run-id Run ID + --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] + --resume Resume from checkpoint instead of fresh start + --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] + --quiet Suppress non-essential output [env: FABRO_QUIET=] + --verbose Enable verbose output [env: FABRO_VERBOSE=] + --storage-dir Storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]] + --server-url Server URL (overrides server.base_url from user.toml) [env: FABRO_SERVER_URL=] + -h, --help Print help ----- stderr ----- "); } -fn launcher_path(context: &fabro_test::TestContext, run_id: &str) -> std::path::PathBuf { - context - .storage_dir - .join("launchers") - .join(format!("{run_id}.json")) -} - #[test] -fn detached_uses_cached_graph_after_source_deleted() { +fn runner_uses_cached_graph_after_source_deleted() { let context = test_context!(); let run_id = unique_run_id(); let workflow_path = context.temp_dir.join("workflow.fabro"); @@ -77,15 +68,7 @@ digraph CachedGraph { context .command() - .args([ - "__detached", - "--run-id", - run_id.as_str(), - "--run-dir", - run_dir.to_str().unwrap(), - "--launcher-path", - launcher_path(&context, &run_id).to_str().unwrap(), - ]) + .args(["__runner", "--run-id", run_id.as_str()]) .timeout(SHARED_DAEMON_TIMEOUT) .assert() .success(); @@ -110,7 +93,7 @@ digraph CachedGraph { } #[test] -fn detached_uses_snapshotted_app_id_for_github_credentials() { +fn runner_uses_snapshotted_app_id_for_github_credentials() { let context = test_context!(); let run_id = unique_run_id(); let workflow_path = context.temp_dir.join("workflow.fabro"); @@ -167,15 +150,7 @@ digraph GitHubApp { let mut cmd = context.command(); cmd.env("GITHUB_APP_PRIVATE_KEY", "%%%not-base64%%%"); - cmd.args([ - "__detached", - "--run-id", - run_id.as_str(), - "--run-dir", - run_dir.to_str().unwrap(), - "--launcher-path", - launcher_path(&context, &run_id).to_str().unwrap(), - ]); + cmd.args(["__runner", "--run-id", run_id.as_str()]); cmd.timeout(SHARED_DAEMON_TIMEOUT); fabro_snapshot!(context.filters(), cmd, @" success: true @@ -186,7 +161,7 @@ digraph GitHubApp { } #[test] -fn detached_runs_without_run_json_when_run_id_is_explicit() { +fn runner_runs_without_run_json_when_run_id_is_explicit() { let context = test_context!(); let run_id = unique_run_id(); let workflow_path = context.temp_dir.join("workflow.fabro"); @@ -218,15 +193,7 @@ digraph DetachedStoreOnly { let run_dir = context.find_run_dir(&run_id); context .command() - .args([ - "__detached", - "--run-id", - run_id.as_str(), - "--run-dir", - run_dir.to_str().unwrap(), - "--launcher-path", - launcher_path(&context, &run_id).to_str().unwrap(), - ]) + .args(["__runner", "--run-id", run_id.as_str()]) .timeout(SHARED_DAEMON_TIMEOUT) .assert() .success(); @@ -251,7 +218,7 @@ digraph DetachedStoreOnly { } #[test] -fn detached_resume_rejects_completed_run_without_mutating_it() { +fn runner_resume_rejects_completed_run_without_mutating_it() { let context = test_context!(); context.write_temp( "workflow.fabro", @@ -301,7 +268,6 @@ digraph Test { "conclusion_timestamp": before[0]["conclusion"]["timestamp"], "conclusion_status": before[0]["conclusion"]["status"], }); - let run_dir = before_summary["run_dir"].as_str().unwrap().to_string(); fabro_json_snapshot!(context, &before_summary, @r#" { "run_dir": "[RUN_DIR]", @@ -312,16 +278,7 @@ digraph Test { "#); let mut cmd = context.command(); - cmd.args([ - "__detached", - "--run-id", - &run_id, - "--run-dir", - &run_dir, - "--launcher-path", - launcher_path(&context, &run_id).to_str().unwrap(), - "--resume", - ]); + cmd.args(["__runner", "--run-id", &run_id, "--resume"]); cmd.timeout(SHARED_DAEMON_TIMEOUT); fabro_snapshot!(context.filters(), cmd, @" success: true diff --git a/lib/crates/fabro-cli/tests/it/cmd/start.rs b/lib/crates/fabro-cli/tests/it/cmd/start.rs index 8a1c4b234..82bb5a2be 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/start.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/start.rs @@ -15,7 +15,7 @@ fn help() { success: true exit_code: 0 ----- stdout ----- - Start a created workflow run (spawn engine process) + Start a created workflow run on the server Usage: fabro start [OPTIONS] diff --git a/lib/crates/fabro-cli/tests/it/scenario/artifacts.rs b/lib/crates/fabro-cli/tests/it/scenario/artifacts.rs index ca62560e5..adecd4c4d 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/artifacts.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/artifacts.rs @@ -165,7 +165,12 @@ fn artifact_commands_share_populated_run_fixture() { let flat_dest = context.temp_dir.join("artifact-flat"); let mut cp_flat = context.command(); - cp_flat.args(["artifact", "cp", &setup.run.run_id, flat_dest.to_str().unwrap()]); + cp_flat.args([ + "artifact", + "cp", + &setup.run.run_id, + flat_dest.to_str().unwrap(), + ]); fabro_snapshot!(context.filters(), cp_flat, @" success: false exit_code: 1