Brummel/doctate
1
0
Fork 0
Code Issues 9 Pull Requests Actions Packages Projects Releases Wiki Activity

Introduce boot_time to evaluate_state

Browse Source 
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`
This commit is contained in:
Brummel
2026-05-05 15:12:08 +02:00
parent bd133e9944
commit 16ef0bbe78
10 changed files with 392 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/analyze/auto_trigger.rs
+177 -6
View File
@@ -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<crate::analyze
/// `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;
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<FailureMarker
/// Auto-trigger re-analyses may legitimately replace an existing input
/// (e.g. a new recording arrived between two reloads while the previous
/// job was still queued). `.tmp` + fsync + rename keeps it atomic.
pub(crate) async fn write_input_overwrite(path: &Path, input: &AnalysisInput) -> std::io::Result<()> {
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,
)
server/src/analyze/recovery.rs
+50 -13
View File
@@ -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;
server/src/lib.rs
+21
View File
@@ -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<Gazetteer>,
pub boot_time: BootTime,
}
impl FromRef<AppState> for Arc<Config> {
@@ -197,6 +207,12 @@ impl FromRef<AppState> for Arc<Gazetteer> {
}
}
impl FromRef<AppState> 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)
}
server/src/main.rs
+4 -2
View File
@@ -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);
server/src/routes/bulk.rs
+5 -12
View File
@@ -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<Arc<Config>>,
State(analyze_tx): State<AnalyzeSender>,
State(events_tx): State<EventSender>,
State(boot_time): State<crate::BootTime>,
CsrfForm(_form): CsrfForm<AnalyzeRequiredForm>,
) -> Result<Redirect, AppError> {
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;
}
server/src/routes/user_web.rs
+27 -8
View File
@@ -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<EventSender>,
State(http_client): State<reqwest::Client>,
State(vocab): State<Arc<Gazetteer>>,
State(boot_time): State<crate::BootTime>,
Query(query): Query<CaseListQuery>,
) -> Result<Html<String>, 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<EventSender>,
State(http_client): State<reqwest::Client>,
State(vocab): State<Arc<Gazetteer>>,
State(boot_time): State<crate::BootTime>,
CaseIdPath(case_id): CaseIdPath,
Query(query): Query<CaseListQuery>,
) -> Result<Html<String>, 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<UserCaseView> {
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<UserCaseView> {
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.
server/tests/analyze_test.rs
+103
View File
@@ -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"
);
}
server/tests/common/paths.rs
+3
View File
@@ -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).
server/tests/magic_link_test.rs
+1
View File
@@ -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),
}
}
server/tests/oneliner_override_test.rs
+1
View File
@@ -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),
}
}