diff --git a/client-desktop/src/app.rs b/client-desktop/src/app.rs index f226d1a..3685ebd 100644 --- a/client-desktop/src/app.rs +++ b/client-desktop/src/app.rs @@ -521,9 +521,9 @@ impl DoctateApp { let time_str = extract_hhmm(&marker.last_activity_at); let oneliner = match &marker.oneliner { - Some(OnelinerState::Ready { text, .. } | OnelinerState::Manual { text, .. }) => { - text.clone() - } + Some( + OnelinerState::Ready { text, .. } | OnelinerState::Manual { text, .. }, + ) => text.clone(), Some(OnelinerState::Empty { .. }) => "∅".to_owned(), Some(OnelinerState::Error { .. }) => "⚠".to_owned(), None => "⏳".to_owned(), diff --git a/server/src/lib.rs b/server/src/lib.rs index 99a654e..0241c37 100644 --- a/server/src/lib.rs +++ b/server/src/lib.rs @@ -8,6 +8,7 @@ pub mod events; pub mod gazetteer; pub mod magic_link; pub mod models; +pub mod oneliner_locks; pub mod paths; pub mod retention; pub mod routes; @@ -26,6 +27,7 @@ use analyze::AnalyzeSender; use config::Config; use gazetteer::Gazetteer; use magic_link::MagicLinkStore; +use oneliner_locks::OnelinerLocks; use transcribe::TranscribeSender; use web_session::SessionStore; @@ -78,6 +80,7 @@ pub struct PipelineState { pub transcribe_busy: TranscribeBusy, pub transcribe_tx: TranscribeSender, pub oneliner_heal_busy: OnelinerHealBusy, + pub oneliner_locks: OnelinerLocks, } /// Shared application state. Clonable — all fields are cheap to clone @@ -92,6 +95,7 @@ pub struct AppState { pub analyze_busy: AnalyzeBusy, pub transcribe_busy: TranscribeBusy, pub oneliner_heal_busy: OnelinerHealBusy, + pub oneliner_locks: OnelinerLocks, pub events_tx: events::EventSender, pub http_client: reqwest::Client, pub vocab: Arc, @@ -145,6 +149,12 @@ impl FromRef for OnelinerHealBusy { } } +impl FromRef for OnelinerLocks { + fn from_ref(state: &AppState) -> Self { + state.oneliner_locks.clone() + } +} + impl FromRef for PipelineState { fn from_ref(state: &AppState) -> Self { PipelineState { @@ -153,6 +163,7 @@ impl FromRef for PipelineState { transcribe_busy: state.transcribe_busy.clone(), transcribe_tx: state.transcribe_tx.clone(), oneliner_heal_busy: state.oneliner_heal_busy.clone(), + oneliner_locks: state.oneliner_locks.clone(), } } } @@ -200,6 +211,7 @@ pub fn create_router_and_session_store(config: Arc) -> (Router, SessionS analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), oneliner_heal_busy: OnelinerHealBusy(Arc::new(AtomicBool::new(false))), + oneliner_locks: OnelinerLocks::new(), events_tx: events::channel(), http_client: reqwest::Client::new(), vocab: Arc::new(Gazetteer::empty()), diff --git a/server/src/main.rs b/server/src/main.rs index c23960b..125dc57 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -122,6 +122,12 @@ async fn main() { let oneliner_heal_busy: doctate_server::WorkerBusy = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + // Per-`case_dir` mutex registry. Shared by the PUT handler (manual + // override write) and the transcribe worker (post-LLM re-check) + // so a doctor's manual edit always wins over a concurrent + // auto-regen, no matter the timing. + let oneliner_locks = doctate_server::oneliner_locks::OnelinerLocks::new(); + // Transcription pipeline: channel + worker + recovery scan. let (transcribe_tx, transcribe_rx) = transcribe::channel(); let transcribe_busy: doctate_server::WorkerBusy = @@ -133,6 +139,7 @@ async fn main() { transcribe_busy.clone(), vocab.clone(), events_tx.clone(), + oneliner_locks.clone(), )); { let tx = transcribe_tx.clone(); @@ -142,6 +149,7 @@ async fn main() { let vocab = vocab.clone(); let events = events_tx.clone(); let heal_busy = oneliner_heal_busy.clone(); + let locks = oneliner_locks.clone(); tokio::spawn(async move { transcribe::recovery::scan_and_enqueue(&data_path, &tx).await; // Then catch up oneliners for cases that crashed between @@ -150,7 +158,7 @@ async fn main() { // page-load does not spawn a second parallel regen loop. let _guard = doctate_server::BusyGuard::new(heal_busy); transcribe::recovery::regenerate_missing_oneliners( - &data_path, &client, &cfg, &vocab, &events, + &data_path, &client, &cfg, &vocab, &events, &locks, ) .await; }); @@ -185,6 +193,7 @@ async fn main() { analyze_busy: doctate_server::AnalyzeBusy(analyze_busy), transcribe_busy: doctate_server::TranscribeBusy(transcribe_busy), oneliner_heal_busy: doctate_server::OnelinerHealBusy(oneliner_heal_busy), + oneliner_locks, events_tx, http_client, vocab, diff --git a/server/src/oneliner_locks.rs b/server/src/oneliner_locks.rs new file mode 100644 index 0000000..48ed12e --- /dev/null +++ b/server/src/oneliner_locks.rs @@ -0,0 +1,115 @@ +//! Per-`case_dir` async mutex registry. +//! +//! Serializes the read-modify-write sequence around `oneliner.json` +//! so that a manual override (PUT handler) cannot race with a concurrent +//! LLM-driven auto-regeneration (`update_oneliner` in `transcribe::worker`). +//! +//! Concretely: both code paths take `OnelinerLocks::lock_for(case_dir)` +//! before reading-then-writing the state. Different case dirs use +//! independent inner mutexes, so unrelated cases never block each other. +//! +//! The registry is intentionally never pruned. Inner mutex entries are +//! tiny (`Arc>`) and the case-dir set per process is bounded +//! by realistic medical workloads (low thousands within the rolling +//! window). If this ever becomes a memory concern, weak references or +//! a periodic sweeper can be added without touching call sites. + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use tokio::sync::{Mutex, OwnedMutexGuard}; + +/// Newtype wrapper around the per-`case_dir` mutex registry. +/// +/// Cloning is cheap (`Arc` clone). Ships in `AppState` as a regular +/// field; handlers receive it via `State` thanks to a +/// `FromRef` impl on the parent state. +#[derive(Clone, Default)] +pub struct OnelinerLocks(Arc>>>>); + +impl OnelinerLocks { + pub fn new() -> Self { + Self::default() + } + + /// Acquire the lock guarding `case_dir`'s `oneliner.json` read-modify-write + /// sequence. The returned guard owns its own `Arc` to the inner mutex, + /// so callers may carry it across `await` points (PUT handler, worker + /// re-check) without lifetime gymnastics. + pub async fn lock_for(&self, case_dir: &Path) -> OwnedMutexGuard<()> { + let inner = { + let mut map = self.0.lock().await; + map.entry(case_dir.to_path_buf()) + .or_insert_with(|| Arc::new(Mutex::new(()))) + .clone() + }; + inner.lock_owned().await + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + use tokio::time::{Instant, timeout}; + + #[tokio::test] + async fn distinct_case_dirs_do_not_block() { + let locks = OnelinerLocks::new(); + let path_a = Path::new("/tmp/case-a"); + let path_b = Path::new("/tmp/case-b"); + + let _guard_a = locks.lock_for(path_a).await; + // Holding guard_a must NOT prevent locking path_b. + let result = timeout(Duration::from_millis(100), locks.lock_for(path_b)).await; + assert!( + result.is_ok(), + "lock_for(path_b) should not be blocked by lock_for(path_a)" + ); + } + + #[tokio::test] + async fn same_case_dir_serializes() { + let locks = OnelinerLocks::new(); + let path = Path::new("/tmp/case-c"); + + let guard1 = locks.lock_for(path).await; + + // Second lock attempt on same path must wait. + let locks_clone = locks.clone(); + let path_buf = path.to_path_buf(); + let waiter = tokio::spawn(async move { + let start = Instant::now(); + let _g2 = locks_clone.lock_for(&path_buf).await; + start.elapsed() + }); + + // Yield once to let the spawned task try to acquire and park. + tokio::time::sleep(Duration::from_millis(50)).await; + drop(guard1); + + let elapsed = waiter.await.expect("waiter task panicked"); + assert!( + elapsed >= Duration::from_millis(40), + "second lock should have waited at least ~50ms, waited {elapsed:?}" + ); + } + + #[tokio::test] + async fn registry_reuses_inner_mutex_across_calls() { + let locks = OnelinerLocks::new(); + let path = Path::new("/tmp/case-d"); + + // Take and release; second lock must hit the same inner mutex + // (i.e. registry persists the entry). + { + let _g = locks.lock_for(path).await; + } + let map_len_after_first = locks.0.lock().await.len(); + let _g = locks.lock_for(path).await; + let map_len_after_second = locks.0.lock().await.len(); + assert_eq!(map_len_after_first, 1); + assert_eq!(map_len_after_second, 1); + } +} diff --git a/server/src/paths.rs b/server/src/paths.rs index e5fffc8..bd86166 100644 --- a/server/src/paths.rs +++ b/server/src/paths.rs @@ -87,10 +87,7 @@ pub async fn read_oneliner_state( /// `write(tmp) + rename(tmp, final)`. Rename is atomic within a single /// filesystem, so concurrent readers never see a half-serialized JSON /// document. -pub async fn write_oneliner_state( - case_dir: &Path, - state: &OnelinerState, -) -> std::io::Result<()> { +pub async fn write_oneliner_state(case_dir: &Path, state: &OnelinerState) -> std::io::Result<()> { let final_path = case_dir.join(ONELINER_FILENAME); let tmp_path = case_dir.join(format!("{ONELINER_FILENAME}.tmp")); let payload = serde_json::to_vec(state).expect("OnelinerState is infallibly serializable"); diff --git a/server/src/routes/mod.rs b/server/src/routes/mod.rs index 8349af7..55001d9 100644 --- a/server/src/routes/mod.rs +++ b/server/src/routes/mod.rs @@ -5,13 +5,14 @@ mod events; mod health; mod login; mod magic; +mod oneliner_override; mod oneliners; mod upload; pub(crate) mod user_web; pub(crate) mod web; use axum::Router; -use axum::routing::{get, post}; +use axum::routing::{get, post, put}; use crate::AppState; @@ -21,6 +22,10 @@ pub fn api_router() -> Router { .route("/api/debug/whoami", get(debug::handle_whoami)) .route("/api/upload", post(upload::handle_upload)) .route("/api/oneliners", get(oneliners::handle_oneliners)) + .route( + "/api/cases/{case_id}/oneliner", + put(oneliner_override::handle_put_oneliner), + ) .route("/api/auth/magic-link", post(magic::handle_create)) .route("/web/magic", get(magic::handle_consume)) .route( diff --git a/server/src/routes/oneliner_override.rs b/server/src/routes/oneliner_override.rs new file mode 100644 index 0000000..6a2b691 --- /dev/null +++ b/server/src/routes/oneliner_override.rs @@ -0,0 +1,86 @@ +//! `PUT /api/cases/{case_id}/oneliner` — doctor's manual oneliner override. +//! +//! Writes [`OnelinerState::Manual`], which the transcribe worker's +//! `update_oneliner` treats as terminal: the manual text is sticky +//! across new recordings until another manual edit (or explicit reset) +//! replaces it. See plan +//! `plane-die-erweiterung-oneliner-override-pure-goblet.md` for the +//! full concurrency rationale. +//! +//! Authentication: `X-API-Key` header, shared with `/api/upload` and +//! `/api/oneliners` (the GET counterpart). +//! +//! Concurrency: holds [`OnelinerLocks::lock_for`] for the duration of +//! the disk write. The worker's auto-regen path takes the same lock +//! around its post-LLM re-read-and-write sequence, so a doctor's PUT +//! arriving while the LLM is still thinking always wins. + +use axum::Json; +use axum::extract::State; +use doctate_common::oneliners::{OnelinerOverrideRequest, OnelinerState}; +use tracing::info; + +use crate::auth::AuthenticatedUser; +use crate::case_id::CaseIdPath; +use crate::error::AppError; +use crate::events::{self, CaseEventKind, EventSender}; +use crate::oneliner_locks::OnelinerLocks; +use crate::paths; + +/// Hard cap on accepted text length. A oneliner is a single sentence +/// (typical 30–80 chars); 500 leaves comfortable headroom while +/// preventing accidental dumps from a runaway client. +const MAX_ONELINER_TEXT_LEN: usize = 500; + +pub async fn handle_put_oneliner( + user: AuthenticatedUser, + State(locks): State, + State(events_tx): State, + CaseIdPath(case_id): CaseIdPath, + Json(req): Json, +) -> Result, AppError> { + let trimmed = req.text.trim(); + if trimmed.is_empty() { + return Err(AppError::BadRequest( + "oneliner text must not be empty".into(), + )); + } + if trimmed.chars().count() > MAX_ONELINER_TEXT_LEN { + return Err(AppError::BadRequest(format!( + "oneliner text exceeds {MAX_ONELINER_TEXT_LEN} characters" + ))); + } + + let case_id_str = case_id.to_string(); + let case_dir = user.data_dir.join(&case_id_str); + if !tokio::fs::try_exists(&case_dir).await.unwrap_or(false) { + return Err(AppError::NotFound(format!("case {case_id_str} not found"))); + } + + let state = OnelinerState::Manual { + text: trimmed.to_owned(), + set_at: doctate_common::now_rfc3339(), + }; + + // RAII guard — released on function exit. Same lock the worker takes + // for its re-check-then-write window, so a concurrent auto-regen + // cannot overwrite this manual edit. + let _guard = locks.lock_for(&case_dir).await; + paths::write_oneliner_state(&case_dir, &state).await?; + + info!( + user = %user.slug, + case_id = %case_id_str, + len = trimmed.chars().count(), + "oneliner manual override set" + ); + + events::emit( + &events_tx, + user.slug.clone(), + case_id_str, + CaseEventKind::OnelinerUpdated, + ); + + Ok(Json(state)) +} diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index 4d66b9b..4f254ac 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -451,6 +451,7 @@ impl PipelineState { let vocab = Arc::clone(vocab); let events_tx = events_tx.clone(); let busy = Arc::clone(&self.oneliner_heal_busy.0); + let locks = self.oneliner_locks.clone(); info!(user = %slug, "oneliner heal spawned"); tokio::spawn(async move { // `BusyGuard::Drop` resets the flag to `false` — even @@ -465,6 +466,7 @@ impl PipelineState { &config, &vocab, &events_tx, + &locks, ) .await; }); diff --git a/server/src/transcribe/recovery.rs b/server/src/transcribe/recovery.rs index 3bbb02f..4d9305f 100644 --- a/server/src/transcribe/recovery.rs +++ b/server/src/transcribe/recovery.rs @@ -7,6 +7,7 @@ use super::{TranscribeJob, TranscribeSender}; use crate::config::Config; use crate::events::EventSender; use crate::gazetteer::Gazetteer; +use crate::oneliner_locks::OnelinerLocks; use crate::paths; /// Walk `data_path/*/` and enqueue every recording pending transcription for @@ -206,6 +207,7 @@ pub async fn regenerate_missing_oneliners( config: &Config, vocab: &Gazetteer, events_tx: &EventSender, + locks: &OnelinerLocks, ) { let cases = cases_needing_oneliner(data_path).await; if cases.is_empty() { @@ -213,7 +215,8 @@ pub async fn regenerate_missing_oneliners( } info!(count = cases.len(), "Regenerating missing oneliners"); for (case_dir, slug) in cases { - super::worker::update_oneliner(&case_dir, &slug, client, config, vocab, events_tx).await; + super::worker::update_oneliner(&case_dir, &slug, client, config, vocab, events_tx, locks) + .await; } } @@ -228,13 +231,15 @@ pub async fn regenerate_missing_oneliners_for_user( config: &Config, vocab: &Gazetteer, events_tx: &EventSender, + locks: &OnelinerLocks, ) { let cases = cases_needing_oneliner_in(user_root).await; if cases.is_empty() { return; } for case_dir in cases { - super::worker::update_oneliner(&case_dir, slug, client, config, vocab, events_tx).await; + super::worker::update_oneliner(&case_dir, slug, client, config, vocab, events_tx, locks) + .await; } } diff --git a/server/src/transcribe/worker.rs b/server/src/transcribe/worker.rs index 9ab701a..6cad094 100644 --- a/server/src/transcribe/worker.rs +++ b/server/src/transcribe/worker.rs @@ -12,6 +12,7 @@ use super::{TranscribeReceiver, ffmpeg, ollama, whisper}; use crate::config::{Config, WhisperUserSettings}; use crate::events::{self, CaseEventKind, EventSender}; use crate::gazetteer::Gazetteer; +use crate::oneliner_locks::OnelinerLocks; use crate::paths; use crate::{BusyGuard, WorkerBusy}; @@ -26,6 +27,7 @@ pub async fn run( worker_busy: WorkerBusy, vocab: Arc, events_tx: EventSender, + oneliner_locks: OnelinerLocks, ) { info!("Transcription worker started"); let timeout = Duration::from_secs(config.whisper_timeout_seconds); @@ -140,6 +142,7 @@ pub async fn run( &config, &vocab, &events_tx, + &oneliner_locks, ) .await; } @@ -213,14 +216,29 @@ async fn write_duration_sidecar(audio_path: &Path, probe_path: &Path) { /// stuck on "Generating" forever. /// - Still-`Pending` recordings present → no-op; the next transcript /// write re-enters this function with more information. -pub(crate) async fn update_oneliner( +pub async fn update_oneliner( case_dir: &Path, user_slug: &str, client: &reqwest::Client, config: &Config, vocab: &Gazetteer, events_tx: &EventSender, + locks: &OnelinerLocks, ) { + // Early-Latch: a manual override is sticky. Skip the LLM call + // entirely so the doctor's edit cannot be overwritten and Ollama + // does not waste a 60s round-trip on a result that will be dropped. + // Read-only check, no lock needed. + let (existing, _) = paths::read_oneliner_state(case_dir).await; + if matches!(existing, Some(OnelinerState::Manual { .. })) { + info!( + user = %user_slug, + case = %case_dir.display(), + "oneliner manual override is sticky — skip auto-regen" + ); + return; + } + let transcript = match all_transcripts_joined(case_dir).await { Some(t) => t, None => { @@ -237,6 +255,19 @@ pub(crate) async fn update_oneliner( .format(&Rfc3339) .unwrap_or_default(); let state = OnelinerState::Empty { generated_at }; + // Re-check under lock: a PUT may have arrived between + // the early-latch read and now (no LLM call here, but + // the silent-empty resolution still races with PUTs). + let _guard = locks.lock_for(case_dir).await; + let (current, _) = paths::read_oneliner_state(case_dir).await; + if matches!(current, Some(OnelinerState::Manual { .. })) { + info!( + user = %user_slug, + case = %case_dir.display(), + "manual override appeared during silent-empty resolution — drop" + ); + return; + } if let Err(e) = paths::write_oneliner_state(case_dir, &state).await { error!( case = %case_dir.display(), @@ -265,6 +296,10 @@ pub(crate) async fn update_oneliner( .format(&Rfc3339) .unwrap_or_default(); + // LLM call runs WITHOUT the lock — holding the per-case mutex for + // 60s would block any concurrent PUT for the same case. The + // post-LLM re-check (below) catches manual overrides that arrived + // during the call. let state = match ollama::generate_oneliner( client, &config.ollama_url, @@ -298,6 +333,21 @@ pub(crate) async fn update_oneliner( } }; + // Re-check under the per-case lock: if a manual override landed + // while the LLM was thinking, drop the auto result so the doctor's + // edit always wins. The PUT handler holds the same lock around its + // write, so the read-and-write below is atomic vs. that path. + let _guard = locks.lock_for(case_dir).await; + let (current, _) = paths::read_oneliner_state(case_dir).await; + if matches!(current, Some(OnelinerState::Manual { .. })) { + info!( + user = %user_slug, + case = %case_dir.display(), + "manual override appeared during regen — dropping LLM result" + ); + return; + } + if let Err(e) = paths::write_oneliner_state(case_dir, &state).await { error!(case = %case_dir.display(), error = %e, "writing oneliner.json failed"); return; diff --git a/server/tests/failed_only_case_empty_oneliner_test.rs b/server/tests/failed_only_case_empty_oneliner_test.rs index 41ec805..cf362b6 100644 --- a/server/tests/failed_only_case_empty_oneliner_test.rs +++ b/server/tests/failed_only_case_empty_oneliner_test.rs @@ -71,6 +71,7 @@ async fn failed_only_case_settles_to_empty_without_llm_call() { transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), transcribe_tx: tx_t, oneliner_heal_busy: OnelinerHealBusy(heal_busy.clone()), + oneliner_locks: doctate_server::oneliner_locks::OnelinerLocks::new(), }; pipeline diff --git a/server/tests/magic_link_test.rs b/server/tests/magic_link_test.rs index 7af3591..388696f 100644 --- a/server/tests/magic_link_test.rs +++ b/server/tests/magic_link_test.rs @@ -66,6 +66,7 @@ fn build_state_with_stores() -> AppState { analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), oneliner_heal_busy: OnelinerHealBusy(Arc::new(AtomicBool::new(false))), + oneliner_locks: doctate_server::oneliner_locks::OnelinerLocks::new(), events_tx: doctate_server::events::channel(), http_client: reqwest::Client::new(), vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()), diff --git a/server/tests/oneliner_heal_decoupled_test.rs b/server/tests/oneliner_heal_decoupled_test.rs index ae817a3..f5deae4 100644 --- a/server/tests/oneliner_heal_decoupled_test.rs +++ b/server/tests/oneliner_heal_decoupled_test.rs @@ -90,6 +90,7 @@ async fn heal_orphans_returns_fast_and_dedups_parallel_calls() { transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), transcribe_tx: tx_t, oneliner_heal_busy: OnelinerHealBusy(heal_busy.clone()), + oneliner_locks: doctate_server::oneliner_locks::OnelinerLocks::new(), }; // Two back-to-back heal invocations. The first wins the CAS and diff --git a/server/tests/oneliner_override_test.rs b/server/tests/oneliner_override_test.rs new file mode 100644 index 0000000..9d827f3 --- /dev/null +++ b/server/tests/oneliner_override_test.rs @@ -0,0 +1,402 @@ +//! Integration tests for the manual oneliner override: +//! `PUT /api/cases/{case_id}/oneliner`. +//! +//! Covers: +//! - Status codes for happy path / auth / validation errors. +//! - The two correctness invariants of the override: +//! 1. Early-latch: once a `Manual` state exists on disk, the +//! worker's `update_oneliner` skips the LLM call entirely. +//! 2. Re-check-under-lock: a manual PUT that arrives while an +//! auto-regen is in flight wins — the post-LLM result is +//! dropped because the per-case mutex serialises the +//! read-modify-write of `oneliner.json`. +//! - Mutex-serialisation: parallel PUTs on the same case produce a +//! well-formed final state, never a torn write. + +mod common; + +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use std::time::Duration; + +use axum::body::Body; +use axum::http::{Request, StatusCode, header}; +use serde_json::json; +use tower::util::ServiceExt; +use uuid::Uuid; +use wiremock::matchers::{method, path}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +use doctate_common::API_KEY_HEADER; +use doctate_common::oneliners::{ONELINER_FILENAME, OnelinerState}; +use doctate_server::oneliner_locks::OnelinerLocks; +use doctate_server::{ + AnalyzeBusy, AppState, OnelinerHealBusy, TranscribeBusy, analyze, transcribe, web_session, +}; + +use common::{TestConfig, body_json, seed_case, seed_oneliner_state, test_user}; + +const SLUG: &str = "dr_a"; +const API_KEY: &str = "key-dr_a"; + +fn put_request(case_id: &str, body: &str, api_key: Option<&str>) -> Request { + let mut b = Request::builder() + .method("PUT") + .uri(format!("/api/cases/{case_id}/oneliner")) + .header(header::CONTENT_TYPE, "application/json"); + if let Some(k) = api_key { + b = b.header(API_KEY_HEADER, k); + } + b.body(Body::from(body.to_owned())).unwrap() +} + +/// Build a full `AppState` with the given config; returns both the +/// state (for direct access to `oneliner_locks`, `events_tx`, etc.) +/// and a router that wraps it. +fn build_app(cfg: Arc) -> AppState { + let (transcribe_tx, _transcribe_rx) = transcribe::channel(); + let (analyze_tx, _analyze_rx) = analyze::channel(); + AppState { + config: cfg, + transcribe_tx, + analyze_tx, + session_store: web_session::new_store(), + magic_link_store: doctate_server::magic_link::new_store(), + analyze_busy: AnalyzeBusy(Arc::new(AtomicBool::new(false))), + transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), + oneliner_heal_busy: OnelinerHealBusy(Arc::new(AtomicBool::new(false))), + oneliner_locks: OnelinerLocks::new(), + events_tx: doctate_server::events::channel(), + http_client: reqwest::Client::new(), + vocab: Arc::new(doctate_server::gazetteer::Gazetteer::empty()), + } +} + +// ===================================================================== +// 1. Happy path +// ===================================================================== + +#[tokio::test] +async fn put_writes_manual_state_to_disk() { + let cfg = TestConfig::new() + .with_label("oneliner-override-ok") + .with_user(test_user(SLUG)) + .build(); + let case_id = Uuid::new_v4().to_string(); + let case_dir = seed_case(&cfg.data_path, SLUG, &case_id); + + let state = build_app(cfg); + let app = doctate_server::create_router_with_state(state); + + let resp = app + .oneshot(put_request( + &case_id, + r#"{"text":"55 J., Knieschmerz li."}"#, + Some(API_KEY), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + + let body = body_json(resp).await; + assert_eq!(body["kind"], "manual"); + assert_eq!(body["text"], "55 J., Knieschmerz li."); + assert!(body["set_at"].as_str().is_some_and(|s| !s.is_empty())); + + // On-disk state must match. + let bytes = tokio::fs::read(case_dir.join(ONELINER_FILENAME)) + .await + .expect("oneliner.json missing"); + let on_disk: OnelinerState = serde_json::from_slice(&bytes).unwrap(); + match on_disk { + OnelinerState::Manual { text, .. } => { + assert_eq!(text, "55 J., Knieschmerz li."); + } + other => panic!("expected Manual, got {other:?}"), + } +} + +// ===================================================================== +// 2-5. Error paths +// ===================================================================== + +#[tokio::test] +async fn put_unknown_case_returns_404() { + let cfg = TestConfig::new() + .with_label("oneliner-override-404") + .with_user(test_user(SLUG)) + .build(); + let unknown_case = Uuid::new_v4().to_string(); + let app = doctate_server::create_router_with_state(build_app(cfg)); + + let resp = app + .oneshot(put_request(&unknown_case, r#"{"text":"x"}"#, Some(API_KEY))) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); +} + +#[tokio::test] +async fn put_without_api_key_returns_401() { + let cfg = TestConfig::new() + .with_label("oneliner-override-401") + .with_user(test_user(SLUG)) + .build(); + let case_id = Uuid::new_v4().to_string(); + seed_case(&cfg.data_path, SLUG, &case_id); + let app = doctate_server::create_router_with_state(build_app(cfg)); + + let resp = app + .oneshot(put_request(&case_id, r#"{"text":"x"}"#, None)) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} + +#[tokio::test] +async fn put_empty_text_returns_400() { + let cfg = TestConfig::new() + .with_label("oneliner-override-empty") + .with_user(test_user(SLUG)) + .build(); + let case_id = Uuid::new_v4().to_string(); + seed_case(&cfg.data_path, SLUG, &case_id); + let app = doctate_server::create_router_with_state(build_app(cfg)); + + // Whitespace-only also collapses to empty after trim. + let resp = app + .oneshot(put_request( + &case_id, + r#"{"text":" \t\n"}"#, + Some(API_KEY), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn put_too_long_text_returns_400() { + let cfg = TestConfig::new() + .with_label("oneliner-override-long") + .with_user(test_user(SLUG)) + .build(); + let case_id = Uuid::new_v4().to_string(); + seed_case(&cfg.data_path, SLUG, &case_id); + let app = doctate_server::create_router_with_state(build_app(cfg)); + + let big = "a".repeat(501); + let body = json!({ "text": big }).to_string(); + let resp = app + .oneshot(put_request(&case_id, &body, Some(API_KEY))) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +// ===================================================================== +// 6. Early-latch: pre-existing Manual blocks LLM call +// ===================================================================== + +#[tokio::test] +async fn early_latch_skips_llm_call_when_manual_set() { + // Mock-Ollama with hit counter. After the manual-latched run we + // expect zero invocations. + let mock = MockServer::start().await; + let hits = Arc::new(AtomicUsize::new(0)); + let hits_for_mock = hits.clone(); + Mock::given(method("POST")) + .and(path("/api/chat")) + .respond_with(move |_req: &wiremock::Request| { + hits_for_mock.fetch_add(1, Ordering::Relaxed); + ResponseTemplate::new(200).set_body_json(json!({ + "message": {"content": "auto-generated"} + })) + }) + .mount(&mock) + .await; + + let cfg = TestConfig::new() + .with_label("oneliner-override-early-latch") + .with_user(test_user(SLUG)) + .with_ollama(mock.uri()) + .build(); + let case_id = Uuid::new_v4().to_string(); + let case_dir = seed_case(&cfg.data_path, SLUG, &case_id); + + // Pre-seed a Manual state directly on disk, simulating an earlier PUT. + seed_oneliner_state( + &case_dir, + &OnelinerState::Manual { + text: "manual-text".into(), + set_at: "2026-04-26T10:00:00Z".into(), + }, + ); + // Also seed a transcript so update_oneliner has content to pass to LLM + // (would call without latch). + std::fs::write(case_dir.join("2026-04-26T11-00-00Z.m4a"), b"audio-bytes").unwrap(); + std::fs::write( + case_dir.join("2026-04-26T11-00-00Z.transcript.txt"), + "Patient hat Fieber.", + ) + .unwrap(); + + let locks = OnelinerLocks::new(); + let events_tx = doctate_server::events::channel(); + let vocab = doctate_server::gazetteer::Gazetteer::empty(); + + transcribe::worker::update_oneliner( + &case_dir, + SLUG, + &reqwest::Client::new(), + &cfg, + &vocab, + &events_tx, + &locks, + ) + .await; + + assert_eq!( + hits.load(Ordering::Relaxed), + 0, + "Manual override must skip LLM entirely" + ); + + // Disk still has the manual state — untouched. + let bytes = tokio::fs::read(case_dir.join(ONELINER_FILENAME)) + .await + .unwrap(); + let on_disk: OnelinerState = serde_json::from_slice(&bytes).unwrap(); + assert!(matches!(on_disk, OnelinerState::Manual { .. })); +} + +// ===================================================================== +// 7. Race: manual PUT during in-flight LLM regen wins +// ===================================================================== + +#[tokio::test] +async fn re_check_drops_llm_result_when_manual_appears_during_regen() { + // Mock-Ollama with artificial delay so the test window is wide + // enough to inject a PUT before the LLM result lands. + let mock = MockServer::start().await; + Mock::given(method("POST")) + .and(path("/api/chat")) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(json!({ + "message": {"content": "auto-from-llm"} + })) + .set_delay(Duration::from_millis(500)), + ) + .mount(&mock) + .await; + + let cfg = TestConfig::new() + .with_label("oneliner-override-race") + .with_user(test_user(SLUG)) + .with_ollama(mock.uri()) + .build(); + let case_id = Uuid::new_v4().to_string(); + let case_dir = seed_case(&cfg.data_path, SLUG, &case_id); + + // Seed transcript so update_oneliner enters the LLM branch. + std::fs::write(case_dir.join("2026-04-26T11-00-00Z.m4a"), b"audio-bytes").unwrap(); + std::fs::write( + case_dir.join("2026-04-26T11-00-00Z.transcript.txt"), + "Patient klagt über Schwindel.", + ) + .unwrap(); + + let locks = OnelinerLocks::new(); + let events_tx = doctate_server::events::channel(); + + // Spawn the regen — it will wait ~500ms inside Mock-Ollama before + // reaching the re-check-under-lock branch. + let regen_cfg = cfg.clone(); + let regen_locks = locks.clone(); + let regen_events = events_tx.clone(); + let regen_dir = case_dir.clone(); + let regen = tokio::spawn(async move { + let vocab = doctate_server::gazetteer::Gazetteer::empty(); + transcribe::worker::update_oneliner( + ®en_dir, + SLUG, + &reqwest::Client::new(), + ®en_cfg, + &vocab, + ®en_events, + ®en_locks, + ) + .await; + }); + + // Give the regen a moment to enter the Ollama call, then write + // the manual override directly through the same lock — emulates + // the PUT handler. + tokio::time::sleep(Duration::from_millis(150)).await; + { + let _guard = locks.lock_for(&case_dir).await; + let manual = OnelinerState::Manual { + text: "doctor-wins".into(), + set_at: "2026-04-26T11:01:00Z".into(), + }; + doctate_server::paths::write_oneliner_state(&case_dir, &manual) + .await + .unwrap(); + } + + regen.await.unwrap(); + + // Final state must be the manual one — the auto result was dropped. + let bytes = tokio::fs::read(case_dir.join(ONELINER_FILENAME)) + .await + .unwrap(); + let on_disk: OnelinerState = serde_json::from_slice(&bytes).unwrap(); + match on_disk { + OnelinerState::Manual { text, .. } => assert_eq!(text, "doctor-wins"), + other => panic!("manual must win race, got {other:?}"), + } +} + +// ===================================================================== +// 8. Mutex serialises parallel PUTs +// ===================================================================== + +#[tokio::test] +async fn parallel_puts_serialize_via_mutex() { + let cfg = TestConfig::new() + .with_label("oneliner-override-parallel") + .with_user(test_user(SLUG)) + .build(); + let case_id = Uuid::new_v4().to_string(); + let case_dir = seed_case(&cfg.data_path, SLUG, &case_id); + + let app = doctate_server::create_router_with_state(build_app(cfg)); + + // Two concurrent PUTs on the same case. + let (r1, r2) = tokio::join!( + app.clone() + .oneshot(put_request(&case_id, r#"{"text":"first"}"#, Some(API_KEY),)), + app.oneshot(put_request(&case_id, r#"{"text":"second"}"#, Some(API_KEY),)), + ); + + assert_eq!(r1.unwrap().status(), StatusCode::OK); + assert_eq!(r2.unwrap().status(), StatusCode::OK); + + // Final on-disk state must be valid JSON of the Manual variant + // and contain exactly one of the two texts (no torn write, no + // mixed content). + let bytes = tokio::fs::read(case_dir.join(ONELINER_FILENAME)) + .await + .unwrap(); + let on_disk: OnelinerState = serde_json::from_slice(&bytes).unwrap(); + match on_disk { + OnelinerState::Manual { text, .. } => { + assert!( + text == "first" || text == "second", + "expected one of the two texts, got {text:?}" + ); + } + other => panic!("expected Manual, got {other:?}"), + } +} diff --git a/server/tests/silent_case_empty_test.rs b/server/tests/silent_case_empty_test.rs index e823f76..c534425 100644 --- a/server/tests/silent_case_empty_test.rs +++ b/server/tests/silent_case_empty_test.rs @@ -73,6 +73,7 @@ async fn silent_only_case_settles_to_empty_without_llm_call() { transcribe_busy: TranscribeBusy(Arc::new(AtomicBool::new(false))), transcribe_tx: tx_t, oneliner_heal_busy: OnelinerHealBusy(heal_busy.clone()), + oneliner_locks: doctate_server::oneliner_locks::OnelinerLocks::new(), }; pipeline diff --git a/server/tests/transcribe_test.rs b/server/tests/transcribe_test.rs index 23d812e..72d499b 100644 --- a/server/tests/transcribe_test.rs +++ b/server/tests/transcribe_test.rs @@ -312,6 +312,7 @@ async fn worker_renames_audio_to_failed_on_permanent_whisper_error() { busy, vocab, doctate_server::events::channel(), + doctate_server::oneliner_locks::OnelinerLocks::new(), ) .await; @@ -360,6 +361,7 @@ async fn worker_leaves_m4a_intact_on_transient_whisper_error() { busy, vocab, doctate_server::events::channel(), + doctate_server::oneliner_locks::OnelinerLocks::new(), ) .await; @@ -424,6 +426,7 @@ async fn transcribe_worker_normalizes_whisper_output() { busy, vocab, doctate_server::events::channel(), + doctate_server::oneliner_locks::OnelinerLocks::new(), ) .await; diff --git a/server/tests/transient_failure_retries_test.rs b/server/tests/transient_failure_retries_test.rs index 8ada14d..c6b7ded 100644 --- a/server/tests/transient_failure_retries_test.rs +++ b/server/tests/transient_failure_retries_test.rs @@ -84,6 +84,7 @@ async fn transient_whisper_failure_is_reenqueued_and_recovers() { // Spawn the transcribe worker in the background; it processes jobs // until the channel is closed at test teardown. + let oneliner_locks = doctate_server::oneliner_locks::OnelinerLocks::new(); let worker = tokio::spawn(transcribe::worker::run( rx_t, config.clone(), @@ -91,6 +92,7 @@ async fn transient_whisper_failure_is_reenqueued_and_recovers() { transcribe_busy.clone(), vocab.clone(), events_tx.clone(), + oneliner_locks.clone(), )); let pipeline = PipelineState { @@ -99,6 +101,7 @@ async fn transient_whisper_failure_is_reenqueued_and_recovers() { transcribe_busy: TranscribeBusy(transcribe_busy.clone()), transcribe_tx: tx_t.clone(), oneliner_heal_busy: OnelinerHealBusy(heal_busy.clone()), + oneliner_locks, }; // Act 1: enqueue the initial job, wait for the 503 failure.