clawmates/crates/cm-api/src/harvest.rs

//! One run of the library: find, skip what we have, shelve the rest.
//!
//! This is the piece that makes the others a *job* rather than parts on a
//! bench. Order matters and it is deliberate:
//!
//! 1. **search** arXiv for candidates
//! 2. **skip** everything already on the checkmark list — before any download
//! 3. **fetch** the PDF for what is left, and verify it really is a PDF
//! 4. **shelve** it in the blob store
//! 5. **catalogue** it: write the vault note
//! 6. **check it off** so next week skips it
//!
//! Step 2 comes before step 3 on purpose. Checking after downloading would
//! still dedupe the catalogue, but it would re-download every paper we already
//! have, every week, forever — and the whole point of the checkmark list is to
//! not do the work twice.
//!
//! # Nothing new is a success, not a failure
//!
//! A weekly run that finds no new papers has worked correctly. A run that
//! *crashed* has not. [`Harvest`] keeps those apart, because collapsing them
//! is precisely the "reported success while doing nothing" shape that this
//! codebase has been bitten by repeatedly. `shelved == 0` with `failed.empty()`
//! is a quiet week; `shelved == 0` with failures is a broken run.

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

use uuid::Uuid;

use crate::corpus;
use crate::papers::{self, Paper};

/// What one run did. Every number here is observed, not claimed.
#[derive(Debug, Default, Clone)]
pub struct Harvest {
    /// Papers the search returned.
    pub candidates: usize,
    /// Of those, how many were already on the checkmark list.
    pub already_had: usize,
    /// Successfully downloaded, shelved and catalogued.
    pub shelved: Vec<String>,
    /// `(source_id, why)` for each paper that could not be shelved.
    pub failed: Vec<(String, String)>,
    /// Vault-relative paths of the notes written.
    pub notes_written: Vec<String>,
    /// The papers actually shelved this run, in shelve order.
    ///
    /// `shelved` carries only source ids, which is all the seen-set needs. The
    /// run manifest a Continuous Research mission hands its agents needs the
    /// title and abstract too, and re-reading them back out of the notes we
    /// just wrote would be a parse of our own output — one more place for the
    /// two to drift.
    pub papers: Vec<crate::papers::Paper>,
}

impl Harvest {
    /// Did this run add anything? The verification predicate for a continuous
    /// research mission: a run that contributes no new source has produced
    /// nothing, whatever its transcript says.
    pub fn added_anything(&self) -> bool {
        !self.shelved.is_empty()
    }

    /// A run is healthy if nothing errored — including a run that found
    /// nothing new, which is the normal state of a mature library.
    pub fn healthy(&self) -> bool {
        self.failed.is_empty()
    }

    pub fn summary(&self) -> String {
        format!(
            "{} candidates, {} already held, {} shelved, {} failed",
            self.candidates,
            self.already_had,
            self.shelved.len(),
            self.failed.len()
        )
    }
}

/// Where a library lives: its records, its shelf, and its catalogue.
///
/// Grouped rather than passed as loose arguments because these five always
/// travel together and always describe one library — splitting them at a call
/// site is how a run ends up shelving into one place and cataloguing into
/// another.
pub struct Library<'a> {
    pub pool: &'a sqlx::PgPool,
    /// The shelf: where PDFs are stored.
    pub blobs: &'a Arc<dyn cm_files::BlobStore>,
    pub workspace_id: Uuid,
    /// Which checkmark list, e.g. `"valhalla-vault"`.
    pub corpus_id: &'a str,
    /// Checkout the catalogue notes are written into.
    pub vault_root: &'a Path,
}

/// Shelve a specific set of papers. Split from [`run`] so the skip/shelve
/// logic is testable without reaching arXiv.
pub async fn shelve(
    lib: &Library<'_>,
    candidates: &[Paper],
    mission_id: Option<Uuid>,
) -> Result<Harvest, String> {
    let Library { pool, blobs, workspace_id, corpus_id, vault_root } = *lib;
    let mut out = Harvest {
        candidates: candidates.len(),
        ..Default::default()
    };

    // One round trip for the whole batch rather than one query per paper.
    let ids: Vec<String> = candidates.iter().map(Paper::source_id).collect();
    let fresh: std::collections::HashSet<String> =
        corpus::unseen(pool, workspace_id, corpus_id, &ids)
            .await?
            .into_iter()
            .collect();
    out.already_had = candidates.len() - fresh.len();

    for paper in candidates {
        let sid = paper.source_id();
        if !fresh.contains(&sid) {
            continue;
        }

        // Fetch first. If the PDF cannot be had, nothing is recorded — the
        // paper stays unseen so a later run retries it, rather than being
        // checked off with an empty shelf slot behind it.
        let bytes = match papers::fetch_pdf(paper).await {
            Ok(b) => b,
            Err(e) => {
                out.failed.push((sid, e));
                continue;
            }
        };

        let key = paper.blob_key();
        if let Err(e) = blobs.put(&key, &bytes).await {
            out.failed.push((sid, format!("shelve {key}: {e}")));
            continue;
        }

        // Catalogue note next to the shelf. Written into the vault checkout;
        // committing and pushing it is the caller's job, through the delivery
        // path that already exists.
        let note = papers::catalogue_note(paper, &key);
        let note_path = vault_root.join(paper.note_path());
        if let Some(parent) = note_path.parent() {
            if let Err(e) = std::fs::create_dir_all(parent) {
                out.failed.push((sid, format!("create {}: {e}", parent.display())));
                continue;
            }
        }
        if let Err(e) = std::fs::write(&note_path, &note) {
            out.failed
                .push((sid, format!("write {}: {e}", note_path.display())));
            continue;
        }

        // Check it off LAST. If anything above failed we did not get the
        // paper, and marking it seen would mean never trying again.
        corpus::record(
            pool,
            workspace_id,
            corpus_id,
            "source",
            &sid,
            Some(&paper.title),
            Some(&paper.note_path()),
            Some(&format!("https://arxiv.org/abs/{}", paper.arxiv_id)),
            &corpus::content_hash(&note),
            mission_id,
        )
        .await?;

        out.notes_written.push(paper.note_path());
        out.papers.push(paper.clone());
        out.shelved.push(sid);
    }

    Ok(out)
}

/// A full run: search arXiv, then shelve whatever is new.
pub async fn run(
    lib: &Library<'_>,
    query: &str,
    limit: usize,
    mission_id: Option<Uuid>,
) -> Result<Harvest, String> {
    let candidates = papers::search(query, limit).await?;
    let harvest = shelve(lib, &candidates, mission_id).await?;
    let corpus_id = lib.corpus_id;
    eprintln!("harvest[{corpus_id}] query={query:?} → {}", harvest.summary());
    for (sid, why) in &harvest.failed {
        eprintln!("harvest[{corpus_id}] FAILED {sid}: {why}");
    }
    Ok(harvest)
}

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

    #[test]
    fn a_quiet_week_is_healthy_but_adds_nothing() {
        let quiet = Harvest {
            candidates: 5,
            already_had: 5,
            ..Default::default()
        };
        assert!(quiet.healthy(), "finding nothing new is not an error");
        assert!(
            !quiet.added_anything(),
            "but it must not count as having produced something"
        );

        let broken = Harvest {
            candidates: 5,
            already_had: 0,
            failed: vec![("arxiv:1".into(), "timeout".into())],
            ..Default::default()
        };
        assert!(!broken.healthy());
        assert!(!broken.added_anything());

        let good = Harvest {
            candidates: 5,
            already_had: 4,
            shelved: vec!["arxiv:2".into()],
            ..Default::default()
        };
        assert!(good.healthy() && good.added_anything());
    }
}