recall-echo 4.4.0

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.

//! Background entity extraction inside the graph daemon.
//!
//! recall-echo's claim is that the memory lifecycle is mechanical rather than
//! on the honor system. That was true of episodes — `SessionEnd` ingests them
//! through a hook — and false of everything built on top of them: entities,
//! relationships, confidence and provenance only appeared when a human
//! remembered to run `recall-echo graph extract`. This module closes that gap.
//!
//! The daemon is the natural home for the pass: it is already long-lived, it
//! already owns the embedded store, and it already knows when nobody is using
//! it. Once the machine has been quiet for `[extraction] idle_after_secs`, the
//! worker takes a small batch of un-extracted archives and runs the *same*
//! extraction the CLI runs — this module decides *when* extraction happens,
//! never *how*.
//!
//! Discipline:
//!
//! - **No lock is held across a batch.** The worker shares the daemon's
//!   `Arc<GraphMemory>` like any connection task; a request that arrives is
//!   served concurrently, and the batch stops at the next unit boundary.
//! - **Crash-only, per unit.** `extracted` flips one archive at a time, after
//!   that archive's extraction succeeded, so an interrupted batch leaves a
//!   store that simply has work left to do.
//! - **Shutdown wins.** Every wait and every unit runs under
//!   [`ShutdownSignal`], so an admin operation taking the store never waits for
//!   a batch to finish.

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;

use crate::config::ExtractionSection;
use crate::graph::error::GraphError;
use crate::graph::llm::LlmProvider;
use crate::graph::{GraphMemory, IngestContext};
use crate::serve::{BackgroundGuard, DaemonLog, ExtractionState, IdleTracker, ShutdownSignal};

/// Longest the worker sleeps between quiet checks.
const MAX_POLL: Duration = Duration::from_secs(30);
/// Shortest it sleeps, so a short `idle_after_secs` stays responsive without
/// spinning.
const MIN_POLL: Duration = Duration::from_millis(100);
/// Attempts one archive gets before it is quarantined. The CLI uses the same
/// rule: one retry, then set it aside rather than retry it forever.
const MAX_UNIT_ATTEMPTS: u32 = 2;
/// Consecutive failed units before the worker concludes the provider is wedged
/// and stops for the daemon's remaining life.
const MAX_CONSECUTIVE_FAILURES: u32 = 3;

// ── Clock ────────────────────────────────────────────────────────────────

/// Source of the current instant. Injectable so scheduling can be tested
/// without waiting for real time to pass.
pub trait Clock: Send + Sync {
    fn now(&self) -> Instant;
}

/// The real clock.
#[derive(Debug, Default, Clone, Copy)]
pub struct SystemClock;

impl Clock for SystemClock {
    fn now(&self) -> Instant {
        Instant::now()
    }
}

// ── The unit of work ─────────────────────────────────────────────────────

/// What one background batch did to one archive.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct UnitReport {
    pub entities: u32,
    pub relationships: u32,
    /// Steps that failed inside an archive that still yielded something. An
    /// archive where *every* step failed is not a report — it is an `Err`
    /// from [`ExtractionUnit::extract`], so it is retried and never marked.
    pub warnings: Vec<String>,
}

/// One archive's worth of extraction, as the scheduler sees it.
///
/// Kept behind a trait so the worker's scheduling — when it waits, when it
/// yields, when it stops — can be exercised without a store or a model.
#[async_trait]
pub trait ExtractionUnit: Send + Sync {
    /// Log numbers still awaiting extraction, at most `limit` of them.
    async fn pending(&self, limit: usize) -> Result<Vec<u32>, GraphError>;

    /// Extract one archive and mark it extracted. Marking is the last thing it
    /// does, so a failure or an interruption leaves the archive pending.
    async fn extract(&self, log_number: u32) -> Result<UnitReport, GraphError>;

    /// Stop offering this archive: it has failed [`MAX_UNIT_ATTEMPTS`] times.
    async fn quarantine(&self, log_number: u32, reason: &str);
}

// ── Schedule ─────────────────────────────────────────────────────────────

/// When and how much the worker extracts.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Schedule {
    /// Quiet period before a batch may start.
    pub idle_after: Duration,
    /// Archives per batch.
    pub batch_size: usize,
    /// How often the worker re-checks whether the daemon is quiet.
    pub poll_interval: Duration,
}

impl Schedule {
    /// The schedule described by an `[extraction]` section.
    #[must_use]
    pub fn from_config(config: &ExtractionSection) -> Self {
        let idle_after = config.idle_after();
        Self {
            idle_after,
            batch_size: config.effective_batch_size(),
            poll_interval: (idle_after / 4).clamp(MIN_POLL, MAX_POLL),
        }
    }
}

/// Whether background extraction should run at all in this daemon.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Plan {
    /// Run, on this schedule.
    Run(Schedule),
    /// Do not run, for this reason.
    Off(String),
}

/// Decide whether this daemon extracts in the background.
///
/// `graph_mode` is the `[graph] mode` of the memory directory. Server mode is
/// refused deliberately: there, clients talk to SurrealDB directly and never
/// start or consult a daemon, so a daemon's idea of "quiet" is not a fact
/// about the user's activity at all — it is a fact about a socket nobody is
/// connected to. A worker there would extract continuously, against a store
/// other processes are writing to, on behalf of a user who has no reason to
/// know the daemon exists. `graph extract` remains the way to extract in
/// server mode.
#[must_use]
pub fn plan(config: &ExtractionSection, graph_mode: &str) -> Plan {
    if !config.background_enabled {
        return Plan::Off("[extraction] background_enabled = false".into());
    }
    if graph_mode == "server" {
        return Plan::Off("[graph] mode = \"server\" — use `graph extract`".into());
    }
    Plan::Run(Schedule::from_config(config))
}

// ── Worker ───────────────────────────────────────────────────────────────

/// Everything the worker shares with the rest of the daemon.
pub struct WorkerContext {
    pub idle: Arc<IdleTracker>,
    pub shutdown: Arc<ShutdownSignal>,
    pub state: Arc<ExtractionState>,
    pub log: Arc<DaemonLog>,
    pub clock: Arc<dyn Clock>,
}

/// How a batch ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum BatchOutcome {
    /// Nothing to extract.
    NoWork,
    /// At least one archive was extracted.
    Worked,
    /// Shutdown arrived.
    Stopped,
    /// The provider has failed too many times in a row.
    Wedged,
}

/// The background extraction loop.
pub struct ExtractionWorker {
    schedule: Schedule,
    context: WorkerContext,
    /// Failed attempts per log number, within this daemon's life.
    attempts: HashMap<u32, u32>,
    /// Log numbers this daemon has given up on.
    skipped: HashSet<u32>,
    consecutive_failures: u32,
}

impl ExtractionWorker {
    #[must_use]
    pub fn new(schedule: Schedule, context: WorkerContext) -> Self {
        Self {
            schedule,
            context,
            attempts: HashMap::new(),
            skipped: HashSet::new(),
            consecutive_failures: 0,
        }
    }

    /// Extract in the background until the daemon shuts down.
    pub async fn run(mut self, unit: Arc<dyn ExtractionUnit>) {
        self.context.state.enable();
        loop {
            if self
                .context
                .shutdown
                .sleep_until_stopped(self.schedule.poll_interval)
                .await
            {
                break;
            }
            if !self.is_quiet() {
                continue;
            }
            match self.run_batch(unit.as_ref()).await {
                BatchOutcome::NoWork | BatchOutcome::Worked => {}
                BatchOutcome::Stopped => break,
                BatchOutcome::Wedged => {
                    let reason = format!(
                        "extraction failed {MAX_CONSECUTIVE_FAILURES} times in a row — \
                         not retrying until the daemon restarts"
                    );
                    self.context
                        .log
                        .log(&format!("background extraction off: {reason}"));
                    self.context.state.disable(reason);
                    return;
                }
            }
        }
        self.context.state.disable("daemon stopping");
    }

    fn is_quiet(&self) -> bool {
        self.context
            .idle
            .is_quiet_at(self.context.clock.now(), self.schedule.idle_after)
    }

    /// Extract up to `batch_size` archives, yielding between each.
    async fn run_batch(&mut self, unit: &dyn ExtractionUnit) -> BatchOutcome {
        let pending = match self.take_pending(unit).await {
            Some(pending) if !pending.is_empty() => pending,
            Some(_) => return BatchOutcome::NoWork,
            None => return BatchOutcome::Stopped,
        };

        // From here on the daemon may not idle-exit: the batch is in flight.
        let _busy = BackgroundGuard::new(Arc::clone(&self.context.idle));
        let started = self.context.clock.now();
        let mut extracted = 0u64;
        let mut outcome = BatchOutcome::Worked;

        for log_number in pending {
            if self.context.shutdown.is_triggered() {
                outcome = BatchOutcome::Stopped;
                break;
            }
            // A hot request beats background work, always. The rest of the
            // batch waits for the next quiet period.
            if self.context.idle.has_connections() {
                break;
            }

            match self.context.shutdown.guard(unit.extract(log_number)).await {
                None => {
                    outcome = BatchOutcome::Stopped;
                    break;
                }
                Some(Ok(report)) => {
                    extracted += 1;
                    self.consecutive_failures = 0;
                    self.attempts.remove(&log_number);
                    self.context.log.log(&format!(
                        "extracted log {log_number:03} in the background: \
                         +{} entities, {} relationships",
                        report.entities, report.relationships
                    ));
                    if let Some(first) = report.warnings.first() {
                        self.context.log.log(&format!(
                            "extraction warnings on log {log_number:03}: \
                             {} warning{}; first: {}",
                            report.warnings.len(),
                            if report.warnings.len() == 1 { "" } else { "s" },
                            one_line(first),
                        ));
                    }
                }
                Some(Err(err)) => {
                    if self.record_failure(unit, log_number, &err).await {
                        outcome = BatchOutcome::Wedged;
                        break;
                    }
                }
            }

            // Hand the runtime back between units so a request that arrived
            // mid-batch is picked up before the next archive starts.
            tokio::task::yield_now().await;
        }

        self.finish_batch(extracted, started);
        outcome
    }

    /// Log numbers worth attempting, or `None` when shutdown arrived.
    async fn take_pending(&mut self, unit: &dyn ExtractionUnit) -> Option<Vec<u32>> {
        // Over-fetch, then drop what this daemon has given up on, so a
        // quarantined archive cannot occupy a batch slot forever.
        let limit = self.schedule.batch_size + self.skipped.len();
        match self.context.shutdown.guard(unit.pending(limit)).await? {
            Ok(pending) => Some(
                pending
                    .into_iter()
                    .filter(|log_number| !self.skipped.contains(log_number))
                    .take(self.schedule.batch_size)
                    .collect(),
            ),
            Err(err) => {
                self.context
                    .log
                    .log(&format!("background extraction: cannot list work: {err}"));
                self.context.state.record_error(err.to_string());
                Some(Vec::new())
            }
        }
    }

    /// Account for a failed archive. `true` means the provider looks wedged.
    async fn record_failure(
        &mut self,
        unit: &dyn ExtractionUnit,
        log_number: u32,
        err: &GraphError,
    ) -> bool {
        self.consecutive_failures += 1;
        let attempts = self.attempts.entry(log_number).or_insert(0);
        *attempts += 1;

        self.context.state.record_error(err.to_string());
        if *attempts >= MAX_UNIT_ATTEMPTS {
            self.skipped.insert(log_number);
            unit.quarantine(log_number, &err.to_string()).await;
            self.context.log.log(&format!(
                "background extraction quarantined log {log_number:03} after \
                 {MAX_UNIT_ATTEMPTS} attempts: {err}"
            ));
        } else {
            self.context.log.log(&format!(
                "background extraction failed on log {log_number:03}: {err}"
            ));
        }

        self.consecutive_failures >= MAX_CONSECUTIVE_FAILURES
    }

    /// Record what the batch did — and only then let the idle clock restart,
    /// so a daemon with a backlog stays alive and one without it does not.
    fn finish_batch(&self, extracted: u64, started: Instant) {
        if extracted == 0 {
            return;
        }
        let finished = self.context.clock.now();
        let elapsed = finished.saturating_duration_since(started);
        self.context
            .state
            .record_batch(extracted, elapsed, finished);
        self.context.idle.touch_at(Instant::now());
        self.context.log.log(&format!(
            "background extraction: {extracted} archives in {}ms",
            elapsed.as_millis()
        ));
    }
}

// ── The real unit: an archive on disk, a store, a provider ───────────────

/// Extraction against the daemon's own store, with a provider built from
/// config. Runs exactly what `recall-echo graph extract` runs.
pub struct GraphExtractionUnit {
    graph: Arc<GraphMemory>,
    llm: Box<dyn LlmProvider>,
    conversations_dir: PathBuf,
    quarantine_path: PathBuf,
}

impl GraphExtractionUnit {
    #[must_use]
    pub fn new(
        graph: Arc<GraphMemory>,
        llm: Box<dyn LlmProvider>,
        conversations_dir: PathBuf,
        quarantine_path: PathBuf,
    ) -> Self {
        Self {
            graph,
            llm,
            conversations_dir,
            quarantine_path,
        }
    }
}

#[async_trait]
impl ExtractionUnit for GraphExtractionUnit {
    async fn pending(&self, limit: usize) -> Result<Vec<u32>, GraphError> {
        let quarantined = read_quarantine(&self.quarantine_path);
        Ok(self
            .graph
            .unextracted_log_numbers()
            .await?
            .into_iter()
            .filter_map(|log_number| u32::try_from(log_number).ok())
            .filter(|log_number| !quarantined.contains(log_number))
            .take(limit)
            .collect())
    }

    async fn extract(&self, log_number: u32) -> Result<UnitReport, GraphError> {
        let path = crate::graph_cli::find_archive_file(&self.conversations_dir, log_number)
            .map_err(|err| GraphError::NotFound(err.to_string()))?;
        let content = std::fs::read_to_string(&path)?;
        let (session_id, _) = crate::graph_cli::extract_archive_metadata(&content, &path);
        let context = IngestContext::new(session_id, Some(log_number));

        let report = self
            .graph
            .extract_from_archive(&content, &context, self.llm.as_ref())
            .await?;
        // Decide before marking: a total failure is an Err and stays pending.
        let outcome = unit_outcome(report)?;
        self.graph.mark_extracted(log_number).await?;
        Ok(outcome)
    }

    async fn quarantine(&self, log_number: u32, _reason: &str) {
        use std::io::Write as _;

        // Best effort. The worker's in-memory skip list already holds for this
        // daemon's life, so a failed write costs one retry after a restart.
        let _ = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.quarantine_path)
            .and_then(|mut file| writeln!(file, "{log_number:03}"));
    }
}

/// What an archive's report means to the worker.
///
/// A total failure — every chunk's extraction call failed — is the
/// provider's failure, not an empty archive: it becomes an `Err`, so the
/// caller never marks the archive and the worker retries, quarantines, and
/// — if it keeps happening — says so and stops. Anything else is a report,
/// carrying the steps that failed as warnings.
fn unit_outcome(report: crate::graph::types::IngestionReport) -> Result<UnitReport, GraphError> {
    if report.is_total_failure() {
        return Err(GraphError::Llm(all_failed_message(&report)));
    }
    Ok(UnitReport {
        entities: report.entities_created + report.entities_merged,
        relationships: report.relationships_created,
        warnings: report.errors,
    })
}

/// The one line the daemon log gets for an archive whose every chunk failed.
/// Names the count and the first error, which is the one that says what is
/// wrong with the provider.
fn all_failed_message(report: &crate::graph::types::IngestionReport) -> String {
    let first = report.errors.first().map_or("", String::as_str);
    format!(
        "all {} extraction chunk{} failed, nothing extracted; first: {}",
        report.chunks_failed,
        if report.chunks_failed == 1 { "" } else { "s" },
        one_line(first),
    )
}

/// Provider stderr goes into the daemon log and `graph status`: keep it on
/// one line, free of control characters, and short. A CLI that prints a
/// newline or an escape sequence must not forge a log line or repaint the
/// terminal of whoever reads the status.
fn one_line(text: &str) -> String {
    const MAX: usize = 300;
    let mut out: String = text
        .chars()
        .map(|c| if c.is_control() && c != '\t' { ' ' } else { c })
        .collect();
    if out.chars().count() > MAX {
        out = out.chars().take(MAX).collect::<String>() + "…";
    }
    out
}

/// Log numbers a previous run set aside. Absent or unreadable means none.
fn read_quarantine(path: &Path) -> HashSet<u32> {
    std::fs::read_to_string(path)
        .unwrap_or_default()
        .lines()
        .filter_map(|line| line.trim().parse().ok())
        .collect()
}

// ── Wiring ───────────────────────────────────────────────────────────────

/// Everything [`spawn`] needs from the daemon.
pub struct Setup {
    pub memory_dir: PathBuf,
    pub graph: Arc<GraphMemory>,
    pub idle: Arc<IdleTracker>,
    pub shutdown: Arc<ShutdownSignal>,
    pub state: Arc<ExtractionState>,
    pub log: Arc<DaemonLog>,
}

/// Start the background worker for a daemon, if it should have one.
///
/// Every refusal is quiet, final and stated once: a missing provider, a
/// missing key, a missing conversations directory and a config opt-out all
/// end here, with a line in the daemon log and a reason on the wire — never a
/// retry loop, and never surprise API spend.
pub fn spawn(setup: Setup) -> Option<tokio::task::JoinHandle<()>> {
    let config = crate::config::load_from_dir(&setup.memory_dir);
    let mode = crate::serve_client::graph_mode(&setup.memory_dir);

    let schedule = match plan(&config.extraction, &mode) {
        Plan::Run(schedule) => schedule,
        Plan::Off(reason) => return refuse(&setup, &reason),
    };

    let conversations_dir = match crate::graph_cli::find_conversations_dir(&setup.memory_dir) {
        Ok(dir) => dir,
        Err(err) => return refuse(&setup, &format!("no archives to extract ({err})")),
    };

    // The daemon is started with an allowlisted environment that deliberately
    // excludes API keys, so an API-key provider fails here — which is the
    // point: an auto-started daemon can only spend what a subscription already
    // covers. Running `serve --foreground` with a key exported is the explicit
    // way to opt an API provider in.
    let handle =
        match crate::llm_provider::create_provider_with_binary(&setup.memory_dir, None, None) {
            Ok(handle) => handle,
            Err(err) => return refuse(&setup, &format!("no usable LLM provider ({err})")),
        };
    let (llm, model, binary) = (handle.llm, handle.model, handle.binary);

    if let Some(timeout) = setup.idle.timeout() {
        if schedule.idle_after >= timeout {
            setup.log.log(&format!(
                "warning: [extraction] idle_after_secs ({}s) is not shorter than \
                 [serve] idle_timeout_secs ({}s) — the daemon exits before it extracts",
                schedule.idle_after.as_secs(),
                timeout.as_secs()
            ));
        }
    }

    let binary = binary.map_or_else(String::new, |p| format!(" ({})", p.display()));
    setup.log.log(&format!(
        "background extraction on: {} provider{binary}, model {}, every {}s of quiet, {} archives per batch",
        config.llm.provider,
        if model.is_empty() { "default" } else { &model },
        schedule.idle_after.as_secs(),
        schedule.batch_size,
    ));

    let unit: Arc<dyn ExtractionUnit> = Arc::new(GraphExtractionUnit::new(
        Arc::clone(&setup.graph),
        llm,
        conversations_dir,
        setup
            .memory_dir
            .join("graph")
            .join("extraction-quarantine.txt"),
    ));
    let worker = ExtractionWorker::new(
        schedule,
        WorkerContext {
            idle: setup.idle,
            shutdown: setup.shutdown,
            state: setup.state,
            log: setup.log,
            clock: Arc::new(SystemClock),
        },
    );
    Some(tokio::spawn(worker.run(unit)))
}

/// Say once why no background extraction runs, and run none.
fn refuse(setup: &Setup, reason: &str) -> Option<tokio::task::JoinHandle<()>> {
    setup
        .log
        .log(&format!("background extraction off: {reason}"));
    setup.state.disable(reason);
    None
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    fn config(background_enabled: bool) -> ExtractionSection {
        ExtractionSection {
            background_enabled,
            ..ExtractionSection::default()
        }
    }

    #[test]
    fn the_default_plan_runs_on_the_configured_schedule() {
        let Plan::Run(schedule) = plan(&config(true), "embedded") else {
            panic!("the default config must run");
        };
        assert_eq!(schedule.idle_after, Duration::from_secs(120));
        assert_eq!(schedule.batch_size, 3);
        assert_eq!(schedule.poll_interval, Duration::from_secs(30));
    }

    #[test]
    fn opting_out_turns_the_worker_off() {
        let Plan::Off(reason) = plan(&config(false), "embedded") else {
            panic!("background_enabled = false must be honored");
        };
        assert!(reason.contains("background_enabled"), "{reason}");
    }

    #[test]
    fn server_mode_never_extracts_in_the_background() {
        let Plan::Off(reason) = plan(&config(true), "server") else {
            panic!("server mode has no daemon to schedule against");
        };
        assert!(reason.contains("server"), "{reason}");
    }

    #[test]
    fn poll_interval_is_bounded_at_both_ends() {
        let fast = Schedule::from_config(&ExtractionSection {
            idle_after_secs: 0,
            ..ExtractionSection::default()
        });
        assert_eq!(fast.poll_interval, MIN_POLL);

        let slow = Schedule::from_config(&ExtractionSection {
            idle_after_secs: 86_400,
            ..ExtractionSection::default()
        });
        assert_eq!(slow.poll_interval, MAX_POLL);
    }

    #[test]
    fn a_batch_of_zero_would_never_extract_so_it_is_one() {
        let schedule = Schedule::from_config(&ExtractionSection {
            batch_size: 0,
            ..ExtractionSection::default()
        });
        assert_eq!(schedule.batch_size, 1);
    }

    // ── Outcome of a report ──────────────────────────────────────────

    fn report_with(
        total: u32,
        failed: u32,
        errors: &[&str],
    ) -> crate::graph::types::IngestionReport {
        crate::graph::types::IngestionReport {
            chunks_total: total,
            chunks_failed: failed,
            entities_created: 2,
            relationships_created: 1,
            errors: errors.iter().map(|e| (*e).to_string()).collect(),
            ..Default::default()
        }
    }

    #[test]
    fn a_total_failure_is_an_err_that_names_the_first_error() {
        let mut report = report_with(
            3,
            3,
            &[
                "extraction chunk 0: failed to spawn claude: No such file",
                "x",
                "y",
            ],
        );
        report.entities_created = 0;
        report.relationships_created = 0;
        let err = unit_outcome(report).expect_err("total failure");
        assert_eq!(
            err.to_string(),
            "llm error: all 3 extraction chunks failed, nothing extracted; first: extraction chunk 0: failed to spawn claude: No such file"
        );
    }

    #[test]
    fn a_partial_failure_is_a_report_with_warnings() {
        let outcome = unit_outcome(report_with(3, 1, &["extraction chunk 2: empty output"]))
            .expect("partial");
        assert_eq!(outcome.entities, 2);
        assert_eq!(outcome.relationships, 1);
        assert_eq!(
            outcome.warnings,
            vec!["extraction chunk 2: empty output".to_string()]
        );
    }

    #[test]
    fn a_clean_report_has_no_warnings() {
        let outcome = unit_outcome(report_with(3, 0, &[])).expect("clean");
        assert!(outcome.warnings.is_empty());
    }

    #[test]
    fn control_characters_in_provider_output_never_reach_the_log() {
        // Built from the char so the theme source scan (only theme.rs may
        // spell an escape) does not need an exemption for this fixture.
        const ESC: char = '\u{1b}';
        assert_eq!(one_line(&format!("a\nb{ESC}[31mc\td")), "a b [31mc\td");
        let long = "x".repeat(400);
        let out = one_line(&long);
        assert_eq!(out.chars().count(), 301);
        assert!(out.ends_with('…'));
    }

    #[test]
    fn a_single_failed_chunk_reads_in_the_singular() {
        let mut report = report_with(1, 1, &["e"]);
        report.entities_created = 0;
        report.relationships_created = 0;
        assert!(all_failed_message(&report).starts_with("all 1 extraction chunk failed"));
    }

    // ── Warnings line, with a fake unit. The retry/quarantine/wedged
    // ladder itself is covered in tests/serve_extract.rs. ─────────────

    /// A unit whose `extract` answers are scripted per call.
    struct FakeUnit {
        pending: Vec<u32>,
        answers: Mutex<Vec<Result<UnitReport, GraphError>>>,
        quarantined: Mutex<Vec<(u32, String)>>,
    }

    impl FakeUnit {
        fn new(pending: Vec<u32>, answers: Vec<Result<UnitReport, GraphError>>) -> Self {
            Self {
                pending,
                answers: Mutex::new(answers),
                quarantined: Mutex::new(Vec::new()),
            }
        }

        fn quarantined(&self) -> Vec<u32> {
            self.quarantined
                .lock()
                .unwrap()
                .iter()
                .map(|(n, _)| *n)
                .collect()
        }
    }

    #[async_trait]
    impl ExtractionUnit for FakeUnit {
        async fn pending(&self, limit: usize) -> Result<Vec<u32>, GraphError> {
            Ok(self.pending.iter().copied().take(limit).collect())
        }

        async fn extract(&self, _log_number: u32) -> Result<UnitReport, GraphError> {
            let mut answers = self.answers.lock().unwrap();
            if answers.is_empty() {
                return Err(GraphError::Llm("script exhausted".into()));
            }
            answers.remove(0)
        }

        async fn quarantine(&self, log_number: u32, reason: &str) {
            self.quarantined
                .lock()
                .unwrap()
                .push((log_number, reason.to_string()));
        }
    }

    struct Harness {
        worker: ExtractionWorker,
        log_path: std::path::PathBuf,
        _dir: tempfile::TempDir,
    }

    fn harness() -> Harness {
        let dir = tempfile::tempdir().expect("tempdir");
        let log_path = dir.path().join("daemon.log");
        let long_ago = Instant::now() - Duration::from_secs(3600);
        let worker = ExtractionWorker::new(
            Schedule {
                idle_after: Duration::from_secs(1),
                batch_size: 1,
                poll_interval: MIN_POLL,
            },
            WorkerContext {
                idle: Arc::new(IdleTracker::new_at(None, long_ago)),
                shutdown: Arc::new(ShutdownSignal::new()),
                state: ExtractionState::shared(),
                log: Arc::new(DaemonLog::open(&log_path, false)),
                clock: Arc::new(SystemClock),
            },
        );
        Harness {
            worker,
            log_path,
            _dir: dir,
        }
    }

    fn log_text(h: &Harness) -> String {
        std::fs::read_to_string(&h.log_path).unwrap_or_default()
    }

    #[tokio::test]
    async fn warnings_on_a_yielding_archive_get_their_own_log_line() {
        let unit = FakeUnit::new(
            vec![12],
            vec![Ok(UnitReport {
                entities: 4,
                relationships: 1,
                warnings: vec![
                    "extraction chunk 2: claude returned empty output".into(),
                    "dedup 'x': timeout".into(),
                ],
            })],
        );
        let mut h = harness();
        assert_eq!(h.worker.run_batch(&unit).await, BatchOutcome::Worked);
        let log = log_text(&h);
        assert!(log.contains("extracted log 012 in the background: +4 entities, 1 relationships"));
        assert!(log.contains(
            "extraction warnings on log 012: 2 warnings; first: extraction chunk 2: claude returned empty output"
        ));
        assert_eq!(
            log.matches("extracted log 012").count(),
            1,
            "one yield line per archive"
        );
        assert!(unit.quarantined().is_empty());
    }

    #[tokio::test]
    async fn a_clean_archive_logs_no_warning_line() {
        let unit = FakeUnit::new(vec![5], vec![Ok(UnitReport::default())]);
        let mut h = harness();
        h.worker.run_batch(&unit).await;
        let log = log_text(&h);
        assert!(log.contains("extracted log 005 in the background: +0 entities, 0 relationships"));
        assert!(!log.contains("warning"));
    }

    #[test]
    fn quarantined_log_numbers_survive_a_restart() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("extraction-quarantine.txt");
        assert!(read_quarantine(&path).is_empty());

        std::fs::write(&path, "007\n12\nnot-a-number\n").unwrap();
        let quarantined = read_quarantine(&path);
        assert!(quarantined.contains(&7));
        assert!(quarantined.contains(&12));
        assert_eq!(quarantined.len(), 2);
    }
}