feat: Add auto-trigger for LLM analysis
Introduces a new module `auto_trigger` responsible for opportunistically initiating LLM analysis jobs. This mechanism scans case directories for new or updated recordings and transcripts, automatically creating and queuing analysis jobs when appropriate. Key components: - `evaluate_case`: Pure function to determine if a case is ready for analysis based on recording status, transcriptions, document modification times, and existing job/failure markers. - `try_enqueue`: Orchestrates the analysis job creation and queuing process if `evaluate_case` returns `AutoDecision::Enqueue`. - `try_enqueue_all_for_user`: Iterates through all user cases to trigger `try_enqueue` for eligible ones. - `write_failure_marker`/`remove_failure_marker`: Handles persistent recording of analysis failures and their cleanup. This feature aims to reduce manual intervention by automatically analyzing new or changed case data.
This commit is contained in:
@@ -0,0 +1,531 @@
|
||||
//! Opportunistic auto-trigger for LLM analysis.
|
||||
//!
|
||||
//! Called by the `/web/cases` and `/web/cases/{id}` handlers on every
|
||||
//! request. The SSE-driven `location.reload()` in the UI turns every
|
||||
//! relevant state change (upload, transcript, document) into a fresh
|
||||
//! handler invocation, so no background timer is needed.
|
||||
//!
|
||||
//! The hot path is a pure `evaluate_case` pre-condition check; the
|
||||
//! `try_enqueue` wrapper performs the actual side effects.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::Arc;
|
||||
use std::time::SystemTime;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use time::OffsetDateTime;
|
||||
use time::format_description::well_known::Rfc3339;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use tracing::warn;
|
||||
|
||||
use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeJob, AnalyzeSender, DOCUMENT_FILE};
|
||||
use crate::config::Config;
|
||||
use crate::events::{self, CaseEventKind, EventSender};
|
||||
use crate::routes::case_actions::build_analysis_input;
|
||||
|
||||
/// Filename of the per-case failure marker. Written by the worker on
|
||||
/// error, deleted on success. Presence with a matching
|
||||
/// `last_recording_mtime` tells us "already tried this exact input, do
|
||||
/// not retry until the input changes".
|
||||
pub const FAILURE_MARKER_FILE: &str = ".analysis_failed.json";
|
||||
|
||||
/// Persistent failure record. JSON so it is human-readable and carries
|
||||
/// a reason for post-mortem.
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
pub struct FailureMarker {
|
||||
pub last_recording_mtime: String,
|
||||
pub reason: String,
|
||||
pub failed_at: String,
|
||||
}
|
||||
|
||||
/// Outcome of the pre-condition check. `Skip` carries a static string so
|
||||
/// the common "nothing to do" path allocates nothing.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum AutoDecision {
|
||||
Enqueue,
|
||||
Skip(&'static str),
|
||||
}
|
||||
|
||||
/// Pure pre-condition check: decide whether this case is due for an
|
||||
/// automatic analysis. Does file I/O but no channel sends and no
|
||||
/// state mutations — safe to call redundantly.
|
||||
pub async fn evaluate_case(case_dir: &Path) -> AutoDecision {
|
||||
let scan = scan_m4as(case_dir).await;
|
||||
|
||||
if !scan.saw_any {
|
||||
return AutoDecision::Skip("no recordings");
|
||||
}
|
||||
if !scan.all_transcribed {
|
||||
return AutoDecision::Skip("not all transcribed");
|
||||
}
|
||||
let Some(latest_mtime) = scan.latest_mtime else {
|
||||
return AutoDecision::Skip("no recordings");
|
||||
};
|
||||
|
||||
// Any existing analysis_input.json means a job is queued or running.
|
||||
// Worker removes the file on success — next reload re-evaluates.
|
||||
if tokio::fs::try_exists(case_dir.join(ANALYSIS_INPUT_FILE))
|
||||
.await
|
||||
.unwrap_or(false)
|
||||
{
|
||||
return AutoDecision::Skip("job queued");
|
||||
}
|
||||
|
||||
// A failure marker matching the current input signature means we
|
||||
// already tried and the input has not changed since.
|
||||
if let Some(marker) = read_failure_marker(case_dir).await
|
||||
&& marker.last_recording_mtime == rfc3339_of(latest_mtime)
|
||||
{
|
||||
return AutoDecision::Skip("previously failed");
|
||||
}
|
||||
|
||||
// Fresh document older than the newest recording → stale, re-analyse.
|
||||
// Fresh document newer or equal → nothing to do.
|
||||
let doc_path = case_dir.join(DOCUMENT_FILE);
|
||||
if let Ok(meta) = tokio::fs::metadata(&doc_path).await
|
||||
&& let Ok(doc_mtime) = meta.modified()
|
||||
&& doc_mtime >= latest_mtime
|
||||
{
|
||||
return AutoDecision::Skip("analysis current");
|
||||
}
|
||||
|
||||
AutoDecision::Enqueue
|
||||
}
|
||||
|
||||
/// Evaluate and, on `Enqueue`, prepare and send the job. Returns `true`
|
||||
/// iff a job was actually sent. Silent on routine skips; `warn!` only on
|
||||
/// actual failures (build_input, write, send).
|
||||
pub async fn try_enqueue(
|
||||
case_dir: &Path,
|
||||
tx: &AnalyzeSender,
|
||||
config: &Arc<Config>,
|
||||
events_tx: &EventSender,
|
||||
) -> bool {
|
||||
if !config.llm_configured() {
|
||||
return false;
|
||||
}
|
||||
|
||||
match evaluate_case(case_dir).await {
|
||||
AutoDecision::Skip(_) => return false,
|
||||
AutoDecision::Enqueue => {}
|
||||
}
|
||||
|
||||
let input = match build_analysis_input(case_dir).await {
|
||||
Ok(i) => i,
|
||||
Err(e) => {
|
||||
warn!(case = %case_dir.display(), error = ?e, "auto-analyze: build input failed");
|
||||
return false;
|
||||
}
|
||||
};
|
||||
|
||||
// Re-analyse path: delete stale document.md up front so the worker's
|
||||
// "document exists → skip" guard does not short-circuit, and the UI
|
||||
// stops showing the outdated text during the render race.
|
||||
let doc_path = case_dir.join(DOCUMENT_FILE);
|
||||
if let Err(e) = remove_if_exists(&doc_path).await {
|
||||
warn!(case = %case_dir.display(), error = %e, "auto-analyze: remove old doc failed");
|
||||
return false;
|
||||
}
|
||||
|
||||
let input_path = case_dir.join(ANALYSIS_INPUT_FILE);
|
||||
if let Err(e) = write_input_overwrite(&input_path, &input).await {
|
||||
warn!(case = %case_dir.display(), error = %e, "auto-analyze: write input failed");
|
||||
return false;
|
||||
}
|
||||
|
||||
if tx
|
||||
.send(AnalyzeJob {
|
||||
case_dir: case_dir.to_path_buf(),
|
||||
})
|
||||
.is_err()
|
||||
{
|
||||
warn!(case = %case_dir.display(), "auto-analyze: send failed (worker gone)");
|
||||
return false;
|
||||
}
|
||||
|
||||
events::emit(
|
||||
events_tx,
|
||||
events::user_slug_of(case_dir),
|
||||
events::case_id_of(case_dir),
|
||||
CaseEventKind::AnalysisQueued,
|
||||
);
|
||||
true
|
||||
}
|
||||
|
||||
/// Iterate over every non-deleted UUID-named case directory under
|
||||
/// `user_root` and run `try_enqueue` on each. Safe to call on every
|
||||
/// `/web/cases` render — skips are cheap filesystem stats.
|
||||
pub async fn try_enqueue_all_for_user(
|
||||
user_root: &Path,
|
||||
tx: &AnalyzeSender,
|
||||
config: &Arc<Config>,
|
||||
events_tx: &EventSender,
|
||||
) {
|
||||
let Ok(mut entries) = tokio::fs::read_dir(user_root).await else {
|
||||
return;
|
||||
};
|
||||
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||
let Ok(name) = entry.file_name().into_string() else {
|
||||
continue;
|
||||
};
|
||||
if uuid::Uuid::parse_str(&name).is_err() {
|
||||
continue;
|
||||
}
|
||||
let case_path = entry.path();
|
||||
if !case_path.is_dir() {
|
||||
continue;
|
||||
}
|
||||
if crate::paths::is_deleted(&case_path).await {
|
||||
continue;
|
||||
}
|
||||
try_enqueue(&case_path, tx, config, events_tx).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// Persist a failure marker atomically. Called by the worker after a
|
||||
/// terminal failure. `last_recording_mtime` must equal the value read
|
||||
/// from the `AnalysisInput` so `evaluate_case` can later detect an
|
||||
/// unchanged input.
|
||||
pub async fn write_failure_marker(
|
||||
case_dir: &Path,
|
||||
last_recording_mtime: &str,
|
||||
reason: &str,
|
||||
) -> std::io::Result<()> {
|
||||
let marker = FailureMarker {
|
||||
last_recording_mtime: last_recording_mtime.to_owned(),
|
||||
reason: reason.to_owned(),
|
||||
failed_at: doctate_common::now_rfc3339(),
|
||||
};
|
||||
let bytes = serde_json::to_vec_pretty(&marker)
|
||||
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
|
||||
let path = case_dir.join(FAILURE_MARKER_FILE);
|
||||
let tmp = path.with_extension("json.tmp");
|
||||
let mut f = tokio::fs::File::create(&tmp).await?;
|
||||
f.write_all(&bytes).await?;
|
||||
f.sync_all().await?;
|
||||
drop(f);
|
||||
tokio::fs::rename(&tmp, &path).await
|
||||
}
|
||||
|
||||
/// Best-effort delete. Called by the worker after a successful run so
|
||||
/// the next `evaluate_case` does not mistake a resolved failure for a
|
||||
/// pending one.
|
||||
pub async fn remove_failure_marker(case_dir: &Path) {
|
||||
let _ = tokio::fs::remove_file(case_dir.join(FAILURE_MARKER_FILE)).await;
|
||||
}
|
||||
|
||||
struct M4aScan {
|
||||
saw_any: bool,
|
||||
all_transcribed: bool,
|
||||
latest_mtime: Option<SystemTime>,
|
||||
}
|
||||
|
||||
/// One-pass directory walk collecting everything `evaluate_case` needs.
|
||||
/// `.m4a.failed` entries are skipped because they did not contribute a
|
||||
/// transcript and will not contribute one in the future.
|
||||
async fn scan_m4as(case_dir: &Path) -> M4aScan {
|
||||
let mut scan = M4aScan {
|
||||
saw_any: false,
|
||||
all_transcribed: true,
|
||||
latest_mtime: None,
|
||||
};
|
||||
|
||||
let mut entries = match tokio::fs::read_dir(case_dir).await {
|
||||
Ok(r) => r,
|
||||
Err(_) => return scan,
|
||||
};
|
||||
|
||||
while let Ok(Some(entry)) = entries.next_entry().await {
|
||||
let Ok(name) = entry.file_name().into_string() else {
|
||||
continue;
|
||||
};
|
||||
if !name.ends_with(".m4a") {
|
||||
continue;
|
||||
}
|
||||
scan.saw_any = true;
|
||||
|
||||
if let Ok(meta) = entry.metadata().await
|
||||
&& let Ok(mtime) = meta.modified()
|
||||
{
|
||||
scan.latest_mtime = Some(match scan.latest_mtime {
|
||||
Some(prev) if prev >= mtime => prev,
|
||||
_ => mtime,
|
||||
});
|
||||
}
|
||||
|
||||
let transcript = entry.path().with_extension("transcript.txt");
|
||||
if !tokio::fs::try_exists(&transcript).await.unwrap_or(false) {
|
||||
scan.all_transcribed = false;
|
||||
}
|
||||
}
|
||||
|
||||
scan
|
||||
}
|
||||
|
||||
async fn read_failure_marker(case_dir: &Path) -> Option<FailureMarker> {
|
||||
let bytes = tokio::fs::read(case_dir.join(FAILURE_MARKER_FILE))
|
||||
.await
|
||||
.ok()?;
|
||||
serde_json::from_slice(&bytes).ok()
|
||||
}
|
||||
|
||||
async fn remove_if_exists(path: &Path) -> std::io::Result<()> {
|
||||
match tokio::fs::remove_file(path).await {
|
||||
Ok(_) => Ok(()),
|
||||
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
/// Overwrite-capable sibling of `case_actions::write_input_create_new`.
|
||||
/// 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.
|
||||
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");
|
||||
let mut f = tokio::fs::File::create(&tmp).await?;
|
||||
f.write_all(&bytes).await?;
|
||||
f.sync_all().await?;
|
||||
drop(f);
|
||||
tokio::fs::rename(&tmp, path).await
|
||||
}
|
||||
|
||||
/// Format an mtime in the same RFC3339 shape that
|
||||
/// `case_actions::build_analysis_input` writes, so equality comparison
|
||||
/// against a stored marker is string-exact.
|
||||
fn rfc3339_of(t: SystemTime) -> String {
|
||||
OffsetDateTime::from(t).format(&Rfc3339).unwrap_or_default()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use filetime::{FileTime, set_file_mtime};
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use tokio::fs;
|
||||
|
||||
async fn touch(path: &std::path::Path) {
|
||||
fs::write(path, b"").await.unwrap();
|
||||
}
|
||||
|
||||
fn set_mtime(path: &std::path::Path, t: SystemTime) {
|
||||
set_file_mtime(path, FileTime::from_system_time(t)).unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn empty_dir_skips() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
AutoDecision::Skip("no recordings")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_transcript_skips() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
|
||||
touch(&dir.path().join("2026-01-01T11-00-00Z.m4a")).await;
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
AutoDecision::Skip("not all transcribed")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn all_transcribed_no_document_enqueues() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
|
||||
assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn document_newer_than_recordings_skips() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
|
||||
touch(&m4a).await;
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
|
||||
let doc = dir.path().join(DOCUMENT_FILE);
|
||||
touch(&doc).await;
|
||||
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&m4a, base);
|
||||
set_mtime(&doc, base + Duration::from_secs(60));
|
||||
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
AutoDecision::Skip("analysis current")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn document_older_than_recordings_enqueues() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
|
||||
touch(&m4a).await;
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
|
||||
let doc = dir.path().join(DOCUMENT_FILE);
|
||||
touch(&doc).await;
|
||||
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&doc, base);
|
||||
set_mtime(&m4a, base + Duration::from_secs(60));
|
||||
|
||||
assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn analysis_input_present_skips() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
|
||||
touch(&dir.path().join(ANALYSIS_INPUT_FILE)).await;
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
AutoDecision::Skip("job queued")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn failure_marker_matching_mtime_skips() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
|
||||
touch(&m4a).await;
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
|
||||
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&m4a, base);
|
||||
|
||||
let marker = FailureMarker {
|
||||
last_recording_mtime: rfc3339_of(base),
|
||||
reason: "test".into(),
|
||||
failed_at: "2026-01-01T10-05-00Z".into(),
|
||||
};
|
||||
fs::write(
|
||||
dir.path().join(FAILURE_MARKER_FILE),
|
||||
serde_json::to_vec(&marker).unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
AutoDecision::Skip("previously failed")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn failure_marker_stale_mtime_enqueues() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
let m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
|
||||
touch(&m4a).await;
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.transcript.txt")).await;
|
||||
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&m4a, base);
|
||||
|
||||
// Marker records an older mtime → input has since changed.
|
||||
let marker = FailureMarker {
|
||||
last_recording_mtime: rfc3339_of(base - Duration::from_secs(3600)),
|
||||
reason: "test".into(),
|
||||
failed_at: "2025-12-31T23-00-00Z".into(),
|
||||
};
|
||||
fs::write(
|
||||
dir.path().join(FAILURE_MARKER_FILE),
|
||||
serde_json::to_vec(&marker).unwrap(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn m4a_failed_is_ignored() {
|
||||
let dir = TempDir::new().unwrap();
|
||||
// Only .m4a.failed present — counts as "no usable recordings".
|
||||
touch(&dir.path().join("2026-01-01T10-00-00Z.m4a.failed")).await;
|
||||
assert_eq!(
|
||||
evaluate_case(dir.path()).await,
|
||||
AutoDecision::Skip("no recordings")
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn try_enqueue_all_sends_only_eligible_cases() {
|
||||
use crate::analyze::channel as analyze_channel;
|
||||
use crate::config::Config;
|
||||
use crate::events::channel as events_channel;
|
||||
|
||||
let user_root = TempDir::new().unwrap();
|
||||
|
||||
// Eligible: all transcribed, no document, plausible content.
|
||||
let eligible_id = "11111111-1111-1111-1111-111111111111";
|
||||
let eligible = user_root.path().join(eligible_id);
|
||||
fs::create_dir_all(&eligible).await.unwrap();
|
||||
fs::write(eligible.join("2026-01-01T10-00-00Z.m4a"), b"fake")
|
||||
.await
|
||||
.unwrap();
|
||||
fs::write(
|
||||
eligible.join("2026-01-01T10-00-00Z.transcript.txt"),
|
||||
b"patient mit brustschmerz",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Not eligible: has a fresh document.
|
||||
let current_id = "22222222-2222-2222-2222-222222222222";
|
||||
let current = user_root.path().join(current_id);
|
||||
fs::create_dir_all(¤t).await.unwrap();
|
||||
let m4a = current.join("2026-01-01T10-00-00Z.m4a");
|
||||
fs::write(&m4a, b"fake").await.unwrap();
|
||||
fs::write(
|
||||
current.join("2026-01-01T10-00-00Z.transcript.txt"),
|
||||
b"stable",
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let doc = current.join(DOCUMENT_FILE);
|
||||
fs::write(&doc, b"doc").await.unwrap();
|
||||
let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
|
||||
set_mtime(&m4a, base);
|
||||
set_mtime(&doc, base + Duration::from_secs(60));
|
||||
|
||||
// Ignored: non-UUID directory name.
|
||||
let junk = user_root.path().join("not-a-uuid");
|
||||
fs::create_dir_all(&junk).await.unwrap();
|
||||
fs::write(junk.join("2026-01-01T10-00-00Z.m4a"), b"fake")
|
||||
.await
|
||||
.unwrap();
|
||||
fs::write(junk.join("2026-01-01T10-00-00Z.transcript.txt"), b"x")
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let (tx, mut rx) = analyze_channel();
|
||||
let events_tx = events_channel();
|
||||
let mut cfg = Config::test_default();
|
||||
cfg.llm_url = "http://localhost:9999".into();
|
||||
cfg.llm_model = "test".into();
|
||||
let cfg = Arc::new(cfg);
|
||||
|
||||
try_enqueue_all_for_user(user_root.path(), &tx, &cfg, &events_tx).await;
|
||||
|
||||
// Drop the sender half so recv closes after draining.
|
||||
drop(tx);
|
||||
let mut received = Vec::new();
|
||||
while let Some(job) = rx.recv().await {
|
||||
received.push(job.case_dir);
|
||||
}
|
||||
assert_eq!(received.len(), 1, "expected exactly one enqueued job");
|
||||
assert!(
|
||||
received[0].ends_with(eligible_id),
|
||||
"expected eligible case, got {}",
|
||||
received[0].display()
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ use std::path::PathBuf;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
pub mod auto_trigger;
|
||||
pub mod llm;
|
||||
pub mod prompt;
|
||||
pub mod recovery;
|
||||
|
||||
@@ -5,7 +5,9 @@ use std::time::Duration;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use tracing::{error, info, warn};
|
||||
|
||||
use super::{ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, llm, prompt};
|
||||
use super::{
|
||||
ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, auto_trigger, llm, prompt,
|
||||
};
|
||||
use crate::config::Config;
|
||||
use crate::events::{self, CaseEventKind, EventSender};
|
||||
use crate::gazetteer::Gazetteer;
|
||||
@@ -69,11 +71,18 @@ async fn process(
|
||||
let stub = b"_Keine verwertbaren Aufnahmen (alle Aufnahmen still)._\n";
|
||||
if let Err(e) = write_atomic(&document_path, stub).await {
|
||||
error!(path = %document_path.display(), error = %e, "writing stub document failed");
|
||||
let _ = auto_trigger::write_failure_marker(
|
||||
case_dir,
|
||||
&input.last_recording_mtime,
|
||||
&format!("stub write: {e}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
if let Err(e) = tokio::fs::remove_file(&input_path).await {
|
||||
warn!(path = %input_path.display(), error = %e, "removing analysis input failed");
|
||||
}
|
||||
auto_trigger::remove_failure_marker(case_dir).await;
|
||||
info!(case = %case_dir.display(), "analysis done (stub, no recordings)");
|
||||
events::emit(
|
||||
events_tx,
|
||||
@@ -117,6 +126,12 @@ async fn process(
|
||||
// Do not log the response body — LlmError::Display is already
|
||||
// redacted, but reinforce the rule here for future readers.
|
||||
error!(case = %case_dir.display(), error = %e, "llm call failed");
|
||||
let _ = auto_trigger::write_failure_marker(
|
||||
case_dir,
|
||||
&input.last_recording_mtime,
|
||||
&format!("llm: {e}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
@@ -128,6 +143,12 @@ async fn process(
|
||||
|
||||
if let Err(e) = write_atomic(&document_path, document.as_bytes()).await {
|
||||
error!(path = %document_path.display(), error = %e, "writing document failed");
|
||||
let _ = auto_trigger::write_failure_marker(
|
||||
case_dir,
|
||||
&input.last_recording_mtime,
|
||||
&format!("write document: {e}"),
|
||||
)
|
||||
.await;
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -137,6 +158,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");
|
||||
}
|
||||
auto_trigger::remove_failure_marker(case_dir).await;
|
||||
|
||||
info!(
|
||||
case = %case_dir.display(),
|
||||
|
||||
@@ -2,6 +2,7 @@ use axum::http::StatusCode;
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use serde_json::json;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum AppError {
|
||||
Unauthorized,
|
||||
BadRequest(String),
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
mod bulk;
|
||||
mod case_actions;
|
||||
pub(crate) mod case_actions;
|
||||
mod debug;
|
||||
mod events;
|
||||
mod health;
|
||||
|
||||
@@ -10,11 +10,14 @@ use tracing::{info, warn};
|
||||
use time::{Date, OffsetDateTime, format_description::well_known::Rfc3339};
|
||||
|
||||
use crate::PipelineState;
|
||||
use crate::analyze::{ANALYSIS_INPUT_FILE, DOCUMENT_FILE, recovery as analyze_recovery};
|
||||
use crate::analyze::{
|
||||
ANALYSIS_INPUT_FILE, DOCUMENT_FILE, auto_trigger, recovery as analyze_recovery,
|
||||
};
|
||||
use crate::auth::AuthenticatedWebUser;
|
||||
use crate::case_id::{CaseId, CaseIdPath};
|
||||
use crate::config::Config;
|
||||
use crate::error::AppError;
|
||||
use crate::events::EventSender;
|
||||
use crate::routes::case_actions::read_document;
|
||||
use crate::routes::web::{RecordingView, scan_recordings};
|
||||
use crate::transcribe::recovery as transcribe_recovery;
|
||||
@@ -228,9 +231,16 @@ pub async fn handle_my_cases(
|
||||
user: AuthenticatedWebUser,
|
||||
State(config): State<Arc<Config>>,
|
||||
State(pipeline): State<PipelineState>,
|
||||
State(events_tx): State<EventSender>,
|
||||
) -> Result<Html<String>, AppError> {
|
||||
let user_root = config.data_path.join(&user.slug);
|
||||
pipeline.heal_orphans_if_idle(&user_root, &user.slug).await;
|
||||
// Auto-analysis: hand every eligible case to the worker before we
|
||||
// render. The common case is "nothing to do" and costs a handful of
|
||||
// stat-calls per case; actual enqueues happen only when pre-conditions
|
||||
// flip (new transcripts in, stale document, ...).
|
||||
auto_trigger::try_enqueue_all_for_user(&user_root, &pipeline.analyze_tx, &config, &events_tx)
|
||||
.await;
|
||||
|
||||
let busy = pipeline.analyze_busy.0.load(Ordering::Acquire);
|
||||
let cases = scan_user_cases(&config, &user.slug, busy).await;
|
||||
@@ -263,6 +273,7 @@ pub async fn handle_case_page(
|
||||
user: AuthenticatedWebUser,
|
||||
State(config): State<Arc<Config>>,
|
||||
State(pipeline): State<PipelineState>,
|
||||
State(events_tx): State<EventSender>,
|
||||
CaseIdPath(case_id): CaseIdPath,
|
||||
) -> Result<Html<String>, AppError> {
|
||||
let user_root = config.data_path.join(&user.slug);
|
||||
@@ -275,6 +286,10 @@ pub async fn handle_case_page(
|
||||
"case page (possible IDOR probe)",
|
||||
)
|
||||
.await?;
|
||||
// Auto-analysis trigger for deep-link navigation: same evaluation as
|
||||
// the list view. Document render below still wins the race if the
|
||||
// worker happens to finish synchronously.
|
||||
auto_trigger::try_enqueue(&case_dir, &pipeline.analyze_tx, &config, &events_tx).await;
|
||||
let case_id_str = case_id.to_string();
|
||||
info!(slug = %user.slug, case_id = %case_id, "case page viewed");
|
||||
|
||||
|
||||
@@ -157,6 +157,10 @@ try {
|
||||
timer = setTimeout(() => location.reload(), 300);
|
||||
};
|
||||
const es = new EventSource('/web/events');
|
||||
// Release the socket-pool slot on navigation. Without this, Chrome
|
||||
// keeps the SSE connection "draining" after unload; six rapid nav
|
||||
// cycles exhaust the 6-per-origin pool and stall further requests.
|
||||
window.addEventListener('pagehide', () => es.close());
|
||||
es.addEventListener('case', (msg) => {
|
||||
let evt;
|
||||
try { evt = JSON.parse(msg.data); } catch (_) { return; }
|
||||
|
||||
@@ -275,6 +275,10 @@ try {
|
||||
}, 300);
|
||||
};
|
||||
const es = new EventSource('/web/events');
|
||||
// Release the socket-pool slot on navigation. Without this, Chrome
|
||||
// keeps the SSE connection "draining" after unload; six rapid nav
|
||||
// cycles exhaust the 6-per-origin pool and stall further requests.
|
||||
window.addEventListener('pagehide', () => es.close());
|
||||
es.addEventListener('case', (msg) => {
|
||||
let evt;
|
||||
try { evt = JSON.parse(msg.data); } catch (_) { return; }
|
||||
|
||||
@@ -162,6 +162,10 @@ try {
|
||||
timer = setTimeout(() => location.reload(), 300);
|
||||
};
|
||||
const es = new EventSource('/web/events');
|
||||
// Release the socket-pool slot on navigation. Without this, Chrome
|
||||
// keeps the SSE connection "draining" after unload; six rapid nav
|
||||
// cycles exhaust the 6-per-origin pool and stall further requests.
|
||||
window.addEventListener('pagehide', () => es.close());
|
||||
es.addEventListener('case', () => scheduleReload());
|
||||
})();
|
||||
</script>
|
||||
|
||||
Reference in New Issue
Block a user