Refactor: Track analysis jobs via InFlight struct

This commit replaces the disk-based `analysis_input.json` marker for
in-progress analysis jobs with an in-memory
`Arc<RwLock<HashSet<PathBuf>>>`.

Previously, the existence of `analysis_input.json` was used as a signal
that a case was queued or being analyzed. This led to stale "Queued"
states after server restarts, as the disk file would persist while the
server process tracking it had vanished.

The new `InFlight` struct, held in `AppState`, provides a process-local
marker. This ensures that:

- Server restarts correctly reset the state, as the `HashSet` is
  cleared.
- Race conditions for triggering analysis are handled by the `try_claim`
  method, replacing the previous reliance on file system `O_EXCL` flags.
- The `AnalyzeJob` payload now travels with the job over the channel,
  eliminating the need for workers to re-read `analysis_input.json` from
  disk.

This change simplifies state management and improves the reliability of
analysis job tracking.

server/src/analyze/auto_trigger.rs

@@ -17,7 +17,7 @@ use time::format_description::well_known::Rfc3339;
 use tokio::io::AsyncWriteExt;
 use tracing::warn;
 
-use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
+use super::{AnalyzeJob, AnalyzeSender, DOCUMENT_FILE, InFlight};
 use crate::events::{self, CaseEventKind, EventSender};
 use crate::paths;
 use crate::routes::case_actions::build_analysis_input;
@@ -155,7 +155,7 @@ pub enum CaseAnalysisState {
     Current,
 }
 
-/// Compute the full per-case lifecycle state. Pure file-I/O, no
+/// Compute the full per-case lifecycle state. Pure observation, no
 /// mutations or channel sends — safe to call redundantly per render.
 ///
 /// `idle_threshold` is the time without new recordings after which a
@@ -172,11 +172,16 @@ pub enum CaseAnalysisState {
 /// fresh server start does not instantly classify every old case as
 /// `Idle` (and trigger a thundering herd of LLM calls on the first
 /// `/cases` render). Pass `SystemTime::UNIX_EPOCH` to disable.
+///
+/// `in_flight` is the process-local "queued or running" set. It replaces
+/// the previous on-disk `analysis_input.json` marker, which survived
+/// crashes and left cases stuck in `Queued` after a restart.
 pub async fn evaluate_state(
     case_dir: &Path,
     idle_threshold: std::time::Duration,
     now: SystemTime,
     boot_time: SystemTime,
+    in_flight: &InFlight,
 ) -> CaseAnalysisState {
     let scan = scan_m4as(case_dir).await;
 
@@ -190,11 +195,9 @@ pub async fn evaluate_state(
         return CaseAnalysisState::NoRecordings;
     };
 
-    // Any existing analysis_input.json means a job is queued or running.
+    // A claimed slot means a job is queued or in flight in this very
-    if tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE))
+    // process. Empty on every server boot — no more phantom Queued.
-        .await
+    if in_flight.contains(case_dir) {
-        .unwrap_or(false)
-    {
         return CaseAnalysisState::Queued;
     }
 
@@ -244,20 +247,24 @@ pub(crate) async fn read_document_meta(case_dir: &Path) -> Option<crate::analyze
     serde_json::from_slice(&bytes).ok()
 }
 
-/// Backwards-compatible binary decision: kept for the existing
+/// Backwards-compatible binary decision: kept for the existing unit
-/// `try_enqueue` callsites and unit tests that predate the typed state
+/// tests that predate the typed state machine. New callers should
-/// machine. New callers should consume [`evaluate_state`] directly.
+/// consume [`evaluate_state`] directly.
 ///
 /// Reduction: `Idle` → `Enqueue`, everything else → `Skip(<tag>)`.
 /// `Required` is also mapped to `Skip` here — but with
 /// `Duration::ZERO` as the threshold, `Required` collapses into `Idle`
 /// in practice, so the eager pre-refactor semantics are preserved.
+///
+/// Uses an empty in-flight set: tests that exercise the `Queued` branch
+/// supply their own set via `evaluate_state` directly.
 pub async fn evaluate_case(case_dir: &Path) -> AutoDecision {
     let state = evaluate_state(
         case_dir,
         std::time::Duration::ZERO,
         SystemTime::now(),
         SystemTime::UNIX_EPOCH,
+        &InFlight::new(),
     )
     .await;
     match state {
@@ -290,18 +297,27 @@ pub async fn try_enqueue(
     idle_threshold: std::time::Duration,
     now: SystemTime,
     boot_time: SystemTime,
+    in_flight: &InFlight,
     tx: &AnalyzeSender,
     events_tx: &EventSender,
 ) -> bool {
-    match evaluate_state(case_dir, idle_threshold, now, boot_time).await {
+    match evaluate_state(case_dir, idle_threshold, now, boot_time, in_flight).await {
         CaseAnalysisState::Idle { .. } => {}
         _ => return false,
     }
 
+    // Atomic claim → race-guard against concurrent triggers. If another
+    // caller (manual button, second tab) just grabbed the slot, skip
+    // silently — they will send the job.
+    if !in_flight.try_claim(case_dir) {
+        return false;
+    }
+
     let input = match build_analysis_input(case_dir, "").await {
         Ok(i) => i,
         Err(e) => {
             warn!(case = %case_dir.display(), error = ?e, "auto-analyze: build input failed");
+            in_flight.release(case_dir);
             return false;
         }
     };
@@ -310,19 +326,15 @@ pub async fn try_enqueue(
     // the UI keeps rendering the last successful analysis as "letztes
     // Dokument" with a Stale-banner. The worker overwrites the file
     // atomically once the new analysis completes.
-    let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
-    if let Err(e) = write_input_overwrite(&input_path, &input).await {
-        warn!(case = %case_dir.display(), error = %e, "auto-analyze: write input failed");
-        return false;
-    }
-
     if tx
         .send(AnalyzeJob {
             case_dir: case_dir.to_path_buf(),
+            input,
         })
         .is_err()
     {
         warn!(case = %case_dir.display(), "auto-analyze: send failed (worker gone)");
+        in_flight.release(case_dir);
         return false;
     }
 
@@ -343,6 +355,7 @@ pub async fn try_enqueue_all_for_user(
     idle_threshold: std::time::Duration,
     now: SystemTime,
     boot_time: SystemTime,
+    in_flight: &InFlight,
     tx: &AnalyzeSender,
     events_tx: &EventSender,
 ) {
@@ -363,7 +376,16 @@ pub async fn try_enqueue_all_for_user(
         if crate::paths::is_closed(&case_path).await {
             continue;
         }
-        try_enqueue(&case_path, idle_threshold, now, boot_time, tx, events_tx).await;
+        try_enqueue(
+            &case_path,
+            idle_threshold,
+            now,
+            boot_time,
+            in_flight,
+            tx,
+            events_tx,
+        )
+        .await;
     }
 }
 
@@ -511,24 +533,6 @@ pub(crate) async fn read_failure_marker(case_dir: &Path) -> Option<FailureMarker
     serde_json::from_slice(&bytes).ok()
 }
 
-/// Overwrite-capable sibling of `case_actions::write_input_create_new`.
-/// Auto-trigger re-analyses may legitimately replace an existing input
-/// (e.g. a new recording arrived between two reloads while the previous
-/// job was still queued). `.tmp` + fsync + rename keeps it atomic.
-pub(crate) async fn write_input_overwrite(
-    path: &Path,
-    input: &AnalysisInput,
-) -> std::io::Result<()> {
-    let bytes = serde_json::to_vec_pretty(input)
-        .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
-    let tmp = path.with_extension("json.tmp");
-    let mut f = tokio::fs::File::create(&tmp).await?;
-    f.write_all(&bytes).await?;
-    f.sync_all().await?;
-    drop(f);
-    tokio::fs::rename(&tmp, path).await
-}
-
 /// Format an mtime in the same RFC3339 shape that
 /// `case_actions::build_analysis_input` writes, so equality comparison
 /// against a stored marker is string-exact.
@@ -697,16 +701,45 @@ mod tests {
         );
     }
 
+    /// `Queued` is now derived from the process-local `InFlight` set,
+    /// not from a file on disk. This test exercises that path directly
+    /// via `evaluate_state` (the `evaluate_case` adapter passes an empty
+    /// set, which is exactly the point — orphan files cannot fake
+    /// `Queued` any more).
     #[tokio::test]
-    async fn analysis_input_present_skips() {
+    async fn in_flight_claim_yields_queued() {
         let dir = TempDir::new().unwrap();
         touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
         seed_meta(dir.path(), "2026-01-01T10-00-00Z").await;
-        touch(&dir.path().join(ANALYSIS_INPUT_FILE)).await;
+
-        assert_eq!(
+        let in_flight = InFlight::new();
-            evaluate_case(dir.path()).await,
+        assert!(in_flight.try_claim(dir.path()));
-            AutoDecision::Skip("job queued")
+
-        );
+        let state = evaluate_state(
+            dir.path(),
+            std::time::Duration::ZERO,
+            SystemTime::now(),
+            SystemTime::UNIX_EPOCH,
+            &in_flight,
+        )
+        .await;
+        assert_eq!(state, CaseAnalysisState::Queued);
+    }
+
+    /// Orphan `analysis_input.json` left over from a pre-refactor server
+    /// run must NOT register as `Queued` any more — that was the whole
+    /// reason for the refactor.
+    #[tokio::test]
+    async fn orphan_input_file_does_not_count_as_queued() {
+        let dir = TempDir::new().unwrap();
+        touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
+        seed_meta_with(dir.path(), "2026-01-01T10-00-00Z", "patient mit fieber").await;
+        // Pre-refactor leftover: a stale file the server used to write.
+        fs::write(dir.path().join("analysis_input.json"), b"{}")
+            .await
+            .unwrap();
+
+        assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
     }
 
     #[tokio::test]
@@ -795,6 +828,7 @@ mod tests {
             Duration::from_secs(900),
             base + Duration::from_secs(100),
             SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
         )
         .await;
 
@@ -823,6 +857,7 @@ mod tests {
             Duration::from_secs(900),
             base + Duration::from_secs(1000),
             SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
         )
         .await;
 
@@ -855,7 +890,14 @@ mod tests {
         // Required, not Idle.
         let now = base + Duration::from_secs(30 * 24 * 3600);
         let boot = now - Duration::from_secs(100);
-        let state = evaluate_state(dir.path(), Duration::from_secs(900), now, boot).await;
+        let state = evaluate_state(
+            dir.path(),
+            Duration::from_secs(900),
+            now,
+            boot,
+            &InFlight::new(),
+        )
+        .await;
 
         assert_eq!(
             state,
@@ -867,7 +909,14 @@ mod tests {
         // Once the grace window has elapsed (boot was 1000 s ago) the
         // case promotes to Idle.
         let boot_old = now - Duration::from_secs(1000);
-        let state_old = evaluate_state(dir.path(), Duration::from_secs(900), now, boot_old).await;
+        let state_old = evaluate_state(
+            dir.path(),
+            Duration::from_secs(900),
+            now,
+            boot_old,
+            &InFlight::new(),
+        )
+        .await;
         assert_eq!(
             state_old,
             CaseAnalysisState::Idle {
@@ -891,6 +940,7 @@ mod tests {
 
         let (tx, mut rx) = analyze_channel();
         let events_tx = events_channel();
+        let in_flight = InFlight::new();
 
         // Idle threshold not reached → must NOT enqueue.
         let sent_required = try_enqueue(
@@ -898,6 +948,7 @@ mod tests {
             Duration::from_secs(900),
             base + Duration::from_secs(100),
             SystemTime::UNIX_EPOCH,
+            &in_flight,
             &tx,
             &events_tx,
         )
@@ -907,6 +958,10 @@ mod tests {
             rx.try_recv().is_err(),
             "channel must be empty after Required"
         );
+        assert!(
+            !in_flight.contains(dir.path()),
+            "no claim should exist after Required"
+        );
 
         // Same case, idle threshold passed → must enqueue exactly one job.
         let sent_idle = try_enqueue(
@@ -914,6 +969,7 @@ mod tests {
             Duration::from_secs(900),
             base + Duration::from_secs(1000),
             SystemTime::UNIX_EPOCH,
+            &in_flight,
             &tx,
             &events_tx,
         )
@@ -924,6 +980,55 @@ mod tests {
             "channel must hold one job after Idle"
         );
         assert!(rx.try_recv().is_err(), "channel must hold exactly one job");
+        assert!(
+            in_flight.contains(dir.path()),
+            "claim must persist until the worker drops the slot guard"
+        );
+    }
+
+    /// Concurrent triggers (auto-trigger + manual click in another tab)
+    /// must produce **exactly one** job, not two. The atomic
+    /// `try_claim` is the race-guard.
+    #[tokio::test]
+    async fn try_enqueue_is_idempotent_under_concurrent_triggers() {
+        use crate::analyze::channel as analyze_channel;
+        use crate::events::channel as events_channel;
+
+        let dir = TempDir::new().unwrap();
+        let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
+        touch(&m4a).await;
+        seed_meta_with(dir.path(), "2026-01-01T10-00-00Z", "patient mit fieber").await;
+        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
+        set_mtime(&m4a, base);
+
+        let (tx, mut rx) = analyze_channel();
+        let events_tx = events_channel();
+        let in_flight = InFlight::new();
+
+        let a = try_enqueue(
+            dir.path(),
+            Duration::from_secs(900),
+            base + Duration::from_secs(1000),
+            SystemTime::UNIX_EPOCH,
+            &in_flight,
+            &tx,
+            &events_tx,
+        )
+        .await;
+        let b = try_enqueue(
+            dir.path(),
+            Duration::from_secs(900),
+            base + Duration::from_secs(1000),
+            SystemTime::UNIX_EPOCH,
+            &in_flight,
+            &tx,
+            &events_tx,
+        )
+        .await;
+        assert!(a, "first call must enqueue");
+        assert!(!b, "second call must skip — slot is already claimed");
+        assert!(rx.try_recv().is_ok());
+        assert!(rx.try_recv().is_err(), "exactly one job in the channel");
     }
 
     #[tokio::test]
@@ -968,12 +1073,14 @@ mod tests {
 
         let (tx, mut rx) = analyze_channel();
         let events_tx = events_channel();
+        let in_flight = InFlight::new();
 
         try_enqueue_all_for_user(
             user_root.path(),
             std::time::Duration::ZERO,
             SystemTime::now(),
             SystemTime::UNIX_EPOCH,
+            &in_flight,
             &tx,
             &events_tx,
         )

server/src/analyze/in_flight.rs

@@ -0,0 +1,144 @@
+//! Process-local "this case is queued or in flight" tracking.
+//!
+//! Replaces the previous design where the existence of `analysis_input.json`
+//! on disk doubled as the lifecycle marker. That coupled disk state
+//! (lives forever) to process state (lives one boot session), so a crash
+//! mid-job left the case stuck in `Queued` after the next restart.
+//!
+//! By moving the marker into an [`Arc<RwLock<HashSet<PathBuf>>>`] held in
+//! `AppState`, the marker is reset to empty on every server boot. The
+//! payload that used to live in `analysis_input.json` now travels with
+//! the [`AnalyzeJob`] over the mpsc channel.
+//!
+//! [`AnalyzeJob`]: super::AnalyzeJob
+
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, RwLock};
+
+/// Shared, process-local set of cases currently queued or being analyzed.
+/// Cheap to clone (`Arc`). Operations are non-blocking and short — the
+/// inner `HashSet` is only ever held under the lock for an `insert`,
+/// `remove`, or `contains` call.
+#[derive(Clone, Default)]
+pub struct InFlight(Arc<RwLock<HashSet<PathBuf>>>);
+
+impl InFlight {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Atomically claim a case. Returns `true` if the caller now owns the
+    /// slot (proceed to send a job), `false` if another caller already
+    /// holds it (skip — duplicate trigger). Replaces the `O_EXCL` create
+    /// race-guard the manual-trigger handlers used to rely on.
+    pub fn try_claim(&self, case_dir: &Path) -> bool {
+        self.0
+            .write()
+            .expect("inflight lock poisoned")
+            .insert(case_dir.to_path_buf())
+    }
+
+    /// Release a slot. Idempotent — safe to call from cleanup paths that
+    /// do not know whether the slot was actually held.
+    pub fn release(&self, case_dir: &Path) {
+        self.0
+            .write()
+            .expect("inflight lock poisoned")
+            .remove(case_dir);
+    }
+
+    /// Read-only check used by `evaluate_state` to render the
+    /// `Queued` UI state.
+    pub fn contains(&self, case_dir: &Path) -> bool {
+        self.0
+            .read()
+            .expect("inflight lock poisoned")
+            .contains(case_dir)
+    }
+}
+
+/// RAII slot-release guard. Construct after a successful [`InFlight::try_claim`];
+/// the slot is released when the guard drops. Covers success, every
+/// failure path, and worker panics — analogous to [`crate::BusyGuard`].
+pub struct InFlightGuard {
+    set: InFlight,
+    case_dir: PathBuf,
+}
+
+impl InFlightGuard {
+    pub fn new(set: InFlight, case_dir: PathBuf) -> Self {
+        Self { set, case_dir }
+    }
+}
+
+impl Drop for InFlightGuard {
+    fn drop(&mut self) {
+        self.set.release(&self.case_dir);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn p(s: &str) -> PathBuf {
+        PathBuf::from(s)
+    }
+
+    #[test]
+    fn try_claim_returns_true_first_then_false() {
+        let set = InFlight::new();
+        assert!(set.try_claim(&p("/case/a")));
+        assert!(!set.try_claim(&p("/case/a")));
+        assert!(set.try_claim(&p("/case/b")));
+    }
+
+    #[test]
+    fn release_makes_slot_reclaimable() {
+        let set = InFlight::new();
+        assert!(set.try_claim(&p("/case/a")));
+        set.release(&p("/case/a"));
+        assert!(set.try_claim(&p("/case/a")));
+    }
+
+    #[test]
+    fn release_unknown_slot_is_noop() {
+        let set = InFlight::new();
+        set.release(&p("/case/never-claimed"));
+    }
+
+    #[test]
+    fn contains_reflects_claim_state() {
+        let set = InFlight::new();
+        assert!(!set.contains(&p("/case/a")));
+        set.try_claim(&p("/case/a"));
+        assert!(set.contains(&p("/case/a")));
+        set.release(&p("/case/a"));
+        assert!(!set.contains(&p("/case/a")));
+    }
+
+    #[test]
+    fn guard_releases_on_drop() {
+        let set = InFlight::new();
+        assert!(set.try_claim(&p("/case/a")));
+        {
+            let _g = InFlightGuard::new(set.clone(), p("/case/a"));
+            assert!(set.contains(&p("/case/a")));
+        }
+        assert!(!set.contains(&p("/case/a")));
+    }
+
+    #[test]
+    fn guard_releases_on_panic_unwind() {
+        let set = InFlight::new();
+        set.try_claim(&p("/case/a"));
+        let set_clone = set.clone();
+        let result = std::panic::catch_unwind(move || {
+            let _g = InFlightGuard::new(set_clone, p("/case/a"));
+            panic!("simulated worker panic");
+        });
+        assert!(result.is_err());
+        assert!(!set.contains(&p("/case/a")));
+    }
+}

server/src/analyze/mod.rs

@@ -5,18 +5,20 @@ use tokio::sync::mpsc;
 
 pub mod auto_trigger;
 pub mod backend;
+pub mod in_flight;
 pub mod llm;
 pub mod prompt;
 pub mod recovery;
 pub mod render;
 pub mod worker;
 
-/// Single-document-per-case filenames. A re-analysis overwrites the
+pub use in_flight::{InFlight, InFlightGuard};
+
+/// Single-document-per-case filename. A re-analysis overwrites the
 /// document in place; the file is a typed JSON envelope so the
 /// `covered_mtime` snapshot stays glued to the markdown body it was
 /// produced from.
 pub const DOCUMENT_FILE: &str = "document.json";
-pub const ANALYSIS_INPUT_FILE: &str = "analysis_input.json";
 
 /// Persisted analysis output for a case. The `covered_mtime` field is
 /// the snapshot mtime of the recordings the LLM saw — *not* the
@@ -33,13 +35,15 @@ pub struct Document {
     pub content_md: String,
 }
 
-/// A single request to analyze a case. Payload is tiny (a path), so the
+/// A single request to analyze a case. Carries the full [`AnalysisInput`]
-/// channel is unbounded — persistence lives in `analysis_input.json` on
+/// payload over the channel so the worker does not need to read it back
-/// disk, not in the queue. Recovery re-enqueues any inputs the server may
+/// from disk. Lifecycle is process-local — see [`InFlight`] — so a server
-/// have held at crash time.
+/// restart between send and worker pickup just drops the job; the next
+/// `/cases` render re-triggers via the normal `Required → Idle` path.
 #[derive(Debug, Clone)]
 pub struct AnalyzeJob {
     pub case_dir: PathBuf,
+    pub input: AnalysisInput,
 }
 
 pub type AnalyzeSender = mpsc::UnboundedSender<AnalyzeJob>;
@@ -49,10 +53,9 @@ pub fn channel() -> (AnalyzeSender, AnalyzeReceiver) {
     mpsc::unbounded_channel()
 }
 
-/// Persistent input for a single analysis run. Written by the analyze
+/// Input payload for a single analysis run. Built by the trigger
-/// handler, consumed by the analyze worker, removed after a successful
+/// (handler or auto-trigger), shipped to the worker via [`AnalyzeJob`].
-/// document write.
+#[derive(Debug, Clone, Serialize, Deserialize)]
-#[derive(Debug, Serialize, Deserialize)]
 pub struct AnalysisInput {
     /// Latest filesystem mtime (server clock) of any `.m4a` in the case at
     /// the moment the analyze button was clicked. Carried through for
@@ -65,7 +68,7 @@ pub struct AnalysisInput {
     pub backend_id: String,
 }
 
-#[derive(Debug, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct RecordingInput {
     /// Recording timestamp from the watch (filename-derived, ISO-8601).
     pub recorded_at: String,

server/src/analyze/recovery.rs

@@ -13,6 +13,7 @@ use std::time::{Duration, SystemTime};
 
 use tracing::info;
 
+use super::InFlight;
 use super::auto_trigger::{CaseAnalysisState, evaluate_state};
 
 /// Aggregate counts over every UUID-named, non-closed case under a user
@@ -64,14 +65,21 @@ pub async fn boot_scan_state(
     idle_threshold: Duration,
     now: SystemTime,
     boot_time: SystemTime,
+    in_flight: &InFlight,
 ) -> BootStateSummary {
     let mut total = BootStateSummary::default();
     let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
         return total;
     };
     while let Ok(Some(user_entry)) = users.next_entry().await {
-        let per_user =
+        let per_user = compute_state_for_user(
-            compute_state_for_user(&user_entry.path(), idle_threshold, now, boot_time).await;
+            &user_entry.path(),
+            idle_threshold,
+            now,
+            boot_time,
+            in_flight,
+        )
+        .await;
         total.add(per_user);
     }
     info!(
@@ -92,6 +100,7 @@ pub async fn compute_state_for_user(
     idle_threshold: Duration,
     now: SystemTime,
     boot_time: SystemTime,
+    in_flight: &InFlight,
 ) -> BootStateSummary {
     let mut summary = BootStateSummary::default();
     let Ok(mut cases) = tokio::fs::read_dir(user_root).await else {
@@ -108,7 +117,7 @@ pub async fn compute_state_for_user(
         if uuid::Uuid::parse_str(&name).is_err() {
             continue;
         }
-        let state = evaluate_state(&case_dir, idle_threshold, now, boot_time).await;
+        let state = evaluate_state(&case_dir, idle_threshold, now, boot_time, in_flight).await;
         summary.record(&state);
     }
     summary
@@ -120,7 +129,7 @@ mod tests {
     use crate::analyze::auto_trigger::{
         FAILURE_MARKER_FILE, FailureMarker, rfc3339_of as auto_rfc3339,
     };
-    use crate::analyze::{ANALYSIS_INPUT_FILE, AnalysisInput, DOCUMENT_FILE, Document};
+    use crate::analyze::{DOCUMENT_FILE, Document};
     use filetime::{FileTime, set_file_mtime};
     use tempfile::TempDir;
     use tokio::fs;
@@ -174,20 +183,6 @@ mod tests {
         .unwrap();
     }
 
-    async fn write_input(case_dir: &Path, mtime: SystemTime) {
-        let input = AnalysisInput {
-            last_recording_mtime: auto_rfc3339(mtime),
-            recordings: Vec::new(),
-            backend_id: String::new(),
-        };
-        fs::write(
-            case_dir.join(ANALYSIS_INPUT_FILE),
-            serde_json::to_vec(&input).unwrap(),
-        )
-        .await
-        .unwrap();
-    }
-
     async fn write_marker(case_dir: &Path, mtime: SystemTime) {
         let marker = FailureMarker {
             last_recording_mtime: auto_rfc3339(mtime),
@@ -208,17 +203,23 @@ mod tests {
         SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000)
     }
 
+    /// `Queued` is now driven by the in-process [`InFlight`] set, not
+    /// by a file on disk. A claimed slot for a transcribed case must
+    /// surface as `queued`.
     #[tokio::test]
-    async fn pending_input_counts_as_queued() {
+    async fn in_flight_claim_counts_as_queued() {
         let user_root = TempDir::new().unwrap();
         let case = seed_transcribed_case(user_root.path(), 1, base_time()).await;
-        write_input(&case, base_time()).await;
+
+        let in_flight = InFlight::new();
+        assert!(in_flight.try_claim(&case));
 
         let summary = compute_state_for_user(
             user_root.path(),
             Duration::ZERO,
             base_time(),
             SystemTime::UNIX_EPOCH,
+            &in_flight,
         )
         .await;
 
@@ -227,6 +228,32 @@ mod tests {
         assert_eq!(summary.required, 0);
     }
 
+    /// Pre-refactor leftover: a stale `analysis_input.json` on disk (no
+    /// claim in [`InFlight`]) must NOT register as `Queued` any more —
+    /// that was the whole reason for the refactor. The case still
+    /// produces a transcribed-but-no-document state, so it counts as
+    /// `idle` (zero idle threshold collapses Required→Idle).
+    #[tokio::test]
+    async fn orphan_input_file_does_not_count_as_queued() {
+        let user_root = TempDir::new().unwrap();
+        let case = seed_transcribed_case(user_root.path(), 7, base_time()).await;
+        fs::write(case.join("analysis_input.json"), b"{}")
+            .await
+            .unwrap();
+
+        let summary = compute_state_for_user(
+            user_root.path(),
+            Duration::ZERO,
+            base_time(),
+            SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
+        )
+        .await;
+
+        assert_eq!(summary.queued, 0);
+        assert_eq!(summary.idle, 1);
+    }
+
     #[tokio::test]
     async fn document_covering_latest_counts_as_current() {
         let user_root = TempDir::new().unwrap();
@@ -238,6 +265,7 @@ mod tests {
             Duration::ZERO,
             base_time(),
             SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
         )
         .await;
 
@@ -258,6 +286,7 @@ mod tests {
             Duration::ZERO,
             base_time(),
             SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
         )
         .await;
 
@@ -277,6 +306,7 @@ mod tests {
             Duration::ZERO,
             base_time(),
             SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
         )
         .await;
 
@@ -298,6 +328,7 @@ mod tests {
             Duration::ZERO,
             base_time(),
             SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
         )
         .await;
 
@@ -316,6 +347,7 @@ mod tests {
             Duration::from_secs(900),
             base_time() + Duration::from_secs(100),
             SystemTime::UNIX_EPOCH,
+            &InFlight::new(),
         )
         .await;
 

server/src/analyze/worker.rs

@@ -7,7 +7,8 @@ use tracing::{error, info, warn};
 use super::auto_trigger::FailureDetails;
 use super::backend::{default_backend, find_backend};
 use super::{
-    ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, auto_trigger, llm, prompt,
+    AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, InFlight, InFlightGuard, auto_trigger, llm,
+    prompt,
 };
 use crate::events::{self, CaseEventKind, EventSender};
 use crate::gazetteer::Gazetteer;
@@ -20,14 +21,18 @@ pub async fn run(
     mut rx: AnalyzeReceiver,
     client: reqwest::Client,
     worker_busy: WorkerBusy,
+    in_flight: InFlight,
     vocab: Arc<Gazetteer>,
     events_tx: EventSender,
 ) {
     info!(vocab_entries = vocab.len(), "Analyze worker started");
 
     while let Some(job) = rx.recv().await {
-        let _guard = BusyGuard::new(worker_busy.clone());
+        let _busy = BusyGuard::new(worker_busy.clone());
-        process(&job.case_dir, &client, &vocab, &events_tx).await;
+        // RAII guard: releases the in-flight slot on every exit path —
+        // success, all failure branches, and panics during the LLM call.
+        let _slot = InFlightGuard::new(in_flight.clone(), job.case_dir.clone());
+        process(&job.case_dir, job.input, &client, &vocab, &events_tx).await;
     }
 
     warn!("Analyze worker stopped (channel closed)");
@@ -35,33 +40,18 @@ pub async fn run(
 
 async fn process(
     case_dir: &Path,
+    input: AnalysisInput,
     client: &reqwest::Client,
     vocab: &Gazetteer,
     events_tx: &EventSender,
 ) {
-    let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
     let document_path = case_dir.join(DOCUMENT_FILE);
 
     // The previous "document exists → skip" guard has been removed:
     // re-analysis is a normal path now (auto-trigger Idle, manual
     // Aktualisieren-Bulk, Neu analysieren-Button), and the previous
     // document.json must be overwritten in place to keep "letztes
-    // Dokument" visible until the new run lands. Recovery still
+    // Dokument" visible until the new run lands.
-    // pre-filters cases with an existing document at recovery.rs:37.
-
-    let input: AnalysisInput = match tokio::fs::read(&input_path).await {
-        Ok(bytes) => match serde_json::from_slice(&bytes) {
-            Ok(v) => v,
-            Err(e) => {
-                error!(path = %input_path.display(), error = %e, "analysis input parse failed");
-                return;
-            }
-        },
-        Err(e) => {
-            error!(path = %input_path.display(), error = %e, "analysis input read failed");
-            return;
-        }
-    };
 
     // Silent-only case: no usable transcripts. Skip LLM, write a stub.
     if input.recordings.is_empty() {
@@ -86,9 +76,6 @@ async fn process(
             emit_analysis_failed(events_tx, case_dir);
             return;
         }
-        if let Err(e) = tokio::fs::remove_file(&input_path).await {
-            warn!(path = %input_path.display(), error = %e, "removing analysis input failed");
-        }
         auto_trigger::remove_failure_marker(case_dir).await;
         info!(case = %case_dir.display(), "analysis done (stub, no recordings)");
         events::emit(
@@ -112,10 +99,6 @@ async fn process(
         "sending to llm"
     );
 
-    // NOTE: If the server crashes between a successful LLM response and the
-    // document write below, the recovery scan will re-enqueue this job and
-    // the call will be made again. Accepted cost — rare event, small
-    // per-call price. Response-caching in a `.tmp` file would prevent it.
     let document = match llm::chat_once(client, backend, &user_content).await {
         Ok(text) => text,
         Err(e) => {
@@ -205,12 +188,6 @@ async fn process(
         return;
     }
 
-    // Job done — drop the queue marker so the live "analyzing" flag
-    // becomes accurate immediately and recovery on next boot does not
-    // re-enqueue this version.
-    if let Err(e) = tokio::fs::remove_file(&input_path).await {
-        warn!(path = %input_path.display(), error = %e, "removing analysis input failed");
-    }
     auto_trigger::remove_failure_marker(case_dir).await;
 
     info!(

server/src/lib.rs

@@ -26,7 +26,7 @@ use axum::extract::FromRef;
 use axum::http::{HeaderName, HeaderValue};
 use tower_http::set_header::SetResponseHeaderLayer;
 
-use analyze::AnalyzeSender;
+use analyze::{AnalyzeSender, InFlight};
 use config::Config;
 use gazetteer::Gazetteer;
 use magic_link::MagicLinkStore;
@@ -114,6 +114,10 @@ pub struct AppState {
     pub http_client: reqwest::Client,
     pub vocab: Arc<Gazetteer>,
     pub boot_time: BootTime,
+    /// Process-local set of cases currently queued or being analyzed.
+    /// Replaces the old "does `analysis_input.json` exist on disk?"
+    /// marker so a server restart never reports stale `Queued` states.
+    pub analyze_in_flight: InFlight,
 }
 
 impl FromRef<AppState> for Arc<Config> {
@@ -213,6 +217,12 @@ impl FromRef<AppState> for BootTime {
     }
 }
 
+impl FromRef<AppState> for InFlight {
+    fn from_ref(state: &AppState) -> Self {
+        state.analyze_in_flight.clone()
+    }
+}
+
 /// Test/simple entrypoint: jobs pushed into either channel are dropped
 /// because the receivers are not retained. Use [`create_router_with_state`]
 /// from `main.rs` where real workers own the receivers.
@@ -241,9 +251,29 @@ pub fn create_router_and_session_store_with_settings(
     config: Arc<Config>,
     settings: Arc<Settings>,
 ) -> (Router, SessionStore) {
+    let (router, session_store, _, _) = create_router_with_full_handles(config, settings);
+    (router, session_store)
+}
+
+/// Test-friendly variant: returns the router, session store, the
+/// process-local [`InFlight`] set, and the receiver half of the
+/// analyze channel. Tests that need to inspect the [`AnalyzeJob`]
+/// payload (recordings, backend_id) consume the receiver to read the
+/// job a handler just sent. Tests that only care about the HTTP
+/// response keep using [`create_router_and_session_store_with_settings`].
+pub fn create_router_with_full_handles(
+    config: Arc<Config>,
+    settings: Arc<Settings>,
+) -> (
+    Router,
+    SessionStore,
+    analyze::InFlight,
+    analyze::AnalyzeReceiver,
+) {
     let (transcribe_tx, _transcribe_rx) = transcribe::channel();
-    let (analyze_tx, _analyze_rx) = analyze::channel();
+    let (analyze_tx, analyze_rx) = analyze::channel();
     let session_store = web_session::new_store();
+    let in_flight = InFlight::new();
     let router = create_router_with_state(AppState {
         config,
         settings,
@@ -263,8 +293,9 @@ pub fn create_router_and_session_store_with_settings(
         // anyway. Tests that exercise the grace-window itself construct
         // an `AppState` with an explicit recent `BootTime`.
         boot_time: BootTime(std::time::SystemTime::UNIX_EPOCH),
+        analyze_in_flight: in_flight.clone(),
     });
-    (router, session_store)
+    (router, session_store, in_flight, analyze_rx)
 }
 
 pub fn create_router_with_state(state: AppState) -> Router {

server/src/main.rs

@@ -190,10 +190,12 @@ async fn main() {
     let (analyze_tx, analyze_rx) = analyze::channel();
     let analyze_busy: doctate_server::WorkerBusy =
         std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let analyze_in_flight = analyze::InFlight::new();
     tokio::spawn(analyze::worker::run(
         analyze_rx,
         http_client.clone(),
         analyze_busy.clone(),
+        analyze_in_flight.clone(),
         vocab.clone(),
         events_tx.clone(),
     ));
@@ -201,12 +203,14 @@ async fn main() {
     {
         let data_path = config.data_path.clone();
         let idle_threshold = std::time::Duration::from_secs(config.analysis_idle_threshold_secs);
+        let in_flight = analyze_in_flight.clone();
         tokio::spawn(async move {
             analyze::recovery::boot_scan_state(
                 &data_path,
                 idle_threshold,
                 std::time::SystemTime::now(),
                 server_boot_time,
+                &in_flight,
             )
             .await;
         });
@@ -227,6 +231,7 @@ async fn main() {
         http_client,
         vocab,
         boot_time: doctate_server::BootTime(server_boot_time),
+        analyze_in_flight,
     };
 
     let addr = format!("0.0.0.0:{}", config.server_port);

server/src/routes/bulk.rs

@@ -8,9 +8,9 @@ use doctate_common::timestamp::now_rfc3339;
 use serde::Deserialize;
 use tracing::{info, warn};
 
-use crate::analyze::auto_trigger::{self, CaseAnalysisState, write_input_overwrite};
+use crate::analyze::auto_trigger::{self, CaseAnalysisState};
 use crate::analyze::backend::any_backend_available;
-use crate::analyze::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender};
+use crate::analyze::{AnalyzeJob, AnalyzeSender, InFlight};
 use crate::auth::AuthenticatedWebUser;
 use crate::config::Config;
 use crate::csrf::{CsrfForm, HasCsrfToken};
@@ -18,9 +18,7 @@ use crate::error::AppError;
 use crate::events::{self, CaseEventKind, EventSender};
 use crate::oneliner_locks::OnelinerLocks;
 use crate::paths::{CloseMarker, write_close_marker};
-use crate::routes::case_actions::{
+use crate::routes::case_actions::{build_analysis_input, reset_case_artefacts};
-    build_analysis_input, reset_case_artefacts, write_input_create_new,
-};
 use crate::routes::user_web::locate_case;
 
 #[derive(Debug, Deserialize)]
@@ -47,6 +45,7 @@ pub async fn handle_bulk_action(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
     State(analyze_tx): State<AnalyzeSender>,
+    State(in_flight): State<InFlight>,
     State(events_tx): State<EventSender>,
     State(locks): State<OnelinerLocks>,
     CsrfForm(form): CsrfForm<BulkForm>,
@@ -73,6 +72,7 @@ pub async fn handle_bulk_action(
         BulkAction::Analyze => {
             bulk_analyze(
                 &analyze_tx,
+                &in_flight,
                 &events_tx,
                 &user.slug,
                 &user_root,
@@ -82,7 +82,15 @@ pub async fn handle_bulk_action(
         }
         BulkAction::Close => bulk_close(&events_tx, &user.slug, &user_root, &form.case_ids).await,
         BulkAction::Reset => {
-            bulk_reset(&events_tx, &locks, &user.slug, &user_root, &form.case_ids).await
+            bulk_reset(
+                &events_tx,
+                &locks,
+                &in_flight,
+                &user.slug,
+                &user_root,
+                &form.case_ids,
+            )
+            .await
         }
     }
 
@@ -91,6 +99,7 @@ pub async fn handle_bulk_action(
 
 async fn bulk_analyze(
     analyze_tx: &AnalyzeSender,
+    in_flight: &InFlight,
     events_tx: &EventSender,
     slug: &str,
     user_root: &std::path::Path,
@@ -113,30 +122,31 @@ async fn bulk_analyze(
             skipped += 1;
             continue;
         };
-        // The previous document.json is left in place: the UI keeps
+        if !in_flight.try_claim(&case_dir) {
-        // rendering it as "letztes Dokument" with a stale-banner until
+            // A job for this case is already queued or running.
-        // the worker overwrites it atomically on the next successful run.
+            skipped += 1;
+            continue;
+        }
         let input = match build_analysis_input(&case_dir, "").await {
             Ok(i) => i,
             Err(_) => {
                 warn!(slug = %slug, case_id = %case_id, "bulk-analyze: build_analysis_input failed");
+                in_flight.release(&case_dir);
                 skipped += 1;
                 continue;
             }
         };
-        let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
-        if write_input_create_new(&input_path, &input).await.is_err() {
-            warn!(slug = %slug, case_id = %case_id, "bulk-analyze: write input failed");
-            skipped += 1;
-            continue;
-        }
         if analyze_tx
             .send(AnalyzeJob {
                 case_dir: case_dir.clone(),
+                input,
             })
             .is_err()
         {
             warn!(slug = %slug, case_id = %case_id, "bulk-analyze: send failed (worker gone)");
+            in_flight.release(&case_dir);
+            skipped += 1;
+            continue;
         }
         events::emit(events_tx, slug, case_id, CaseEventKind::AnalysisQueued);
         ok += 1;
@@ -181,6 +191,7 @@ async fn bulk_close(
 async fn bulk_reset(
     events_tx: &EventSender,
     locks: &OnelinerLocks,
+    in_flight: &InFlight,
     slug: &str,
     user_root: &std::path::Path,
     case_ids: &[String],
@@ -198,6 +209,7 @@ async fn bulk_reset(
             skipped += 1;
             continue;
         };
+        in_flight.release(&case_dir);
         if let Err(e) = reset_case_artefacts(&case_dir, locks).await {
             warn!(slug = %slug, case_id = %case_id, error = %e, "bulk-reset: failed");
             skipped += 1;
@@ -234,6 +246,7 @@ pub async fn handle_analyze_required(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
     State(analyze_tx): State<AnalyzeSender>,
+    State(in_flight): State<InFlight>,
     State(events_tx): State<EventSender>,
     State(boot_time): State<crate::BootTime>,
     CsrfForm(_form): CsrfForm<AnalyzeRequiredForm>,
@@ -276,7 +289,9 @@ pub async fn handle_analyze_required(
         // are intentionally left alone (Failed → manual per-backend retry
         // path on the case page; Queued → already in flight; Current →
         // nothing to do).
-        match auto_trigger::evaluate_state(&case_dir, idle_threshold, now, boot_time.0).await {
+        match auto_trigger::evaluate_state(&case_dir, idle_threshold, now, boot_time.0, &in_flight)
+            .await
+        {
             CaseAnalysisState::Required { .. } | CaseAnalysisState::Idle { .. } => {}
             _ => {
                 skipped += 1;
@@ -284,27 +299,30 @@ pub async fn handle_analyze_required(
             }
         }
 
+        if !in_flight.try_claim(&case_dir) {
+            // A concurrent trigger raced us — they will send the job.
+            skipped += 1;
+            continue;
+        }
+
         let input = match build_analysis_input(&case_dir, "").await {
             Ok(i) => i,
             Err(_) => {
                 warn!(slug = %user.slug, case = %case_dir.display(), "analyze-required: build_analysis_input failed");
+                in_flight.release(&case_dir);
                 skipped += 1;
                 continue;
             }
         };
-        let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
-        if let Err(e) = write_input_overwrite(&input_path, &input).await {
-            warn!(slug = %user.slug, case = %case_dir.display(), error = %e, "analyze-required: write input failed");
-            skipped += 1;
-            continue;
-        }
         if analyze_tx
             .send(AnalyzeJob {
                 case_dir: case_dir.clone(),
+                input,
             })
             .is_err()
         {
             warn!(slug = %user.slug, case = %case_dir.display(), "analyze-required: send failed (worker gone)");
+            in_flight.release(&case_dir);
             skipped += 1;
             continue;
         }

server/src/routes/case_actions.rs

@@ -8,7 +8,6 @@ use axum::response::Redirect;
 use serde::Deserialize;
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
-use tokio::io::AsyncWriteExt;
 use tracing::{info, warn};
 
 use doctate_common::timestamp::{filename_stem_to_recorded_at, now_rfc3339};
@@ -17,7 +16,7 @@ use doctate_common::{RECORDING_META_SUFFIX, TranscriptState};
 use crate::analyze::auto_trigger::FAILURE_MARKER_FILE;
 use crate::analyze::backend::{any_backend_available, find_backend};
 use crate::analyze::{
-    ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE, RecordingInput,
+    AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE, InFlight, RecordingInput,
 };
 use crate::auth::AuthenticatedWebUser;
 use crate::case_id::CaseIdPath;
@@ -54,16 +53,15 @@ impl HasCsrfToken for AnalyzeForm {
 
 /// POST /cases/{case_id}/analyze
 ///
-/// Trigger an LLM analysis for the case. If no document exists yet, writes
+/// Trigger an LLM analysis for the case. Race protection via the
-/// `analysis_input_v1.json`; otherwise writes `analysis_input_v{N+1}.json`
+/// process-local [`InFlight`] set: a duplicate trigger while a job is
-/// where N is the current latest document version. The worker picks up the
+/// already queued or running returns 409. Available to all logged-in
-/// new input and produces `document_v{version}.md`. Available to all logged-in
+/// users; the optional `backend` form field is admin-only.
-/// users; the optional `backend` form field is admin-only. Race protection
-/// via `create_new` on the input file.
 pub async fn handle_analyze_case(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
     State(analyze_tx): State<AnalyzeSender>,
+    State(in_flight): State<InFlight>,
     State(events_tx): State<EventSender>,
     CaseIdPath(case_id): CaseIdPath,
     CsrfForm(form): CsrfForm<AnalyzeForm>,
@@ -96,47 +94,56 @@ pub async fn handle_analyze_case(
     let user_root = config.data_path.join(&user.slug);
     let case_dir = locate_case_or_404(&user_root, &case_id, &user.slug, "analyze").await?;
 
+    // Atomic claim. Concurrent triggers (manual + auto-trigger,
+    // duplicate clicks, second tab) hit this and the loser bails with
+    // 409 — the winner sends a single job.
+    if !in_flight.try_claim(&case_dir) {
+        return Err(AppError::Conflict(
+            "Analyse läuft bereits für diesen Fall".into(),
+        ));
+    }
+
     // Re-analyze: delete the existing document first so the UI stops showing
     // the stale "ausgewertet" state while the worker re-computes. Ignore
     // NotFound (first-time analysis path).
     let document_path = case_dir.join(DOCUMENT_FILE);
-    match tokio::fs::remove_file(&document_path).await {
+    if let Err(e) = tokio::fs::remove_file(&document_path).await
-        Ok(_) => {}
+        && e.kind() != std::io::ErrorKind::NotFound
-        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
+    {
-        Err(e) => return Err(AppError::Internal(format!("remove old document: {e}"))),
+        in_flight.release(&case_dir);
+        return Err(AppError::Internal(format!("remove old document: {e}")));
     }
 
-    let input = build_analysis_input(&case_dir, &backend_id).await?;
+    let input = match build_analysis_input(&case_dir, &backend_id).await {
-    let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
+        Ok(i) => i,
+        Err(e) => {
+            in_flight.release(&case_dir);
+            return Err(e);
+        }
+    };
 
-    // Manual retry path: a failure marker means the worker has explicitly
+    let recordings_count = input.recordings.len();
-    // given up on the prior input. Drop the stale sentinel so create_new
-    // does not reject the legitimate retry as "Analyse läuft bereits".
-    // The marker itself is left in place — the worker overwrites it on
-    // re-failure and removes it on success, so the UI's "analyzing" flag
-    // (driven by input file presence) takes over from the next render.
-    if case_dir.join(FAILURE_MARKER_FILE).exists() {
-        remove_if_exists(&input_path).await?;
-    }
 
-    write_input_create_new(&input_path, &input).await?;
+    // Unbounded send: only fails if the worker is gone (shutdown race).
-
+    // Release the slot so a future request can re-trigger; surface the
-    // Unbounded send: only fails if the worker has gone away (programmer error
+    // failure as a log line rather than 500 to keep the UI honest —
-    // or shutdown race). Log but don't fail the request — the file is already
+    // the user lands on the case page and the missing document is
-    // on disk and recovery will pick it up on next startup.
+    // self-evident.
     if analyze_tx
         .send(AnalyzeJob {
             case_dir: case_dir.clone(),
+            input,
         })
         .is_err()
     {
         warn!(case_id = %case_id, "analyze send failed (worker gone)");
+        in_flight.release(&case_dir);
     }
 
     info!(
         slug = %user.slug,
         case_id = %case_id,
-        recordings = input.recordings.len(),
+        recordings = recordings_count,
         "analyze requested; analysis enqueued"
     );
 
@@ -343,40 +350,6 @@ async fn collect_m4as(case_dir: &Path) -> Result<Vec<(PathBuf, SystemTime)>, App
     Ok(out)
 }
 
-/// Serialize `input` and write it to `path` using `create_new` to atomically
-/// reserve the destination. Returns `Conflict` if another close already won
-/// the race.
-///
-/// Trade-off: if the process crashes between `open` and `write_all`/`sync_all`,
-/// a zero-byte or partially-written JSON remains. The worker will fail to
-/// deserialize it and log an error — manual cleanup required. For MVP the
-/// simpler check-and-create is acceptable.
-pub(crate) async fn write_input_create_new(
-    path: &Path,
-    input: &AnalysisInput,
-) -> Result<(), AppError> {
-    let bytes = serde_json::to_vec_pretty(input)
-        .map_err(|e| AppError::Internal(format!("serialize input: {e}")))?;
-
-    let mut f = match tokio::fs::OpenOptions::new()
-        .write(true)
-        .create_new(true)
-        .open(path)
-        .await
-    {
-        Ok(f) => f,
-        Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
-            return Err(AppError::Conflict(
-                "Analyse läuft bereits für diesen Fall".into(),
-            ));
-        }
-        Err(e) => return Err(AppError::Internal(e.to_string())),
-    };
-    f.write_all(&bytes).await?;
-    f.sync_all().await?;
-    Ok(())
-}
-
 /// Read `case_dir/document.json` and return the embedded markdown
 /// body. Returns `None` if no document exists or the file is not valid
 /// JSON for the [`Document`] envelope. A parse error is logged so a
@@ -394,9 +367,12 @@ pub(crate) async fn read_document(case_dir: &Path) -> Option<String> {
 }
 
 /// Reset a case to its raw audio: delete derived artefacts
-/// (transcripts, analysis input, document, non-Manual oneliner) and rename every
+/// (transcripts, document, non-Manual oneliner, failure marker) and
-/// `*.m4a.failed` back to `*.m4a` so the transcribe self-heal picks it
+/// rename every `*.m4a.failed` back to `*.m4a` so the transcribe
-/// up again. Idempotent: missing files are no-ops.
+/// self-heal picks it up again. Idempotent: missing files are no-ops.
+/// The caller is responsible for releasing any in-flight slot for this
+/// case via [`InFlight::release`] — `reset_case_artefacts` itself is
+/// pure file I/O.
 ///
 /// `oneliner.json` is special: a [`OnelinerState::Manual`] override is
 /// preserved (per the sticky-Manual rule). Routed through
@@ -406,7 +382,11 @@ pub(crate) async fn reset_case_artefacts(
     case_dir: &Path,
     locks: &OnelinerLocks,
 ) -> std::io::Result<()> {
-    for name in [DOCUMENT_FILE, ANALYSIS_INPUT_FILE] {
+    // Failure marker is part of the lifecycle state — without removing
+    // it, a stale `<input-mtime>`-pinned marker would keep classifying
+    // the case as `Failed` until a new recording arrives, even though
+    // the user has explicitly chosen to start over.
+    for name in [DOCUMENT_FILE, FAILURE_MARKER_FILE] {
         match tokio::fs::remove_file(case_dir.join(name)).await {
             Ok(_) => {}
             Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
@@ -552,6 +532,7 @@ pub async fn handle_reset_case(
     State(config): State<Arc<Config>>,
     State(events_tx): State<EventSender>,
     State(locks): State<OnelinerLocks>,
+    State(in_flight): State<InFlight>,
     CaseIdPath(case_id): CaseIdPath,
     CsrfForm(form): CsrfForm<CsrfOnlyForm>,
 ) -> Result<Redirect, AppError> {
@@ -562,6 +543,14 @@ pub async fn handle_reset_case(
     let user_root = config.data_path.join(&user.slug);
     let case_dir = locate_case_or_404(&user_root, &case_id, &user.slug, "reset").await?;
 
+    // A worker job may still be running against the old artefacts; the
+    // RAII `InFlightGuard` will release the slot when it completes. We
+    // also clear the slot here so the UI immediately stops showing
+    // "wird analysiert" — the worker's fresh-write path is harmless,
+    // it will just produce a document the user is about to overwrite
+    // anyway.
+    in_flight.release(&case_dir);
+
     reset_case_artefacts(&case_dir, &locks)
         .await
         .map_err(|e| AppError::Internal(format!("reset: {e}")))?;
@@ -696,6 +685,7 @@ pub async fn handle_delete_recording(
     State(config): State<Arc<Config>>,
     State(events_tx): State<EventSender>,
     State(locks): State<OnelinerLocks>,
+    State(in_flight): State<InFlight>,
     CaseIdPath(case_id): CaseIdPath,
     CsrfForm(form): CsrfForm<DeleteRecordingForm>,
 ) -> Result<Redirect, AppError> {
@@ -709,15 +699,18 @@ pub async fn handle_delete_recording(
         .trim_end_matches(".failed")
         .trim_end_matches(".m4a");
 
+    // The recording set is changing — any pending analysis is now stale.
+    // Release the in-flight slot so the next render auto-triggers fresh.
+    in_flight.release(&case_dir);
+
     // Audio and its single metadata sidecar. NotFound on any of these
     // is fine: the JSON sidecar may never have been written (the
     // worker hadn't reached the atomic write yet, or the file got
     // deleted twice).
-    let targets: [PathBuf; 4] = [
+    let targets: [PathBuf; 3] = [
         case_dir.join(&form.filename),
         case_dir.join(format!("{stem}.{RECORDING_META_SUFFIX}")),
         case_dir.join(DOCUMENT_FILE),
-        case_dir.join(ANALYSIS_INPUT_FILE),
     ];
     for p in &targets {
         remove_if_exists(p).await?;

server/src/routes/user_web.rs

@@ -13,7 +13,7 @@ use tracing::{info, warn};
 
 use time::{Date, OffsetDateTime, format_description::well_known::Rfc3339};
 
-use crate::analyze::{ANALYSIS_INPUT_FILE, DOCUMENT_FILE, auto_trigger};
+use crate::analyze::{DOCUMENT_FILE, InFlight, auto_trigger};
 use crate::auth::AuthenticatedWebUser;
 use crate::case_id::{CaseId, CaseIdPath};
 use crate::config::Config;
@@ -479,12 +479,13 @@ async fn compute_flags(
     idle_threshold: std::time::Duration,
     now_systime: std::time::SystemTime,
     boot_time: std::time::SystemTime,
+    in_flight: &InFlight,
 ) -> CaseFlags {
     let has_document = any_document_exists(case_dir).await;
-    // Honest status: an input file alone does not mean a job is in flight.
+    // Honest status: a claimed in-flight slot is the only authoritative
-    // The worker may be idle and the file an orphan from a crash. Page-level
+    // signal that a job is queued or running. The set is process-local
-    // self-heal (see `heal_orphans_if_idle`) re-enqueues those.
+    // so it cannot survive a crash — no orphan markers any more.
-    let analyzing = !has_document && worker_busy && any_analysis_input_exists(case_dir).await;
+    let analyzing = !has_document && worker_busy && in_flight.contains(case_dir);
 
     let non_failed: Vec<&RecordingView> = recordings.iter().filter(|r| !r.failed).collect();
     let all_transcribed =
@@ -521,7 +522,8 @@ async fn compute_flags(
     };
 
     let state =
-        auto_trigger::evaluate_state(case_dir, idle_threshold, now_systime, boot_time).await;
+        auto_trigger::evaluate_state(case_dir, idle_threshold, now_systime, boot_time, in_flight)
+            .await;
     let analysis_stale_with_doc = matches!(
         state,
         auto_trigger::CaseAnalysisState::Required { has_document: true }
@@ -538,14 +540,6 @@ async fn compute_flags(
     }
 }
 
-/// True iff `analysis_input.json` exists. The worker removes it after a
-/// successful document write, so existence implies a pending or in-flight job.
-pub(crate) async fn any_analysis_input_exists(case_dir: &Path) -> bool {
-    tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE))
-        .await
-        .unwrap_or(false)
-}
-
 /// True iff `document.md` exists.
 pub(crate) async fn any_document_exists(case_dir: &Path) -> bool {
     tokio::fs::try_exists(case_dir.join(DOCUMENT_FILE))
@@ -629,6 +623,7 @@ pub async fn handle_my_cases(
     State(http_client): State<reqwest::Client>,
     State(vocab): State<Arc<Gazetteer>>,
     State(boot_time): State<crate::BootTime>,
+    State(in_flight): State<InFlight>,
     Query(query): Query<CaseListQuery>,
 ) -> Result<Html<String>, AppError> {
     let user_root = config.data_path.join(&user.slug);
@@ -653,6 +648,7 @@ pub async fn handle_my_cases(
         idle_threshold,
         now,
         boot_time.0,
+        &in_flight,
         &pipeline.analyze_tx,
         &events_tx,
     )
@@ -679,6 +675,7 @@ pub async fn handle_my_cases(
         retention.auto_delete_days,
         crate::analyze::backend::any_backend_available(),
         boot_time.0,
+        &in_flight,
     )
     .await;
     let total = cases.len();
@@ -737,6 +734,7 @@ pub async fn handle_case_page(
     State(http_client): State<reqwest::Client>,
     State(vocab): State<Arc<Gazetteer>>,
     State(boot_time): State<crate::BootTime>,
+    State(in_flight): State<InFlight>,
     CaseIdPath(case_id): CaseIdPath,
     Query(query): Query<CaseListQuery>,
 ) -> Result<Html<String>, AppError> {
@@ -780,6 +778,7 @@ pub async fn handle_case_page(
         idle_threshold,
         now,
         boot_time.0,
+        &in_flight,
         &pipeline.analyze_tx,
         &events_tx,
     )
@@ -809,6 +808,7 @@ pub async fn handle_case_page(
         idle_threshold,
         now,
         boot_time.0,
+        &in_flight,
     )
     .await;
 
@@ -1015,6 +1015,7 @@ async fn scan_user_cases(
     auto_delete_days: u32,
     llm_configured: bool,
     boot_time: std::time::SystemTime,
+    in_flight: &InFlight,
 ) -> Vec<UserCaseView> {
     let user_root = config.data_path.join(slug);
     let mut cases = Vec::new();
@@ -1040,6 +1041,7 @@ async fn scan_user_cases(
             idle_threshold,
             now_systime,
             boot_time,
+            in_flight,
         )
         .await
         {
@@ -1086,6 +1088,7 @@ async fn compute_case_view(
     idle_threshold: std::time::Duration,
     now_systime: std::time::SystemTime,
     boot_time: std::time::SystemTime,
+    in_flight: &InFlight,
 ) -> Option<UserCaseView> {
     let recordings = scan_recordings(case_path).await;
     if recordings.is_empty() {
@@ -1111,11 +1114,12 @@ async fn compute_case_view(
     let (is_closed, days_until_purge) = closed_state(case_path, auto_delete_days, now).await;
 
     // The typed lifecycle is the single source of truth for both the
-    // badge and the "is the Analysieren-button enabled?"
+    // badge and the "is the Analysieren-button enabled?" decision: a
-    // decision: a case in `Queued` (i.e. `analysis_input.json` on disk)
+    // case in `Queued` (claim held in [`InFlight`]) must not let the
-    // must not let the user enqueue a duplicate job.
+    // user enqueue a duplicate job.
     let state =
-        auto_trigger::evaluate_state(case_path, idle_threshold, now_systime, boot_time).await;
+        auto_trigger::evaluate_state(case_path, idle_threshold, now_systime, boot_time, in_flight)
+            .await;
     let analysis_in_flight = matches!(state, auto_trigger::CaseAnalysisState::Queued);
     // Closed cases must never advertise as "Analyse erforderlich" — the
     // whole point of closing a case is to freeze it. Suppress the badge

server/tests/analyze_test.rs

@@ -188,7 +188,7 @@ async fn analyze_case_without_llm_returns_503() {
 // ---------------------------------------------------------------------
 
 #[tokio::test]
-async fn analyze_case_happy_path_writes_input_and_redirects() {
+async fn analyze_case_happy_path_sends_job_and_redirects() {
     let (cfg, settings) = cfg_llm("g");
     let case_id = "11111111-1111-1111-1111-111111111111";
     let case_dir = seed_case(&cfg.data_path, "dr_a", case_id);
@@ -203,7 +203,8 @@ async fn analyze_case_happy_path_writes_input_and_redirects() {
         Some("Korrektur: links, nicht rechts."),
     );
 
-    let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
+    let (app, store, in_flight, mut analyze_rx) =
+        doctate_server::create_router_with_full_handles(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
 
     let resp = app
@@ -225,17 +226,26 @@ async fn analyze_case_happy_path_writes_input_and_redirects() {
         format!("/cases/{case_id}")
     );
 
-    let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
+    // Slot is claimed for the duration of the worker's run — proxied
-    assert!(input_path.exists(), "analysis_input.json missing");
+    // here by the receiver still holding the un-consumed job.
+    assert!(
+        in_flight.contains(&case_dir),
+        "in-flight slot must be held until the worker drops it"
+    );
 
-    let raw = std::fs::read_to_string(&input_path).unwrap();
+    let job = analyze_rx
-    let parsed: Value = serde_json::from_str(&raw).unwrap();
+        .try_recv()
-    let recs = parsed["recordings"].as_array().unwrap();
+        .expect("handler must enqueue exactly one job");
-    assert_eq!(recs.len(), 2);
+    assert_eq!(job.case_dir, case_dir);
-    assert_eq!(recs[0]["recorded_at"], "2026-04-15T10:00:00Z");
+    assert_eq!(job.input.recordings.len(), 2);
-    assert_eq!(recs[0]["text"], "Patient klagt über Knie.");
+    assert_eq!(job.input.recordings[0].recorded_at, "2026-04-15T10:00:00Z");
-    assert_eq!(recs[1]["recorded_at"], "2026-04-15T10:05:00Z");
+    assert_eq!(job.input.recordings[0].text, "Patient klagt über Knie.");
-    assert!(parsed["last_recording_mtime"].is_string());
+    assert_eq!(job.input.recordings[1].recorded_at, "2026-04-15T10:05:00Z");
+    assert!(!job.input.last_recording_mtime.is_empty());
+    assert!(
+        analyze_rx.try_recv().is_err(),
+        "exactly one job must be enqueued"
+    );
 }
 
 #[tokio::test]
@@ -245,7 +255,12 @@ async fn analyze_case_second_time_returns_409() {
     let case_dir = seed_case(&cfg.data_path, "dr_a", case_id);
     seed_recording(&case_dir, "2026-04-15T10-00-00Z", Some("text"));
 
-    let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
+    // Hold the receiver alive across both requests — otherwise the
+    // first send fails, the slot is released, and the second click
+    // would be accepted as a fresh trigger. In production the worker
+    // keeps the slot held via `InFlightGuard` until it finishes.
+    let (app, store, _in_flight, _analyze_rx) =
+        doctate_server::create_router_with_full_handles(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
 
     let first = app
@@ -273,32 +288,29 @@ async fn analyze_case_second_time_returns_409() {
 }
 
 /// Bug regression (case c414cf52, 2026-05-03): a prior LLM run failed,
-/// leaving `analysis_input.json` AND `.analysis_failed.json` on disk.
+/// leaving `.analysis_failed.json` on disk. The user clicks "Erneut
-/// The user clicks "Erneut versuchen" — the handler must accept the
+/// versuchen" — the handler must accept the retry. With the in-flight
-/// retry instead of rejecting it as 409 Conflict ("Analyse läuft
+/// marker now process-local, no on-disk sentinel can falsely block the
-/// bereits"). The failure marker is the explicit "worker has given up"
+/// retry; the explicit user click claims a fresh slot and sends a job
-/// signal that legitimises overwriting the stale sentinel.
+/// with the current transcripts.
 #[tokio::test]
-async fn analyze_retry_after_failure_marker_overwrites_stale_input() {
+async fn analyze_retry_after_failure_marker_succeeds() {
     let (cfg, settings) = cfg_llm("retry-after-fail");
     let case_id = "11111111-1111-1111-1111-111111111111";
     let case_dir = seed_case(&cfg.data_path, "dr_a", case_id);
     seed_recording(&case_dir, "2026-04-15T10-00-00Z", Some("frischer Text"));
 
-    // Simulate the post-failure on-disk state: stale input from the
+    // Failure marker from the prior aborted attempt remains on disk —
-    // attempt the worker gave up on, plus the matching failure marker.
+    // the new model leaves it in place; the worker overwrites it on
-    std::fs::write(
+    // re-failure or removes it on success.
-        case_dir.join(ANALYSIS_INPUT_FILE),
-        r#"{"last_recording_mtime":"2026-04-15T10:00:00Z","recordings":[{"recorded_at":"2026-04-15T10:00:00Z","text":"alter Text"}],"backend_id":""}"#,
-    )
-    .unwrap();
     std::fs::write(
         case_dir.join(".analysis_failed.json"),
         r#"{"last_recording_mtime":"2026-04-15T10:00:00Z","reason":"llm: connect timeout","failed_at":"2026-04-15T10:05:00Z"}"#,
     )
     .unwrap();
 
-    let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
+    let (app, store, _in_flight, mut analyze_rx) =
+        doctate_server::create_router_with_full_handles(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
 
     let resp = app
@@ -316,13 +328,13 @@ async fn analyze_retry_after_failure_marker_overwrites_stale_input() {
         "retry with failure marker present must succeed, not 409"
     );
 
-    // Input was rewritten with the fresh transcript — the stale "alter Text"
+    // The job carries the fresh transcript — there is no on-disk state
-    // must be gone.
+    // to validate any more, the channel is the truth.
-    let raw = std::fs::read_to_string(case_dir.join(ANALYSIS_INPUT_FILE)).unwrap();
+    let job = analyze_rx
-    let parsed: Value = serde_json::from_str(&raw).unwrap();
+        .try_recv()
-    let recs = parsed["recordings"].as_array().unwrap();
+        .expect("retry must enqueue a fresh job");
-    assert_eq!(recs.len(), 1);
+    assert_eq!(job.input.recordings.len(), 1);
-    assert_eq!(recs[0]["text"], "frischer Text");
+    assert_eq!(job.input.recordings[0].text, "frischer Text");
 }
 
 #[tokio::test]
@@ -334,7 +346,8 @@ async fn analyze_case_skips_blank_transcript_from_recordings() {
     seed_recording(&case_dir, "2026-04-15T10-05-00Z", Some("   ")); // blank
     seed_recording(&case_dir, "2026-04-15T10-10-00Z", Some("auch echt"));
 
-    let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
+    let (app, store, _in_flight, mut analyze_rx) =
+        doctate_server::create_router_with_full_handles(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
 
     let resp = app
@@ -348,10 +361,13 @@ async fn analyze_case_skips_blank_transcript_from_recordings() {
         .unwrap();
     assert_eq!(resp.status(), StatusCode::SEE_OTHER);
 
-    let raw = std::fs::read_to_string(case_dir.join(ANALYSIS_INPUT_FILE)).unwrap();
+    let job = analyze_rx.try_recv().expect("job must be enqueued");
-    let parsed: Value = serde_json::from_str(&raw).unwrap();
+    assert_eq!(
-    let recs = parsed["recordings"].as_array().unwrap();
+        job.input.recordings.len(),
-    assert_eq!(recs.len(), 2, "blank transcript must be filtered out");
+        2,
+        "blank transcript must be filtered out"
+    );
+    let _ = case_dir; // name kept for grep symmetry with the seed calls
 }
 
 // ---------------------------------------------------------------------
@@ -376,17 +392,14 @@ async fn analyze_worker_writes_document_via_wiremock() {
     let case_dir = tmp.join("dr_a/11111111-1111-1111-1111-111111111111");
     std::fs::create_dir_all(&case_dir).unwrap();
 
-    let input = json!({
+    let input = analyze::AnalysisInput {
-        "last_recording_mtime": "2026-04-15T10:00:00Z",
+        last_recording_mtime: "2026-04-15T10:00:00Z".to_owned(),
-        "recordings": [
+        recordings: vec![analyze::RecordingInput {
-            { "recorded_at": "2026-04-15T10:00:00Z", "text": "Patient mit Knie." }
+            recorded_at: "2026-04-15T10:00:00Z".to_owned(),
-        ]
+            text: "Patient mit Knie.".to_owned(),
-    });
+        }],
-    std::fs::write(
+        backend_id: String::new(),
-        case_dir.join(ANALYSIS_INPUT_FILE),
+    };
-        serde_json::to_vec_pretty(&input).unwrap(),
-    )
-    .unwrap();
 
     let mock = MockServer::start().await;
     Mock::given(method("POST"))
@@ -410,16 +423,19 @@ async fn analyze_worker_writes_document_via_wiremock() {
     let client = reqwest::Client::new();
     let busy = Arc::new(std::sync::atomic::AtomicBool::new(false));
     let vocab = Arc::new(doctate_server::gazetteer::Gazetteer::empty());
+    let in_flight = analyze::InFlight::new();
     let handle = tokio::spawn(analyze::worker::run(
         rx,
         client,
         busy,
+        in_flight,
         vocab,
         doctate_server::events::channel(),
     ));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),
+        input,
     })
     .unwrap();
 
@@ -455,17 +471,14 @@ async fn analyze_worker_normalizes_llm_output() {
     let case_dir = tmp.join("dr_a/22222222-2222-2222-2222-222222222222");
     std::fs::create_dir_all(&case_dir).unwrap();
 
-    let input = json!({
+    let input = analyze::AnalysisInput {
-        "last_recording_mtime": "2026-04-15T10:00:00Z",
+        last_recording_mtime: "2026-04-15T10:00:00Z".to_owned(),
-        "recordings": [
+        recordings: vec![analyze::RecordingInput {
-            { "recorded_at": "2026-04-15T10:00:00Z", "text": "Patient mit Blutung im Zerebrum." }
+            recorded_at: "2026-04-15T10:00:00Z".to_owned(),
-        ]
+            text: "Patient mit Blutung im Zerebrum.".to_owned(),
-    });
+        }],
-    std::fs::write(
+        backend_id: String::new(),
-        case_dir.join(ANALYSIS_INPUT_FILE),
+    };
-        serde_json::to_vec_pretty(&input).unwrap(),
-    )
-    .unwrap();
 
     // Vocab directory with a single anatomy entry.
     let vocab_dir = unique_tmpdir("analyze-vocab");
@@ -501,16 +514,19 @@ async fn analyze_worker_normalizes_llm_output() {
     let (tx, rx) = analyze::channel();
     let client = reqwest::Client::new();
     let busy = Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let in_flight = analyze::InFlight::new();
     let handle = tokio::spawn(analyze::worker::run(
         rx,
         client,
         busy,
+        in_flight,
         vocab,
         doctate_server::events::channel(),
     ));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),
+        input,
     })
     .unwrap();
 
@@ -613,23 +629,27 @@ async fn replace_real_case_dry() {
 // Recovery
 // ---------------------------------------------------------------------
 
-/// Boot-scan is observation-only after the typed-state refactor: even a
+/// Boot-scan is observation-only after the typed-state refactor: it
-/// case sitting in the `Queued` state on disk must NOT be re-enqueued
+/// must never enqueue jobs. With the in-flight set replacing on-disk
-/// at startup. The per-page-render `try_enqueue_all_for_user` is the
+/// markers, an orphan `analysis_input.json` left over from a
-/// only auto-trigger.
+/// pre-refactor server run is a no-op: the empty `InFlight` set means
+/// the case is not classified as `Queued`.
 #[tokio::test]
 async fn recovery_does_not_enqueue_at_boot() {
     let tmp = unique_tmpdir("analyze-r1");
     let case_dir = tmp.join("dr_a/11111111-1111-1111-1111-111111111111");
     std::fs::create_dir_all(&case_dir).unwrap();
+    // Pre-refactor leftover; ignored by the new model.
     std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap();
 
     let (_tx, mut rx) = analyze::channel();
+    let in_flight = analyze::InFlight::new();
     analyze::recovery::boot_scan_state(
         &tmp,
         std::time::Duration::ZERO,
         std::time::SystemTime::now(),
         std::time::SystemTime::UNIX_EPOCH,
+        &in_flight,
     )
     .await;
 
@@ -644,7 +664,7 @@ async fn recovery_does_not_enqueue_at_boot() {
 // ---------------------------------------------------------------------
 
 #[tokio::test]
-async fn analyze_deletes_old_document_and_writes_fresh_input() {
+async fn analyze_deletes_old_document_and_sends_fresh_job() {
     let (cfg, settings) = cfg_llm("rn-5");
     let case_id = "11111111-1111-1111-1111-111111111111";
     let case_dir = seed_case(&cfg.data_path, "dr_a", case_id);
@@ -652,7 +672,8 @@ async fn analyze_deletes_old_document_and_writes_fresh_input() {
     seed_recording(&case_dir, "2026-04-15T10-00-00Z", Some("erste Aufnahme"));
     seed_recording(&case_dir, "2026-04-15T10-05-00Z", Some("zweite Aufnahme"));
 
-    let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
+    let (app, store, _in_flight, mut analyze_rx) =
+        doctate_server::create_router_with_full_handles(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
 
     let resp = app
@@ -670,11 +691,8 @@ async fn analyze_deletes_old_document_and_writes_fresh_input() {
         !case_dir.join(DOCUMENT_FILE).exists(),
         "old document must be deleted before re-analysis"
     );
-    let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
+    let job = analyze_rx.try_recv().expect("re-analysis must enqueue");
-    assert!(input_path.exists(), "analysis_input.json missing");
+    assert_eq!(job.input.recordings.len(), 2);
-    let parsed: Value =
-        serde_json::from_str(&std::fs::read_to_string(&input_path).unwrap()).unwrap();
-    assert_eq!(parsed["recordings"].as_array().unwrap().len(), 2);
 }
 
 // ---------------------------------------------------------------------
@@ -1211,7 +1229,8 @@ async fn bulk_analyze_processes_each_selected_case() {
     seed_recording(&dir_a, "2026-04-15T10-00-00Z", Some("a"));
     seed_recording(&dir_b, "2026-04-15T10-05-00Z", Some("b"));
 
-    let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
+    let (app, store, in_flight, mut analyze_rx) =
+        doctate_server::create_router_with_full_handles(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
 
     let body = format!(
@@ -1224,8 +1243,15 @@ async fn bulk_analyze_processes_each_selected_case() {
         .unwrap();
     assert_eq!(resp.status(), StatusCode::SEE_OTHER);
 
-    assert!(dir_a.join(ANALYSIS_INPUT_FILE).exists());
+    assert!(in_flight.contains(&dir_a));
-    assert!(dir_b.join(ANALYSIS_INPUT_FILE).exists());
+    assert!(in_flight.contains(&dir_b));
+    let mut received = Vec::new();
+    while let Ok(job) = analyze_rx.try_recv() {
+        received.push(job.case_dir);
+    }
+    assert_eq!(received.len(), 2, "both selected cases must be enqueued");
+    assert!(received.iter().any(|p| p == &dir_a));
+    assert!(received.iter().any(|p| p == &dir_b));
 }
 
 #[tokio::test]
@@ -1233,18 +1259,20 @@ async fn recovery_skips_closed_cases() {
     let tmp = unique_tmpdir("analyze-rec-closed");
     let case_dir = tmp.join("dr_a/11111111-1111-1111-1111-111111111111");
     std::fs::create_dir_all(&case_dir).unwrap();
-    std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap();
     std::fs::write(
         case_dir.join(CLOSE_MARKER),
         r#"{"closed_at":"2026-04-15T10:00:00Z"}"#,
     )
     .unwrap();
 
+    let in_flight = analyze::InFlight::new();
+    in_flight.try_claim(&case_dir);
     let summary = analyze::recovery::boot_scan_state(
         &tmp,
         std::time::Duration::ZERO,
         std::time::SystemTime::now(),
         std::time::SystemTime::UNIX_EPOCH,
+        &in_flight,
     )
     .await;
 
@@ -1276,7 +1304,15 @@ async fn reset_case_clears_derived_and_unfails_audio() {
     )
     .unwrap();
     write_document_for_test(&dir, "# Doc\n", "1970-01-01T00:00:00Z");
-    std::fs::write(dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap();
+    // Failure marker from a prior aborted run — the bonus-fix says
+    // reset must wipe this too, otherwise the case stays classified
+    // as `Failed` forever after the user explicitly requested a fresh
+    // start.
+    std::fs::write(
+        dir.join(".analysis_failed.json"),
+        r#"{"last_recording_mtime":"2026-04-15T09:00:00Z","reason":"x","failed_at":"2026-04-15T09:01:00Z"}"#,
+    )
+    .unwrap();
 
     let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
@@ -1304,7 +1340,10 @@ async fn reset_case_clears_derived_and_unfails_audio() {
     assert!(!dir.join("2026-04-15T09-00-00Z.json").exists());
     assert!(!dir.join(ONELINER_FILENAME).exists());
     assert!(!dir.join(DOCUMENT_FILE).exists());
-    assert!(!dir.join(ANALYSIS_INPUT_FILE).exists());
+    assert!(
+        !dir.join(".analysis_failed.json").exists(),
+        "reset must clear the failure marker so the case is no longer Failed"
+    );
 }
 
 #[tokio::test]
@@ -1440,17 +1479,14 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() {
     let case_dir = tmp.join("dr_a/22222222-2222-2222-2222-222222222222");
     std::fs::create_dir_all(&case_dir).unwrap();
 
-    let input = json!({
+    let input = analyze::AnalysisInput {
-        "last_recording_mtime": "2026-04-16T10:00:00Z",
+        last_recording_mtime: "2026-04-16T10:00:00Z".to_owned(),
-        "recordings": [
+        recordings: vec![analyze::RecordingInput {
-            { "recorded_at": "2026-04-16T10:00:00Z", "text": "Test." }
+            recorded_at: "2026-04-16T10:00:00Z".to_owned(),
-        ]
+            text: "Test.".to_owned(),
-    });
+        }],
-    std::fs::write(
+        backend_id: String::new(),
-        case_dir.join(ANALYSIS_INPUT_FILE),
+    };
-        serde_json::to_vec_pretty(&input).unwrap(),
-    )
-    .unwrap();
 
     let mock = MockServer::start().await;
     Mock::given(method("POST"))
@@ -1483,16 +1519,19 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() {
     let client = reqwest::Client::new();
     let busy = Arc::new(std::sync::atomic::AtomicBool::new(false));
     let vocab = Arc::new(doctate_server::gazetteer::Gazetteer::empty());
+    let in_flight = analyze::InFlight::new();
     let handle = tokio::spawn(analyze::worker::run(
         rx,
         client,
         busy,
+        in_flight,
         vocab,
         doctate_server::events::channel(),
     ));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),
+        input,
     })
     .unwrap();
 
@@ -1518,9 +1557,8 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() {
 
 /// `POST /cases/analyze-required` walks every UUID-named case and force-
 /// enqueues those in `Required`/`Idle`. `Failed` and `Current` cases must
-/// stay untouched. We probe the resulting state via the on-disk
+/// stay untouched. The InFlight set is the canonical "would have been
-/// `analysis_input.json` marker — written before the channel send, so its
+/// enqueued" signal; the channel receiver lets us count delivered jobs.
-/// presence is the canonical "would have been enqueued" signal.
 #[tokio::test]
 async fn analyze_required_endpoint_enqueues_only_required_cases() {
     use doctate_server::analyze::auto_trigger::FailureMarker;
@@ -1570,7 +1608,8 @@ async fn analyze_required_endpoint_enqueues_only_required_cases() {
     )
     .unwrap();
 
-    let (app, store) = doctate_server::create_router_and_session_store_with_settings(cfg, settings);
+    let (app, store, in_flight, mut analyze_rx) =
+        doctate_server::create_router_with_full_handles(cfg, settings);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
 
     let resp = app
@@ -1584,31 +1623,42 @@ async fn analyze_required_endpoint_enqueues_only_required_cases() {
         "must redirect back to the case list"
     );
 
-    // Required → analysis_input.json was written, the case is now Queued.
+    let req_path = user_root.join(req_id);
+    let cur_path = user_root.join(cur_id);
+    let fail_path = user_root.join(fail_id);
+
+    // Required → claim held + job in channel.
     assert!(
-        user_root.join(req_id).join(ANALYSIS_INPUT_FILE).exists(),
+        in_flight.contains(&req_path),
-        "Required case must have been enqueued"
+        "Required case must have a claim"
     );
-    // Current → no enqueue, document.json untouched.
+    // Current → no claim, document.json untouched.
     assert!(
-        !user_root.join(cur_id).join(ANALYSIS_INPUT_FILE).exists(),
+        !in_flight.contains(&cur_path),
-        "Current case must not be enqueued"
+        "Current case must not be claimed"
     );
     assert!(
-        user_root.join(cur_id).join(DOCUMENT_FILE).exists(),
+        cur_path.join(DOCUMENT_FILE).exists(),
         "Current case's document must remain in place"
     );
-    // Failed → no enqueue. Marker stays so the user sees the failure
+    // Failed → no claim. Marker stays so the user sees the failure
     // banner on the case page and can retry per-backend manually.
     assert!(
-        !user_root.join(fail_id).join(ANALYSIS_INPUT_FILE).exists(),
+        !in_flight.contains(&fail_path),
-        "Failed case must not be enqueued"
+        "Failed case must not be claimed"
     );
     assert!(
-        user_root
+        fail_path.join(".analysis_failed.json").exists(),
-            .join(fail_id)
-            .join(".analysis_failed.json")
-            .exists(),
         "failure marker must persist"
     );
+
+    let mut received = Vec::new();
+    while let Ok(job) = analyze_rx.try_recv() {
+        received.push(job.case_dir);
+    }
+    assert_eq!(
+        received,
+        vec![req_path],
+        "only the Required case must be enqueued"
+    );
 }

server/tests/common/artefacts.rs

@@ -11,9 +11,16 @@
 use std::path::Path;
 
 pub use doctate_common::ONELINER_FILENAME;
-pub use doctate_server::analyze::{ANALYSIS_INPUT_FILE, DOCUMENT_FILE, Document};
+pub use doctate_server::analyze::{DOCUMENT_FILE, Document};
 pub use doctate_server::paths::CLOSE_MARKER;
 
+/// Legacy filename of the per-case analysis input. The server no
+/// longer writes or reads it — the in-flight marker lives in `InFlight`
+/// and the payload travels with the `AnalyzeJob`. This constant is
+/// retained only for tests that simulate an orphan file from a
+/// pre-refactor server. New tests should not seed this file.
+pub const ANALYSIS_INPUT_FILE: &str = "analysis_input.json";
+
 /// Seed a `document.json` for tests. `covered_mtime` is the RFC3339
 /// snapshot mtime the document is supposed to "cover" — for tests that
 /// don't care about the auto-trigger state machine, pass any RFC3339

server/tests/delete_recording_test.rs

@@ -14,9 +14,9 @@ use doctate_common::oneliners::OnelinerState;
 use tower::util::ServiceExt;
 
 use common::{
-    ANALYSIS_INPUT_FILE, DOCUMENT_FILE, ONELINER_FILENAME, TestConfig, csrf_form_post,
+    DOCUMENT_FILE, ONELINER_FILENAME, TestConfig, csrf_form_post, login_with_csrf, paths,
-    login_with_csrf, paths, seed_case, seed_oneliner_ready, seed_oneliner_state,
+    seed_case, seed_oneliner_ready, seed_oneliner_state, seed_recording_with_sidecars, test_user,
-    seed_recording_with_sidecars, test_user, write_document_for_test,
+    write_document_for_test,
 };
 
 fn delete_body(filename: &str) -> String {
@@ -38,7 +38,6 @@ async fn delete_recording_removes_file_and_sidecars_and_invalidates_derived() {
     // delete — keeps the original contract „auto state must be wiped".
     seed_oneliner_ready(&case_dir, "knee, left, pain");
     write_document_for_test(&case_dir, "# stale", "1970-01-01T00:00:00Z");
-    std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), b"{}").unwrap();
 
     let (app, store) = doctate_server::create_router_and_session_store(cfg);
     let (cookie, csrf) = login_with_csrf(&app, &store, "dr_a", "s").await;
@@ -91,10 +90,6 @@ async fn delete_recording_removes_file_and_sidecars_and_invalidates_derived() {
         !case_dir.join(DOCUMENT_FILE).exists(),
         "document.md must be cleared"
     );
-    assert!(
-        !case_dir.join(ANALYSIS_INPUT_FILE).exists(),
-        "analysis_input.json must be cleared"
-    );
 }
 
 #[tokio::test]

server/tests/magic_link_test.rs

@@ -72,6 +72,7 @@ fn build_state_with_stores() -> AppState {
         http_client: reqwest::Client::new(),
         vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()),
         boot_time: doctate_server::BootTime(std::time::SystemTime::UNIX_EPOCH),
+        analyze_in_flight: doctate_server::analyze::InFlight::new(),
     }
 }
 

server/tests/oneliner_override_test.rs

@@ -85,6 +85,7 @@ fn build_app_with_events_tx(
         http_client: reqwest::Client::new(),
         vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()),
         boot_time: doctate_server::BootTime(std::time::SystemTime::UNIX_EPOCH),
+        analyze_in_flight: doctate_server::analyze::InFlight::new(),
     }
 }