Refactor busy flags and worker guards

The `WorkerBusy` type was previously a single `Arc<AtomicBool>` shared
between the analyze and transcribe workers. This commit refactors this
to:

- Introduce `AnalyzeBusy` and `TranscribeBusy` newtype wrappers around
  `WorkerBusy` to distinguish between the two flags. This allows
  `axum::extract::State` to target them individually.
- Move the `BusyGuard` RAII guard into `src/lib.rs` and make it generic
  to work with any `WorkerBusy` instance.
- Update the `main.rs`, `worker.rs`, and `routes/user_web.rs` files to
  use the new types and guards, ensuring correct state management for
  both pipelines.
- Enhance the transcription recovery scan
  (`transcribe::recovery::scan_and_enqueue`) to iterate over user
  directories and call `enqueue_pending_for_user` for each, making it
  more robust and aligned with the per-user self-healing mechanism.
- Add a check for `transcribe_busy` in the `case_detail.html` template
  to correctly display "Transkription läuft…" only when the transcribe
  worker is actually active.

server/src/analyze/worker.rs

@@ -1,5 +1,4 @@
 use std::path::Path;
-use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -8,25 +7,7 @@ use tracing::{error, info, warn};
 
 use super::{llm, prompt, AnalysisInput, AnalyzeReceiver};
 use crate::config::Config;
-use crate::WorkerBusy;
-
-/// RAII guard that flips `WorkerBusy` to `true` on construction and back to
-/// `false` on drop. Panic-safe: even if `process()` unwinds, the flag is
-/// cleared.
-struct BusyGuard(WorkerBusy);
-
-impl BusyGuard {
-    fn new(busy: WorkerBusy) -> Self {
-        busy.store(true, Ordering::Release);
-        Self(busy)
-    }
-}
-
-impl Drop for BusyGuard {
-    fn drop(&mut self) {
-        self.0.store(false, Ordering::Release);
-    }
-}
+use crate::{BusyGuard, WorkerBusy};
 
 /// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
 /// but sequential processing keeps the architecture simple and shields the

server/src/lib.rs

@@ -8,7 +8,7 @@ pub mod routes;
 pub mod transcribe;
 pub mod web_session;
 
-use std::sync::atomic::AtomicBool;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
 
 use axum::extract::FromRef;
@@ -19,12 +19,36 @@ use config::Config;
 use transcribe::TranscribeSender;
 use web_session::SessionStore;
 
-/// Live "is the analyze worker currently processing a job?" flag.
-/// Set true before each `process()` call, false after. UI uses it to
-/// distinguish a real in-flight analysis from a stale `analysis_input_v*.json`
-/// orphan left over from a crash.
+/// Live "is this worker currently processing a job?" flag.
+/// Set true before each job, false after. UI uses it to distinguish a real
+/// in-flight job from a stale on-disk marker (orphan input, missing
+/// transcript) left over from a crash.
 pub type WorkerBusy = Arc<AtomicBool>;
 
+/// Newtype wrappers so `FromRef<AppState>` can target the two busy-flags
+/// separately even though both are `Arc<AtomicBool>` underneath.
+#[derive(Clone)]
+pub struct AnalyzeBusy(pub WorkerBusy);
+#[derive(Clone)]
+pub struct TranscribeBusy(pub WorkerBusy);
+
+/// RAII guard that flips a [`WorkerBusy`] true on construction and false on
+/// drop — panic-safe. Shared across analyze + transcribe workers.
+pub struct BusyGuard(WorkerBusy);
+
+impl BusyGuard {
+    pub fn new(busy: WorkerBusy) -> Self {
+        busy.store(true, Ordering::Release);
+        Self(busy)
+    }
+}
+
+impl Drop for BusyGuard {
+    fn drop(&mut self) {
+        self.0.store(false, Ordering::Release);
+    }
+}
+
 /// Shared application state. Clonable — all fields are cheap to clone
 /// (`Arc` and `mpsc::Sender`).
 #[derive(Clone)]
@@ -33,7 +57,8 @@ pub struct AppState {
     pub transcribe_tx: TranscribeSender,
     pub analyze_tx: AnalyzeSender,
     pub session_store: SessionStore,
-    pub worker_busy: WorkerBusy,
+    pub analyze_busy: AnalyzeBusy,
+    pub transcribe_busy: TranscribeBusy,
 }
 
 impl FromRef<AppState> for Arc<Config> {
@@ -60,9 +85,15 @@ impl FromRef<AppState> for SessionStore {
     }
 }
 
-impl FromRef<AppState> for WorkerBusy {
+impl FromRef<AppState> for AnalyzeBusy {
     fn from_ref(state: &AppState) -> Self {
-        state.worker_busy.clone()
+        state.analyze_busy.clone()
+    }
+}
+
+impl FromRef<AppState> for TranscribeBusy {
+    fn from_ref(state: &AppState) -> Self {
+        state.transcribe_busy.clone()
     }
 }
 
@@ -77,7 +108,8 @@ pub fn create_router(config: Arc<Config>) -> Router {
         transcribe_tx,
         analyze_tx,
         session_store: web_session::new_store(),
-        worker_busy: Arc::new(AtomicBool::new(false)),
+        analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))),
+        transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))),
     })
 }
 

server/src/main.rs

@@ -66,10 +66,13 @@ async fn main() {
 
     // Transcription pipeline: channel + worker + recovery scan.
     let (transcribe_tx, transcribe_rx) = transcribe::channel();
+    let transcribe_busy: doctate_server::WorkerBusy =
+        std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
     tokio::spawn(transcribe::worker::run(
         transcribe_rx,
         config.clone(),
         http_client.clone(),
+        transcribe_busy.clone(),
     ));
     {
         let tx = transcribe_tx.clone();
@@ -81,13 +84,13 @@ async fn main() {
 
     // Analyze pipeline: channel + worker + recovery scan.
     let (analyze_tx, analyze_rx) = analyze::channel();
-    let worker_busy: doctate_server::WorkerBusy =
+    let analyze_busy: doctate_server::WorkerBusy =
         std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
     tokio::spawn(analyze::worker::run(
         analyze_rx,
         config.clone(),
         http_client.clone(),
-        worker_busy.clone(),
+        analyze_busy.clone(),
     ));
     {
         let tx = analyze_tx.clone();
@@ -102,7 +105,8 @@ async fn main() {
         transcribe_tx,
         analyze_tx,
         session_store: doctate_server::web_session::new_store(),
-        worker_busy,
+        analyze_busy: doctate_server::AnalyzeBusy(analyze_busy),
+        transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy),
     };
 
     let addr = format!("0.0.0.0:{}", config.server_port);

server/src/routes/user_web.rs

@@ -12,7 +12,8 @@ use crate::auth::AuthenticatedWebUser;
 use crate::config::Config;
 use crate::error::AppError;
 use crate::routes::web::{scan_recordings, RecordingView};
-use crate::WorkerBusy;
+use crate::transcribe::{recovery as transcribe_recovery, TranscribeSender};
+use crate::{AnalyzeBusy, TranscribeBusy};
 
 struct UserCaseView {
     case_id: String,
@@ -67,6 +68,10 @@ struct CaseDetailTemplate {
     llm_missing: bool,
     analyzing: bool,
     has_document: bool,
+    /// True iff the transcribe worker is currently running. Gate for the
+    /// per-recording "Transkription läuft…" label — suppresses the lie when
+    /// a transcript is missing but no worker is active.
+    transcribe_busy: bool,
 }
 
 /// Flags derived from filesystem state + config.
@@ -142,30 +147,45 @@ pub(crate) async fn any_document_exists(case_dir: &Path) -> bool {
     false
 }
 
-/// Self-heal: if the analyze worker is idle but orphan inputs (an input file
-/// without its document) exist for this user, re-enqueue them. Page-load is
-/// the trigger; no cron, no periodic task.
+/// Self-heal: if a worker is idle but orphans exist on disk for this user,
+/// re-enqueue them. Page-load is the trigger; no cron, no periodic task.
+/// Runs for both pipelines so a page-load on either view cleans both.
 async fn heal_orphans_if_idle(
     user_root: &Path,
-    worker_busy: &WorkerBusy,
+    slug: &str,
+    analyze_busy: &AnalyzeBusy,
     analyze_tx: &AnalyzeSender,
+    transcribe_busy: &TranscribeBusy,
+    transcribe_tx: &TranscribeSender,
 ) {
-    if worker_busy.load(Ordering::Acquire) {
-        return;
+    if !analyze_busy.0.load(Ordering::Acquire) {
+        analyze_recovery::enqueue_pending_for_user(user_root, analyze_tx).await;
+    }
+    if !transcribe_busy.0.load(Ordering::Acquire) {
+        transcribe_recovery::enqueue_pending_for_user(user_root, slug, transcribe_tx).await;
     }
-    analyze_recovery::enqueue_pending_for_user(user_root, analyze_tx).await;
 }
 
 pub async fn handle_my_cases(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
-    State(worker_busy): State<WorkerBusy>,
+    State(analyze_busy): State<AnalyzeBusy>,
     State(analyze_tx): State<AnalyzeSender>,
+    State(transcribe_busy): State<TranscribeBusy>,
+    State(transcribe_tx): State<TranscribeSender>,
 ) -> Result<Html<String>, AppError> {
     let user_root = config.data_path.join(&user.slug);
-    heal_orphans_if_idle(&user_root, &worker_busy, &analyze_tx).await;
+    heal_orphans_if_idle(
+        &user_root,
+        &user.slug,
+        &analyze_busy,
+        &analyze_tx,
+        &transcribe_busy,
+        &transcribe_tx,
+    )
+    .await;
 
-    let busy = worker_busy.load(Ordering::Acquire);
+    let busy = analyze_busy.0.load(Ordering::Acquire);
     let cases = scan_user_cases(&config, &user.slug, busy).await;
     let undo_count = crate::routes::case_actions::summarize_latest_batch(&user_root)
         .await
@@ -186,8 +206,10 @@ pub async fn handle_my_cases(
 pub async fn handle_case_detail(
     user: AuthenticatedWebUser,
     State(config): State<Arc<Config>>,
-    State(worker_busy): State<WorkerBusy>,
+    State(analyze_busy): State<AnalyzeBusy>,
     State(analyze_tx): State<AnalyzeSender>,
+    State(transcribe_busy): State<TranscribeBusy>,
+    State(transcribe_tx): State<TranscribeSender>,
     AxumPath(case_id): AxumPath<String>,
 ) -> Result<Html<String>, AppError> {
     uuid::Uuid::parse_str(&case_id)
@@ -195,7 +217,15 @@ pub async fn handle_case_detail(
 
     // IDOR guard: case_dir must live under the session's user_slug.
     let user_root = config.data_path.join(&user.slug);
-    heal_orphans_if_idle(&user_root, &worker_busy, &analyze_tx).await;
+    heal_orphans_if_idle(
+        &user_root,
+        &user.slug,
+        &analyze_busy,
+        &analyze_tx,
+        &transcribe_busy,
+        &transcribe_tx,
+    )
+    .await;
 
     let case_dir = match locate_case(&user_root, &case_id).await {
         Some(p) => p,
@@ -213,8 +243,9 @@ pub async fn handle_case_detail(
         .map(|s| s.trim().to_string())
         .filter(|s| !s.is_empty());
 
-    let busy = worker_busy.load(Ordering::Acquire);
-    let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), busy).await;
+    let a_busy = analyze_busy.0.load(Ordering::Acquire);
+    let t_busy = transcribe_busy.0.load(Ordering::Acquire);
+    let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), a_busy).await;
     let case_id_short = case_id.chars().take(8).collect();
 
     CaseDetailTemplate {
@@ -227,6 +258,7 @@ pub async fn handle_case_detail(
         llm_missing: flags.llm_missing,
         analyzing: flags.analyzing,
         has_document: flags.has_document,
+        transcribe_busy: t_busy,
     }
     .render()
     .map(Html)

server/src/transcribe/recovery.rs

@@ -1,85 +1,81 @@
-use std::path::{Path, PathBuf};
+use std::path::Path;
 
 use tracing::{info, warn};
 
 use super::{TranscribeJob, TranscribeSender};
 
-/// Walk `data_path/*/*/*.m4a` and enqueue every recording that does not
-/// yet have a `.transcript.txt` sibling.
-///
-/// Intended to run once at startup (spawn as a task — sends may back-pressure
-/// if many recordings are pending and the queue fills).
+/// Walk `data_path/*/` and enqueue every recording pending transcription for
+/// every user. Intended for startup; the same primitive backs the per-user
+/// self-heal triggered by web handlers.
 pub async fn scan_and_enqueue(data_path: &Path, tx: &TranscribeSender) {
-    let pending = collect_pending(data_path).await;
-    let total = pending.len();
-    info!(count = total, "Recovery scan found pending recordings");
+    let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
+        return;
+    };
+    // Collect + sort by slug so enqueue order is deterministic across
+    // users (within a user, the primitive sorts by timestamp).
+    let mut user_dirs: Vec<(String, std::path::PathBuf)> = Vec::new();
+    while let Ok(Some(user_entry)) = users.next_entry().await {
+        if let Some(slug) = user_entry.file_name().to_str().map(str::to_owned) {
+            user_dirs.push((slug, user_entry.path()));
+        }
+    }
+    user_dirs.sort_by(|a, b| a.0.cmp(&b.0));
 
-    for (user_slug, audio_path) in pending {
+    let mut total = 0;
+    for (slug, path) in user_dirs {
+        total += enqueue_pending_for_user(&path, &slug, tx).await;
+    }
+    info!(count = total, "Transcribe recovery scan finished");
+}
+
+/// Scan a single `<user_root>/<case>/*.m4a` set and enqueue every recording
+/// whose `.transcript.txt` is missing. Returns the number sent. Channel send
+/// is awaited (bounded channel); only fails if the worker is gone.
+pub async fn enqueue_pending_for_user(
+    user_root: &Path,
+    slug: &str,
+    tx: &TranscribeSender,
+) -> usize {
+    let Ok(mut cases) = tokio::fs::read_dir(user_root).await else {
+        return 0;
+    };
+    let mut pending: Vec<std::path::PathBuf> = Vec::new();
+    while let Ok(Some(case_entry)) = cases.next_entry().await {
+        let case_dir = case_entry.path();
+        if !case_dir.is_dir() || crate::paths::is_deleted(&case_dir).await {
+            continue;
+        }
+        let Ok(mut files) = tokio::fs::read_dir(&case_dir).await else {
+            continue;
+        };
+        while let Ok(Some(file)) = files.next_entry().await {
+            let path = file.path();
+            if path.extension().and_then(|s| s.to_str()) != Some("m4a") {
+                continue;
+            }
+            if path.with_extension("transcript.txt").exists() {
+                continue;
+            }
+            pending.push(path);
+        }
+    }
+    // Oldest first — filenames are ISO timestamps so path order = chronological.
+    pending.sort();
+
+    let mut sent = 0;
+    for path in pending {
         if tx
             .send(TranscribeJob {
-                audio_path: audio_path.clone(),
-                user_slug: user_slug.clone(),
+                audio_path: path.clone(),
+                user_slug: slug.to_owned(),
             })
             .await
             .is_err()
         {
-            warn!(file = %audio_path.display(), "Recovery send failed (worker gone)");
-            break;
+            warn!(file = %path.display(), "recovery send failed (worker gone)");
+            return sent;
         }
+        sent += 1;
     }
-}
-
-/// Walks `$data_path/<slug>/<case>/*.m4a` and returns `(slug, audio_path)`
-/// tuples for every recording without a `.transcript.txt` sibling. The slug
-/// comes from the top-level user directory name — that's the only ownership
-/// signal after a restart (auth context is gone).
-async fn collect_pending(data_path: &Path) -> Vec<(String, PathBuf)> {
-    let mut out: Vec<(String, PathBuf)> = Vec::new();
-
-    let mut users = match tokio::fs::read_dir(data_path).await {
-        Ok(r) => r,
-        Err(_) => return out,
-    };
-
-    while let Ok(Some(user_entry)) = users.next_entry().await {
-        let Some(slug) = user_entry.file_name().to_str().map(str::to_owned) else {
-            continue;
-        };
-        let mut cases = match tokio::fs::read_dir(user_entry.path()).await {
-            Ok(r) => r,
-            Err(_) => continue,
-        };
-
-        while let Ok(Some(case_entry)) = cases.next_entry().await {
-            let case_dir = case_entry.path();
-            if !case_dir.is_dir() {
-                continue;
-            }
-            if crate::paths::is_deleted(&case_dir).await {
-                continue;
-            }
-
-            let mut files = match tokio::fs::read_dir(&case_dir).await {
-                Ok(r) => r,
-                Err(_) => continue,
-            };
-
-            while let Ok(Some(file)) = files.next_entry().await {
-                let path = file.path();
-                if path.extension().and_then(|s| s.to_str()) != Some("m4a") {
-                    continue;
-                }
-                let transcript = path.with_extension("transcript.txt");
-                if !transcript.exists() {
-                    out.push((slug.clone(), path));
-                }
-            }
-        }
-    }
-
-    // Oldest first (filenames are ISO timestamps). Path ordering puts same-user
-    // entries together, which is fine — we still process strictly by timestamp
-    // within a user directory.
-    out.sort_by(|a, b| a.1.cmp(&b.1));
-    out
+    sent
 }

server/src/transcribe/worker.rs

@@ -6,16 +6,23 @@ use tracing::{error, info, warn};
 
 use super::{ffmpeg, ollama, whisper, TranscribeReceiver};
 use crate::config::{Config, WhisperUserSettings};
+use crate::{BusyGuard, WorkerBusy};
 
 const ONELINER_TIMEOUT: Duration = Duration::from_secs(60);
 
 /// Consume transcription jobs sequentially. Whisper holds the GPU, so parallelism
 /// here would only cause contention. One job in flight at a time.
-pub async fn run(mut rx: TranscribeReceiver, config: Arc<Config>, client: reqwest::Client) {
+pub async fn run(
+    mut rx: TranscribeReceiver,
+    config: Arc<Config>,
+    client: reqwest::Client,
+    worker_busy: WorkerBusy,
+) {
     info!("Transcription worker started");
     let timeout = Duration::from_secs(config.whisper_timeout_seconds);
 
     while let Some(job) = rx.recv().await {
+        let _guard = BusyGuard::new(worker_busy.clone());
         let audio_path = job.audio_path;
         let transcript_path = audio_path.with_extension("transcript.txt");
 

server/templates/case_detail.html

@@ -69,7 +69,11 @@ header form { margin: 0; }
 {% when Some with (t) %}
 <div class="transcript">{{ t }}</div>
 {% when None %}
+{% if transcribe_busy %}
 <div class="pending">Transkription läuft…</div>
+{% else %}
+<div class="pending">Transkription ausstehend.</div>
+{% endif %}
 {% endmatch %}
 {% endif %}
 </div>

server/tests/transcribe_test.rs

@@ -286,7 +286,8 @@ async fn worker_renames_audio_to_failed_on_whisper_error() {
     drop(tx); // close channel so the worker loop exits after processing the job
 
     let client = reqwest::Client::new();
-    transcribe::worker::run(rx, config, client).await;
+    let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    transcribe::worker::run(rx, config, client, busy).await;
 
     // Audio must have been renamed so recovery skips it next time.
     assert!(!audio.exists(), "original .m4a still present");