sessions: an import writes its archive's search

cli/GUIDE.md

@@ -880,7 +880,10 @@ from Hermes' database) replaces the older; an older copy never does. It says wha
 already had, and what it left out and why: at most 18,000 conversations
 and 1 GiB an archive (an import past either adds nothing), 8 MiB a
 conversation. A tool's output and an agent's reasoning are cut to 2,000
-characters; what was said, never. An agent runs it for its owner (into
+characters; what was said, never. Each import also writes the archive's
+search, a section a turn of each conversation, and puts it right when it
+has fallen behind (on a host without search yet, it says so and skips
+it). An agent runs it for its owner (into
 their archive, never one of its own), and so reads it: `fragment sync
 <archive> --dir ~/archive --mode pull`, then your own tools over
 `~/archive/sessions/`. A conversation's file (format

cli/src/session/mod.rs

@@ -17,6 +17,7 @@
 
 mod chats;
 mod hermes;
+mod search;
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::path::{Path, PathBuf};
@@ -84,6 +85,8 @@ pub struct Imported {
     pub problems: Vec<String>,
     /// What the archive holds now, by kind.
     pub holds: BTreeMap<&'static str, usize>,
+    /// What it did to the archive's search (search.rs).
+    pub search: search::Searched,
 }
 
 #[derive(Debug, Serialize)]
@@ -264,8 +267,9 @@ pub fn run(c: &Client, sub: SessionCmd) -> Result<Imported> {
     for e in written.entries.values() {
         *holds.entry(e.kind.as_str()).or_default() += 1;
     }
+    let search = search::sync(c, &archive, &spool);
     let refused = std::mem::take(&mut spool.refused);
-    Ok(Imported { archive, made, added: written.added, updated: written.updated, unchanged: written.unchanged, refused, problems, holds })
+    Ok(Imported { archive, made, added: written.added, updated: written.updated, unchanged: written.unchanged, refused, problems, holds, search })
 }
 
 /// Where a path a person names reads from.
@@ -565,6 +569,24 @@ pub fn print(r: &Imported) {
         noun(r.holds["cron"], "scheduled run", "scheduled runs"),
         noun(r.holds["subagent"], "side task", "side tasks")
     );
+    match &r.search {
+        search::Searched::Skipped { why } => println!("Search: skipped ({why}); the next import puts it right."),
+        search::Searched::Written { written, unchanged, overtaken, left_out, refused, refused_count } => {
+            println!("Search: {written} written, {unchanged} up to date.");
+            if *overtaken > 0 {
+                println!("  {overtaken} written meanwhile by another import, left as it wrote them.");
+            }
+            for one in refused {
+                println!("  not searchable: {} ({})", one.id, one.why);
+            }
+            if *refused_count > refused.len() {
+                println!("  and {} more not searchable.", refused_count - refused.len());
+            }
+            if *left_out > 0 {
+                println!("  {} sections past search's limits left out.", left_out);
+            }
+        }
+    }
 }
 
 /// A fresh folder in the system's temporary place (a test's).
@@ -895,6 +917,108 @@ mod tests {
         assert!(!stale.exists() && fresh.exists() && other.exists());
     }
 
+    /// Goal: an import leaves its archive's search holding each of its
+    /// conversations as it is, and nothing else of the archive's: valid,
+    /// replay, update, an index lost (repaired from the archive's own
+    /// files), a conversation gone, and a host without search (skipped, the
+    /// import standing). Method: the search fake behind the host's routes.
+    #[test]
+    fn an_import_puts_its_archives_search_right() {
+        let (mock, c) = archive();
+        let upserts = || mock.take_requests("").get("POST upsert").copied().unwrap_or(0);
+        write(&c, NAME, &spooled(vec![session("a", Kind::Chat, "2026-10-01T09:00:00Z", &["hello", "hi"])])).unwrap();
+        let skipped = search::sync(&c, NAME, &spooled(vec![]));
+        assert!(matches!(&skipped, search::Searched::Skipped { why } if why.contains("no search")), "{skipped:?}");
+
+        mock.enable_search(NAME);
+        let spool = spooled(vec![session("a", Kind::Chat, "2026-10-01T09:00:00Z", &["hello", "hi"]), session("r", Kind::Cron, "2026-10-02T06:00:00Z", &["digest"])]);
+        write(&c, NAME, &spool).unwrap();
+        upserts();
+        let s = search::sync(&c, NAME, &spool);
+        assert!(matches!(s, search::Searched::Written { written: 2, unchanged: 0, overtaken: 0, .. }), "{s:?}");
+        let held = |source: &str| mock.search_index(NAME, |ix| ix.sources.get(source).cloned()).flatten();
+        let a = held("sessions/hermes-a.json").expect("a is searched");
+        assert_eq!(a.revision, sha(&file_of(session("a", Kind::Chat, "2026-10-01T09:00:00Z", &["hello", "hi"]))));
+        assert_eq!(a.sections[0].text, "Person: hello\nAgent: hi");
+        assert_eq!(a.fields["kind"], "chat");
+        assert_eq!(upserts(), 2);
+
+        write(&c, NAME, &spool).unwrap();
+        let again = search::sync(&c, NAME, &spooled(vec![]));
+        assert!(matches!(again, search::Searched::Written { written: 0, unchanged: 2, .. }), "{again:?}");
+        assert_eq!(upserts(), 0, "a replay writes nothing");
+
+        let later = spooled(vec![session("a", Kind::Chat, "2026-10-03T09:00:00Z", &["hello", "hi", "and peas?"])]);
+        write(&c, NAME, &later).unwrap();
+        assert!(matches!(search::sync(&c, NAME, &later), search::Searched::Written { written: 1, unchanged: 1, .. }));
+        assert_eq!(held("sessions/hermes-a.json").unwrap().sections.len(), 2);
+
+        mock.search_index(NAME, |ix| ix.sources.clear());
+        mock.external_commit(NAME, "main", &[(session::session_path("hermes-r").as_str(), None)], "a conversation removed");
+        write(&c, NAME, &spooled(vec![])).unwrap();
+        let other = fragment_fakes::search::Held { revision: "1".into(), title: None, at: None, fields: BTreeMap::new(), sections: vec![] };
+        mock.search_index(NAME, |ix| {
+            ix.sources.insert("sessions/hermes-gone.json".into(), other.clone());
+            ix.sources.insert("notes/x.md".into(), other);
+        });
+        let repaired = search::sync(&c, NAME, &spooled(vec![]));
+        assert!(matches!(repaired, search::Searched::Written { written: 1, unchanged: 0, .. }), "{repaired:?}");
+        assert!(held("sessions/hermes-a.json").is_some(), "read again from the archive");
+        assert!(held("sessions/hermes-gone.json").is_some() && held("notes/x.md").is_some(), "nothing deleted: the platform forgets a source when its file goes");
+    }
+
+    /// Goal: an import that pauses between its writes and its search, while
+    /// another writes after it and puts search right, undoes nothing of the
+    /// other's search: no older copy written back over a newer one, no
+    /// conversation the other added removed. Method: import A writes `a`
+    /// and pauses; B writes a newer `a` and a new `b`, then its search; A's
+    /// search then.
+    #[test]
+    fn an_import_overtaken_undoes_nothing_of_a_later_ones_search() {
+        let (mock, c) = archive();
+        mock.enable_search(NAME);
+        let first = || session("a", Kind::Chat, "2026-10-01T09:00:00Z", &["hello"]);
+        let newer = || session("a", Kind::Chat, "2026-10-03T09:00:00Z", &["hello", "and peas?"]);
+        let added = || session("b", Kind::Chat, "2026-10-03T10:00:00Z", &["beans"]);
+        let held = |source: &str| mock.search_index(NAME, |ix| ix.sources.get(source).map(|h| h.revision.clone())).flatten();
+        let a_spool = spooled(vec![first()]);
+        write(&c, NAME, &a_spool).unwrap();
+        let b_spool = spooled(vec![newer(), added()]);
+        write(&c, NAME, &b_spool).unwrap();
+        search::sync(&c, NAME, &b_spool);
+        let resumed = search::sync(&c, NAME, &a_spool);
+        assert_eq!(held("sessions/hermes-a.json"), Some(sha(&file_of(newer()))), "no older copy written back: {resumed:?}");
+        assert_eq!(held("sessions/hermes-b.json"), Some(sha(&file_of(added()))), "no conversation the other added removed: {resumed:?}");
+    }
+
+    /// Goal: a conversation another import wrote to search between this
+    /// one's reading search and writing it is left as the other wrote it:
+    /// this one's write names the revision it read (`replaces`), and is
+    /// refused and counted overtaken. Method: search read holding `a`'s
+    /// first copy; another import's newer copy written to the archive and
+    /// search since; this import's write of the first copy then.
+    #[test]
+    fn a_write_overtaken_leaves_the_others_as_it_wrote_it() {
+        let (mock, c) = archive();
+        mock.enable_search(NAME);
+        let first = || session("a", Kind::Chat, "2026-10-01T09:00:00Z", &["hello"]);
+        let newer = || session("a", Kind::Chat, "2026-10-03T09:00:00Z", &["hello", "and peas?"]);
+        let spool = spooled(vec![first()]);
+        write(&c, NAME, &spool).unwrap();
+        search::sync(&c, NAME, &spool);
+        let read = sha(&file_of(first()));
+        let other = spooled(vec![newer()]);
+        write(&c, NAME, &other).unwrap();
+        search::sync(&c, NAME, &other);
+        let storage = CodeStorage::connect(&c, NAME).unwrap();
+        let head = storage.branch_head(MAIN).unwrap().unwrap();
+        let entry = spool.entries["hermes-a"].clone();
+        let wrote = search::write_one(&c, NAME, &spool, &storage, &Blobs::new(&c, NAME), &head, "hermes-a", &entry, &read).unwrap();
+        assert_eq!(wrote, search::Wrote::Overtaken);
+        let now = mock.search_index(NAME, |ix| ix.sources["sessions/hermes-a.json"].revision.clone()).unwrap();
+        assert_eq!(now, sha(&file_of(newer())), "the other's write stands");
+    }
+
     /// Goal: a newer copy of a conversation replaces the one held, and an
     /// older copy never does. Method: a chat, then it later with a reply
     /// more, then the first copy again.

cli/src/session/search.rs

@@ -0,0 +1,209 @@
+//! An archive's conversations into the platform's search, written by the
+//! import that holds their bytes (fragment_core::session_search makes each
+//! one's sections) through the search write API: `GET
+//! /api/f/{name}/search/sources` and `POST …/search/upsert`.
+//!
+//! After an import's writes, the archive's search is reconciled with the
+//! archive **as it stands then**, not as this import left it: another
+//! import may have written since (newer copies, conversations of its own).
+//! Its indexes at main's head say each conversation's sha256, and its files
+//! changed since them are read (`held`, `found`); each conversation whose
+//! sha256 is not the revision search holds for its file is written, its
+//! bytes read at that head and its revision theirs. Each write names the
+//! revision it replaces, as search was read holding it (`replaces`), so a
+//! write another import made since is never written over: that one is
+//! left as it is (`overtaken`). Nothing is deleted here: the platform
+//! forgets a conversation's source when a move of main deletes its file
+//! (cell/src/searchable.rs), and what this import did not see is no
+//! evidence of a file gone. So an import that writes nothing still puts
+//! search right, and an archive made before its host had search is
+//! searched from its next import. A host without search (its routes 404),
+//! or one that cannot be reached, is said and skipped: the import stands.
+
+use std::collections::BTreeMap;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use anyhow::Result;
+use fragment_core::blob;
+use fragment_core::session::{self, Entry, Session};
+use fragment_core::{searchable, session_search};
+use fragment_proto::search::SOURCES_PAGE_MAX;
+use serde::{Deserialize, Serialize};
+
+use super::{said_id, Spool};
+use crate::api::{encode_q, Client};
+use crate::blobs::Blobs;
+use crate::codestorage::CodeStorage;
+
+/// Sources one listing asks for (the API's most).
+const PAGE: usize = SOURCES_PAGE_MAX;
+/// Pages read, at most: an archive's conversations, and a little more.
+const PAGES_MAX: usize = session::SESSIONS_MAX / PAGE + 4;
+/// Conversations written at once.
+const WRITERS: usize = 8;
+/// Conversations a refusal is said for, at most (the rest counted).
+const REFUSED_SAID: usize = 20;
+
+/// What an import did to its archive's search.
+#[derive(Debug, Serialize)]
+#[serde(tag = "state", rename_all = "snake_case")]
+pub enum Searched {
+    /// Nothing written: the host has no search, or it could not be reached.
+    Skipped { why: String },
+    Written {
+        written: usize,
+        unchanged: usize,
+        /// Conversations another import wrote to search between this one's
+        /// reading it and writing them: left as that one wrote them.
+        overtaken: usize,
+        /// Sections past search's limits, left out of their conversations.
+        left_out: usize,
+        /// Conversations search refused or could not take, each with why
+        /// (at most `REFUSED_SAID`), and how many in all.
+        refused: Vec<super::RefusedOne>,
+        refused_count: usize,
+    },
+}
+
+#[derive(Deserialize)]
+struct Page {
+    sources: Vec<Listed>,
+    next: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct Listed {
+    source: String,
+    revision: String,
+}
+
+/// The revision search holds of each source, or why there is none to read.
+fn listed(c: &Client, name: &str) -> std::result::Result<BTreeMap<String, String>, String> {
+    let mut held = BTreeMap::new();
+    let mut after: Option<String> = None;
+    for _ in 0..PAGES_MAX {
+        let from = after.as_deref().map(|a| format!("&after={}", encode_q(a))).unwrap_or_default();
+        let resp = c.get(&format!("/api/f/{name}/search/sources?limit={PAGE}{from}")).map_err(|e| format!("{e:#}"))?;
+        if resp.status == 404 {
+            return Err("this host has no search yet".into());
+        }
+        let page: Page = c.call_as(resp).map_err(|e| format!("{e:#}"))?;
+        held.extend(page.sources.into_iter().map(|s| (s.source, s.revision)));
+        match page.next {
+            None => return Ok(held),
+            Some(n) if after.as_ref().is_some_and(|a| n <= *a) => return Err(format!("search's sources stopped moving on at {}", said_id(&n))),
+            Some(n) => after = Some(n),
+        }
+    }
+    Err(format!("search lists more than {} sources", PAGES_MAX * PAGE))
+}
+
+/// One conversation's write: its sections the limits left out, or
+/// another's write since this import read search.
+#[derive(Debug, PartialEq)]
+pub(super) enum Wrote {
+    Written(usize),
+    Overtaken,
+}
+
+/// The archive at main's head as it stands now: the head, and each
+/// conversation its `sessions/` holds, by key, as its indexes describe it
+/// or, where they may not, as its file is (`found`).
+fn archive_now(c: &Client, name: &str, storage: &CodeStorage, blobs: &Blobs) -> Result<(Option<String>, BTreeMap<String, Entry>)> {
+    let h = super::held(storage, blobs)?;
+    let found: BTreeMap<String, Entry> = super::found(c, name, storage, &h, &BTreeMap::new())?.into_iter().map(|e| (e.key.clone(), e)).collect();
+    let entries = h.stored.keys().filter_map(|k| found.get(k).or_else(|| h.indexed.get(k)).map(|e| (k.clone(), e.clone()))).collect();
+    Ok((h.head, entries))
+}
+
+/// `name`'s search put right with the archive as it stands now, each
+/// conversation read from `spool` when the import holds those bytes, else
+/// from the archive at the head read.
+pub fn sync(c: &Client, name: &str, spool: &Spool) -> Searched {
+    let held = match listed(c, name) {
+        Ok(held) => held,
+        Err(why) => return Searched::Skipped { why },
+    };
+    let storage = match CodeStorage::connect(c, name) {
+        Ok(s) => s,
+        Err(e) => return Searched::Skipped { why: format!("{e:#}") },
+    };
+    let (head, entries) = match archive_now(c, name, &storage, &Blobs::new(c, name)) {
+        Ok(now) => now,
+        Err(e) => return Searched::Skipped { why: format!("reading the archive: {e:#}") },
+    };
+    let Some(head) = head else { return Searched::Written { written: 0, unchanged: 0, overtaken: 0, left_out: 0, refused: Vec::new(), refused_count: 0 } };
+    let todo: Vec<(&String, &Entry)> = entries.iter().filter(|(k, e)| held.get(&session::session_path(k)) != Some(&e.sha256)).collect();
+    // a refusal that would be every one's (not allowed, no route): stop
+    let stop = AtomicBool::new(false);
+    let write = |todo: &[(&String, &Entry)]| -> Vec<std::result::Result<Wrote, (String, String)>> {
+        let blobs = Blobs::new(c, name);
+        let mut out = Vec::new();
+        for (key, e) in todo {
+            if stop.load(Ordering::Relaxed) {
+                break;
+            }
+            let replaces = held.get(&session::session_path(key)).map_or("", String::as_str);
+            out.push(write_one(c, name, spool, &storage, &blobs, &head, key, e, replaces).map_err(|(status, why)| {
+                if matches!(status, Some(401 | 403 | 404)) {
+                    stop.store(true, Ordering::Relaxed);
+                }
+                (said_id(key), why)
+            }));
+        }
+        out
+    };
+    let per = todo.len().div_ceil(WRITERS).max(1);
+    let results: Vec<_> = std::thread::scope(|scope| {
+        let writers: Vec<_> = todo.chunks(per).map(|part| scope.spawn(move || write(part))).collect();
+        writers.into_iter().flat_map(|w| w.join().unwrap_or_else(|_| vec![Err(("a writer".into(), "panicked".into()))])).collect()
+    });
+    let (mut written, mut overtaken, mut left_out, mut refused, mut refused_count) = (0, 0, 0, Vec::new(), 0);
+    for r in results {
+        match r {
+            Ok(Wrote::Written(left)) => {
+                written += 1;
+                left_out += left;
+            }
+            Ok(Wrote::Overtaken) => overtaken += 1,
+            Err((id, why)) => {
+                refused_count += 1;
+                if refused.len() < REFUSED_SAID {
+                    refused.push(super::RefusedOne { id, why });
+                }
+            }
+        }
+    }
+    if stop.load(Ordering::Relaxed) {
+        let why = refused.first().map_or_else(String::new, |r| r.why.clone());
+        return Searched::Skipped { why: format!("search refused this archive's writes: {why}") };
+    }
+    Searched::Written { written, unchanged: entries.len() - todo.len(), overtaken, left_out, refused, refused_count }
+}
+
+/// One conversation written, its bytes as the import spooled them when
+/// they are the archive's at `head`, else read at `head`; its revision their
+/// own sha256, written only over `replaces` (the revision search was read
+/// holding, `""` for none). A refusal says its status (when search
+/// answered) and why.
+#[allow(clippy::too_many_arguments)]
+pub(super) fn write_one(c: &Client, name: &str, spool: &Spool, storage: &CodeStorage, blobs: &Blobs, head: &str, key: &str, e: &Entry, replaces: &str) -> std::result::Result<Wrote, (Option<u16>, String)> {
+    let path = session::session_path(key);
+    let bytes = if spool.entries.get(key).is_some_and(|s| s.sha256 == e.sha256) {
+        std::fs::read(spool.path(key)).map_err(|err| (None, format!("reading it back: {err}")))?
+    } else {
+        storage.read_file(&path, head).map_err(anyhow::Error::new).and_then(|b| blobs.resolve(b)).map_err(|err| (None, format!("reading it: {err:#}")))?
+    };
+    let s: Session = serde_json::from_slice(&bytes).map_err(|err| (None, format!("it is no conversation: {err}")))?;
+    let mut indexed = session_search::upsert_of(&path, &blob::sha256_hex(&bytes), &s);
+    indexed.upsert.replaces = Some(replaces.to_string());
+    // what the platform would refuse is refused here, unsent
+    searchable::check(&indexed.upsert).map_err(|why| (None, why))?;
+    match c.post_json_by_id(&format!("/api/f/{name}/search/upsert"), &indexed.upsert) {
+        Ok(r) if r.ok() => Ok(Wrote::Written(indexed.left)),
+        // another import's write since this one read search: theirs stands
+        Ok(r) if r.status == 409 => Ok(Wrote::Overtaken),
+        Ok(r) => Err((Some(r.status), r.refusal().msg)),
+        Err(err) => Err((None, format!("{err:#}"))),
+    }
+}

crates/core/src/lib.rs

@@ -57,6 +57,7 @@ pub mod search;
 pub mod seal;
 pub mod secrets;
 pub mod session;
+pub mod session_search;
 pub mod secrets_store;
 pub mod webpush;
 pub mod site;