osobh/clawmates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(corpus): record what a continuous mission has already covered

Browse Source 
Slice 2 of the adopt-or-build plan. A recurring mission's hard problem is
not running the agent — that is 23 seconds — it is knowing what it did
last time. This repository already tried continuous research once:
migrations 0030-0044 built research_topics/loops, 0053 dropped them all,
and the reason they could not survive is that research_topics carried a
status lifecycle but no seen-set. It could run forever and never know
what it had covered.

Two kinds of row, because the real vault forced it. The plan assumed
notes carry arxiv:/doi:/url: frontmatter. Measured against the actual
valhalla-vault: 416 notes, 145 with frontmatter, and ZERO with any of
those keys — the dominant keys are repo-sync metadata (node, org, gitea)
and course fields (presenter, session). An ingester keyed only on
external identity would have indexed nothing, which is the same shape of
failure as everything else found this week. So `note` rows record
coverage (keyed by path) and `source` rows record consumption (keyed by
natural id); a continuous mission needs both.

Two decisions the data forced:

- `source:` is deliberately NOT an identity key. The vault uses it for
  local paths of course material (/Users/quantum/Downloads/...), which is
  provenance, not citable identity. Accepting it would fill the seen-set
  with 25 rows keyed on a laptop path.
- The hash covers the body, not the whole file. Repo-sync notes rewrite
  updated:/size_kb: on every sync without the prose changing; hashing the
  file would report 103 phantom edits per run and make "unchanged"
  meaningless.

Authoritative in Postgres rather than ZeroClaw memory, per the Slice 1
spike: memory is agent-scoped and mission agents are ephemeral
claw_<uuid> aliases (~100 already present). A seen-set that disappears
with the agent that wrote it is not a seen-set. The spike did find that
POST /api/memory upserts by key, so mirroring content there later would
inherit idempotence for free if keyed by source_id.

Verified against the live 416-note vault, not a fixture:
  PASS1 { scanned: 416, inserted: 416, updated: 0, unchanged: 0 }
  PASS2 { scanned: 416, inserted: 0,   updated: 0, unchanged: 416 }

382 tests, clippy clean.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Omar Sobh
2026-08-03 07:00:21 -07:00
co-authored by Claude Opus 5
parent 2380c2cb0b
commit 6e5ccc25a6
4 changed files with 737 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/cm-api/src/corpus.rs
+466
View File
@@ -0,0 +1,466 @@
//! What a continuous mission has already covered.
//!
//! A recurring mission's hard problem is not running the agent — that is 23
//! seconds — it is knowing what it already did last time. A research mission
//! with no memory of prior runs resurfaces the same papers forever and reports
//! success every time.
//!
//! This module keeps that record. It is deliberately small: an index derived
//! from the corpus, never the corpus itself. The vault is the source of truth,
//! the index is rebuildable, and a hand-edited note is never "wrong".
//!
//! # Two kinds, because the real vault forced it
//!
//! The plan assumed notes would carry `arxiv:` / `doi:` / `url:` frontmatter.
//! Measured against the actual vault: **416 notes, 145 with frontmatter, and
//! zero with any of those keys.** The dominant keys are repo-sync metadata
//! (`node`, `org`, `gitea`) and course-note fields (`presenter`, `session`).
//! An ingester keyed only on external identity would have indexed nothing —
//! the same shape of failure as everything else this week.
//!
//! So `note` rows record coverage (what the vault already contains, keyed by
//! path) and `source` rows record consumption (external things a mission
//! read, keyed by natural id). They answer different questions and a
//! continuous mission needs both: "have I already written about this topic?"
//! and "have I already read this paper?".
use sha2::{Digest, Sha256};
use uuid::Uuid;
/// A note parsed out of the vault, ready to be indexed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ParsedNote {
/// Vault-relative path, used as identity for `kind = 'note'`.
pub path: String,
pub title: Option<String>,
pub content_hash: String,
/// An external identity the note declares for itself, if any. Nothing in
/// the vault does this today; missions writing new notes are expected to.
pub declared_source_id: Option<String>,
}
impl ParsedNote {
/// `note:<path>` — the `source_id` this note occupies in the index.
pub fn source_id(&self) -> String {
format!("note:{}", self.path)
}
}
/// Hash content for change detection. Not a dedupe key — identity is
/// `source_id`; this only distinguishes "unchanged" from "edited".
pub fn content_hash(body: &str) -> String {
let mut h = Sha256::new();
h.update(body.as_bytes());
format!("{:x}", h.finalize())
}
/// Split YAML frontmatter from the body.
///
/// Returns `(frontmatter, body)`. A note without frontmatter — 271 of the 416
/// in the real vault — yields `("", whole file)` rather than being skipped.
/// Skipping them would drop two thirds of the corpus on the floor.
fn split_frontmatter(text: &str) -> (&str, &str) {
let Some(rest) = text.strip_prefix("---") else {
return ("", text);
};
let rest = rest.strip_prefix('\n').unwrap_or(rest);
match rest.find("\n---") {
Some(end) => {
let body = &rest[end + 4..];
(&rest[..end], body.strip_prefix('\n').unwrap_or(body))
}
// An opening fence with no close is malformed; treat the whole file as
// body rather than swallowing it as frontmatter.
None => ("", text),
}
}
/// Read one scalar key out of a frontmatter block.
///
/// Deliberately not a YAML parser. The vault's frontmatter is flat
/// `key: value` with occasional quotes and one list (`tags`), and pulling in a
/// YAML dependency to read three keys would be more surface than it is worth.
fn frontmatter_value<'a>(fm: &'a str, key: &str) -> Option<&'a str> {
for line in fm.lines() {
let line = line.trim();
let Some((k, v)) = line.split_once(':') else {
continue;
};
if !k.trim().eq_ignore_ascii_case(key) {
continue;
}
let v = v.trim().trim_matches('"').trim_matches('\'').trim();
if !v.is_empty() {
return Some(v);
}
}
None
}
/// Which frontmatter keys may declare an external identity, in priority order.
///
/// None of these appear in the vault today. They are the contract for notes
/// that missions write from here on, and the reason a `source:` key is NOT in
/// the list: the vault already uses `source:` for local filesystem paths of
/// course material (`/Users/quantum/Downloads/...`), which is provenance, not
/// a citable external identity. Treating it as one would fill the seen-set
/// with 25 rows keyed on a laptop path.
const IDENTITY_KEYS: &[&str] = &["source_id", "arxiv", "doi", "url", "permalink"];
/// Parse a note. `path` must be vault-relative.
pub fn parse_note(path: &str, text: &str) -> ParsedNote {
let (fm, body) = split_frontmatter(text);
let declared_source_id = IDENTITY_KEYS.iter().find_map(|k| {
frontmatter_value(fm, k).map(|v| {
// `source_id` is already qualified; the others name their scheme.
if *k == "source_id" || v.contains(':') {
v.to_string()
} else {
format!("{k}:{v}")
}
})
});
// Title: the first markdown H1, else the filename stem. Frontmatter has no
// consistent title key in this vault.
let title = body
.lines()
.find_map(|l| l.strip_prefix("# ").map(str::trim))
.filter(|t| !t.is_empty())
.map(str::to_string)
.or_else(|| {
std::path::Path::new(path)
.file_stem()
.map(|s| s.to_string_lossy().into_owned())
});
ParsedNote {
path: path.to_string(),
title,
// Hash the body, not the whole file: re-syncing a repo note rewrites
// `updated:`/`size_kb:` in frontmatter without the prose changing, and
// that should not read as an edit.
content_hash: content_hash(body),
declared_source_id,
}
}
/// What a re-index actually did. `unchanged` is the number that matters: on a
/// vault nobody edited it should equal the note count.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct IndexStats {
pub scanned: usize,
pub inserted: usize,
pub updated: usize,
pub unchanged: usize,
}
/// Walk a checkout and index every markdown note.
///
/// Skips `.git` and Obsidian's own `.obsidian` config directory — indexing an
/// editor's workspace state as knowledge would be noise.
pub fn collect_notes(root: &std::path::Path) -> Vec<ParsedNote> {
fn walk(dir: &std::path::Path, root: &std::path::Path, out: &mut Vec<ParsedNote>) {
let Ok(entries) = std::fs::read_dir(dir) else {
return;
};
for entry in entries.flatten() {
let path = entry.path();
let name = entry.file_name();
let name = name.to_string_lossy();
if name.starts_with('.') {
continue;
}
if path.is_dir() {
walk(&path, root, out);
} else if path.extension().and_then(|e| e.to_str()) == Some("md") {
let Ok(text) = std::fs::read_to_string(&path) else {
continue;
};
let rel = path
.strip_prefix(root)
.unwrap_or(&path)
.to_string_lossy()
.into_owned();
out.push(parse_note(&rel, &text));
}
}
}
let mut out = Vec::new();
walk(root, root, &mut out);
out.sort_by(|a, b| a.path.cmp(&b.path));
out
}
/// Upsert one item. Returns whether the row was new.
#[allow(clippy::too_many_arguments)]
pub async fn record(
pool: &sqlx::PgPool,
workspace_id: Uuid,
corpus_id: &str,
kind: &str,
source_id: &str,
title: Option<&str>,
path: Option<&str>,
url: Option<&str>,
content_hash: &str,
mission_id: Option<Uuid>,
) -> Result<bool, String> {
// `last_seen_at` always moves; `first_seen_at` and `mission_id` never do.
// The first mission to find a source keeps the credit, which is what makes
// "did THIS run contribute anything new" answerable.
let row: (bool,) = sqlx::query_as(
"INSERT INTO corpus_items
(id, workspace_id, corpus_id, kind, source_id, title, path, url,
content_hash, mission_id)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
ON CONFLICT (workspace_id, corpus_id, source_id) DO UPDATE
SET last_seen_at = now(),
title = COALESCE(EXCLUDED.title, corpus_items.title),
path = COALESCE(EXCLUDED.path, corpus_items.path),
url = COALESCE(EXCLUDED.url, corpus_items.url),
content_hash = EXCLUDED.content_hash
RETURNING (xmax = 0) AS inserted",
)
.bind(Uuid::now_v7())
.bind(workspace_id)
.bind(corpus_id)
.bind(kind)
.bind(source_id)
.bind(title)
.bind(path)
.bind(url)
.bind(content_hash)
.bind(mission_id)
.fetch_one(pool)
.await
.map_err(|e| format!("record corpus item {source_id}: {e}"))?;
Ok(row.0)
}
/// Has this corpus already seen this `source_id`?
pub async fn seen(
pool: &sqlx::PgPool,
workspace_id: Uuid,
corpus_id: &str,
source_id: &str,
) -> Result<bool, String> {
// `SELECT 1` is INT4; binding it as i64 fails to decode.
let row: Option<(i32,)> = sqlx::query_as(
"SELECT 1 FROM corpus_items
WHERE workspace_id = $1 AND corpus_id = $2 AND source_id = $3",
)
.bind(workspace_id)
.bind(corpus_id)
.bind(source_id)
.fetch_optional(pool)
.await
.map_err(|e| format!("seen({source_id}): {e}"))?;
Ok(row.is_some())
}
/// Of these candidate ids, which has this corpus NOT seen?
///
/// The shape a research agent actually needs: it has ten search hits and wants
/// to know which are worth fetching. One round trip, not ten.
pub async fn unseen(
pool: &sqlx::PgPool,
workspace_id: Uuid,
corpus_id: &str,
candidates: &[String],
) -> Result<Vec<String>, String> {
if candidates.is_empty() {
return Ok(Vec::new());
}
let rows: Vec<(String,)> = sqlx::query_as(
"SELECT source_id FROM corpus_items
WHERE workspace_id = $1 AND corpus_id = $2 AND source_id = ANY($3)",
)
.bind(workspace_id)
.bind(corpus_id)
.bind(candidates)
.fetch_all(pool)
.await
.map_err(|e| format!("unseen: {e}"))?;
let known: std::collections::HashSet<String> = rows.into_iter().map(|r| r.0).collect();
Ok(candidates
.iter()
.filter(|c| !known.contains(*c))
.cloned()
.collect())
}
/// Index every note in a checkout. Idempotent by construction.
pub async fn index_vault(
pool: &sqlx::PgPool,
workspace_id: Uuid,
corpus_id: &str,
root: &std::path::Path,
) -> Result<IndexStats, String> {
let notes = collect_notes(root);
let mut stats = IndexStats {
scanned: notes.len(),
..Default::default()
};
for note in &notes {
let existing: Option<(String,)> = sqlx::query_as(
"SELECT content_hash FROM corpus_items
WHERE workspace_id = $1 AND corpus_id = $2 AND source_id = $3",
)
.bind(workspace_id)
.bind(corpus_id)
.bind(note.source_id())
.fetch_optional(pool)
.await
.map_err(|e| format!("lookup {}: {e}", note.path))?;
match existing {
Some((hash,)) if hash == note.content_hash => {
stats.unchanged += 1;
continue;
}
Some(_) => stats.updated += 1,
None => stats.inserted += 1,
}
record(
pool,
workspace_id,
corpus_id,
"note",
&note.source_id(),
note.title.as_deref(),
Some(&note.path),
None,
&note.content_hash,
None,
)
.await?;
// A note that declares an external identity also registers as a
// consumed source, so a later mission does not re-read what an
// earlier one already wrote up.
if let Some(sid) = &note.declared_source_id {
record(
pool,
workspace_id,
corpus_id,
"source",
sid,
note.title.as_deref(),
Some(&note.path),
None,
&note.content_hash,
None,
)
.await?;
}
}
Ok(stats)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn frontmatter_is_split_from_body() {
let (fm, body) = split_frontmatter("---\ntype: lecture\n---\n# Title\n\ntext\n");
assert_eq!(fm, "type: lecture");
assert!(body.starts_with("# Title"));
}
/// 271 of the vault's 416 notes have no frontmatter. Dropping them would
/// discard two thirds of the corpus.
#[test]
fn a_note_without_frontmatter_is_still_a_note() {
let (fm, body) = split_frontmatter("# Plain\n\nno frontmatter here\n");
assert_eq!(fm, "");
assert!(body.starts_with("# Plain"));
let n = parse_note("Daily/x.md", "# Plain\n\nbody\n");
assert_eq!(n.title.as_deref(), Some("Plain"));
assert_eq!(n.declared_source_id, None);
}
/// An unterminated fence must not swallow the file.
#[test]
fn malformed_frontmatter_is_treated_as_body() {
let (fm, body) = split_frontmatter("---\nbroken: yes\nno closing fence\n");
assert_eq!(fm, "");
assert!(body.contains("no closing fence"));
}
/// The vault's real `source:` values are local filesystem paths of course
/// material. Treating those as citable identity would fill the seen-set
/// with 25 rows keyed on a laptop path.
#[test]
fn a_local_source_path_is_not_an_external_identity() {
let note = parse_note(
"50 APESS 2026/Lectures/talk.md",
"---\nsource: \"/Users/quantum/Downloads/Material_APESS_2026/x.pdf\"\n\
date: 2026-07-27\ntype: lecture\n---\n# Agentic Design\n",
);
assert_eq!(
note.declared_source_id, None,
"a Downloads path is provenance, not a citable source id"
);
assert_eq!(note.title.as_deref(), Some("Agentic Design"));
assert_eq!(note.source_id(), "note:50 APESS 2026/Lectures/talk.md");
}
#[test]
fn declared_identities_are_scheme_qualified() {
let a = parse_note("p.md", "---\narxiv: 2401.12345\n---\n# T\n");
assert_eq!(a.declared_source_id.as_deref(), Some("arxiv:2401.12345"));
let d = parse_note("p.md", "---\ndoi: 10.1000/xyz\n---\n# T\n");
assert_eq!(d.declared_source_id.as_deref(), Some("doi:10.1000/xyz"));
// Already-qualified values are not double-prefixed.
let s = parse_note("p.md", "---\nsource_id: arxiv:2401.99999\n---\n# T\n");
assert_eq!(s.declared_source_id.as_deref(), Some("arxiv:2401.99999"));
// A URL carries its own scheme and must not become `url:https:...`.
let u = parse_note("p.md", "---\nurl: https://example.com/p\n---\n# T\n");
assert_eq!(
u.declared_source_id.as_deref(),
Some("https://example.com/p")
);
}
/// Repo-sync notes rewrite `updated:`/`size_kb:` on every sync without the
/// prose changing. Hashing the whole file would report 103 phantom edits
/// per run and make "unchanged" meaningless.
#[test]
fn frontmatter_churn_does_not_count_as_an_edit() {
let a = parse_note("Repos/x.md", "---\nupdated: 2026-08-01\nsize_kb: 12\n---\n# X\n\nbody\n");
let b = parse_note("Repos/x.md", "---\nupdated: 2026-08-03\nsize_kb: 14\n---\n# X\n\nbody\n");
assert_eq!(a.content_hash, b.content_hash);
let c = parse_note("Repos/x.md", "---\nupdated: 2026-08-03\n---\n# X\n\nDIFFERENT\n");
assert_ne!(a.content_hash, c.content_hash, "real edits must be visible");
}
#[test]
fn note_identity_is_its_path() {
let n = parse_note("30 Resources/a b.md", "# A\n");
assert_eq!(n.source_id(), "note:30 Resources/a b.md");
}
#[test]
fn collect_skips_dotfiles_and_non_markdown() {
let tmp = tempfile::tempdir().unwrap();
let root = tmp.path();
std::fs::create_dir_all(root.join(".obsidian")).unwrap();
std::fs::create_dir_all(root.join("Daily")).unwrap();
std::fs::write(root.join(".obsidian/workspace.md"), "# editor state\n").unwrap();
std::fs::write(root.join("Daily/note.md"), "# Real\n").unwrap();
std::fs::write(root.join("image.png"), "notmd").unwrap();
let notes = collect_notes(root);
assert_eq!(notes.len(), 1, "only the real note: {notes:?}");
assert_eq!(notes[0].path, "Daily/note.md");
}
}