//! Opportunistic auto-trigger for LLM analysis.
//!
//! Called by the `/cases` and `/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::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::{AnalyzeJob, AnalyzeSender, DOCUMENT_FILE, InFlight};
use crate::events::{self, CaseEventKind, EventSender};
use crate::paths;
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,
    /// Id of the backend that failed (when known). Old markers without
    /// the field deserialize to `None`.
    #[serde(default)]
    pub backend_id: Option<String>,
    /// Structured diagnostics — populated for LLM failures where the
    /// provider response carries `finish_reason`/`usage`. Old markers
    /// without the field deserialize to `None` (no migration needed).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub details: Option<FailureDetails>,
}

/// Structured failure diagnostics. All fields optional so the same struct
/// works for empty-content failures (everything populated) and for HTTP /
/// transport errors (only `model_id` known).
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct FailureDetails {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub model_id: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub finish_reason: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub prompt_tokens: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub completion_tokens: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reasoning_tokens: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_completion_tokens: Option<u32>,
}

impl FailureDetails {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn with_model_id(mut self, v: impl Into<String>) -> Self {
        self.model_id = Some(v.into());
        self
    }

    pub fn with_finish_reason(mut self, v: impl Into<String>) -> Self {
        self.finish_reason = Some(v.into());
        self
    }

    pub fn with_prompt_tokens(mut self, v: u32) -> Self {
        self.prompt_tokens = Some(v);
        self
    }

    pub fn with_completion_tokens(mut self, v: u32) -> Self {
        self.completion_tokens = Some(v);
        self
    }

    pub fn with_reasoning_tokens(mut self, v: u32) -> Self {
        self.reasoning_tokens = Some(v);
        self
    }

    pub fn with_max_completion_tokens(mut self, v: u32) -> Self {
        self.max_completion_tokens = Some(v);
        self
    }

    fn is_empty(&self) -> bool {
        self.model_id.is_none()
            && self.finish_reason.is_none()
            && self.prompt_tokens.is_none()
            && self.completion_tokens.is_none()
            && self.reasoning_tokens.is_none()
            && self.max_completion_tokens.is_none()
    }
}

/// Outcome of the pre-condition check. `Skip` carries a static string so
/// the common "nothing to do" path allocates nothing.
///
/// Adapter over [`CaseAnalysisState`]: the auto-trigger reduces the rich
/// state machine to a binary "fire or skip" decision. UI layers that need
/// the full state should call [`evaluate_state`] directly.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AutoDecision {
    Enqueue,
    Skip(&'static str),
}

/// Per-case lifecycle state, computed from the artefacts on disk plus the
/// idle threshold. The variants below are precedence-ordered: the first
/// matching condition wins, so a `Failed` case with a stale failure
/// marker is reported as `Required`, not `Failed`.
///
/// `has_document: bool` on `Required`/`Idle` lets the UI decide whether
/// to render the previously-generated document (= „letztes Dokument")
/// alongside the „Analyse erforderlich"-banner.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum CaseAnalysisState {
    /// No `.m4a` files in the directory at all (e.g. fresh case before
    /// the first upload completes).
    NoRecordings,
    /// At least one recording is still missing its transcript sidecar.
    Pending,
    /// `analysis_input.json` exists — a job is queued or in flight.
    Queued,
    /// Failure marker matches the current `latest_mtime`. Stays `Failed`
    /// until either the input changes (→ `Required`) or the user wipes
    /// the marker via the manual retry path.
    Failed,
    /// Document does not yet cover the newest recording, but the
    /// idle-threshold has not elapsed yet — the user may still be
    /// dictating. Auto-trigger waits.
    Required { has_document: bool },
    /// Like `Required`, plus the idle window has elapsed. The auto-
    /// trigger fires.
    Idle { has_document: bool },
    /// Document on disk covers exactly the current recording set
    /// (`covered_mtime == latest_mtime`).
    Current,
}

/// Compute the full per-case lifecycle state. Pure observation, no
/// mutations or channel sends — safe to call redundantly per render.
///
/// `idle_threshold` is the time without new recordings after which a
/// stale-document case is auto-promoted from `Required` to `Idle`. Pass
/// `Duration::ZERO` to disable the wait (every Required becomes Idle
/// immediately — the historical eager behaviour, used by the
/// [`evaluate_case`] adapter and most existing tests).
///
/// `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.
///
/// `in_flight` is the process-local "queued or running" set. It replaces
/// the previous on-disk `analysis_input.json` marker, which survived
/// crashes and left cases stuck in `Queued` after a restart.
pub async fn evaluate_state(
    case_dir: &Path,
    idle_threshold: std::time::Duration,
    now: SystemTime,
    boot_time: SystemTime,
    in_flight: &InFlight,
) -> CaseAnalysisState {
    let scan = scan_m4as(case_dir).await;

    if !scan.saw_any {
        return CaseAnalysisState::NoRecordings;
    }
    if !scan.all_transcribed {
        return CaseAnalysisState::Pending;
    }
    let Some(latest_mtime) = scan.latest_mtime else {
        return CaseAnalysisState::NoRecordings;
    };

    // A claimed slot means a job is queued or in flight in this very
    // process. Empty on every server boot — no more phantom Queued.
    if in_flight.contains(case_dir) {
        return CaseAnalysisState::Queued;
    }

    let latest_rfc = rfc3339_of(latest_mtime);

    // Failure marker matching the current input signature → previous
    // attempt for *this* recording set failed, do not retry until the
    // input changes (or the user clears the marker manually).
    if let Some(marker) = read_failure_marker(case_dir).await
        && marker.last_recording_mtime == latest_rfc
    {
        return CaseAnalysisState::Failed;
    }

    let doc_meta = read_document_meta(case_dir).await;
    let has_document = doc_meta.is_some();

    if let Some(doc) = doc_meta
        && doc.covered_mtime == latest_rfc
    {
        return CaseAnalysisState::Current;
    }

    // Stale document (or no document at all): differentiate between
    // "user might still be dictating" (Required) and "long enough quiet,
    // 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(seen_at)
        .unwrap_or(std::time::Duration::ZERO);
    if elapsed >= idle_threshold {
        CaseAnalysisState::Idle { has_document }
    } else {
        CaseAnalysisState::Required { has_document }
    }
}

/// Read and parse `document.json` if present. Returns `None` for
/// missing files or parse failures (a corrupt envelope makes the case
/// appear as „has no document", which downgrades it to `Required` and
/// triggers a fresh analysis on the next idle window).
pub(crate) async fn read_document_meta(case_dir: &Path) -> Option<crate::analyze::Document> {
    let bytes = tokio::fs::read(case_dir.join(DOCUMENT_FILE)).await.ok()?;
    serde_json::from_slice(&bytes).ok()
}

/// Backwards-compatible binary decision: kept for the existing unit
/// tests that predate the typed state machine. New callers should
/// consume [`evaluate_state`] directly.
///
/// Reduction: `Idle` → `Enqueue`, everything else → `Skip(<tag>)`.
/// `Required` is also mapped to `Skip` here — but with
/// `Duration::ZERO` as the threshold, `Required` collapses into `Idle`
/// in practice, so the eager pre-refactor semantics are preserved.
///
/// Uses an empty in-flight set: tests that exercise the `Queued` branch
/// supply their own set via `evaluate_state` directly.
pub async fn evaluate_case(case_dir: &Path) -> AutoDecision {
    let state = evaluate_state(
        case_dir,
        std::time::Duration::ZERO,
        SystemTime::now(),
        SystemTime::UNIX_EPOCH,
        &InFlight::new(),
    )
    .await;
    match state {
        CaseAnalysisState::NoRecordings => AutoDecision::Skip("no recordings"),
        CaseAnalysisState::Pending => AutoDecision::Skip("not all transcribed"),
        CaseAnalysisState::Queued => AutoDecision::Skip("job queued"),
        CaseAnalysisState::Failed => AutoDecision::Skip("previously failed"),
        CaseAnalysisState::Current => AutoDecision::Skip("analysis current"),
        CaseAnalysisState::Idle { .. } => AutoDecision::Enqueue,
        CaseAnalysisState::Required { .. } => AutoDecision::Skip("waiting for idle"),
    }
}

/// Evaluate and, on `Idle`, 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).
///
/// `idle_threshold` and `now` flow through to [`evaluate_state`]: only
/// `Idle` (= stale document AND quiet window elapsed) fires. `Required`
/// returns `false` — the user-facing "Aktualisieren"-bulk endpoint is
/// the manual override path for those.
///
/// LLM-availability is *not* checked here — callers (web handlers) check
/// via `analyze::backend::any_backend_available()` before deciding to
/// render the analyze affordance at all. If the worker dequeues a job
/// without an available backend it will fail at the LLM call and write a
/// failure marker, which is graceful degradation rather than data loss.
pub async fn try_enqueue(
    case_dir: &Path,
    idle_threshold: std::time::Duration,
    now: SystemTime,
    boot_time: SystemTime,
    in_flight: &InFlight,
    tx: &AnalyzeSender,
    events_tx: &EventSender,
) -> bool {
    match evaluate_state(case_dir, idle_threshold, now, boot_time, in_flight).await {
        CaseAnalysisState::Idle { .. } => {}
        _ => return false,
    }

    // Atomic claim → race-guard against concurrent triggers. If another
    // caller (manual button, second tab) just grabbed the slot, skip
    // silently — they will send the job.
    if !in_flight.try_claim(case_dir) {
        return false;
    }

    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");
            in_flight.release(case_dir);
            return false;
        }
    };

    // The previous document.json stays in place: while the worker runs,
    // the UI keeps rendering the last successful analysis as "letztes
    // Dokument" with a Stale-banner. The worker overwrites the file
    // atomically once the new analysis completes.
    if tx
        .send(AnalyzeJob {
            case_dir: case_dir.to_path_buf(),
            input,
        })
        .is_err()
    {
        warn!(case = %case_dir.display(), "auto-analyze: send failed (worker gone)");
        in_flight.release(case_dir);
        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
/// `/cases` render — skips are cheap filesystem stats.
pub async fn try_enqueue_all_for_user(
    user_root: &Path,
    idle_threshold: std::time::Duration,
    now: SystemTime,
    boot_time: SystemTime,
    in_flight: &InFlight,
    tx: &AnalyzeSender,
    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_closed(&case_path).await {
            continue;
        }
        try_enqueue(
            &case_path,
            idle_threshold,
            now,
            boot_time,
            in_flight,
            tx,
            events_tx,
        )
        .await;
    }
}

/// Builder for a failure marker write. Use `failure_marker(...)` to start.
///
/// Fluent so the worker's call sites stay readable when carrying many
/// optional diagnostic fields (`finish_reason`, `usage`, `model_id`).
/// Consume the builder with `.write().await`.
pub struct FailureMarkerWriter<'a> {
    case_dir: &'a Path,
    last_recording_mtime: &'a str,
    reason: String,
    backend_id: Option<String>,
    details: FailureDetails,
}

impl<'a> FailureMarkerWriter<'a> {
    pub fn with_backend_id(mut self, id: impl Into<String>) -> Self {
        self.backend_id = Some(id.into());
        self
    }

    pub fn with_details(mut self, details: FailureDetails) -> Self {
        self.details = details;
        self
    }

    /// Persist atomically: write to `<path>.tmp`, fsync, rename.
    /// `last_recording_mtime` must equal the value read from the
    /// `AnalysisInput` so `evaluate_case` can later detect an unchanged
    /// input.
    pub async fn write(self) -> std::io::Result<()> {
        let marker = FailureMarker {
            last_recording_mtime: self.last_recording_mtime.to_owned(),
            reason: self.reason,
            failed_at: doctate_common::now_rfc3339(),
            backend_id: self.backend_id,
            details: if self.details.is_empty() {
                None
            } else {
                Some(self.details)
            },
        };
        let bytes = serde_json::to_vec_pretty(&marker)
            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
        let path = self.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
    }
}

/// Start a failure-marker write. The minimum a caller must supply is the
/// case directory, the input mtime (so `evaluate_case` can later detect
/// "input changed since failure"), and a human-readable reason. Backend
/// id and structured details are added fluently.
pub fn failure_marker<'a>(
    case_dir: &'a Path,
    last_recording_mtime: &'a str,
    reason: impl Into<String>,
) -> FailureMarkerWriter<'a> {
    FailureMarkerWriter {
        case_dir,
        last_recording_mtime,
        reason: reason.into(),
        backend_id: None,
        details: FailureDetails::default(),
    }
}

/// 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,
            });
        }

        // Analyze gate: "all transcribed" means every m4a has passed
        // through the transcriber, regardless of whether that produced
        // `Content` or `Silent` output. Only `Pending` (no sidecar)
        // holds the trigger back.
        if paths::read_transcript_state(&entry.path())
            .await
            .is_pending()
        {
            scan.all_transcribed = false;
        }
    }

    scan
}

/// Read and parse the per-case failure marker. `pub(crate)` so the web
/// layer can surface the failure state in the case page (otherwise a
/// transport-level LLM error leaves the user with no UI affordance to
/// retry — the auto-trigger sees the marker and skips, but no document
/// exists yet either, so the existing `Neu analysieren` button is not
/// rendered).
pub(crate) 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()
}

/// 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.
pub(crate) fn rfc3339_of(t: SystemTime) -> String {
    OffsetDateTime::from(t).format(&Rfc3339).unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::analyze::Document;
    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();
    }

    /// Seed a `document.json` whose `covered_mtime` matches the supplied
    /// `SystemTime`. Use `set_mtime` on `.m4a` files separately to drive
    /// the latest-mtime → covered-mtime comparison in [`evaluate_state`].
    async fn write_document_covering(case_dir: &std::path::Path, covered_mtime: SystemTime) {
        let covered_rfc = rfc3339_of(covered_mtime);
        let doc = Document {
            covered_mtime: covered_rfc.clone(),
            written_at: covered_rfc,
            backend_id: "test".into(),
            content_md: "doc".into(),
        };
        fs::write(
            case_dir.join(DOCUMENT_FILE),
            serde_json::to_vec(&doc).unwrap(),
        )
        .await
        .unwrap();
    }

    /// Seed a `<stem>.json` next to a (presumed already-touched)
    /// `<stem>.m4a`. Variant is `Transcript::Silent` — these tests
    /// only care that the recording is in *some* terminal state
    /// (transcribed), not what was actually said. Tests that need
    /// real content call `seed_meta_with` with the desired variant.
    async fn seed_meta(case_dir: &std::path::Path, stem: &str) {
        crate::paths::write_recording_meta_sync(
            case_dir,
            stem,
            doctate_common::Transcript::Silent,
            None,
        );
    }

    async fn seed_meta_with(case_dir: &std::path::Path, stem: &str, text: &str) {
        crate::paths::write_recording_meta_sync(
            case_dir,
            stem,
            doctate_common::Transcript::Content {
                text: text.to_owned(),
            },
            None,
        );
    }

    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;
        seed_meta(dir.path(), "2026-01-01T10-00-00Z").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;
        seed_meta(dir.path(), "2026-01-01T10-00-00Z").await;
        assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
    }

    #[tokio::test]
    async fn document_covered_equal_latest_returns_current() {
        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);
        write_document_covering(dir.path(), base).await;

        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Skip("analysis current")
        );
    }

    #[tokio::test]
    async fn document_covered_older_than_recordings_enqueues() {
        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 + Duration::from_secs(60));
        // Document covers an earlier snapshot — a recording arrived since.
        write_document_covering(dir.path(), base).await;

        assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
    }

    /// Burst race: when a recording arrives DURING the LLM call for an
    /// earlier recording, the worker writes a `document.json` whose
    /// `covered_mtime` snapshot only includes the recordings that were
    /// on disk when the analyze job started. A later-arriving recording
    /// (B) is reflected by a strictly larger `latest_mtime` on disk than
    /// the document's `covered_mtime`, so [`evaluate_state`] correctly
    /// returns `Required` (and `evaluate_case` enqueues, given a zero
    /// idle threshold).
    #[tokio::test]
    async fn recording_arrived_during_llm_call_re_enqueues() {
        let dir = TempDir::new().unwrap();

        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);

        // A: dictated and transcribed before the LLM call started.
        let a_m4a = dir.path().join("2026-01-01T10-00-00Z.m4a");
        touch(&a_m4a).await;
        seed_meta_with(dir.path(), "2026-01-01T10-00-00Z", "alpha").await;
        set_mtime(&a_m4a, base);

        // B: arrived 5s later, while A's LLM call was already in flight.
        let b_m4a = dir.path().join("2026-01-01T10-00-05Z.m4a");
        touch(&b_m4a).await;
        seed_meta_with(dir.path(), "2026-01-01T10-00-05Z", "bravo").await;
        set_mtime(&b_m4a, base + Duration::from_secs(5));

        // document.json: covers only A's snapshot. Wallclock-mtime of
        // the file itself would be misleading (later than B), but the
        // typed `covered_mtime` field pins the truth — the contained
        // markdown describes A only.
        write_document_covering(dir.path(), base).await;

        assert_eq!(
            evaluate_case(dir.path()).await,
            AutoDecision::Enqueue,
            "B was added during A's LLM call; the document cannot include it, so a re-analysis must be enqueued"
        );
    }

    /// `Queued` is now derived from the process-local `InFlight` set,
    /// not from a file on disk. This test exercises that path directly
    /// via `evaluate_state` (the `evaluate_case` adapter passes an empty
    /// set, which is exactly the point — orphan files cannot fake
    /// `Queued` any more).
    #[tokio::test]
    async fn in_flight_claim_yields_queued() {
        let dir = TempDir::new().unwrap();
        touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
        seed_meta(dir.path(), "2026-01-01T10-00-00Z").await;

        let in_flight = InFlight::new();
        assert!(in_flight.try_claim(dir.path()));

        let state = evaluate_state(
            dir.path(),
            std::time::Duration::ZERO,
            SystemTime::now(),
            SystemTime::UNIX_EPOCH,
            &in_flight,
        )
        .await;
        assert_eq!(state, CaseAnalysisState::Queued);
    }

    /// Orphan `analysis_input.json` left over from a pre-refactor server
    /// run must NOT register as `Queued` any more — that was the whole
    /// reason for the refactor.
    #[tokio::test]
    async fn orphan_input_file_does_not_count_as_queued() {
        let dir = TempDir::new().unwrap();
        touch(&dir.path().join("2026-01-01T10-00-00Z.m4a")).await;
        seed_meta_with(dir.path(), "2026-01-01T10-00-00Z", "patient mit fieber").await;
        // Pre-refactor leftover: a stale file the server used to write.
        fs::write(dir.path().join("analysis_input.json"), b"{}")
            .await
            .unwrap();

        assert_eq!(evaluate_case(dir.path()).await, AutoDecision::Enqueue);
    }

    #[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;
        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);

        let marker = FailureMarker {
            last_recording_mtime: rfc3339_of(base),
            reason: "test".into(),
            failed_at: "2026-01-01T10-05-00Z".into(),
            backend_id: None,
            details: None,
        };
        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;
        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);

        // 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(),
            backend_id: None,
            details: None,
        };
        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 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,
            &InFlight::new(),
        )
        .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,
            &InFlight::new(),
        )
        .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,
            &InFlight::new(),
        )
        .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,
            &InFlight::new(),
        )
        .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();
        let in_flight = InFlight::new();

        // 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,
            &in_flight,
            &tx,
            &events_tx,
        )
        .await;
        assert!(!sent_required, "Required state must not enqueue");
        assert!(
            rx.try_recv().is_err(),
            "channel must be empty after Required"
        );
        assert!(
            !in_flight.contains(dir.path()),
            "no claim should exist 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,
            &in_flight,
            &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");
        assert!(
            in_flight.contains(dir.path()),
            "claim must persist until the worker drops the slot guard"
        );
    }

    /// Concurrent triggers (auto-trigger + manual click in another tab)
    /// must produce **exactly one** job, not two. The atomic
    /// `try_claim` is the race-guard.
    #[tokio::test]
    async fn try_enqueue_is_idempotent_under_concurrent_triggers() {
        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();
        let in_flight = InFlight::new();

        let a = try_enqueue(
            dir.path(),
            Duration::from_secs(900),
            base + Duration::from_secs(1000),
            SystemTime::UNIX_EPOCH,
            &in_flight,
            &tx,
            &events_tx,
        )
        .await;
        let b = try_enqueue(
            dir.path(),
            Duration::from_secs(900),
            base + Duration::from_secs(1000),
            SystemTime::UNIX_EPOCH,
            &in_flight,
            &tx,
            &events_tx,
        )
        .await;
        assert!(a, "first call must enqueue");
        assert!(!b, "second call must skip — slot is already claimed");
        assert!(rx.try_recv().is_ok());
        assert!(rx.try_recv().is_err(), "exactly one job in the channel");
    }

    #[tokio::test]
    async fn try_enqueue_all_sends_only_eligible_cases() {
        use crate::analyze::channel as analyze_channel;
        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();
        seed_meta_with(
            &eligible,
            "2026-01-01T10-00-00Z",
            "patient mit brustschmerz",
        )
        .await;

        // Not eligible: document already covers the latest recording.
        let current_id = "22222222-2222-2222-2222-222222222222";
        let current = user_root.path().join(current_id);
        fs::create_dir_all(&current).await.unwrap();
        let m4a = current.join("2026-01-01T10-00-00Z.m4a");
        fs::write(&m4a, b"fake").await.unwrap();
        seed_meta_with(&current, "2026-01-01T10-00-00Z", "stable").await;
        let base = SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000);
        set_mtime(&m4a, base);
        write_document_covering(&current, base).await;

        // 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();
        seed_meta_with(&junk, "2026-01-01T10-00-00Z", "x").await;

        let (tx, mut rx) = analyze_channel();
        let events_tx = events_channel();
        let in_flight = InFlight::new();

        try_enqueue_all_for_user(
            user_root.path(),
            std::time::Duration::ZERO,
            SystemTime::now(),
            SystemTime::UNIX_EPOCH,
            &in_flight,
            &tx,
            &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()
        );
    }
}
