doctate/server/src/analyze/worker.rs

use std::path::Path;
use std::sync::Arc;

use tokio::io::AsyncWriteExt;
use tracing::{error, info, warn};

use super::auto_trigger::FailureDetails;
use super::backend::{default_backend, find_backend};
use super::{
    ANALYSIS_INPUT_FILE, AnalysisInput, AnalyzeReceiver, DOCUMENT_FILE, auto_trigger, llm, prompt,
};
use crate::events::{self, CaseEventKind, EventSender};
use crate::gazetteer::Gazetteer;
use crate::{BusyGuard, WorkerBusy};

/// Consume analyze jobs sequentially. The external LLM tolerates parallelism,
/// but sequential processing keeps the architecture simple and shields the
/// provider from bursts. Trivial to lift later via `tokio::spawn(process(..))`.
pub async fn run(
    mut rx: AnalyzeReceiver,
    client: reqwest::Client,
    worker_busy: WorkerBusy,
    vocab: Arc<Gazetteer>,
    events_tx: EventSender,
) {
    info!(vocab_entries = vocab.len(), "Analyze worker started");

    while let Some(job) = rx.recv().await {
        let _guard = BusyGuard::new(worker_busy.clone());
        process(&job.case_dir, &client, &vocab, &events_tx).await;
    }

    warn!("Analyze worker stopped (channel closed)");
}

async fn process(
    case_dir: &Path,
    client: &reqwest::Client,
    vocab: &Gazetteer,
    events_tx: &EventSender,
) {
    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.
    if document_path.exists() {
        return;
    }

    let input: AnalysisInput = match tokio::fs::read(&input_path).await {
        Ok(bytes) => match serde_json::from_slice(&bytes) {
            Ok(v) => v,
            Err(e) => {
                error!(path = %input_path.display(), error = %e, "analysis input parse failed");
                return;
            }
        },
        Err(e) => {
            error!(path = %input_path.display(), error = %e, "analysis input read failed");
            return;
        }
    };

    // Silent-only case: no usable transcripts. Skip LLM, write a stub.
    if input.recordings.is_empty() {
        let stub_md = "_Keine verwertbaren Aufnahmen (alle Aufnahmen still)._\n";
        let doc = super::Document {
            covered_mtime: input.last_recording_mtime.clone(),
            written_at: doctate_common::timestamp::now_rfc3339(),
            backend_id: String::new(),
            content_md: stub_md.to_owned(),
        };
        let stub_bytes =
            serde_json::to_vec_pretty(&doc).expect("Document with String fields always serializes");
        if let Err(e) = write_atomic(&document_path, &stub_bytes).await {
            error!(path = %document_path.display(), error = %e, "writing stub document failed");
            let _ = auto_trigger::failure_marker(
                case_dir,
                &input.last_recording_mtime,
                format!("stub write: {e}"),
            )
            .write()
            .await;
            emit_analysis_failed(events_tx, case_dir);
            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,
            events::user_slug_of(case_dir),
            events::case_id_of(case_dir),
            CaseEventKind::DocumentReady,
        );
        return;
    }

    let user_content = prompt::render_prompt(&input);
    let total_chars = user_content.chars().count();
    let backend = find_backend(&input.backend_id).unwrap_or_else(default_backend);
    info!(
        case = %case_dir.display(),
        recording_count = input.recordings.len(),
        total_chars,
        backend_id = %backend.id,
        model = %backend.model_id,
        "sending to llm"
    );

    // NOTE: If the server crashes between a successful LLM response and the
    // document write below, the recovery scan will re-enqueue this job and
    // the call will be made again. Accepted cost — rare event, small
    // per-call price. Response-caching in a `.tmp` file would prevent it.
    let document = match llm::chat_once(client, backend, &user_content).await {
        Ok(text) => text,
        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(), backend_id = %backend.id, error = %e, "llm call failed");
            let mut details = FailureDetails::new().with_model_id(&backend.model_id);
            if let Some(d) = e.empty_content_details() {
                details = details.with_max_completion_tokens(d.max_completion_tokens);
                if let Some(fr) = d.finish_reason {
                    details = details.with_finish_reason(fr);
                }
                if let Some(v) = d.prompt_tokens {
                    details = details.with_prompt_tokens(v);
                }
                if let Some(v) = d.completion_tokens {
                    details = details.with_completion_tokens(v);
                }
                if let Some(v) = d.reasoning_tokens {
                    details = details.with_reasoning_tokens(v);
                }
            }
            let _ = auto_trigger::failure_marker(
                case_dir,
                &input.last_recording_mtime,
                format!("llm: {e}"),
            )
            .with_backend_id(&backend.id)
            .with_details(details)
            .write()
            .await;
            emit_analysis_failed(events_tx, case_dir);
            return;
        }
    };

    // Post-AI terminology normalization: enforce vocab-canonical forms
    // over the LLM's training priors (e.g. rewrite anglicized INNs like
    // `Amiodarone` back to `Amiodaron`).
    let document = vocab.replace(&document);

    // Silent-deletion guard: schema-aware backends can return formally valid
    // but empty envelopes (`{"document": ""}`) when the prompt is incompatible
    // with the format constraint. Without this guard the worker would persist
    // a Document with empty `content_md`, drop the failure marker, and the
    // user would see a blank note with no diagnostic. Observed on Llama 3.1
    // 405B FP8 (test T4, 2026-05-03) and previously on Mistral 24B.
    if document.trim().is_empty() {
        error!(
            case = %case_dir.display(),
            backend_id = %backend.id,
            "llm returned empty document — silent deletion guard triggered"
        );
        let _ = auto_trigger::failure_marker(
            case_dir,
            &input.last_recording_mtime,
            "llm returned empty document",
        )
        .with_backend_id(&backend.id)
        .with_details(FailureDetails::new().with_model_id(&backend.model_id))
        .write()
        .await;
        emit_analysis_failed(events_tx, case_dir);
        return;
    }

    let doc = super::Document {
        covered_mtime: input.last_recording_mtime.clone(),
        written_at: doctate_common::timestamp::now_rfc3339(),
        backend_id: backend.id.clone(),
        content_md: document.clone(),
    };
    let doc_bytes =
        serde_json::to_vec_pretty(&doc).expect("Document with String fields always serializes");
    if let Err(e) = write_atomic(&document_path, &doc_bytes).await {
        error!(path = %document_path.display(), error = %e, "writing document failed");
        let _ = auto_trigger::failure_marker(
            case_dir,
            &input.last_recording_mtime,
            format!("write document: {e}"),
        )
        .with_backend_id(&backend.id)
        .with_details(FailureDetails::new().with_model_id(&backend.model_id))
        .write()
        .await;
        emit_analysis_failed(events_tx, case_dir);
        return;
    }

    // Job done — drop the queue marker so the live "analyzing" flag
    // becomes accurate immediately and recovery on next boot does not
    // re-enqueue this version.
    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(),
        bytes = document.len(),
        "analysis done"
    );

    events::emit(
        events_tx,
        events::user_slug_of(case_dir),
        events::case_id_of(case_dir),
        CaseEventKind::DocumentReady,
    );
}

/// Notify subscribed browsers that an analysis attempt failed. Called
/// after the failure marker is on disk so a debounced reload sees the
/// fresh banner immediately instead of leaving the page on the
/// "Wird analysiert …" placeholder.
fn emit_analysis_failed(events_tx: &EventSender, case_dir: &Path) {
    events::emit(
        events_tx,
        events::user_slug_of(case_dir),
        events::case_id_of(case_dir),
        CaseEventKind::AnalysisFailed,
    );
}

/// Write `bytes` to `path` atomically: write to `<path>.tmp`, fsync, rename.
/// A crash between write and rename leaves no half-written target file — the
/// state machine simply sees the job as still pending and retries.
async fn write_atomic(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
    let tmp = path.with_extension({
        let mut ext = path
            .extension()
            .map(|s| s.to_os_string())
            .unwrap_or_default();
        ext.push(".tmp");
        ext
    });

    let mut file = tokio::fs::File::create(&tmp).await?;
    file.write_all(bytes).await?;
    file.sync_all().await?;
    drop(file);
    tokio::fs::rename(&tmp, path).await?;
    Ok(())
}