From 4423e65e4a3884487d4677764d4cc61e730f6e01 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 5 Apr 2026 03:02:16 -0400 Subject: [PATCH] fix(cli): route shared-test run lookup through server --- .../fabro-cli/src/commands/run/attach.rs | 120 +++++++++++------- lib/crates/fabro-cli/src/server_client.rs | 39 ++++++ lib/crates/fabro-cli/tests/it/cmd/attach.rs | 54 ++++++-- lib/crates/fabro-cli/tests/it/cmd/support.rs | 27 +++- lib/crates/fabro-workflow/src/run_lookup.rs | 21 +-- 5 files changed, 198 insertions(+), 63 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 88ba3efa7..6873a8c36 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -8,7 +8,8 @@ use std::time::{Duration, Instant}; use anyhow::Result; use fabro_types::{EventBody, RunEvent, RunId}; -use fabro_interview::{AnswerValue, ConsoleInterviewer}; +use fabro_api::types; +use fabro_interview::{AnswerValue, ConsoleInterviewer, Question, QuestionOption, QuestionType}; use fabro_store::{EventEnvelope, RuntimeState}; use fabro_util::json::normalize_json_value; use fabro_util::terminal::Styles; @@ -97,7 +98,6 @@ async fn attach_run_server( json_output: bool, ) -> Result { let runtime_state = RuntimeState::new(run_dir); - let runtime_interview_paths = InterviewPaths::from_runtime_state(&runtime_state); let mut engine_guard = engine_child.map(EngineChildGuard::new); @@ -118,6 +118,12 @@ async fn attach_run_server( emit_progress_line(&mut progress_ui, line, json_output)?; } + 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(); @@ -181,50 +187,31 @@ async fn attach_run_server( continue; } - // Check for interview request - if runtime_interview_paths.request_path.exists() { - let interview_paths = &runtime_interview_paths; - if !interview_paths.response_path.exists() { - if json_output { - defuse_engine_child(&mut engine_guard); - eprintln!("{JSON_INTERVIEW_MESSAGE}"); - return Ok(ExitCode::from(1)); - } - if let Some(_claim_guard) = - InterviewClaimGuard::acquire(&interview_paths.claim_path) - { - if let Ok(request_data) = std::fs::read_to_string(&interview_paths.request_path) - { - if let Ok(question) = - serde_json::from_str::(&request_data) - { - // Hide progress bars during interview - hide_progress(&mut progress_ui, json_output); - - // Prompt user via ConsoleInterviewer - let interviewer = ConsoleInterviewer::new(styles); - let answer = - fabro_interview::Interviewer::ask(&interviewer, question).await; - - // Show progress bars again before any return path. - 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)); - } - - write_interview_response_atomically( - &interview_paths.response_path, - &answer, - )?; - } - } - } + // 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)); } + + 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) { + if let Some(guard) = engine_guard.as_mut() { + guard.defuse(); + } + 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 @@ -289,6 +276,49 @@ async fn attach_run_server( }) } +fn api_question_to_question(question: &types::ApiQuestion) -> Question { + let question_type = match question.question_type { + types::QuestionType::YesNo => QuestionType::YesNo, + types::QuestionType::MultipleChoice => QuestionType::MultipleChoice, + types::QuestionType::MultiSelect => QuestionType::MultiSelect, + types::QuestionType::Freeform => QuestionType::Freeform, + types::QuestionType::Confirmation => QuestionType::Confirmation, + }; + let mut converted = Question::new(question.text.clone(), question_type); + converted.options = question + .options + .iter() + .map(|option| QuestionOption { + key: option.key.clone(), + label: option.label.clone(), + }) + .collect(); + converted.allow_freeform = question.allow_freeform; + converted +} + +async fn submit_server_interview_answer( + client: &server_client::ServerStoreClient, + run_id: &RunId, + qid: &str, + answer: &fabro_interview::Answer, +) -> Result { + let (value, selected_option_key, selected_option_keys) = match &answer.value { + AnswerValue::Text(text) => (Some(text.clone()), None, Vec::new()), + AnswerValue::Selected(key) => (None, Some(key.clone()), Vec::new()), + AnswerValue::MultiSelected(keys) => (None, None, keys.clone()), + AnswerValue::Yes => (Some("yes".to_string()), None, Vec::new()), + AnswerValue::No => (Some("no".to_string()), None, Vec::new()), + AnswerValue::Aborted | AnswerValue::Skipped | AnswerValue::Timeout => { + return Ok(false); + } + }; + client + .submit_run_answer(run_id, qid, value, selected_option_key, selected_option_keys) + .await?; + Ok(true) +} + async fn flush_remaining_server_events( client: &server_client::ServerStoreClient, run_id: &RunId, diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index 1850335cf..03be42009 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -235,6 +235,45 @@ impl ServerStoreClient { .collect::>>() } + pub(crate) async fn list_run_questions( + &self, + run_id: &RunId, + ) -> Result> { + let response = self + .client + .list_run_questions() + .id(run_id.to_string()) + .page_limit(100) + .page_offset(0) + .send() + .await + .map_err(map_api_error)?; + Ok(response.into_inner().data) + } + + pub(crate) async fn submit_run_answer( + &self, + run_id: &RunId, + qid: &str, + value: Option, + selected_option_key: Option, + selected_option_keys: Vec, + ) -> Result<()> { + self.client + .submit_run_answer() + .id(run_id.to_string()) + .qid(qid) + .body(types::SubmitAnswerRequest { + value, + selected_option_key, + selected_option_keys, + }) + .send() + .await + .map_err(map_api_error)?; + Ok(()) + } + pub(crate) async fn append_run_event(&self, run_id: &RunId, event: &RunEvent) -> Result<()> { let body: types::RunEvent = convert_type(event)?; self.client diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index bc0ce8cb8..9f0616e53 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -250,14 +250,30 @@ fn attach_json_errors_without_prompting_for_human_input() { scopeguard::defer! { let _ = context.command().args(["rm", "--force", &cleanup_run_id]).output(); } - let run_dir = context.find_run_dir(&run_id); - - let request_path = run_dir.join("runtime/interview_request.json"); let deadline = std::time::Instant::now() + SHARED_DAEMON_TIMEOUT; - while !request_path.exists() { + loop { + let logs_output = context + .command() + .args(["logs", &run_id, "--json"]) + .output() + .expect("logs should execute"); + assert!(logs_output.status.success(), "logs should succeed"); + let log_events: Vec = String::from_utf8(logs_output.stdout) + .expect("stdout should be UTF-8") + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str(line).expect("log line should be valid JSON")) + .collect(); + if log_events.iter().any(|event| { + event["event"] == "stage.started" + && event["node_id"] == "approve" + && event["properties"]["handler_type"] == "human" + }) { + break; + } assert!( std::time::Instant::now() < deadline, - "timed out waiting for interview request for {run_id}" + "timed out waiting for human gate to start for {run_id}" ); std::thread::sleep(std::time::Duration::from_millis(50)); } @@ -276,12 +292,34 @@ fn attach_json_errors_without_prompting_for_human_input() { !stderr.contains("Approve?"), "attach should not prompt on stderr" ); + let logs_output = context + .command() + .args(["logs", &run_id, "--json"]) + .output() + .expect("logs should execute"); + assert!(logs_output.status.success(), "logs should succeed"); + let log_events: Vec = String::from_utf8(logs_output.stdout) + .expect("stdout should be UTF-8") + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| serde_json::from_str(line).expect("log line should be valid JSON")) + .collect(); assert!( - request_path.exists(), - "the run should still be waiting on the interview request" + log_events.iter().any(|event| { + event["event"] == "stage.started" + && event["node_id"] == "approve" + && event["properties"]["handler_type"] == "human" + }), + "the run should still be waiting on the human gate" ); assert!( - !run_dir.join("runtime/interview_response.json").exists(), + !log_events.iter().any(|event| { + event["node_id"] == "approve" + && matches!( + event["event"].as_str(), + Some("stage.completed" | "stage.failed" | "interview.completed") + ) + }), "attach --json should not answer the interview" ); diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 655e1a2d6..dd3659b91 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -528,8 +528,13 @@ pub(crate) fn run_count_for_test_case(context: &TestContext) -> usize { } fn run_dirs_for_test_case(context: &TestContext) -> Vec { - let runs: Vec = - block_on(get_server_json_for_storage(&context.storage_dir, "/api/v1/runs")); + let runs: Option> = block_on(try_get_server_json_for_storage( + &context.storage_dir, + "/api/v1/runs", + )); + let Some(runs) = runs else { + return Vec::new(); + }; runs.into_iter() .filter(|run| { run.labels @@ -623,6 +628,24 @@ async fn get_server_json(run_dir: &Path, path: & get_server_json_for_storage(storage_dir, path).await } +async fn try_get_server_json_for_storage( + storage_dir: &Path, + path: &str, +) -> Option { + if !storage_dir.join("fabro.sock").exists() { + return None; + } + let response = server_http_client(storage_dir) + .get(format!("http://fabro{path}")) + .send() + .await + .ok()?; + if !response.status().is_success() { + return None; + } + response.json::().await.ok() +} + async fn get_server_json_for_storage( storage_dir: &Path, path: &str, diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index adecc32e9..cf15ab78f 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -347,15 +347,20 @@ fn resolve_run_from_infos(runs: &[RunInfo], identifier: &str) -> Result let id_lower = identifier.to_lowercase(); let id_collapsed = collapse_separators(&id_lower); - let workflow_match = runs.iter().filter(|run| !run.is_orphan).find(|run| { - if let Some(slug) = run.workflow_slug() { - if slug.to_lowercase() == id_lower { - return true; + let workflow_match = runs + .iter() + .filter(|run| !run.is_orphan) + .filter(|run| { + if let Some(slug) = run.workflow_slug() { + if slug.to_lowercase() == id_lower { + return true; + } } - } - let name_lower = run.workflow_name().to_lowercase(); - name_lower.contains(&id_lower) || collapse_separators(&name_lower).contains(&id_collapsed) - }); + let name_lower = run.workflow_name().to_lowercase(); + name_lower.contains(&id_lower) + || collapse_separators(&name_lower).contains(&id_collapsed) + }) + .max_by_key(|run| run.run_id().created_at()); match workflow_match { Some(run) => Ok(run.clone()),