fix: skip already-failed inputs in analyze recovery scan

`recovery::enqueue_pending_for_user` previously enqueued any case with
`analysis_input.json` and no `document.md`, ignoring the failure marker
that `auto_trigger::evaluate_case` already honours. A timed-out LLM
call (e.g. Llama 3.1 405B at ~20 tok/s with default 4096 max_tokens)
left the input in place, the next case-list reload re-enqueued it, and
the worker hung another full timeout window — starving fresh user
clicks behind a failed retry loop.

Recovery now mirrors the same skip: if a marker exists whose
`last_recording_mtime` matches the input's, treat it as "already
tried, do not retry until input changes". Fresh recordings bump the
mtime and recovery picks the case back up.

server/src/analyze/recovery.rs

@@ -2,7 +2,8 @@ use std::path::Path;
 
 use tracing::{info, warn};
 
-use super::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
+use super::auto_trigger::read_failure_marker;
+use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
 
 /// Walk `data_path/*/` and enqueue every pending analysis for every user.
 /// Intended to run once at startup; the same primitive backs the per-user
@@ -36,6 +37,15 @@ pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> u
         if case_dir.join(DOCUMENT_FILE).exists() {
             continue;
         }
+        // A failure marker matching the input's `last_recording_mtime`
+        // means the worker has already tried this exact input and failed
+        // (e.g. LLM timeout). Re-enqueueing here would block the worker
+        // for another full timeout window and starve fresh user clicks.
+        // `auto_trigger::evaluate_case` performs the same skip; recovery
+        // must mirror it.
+        if input_already_failed(&case_dir).await {
+            continue;
+        }
         if tx
             .send(AnalyzeJob {
                 case_dir: case_dir.clone(),
@@ -49,3 +59,157 @@ pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> u
     }
     sent
 }
+
+/// Returns `true` iff a failure marker exists whose
+/// `last_recording_mtime` equals the one in `analysis_input.json`. Any
+/// I/O or parse error is treated as "no usable marker" — recovery then
+/// proceeds with the enqueue, matching the optimistic intent of the
+/// outer scan.
+async fn input_already_failed(case_dir: &Path) -> bool {
+    let Some(marker) = read_failure_marker(case_dir).await else {
+        return false;
+    };
+    let Ok(bytes) = tokio::fs::read(case_dir.join(ANALYSIS_INPUT_FILE)).await else {
+        return false;
+    };
+    let Ok(input) = serde_json::from_slice::<AnalysisInput>(&bytes) else {
+        return false;
+    };
+    marker.last_recording_mtime == input.last_recording_mtime
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::analyze::auto_trigger::{FAILURE_MARKER_FILE, FailureMarker};
+    use crate::analyze::{RecordingInput, channel as analyze_channel};
+    use tempfile::TempDir;
+    use tokio::fs;
+
+    fn make_input(mtime: &str) -> AnalysisInput {
+        AnalysisInput {
+            last_recording_mtime: mtime.to_owned(),
+            recordings: vec![RecordingInput {
+                recorded_at: mtime.to_owned(),
+                text: "ignored".into(),
+            }],
+            backend_id: String::new(),
+        }
+    }
+
+    async fn write_input(case_dir: &Path, input: &AnalysisInput) {
+        fs::write(
+            case_dir.join(ANALYSIS_INPUT_FILE),
+            serde_json::to_vec(input).unwrap(),
+        )
+        .await
+        .unwrap();
+    }
+
+    async fn write_marker(case_dir: &Path, mtime: &str) {
+        let marker = FailureMarker {
+            last_recording_mtime: mtime.to_owned(),
+            reason: "test".into(),
+            failed_at: "2026-05-03T00:00:00Z".into(),
+            backend_id: None,
+        };
+        fs::write(
+            case_dir.join(FAILURE_MARKER_FILE),
+            serde_json::to_vec(&marker).unwrap(),
+        )
+        .await
+        .unwrap();
+    }
+
+    /// Drain the receiver after sender drop — same pattern as
+    /// `auto_trigger::tests::try_enqueue_all_sends_only_eligible_cases`.
+    async fn drain(mut rx: crate::analyze::AnalyzeReceiver) -> Vec<std::path::PathBuf> {
+        let mut out = Vec::new();
+        while let Some(job) = rx.recv().await {
+            out.push(job.case_dir);
+        }
+        out
+    }
+
+    #[tokio::test]
+    async fn enqueues_pending_input_without_doc_or_marker() {
+        let user_root = TempDir::new().unwrap();
+        let case_id = "11111111-1111-1111-1111-111111111111";
+        let case = user_root.path().join(case_id);
+        fs::create_dir_all(&case).await.unwrap();
+        write_input(&case, &make_input("2026-05-03T10:00:00Z")).await;
+
+        let (tx, rx) = analyze_channel();
+        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
+        drop(tx);
+
+        assert_eq!(sent, 1);
+        let received = drain(rx).await;
+        assert_eq!(received.len(), 1);
+        assert!(received[0].ends_with(case_id));
+    }
+
+    #[tokio::test]
+    async fn skips_when_document_already_exists() {
+        let user_root = TempDir::new().unwrap();
+        let case = user_root
+            .path()
+            .join("22222222-2222-2222-2222-222222222222");
+        fs::create_dir_all(&case).await.unwrap();
+        write_input(&case, &make_input("2026-05-03T10:00:00Z")).await;
+        fs::write(case.join(DOCUMENT_FILE), b"already analysed")
+            .await
+            .unwrap();
+
+        let (tx, rx) = analyze_channel();
+        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
+        drop(tx);
+
+        assert_eq!(sent, 0);
+        assert!(drain(rx).await.is_empty());
+    }
+
+    /// Regression: a stuck Llama job times out, leaves `analysis_input.json`
+    /// behind, and writes a failure marker. Without this skip the next
+    /// page reload re-enqueues the same job and blocks the worker for
+    /// another full timeout — see Bug 1 (case c414cf52, 2026-05-03).
+    #[tokio::test]
+    async fn skips_when_failure_marker_matches_input_mtime() {
+        let user_root = TempDir::new().unwrap();
+        let case = user_root
+            .path()
+            .join("33333333-3333-3333-3333-333333333333");
+        fs::create_dir_all(&case).await.unwrap();
+        let mtime = "2026-05-03T10:00:00Z";
+        write_input(&case, &make_input(mtime)).await;
+        write_marker(&case, mtime).await;
+
+        let (tx, rx) = analyze_channel();
+        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
+        drop(tx);
+
+        assert_eq!(sent, 0, "must not re-enqueue an already-failed input");
+        assert!(drain(rx).await.is_empty());
+    }
+
+    /// A new recording bumps `last_recording_mtime`, so the old marker
+    /// no longer applies and recovery should re-enqueue.
+    #[tokio::test]
+    async fn enqueues_when_failure_marker_is_stale() {
+        let user_root = TempDir::new().unwrap();
+        let case_id = "44444444-4444-4444-4444-444444444444";
+        let case = user_root.path().join(case_id);
+        fs::create_dir_all(&case).await.unwrap();
+        write_input(&case, &make_input("2026-05-03T11:00:00Z")).await;
+        write_marker(&case, "2026-05-03T10:00:00Z").await;
+
+        let (tx, rx) = analyze_channel();
+        let sent = enqueue_pending_for_user(user_root.path(), &tx).await;
+        drop(tx);
+
+        assert_eq!(sent, 1);
+        let received = drain(rx).await;
+        assert_eq!(received.len(), 1);
+        assert!(received[0].ends_with(case_id));
+    }
+}