Refactor analyze recovery and worker busy status

The analyze recovery scan has been refactored to iterate through user
directories and enqueue pending analysis jobs more efficiently. The
`WorkerBusy` status has been introduced as an `Arc<AtomicBool>` to track
whether the analyze worker is currently processing a job.

This `WorkerBusy` flag is used in the `analyze::worker::run` function
and managed by a `BusyGuard` RAII struct, ensuring the flag is correctly
set and unset even in case of panics.

The `handle_my_cases` and `handle_case_detail` handlers in `user_web.rs`
now utilize the `WorkerBusy` flag to accurately display the "analyzing"
status and to trigger a self-heal of orphaned analysis inputs when the
worker is idle.

Additionally, the `WorkerBusy` type is now exported from
`server/src/lib.rs` to be accessible by other modules. The test cases
have been updated to include the `WorkerBusy` parameter when spawning
the analyze worker.

server/src/analyze/recovery.rs

@@ -1,78 +1,65 @@
-use std::path::{Path, PathBuf};
+use std::path::Path;
 
 use tracing::{info, warn};
 
 use super::{AnalyzeJob, AnalyzeSender};
 
-/// Walk `data_path/*/*/analysis_input_v*.json` and enqueue every input
-/// that does not yet have a matching `document_v*.md` sibling.
-///
-/// Intended to run once at startup. Send is synchronous on the unbounded
-/// channel — only fails if the worker is gone, which is a programming error.
+/// Walk `data_path/*/` and enqueue every pending analysis for every user.
+/// Intended to run once at startup; the same primitive backs the per-user
+/// self-heal triggered by web handlers.
 pub async fn scan_and_enqueue(data_path: &Path, tx: &AnalyzeSender) {
-    let pending = collect_pending(data_path).await;
-    let total = pending.len();
-    info!(count = total, "Analyze recovery scan found pending inputs");
-
-    for (case_dir, version) in pending {
-        if tx.send(AnalyzeJob { case_dir: case_dir.clone(), version }).is_err() {
-            warn!(case = %case_dir.display(), "Recovery send failed (worker gone)");
-            break;
-        }
+    let Ok(mut users) = tokio::fs::read_dir(data_path).await else {
+        return;
+    };
+    let mut total = 0;
+    while let Ok(Some(user_entry)) = users.next_entry().await {
+        total += enqueue_pending_for_user(&user_entry.path(), tx).await;
     }
+    info!(count = total, "Analyze recovery scan finished");
 }
 
-/// Returns `(case_dir, version)` tuples for every `analysis_input_v{N}.json`
-/// whose `document_v{N}.md` is missing.
-async fn collect_pending(data_path: &Path) -> Vec<(PathBuf, u32)> {
-    let mut out: Vec<(PathBuf, u32)> = Vec::new();
-
-    let mut users = match tokio::fs::read_dir(data_path).await {
-        Ok(r) => r,
-        Err(_) => return out,
+/// Scan a single `<user_root>/<case>/analysis_input_v*.json` set and enqueue
+/// every input whose `document_v{N}.md` is missing. Returns the number sent.
+/// Channel send is synchronous on the unbounded channel — only fails if the
+/// worker is gone, which is a programming error.
+pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> usize {
+    let Ok(mut cases) = tokio::fs::read_dir(user_root).await else {
+        return 0;
     };
-
-    while let Ok(Some(user_entry)) = users.next_entry().await {
-        let mut cases = match tokio::fs::read_dir(user_entry.path()).await {
-            Ok(r) => r,
-            Err(_) => continue,
+    let mut sent = 0;
+    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(case_entry)) = cases.next_entry().await {
-            let case_dir = case_entry.path();
-            if !case_dir.is_dir() {
+        while let Ok(Some(file)) = files.next_entry().await {
+            let path = file.path();
+            let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
                 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();
-                let Some(name) = path.file_name().and_then(|s| s.to_str()) else {
-                    continue;
-                };
-                let Some(version) = parse_input_version(name) else {
-                    continue;
-                };
-                let document = case_dir.join(format!("document_v{version}.md"));
-                if !document.exists() {
-                    out.push((case_dir.clone(), version));
-                }
+            let Some(version) = parse_input_version(name) else {
+                continue;
+            };
+            if case_dir.join(format!("document_v{version}.md")).exists() {
+                continue;
             }
+            if tx
+                .send(AnalyzeJob {
+                    case_dir: case_dir.clone(),
+                    version,
+                })
+                .is_err()
+            {
+                warn!(case = %case_dir.display(), "recovery send failed (worker gone)");
+                return sent;
+            }
+            sent += 1;
         }
     }
-
-    // Oldest case directories first — not strictly required for correctness,
-    // but gives deterministic recovery order. Cases are UUIDs so sorting is
-    // purely lexicographic, which is fine.
-    out.sort();
-    out
+    sent
 }
 
 /// Parse `analysis_input_v{N}.json` and return `N` (as u32) if the pattern

server/src/analyze/worker.rs

@@ -1,4 +1,5 @@
 use std::path::Path;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -7,15 +8,40 @@ 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);
+    }
+}
 
 /// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
 /// but sequential processing keeps the architecture simple and shields the
 /// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`.
-pub async fn run(mut rx: AnalyzeReceiver, config: Arc<Config>, client: reqwest::Client) {
+pub async fn run(
+    mut rx: AnalyzeReceiver,
+    config: Arc<Config>,
+    client: reqwest::Client,
+    worker_busy: WorkerBusy,
+) {
     info!("Analyze worker started");
     let timeout = Duration::from_secs(config.llm_timeout_seconds);
 
     while let Some(job) = rx.recv().await {
+        let _guard = BusyGuard::new(worker_busy.clone());
         process(&job.case_dir, job.version, &config, &client, timeout).await;
     }
 
@@ -100,6 +126,13 @@ 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");
+    }
+
     info!(
         case = %case_dir.display(),
         version,

server/src/lib.rs

@@ -8,6 +8,7 @@ pub mod routes;
 pub mod transcribe;
 pub mod web_session;
 
+use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
 
 use axum::extract::FromRef;
@@ -18,6 +19,12 @@ 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.
+pub type WorkerBusy = Arc<AtomicBool>;
+
 /// Shared application state. Clonable — all fields are cheap to clone
 /// (`Arc` and `mpsc::Sender`).
 #[derive(Clone)]
@@ -26,6 +33,7 @@ pub struct AppState {
     pub transcribe_tx: TranscribeSender,
     pub analyze_tx: AnalyzeSender,
     pub session_store: SessionStore,
+    pub worker_busy: WorkerBusy,
 }
 
 impl FromRef<AppState> for Arc<Config> {
@@ -52,6 +60,12 @@ impl FromRef<AppState> for SessionStore {
     }
 }
 
+impl FromRef<AppState> for WorkerBusy {
+    fn from_ref(state: &AppState) -> Self {
+        state.worker_busy.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.
@@ -63,6 +77,7 @@ 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)),
     })
 }
 

server/src/main.rs

@@ -81,10 +81,13 @@ async fn main() {
 
     // Analyze pipeline: channel + worker + recovery scan.
     let (analyze_tx, analyze_rx) = analyze::channel();
+    let worker_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(),
     ));
     {
         let tx = analyze_tx.clone();
@@ -99,6 +102,7 @@ async fn main() {
         transcribe_tx,
         analyze_tx,
         session_store: doctate_server::web_session::new_store(),
+        worker_busy,
     };
 
     let addr = format!("0.0.0.0:{}", config.server_port);

server/src/routes/user_web.rs

@@ -1,4 +1,5 @@
 use std::path::Path;
+use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use askama::Template;
@@ -6,15 +7,19 @@ use axum::extract::{Path as AxumPath, State};
 use axum::response::Html;
 use tracing::{info, warn};
 
+use crate::analyze::{recovery as analyze_recovery, AnalyzeSender};
 use crate::auth::AuthenticatedWebUser;
 use crate::config::Config;
 use crate::error::AppError;
 use crate::routes::web::{scan_recordings, RecordingView};
+use crate::WorkerBusy;
 
 struct UserCaseView {
     case_id: String,
-    case_id_short: String,
     most_recent: String,
+    /// HH:MM extracted from `most_recent` filename (e.g. "10:32"),
+    /// empty string if filename does not match the expected timestamp form.
+    time_hms: String,
     oneliner: Option<String>,
     recordings_count: usize,
     analyzing: bool,
@@ -25,6 +30,19 @@ struct UserCaseView {
     can_analyze: bool,
 }
 
+/// Extract `HH:MM` from a recording filename of the form
+/// `YYYY-MM-DDTHH-MM-SSZ.m4a[.failed]`. Returns `""` on any mismatch.
+fn extract_hhmm(filename: &str) -> String {
+    let (Some(h), Some(m)) = (filename.get(11..13), filename.get(14..16)) else {
+        return String::new();
+    };
+    if h.bytes().all(|b| b.is_ascii_digit()) && m.bytes().all(|b| b.is_ascii_digit()) {
+        format!("{h}:{m}")
+    } else {
+        String::new()
+    }
+}
+
 #[derive(Template)]
 #[template(path = "my_cases.html")]
 struct MyCasesTemplate {
@@ -33,6 +51,8 @@ struct MyCasesTemplate {
     /// Number of cases that "Undo last delete" would restore. 0 hides the
     /// undo button entirely.
     undo_count: usize,
+    /// True iff the session user is an admin — toggles full case_id display.
+    is_admin: bool,
 }
 
 #[derive(Template)]
@@ -65,12 +85,13 @@ async fn compute_flags(
     case_dir: &Path,
     recordings: &[RecordingView],
     llm_configured: bool,
+    worker_busy: bool,
 ) -> CaseFlags {
     let has_document = any_document_exists(case_dir).await;
-    let analyzing = !has_document
-        && tokio::fs::try_exists(case_dir.join("analysis_input_v1.json"))
-            .await
-            .unwrap_or(false);
+    // Honest status: an input file alone does not mean a job is in flight.
+    // The worker may be idle and the file an orphan from a crash. Page-level
+    // self-heal (see `heal_orphans_if_idle`) re-enqueues those.
+    let analyzing = !has_document && worker_busy && any_analysis_input_exists(case_dir).await;
 
     let non_failed: Vec<&RecordingView> = recordings.iter().filter(|r| !r.failed).collect();
     let all_transcribed =
@@ -86,6 +107,24 @@ async fn compute_flags(
     }
 }
 
+/// True iff the case directory contains at least one `analysis_input_v*.json`.
+/// The worker removes the input after successful document write, so existence
+/// implies a pending or in-flight analyze job.
+pub(crate) async fn any_analysis_input_exists(case_dir: &Path) -> bool {
+    let mut entries = match tokio::fs::read_dir(case_dir).await {
+        Ok(r) => r,
+        Err(_) => return false,
+    };
+    while let Ok(Some(entry)) = entries.next_entry().await {
+        let name = entry.file_name();
+        let Some(name_str) = name.to_str() else { continue };
+        if name_str.starts_with("analysis_input_v") && name_str.ends_with(".json") {
+            return true;
+        }
+    }
+    false
+}
+
 /// True iff the case directory contains at least one `document_v{N}.md`.
 /// We don't care about N here — any version means "Ausgewertet" for the UI.
 pub(crate) async fn any_document_exists(case_dir: &Path) -> bool {
@@ -103,20 +142,41 @@ 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.
+async fn heal_orphans_if_idle(
+    user_root: &Path,
+    worker_busy: &WorkerBusy,
+    analyze_tx: &AnalyzeSender,
+) {
+    if worker_busy.load(Ordering::Acquire) {
+        return;
+    }
+    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_tx): State<AnalyzeSender>,
 ) -> Result<Html<String>, AppError> {
-    let cases = scan_user_cases(&config, &user.slug).await;
     let user_root = config.data_path.join(&user.slug);
+    heal_orphans_if_idle(&user_root, &worker_busy, &analyze_tx).await;
+
+    let busy = worker_busy.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
         .map(|(_, n)| n)
         .unwrap_or(0);
+    let is_admin = user.is_admin();
     MyCasesTemplate {
         slug: user.slug,
         cases,
         undo_count,
+        is_admin,
     }
     .render()
     .map(Html)
@@ -126,6 +186,8 @@ 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_tx): State<AnalyzeSender>,
     AxumPath(case_id): AxumPath<String>,
 ) -> Result<Html<String>, AppError> {
     uuid::Uuid::parse_str(&case_id)
@@ -133,6 +195,8 @@ 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;
+
     let case_dir = match locate_case(&user_root, &case_id).await {
         Some(p) => p,
         None => {
@@ -149,7 +213,8 @@ pub async fn handle_case_detail(
         .map(|s| s.trim().to_string())
         .filter(|s| !s.is_empty());
 
-    let flags = compute_flags(&case_dir, &recordings, config.llm_configured()).await;
+    let busy = worker_busy.load(Ordering::Acquire);
+    let flags = compute_flags(&case_dir, &recordings, config.llm_configured(), busy).await;
     let case_id_short = case_id.chars().take(8).collect();
 
     CaseDetailTemplate {
@@ -186,7 +251,7 @@ pub(crate) async fn locate_case(user_root: &Path, case_id: &str) -> Option<std::
 /// recent recording filename (lexicographic ≈ chronological since
 /// timestamps are ISO-8601), newest first. Soft-deleted cases (with
 /// `.deleted` marker) are excluded.
-async fn scan_user_cases(config: &Config, slug: &str) -> Vec<UserCaseView> {
+async fn scan_user_cases(config: &Config, slug: &str, worker_busy: bool) -> Vec<UserCaseView> {
     let user_root = config.data_path.join(slug);
     let llm_configured = config.llm_configured();
     let mut cases = Vec::new();
@@ -230,7 +295,6 @@ async fn scan_user_cases(config: &Config, slug: &str) -> Vec<UserCaseView> {
                 .filter(|r| !r.failed)
                 .all(|r| r.transcript.is_some());
 
-        let case_id_short = case_id.chars().take(8).collect();
         let oneliner = tokio::fs::read_to_string(case_path.join("oneliner.txt"))
             .await
             .ok()
@@ -238,16 +302,15 @@ async fn scan_user_cases(config: &Config, slug: &str) -> Vec<UserCaseView> {
             .filter(|s| !s.is_empty());
 
         let has_document = any_document_exists(&case_path).await;
-        let analyzing = !has_document
-            && tokio::fs::try_exists(case_path.join("analysis_input_v1.json"))
-                .await
-                .unwrap_or(false);
+        let analyzing =
+            !has_document && worker_busy && any_analysis_input_exists(&case_path).await;
         let can_analyze = !analyzing && all_transcribed && llm_configured;
 
+        let time_hms = extract_hhmm(&most_recent);
         cases.push(UserCaseView {
             case_id,
-            case_id_short,
             most_recent,
+            time_hms,
             oneliner,
             recordings_count,
             analyzing,

server/templates/my_cases.html

@@ -18,10 +18,17 @@ section h2 { font-size: 1.1em; color: #555; border-bottom: 1px solid #ddd; paddi
 .case-row .check { align-self: start; padding-top: 0.25em; }
 .case-row .body { min-width: 0; }
 .case-row .body a { color: inherit; text-decoration: none; }
-.case-row .body a:hover h3 { text-decoration: underline; }
+.case-row .body a:hover .line1 { text-decoration: underline; }
 .case-row .body h3 { margin: 0 0 0.2em 0; font-size: 1em; font-weight: 600; }
 .case-row .body small { color: #666; font-family: monospace; font-size: 0.85em; }
 .case-row .oneliner { color: #333; margin: 0.2em 0 0; }
+.case-row .line1 { font-size: 1em; margin: 0; }
+.case-row .line1 .time { font-family: monospace; font-weight: bold; color: #444; }
+.case-row .line2 { margin: 0.2em 0 0; color: #555; font-size: 0.95em; display: flex; align-items: center; gap: 0.5em; }
+.case-row .uuid { color: #999; font-family: monospace; font-size: 0.7em; margin-top: 0.2em; }
+.status-badge { display: inline-block; padding: 0.05em 0.5em; border-radius: 999px; font-size: 0.75em; font-weight: bold; color: white; }
+.status-badge.open { background: #e24a4a; }
+.status-badge.done { background: #7ed321; }
 .case-row .actions { display: flex; gap: 0.4em; }
 .case-row .actions button { font-size: 0.85em; padding: 0.35em 0.8em; }
 
@@ -60,13 +67,9 @@ section h2 { font-size: 1.1em; color: #555; border-bottom: 1px solid #ddd; paddi
 <div class="check"><input type="checkbox" name="case_id" value="{{ case.case_id }}"></div>
 <div class="body">
 {% if case.has_document %}<a href="/web/cases/{{ case.case_id }}/document">{% else %}<a href="/web/cases/{{ case.case_id }}">{% endif %}
-<h3>{{ case.case_id_short }} — {{ case.recordings_count }} Aufnahme(n){% if case.has_document %}<span class="label done-doc">ausgewertet</span>{% else if case.analyzing %}<span class="label analyzing">wird analysiert</span>{% endif %}</h3>
-{% match case.oneliner %}
-{% when Some with (t) %}
-<div class="oneliner">{{ t }}</div>
-{% when None %}
-{% endmatch %}
-<small>{{ case.most_recent }}</small>
+<div class="line1"><span class="time">{{ case.time_hms }}</span>{% match case.oneliner %}{% when Some with (t) %} — {{ t }}{% when None %}{% endmatch %}</div>
+<div class="line2">{{ case.recordings_count }} Aufnahmen{% if case.has_document %} — <span class="status-badge done">ausgewertet</span>{% else %} — <span class="status-badge open">offen</span>{% endif %}{% if case.analyzing %} <span class="label analyzing">wird analysiert</span>{% endif %}</div>
+{% if is_admin %}<div class="uuid">{{ case.case_id }}</div>{% endif %}
 </a>
 </div>
 <div class="actions">

server/tests/analyze_test.rs

@@ -327,7 +327,8 @@ async fn analyze_worker_writes_document_via_wiremock() {
     let config = config_with_llm(tmp.clone(), mock.uri());
     let (tx, rx) = analyze::channel();
     let client = reqwest::Client::new();
-    let handle = tokio::spawn(analyze::worker::run(rx, config, client));
+    let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let handle = tokio::spawn(analyze::worker::run(rx, config, client, busy));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),
@@ -480,7 +481,8 @@ async fn analyze_v2_worker_writes_document_v2_via_wiremock() {
 
     let config = config_with_llm(tmp.clone(), mock.uri());
     let (tx, rx) = analyze::channel();
-    let handle = tokio::spawn(analyze::worker::run(rx, config, reqwest::Client::new()));
+    let busy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let handle = tokio::spawn(analyze::worker::run(rx, config, reqwest::Client::new(), busy));
 
     tx.send(analyze::AnalyzeJob {
         case_dir: case_dir.clone(),