scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix(cli): route shared-test run lookup through server

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-05 03:02:16 -04:00
parent 50aff2787b
commit 4423e65e4a
No known key found for this signature in database
5 changed files with 198 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

120
lib/crates/fabro-cli/src/commands/run/attach.rs
View file

@ -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<ExitCode> {
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<u32> = 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::<fabro_interview::Question>(&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<bool> {
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,

39
lib/crates/fabro-cli/src/server_client.rs
View file

@ -235,6 +235,45 @@ impl ServerStoreClient {
.collect::<Result<Vec<_>>>()
}
pub(crate) async fn list_run_questions(
&self,
run_id: &RunId,
) -> Result<Vec<types::ApiQuestion>> {
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<String>,
selected_option_key: Option<String>,
selected_option_keys: Vec<String>,
) -> 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

54
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -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<Value> = 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<Value> = 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"
);

27
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -528,8 +528,13 @@ pub(crate) fn run_count_for_test_case(context: &TestContext) -> usize {
}
fn run_dirs_for_test_case(context: &TestContext) -> Vec<PathBuf> {
let runs: Vec<RunSummaryRecord> =
block_on(get_server_json_for_storage(&context.storage_dir, "/api/v1/runs"));
let runs: Option<Vec<RunSummaryRecord>> = 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<T: serde::de::DeserializeOwned>(run_dir: &Path, path: &
get_server_json_for_storage(storage_dir, path).await
}
async fn try_get_server_json_for_storage<T: serde::de::DeserializeOwned>(
storage_dir: &Path,
path: &str,
) -> Option<T> {
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::<T>().await.ok()
}
async fn get_server_json_for_storage<T: serde::de::DeserializeOwned>(
storage_dir: &Path,
path: &str,

21
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -347,15 +347,20 @@ fn resolve_run_from_infos(runs: &[RunInfo], identifier: &str) -> Result<RunInfo>
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()),