feat(harvest): one run — find, skip what we hold, shelve the rest

Turns the parts into a job. Order is the point: the checkmark list is
consulted BEFORE anything downloads. Checking afterwards would still
dedupe the catalogue while re-downloading every paper we already have,
every week, forever.

Two properties the tests pin down, both learned the hard way this week:

- A quiet week is not a failure. `shelved == 0` with no errors is a
  healthy run against a mature library; `shelved == 0` with errors is
  broken. Harvest::healthy() and ::added_anything() keep those apart
  rather than collapsing them into one ambiguous "did nothing".
- A failed download leaves the paper UNSEEN. Checking it off before the
  PDF is safely shelved would mean one transient network error retires
  that paper permanently. The checkmark is written last, after the bytes
  and the note are both on disk.

The skip test gives every candidate a pdf_url pointing at a closed port,
so if the skip ever regresses the test fails loudly instead of quietly
re-fetching.

Live end-to-end against arXiv, run twice:
  RUN1  3 candidates, 0 already held, 3 shelved, 0 failed
  RUN2  3 candidates, 3 already held, 0 shelved, 0 failed

Library<'_> groups the five values that always describe one library;
passing them loose is how a run shelves into one place and catalogues
into another (also silences clippy::too_many_arguments honestly rather
than by allow).

391 tests, clippy clean.

Co-Authored-By: Claude Opus 5 <[email protected]>

Cargo.lock

@@ -946,6 +946,7 @@ dependencies = [
  "cm-config",
  "cm-db",
  "cm-domain",
+ "cm-files",
  "cm-llm",
  "cm-orchestrator",
  "cm-runtime",

crates/cm-api/Cargo.toml

@@ -32,6 +32,7 @@ cm-brain = { path = "../cm-brain" }
 cm-config = { path = "../cm-config" }
 cm-db = { path = "../cm-db" }
 cm-domain = { path = "../cm-domain" }
+cm-files = { path = "../cm-files" }
 cm-llm = { path = "../cm-llm" }
 cm-orchestrator = { path = "../cm-orchestrator", features = ["provider"] }
 cm-runtime = { path = "../cm-runtime" }

crates/cm-api/src/harvest.rs

@@ -0,0 +1,227 @@
+//! 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>,
+}
+
+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.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());
+    }
+}

crates/cm-api/src/lib.rs

@@ -17,6 +17,7 @@ mod mcp_skills;
 pub mod mission_orchestrator;
 pub mod mission_refiner;
 pub mod corpus;
+pub mod harvest;
 pub mod mission_delivery;
 pub mod papers;
 pub mod phase_config;

crates/cm-api/tests/harvest_run.rs

@@ -0,0 +1,153 @@
+//! A second run must not re-download what the first run already shelved.
+
+use cm_api::{corpus, harvest, papers::Paper};
+use std::sync::Arc;
+use uuid::Uuid;
+
+async fn workspace(pool: &sqlx::PgPool) -> Uuid {
+    let ws = Uuid::now_v7();
+    sqlx::query("INSERT INTO workspaces (id, name, plan) VALUES ($1,'t','team')")
+        .bind(ws)
+        .execute(pool)
+        .await
+        .unwrap();
+    ws
+}
+
+fn paper(id: &str) -> Paper {
+    Paper {
+        arxiv_id: id.into(),
+        title: format!("Paper {id}"),
+        authors: vec!["Ada Lovelace".into()],
+        summary: "A summary.".into(),
+        published: "2026-01-15T10:00:00Z".into(),
+        // Deliberately unreachable: if the skip works, this is never fetched.
+        pdf_url: "http://127.0.0.1:1/never.pdf".into(),
+    }
+}
+
+/// The load-bearing behaviour. Every candidate is already on the checkmark
+/// list, and every `pdf_url` points at a closed port — so if the run tries to
+/// download anything at all, it fails loudly instead of passing quietly.
+#[tokio::test]
+async fn papers_we_already_hold_are_never_downloaded_again() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = workspace(&pool).await;
+    let tmp = tempfile::tempdir().unwrap();
+    let blobs: Arc<dyn cm_files::BlobStore> =
+        Arc::new(cm_files::LocalBlobStore::new(tmp.path().join("blobs")));
+    let vault = tmp.path().join("vault");
+
+    let candidates = vec![paper("2401.11111"), paper("2401.22222")];
+    for p in &candidates {
+        corpus::record(
+            &pool, ws, "lib", "source", &p.source_id(),
+            Some(&p.title), None, None, "h", None,
+        )
+        .await
+        .unwrap();
+    }
+
+    let lib = harvest::Library {
+        pool: &pool, blobs: &blobs, workspace_id: ws,
+        corpus_id: "lib", vault_root: &vault,
+    };
+    let h = harvest::shelve(&lib, &candidates, None).await.unwrap();
+
+    assert_eq!(h.candidates, 2);
+    assert_eq!(h.already_had, 2, "both were already held");
+    assert!(h.shelved.is_empty());
+    assert!(
+        h.failed.is_empty(),
+        "nothing should have been fetched at all, but got: {:?}",
+        h.failed
+    );
+    assert!(h.healthy(), "a fully-known batch is a healthy quiet week");
+    assert!(!h.added_anything(), "and it added nothing");
+    assert!(!vault.exists(), "no notes written for papers we already had");
+}
+
+/// A paper that cannot be downloaded must NOT be checked off — otherwise one
+/// transient network failure means that paper is never retried.
+#[tokio::test]
+async fn a_failed_download_leaves_the_paper_unseen_for_next_time() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = workspace(&pool).await;
+    let tmp = tempfile::tempdir().unwrap();
+    let blobs: Arc<dyn cm_files::BlobStore> =
+        Arc::new(cm_files::LocalBlobStore::new(tmp.path().join("blobs")));
+    let vault = tmp.path().join("vault");
+
+    let candidates = vec![paper("2401.33333")];
+    let lib = harvest::Library {
+        pool: &pool, blobs: &blobs, workspace_id: ws,
+        corpus_id: "lib", vault_root: &vault,
+    };
+    let h = harvest::shelve(&lib, &candidates, None).await.unwrap();
+
+    assert_eq!(h.already_had, 0);
+    assert!(h.shelved.is_empty());
+    assert_eq!(h.failed.len(), 1, "the unreachable fetch must be reported");
+    assert!(!h.healthy(), "a failed fetch is not a quiet week");
+
+    assert!(
+        !corpus::seen(&pool, ws, "lib", "arxiv:2401.33333")
+            .await
+            .unwrap(),
+        "a paper we failed to get must stay unseen so a later run retries it"
+    );
+}
+
+/// Live end-to-end: search arXiv, shelve genuinely new papers, then confirm a
+/// second identical run adds nothing. Ignored by default (network + Postgres):
+/// `cargo test -p cm-api --test harvest_run live_ -- --ignored --nocapture`
+#[tokio::test]
+#[ignore]
+async fn live_end_to_end_run_then_rerun() {
+    let pool = cm_testkit::test_pool().await;
+    let ws = workspace(&pool).await;
+    let tmp = tempfile::tempdir().unwrap();
+    let blobs: Arc<dyn cm_files::BlobStore> =
+        Arc::new(cm_files::LocalBlobStore::new(tmp.path().join("blobs")));
+    let vault = tmp.path().join("vault");
+
+    let lib = harvest::Library {
+        pool: &pool, blobs: &blobs, workspace_id: ws,
+        corpus_id: "lib", vault_root: &vault,
+    };
+    let first = harvest::run(&lib, "all:agentic AND all:topology", 3, None)
+        .await
+        .unwrap();
+    println!("RUN1 {}", first.summary());
+    for n in &first.notes_written {
+        println!("  note: {n}");
+    }
+    assert!(first.healthy(), "failures: {:?}", first.failed);
+    assert!(first.added_anything(), "first run should find something new");
+
+    // Every note must be readable back through the corpus parser, or the
+    // catalogue cannot rebuild the checkmark list.
+    for rel in &first.notes_written {
+        let text = std::fs::read_to_string(vault.join(rel)).unwrap();
+        let parsed = corpus::parse_note(rel, &text);
+        assert!(
+            parsed
+                .declared_source_id
+                .as_deref()
+                .is_some_and(|s| s.starts_with("arxiv:")),
+            "note {rel} lost its identity"
+        );
+    }
+
+    let second = harvest::run(&lib, "all:agentic AND all:topology", 3, None)
+        .await
+        .unwrap();
+    println!("RUN2 {}", second.summary());
+    assert!(second.healthy());
+    assert!(
+        !second.added_anything(),
+        "a rerun must add nothing — got {:?}",
+        second.shelved
+    );
+    assert_eq!(second.already_had, second.candidates);
+}