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

Refactor analysis file names

Browse Source 
The versioning of analysis input and document files
(`analysis_input_v{N}.json`, `document_v{N}.md`) has been removed. All
analysis inputs will now use `analysis_input.json` and generated
documents will use `document.md`.

This simplifies file management, as there's no longer a need to track
and manage multiple versions of these files within a case directory. The
analysis worker will now overwrite the existing `document.md` if it
exists, ensuring that the latest analysis result is always present. The
`version` field has also been removed from `AnalyzeJob` and
`AnalysisInput`.
This commit is contained in:
Brummel
2026-04-16 01:56:55 +02:00
parent 928c0659c1
commit 8c6b2eeaa4
9 changed files with 111 additions and 281 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/src/analyze/mod.rs
+12 -13
View File
@@ -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<AnalyzeJob>;
@@ -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<RecordingInput>,
}
server/src/analyze/prompt.rs
-1
View File
@@ -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()
server/src/analyze/recovery.rs
+17 -61
View File
@@ -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 `<user_root>/<case>/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<u32> {
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::<u32>().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);
}
}
server/src/analyze/worker.rs
+6 -9
View File
@@ -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"
);