diff --git a/AGENTS.md b/AGENTS.md index 524542707..921bbc55b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -14,6 +14,8 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co - `cargo fmt --check --all` — check formatting - `cargo clippy --workspace -- -D warnings` — lint +macOS note: if `cargo nextest run` fails with `Too many open files (os error 24)` / `EMFILE`, raise the shell's soft FD limit before running tests, for example `ulimit -n 4096 && cargo nextest run --workspace`. Some terminals and inherited agent sessions start with `ulimit -n 256`, which is too low for the shared CLI test daemon under parallel nextest load. + ### TypeScript (fabro-web) - `cd apps/fabro-web && bun run dev` — start React dev server - `cd apps/fabro-web && bun test` — run tests diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index e682bb519..7c910e0d8 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -62,6 +62,10 @@ pub(crate) async fn attach_run_with_client( json_output: bool, ) -> Result { let state = client.get_run_state(run_id).await?; + let auto_approve = state + .run + .as_ref() + .is_some_and(|record| record.settings.auto_approve_enabled()); let verbose = state .run .as_ref() @@ -87,6 +91,7 @@ pub(crate) async fn attach_run_with_client( attach_live_run_with_client( client, run_id, + auto_approve, verbose, replay_events, stream, @@ -119,6 +124,7 @@ fn replay_run_with_client( async fn attach_live_run_with_client( client: &server_client::ServerStoreClient, run_id: &RunId, + auto_approve: bool, verbose: bool, existing_events: Vec, mut stream: server_client::RunAttachEventStream, @@ -136,9 +142,15 @@ async fn attach_live_run_with_client( emit_progress_line(&mut progress_ui, &line, json_output)?; } - if let Some(exit_code) = - handle_pending_server_interview(client, run_id, &mut progress_ui, styles, json_output) - .await? + if let Some(exit_code) = handle_pending_server_interview( + client, + run_id, + auto_approve, + &mut progress_ui, + styles, + json_output, + ) + .await? { return Ok(exit_code); } @@ -170,6 +182,7 @@ async fn attach_live_run_with_client( if let Some(exit_code) = handle_pending_server_interview( client, run_id, + auto_approve, &mut progress_ui, styles, json_output, @@ -185,6 +198,7 @@ async fn attach_live_run_with_client( async fn handle_pending_server_interview( client: &server_client::ServerStoreClient, run_id: &RunId, + auto_approve: bool, progress_ui: &mut run_progress::ProgressUI, styles: &'static Styles, json_output: bool, @@ -193,10 +207,13 @@ async fn handle_pending_server_interview( return Ok(None); }; - if json_output { + if json_pending_interview_requires_manual_input(json_output, auto_approve) { eprintln!("{JSON_INTERVIEW_MESSAGE}"); return Ok(Some(ExitCode::from(1))); } + if json_output { + return Ok(None); + } hide_progress(progress_ui, json_output); let interviewer = ConsoleInterviewer::new(styles); @@ -245,7 +262,7 @@ fn api_question_to_question(question: &types::ApiQuestion) -> Question { types::QuestionType::Confirmation => QuestionType::Confirmation, }; let mut converted = Question::new(question.text.clone(), question_type); - converted.id = question.id.clone(); + converted.id.clone_from(&question.id); converted.options = question .options .iter() @@ -255,9 +272,11 @@ fn api_question_to_question(question: &types::ApiQuestion) -> Question { }) .collect(); converted.allow_freeform = question.allow_freeform; - converted.stage = question.stage.clone(); + converted.stage.clone_from(&question.stage); converted.timeout_seconds = question.timeout_seconds; - converted.context_display = question.context_display.clone(); + converted + .context_display + .clone_from(&question.context_display); converted } @@ -289,6 +308,10 @@ async fn submit_server_interview_answer( Ok(true) } +fn json_pending_interview_requires_manual_input(json_output: bool, auto_approve: bool) -> bool { + json_output && !auto_approve +} + fn state_is_terminal(state: &server_client::RunProjection) -> bool { state.conclusion.is_some() || state @@ -493,4 +516,14 @@ mod tests { assert!(answer_requires_reattach(&skipped)); assert!(!answer_requires_reattach(&answered)); } + + #[test] + fn json_pending_interview_requires_manual_input_when_auto_approve_is_disabled() { + assert!(json_pending_interview_requires_manual_input(true, false)); + } + + #[test] + fn json_pending_interview_does_not_require_manual_input_when_auto_approve_is_enabled() { + assert!(!json_pending_interview_requires_manual_input(true, true)); + } } diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 16dd2f592..191c4a1bb 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -878,6 +878,31 @@ fn attach_json_errors_without_prompting_for_human_input() { }, "run_id": "[ULID]", "ts": "[TIMESTAMP]" + }, + { + "event": "interview.started", + "id": "[EVENT_ID]", + "node_id": "approve", + "node_label": "approve", + "properties": { + "allow_freeform": false, + "options": [ + { + "key": "A", + "label": "[A] Approve" + }, + { + "key": "R", + "label": "[R] Revise" + } + ], + "question": "Approve?", + "question_id": "[ULID]", + "question_type": "multiple_choice", + "stage": "approve" + }, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" } ] "#); diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 1a875ade5..350701a1c 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -955,6 +955,43 @@ fn json_run_implies_auto_approve_for_human_gates() { "run_id": "[ULID]", "ts": "[TIMESTAMP]" }, + { + "event": "interview.started", + "id": "[EVENT_ID]", + "node_id": "approve", + "node_label": "approve", + "properties": { + "allow_freeform": false, + "options": [ + { + "key": "A", + "label": "[A] Approve" + }, + { + "key": "R", + "label": "[R] Revise" + } + ], + "question": "Approve?", + "question_id": "[ULID]", + "question_type": "multiple_choice", + "stage": "approve" + }, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" + }, + { + "event": "interview.completed", + "id": "[EVENT_ID]", + "properties": { + "answer": "A", + "duration_ms": "[DURATION_MS]", + "question": "Approve?", + "question_id": "[ULID]" + }, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" + }, { "event": "stage.completed", "id": "[EVENT_ID]", diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 064eb956c..96562240d 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -59,7 +59,7 @@ use tokio::sync::broadcast::error::RecvError; use tokio::sync::mpsc; use tokio::sync::oneshot; use tokio::task::spawn_blocking; -use tokio::time::sleep; +use tokio::time::{sleep, timeout}; use tokio_stream::StreamExt; use tokio_stream::wrappers::{BroadcastStream, UnboundedReceiverStream}; use tower::{ServiceExt, service_fn}; @@ -331,7 +331,7 @@ impl RunAnswerTransport { match self { Self::Subprocess { control_tx } => { let message = WorkerControlEnvelope::interview_answer(qid.to_string(), answer); - tokio::time::timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, control_tx.send(message)) + timeout(WORKER_CONTROL_ENQUEUE_TIMEOUT, control_tx.send(message)) .await .map_err(|_| AnswerTransportError::Timeout)? .map_err(|_| AnswerTransportError::Closed) @@ -489,9 +489,8 @@ impl SlackService { } async fn submit_answer(&self, state: Arc, submission: SlackAnswerSubmission) { - let run_id = match RunId::from_str(&submission.run_id) { - Ok(run_id) => run_id, - Err(_) => return, + let Ok(run_id) = RunId::from_str(&submission.run_id) else { + return; }; let pending = match load_pending_interview(state.as_ref(), run_id, &submission.qid).await { @@ -2991,15 +2990,14 @@ fn validate_answer_for_question( match (&question.question_type, &answer.value) { ( InterviewQuestionType::YesNo | InterviewQuestionType::Confirmation, - fabro_interview::AnswerValue::Yes, + fabro_interview::AnswerValue::Yes | fabro_interview::AnswerValue::No, ) | ( - InterviewQuestionType::YesNo | InterviewQuestionType::Confirmation, - fabro_interview::AnswerValue::No, - ) - | (_, fabro_interview::AnswerValue::Aborted) - | (_, fabro_interview::AnswerValue::Skipped) - | (_, fabro_interview::AnswerValue::Timeout) => Ok(()), + _, + fabro_interview::AnswerValue::Aborted + | fabro_interview::AnswerValue::Skipped + | fabro_interview::AnswerValue::Timeout, + ) => Ok(()), (InterviewQuestionType::MultipleChoice, fabro_interview::AnswerValue::Selected(key)) => { if question.options.iter().any(|option| option.key == *key) { Ok(()) @@ -3067,16 +3065,15 @@ async fn deliver_answer_to_run( } }; - match transport.submit(qid, answer).await { - Ok(()) => Ok(()), - Err(_) => { - release_run_answer_claim(state, run_id, qid); - Err(ApiError::new( - StatusCode::SERVICE_UNAVAILABLE, - "Failed to deliver answer to the active run.", - ) - .into_response()) - } + if let Ok(()) = transport.submit(qid, answer).await { + Ok(()) + } else { + release_run_answer_claim(state, run_id, qid); + Err(ApiError::new( + StatusCode::SERVICE_UNAVAILABLE, + "Failed to deliver answer to the active run.", + ) + .into_response()) } } @@ -3826,6 +3823,20 @@ async fn execute_run_subprocess(state: Arc, run_id: RunId) { } } + let superseded = { + let runs = state.runs.lock().expect("runs lock poisoned"); + runs.get(&run_id) + .is_some_and(|managed_run| managed_run.worker_pid != Some(worker_pid)) + }; + if superseded { + tracing::info!( + run_id = %run_id, + worker_pid, + "Skipping stale worker cleanup for superseded run execution" + ); + return; + } + append_worker_exit_failure(&run_store, run_id, &wait_status).await; let final_state = match run_store.state().await { diff --git a/lib/crates/fabro-slack/src/interaction.rs b/lib/crates/fabro-slack/src/interaction.rs index 6d8acdc1c..8aa82fb84 100644 --- a/lib/crates/fabro-slack/src/interaction.rs +++ b/lib/crates/fabro-slack/src/interaction.rs @@ -40,7 +40,6 @@ pub fn parse_interaction(payload: &Value) -> Option { // Ignore checkbox toggle events — wait for Submit button return None; } - "plain_text_input" => return None, _ => return None, }; diff --git a/lib/crates/fabro-test/src/lib.rs b/lib/crates/fabro-test/src/lib.rs index c9ee10d3c..4daa990f8 100644 --- a/lib/crates/fabro-test/src/lib.rs +++ b/lib/crates/fabro-test/src/lib.rs @@ -46,6 +46,7 @@ static INSTA_FILTERS: &[(&str, &str)] = &[ const MANAGED_STORAGE_MARKER: &str = "# fabro-test managed storage_dir"; const TEST_IN_MEMORY_STORE_ENV: &str = "FABRO_TEST_IN_MEMORY_STORE"; +const SESSION_LOCK_TIMEOUT: Duration = Duration::from_secs(20); #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum TestMode { @@ -267,7 +268,7 @@ fn with_session_lock(root: &Path, f: impl FnOnce() -> T) -> T { ensure_parent_dir(&lock_path); let lock_file = File::create(&lock_path) .unwrap_or_else(|err| panic!("failed to create {}: {err}", lock_path.display())); - let deadline = std::time::Instant::now() + Duration::from_secs(5); + let deadline = std::time::Instant::now() + SESSION_LOCK_TIMEOUT; while !fabro_proc::try_flock_exclusive(&lock_file) .unwrap_or_else(|err| panic!("failed to lock {}: {err}", lock_path.display())) { diff --git a/lib/crates/fabro-workflow/src/handler/human.rs b/lib/crates/fabro-workflow/src/handler/human.rs index 68845b159..49832a9b3 100644 --- a/lib/crates/fabro-workflow/src/handler/human.rs +++ b/lib/crates/fabro-workflow/src/handler/human.rs @@ -13,6 +13,7 @@ use crate::millis_u64; use crate::outcome::{Outcome, OutcomeExt}; use fabro_graphviz::graph::{Graph, Node}; use fabro_interview::{Answer, AnswerValue, Interviewer, Question, QuestionOption, QuestionType}; +use fabro_types::run_event::InterviewOption; use ulid::Ulid; use super::{EngineServices, Handler}; @@ -218,7 +219,7 @@ impl Handler for HumanHandler { options: question .options .iter() - .map(|option| fabro_types::run_event::InterviewOption { + .map(|option| InterviewOption { key: option.key.clone(), label: option.label.clone(), })