diff --git a/server/src/analyze/auto_trigger.rs b/server/src/analyze/auto_trigger.rs index f11e6be..539507a 100644 --- a/server/src/analyze/auto_trigger.rs +++ b/server/src/analyze/auto_trigger.rs @@ -112,56 +112,146 @@ impl FailureDetails { /// Outcome of the pre-condition check. `Skip` carries a static string so /// the common "nothing to do" path allocates nothing. +/// +/// Adapter over [`CaseAnalysisState`]: the auto-trigger reduces the rich +/// state machine to a binary "fire or skip" decision. UI layers that need +/// the full state should call [`evaluate_state`] directly. #[derive(Debug, Clone, PartialEq, Eq)] pub enum AutoDecision { Enqueue, Skip(&'static str), } -/// Pure pre-condition check: decide whether this case is due for an -/// automatic analysis. Does file I/O but no channel sends and no -/// state mutations — safe to call redundantly. -pub async fn evaluate_case(case_dir: &Path) -> AutoDecision { +/// Per-case lifecycle state, computed from the artefacts on disk plus the +/// idle threshold. The variants below are precedence-ordered: the first +/// matching condition wins, so a `Failed` case with a stale failure +/// marker is reported as `Required`, not `Failed`. +/// +/// `has_document: bool` on `Required`/`Idle` lets the UI decide whether +/// to render the previously-generated document (= „letztes Dokument") +/// alongside the „Analyse erforderlich"-banner. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum CaseAnalysisState { + /// No `.m4a` files in the directory at all (e.g. fresh case before + /// the first upload completes). + NoRecordings, + /// At least one recording is still missing its transcript sidecar. + Pending, + /// `analysis_input.json` exists — a job is queued or in flight. + Queued, + /// Failure marker matches the current `latest_mtime`. Stays `Failed` + /// until either the input changes (→ `Required`) or the user wipes + /// the marker via the manual retry path. + Failed, + /// Document does not yet cover the newest recording, but the + /// idle-threshold has not elapsed yet — the user may still be + /// dictating. Auto-trigger waits. + Required { has_document: bool }, + /// Like `Required`, plus the idle window has elapsed. The auto- + /// trigger fires. + Idle { has_document: bool }, + /// Document on disk covers exactly the current recording set + /// (`covered_mtime == latest_mtime`). + Current, +} + +/// Compute the full per-case lifecycle state. Pure file-I/O, no +/// mutations or channel sends — safe to call redundantly per render. +/// +/// `idle_threshold` is the time without new recordings after which a +/// stale-document case is auto-promoted from `Required` to `Idle`. Pass +/// `Duration::ZERO` to disable the wait (every Required becomes Idle +/// immediately — the historical eager behaviour, used by the +/// [`evaluate_case`] adapter and most existing tests). +/// +/// `now` is injected so tests can pin filesystem mtimes against a known +/// reference point. Production callers pass `SystemTime::now()`. +pub async fn evaluate_state( + case_dir: &Path, + idle_threshold: std::time::Duration, + now: SystemTime, +) -> CaseAnalysisState { let scan = scan_m4as(case_dir).await; if !scan.saw_any { - return AutoDecision::Skip("no recordings"); + return CaseAnalysisState::NoRecordings; } if !scan.all_transcribed { - return AutoDecision::Skip("not all transcribed"); + return CaseAnalysisState::Pending; } let Some(latest_mtime) = scan.latest_mtime else { - return AutoDecision::Skip("no recordings"); + return CaseAnalysisState::NoRecordings; }; // Any existing analysis_input.json means a job is queued or running. - // Worker removes the file on success — next reload re-evaluates. if tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE)) .await .unwrap_or(false) { - return AutoDecision::Skip("job queued"); + return CaseAnalysisState::Queued; } - // A failure marker matching the current input signature means we - // already tried and the input has not changed since. + let latest_rfc = rfc3339_of(latest_mtime); + + // Failure marker matching the current input signature → previous + // attempt for *this* recording set failed, do not retry until the + // input changes (or the user clears the marker manually). if let Some(marker) = read_failure_marker(case_dir).await - && marker.last_recording_mtime == rfc3339_of(latest_mtime) + && marker.last_recording_mtime == latest_rfc { - return AutoDecision::Skip("previously failed"); + return CaseAnalysisState::Failed; } - // Fresh document older than the newest recording → stale, re-analyse. - // Fresh document newer or equal → nothing to do. - let doc_path = case_dir.join(DOCUMENT_FILE); - if let Ok(meta) = tokio::fs::metadata(&doc_path).await - && let Ok(doc_mtime) = meta.modified() - && doc_mtime >= latest_mtime + let doc_meta = read_document_meta(case_dir).await; + let has_document = doc_meta.is_some(); + + if let Some(doc) = doc_meta + && doc.covered_mtime == latest_rfc { - return AutoDecision::Skip("analysis current"); + return CaseAnalysisState::Current; } - AutoDecision::Enqueue + // Stale document (or no document at all): differentiate between + // "user might still be dictating" (Required) and "long enough quiet, + // fire" (Idle). + let elapsed = now + .duration_since(latest_mtime) + .unwrap_or(std::time::Duration::ZERO); + if elapsed >= idle_threshold { + CaseAnalysisState::Idle { has_document } + } else { + CaseAnalysisState::Required { has_document } + } +} + +/// Read and parse `document.json` if present. Returns `None` for +/// missing files or parse failures (a corrupt envelope makes the case +/// appear as „has no document", which downgrades it to `Required` and +/// triggers a fresh analysis on the next idle window). +pub(crate) async fn read_document_meta(case_dir: &Path) -> Option { + let bytes = tokio::fs::read(case_dir.join(DOCUMENT_FILE)).await.ok()?; + serde_json::from_slice(&bytes).ok() +} + +/// Backwards-compatible binary decision: kept for the existing +/// `try_enqueue` callsites and unit tests that predate the typed state +/// machine. New callers should consume [`evaluate_state`] directly. +/// +/// Reduction: `Idle` → `Enqueue`, everything else → `Skip()`. +/// `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. +pub async fn evaluate_case(case_dir: &Path) -> AutoDecision { + let state = evaluate_state(case_dir, std::time::Duration::ZERO, SystemTime::now()).await; + match state { + CaseAnalysisState::NoRecordings => AutoDecision::Skip("no recordings"), + CaseAnalysisState::Pending => AutoDecision::Skip("not all transcribed"), + CaseAnalysisState::Queued => AutoDecision::Skip("job queued"), + CaseAnalysisState::Failed => AutoDecision::Skip("previously failed"), + CaseAnalysisState::Current => AutoDecision::Skip("analysis current"), + CaseAnalysisState::Idle { .. } => AutoDecision::Enqueue, + CaseAnalysisState::Required { .. } => AutoDecision::Skip("waiting for idle"), + } } /// Evaluate and, on `Enqueue`, prepare and send the job. Returns `true` @@ -187,15 +277,10 @@ pub async fn try_enqueue(case_dir: &Path, tx: &AnalyzeSender, events_tx: &EventS } }; - // Re-analyse path: delete stale document.md up front so the worker's - // "document exists → skip" guard does not short-circuit, and the UI - // stops showing the outdated text during the render race. - let doc_path = case_dir.join(DOCUMENT_FILE); - if let Err(e) = remove_if_exists(&doc_path).await { - warn!(case = %case_dir.display(), error = %e, "auto-analyze: remove old doc failed"); - return false; - } - + // The previous document.json stays in place: while the worker runs, + // 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"); @@ -394,14 +479,6 @@ pub(crate) async fn read_failure_marker(case_dir: &Path) -> Option std::io::Result<()> { - match tokio::fs::remove_file(path).await { - Ok(_) => Ok(()), - Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()), - Err(e) => Err(e), - } -} - /// 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 @@ -420,13 +497,14 @@ async fn write_input_overwrite(path: &Path, input: &AnalysisInput) -> std::io::R /// 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. -fn rfc3339_of(t: SystemTime) -> String { +pub(crate) fn rfc3339_of(t: SystemTime) -> String { OffsetDateTime::from(t).format(&Rfc3339).unwrap_or_default() } #[cfg(test)] mod tests { use super::*; + use crate::analyze::Document; use filetime::{FileTime, set_file_mtime}; use std::time::Duration; use tempfile::TempDir; @@ -436,6 +514,25 @@ mod tests { fs::write(path, b"").await.unwrap(); } + /// Seed a `document.json` whose `covered_mtime` matches the supplied + /// `SystemTime`. Use `set_mtime` on `.m4a` files separately to drive + /// the latest-mtime → covered-mtime comparison in [`evaluate_state`]. + async fn write_document_covering(case_dir: &std::path::Path, covered_mtime: SystemTime) { + let covered_rfc = rfc3339_of(covered_mtime); + let doc = Document { + covered_mtime: covered_rfc.clone(), + written_at: covered_rfc, + backend_id: "test".into(), + content_md: "doc".into(), + }; + fs::write( + case_dir.join(DOCUMENT_FILE), + serde_json::to_vec(&doc).unwrap(), + ) + .await + .unwrap(); + } + /// Seed a `.json` next to a (presumed already-touched) /// `.m4a`. Variant is `Transcript::Silent` — these tests /// only care that the recording is in *some* terminal state @@ -495,17 +592,15 @@ mod tests { } #[tokio::test] - async fn document_newer_than_recordings_skips() { + async fn document_covered_equal_latest_returns_current() { let dir = TempDir::new().unwrap(); let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a"); touch(&m4a).await; seed_meta(dir.path(), "2026-01-01T10-00-00Z").await; - let doc = dir.path().join(DOCUMENT_FILE); - touch(&doc).await; let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000); set_mtime(&m4a, base); - set_mtime(&doc, base + Duration::from_secs(60)); + write_document_covering(dir.path(), base).await; assert_eq!( evaluate_case(dir.path()).await, @@ -514,32 +609,28 @@ mod tests { } #[tokio::test] - async fn document_older_than_recordings_enqueues() { + async fn document_covered_older_than_recordings_enqueues() { let dir = TempDir::new().unwrap(); let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a"); touch(&m4a).await; seed_meta(dir.path(), "2026-01-01T10-00-00Z").await; - let doc = dir.path().join(DOCUMENT_FILE); - touch(&doc).await; let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000); - set_mtime(&doc, base); set_mtime(&m4a, base + Duration::from_secs(60)); + // Document covers an earlier snapshot — a recording arrived since. + write_document_covering(dir.path(), base).await; assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue); } /// Burst race: when a recording arrives DURING the LLM call for an - /// earlier recording, the worker writes a `document.md` whose mtime - /// is later than the new recording's mtime — even though the - /// document content cannot include that recording (its - /// `analysis_input.json` snapshot was frozen before the recording - /// existed). The current `doc_mtime >= latest_mtime` check would - /// then incorrectly report "analysis current" and the new recording - /// would never make it into a document. - /// - /// This test pins the desired behaviour: `evaluate_case` must return - /// `Enqueue` so the late recording is folded into a fresh document. + /// earlier recording, the worker writes a `document.json` whose + /// `covered_mtime` snapshot only includes the recordings that were + /// on disk when the analyze job started. A later-arriving recording + /// (B) is reflected by a strictly larger `latest_mtime` on disk than + /// the document's `covered_mtime`, so [`evaluate_state`] correctly + /// returns `Required` (and `evaluate_case` enqueues, given a zero + /// idle threshold). #[tokio::test] async fn recording_arrived_during_llm_call_re_enqueues() { let dir = TempDir::new().unwrap(); @@ -558,13 +649,11 @@ mod tests { seed_meta_with(dir.path(), "2026-01-01T10-00-05Z", "bravo").await; set_mtime(&b_m4a, base + Duration::from_secs(5)); - // document.md: written when the LLM responded for A's snapshot. - // Its mtime is newer than both recordings — but its content only - // covers A. analysis_input.json has already been removed by the - // worker on success, and there is no failure marker. - let doc = dir.path().join(DOCUMENT_FILE); - touch(&doc).await; - set_mtime(&doc, base + Duration::from_secs(30)); + // document.json: covers only A's snapshot. Wallclock-mtime of + // the file itself would be misleading (later than B), but the + // typed `covered_mtime` field pins the truth — the contained + // markdown describes A only. + write_document_covering(dir.path(), base).await; assert_eq!( evaluate_case(dir.path()).await, @@ -675,18 +764,16 @@ mod tests { ) .await; - // Not eligible: has a fresh document. + // Not eligible: document already covers the latest recording. let current_id = "22222222-2222-2222-2222-222222222222"; let current = user_root.path().join(current_id); fs::create_dir_all(¤t).await.unwrap(); let m4a = current.join("2026-01-01T10-00-00Z.m4a"); fs::write(&m4a, b"fake").await.unwrap(); seed_meta_with(¤t, "2026-01-01T10-00-00Z", "stable").await; - let doc = current.join(DOCUMENT_FILE); - fs::write(&doc, b"doc").await.unwrap(); let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000); set_mtime(&m4a, base); - set_mtime(&doc, base + Duration::from_secs(60)); + write_document_covering(¤t, base).await; // Ignored: non-UUID directory name. let junk = user_root.path().join("not-a-uuid"); diff --git a/server/src/analyze/recovery.rs b/server/src/analyze/recovery.rs index 14c0a65..5e97d32 100644 --- a/server/src/analyze/recovery.rs +++ b/server/src/analyze/recovery.rs @@ -1,97 +1,159 @@ +//! Boot-time state observation for the analyze pipeline. +//! +//! Pre-refactor this module re-enqueued every case that looked pending +//! at startup. With the typed [`CaseAnalysisState`] machine the per-page +//! `try_enqueue_all_for_user` covers that responsibility lazily — boot +//! is now reduced to a telemetry pass that logs how many cases sit in +//! each observable state, without any worker channel sends. +//! +//! [`CaseAnalysisState`]: super::auto_trigger::CaseAnalysisState + use std::path::Path; +use std::time::{Duration, SystemTime}; -use tracing::{info, warn}; +use tracing::info; -use super::auto_trigger::read_failure_marker; -use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE}; +use super::auto_trigger::{CaseAnalysisState, evaluate_state}; -/// 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 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"); +/// Aggregate counts over every UUID-named, non-closed case under a user +/// root. `NoRecordings` and `Pending` are deliberately not represented: +/// they are transient pre-states that need no operator visibility. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct BootStateSummary { + /// `analysis_input.json` present → a job is queued or in flight. + pub queued: usize, + /// Failure marker matches `latest_mtime` → manual retry needed. + pub failed: usize, + /// Idle threshold elapsed → next page render will auto-enqueue. + pub idle: usize, + /// Document not covering latest, idle threshold not yet reached. + pub required: usize, + /// Document covers `latest_mtime` → nothing to do. + pub current: usize, } -/// Scan a user's case dirs and enqueue any case that has an -/// `analysis_input.json` but no `document.md`. Returns the number sent. -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; +impl BootStateSummary { + fn add(&mut self, other: BootStateSummary) { + self.queued += other.queued; + self.failed += other.failed; + self.idle += other.idle; + self.required += other.required; + self.current += other.current; + } + + fn record(&mut self, state: &CaseAnalysisState) { + match state { + CaseAnalysisState::Queued => self.queued += 1, + CaseAnalysisState::Failed => self.failed += 1, + CaseAnalysisState::Idle { .. } => self.idle += 1, + CaseAnalysisState::Required { .. } => self.required += 1, + CaseAnalysisState::Current => self.current += 1, + CaseAnalysisState::NoRecordings | CaseAnalysisState::Pending => {} + } + } +} + +/// Walk `data_path/*/` and emit a single info-level summary line. The +/// returned summary is mainly for tests; production callers ignore it. +pub async fn boot_scan_state( + data_path: &Path, + idle_threshold: Duration, + now: SystemTime, +) -> 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 = compute_state_for_user(&user_entry.path(), idle_threshold, now).await; + total.add(per_user); + } + info!( + queued = total.queued, + failed = total.failed, + idle = total.idle, + required = total.required, + current = total.current, + "analyze boot state scan finished" + ); + total +} + +/// Aggregate the per-case state for one user root. Skips closed cases +/// and non-UUID directory names. Pure observation — no side effects. +pub async fn compute_state_for_user( + user_root: &Path, + idle_threshold: Duration, + now: SystemTime, +) -> BootStateSummary { + let mut summary = BootStateSummary::default(); + let Ok(mut cases) = tokio::fs::read_dir(user_root).await else { + return summary; }; - 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_closed(&case_dir).await { continue; } - if !case_dir.join(ANALYSIS_INPUT_FILE).exists() { + let Ok(name) = case_entry.file_name().into_string() else { + continue; + }; + if uuid::Uuid::parse_str(&name).is_err() { continue; } - if case_dir.join(DOCUMENT_FILE).exists() { - continue; - } - // A failure marker matching the input's `last_recording_mtime` - // means the worker has already tried this exact input and failed - // (e.g. LLM timeout). Re-enqueueing here would block the worker - // for another full timeout window and starve fresh user clicks. - // `auto_trigger::evaluate_case` performs the same skip; recovery - // must mirror it. - if input_already_failed(&case_dir).await { - continue; - } - if tx - .send(AnalyzeJob { - case_dir: case_dir.clone(), - }) - .is_err() - { - warn!(case = %case_dir.display(), "recovery send failed (worker gone)"); - return sent; - } - sent += 1; + let state = evaluate_state(&case_dir, idle_threshold, now).await; + summary.record(&state); } - sent -} - -/// Returns `true` iff a failure marker exists whose -/// `last_recording_mtime` equals the one in `analysis_input.json`. Any -/// I/O or parse error is treated as "no usable marker" — recovery then -/// proceeds with the enqueue, matching the optimistic intent of the -/// outer scan. -async fn input_already_failed(case_dir: &Path) -> bool { - let Some(marker) = read_failure_marker(case_dir).await else { - return false; - }; - let Ok(bytes) = tokio::fs::read(case_dir.join(ANALYSIS_INPUT_FILE)).await else { - return false; - }; - let Ok(input) = serde_json::from_slice::(&bytes) else { - return false; - }; - marker.last_recording_mtime == input.last_recording_mtime + summary } #[cfg(test)] mod tests { use super::*; - use crate::analyze::auto_trigger::{FAILURE_MARKER_FILE, FailureMarker}; - use crate::analyze::{Document, RecordingInput, channel as analyze_channel}; + 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 filetime::{FileTime, set_file_mtime}; use tempfile::TempDir; use tokio::fs; - async fn write_document(case_dir: &Path, content: &str) { + fn case_id(n: u8) -> String { + let h = format!("{n:08x}"); + format!("{h}-1111-1111-1111-111111111111") + } + + fn set_mtime(path: &Path, t: SystemTime) { + set_file_mtime(path, FileTime::from_system_time(t)).unwrap(); + } + + /// Seed a case with a single transcribed recording at `mtime`. + /// Returns the case_dir for further fixture operations. + async fn seed_transcribed_case(user_root: &Path, n: u8, mtime: SystemTime) -> std::path::PathBuf { + let case = user_root.join(case_id(n)); + fs::create_dir_all(&case).await.unwrap(); + let stem = "2026-01-01T10-00-00Z"; + let m4a = case.join(format!("{stem}.m4a")); + fs::write(&m4a, b"fake").await.unwrap(); + crate::paths::write_recording_meta_sync( + &case, + stem, + doctate_common::Transcript::Content { + text: "transcript".into(), + }, + None, + ); + set_mtime(&m4a, mtime); + case + } + + async fn write_document_covering(case_dir: &Path, covered_mtime: SystemTime) { + let rfc = auto_rfc3339(covered_mtime); let doc = Document { - covered_mtime: "1970-01-01T00:00:00Z".into(), - written_at: "1970-01-01T00:00:00Z".into(), + covered_mtime: rfc.clone(), + written_at: rfc, backend_id: "test".into(), - content_md: content.into(), + content_md: "doc".into(), }; fs::write( case_dir.join(DOCUMENT_FILE), @@ -101,29 +163,23 @@ mod tests { .unwrap(); } - fn make_input(mtime: &str) -> AnalysisInput { - AnalysisInput { - last_recording_mtime: mtime.to_owned(), - recordings: vec![RecordingInput { - recorded_at: mtime.to_owned(), - text: "ignored".into(), - }], + 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(), - } - } - - async fn write_input(case_dir: &Path, input: &AnalysisInput) { + }; fs::write( case_dir.join(ANALYSIS_INPUT_FILE), - serde_json::to_vec(input).unwrap(), + serde_json::to_vec(&input).unwrap(), ) .await .unwrap(); } - async fn write_marker(case_dir: &Path, mtime: &str) { + async fn write_marker(case_dir: &Path, mtime: SystemTime) { let marker = FailureMarker { - last_recording_mtime: mtime.to_owned(), + last_recording_mtime: auto_rfc3339(mtime), reason: "test".into(), failed_at: "2026-05-03T00:00:00Z".into(), backend_id: None, @@ -137,93 +193,96 @@ mod tests { .unwrap(); } - /// Drain the receiver after sender drop — same pattern as - /// `auto_trigger::tests::try_enqueue_all_sends_only_eligible_cases`. - async fn drain(mut rx: crate::analyze::AnalyzeReceiver) -> Vec { - let mut out = Vec::new(); - while let Some(job) = rx.recv().await { - out.push(job.case_dir); - } - out + fn base_time() -> SystemTime { + SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000) } #[tokio::test] - async fn enqueues_pending_input_without_doc_or_marker() { + async fn pending_input_counts_as_queued() { let user_root = TempDir::new().unwrap(); - let case_id = "11111111-1111-1111-1111-111111111111"; - let case = user_root.path().join(case_id); - fs::create_dir_all(&case).await.unwrap(); - write_input(&case, &make_input("2026-05-03T10:00:00Z")).await; + let case = seed_transcribed_case(user_root.path(), 1, base_time()).await; + write_input(&case, base_time()).await; - let (tx, rx) = analyze_channel(); - let sent = enqueue_pending_for_user(user_root.path(), &tx).await; - drop(tx); + let summary = + compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; - assert_eq!(sent, 1); - let received = drain(rx).await; - assert_eq!(received.len(), 1); - assert!(received[0].ends_with(case_id)); + assert_eq!(summary.queued, 1); + assert_eq!(summary.idle, 0); + assert_eq!(summary.required, 0); } #[tokio::test] - async fn skips_when_document_already_exists() { + async fn document_covering_latest_counts_as_current() { let user_root = TempDir::new().unwrap(); - let case = user_root - .path() - .join("22222222-2222-2222-2222-222222222222"); - fs::create_dir_all(&case).await.unwrap(); - write_input(&case, &make_input("2026-05-03T10:00:00Z")).await; - write_document(&case, "already analysed").await; + let case = seed_transcribed_case(user_root.path(), 2, base_time()).await; + write_document_covering(&case, base_time()).await; - let (tx, rx) = analyze_channel(); - let sent = enqueue_pending_for_user(user_root.path(), &tx).await; - drop(tx); + let summary = + compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; - assert_eq!(sent, 0); - assert!(drain(rx).await.is_empty()); + assert_eq!(summary.current, 1); + assert_eq!(summary.queued, 0); } - /// Regression: a stuck Llama job times out, leaves `analysis_input.json` - /// behind, and writes a failure marker. Without this skip the next - /// page reload re-enqueues the same job and blocks the worker for - /// another full timeout — see Bug 1 (case c414cf52, 2026-05-03). + /// Failure marker matching the input → operator must intervene; not + /// retried automatically. #[tokio::test] - async fn skips_when_failure_marker_matches_input_mtime() { + async fn failure_marker_matching_mtime_counts_as_failed() { let user_root = TempDir::new().unwrap(); - let case = user_root - .path() - .join("33333333-3333-3333-3333-333333333333"); - fs::create_dir_all(&case).await.unwrap(); - let mtime = "2026-05-03T10:00:00Z"; - write_input(&case, &make_input(mtime)).await; - write_marker(&case, mtime).await; + let case = seed_transcribed_case(user_root.path(), 3, base_time()).await; + write_marker(&case, base_time()).await; - let (tx, rx) = analyze_channel(); - let sent = enqueue_pending_for_user(user_root.path(), &tx).await; - drop(tx); + let summary = + compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; - assert_eq!(sent, 0, "must not re-enqueue an already-failed input"); - assert!(drain(rx).await.is_empty()); + assert_eq!(summary.failed, 1); } - /// A new recording bumps `last_recording_mtime`, so the old marker - /// no longer applies and recovery should re-enqueue. + /// Stale failure marker (older than current input) → recoverable on + /// next idle tick. #[tokio::test] - async fn enqueues_when_failure_marker_is_stale() { + async fn stale_failure_marker_falls_through_to_idle() { let user_root = TempDir::new().unwrap(); - let case_id = "44444444-4444-4444-4444-444444444444"; - let case = user_root.path().join(case_id); - fs::create_dir_all(&case).await.unwrap(); - write_input(&case, &make_input("2026-05-03T11:00:00Z")).await; - write_marker(&case, "2026-05-03T10:00:00Z").await; + let case = seed_transcribed_case(user_root.path(), 4, base_time()).await; + write_marker(&case, base_time() - Duration::from_secs(3600)).await; - let (tx, rx) = analyze_channel(); - let sent = enqueue_pending_for_user(user_root.path(), &tx).await; - drop(tx); + let summary = + compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; - assert_eq!(sent, 1); - let received = drain(rx).await; - assert_eq!(received.len(), 1); - assert!(received[0].ends_with(case_id)); + assert_eq!(summary.idle, 1); + assert_eq!(summary.failed, 0); + } + + #[tokio::test] + async fn closed_cases_are_skipped() { + let user_root = TempDir::new().unwrap(); + let case = seed_transcribed_case(user_root.path(), 5, base_time()).await; + // Mark closed; the closed-cases scan must skip this entirely. + fs::write(case.join(crate::paths::CLOSE_MARKER), "{}") + .await + .unwrap(); + + let summary = + compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; + + assert_eq!(summary, BootStateSummary::default()); + } + + /// Idle threshold not yet reached → state is `Required`, not `Idle`. + #[tokio::test] + async fn idle_threshold_not_reached_counts_as_required() { + let user_root = TempDir::new().unwrap(); + let _ = seed_transcribed_case(user_root.path(), 6, base_time()).await; + + // `now` is only 100s after the recording — below a 900s threshold. + let summary = compute_state_for_user( + user_root.path(), + Duration::from_secs(900), + base_time() + Duration::from_secs(100), + ) + .await; + + assert_eq!(summary.required, 1); + assert_eq!(summary.idle, 0); } } diff --git a/server/src/analyze/worker.rs b/server/src/analyze/worker.rs index d7a8108..17074ad 100644 --- a/server/src/analyze/worker.rs +++ b/server/src/analyze/worker.rs @@ -42,11 +42,12 @@ async fn process( let input_path = case_dir.join(ANALYSIS_INPUT_FILE); let document_path = case_dir.join(DOCUMENT_FILE); - // If the document already exists, the job is obsolete (e.g. recovery - // re-enqueued a finished case during a race). Skip silently. - if document_path.exists() { - return; - } + // 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 + // 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) { diff --git a/server/src/main.rs b/server/src/main.rs index 2cdbaa0..75858bf 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -184,7 +184,9 @@ async fn main() { }); } - // Analyze pipeline: channel + worker + recovery scan. + // Analyze pipeline: channel + worker. Boot-scan is observation-only — + // the per-render `try_enqueue_all_for_user` covers auto-trigger via + // the typed `CaseAnalysisState` machine. let (analyze_tx, analyze_rx) = analyze::channel(); let analyze_busy: doctate_server::WorkerBusy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); @@ -196,10 +198,16 @@ async fn main() { events_tx.clone(), )); { - let tx = analyze_tx.clone(); let data_path = config.data_path.clone(); + let idle_threshold = + std::time::Duration::from_secs(config.analysis_idle_threshold_secs); tokio::spawn(async move { - analyze::recovery::scan_and_enqueue(&data_path, &tx).await; + analyze::recovery::boot_scan_state( + &data_path, + idle_threshold, + std::time::SystemTime::now(), + ) + .await; }); } diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index e24a284..5e1d62e 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -13,9 +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, recovery as analyze_recovery, -}; +use crate::analyze::{ANALYSIS_INPUT_FILE, DOCUMENT_FILE, auto_trigger}; use crate::auth::AuthenticatedWebUser; use crate::case_id::{CaseId, CaseIdPath}; use crate::config::Config; @@ -519,9 +517,6 @@ impl PipelineState { vocab: &Arc, events_tx: &EventSender, ) { - if !self.analyze_busy.0.load(Ordering::Acquire) { - analyze_recovery::enqueue_pending_for_user(user_root, &self.analyze_tx).await; - } if !self.transcribe_busy.0.load(Ordering::Acquire) { transcribe_recovery::enqueue_pending_for_user(user_root, slug, &self.transcribe_tx) .await; diff --git a/server/tests/analyze_test.rs b/server/tests/analyze_test.rs index 3e8e6f3..44f9568 100644 --- a/server/tests/analyze_test.rs +++ b/server/tests/analyze_test.rs @@ -613,35 +613,28 @@ async fn replace_real_case_dry() { // Recovery // --------------------------------------------------------------------- +/// Boot-scan is observation-only after the typed-state refactor: even a +/// case sitting in the `Queued` state on disk must NOT be re-enqueued +/// at startup. The per-page-render `try_enqueue_all_for_user` is the +/// only auto-trigger. #[tokio::test] -async fn recovery_enqueues_pending_analysis() { +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(); std::fs::write(case_dir.join(ANALYSIS_INPUT_FILE), "{}").unwrap(); - let (tx, mut rx) = analyze::channel(); - analyze::recovery::scan_and_enqueue(&tmp, &tx).await; - - let job = rx.try_recv().expect("expected one job"); - assert_eq!(job.case_dir, case_dir); - assert!(rx.try_recv().is_err(), "no further jobs expected"); -} - -#[tokio::test] -async fn recovery_skips_completed_analysis() { - let tmp = unique_tmpdir("analyze-r2"); - 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(); - write_document_for_test(&case_dir, "done", "1970-01-01T00:00:00Z"); - - let (tx, mut rx) = analyze::channel(); - analyze::recovery::scan_and_enqueue(&tmp, &tx).await; + let (_tx, mut rx) = analyze::channel(); + analyze::recovery::boot_scan_state( + &tmp, + std::time::Duration::ZERO, + std::time::SystemTime::now(), + ) + .await; assert!( rx.try_recv().is_err(), - "completed analysis must not re-enqueue" + "boot scan must not enqueue any jobs in the new state model" ); } @@ -1246,10 +1239,18 @@ async fn recovery_skips_closed_cases() { ) .unwrap(); - let (tx, mut rx) = analyze::channel(); - analyze::recovery::scan_and_enqueue(&tmp, &tx).await; + let summary = analyze::recovery::boot_scan_state( + &tmp, + std::time::Duration::ZERO, + std::time::SystemTime::now(), + ) + .await; - assert!(rx.try_recv().is_err(), "closed case must not be enqueued"); + assert_eq!( + summary, + analyze::recovery::BootStateSummary::default(), + "closed case must not appear in any state bucket" + ); } // ---------------------------------------------------------------------