Merge remote-tracking branch 'origin/main'

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

@@ -245,7 +245,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 +255,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
 }
 

lib/crates/fabro-cli/src/commands/run/runner.rs

@@ -116,14 +116,11 @@ where
 {
     let mut lines = BufReader::new(reader).lines();
     loop {
-        match lines.next_line().await {
-            Ok(Some(line)) => {
-                apply_worker_control_line(&broker, &line).await;
-            }
-            Ok(None) | Err(_) => {
-                broker.abort_all().await;
-                break;
-            }
+        if let Ok(Some(line)) = lines.next_line().await {
+            apply_worker_control_line(&broker, &line).await;
+        } else {
+            broker.abort_all().await;
+            break;
         }
     }
 }

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

@@ -874,6 +874,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

@@ -58,7 +58,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)
@@ -491,16 +491,15 @@ 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 question =
-            match load_pending_interview_question(state.as_ref(), run_id, &submission.qid).await {
-                Ok(question) => question,
-                Err(_) => return,
-            };
+        let Ok(question) =
+            load_pending_interview_question(state.as_ref(), run_id, &submission.qid).await
+        else {
+            return;
+        };
         if validate_answer_for_question(&question, &submission.answer).is_err() {
             return;
         }
@@ -2910,7 +2909,6 @@ fn parse_question_type(question_type: &str) -> QuestionType {
         "yes_no" => QuestionType::YesNo,
         "multiple_choice" => QuestionType::MultipleChoice,
         "multi_select" => QuestionType::MultiSelect,
-        "freeform" => QuestionType::Freeform,
         "confirmation" => QuestionType::Confirmation,
         _ => QuestionType::Freeform,
     }
@@ -2984,11 +2982,16 @@ async fn load_pending_interview_question(
 #[allow(clippy::result_large_err)] // Axum handlers naturally propagate full `Response` errors.
 fn validate_answer_for_question(question: &Question, answer: &Answer) -> Result<(), Response> {
     match (&question.question_type, &answer.value) {
-        (QuestionType::YesNo | QuestionType::Confirmation, fabro_interview::AnswerValue::Yes)
-        | (QuestionType::YesNo | QuestionType::Confirmation, fabro_interview::AnswerValue::No)
-        | (_, fabro_interview::AnswerValue::Aborted)
-        | (_, fabro_interview::AnswerValue::Skipped)
-        | (_, fabro_interview::AnswerValue::Timeout) => Ok(()),
+        (
+            QuestionType::YesNo | QuestionType::Confirmation,
+            fabro_interview::AnswerValue::Yes | fabro_interview::AnswerValue::No,
+        )
+        | (
+            _,
+            fabro_interview::AnswerValue::Aborted
+            | fabro_interview::AnswerValue::Skipped
+            | fabro_interview::AnswerValue::Timeout,
+        ) => Ok(()),
         (QuestionType::MultipleChoice, fabro_interview::AnswerValue::Selected(key)) => {
             if question.options.iter().any(|option| option.key == *key) {
                 Ok(())
@@ -3046,16 +3049,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())
     }
 }
 
@@ -3796,6 +3798,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(),
                     })