Merge remote-tracking branch 'origin/main'

# Conflicts:
#	lib/crates/fabro-cli/src/commands/run/runner.rs
#	lib/crates/fabro-server/src/server.rs

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

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<ExitCode> {
     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<EventEnvelope>,
     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));
+    }
 }

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]"
       }
     ]
     "#);

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]",

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<AppState>, 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<AppState>, 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 {

lib/crates/fabro-slack/src/interaction.rs

@@ -40,7 +40,6 @@ pub fn parse_interaction(payload: &Value) -> Option<SlackAnswerSubmission> {
             // Ignore checkbox toggle events — wait for Submit button
             return None;
         }
-        "plain_text_input" => return None,
         _ => return None,
     };
 

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<T>(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()))
     {

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(),
                     })