From 16ef0bbe78eb323844027e4108b1a6c8b56a606a Mon Sep 17 00:00:00 2001 From: Brummel Date: Tue, 5 May 2026 15:12:08 +0200 Subject: [PATCH] Introduce `boot_time` to `evaluate_state` The `boot_time` parameter is introduced to `evaluate_state` and its callers. This parameter acts as a lower bound for the "seen at" timestamp used in determining the `Idle` state for cases. Previously, the `elapsed` duration was calculated as `now.duration_since(latest_mtime)`. This meant that a server restart could immediately cause old recordings to be classified as `Idle` if their modification times were significantly in the past, potentially triggering a large number of LLM calls upon the first render of the `/cases` page. By using `seen_at = std::cmp::max(latest_mtime, boot_time)`, the idle calculation now ensures that a fresh server start respects a grace period. Old recordings will not satisfy the idle threshold until the elapsed time since the server's boot (or the recording's modification time, whichever is later) exceeds the `idle_threshold`. `SystemTime::UNIX_EPOCH` is used as the default `boot_time` in tests and in `create_router_and_session_store_with_settings`. This preserves existing test behavior where the idle grace period is effectively disabled. For production, `main.rs` now captures the server's actual boot time and passes it down. This change also refactors `evaluate_case` and `try_enqueue` to accept the new `boot_time` parameter. `try_enqueue_all_for_user` is updated to pass `boot_time` along. `boot_scan_state` and `compute_state_for_user` in `recovery.rs` also receive `boot_time`. Additionally, a new test case `analyze_required_endpoint_enqueues_only_required_cases` is added to verify the behavior of the `/cases/analyze-required` endpoint. This endpoint now correctly enqueues only `Required` --- server/src/analyze/auto_trigger.rs | 183 ++++++++++++++++++++++++- server/src/analyze/recovery.rs | 63 +++++++-- server/src/lib.rs | 21 +++ server/src/main.rs | 6 +- server/src/routes/bulk.rs | 17 +-- server/src/routes/user_web.rs | 35 +++-- server/tests/analyze_test.rs | 103 ++++++++++++++ server/tests/common/paths.rs | 3 + server/tests/magic_link_test.rs | 1 + server/tests/oneliner_override_test.rs | 1 + 10 files changed, 392 insertions(+), 41 deletions(-) diff --git a/server/src/analyze/auto_trigger.rs b/server/src/analyze/auto_trigger.rs index bb2af5f..2692518 100644 --- a/server/src/analyze/auto_trigger.rs +++ b/server/src/analyze/auto_trigger.rs @@ -166,10 +166,17 @@ pub enum CaseAnalysisState { /// /// `now` is injected so tests can pin filesystem mtimes against a known /// reference point. Production callers pass `SystemTime::now()`. +/// +/// `boot_time` is the lower bound for the "seen at"-clock: idle is +/// measured against `max(latest_recording_mtime, boot_time)`, so a +/// 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. pub async fn evaluate_state( case_dir: &Path, idle_threshold: std::time::Duration, now: SystemTime, + boot_time: SystemTime, ) -> CaseAnalysisState { let scan = scan_m4as(case_dir).await; @@ -213,9 +220,13 @@ pub async fn evaluate_state( // Stale document (or no document at all): differentiate between // "user might still be dictating" (Required) and "long enough quiet, - // fire" (Idle). + // fire" (Idle). The "seen at" clock is bounded below by the server + // boot time so a restart re-arms the grace window — old recordings + // do not instantly satisfy the threshold just because their mtime + // happens to be ancient. + let seen_at = std::cmp::max(latest_mtime, boot_time); let elapsed = now - .duration_since(latest_mtime) + .duration_since(seen_at) .unwrap_or(std::time::Duration::ZERO); if elapsed >= idle_threshold { CaseAnalysisState::Idle { has_document } @@ -242,7 +253,13 @@ pub(crate) async fn read_document_meta(case_dir: &Path) -> Option AutoDecision { - let state = evaluate_state(case_dir, std::time::Duration::ZERO, SystemTime::now()).await; + let state = evaluate_state( + case_dir, + std::time::Duration::ZERO, + SystemTime::now(), + SystemTime::UNIX_EPOCH, + ) + .await; match state { CaseAnalysisState::NoRecordings => AutoDecision::Skip("no recordings"), CaseAnalysisState::Pending => AutoDecision::Skip("not all transcribed"), @@ -272,10 +289,11 @@ pub async fn try_enqueue( case_dir: &Path, idle_threshold: std::time::Duration, now: SystemTime, + boot_time: SystemTime, tx: &AnalyzeSender, events_tx: &EventSender, ) -> bool { - match evaluate_state(case_dir, idle_threshold, now).await { + match evaluate_state(case_dir, idle_threshold, now, boot_time).await { CaseAnalysisState::Idle { .. } => {} _ => return false, } @@ -324,6 +342,7 @@ pub async fn try_enqueue_all_for_user( user_root: &Path, idle_threshold: std::time::Duration, now: SystemTime, + boot_time: SystemTime, tx: &AnalyzeSender, events_tx: &EventSender, ) { @@ -344,7 +363,7 @@ pub async fn try_enqueue_all_for_user( if crate::paths::is_closed(&case_path).await { continue; } - try_enqueue(&case_path, idle_threshold, now, tx, events_tx).await; + try_enqueue(&case_path, idle_threshold, now, boot_time, tx, events_tx).await; } } @@ -496,7 +515,10 @@ pub(crate) async fn read_failure_marker(case_dir: &Path) -> Option std::io::Result<()> { +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"); @@ -756,6 +778,154 @@ mod tests { ); } + #[tokio::test] + async fn idle_threshold_not_reached_returns_required() { + 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 base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000); + set_mtime(&m4a, base); + + // `now` is 100s after the recording, threshold is 900s → still + // inside the quiet window, so the case is `Required`, not `Idle`. + let state = evaluate_state( + dir.path(), + Duration::from_secs(900), + base + Duration::from_secs(100), + SystemTime::UNIX_EPOCH, + ) + .await; + + assert_eq!( + state, + CaseAnalysisState::Required { + has_document: false + } + ); + } + + #[tokio::test] + async fn idle_threshold_reached_returns_idle() { + 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 base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000); + set_mtime(&m4a, base); + + // `now` is 1000s after the recording, threshold is 900s → quiet + // window has elapsed, the auto-trigger may fire. + let state = evaluate_state( + dir.path(), + Duration::from_secs(900), + base + Duration::from_secs(1000), + SystemTime::UNIX_EPOCH, + ) + .await; + + assert_eq!( + state, + CaseAnalysisState::Idle { + has_document: false + } + ); + } + + /// A fresh server boot must NOT instantly classify ancient + /// recordings as `Idle`. Even when `now - latest_mtime` exceeds the + /// threshold by orders of magnitude, the boot-time lower bound + /// pulls the "seen at"-clock forward and the case stays `Required` + /// until the post-boot grace window has elapsed. + #[tokio::test] + async fn boot_time_reanchors_idle_window_after_restart() { + 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 base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000); + set_mtime(&m4a, base); + + // The recording is 30 days old in wallclock terms — but the + // server only booted 100 s ago. With a 900 s threshold that + // means the grace window has not yet elapsed: state must be + // 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; + + assert_eq!( + state, + CaseAnalysisState::Required { + has_document: false + } + ); + + // 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; + assert_eq!( + state_old, + CaseAnalysisState::Idle { + has_document: false + } + ); + } + + #[tokio::test] + async fn try_enqueue_skips_required_enqueues_idle() { + 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(); + + // Idle threshold not reached → must NOT enqueue. + let sent_required = try_enqueue( + dir.path(), + Duration::from_secs(900), + base + Duration::from_secs(100), + SystemTime::UNIX_EPOCH, + &tx, + &events_tx, + ) + .await; + assert!(!sent_required, "Required state must not enqueue"); + assert!( + rx.try_recv().is_err(), + "channel must be empty after Required" + ); + + // Same case, idle threshold passed → must enqueue exactly one job. + let sent_idle = try_enqueue( + dir.path(), + Duration::from_secs(900), + base + Duration::from_secs(1000), + SystemTime::UNIX_EPOCH, + &tx, + &events_tx, + ) + .await; + assert!(sent_idle, "Idle state must enqueue"); + assert!( + rx.try_recv().is_ok(), + "channel must hold one job after Idle" + ); + assert!(rx.try_recv().is_err(), "channel must hold exactly one job"); + } + #[tokio::test] async fn try_enqueue_all_sends_only_eligible_cases() { use crate::analyze::channel as analyze_channel; @@ -803,6 +973,7 @@ mod tests { user_root.path(), std::time::Duration::ZERO, SystemTime::now(), + SystemTime::UNIX_EPOCH, &tx, &events_tx, ) diff --git a/server/src/analyze/recovery.rs b/server/src/analyze/recovery.rs index 5e97d32..9b6e3ee 100644 --- a/server/src/analyze/recovery.rs +++ b/server/src/analyze/recovery.rs @@ -55,17 +55,23 @@ impl BootStateSummary { /// Walk `data_path/*/` and emit a single info-level summary line. The /// returned summary is mainly for tests; production callers ignore it. +/// +/// `boot_time` is forwarded to [`evaluate_state`]. Boot-scan typically +/// passes `SystemTime::now()` here — the same instant it itself observes +/// the case directories — so old recordings count as "just seen now". pub async fn boot_scan_state( data_path: &Path, idle_threshold: Duration, now: SystemTime, + boot_time: 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; + let per_user = + compute_state_for_user(&user_entry.path(), idle_threshold, now, boot_time).await; total.add(per_user); } info!( @@ -85,6 +91,7 @@ pub async fn compute_state_for_user( user_root: &Path, idle_threshold: Duration, now: SystemTime, + boot_time: SystemTime, ) -> BootStateSummary { let mut summary = BootStateSummary::default(); let Ok(mut cases) = tokio::fs::read_dir(user_root).await else { @@ -101,7 +108,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).await; + let state = evaluate_state(&case_dir, idle_threshold, now, boot_time).await; summary.record(&state); } summary @@ -129,7 +136,11 @@ mod tests { /// 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 { + 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"; @@ -203,8 +214,13 @@ mod tests { let case = seed_transcribed_case(user_root.path(), 1, base_time()).await; write_input(&case, base_time()).await; - let summary = - compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; + let summary = compute_state_for_user( + user_root.path(), + Duration::ZERO, + base_time(), + SystemTime::UNIX_EPOCH, + ) + .await; assert_eq!(summary.queued, 1); assert_eq!(summary.idle, 0); @@ -217,8 +233,13 @@ mod tests { let case = seed_transcribed_case(user_root.path(), 2, base_time()).await; write_document_covering(&case, base_time()).await; - let summary = - compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; + let summary = compute_state_for_user( + user_root.path(), + Duration::ZERO, + base_time(), + SystemTime::UNIX_EPOCH, + ) + .await; assert_eq!(summary.current, 1); assert_eq!(summary.queued, 0); @@ -232,8 +253,13 @@ mod tests { let case = seed_transcribed_case(user_root.path(), 3, base_time()).await; write_marker(&case, base_time()).await; - let summary = - compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; + let summary = compute_state_for_user( + user_root.path(), + Duration::ZERO, + base_time(), + SystemTime::UNIX_EPOCH, + ) + .await; assert_eq!(summary.failed, 1); } @@ -246,8 +272,13 @@ mod tests { let case = seed_transcribed_case(user_root.path(), 4, base_time()).await; write_marker(&case, base_time() - Duration::from_secs(3600)).await; - let summary = - compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; + let summary = compute_state_for_user( + user_root.path(), + Duration::ZERO, + base_time(), + SystemTime::UNIX_EPOCH, + ) + .await; assert_eq!(summary.idle, 1); assert_eq!(summary.failed, 0); @@ -262,8 +293,13 @@ mod tests { .await .unwrap(); - let summary = - compute_state_for_user(user_root.path(), Duration::ZERO, base_time()).await; + let summary = compute_state_for_user( + user_root.path(), + Duration::ZERO, + base_time(), + SystemTime::UNIX_EPOCH, + ) + .await; assert_eq!(summary, BootStateSummary::default()); } @@ -279,6 +315,7 @@ mod tests { user_root.path(), Duration::from_secs(900), base_time() + Duration::from_secs(100), + SystemTime::UNIX_EPOCH, ) .await; diff --git a/server/src/lib.rs b/server/src/lib.rs index f5e4224..b09d326 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -87,6 +87,15 @@ pub struct PipelineState { pub oneliner_locks: OnelinerLocks, } +/// Newtype for the server's start timestamp. Functions as the lower +/// bound for the analyze idle-trigger: a case is only auto-enqueued +/// once `now - max(latest_recording_mtime, boot_time) >= threshold`. +/// Without this, a server restart would treat every case with an +/// older recording as immediately `Idle` and spawn a thundering herd +/// of LLM calls on the first `/cases` render. +#[derive(Clone, Copy, Debug)] +pub struct BootTime(pub std::time::SystemTime); + /// Shared application state. Clonable — all fields are cheap to clone /// (`Arc` and `mpsc::Sender` / `broadcast::Sender`). #[derive(Clone)] @@ -104,6 +113,7 @@ pub struct AppState { pub events_tx: events::EventSender, pub http_client: reqwest::Client, pub vocab: Arc, + pub boot_time: BootTime, } impl FromRef for Arc { @@ -197,6 +207,12 @@ impl FromRef for Arc { } } +impl FromRef for BootTime { + fn from_ref(state: &AppState) -> Self { + state.boot_time + } +} + /// 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. @@ -242,6 +258,11 @@ pub fn create_router_and_session_store_with_settings( events_tx: events::channel(), http_client: reqwest::Client::new(), vocab: Arc::new(Gazetteer::empty()), + // UNIX_EPOCH = "the server has always been up" — disables the + // post-boot grace window in tests, which is what most tests want + // anyway. Tests that exercise the grace-window itself construct + // an `AppState` with an explicit recent `BootTime`. + boot_time: BootTime(std::time::SystemTime::UNIX_EPOCH), }); (router, session_store) } diff --git a/server/src/main.rs b/server/src/main.rs index 75858bf..5bc05e0 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -197,15 +197,16 @@ async fn main() { vocab.clone(), events_tx.clone(), )); + let server_boot_time = std::time::SystemTime::now(); { let data_path = config.data_path.clone(); - let idle_threshold = - std::time::Duration::from_secs(config.analysis_idle_threshold_secs); + let idle_threshold = std::time::Duration::from_secs(config.analysis_idle_threshold_secs); tokio::spawn(async move { analyze::recovery::boot_scan_state( &data_path, idle_threshold, std::time::SystemTime::now(), + server_boot_time, ) .await; }); @@ -225,6 +226,7 @@ async fn main() { events_tx, http_client, vocab, + boot_time: doctate_server::BootTime(server_boot_time), }; let addr = format!("0.0.0.0:{}", config.server_port); diff --git a/server/src/routes/bulk.rs b/server/src/routes/bulk.rs index cfff4e4..748078a 100644 --- a/server/src/routes/bulk.rs +++ b/server/src/routes/bulk.rs @@ -8,9 +8,7 @@ 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, write_input_overwrite}; use crate::analyze::backend::any_backend_available; use crate::analyze::{ANALYSIS_INPUT_FILE, AnalyzeJob, AnalyzeSender}; use crate::auth::AuthenticatedWebUser; @@ -237,6 +235,7 @@ pub async fn handle_analyze_required( State(config): State>, State(analyze_tx): State, State(events_tx): State, + State(boot_time): State, CsrfForm(_form): CsrfForm, ) -> Result { let user_root = config.data_path.join(&user.slug); @@ -247,8 +246,7 @@ pub async fn handle_analyze_required( } let now = std::time::SystemTime::now(); - let idle_threshold = - std::time::Duration::from_secs(config.analysis_idle_threshold_secs); + let idle_threshold = std::time::Duration::from_secs(config.analysis_idle_threshold_secs); let mut ok = 0usize; let mut skipped = 0usize; @@ -278,7 +276,7 @@ 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).await { + match auto_trigger::evaluate_state(&case_dir, idle_threshold, now, boot_time.0).await { CaseAnalysisState::Required { .. } | CaseAnalysisState::Idle { .. } => {} _ => { skipped += 1; @@ -310,12 +308,7 @@ pub async fn handle_analyze_required( skipped += 1; continue; } - events::emit( - &events_tx, - &user.slug, - &name, - CaseEventKind::AnalysisQueued, - ); + events::emit(&events_tx, &user.slug, &name, CaseEventKind::AnalysisQueued); ok += 1; } diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index 46cd288..45de811 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -448,6 +448,7 @@ struct CaseFlags { analysis_stale_with_doc: bool, } +#[allow(clippy::too_many_arguments)] async fn compute_flags( case_dir: &Path, recordings: &[RecordingView], @@ -455,6 +456,7 @@ async fn compute_flags( worker_busy: bool, idle_threshold: std::time::Duration, now_systime: std::time::SystemTime, + boot_time: std::time::SystemTime, ) -> CaseFlags { let has_document = any_document_exists(case_dir).await; // Honest status: an input file alone does not mean a job is in flight. @@ -496,7 +498,8 @@ async fn compute_flags( None }; - let state = auto_trigger::evaluate_state(case_dir, idle_threshold, now_systime).await; + let state = + auto_trigger::evaluate_state(case_dir, idle_threshold, now_systime, boot_time).await; let analysis_stale_with_doc = matches!( state, auto_trigger::CaseAnalysisState::Required { has_document: true } @@ -603,6 +606,7 @@ pub async fn handle_my_cases( State(events_tx): State, State(http_client): State, State(vocab): State>, + State(boot_time): State, Query(query): Query, ) -> Result, AppError> { let user_root = config.data_path.join(&user.slug); @@ -626,6 +630,7 @@ pub async fn handle_my_cases( &user_root, idle_threshold, now, + boot_time.0, &pipeline.analyze_tx, &events_tx, ) @@ -653,6 +658,7 @@ pub async fn handle_my_cases( show_closed, retention.auto_delete_days, crate::analyze::backend::any_backend_available(), + boot_time.0, ) .await; let total = cases.len(); @@ -707,6 +713,7 @@ pub async fn handle_case_page( State(events_tx): State, State(http_client): State, State(vocab): State>, + State(boot_time): State, CaseIdPath(case_id): CaseIdPath, Query(query): Query, ) -> Result, AppError> { @@ -749,6 +756,7 @@ pub async fn handle_case_page( &case_dir, idle_threshold, now, + boot_time.0, &pipeline.analyze_tx, &events_tx, ) @@ -777,6 +785,7 @@ pub async fn handle_case_page( a_busy, idle_threshold, now, + boot_time.0, ) .await; @@ -975,6 +984,7 @@ pub(crate) async fn locate_closed_case_or_404( /// behaviour for the show-closed view. `auto_delete_days` is used to /// compute the per-case `days_until_purge` countdown; pass 0 to /// disable the countdown (maps to `None` on the view). +#[allow(clippy::too_many_arguments)] async fn scan_user_cases( config: &Config, slug: &str, @@ -982,6 +992,7 @@ async fn scan_user_cases( include_closed: bool, auto_delete_days: u32, llm_configured: bool, + boot_time: std::time::SystemTime, ) -> Vec { let user_root = config.data_path.join(slug); let mut cases = Vec::new(); @@ -1007,6 +1018,7 @@ async fn scan_user_cases( now, idle_threshold, now_systime, + boot_time, ) .await { @@ -1053,6 +1065,7 @@ async fn compute_case_view( now: OffsetDateTime, idle_threshold: std::time::Duration, now_systime: std::time::SystemTime, + boot_time: std::time::SystemTime, ) -> Option { let recordings = scan_recordings(case_path).await; if recordings.is_empty() { @@ -1078,15 +1091,21 @@ async fn compute_case_view( let analyzing = !has_document && worker_busy && any_analysis_input_exists(case_path).await; let can_analyze = !analyzing && all_transcribed && llm_configured; - let state = auto_trigger::evaluate_state(case_path, idle_threshold, now_systime).await; - let analysis_required = matches!( - state, - crate::analyze::auto_trigger::CaseAnalysisState::Required { .. } - | crate::analyze::auto_trigger::CaseAnalysisState::Idle { .. } - ); - let (is_closed, days_until_purge) = closed_state(case_path, auto_delete_days, now).await; + let state = + auto_trigger::evaluate_state(case_path, idle_threshold, now_systime, boot_time).await; + // Closed cases must never advertise as "Analyse erforderlich" — the + // whole point of closing a case is to freeze it. The bulk-endpoint + // already skips closed cases in its enqueue loop; this keeps the UI + // classification in lockstep so the badge / required_count match. + let analysis_required = !is_closed + && matches!( + state, + crate::analyze::auto_trigger::CaseAnalysisState::Required { .. } + | crate::analyze::auto_trigger::CaseAnalysisState::Idle { .. } + ); + // One extra read per case with a document. Documents are in the KB // range — negligible next to the existing scan_recordings I/O — and // we skip the read entirely when the stat above said no document. diff --git a/server/tests/analyze_test.rs b/server/tests/analyze_test.rs index 44f9568..9ecbcb6 100644 --- a/server/tests/analyze_test.rs +++ b/server/tests/analyze_test.rs @@ -629,6 +629,7 @@ async fn recovery_does_not_enqueue_at_boot() { &tmp, std::time::Duration::ZERO, std::time::SystemTime::now(), + std::time::SystemTime::UNIX_EPOCH, ) .await; @@ -1243,6 +1244,7 @@ async fn recovery_skips_closed_cases() { &tmp, std::time::Duration::ZERO, std::time::SystemTime::now(), + std::time::SystemTime::UNIX_EPOCH, ) .await; @@ -1509,3 +1511,104 @@ async fn analyze_works_against_ollama_style_endpoint_without_api_key() { drop(tx); let _ = tokio::time::timeout(Duration::from_secs(2), handle).await; } + +// --------------------------------------------------------------------- +// Bulk "Aktualisieren" endpoint +// --------------------------------------------------------------------- + +/// `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 +/// `analysis_input.json` marker — written before the channel send, so its +/// 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; + use time::OffsetDateTime; + use time::format_description::well_known::Rfc3339; + + fn to_rfc3339(t: std::time::SystemTime) -> String { + OffsetDateTime::from(t).format(&Rfc3339).unwrap() + } + + let (cfg, settings) = cfg_llm("ar-1"); + let user_root = cfg.data_path.join("dr_a"); + + // Required: transcribed recording, no document. + let req_id = "11111111-1111-1111-1111-111111111111"; + let req_dir = seed_case(&cfg.data_path, "dr_a", req_id); + seed_recording(&req_dir, "2026-04-15T10-00-00Z", Some("erste Aufnahme")); + + // Current: document covers the recording's mtime exactly. + let cur_id = "22222222-2222-2222-2222-222222222222"; + let cur_dir = seed_case(&cfg.data_path, "dr_a", cur_id); + seed_recording(&cur_dir, "2026-04-15T10-00-00Z", Some("alles okay")); + let cur_m4a_mtime = std::fs::metadata(cur_dir.join("2026-04-15T10-00-00Z.m4a")) + .unwrap() + .modified() + .unwrap(); + write_document_for_test(&cur_dir, "fertig", &to_rfc3339(cur_m4a_mtime)); + + // Failed: failure marker matches latest recording mtime. + let fail_id = "33333333-3333-3333-3333-333333333333"; + let fail_dir = seed_case(&cfg.data_path, "dr_a", fail_id); + seed_recording(&fail_dir, "2026-04-15T10-00-00Z", Some("kaputt")); + let fail_m4a_mtime = std::fs::metadata(fail_dir.join("2026-04-15T10-00-00Z.m4a")) + .unwrap() + .modified() + .unwrap(); + let marker = FailureMarker { + last_recording_mtime: to_rfc3339(fail_m4a_mtime), + reason: "test".into(), + failed_at: "2026-04-15T10-05-00Z".into(), + backend_id: None, + details: None, + }; + std::fs::write( + fail_dir.join(".analysis_failed.json"), + serde_json::to_vec(&marker).unwrap(), + ) + .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; + + let resp = app + .oneshot(csrf_form_post(paths::ANALYZE_REQUIRED, &cookie, &csrf, "")) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::SEE_OTHER); + assert_eq!( + resp.headers().get(header::LOCATION).unwrap(), + "/cases", + "must redirect back to the case list" + ); + + // Required → analysis_input.json was written, the case is now Queued. + assert!( + user_root.join(req_id).join(ANALYSIS_INPUT_FILE).exists(), + "Required case must have been enqueued" + ); + // Current → no enqueue, document.json untouched. + assert!( + !user_root.join(cur_id).join(ANALYSIS_INPUT_FILE).exists(), + "Current case must not be enqueued" + ); + assert!( + user_root.join(cur_id).join(DOCUMENT_FILE).exists(), + "Current case's document must remain in place" + ); + // Failed → no enqueue. 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(), + "Failed case must not be enqueued" + ); + assert!( + user_root + .join(fail_id) + .join(".analysis_failed.json") + .exists(), + "failure marker must persist" + ); +} diff --git a/server/tests/common/paths.rs b/server/tests/common/paths.rs index f3467ab..1d9ede7 100644 --- a/server/tests/common/paths.rs +++ b/server/tests/common/paths.rs @@ -19,6 +19,9 @@ pub const LOGOUT: &str = "/logout"; pub const CASES: &str = "/cases"; /// `POST /cases/bulk` — admin bulk action (close/analyze/reset). pub const BULK: &str = "/cases/bulk"; +/// `POST /cases/analyze-required` — user-facing bulk "Aktualisieren" +/// button: enqueues every Required/Idle case for the authenticated user. +pub const ANALYZE_REQUIRED: &str = "/cases/analyze-required"; /// `POST /cases/purge-closed` — admin hard-delete of closed cases. pub const PURGE_CLOSED: &str = "/cases/purge-closed"; /// `GET /events` — SSE stream (per-user scoped). diff --git a/server/tests/magic_link_test.rs b/server/tests/magic_link_test.rs index 81ba93f..d90b77d 100644 --- a/server/tests/magic_link_test.rs +++ b/server/tests/magic_link_test.rs @@ -71,6 +71,7 @@ fn build_state_with_stores() -> AppState { events_tx: doctate_server::events::channel(), http_client: reqwest::Client::new(), vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()), + boot_time: doctate_server::BootTime(std::time::SystemTime::UNIX_EPOCH), } } diff --git a/server/tests/oneliner_override_test.rs b/server/tests/oneliner_override_test.rs index 05c8c20..1d24a4d 100644 --- a/server/tests/oneliner_override_test.rs +++ b/server/tests/oneliner_override_test.rs @@ -84,6 +84,7 @@ fn build_app_with_events_tx( 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), } }