diff --git a/server/src/analyze/mod.rs b/server/src/analyze/mod.rs index 3caa88d..8d75b06 100644 --- a/server/src/analyze/mod.rs +++ b/server/src/analyze/mod.rs @@ -8,14 +8,18 @@ pub mod prompt; pub mod recovery; pub mod worker; -/// A single request to analyze a case. Payload is tiny (path + u32), so the -/// channel is unbounded — persistence lives in `analysis_input_v{N}.json` on +/// Single-document-per-case filenames. A re-analysis overwrites the +/// document in place (handler deletes it first; worker writes the fresh one). +pub const DOCUMENT_FILE: &str = "document.md"; +pub const ANALYSIS_INPUT_FILE: &str = "analysis_input.json"; + +/// A single request to analyze a case. Payload is tiny (a path), so the +/// channel is unbounded — persistence lives in `analysis_input.json` on /// disk, not in the queue. Recovery re-enqueues any inputs the server may /// have held at crash time. #[derive(Debug, Clone)] pub struct AnalyzeJob { pub case_dir: PathBuf, - pub version: u32, } pub type AnalyzeSender = mpsc::UnboundedSender; @@ -25,19 +29,14 @@ pub fn channel() -> (AnalyzeSender, AnalyzeReceiver) { mpsc::unbounded_channel() } -/// Persistent input for a single analysis run. Written by the close-case -/// handler, consumed by the analyze worker, and left on disk indefinitely as -/// an audit trail of what was sent to the LLM. -/// -/// Timestamps are stored as ISO-8601 strings (lexicographic order matches -/// chronological order, so string comparison is sufficient for the -/// "Nachtrag" detection the UI needs later). +/// Persistent input for a single analysis run. Written by the analyze +/// handler, consumed by the analyze worker, removed after a successful +/// document write. #[derive(Debug, Serialize, Deserialize)] pub struct AnalysisInput { - pub version: u32, /// Latest filesystem mtime (server clock) of any `.m4a` in the case at - /// the moment the close button was clicked. Used later to detect late - /// arrivals that should trigger a re-analysis prompt. + /// the moment the analyze button was clicked. Carried through for + /// diagnostics / future "Nachtrag" detection. pub last_recording_mtime: String, pub recordings: Vec, } diff --git a/server/src/analyze/prompt.rs b/server/src/analyze/prompt.rs index 6c21a43..af2904e 100644 --- a/server/src/analyze/prompt.rs +++ b/server/src/analyze/prompt.rs @@ -34,7 +34,6 @@ mod tests { fn mk(input: Vec<(&str, &str)>) -> AnalysisInput { AnalysisInput { - version: 1, last_recording_mtime: "2026-04-15T10:47:00Z".into(), recordings: input .into_iter() diff --git a/server/src/analyze/recovery.rs b/server/src/analyze/recovery.rs index 62a4903..738c1e9 100644 --- a/server/src/analyze/recovery.rs +++ b/server/src/analyze/recovery.rs @@ -2,7 +2,7 @@ use std::path::Path; use tracing::{info, warn}; -use super::{AnalyzeJob, AnalyzeSender}; +use super::{AnalyzeJob, AnalyzeSender, ANALYSIS_INPUT_FILE, DOCUMENT_FILE}; /// Walk `data_path/*/` and enqueue every pending analysis for every user. /// Intended to run once at startup; the same primitive backs the per-user @@ -18,10 +18,8 @@ pub async fn scan_and_enqueue(data_path: &Path, tx: &AnalyzeSender) { info!(count = total, "Analyze recovery scan finished"); } -/// Scan a single `//analysis_input_v*.json` set and enqueue -/// every input whose `document_v{N}.md` is missing. Returns the number sent. -/// Channel send is synchronous on the unbounded channel — only fails if the -/// worker is gone, which is a programming error. +/// Scan a user's case dirs and enqueue any case that has an +/// `analysis_input.json` but no `document.md`. Returns the number sent. pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> usize { let Ok(mut cases) = tokio::fs::read_dir(user_root).await else { return 0; @@ -32,64 +30,22 @@ pub async fn enqueue_pending_for_user(user_root: &Path, tx: &AnalyzeSender) -> u if !case_dir.is_dir() || crate::paths::is_deleted(&case_dir).await { continue; } - let Ok(mut files) = tokio::fs::read_dir(&case_dir).await else { + if !case_dir.join(ANALYSIS_INPUT_FILE).exists() { continue; - }; - while let Ok(Some(file)) = files.next_entry().await { - let path = file.path(); - let Some(name) = path.file_name().and_then(|s| s.to_str()) else { - continue; - }; - let Some(version) = parse_input_version(name) else { - continue; - }; - if case_dir.join(format!("document_v{version}.md")).exists() { - continue; - } - if tx - .send(AnalyzeJob { - case_dir: case_dir.clone(), - version, - }) - .is_err() - { - warn!(case = %case_dir.display(), "recovery send failed (worker gone)"); - return sent; - } - sent += 1; } + if case_dir.join(DOCUMENT_FILE).exists() { + continue; + } + if tx + .send(AnalyzeJob { + case_dir: case_dir.clone(), + }) + .is_err() + { + warn!(case = %case_dir.display(), "recovery send failed (worker gone)"); + return sent; + } + sent += 1; } sent } - -/// Parse `analysis_input_v{N}.json` and return `N` (as u32) if the pattern -/// matches exactly. Rejects malformed names, leading zeros, and overflow. -fn parse_input_version(filename: &str) -> Option { - let rest = filename.strip_prefix("analysis_input_v")?; - let num = rest.strip_suffix(".json")?; - if num.is_empty() || (num.starts_with('0') && num.len() > 1) { - return None; - } - num.parse::().ok() -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn parses_valid_names() { - assert_eq!(parse_input_version("analysis_input_v1.json"), Some(1)); - assert_eq!(parse_input_version("analysis_input_v42.json"), Some(42)); - } - - #[test] - fn rejects_invalid_names() { - assert_eq!(parse_input_version("analysis_input_v.json"), None); - assert_eq!(parse_input_version("analysis_input_v01.json"), None); - assert_eq!(parse_input_version("analysis_input_vx.json"), None); - assert_eq!(parse_input_version("document_v1.md"), None); - assert_eq!(parse_input_version("analysis_input_v1.txt"), None); - assert_eq!(parse_input_version(""), None); - } -} diff --git a/server/src/analyze/worker.rs b/server/src/analyze/worker.rs index 9ca1f7b..893124d 100644 --- a/server/src/analyze/worker.rs +++ b/server/src/analyze/worker.rs @@ -5,7 +5,7 @@ use std::time::Duration; use tokio::io::AsyncWriteExt; use tracing::{error, info, warn}; -use super::{llm, prompt, AnalysisInput, AnalyzeReceiver}; +use super::{llm, prompt, AnalysisInput, AnalyzeReceiver, ANALYSIS_INPUT_FILE, DOCUMENT_FILE}; use crate::config::Config; use crate::{BusyGuard, WorkerBusy}; @@ -23,7 +23,7 @@ pub async fn run( while let Some(job) = rx.recv().await { let _guard = BusyGuard::new(worker_busy.clone()); - process(&job.case_dir, job.version, &config, &client, timeout).await; + process(&job.case_dir, &config, &client, timeout).await; } warn!("Analyze worker stopped (channel closed)"); @@ -31,13 +31,12 @@ pub async fn run( async fn process( case_dir: &Path, - version: u32, config: &Config, client: &reqwest::Client, timeout: Duration, ) { - let input_path = case_dir.join(format!("analysis_input_v{version}.json")); - let document_path = case_dir.join(format!("document_v{version}.md")); + let input_path = case_dir.join(ANALYSIS_INPUT_FILE); + let document_path = case_dir.join(DOCUMENT_FILE); // If the document already exists, the job is obsolete (e.g. recovery // re-enqueued a finished case during a race). Skip silently. @@ -69,7 +68,7 @@ async fn process( if let Err(e) = tokio::fs::remove_file(&input_path).await { warn!(path = %input_path.display(), error = %e, "removing analysis input failed"); } - info!(case = %case_dir.display(), version, "analysis done (stub, no recordings)"); + info!(case = %case_dir.display(), "analysis done (stub, no recordings)"); return; } @@ -77,7 +76,6 @@ async fn process( let total_chars = user_content.chars().count(); info!( case = %case_dir.display(), - version, recording_count = input.recordings.len(), total_chars, "sending to llm" @@ -106,7 +104,7 @@ async fn process( Err(e) => { // Do not log the response body — LlmError::Display is already // redacted, but reinforce the rule here for future readers. - error!(case = %case_dir.display(), version, error = %e, "llm call failed"); + error!(case = %case_dir.display(), error = %e, "llm call failed"); return; } }; @@ -125,7 +123,6 @@ async fn process( info!( case = %case_dir.display(), - version, bytes = document.len(), "analysis done" ); diff --git a/server/src/routes/bulk.rs b/server/src/routes/bulk.rs index 6dbf072..3f7e656 100644 --- a/server/src/routes/bulk.rs +++ b/server/src/routes/bulk.rs @@ -8,12 +8,12 @@ use time::format_description::well_known::Rfc3339; use time::OffsetDateTime; use tracing::{info, warn}; -use crate::analyze::{AnalyzeJob, AnalyzeSender}; +use crate::analyze::{AnalyzeJob, AnalyzeSender, ANALYSIS_INPUT_FILE, DOCUMENT_FILE}; use crate::auth::AuthenticatedWebUser; use crate::config::Config; use crate::error::AppError; use crate::paths::{write_delete_marker, DeleteMarker}; -use crate::routes::case_actions::{build_analysis_input, find_latest_document, write_input_create_new}; +use crate::routes::case_actions::{build_analysis_input, write_input_create_new}; use crate::routes::user_web::locate_case; #[derive(Debug, Deserialize)] @@ -72,11 +72,19 @@ async fn bulk_analyze( skipped += 1; continue; }; - let version = find_latest_document(&case_dir) - .await - .map(|(v, _)| v + 1) - .unwrap_or(1); - let input = match build_analysis_input(&case_dir, version).await { + // Same order as the single-case handler: drop old document first so + // the UI doesn't keep showing "ausgewertet" during re-analysis. + let document_path = case_dir.join(DOCUMENT_FILE); + match tokio::fs::remove_file(&document_path).await { + Ok(_) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => { + warn!(slug = %slug, case_id = %case_id, error = %e, "bulk-analyze: remove old document failed"); + skipped += 1; + continue; + } + } + let input = match build_analysis_input(&case_dir).await { Ok(i) => i, Err(_) => { warn!(slug = %slug, case_id = %case_id, "bulk-analyze: build_analysis_input failed"); @@ -84,7 +92,7 @@ async fn bulk_analyze( continue; } }; - let input_path = case_dir.join(format!("analysis_input_v{version}.json")); + let input_path = case_dir.join(ANALYSIS_INPUT_FILE); if write_input_create_new(&input_path, &input).await.is_err() { warn!(slug = %slug, case_id = %case_id, "bulk-analyze: write input failed"); skipped += 1; @@ -93,7 +101,6 @@ async fn bulk_analyze( if analyze_tx .send(AnalyzeJob { case_dir: case_dir.clone(), - version, }) .is_err() { diff --git a/server/src/routes/case_actions.rs b/server/src/routes/case_actions.rs index a4c8331..f0df5ba 100644 --- a/server/src/routes/case_actions.rs +++ b/server/src/routes/case_actions.rs @@ -10,7 +10,9 @@ use time::OffsetDateTime; use tokio::io::AsyncWriteExt; use tracing::{info, warn}; -use crate::analyze::{AnalysisInput, AnalyzeJob, AnalyzeSender, RecordingInput}; +use crate::analyze::{ + AnalysisInput, AnalyzeJob, AnalyzeSender, RecordingInput, ANALYSIS_INPUT_FILE, DOCUMENT_FILE, +}; use crate::auth::AuthenticatedWebUser; use crate::config::Config; use crate::error::AppError; @@ -21,7 +23,6 @@ use crate::routes::user_web::locate_case; #[template(path = "document.html")] struct DocumentTemplate { case_id: String, - version: u32, content: String, } @@ -56,13 +57,18 @@ pub async fn handle_analyze_case( } }; - let version = find_latest_document(&case_dir) - .await - .map(|(v, _)| v + 1) - .unwrap_or(1); + // Re-analyze: delete the existing document first so the UI stops showing + // the stale "ausgewertet" state while the worker re-computes. Ignore + // NotFound (first-time analysis path). + let document_path = case_dir.join(DOCUMENT_FILE); + match tokio::fs::remove_file(&document_path).await { + Ok(_) => {} + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(AppError::Internal(format!("remove old document: {e}"))), + } - let input = build_analysis_input(&case_dir, version).await?; - let input_path = case_dir.join(format!("analysis_input_v{version}.json")); + let input = build_analysis_input(&case_dir).await?; + let input_path = case_dir.join(ANALYSIS_INPUT_FILE); write_input_create_new(&input_path, &input).await?; // Unbounded send: only fails if the worker has gone away (programmer error @@ -71,7 +77,6 @@ pub async fn handle_analyze_case( if analyze_tx .send(AnalyzeJob { case_dir: case_dir.clone(), - version, }) .is_err() { @@ -81,7 +86,6 @@ pub async fn handle_analyze_case( info!( slug = %user.slug, case_id = %case_id, - version, recordings = input.recordings.len(), "analyze requested; analysis enqueued" ); @@ -110,29 +114,22 @@ pub async fn handle_document_view( } }; - let (version, content) = find_latest_document(&case_dir) + let content = read_document(&case_dir) .await .ok_or_else(|| AppError::NotFound("Noch kein Dokument erzeugt".into()))?; - DocumentTemplate { - case_id, - version, - content, - } + DocumentTemplate { case_id, content } .render() .map(Html) .map_err(|e| AppError::Internal(format!("Template render failed: {e}"))) } -/// Build an `AnalysisInput` for the given case directory at the given version. +/// Build an `AnalysisInput` for the given case directory. /// Collects all `.m4a` (excluding `.m4a.failed`), reads the matching /// `.transcript.txt` for each, skips blank transcripts, and computes /// `last_recording_mtime` as the max mtime over ALL `.m4a` (including blank /// ones — a late blank addendum still counts as activity). -pub(crate) async fn build_analysis_input( - case_dir: &Path, - version: u32, -) -> Result { +pub(crate) async fn build_analysis_input(case_dir: &Path) -> Result { let m4as = collect_m4as(case_dir).await?; if m4as.is_empty() { return Err(AppError::BadRequest("Keine Aufnahmen im Fall".into())); @@ -170,7 +167,6 @@ pub(crate) async fn build_analysis_input( .map_err(|e| AppError::Internal(format!("format mtime: {e}")))?; Ok(AnalysisInput { - version, last_recording_mtime, recordings, }) @@ -242,36 +238,9 @@ fn filename_stem_to_recorded_at(stem: &str) -> String { } } -/// Scan `case_dir` for `document_v{N}.md`, pick the highest N, return its -/// content together with N. Returns `None` if no document exists. -pub(crate) async fn find_latest_document(case_dir: &Path) -> Option<(u32, String)> { - let mut highest: Option = None; - let mut entries = tokio::fs::read_dir(case_dir).await.ok()?; - while let Ok(Some(entry)) = entries.next_entry().await { - let name = entry.file_name(); - let Some(name_str) = name.to_str() else { continue }; - let Some(version) = parse_document_version(name_str) else { - continue; - }; - highest = match highest { - Some(h) if h >= version => Some(h), - _ => Some(version), - }; - } - let v = highest?; - let content = tokio::fs::read_to_string(case_dir.join(format!("document_v{v}.md"))) - .await - .ok()?; - Some((v, content)) -} - -fn parse_document_version(filename: &str) -> Option { - let rest = filename.strip_prefix("document_v")?; - let num = rest.strip_suffix(".md")?; - if num.is_empty() || (num.starts_with('0') && num.len() > 1) { - return None; - } - num.parse::().ok() +/// Read `case_dir/document.md`. Returns `None` if no document exists. +pub(crate) async fn read_document(case_dir: &Path) -> Option { + tokio::fs::read_to_string(case_dir.join(DOCUMENT_FILE)).await.ok() } /// POST /web/cases/{case_id}/delete @@ -437,18 +406,4 @@ mod tests { ); } - #[test] - fn parses_valid_document_names() { - assert_eq!(parse_document_version("document_v1.md"), Some(1)); - assert_eq!(parse_document_version("document_v42.md"), Some(42)); - } - - #[test] - fn rejects_invalid_document_names() { - assert_eq!(parse_document_version("document_v01.md"), None); - assert_eq!(parse_document_version("document_v.md"), None); - assert_eq!(parse_document_version("document_vx.md"), None); - assert_eq!(parse_document_version("analysis_input_v1.json"), None); - assert_eq!(parse_document_version(""), None); - } } diff --git a/server/src/routes/user_web.rs b/server/src/routes/user_web.rs index 978b5ad..8c3f294 100644 --- a/server/src/routes/user_web.rs +++ b/server/src/routes/user_web.rs @@ -9,7 +9,7 @@ use tracing::{info, warn}; use time::{format_description::well_known::Rfc3339, Date, OffsetDateTime, UtcOffset}; -use crate::analyze::{recovery as analyze_recovery, AnalyzeSender}; +use crate::analyze::{recovery as analyze_recovery, AnalyzeSender, ANALYSIS_INPUT_FILE, DOCUMENT_FILE}; use crate::auth::AuthenticatedWebUser; use crate::config::Config; use crate::error::AppError; @@ -173,39 +173,19 @@ async fn compute_flags( } } -/// True iff the case directory contains at least one `analysis_input_v*.json`. -/// The worker removes the input after successful document write, so existence -/// implies a pending or in-flight analyze job. +/// True iff `analysis_input.json` exists. The worker removes it after a +/// successful document write, so existence implies a pending or in-flight job. pub(crate) async fn any_analysis_input_exists(case_dir: &Path) -> bool { - let mut entries = match tokio::fs::read_dir(case_dir).await { - Ok(r) => r, - Err(_) => return false, - }; - while let Ok(Some(entry)) = entries.next_entry().await { - let name = entry.file_name(); - let Some(name_str) = name.to_str() else { continue }; - if name_str.starts_with("analysis_input_v") && name_str.ends_with(".json") { - return true; - } - } - false + tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE)) + .await + .unwrap_or(false) } -/// True iff the case directory contains at least one `document_v{N}.md`. -/// We don't care about N here — any version means "Ausgewertet" for the UI. +/// True iff `document.md` exists. pub(crate) async fn any_document_exists(case_dir: &Path) -> bool { - let mut entries = match tokio::fs::read_dir(case_dir).await { - Ok(r) => r, - Err(_) => return false, - }; - while let Ok(Some(entry)) = entries.next_entry().await { - let name = entry.file_name(); - let Some(name_str) = name.to_str() else { continue }; - if name_str.starts_with("document_v") && name_str.ends_with(".md") { - return true; - } - } - false + tokio::fs::try_exists(case_dir.join(DOCUMENT_FILE)) + .await + .unwrap_or(false) } /// Self-heal: if a worker is idle but orphans exist on disk for this user, diff --git a/server/templates/document.html b/server/templates/document.html index 94f80e4..2f50fdc 100644 --- a/server/templates/document.html +++ b/server/templates/document.html @@ -2,7 +2,7 @@ -Doctate — Dokument v{{ version }} +Doctate — Dokument