diff --git a/CHANGELOG.md b/CHANGELOG.md index 97cd1ad5..9a1dfd71 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -184,6 +184,28 @@ Notable changes to the native `ai-hist` CLI are documented here. ### Rust API +- `ChangeQuery::session(source, session_id)` restricts a change-feed drain + to one session: the same `Change` values the unfiltered drain reports for + that session, tombstones included, read through each table's + `(source, session_id)` index rather than the whole revision range. It is a + one-shot read (a session's backfill), so naming a consumer alongside it is + `Error::InvalidArgument`. A prompt with no session belongs to no session's + drain, and a trajectory is the session its id names under the `trajectory` + source. +- `SessionStore::session_identities(IdentityQuery { after, limit })` pages + every `(source_name, session_id)` the store holds evidence under, catalogued + or not: the union of the catalog, prompts, events, tool calls, file edits, + markers, relationships (by parent), presences, commit links, connector + observations and trajectories. It is a merge of covering index seeks, so no + payload is read; each page reads one snapshot, and an empty session id is + no session. `SessionIdentity` names a stored session, including one + under a source this build does not know, and is the type + `ChangeQuery::session` holds. `storage::session_identities_after` is the + same read, and now covers every one of those tables rather than the catalog, + prompts and events alone, and reads each page on one snapshot even on an + autocommit connection. `sessions` gains `idx_sessions_identity` on + `(source, session_id)`, added by the first writable open; until then a + read-only store answers `session_identities` with `StaleSchema`. - The change feed reports every evidence table and each row exactly as stored. `ChangeKind` gains `History`, `Presence`, `CommitLink`, `Trajectory`, `SourceObservation` and `ObservationEvidence` (`ALL` lists diff --git a/crates/ai-hist/README.md b/crates/ai-hist/README.md index d0641591..cf23f247 100644 --- a/crates/ai-hist/README.md +++ b/crates/ai-hist/README.md @@ -29,7 +29,7 @@ fn main() -> Result<(), ai_hist::Error> { } ``` -Default features expose `SessionStore` — `open`, `discover`, `sync`, `hydrate`, `watch`, `sessions`, `session`, `changes_since`, `head_revision` — plus `Source::capabilities()`, the typed evidence structs and one `Error` enum; see `docs/sourcing-sdk.md` in the repository. Optional features: `export` (consistent snapshots and change capture), `opencode-backup`, `git-hooks`, `fs-events`. The legacy `delivery` feature aliases `export`; upload jobs and workers are owned by the probe package, `plugins/relayhistory/rust`. Workspace crates enable `unstable-internal` for connection-level maintenance APIs. +Default features expose `SessionStore` — `open`, `discover`, `sync`, `hydrate`, `watch`, `sessions`, `session`, `session_identities`, `changes_since`, `head_revision` — plus `Source::capabilities()`, the typed evidence structs and one `Error` enum; see `docs/sourcing-sdk.md` in the repository. Optional features: `export` (consistent snapshots and change capture), `opencode-backup`, `git-hooks`, `fs-events`. The legacy `delivery` feature aliases `export`; upload jobs and workers are owned by the probe package, `plugins/relayhistory/rust`. Workspace crates enable `unstable-internal` for connection-level maintenance APIs. The embedder guide — store options, the sync lifecycle and its locking, the evidence model, accounting semantics, versioning and feature flags — is [`docs/sourcing-sdk.md`](https://github.com/AgentWorkforce/relayhistory/blob/main/docs/sourcing-sdk.md), and [`examples/rust-consumer`](https://github.com/AgentWorkforce/relayhistory/tree/main/examples/rust-consumer) is a standalone project that consumes this crate from crates.io. Cargo semver is the contract; the crate's default-feature public API is snapshotted in the repository at `crates/ai-hist/public-api.txt` and diffed in CI. diff --git a/crates/ai-hist/public-api.txt b/crates/ai-hist/public-api.txt index c0f6312f..b2201f9d 100644 --- a/crates/ai-hist/public-api.txt +++ b/crates/ai-hist/public-api.txt @@ -315,10 +315,12 @@ pub ai_hist::Change::source_name: alloc::string::String pub ai_hist::ChangeQuery::batch: usize pub ai_hist::ChangeQuery::consumer: core::option::Option pub ai_hist::ChangeQuery::kinds: core::option::Option> +pub ai_hist::ChangeQuery::session: core::option::Option impl ai_hist::ChangeQuery pub fn ai_hist::ChangeQuery::batch(self, usize) -> Self pub fn ai_hist::ChangeQuery::consumer(self, impl core::convert::Into) -> Self pub fn ai_hist::ChangeQuery::kinds(self, impl core::iter::traits::collect::IntoIterator) -> Self +pub fn ai_hist::ChangeQuery::session(self, impl core::convert::Into, impl core::convert::Into) -> Self pub struct ai_hist::Changes impl ai_hist::Changes pub fn ai_hist::Changes::commit(&self) -> anyhow::Result @@ -384,6 +386,12 @@ pub ai_hist::HydrateReport::diagnostics: alloc::vec::Vec pub ai_hist::HydrateReport::related: alloc::vec::Vec pub ai_hist::HydrateReport::session: ai_hist::SessionRef pub ai_hist::HydrateReport::status: ai_hist::HydrateStatus +#[non_exhaustive] pub struct ai_hist::IdentityQuery +pub ai_hist::IdentityQuery::after: core::option::Option +pub ai_hist::IdentityQuery::limit: usize +impl ai_hist::IdentityQuery +pub fn ai_hist::IdentityQuery::after(self, ai_hist::SessionIdentity) -> Self +pub fn ai_hist::IdentityQuery::limit(self, usize) -> Self #[non_exhaustive] pub struct ai_hist::Marker pub ai_hist::Marker::kind: alloc::string::String pub ai_hist::Marker::marker_uid: alloc::string::String @@ -558,6 +566,12 @@ pub ai_hist::SessionFileEdit::tool_name: core::option::Option pub ai_hist::SessionFileEdit::user_modified: core::option::Option +#[non_exhaustive] pub struct ai_hist::SessionIdentity +pub ai_hist::SessionIdentity::session_id: alloc::string::String +pub ai_hist::SessionIdentity::source_name: alloc::string::String +impl ai_hist::SessionIdentity +pub fn ai_hist::SessionIdentity::new(impl core::convert::Into, impl core::convert::Into) -> Self +pub fn ai_hist::SessionIdentity::source(&self) -> core::option::Option pub struct ai_hist::SessionMarker pub ai_hist::SessionMarker::id: i64 pub ai_hist::SessionMarker::kind: alloc::string::String @@ -635,6 +649,8 @@ pub fn ai_hist::SessionStore::session(&self, &ai_hist::SessionRef, ai_hist::Sess pub fn ai_hist::SessionStore::sessions(&self, ai_hist::CatalogQuery) -> ai_hist::CatalogIter pub fn ai_hist::SessionStore::sync(&self, ai_hist::SyncOptions) -> core::result::Result pub fn ai_hist::SessionStore::watch(&self, ai_hist::WatchOptions) -> core::result::Result +impl ai_hist::SessionStore +pub fn ai_hist::SessionStore::session_identities(&self, ai_hist::IdentityQuery) -> core::result::Result, ai_hist::Error> pub struct ai_hist::SessionToolCall pub ai_hist::SessionToolCall::args_json: core::option::Option pub ai_hist::SessionToolCall::id: i64 diff --git a/crates/ai-hist/src/change_feed.rs b/crates/ai-hist/src/change_feed.rs index 7ff4e3ab..7a66acb4 100644 --- a/crates/ai-hist/src/change_feed.rs +++ b/crates/ai-hist/src/change_feed.rs @@ -44,6 +44,7 @@ use crate::discover::{row_to_session, ShallowSession, SESSION_COLUMNS}; use crate::relationship_graph::{map_relationship, SessionRelationship, RELATIONSHIP_COLUMNS}; +use crate::session_identities::SessionIdentity; use crate::session_store::{Error, SessionStore, Source}; use crate::store::{ ensure_columns, migration_applied, open_db, open_db_readonly, row_to_file_edit, @@ -724,6 +725,8 @@ pub struct ChangeQuery { /// Rows per page, clamped to `1..=`[`MAX_CHANGE_BATCH`]; zero means /// [`DEFAULT_CHANGE_BATCH`]. pub batch: usize, + /// Report only this session's changes; see [`ChangeQuery::session`]. + pub session: Option, } impl ChangeQuery { @@ -744,6 +747,32 @@ impl ChangeQuery { self.batch = batch; self } + + /// Report only one session's changes: exactly those whose + /// [`Change::source_name`] and [`Change::session_id`] are these, read + /// through each table's session index rather than the whole revision + /// range. `source` is the stored name; one this build does not know is + /// accepted. + /// + /// That is every kind that stores a session -- the catalog row, events, + /// tool calls, file edits, markers, relationships (under their parent + /// session), prompts, presences, commit links and connector + /// observations -- plus a trajectory, whose session is its own id under + /// the `trajectory` source. A prompt that names no session belongs to no + /// session's drain, and nor does a prompt's delete: a prompt's session is + /// not part of its identity, so its tombstone carries none. Both still + /// reach the unfiltered feed. + /// + /// A session drain is a one-shot read -- the backfill of a session an + /// embedder has just started following -- and cannot name a consumer + /// ([`Error::InvalidArgument`]): a cursor is a position in one stream, + /// and a position reached reading one session accounts for nothing + /// about the others. Each page re-seeks the session, so a long session + /// drains fastest with a large [`ChangeQuery::batch`]. + pub fn session(mut self, source: impl Into, session_id: impl Into) -> Self { + self.session = Some(SessionIdentity::new(source, session_id)); + self + } } /// The drain [`SessionStore::changes_since`] hands back. @@ -761,6 +790,7 @@ pub struct Changes { conn: Connection, kinds: Vec, consumer: Option, + session: Option, batch: usize, head: Watermark, position: Watermark, @@ -859,7 +889,8 @@ impl Changes { // did, and therefore above the cut. The second pass fetches the // rows in `(lo, cut]`, which is exactly the page, because // revisions are unique per write. - let Some(cut) = page_cut(&snapshot, &self.kinds, lo, hi, self.batch)? else { + let session = self.session.as_ref(); + let Some(cut) = page_cut(&snapshot, &self.kinds, session, lo, hi, self.batch)? else { // The key pass itself found nothing left: that, and only // that, is exhaustion. self.exhausted = true; @@ -868,11 +899,14 @@ impl Changes { }; let mut rows: Vec = Vec::with_capacity(self.batch); for kind in &self.kinds { - rows.extend(read_upserts(&snapshot, *kind, lo, cut, self.batch)?); + rows.extend(read_upserts( + &snapshot, *kind, session, lo, cut, self.batch, + )?); } rows.extend(read_tombstones( &snapshot, &self.kinds, + session, lo, cut, self.batch, @@ -929,6 +963,7 @@ impl std::fmt::Debug for Changes { .field("db_path", &self.db_path) .field("kinds", &self.kinds) .field("consumer", &self.consumer) + .field("session", &self.session) .field("batch", &self.batch) .field("head", &self.head) .field("position", &self.position) @@ -988,6 +1023,23 @@ impl SessionStore { .to_string(), )); } + if let Some(session) = &query.session { + if query.consumer.is_some() { + return Err(Error::InvalidArgument( + "changes_since: ChangeQuery::session is a one-shot read and cannot name a \ + consumer; a cursor committed from one session's changes would skip every \ + other session's. Drain the session from Watermark::START without a \ + consumer, and keep the named cursor for the whole feed" + .to_string(), + )); + } + if session.source_name.is_empty() || session.session_id.is_empty() { + return Err(Error::InvalidArgument( + "changes_since: ChangeQuery::session needs a nonempty source and session id" + .to_string(), + )); + } + } let (start, head, stale_cursor) = resolve_start_and_head(&conn, from, query.consumer.as_deref(), &kind_set)?; if from != Watermark::START && from != Watermark::CONSUMER && from.epoch != head.epoch { @@ -1011,6 +1063,7 @@ impl SessionStore { conn, kinds: kind_set.kinds, consumer: query.consumer, + session: query.session, batch, head, position: start, @@ -1092,31 +1145,101 @@ fn resolve_start_and_head( Ok((in_store(start), head, stale_cursor)) } +/// The revision range, and its order, for one page read. +/// +/// Unfiltered, it reads the revision index. Restricted to one session, the +/// session's own `(source, session_id, ...)` index is the narrow one: the +/// range is written `+revision`, which no index can serve, so the planner +/// seeks the session and sorts only that session's rows, rather than walking +/// the whole revision range to discard every other session's. +fn revision_range(first: usize, filtered: bool) -> (String, String) { + let column = if filtered { + format!("+{REVISION_COLUMN}") + } else { + REVISION_COLUMN.to_string() + }; + ( + format!("{column} > ?{first} AND {column} <= ?{}", first + 1), + format!("ORDER BY {column} ASC LIMIT ?{}", first + 2), + ) +} + +/// The session predicate for one kind's upsert reads, over parameters +/// `?4` (source) and `?5` (session id), or nothing when unfiltered. +fn upsert_session_sql(kind: ChangeKind, filtered: bool) -> String { + if !filtered { + return String::new(); + } + let table = kind.table(); + format!( + " AND {} = ?4 AND {}.{} = ?5", + table.source_sql(table.name), + table.name, + table.session + ) +} + +/// The session predicate for tombstone reads, over `?5` and `?6`. +fn tombstone_session_sql(filtered: bool) -> &'static str { + if filtered { + " AND source = ?5 AND session_id = ?6" + } else { + "" + } +} + /// The revision-only page query for one kind: a covering read of the -/// revision index. -fn upsert_key_sql(kind: ChangeKind) -> String { +/// revision index, or of the session index when restricted to one session. +fn upsert_key_sql(kind: ChangeKind, filtered: bool) -> String { + let (range, order) = revision_range(1, filtered); format!( - "SELECT {REVISION_COLUMN} FROM {name} \ - WHERE {REVISION_COLUMN} > ?1 AND {REVISION_COLUMN} <= ?2 \ - ORDER BY {REVISION_COLUMN} ASC LIMIT ?3", + "SELECT {REVISION_COLUMN} FROM {name} WHERE {range}{session} {order}", name = kind.table().name, + session = upsert_session_sql(kind, filtered), ) } -fn tombstone_key_sql() -> String { +fn tombstone_key_sql(filtered: bool) -> String { + let (range, order) = revision_range(2, filtered); format!( "SELECT {REVISION_COLUMN} FROM evidence_tombstones \ - WHERE kind = ?1 AND {REVISION_COLUMN} > ?2 AND {REVISION_COLUMN} <= ?3 \ - ORDER BY {REVISION_COLUMN} ASC LIMIT ?4" + WHERE kind = ?1 AND {range}{session} {order}", + session = tombstone_session_sql(filtered), ) } +/// The parameters a page read binds: the range and limit, then the session +/// when the drain is restricted to one. +fn page_params( + kind: Option, + session: Option<&SessionIdentity>, + lo: u64, + hi: u64, + batch: usize, +) -> Vec { + let mut values: Vec = Vec::with_capacity(6); + if let Some(kind) = kind { + values.push(kind.as_str().to_string().into()); + } + values.extend([ + (lo as i64).into(), + (hi as i64).into(), + (batch as i64).into(), + ]); + if let Some(session) = session { + values.push(session.source_name.clone().into()); + values.push(session.session_id.clone().into()); + } + values +} + /// The highest revision of the next page: the `batch`-th smallest revision /// in `(lo, hi]` across every stream, or `hi` when fewer remain, or `None` /// when nothing does. Reads revisions only. fn page_cut( conn: &Connection, kinds: &[ChangeKind], + session: Option<&SessionIdentity>, lo: u64, hi: u64, batch: usize, @@ -1133,20 +1256,18 @@ fn page_cut( } Ok(()) }; - let range: [rusqlite::types::Value; 3] = [ - (lo as i64).into(), - (hi as i64).into(), - (batch as i64).into(), - ]; + let filtered = session.is_some(); + let range = page_params(None, session, lo, hi, batch); for kind in kinds { - let mut statement = conn.prepare_cached(&upsert_key_sql(*kind))?; + let mut statement = conn.prepare_cached(&upsert_key_sql(*kind, filtered))?; collect(&mut statement, &range)?; } - let mut tombstones = conn.prepare_cached(&tombstone_key_sql())?; + let mut tombstones = conn.prepare_cached(&tombstone_key_sql(filtered))?; for kind in kinds { - let mut values: Vec = vec![kind.as_str().to_string().into()]; - values.extend(range.iter().cloned()); - collect(&mut tombstones, &values)?; + collect( + &mut tombstones, + &page_params(Some(*kind), session, lo, hi, batch), + )?; } if revisions.is_empty() { return Ok(None); @@ -1284,7 +1405,7 @@ fn commit_cursor( /// source, session, record key -- computed by the same SQL the triggers use, /// so an upsert and a delete of one record name it identically; the stored /// columns; and the revision. -fn upsert_sql(kind: ChangeKind, stored: &[String]) -> String { +fn upsert_sql(kind: ChangeKind, stored: &[String], filtered: bool) -> String { let table = kind.table(); let name = table.name; let mut select = Vec::new(); @@ -1299,17 +1420,19 @@ fn upsert_sql(kind: ChangeKind, stored: &[String]) -> String { .iter() .map(|column| format!("{name}.\"{}\"", column.replace('"', "\"\""))), ); + let (range, order) = revision_range(1, filtered); format!( "SELECT {select}, {name}.{REVISION_COLUMN} FROM {name} \ - WHERE {REVISION_COLUMN} > ?1 AND {REVISION_COLUMN} <= ?2 \ - ORDER BY {REVISION_COLUMN} ASC LIMIT ?3", + WHERE {range}{session} {order}", select = select.join(", "), + session = upsert_session_sql(kind, filtered), ) } fn read_upserts( conn: &Connection, kind: ChangeKind, + session: Option<&SessionIdentity>, lo: u64, hi: u64, batch: usize, @@ -1320,10 +1443,11 @@ fn read_upserts( .iter() .map(|column| Arc::from(column.as_str())) .collect(); - let mut statement = conn.prepare_cached(&upsert_sql(kind, &stored))?; + let mut statement = conn.prepare_cached(&upsert_sql(kind, &stored, session.is_some()))?; // Everything before the identity is the typed row. let identity = statement.column_count() - stored.len() - 4; - let rows = statement.query_map(params![lo as i64, hi as i64, batch as i64], |row| { + let values = page_params(None, session, lo, hi, batch); + let rows = statement.query_map(rusqlite::params_from_iter(values), |row| { let evidence = match kind { ChangeKind::Session => EvidenceRow::Session(row_to_session(row)?), ChangeKind::SessionEvent => EvidenceRow::SessionEvent(row_to_session_event(row)?), @@ -1380,27 +1504,29 @@ fn read_upserts( /// list: the tombstone index is `(kind, revision)`, and a single-kind /// equality is what lets the range come back in revision order without a /// temporary sort. -fn tombstone_sql() -> String { +fn tombstone_sql(filtered: bool) -> String { + let (range, order) = revision_range(2, filtered); format!( "SELECT source, session_id, record_key, {REVISION_COLUMN} FROM evidence_tombstones \ - WHERE kind = ?1 AND {REVISION_COLUMN} > ?2 AND {REVISION_COLUMN} <= ?3 \ - ORDER BY {REVISION_COLUMN} ASC LIMIT ?4" + WHERE kind = ?1 AND {range}{session} {order}", + session = tombstone_session_sql(filtered), ) } fn read_tombstones( conn: &Connection, kinds: &[ChangeKind], + session: Option<&SessionIdentity>, lo: u64, hi: u64, batch: usize, ) -> Result> { let mut changes = Vec::new(); - let mut statement = conn.prepare_cached(&tombstone_sql())?; + let mut statement = conn.prepare_cached(&tombstone_sql(session.is_some()))?; for kind in kinds { let table = kind.table(); let rows = statement.query_map( - params![kind.as_str(), lo as i64, hi as i64, batch as i64], + rusqlite::params_from_iter(page_params(Some(*kind), session, lo, hi, batch)), |row| { Ok(( row.get::<_, String>(0)?, @@ -2563,21 +2689,33 @@ mod tests { let mut plans = Vec::new(); for kind in ChangeKind::ALL { let stored = stored_columns(&conn, kind.table().name).unwrap(); - plans.push((kind.table().name, upsert_sql(*kind, &stored), page.clone())); + plans.push(( + kind.table().name, + upsert_sql(*kind, &stored, false), + page.clone(), + )); } let mut tombstone_page = vec![rusqlite::types::Value::from("session_event".to_string())]; tombstone_page.extend(page.iter().cloned()); plans.push(( "evidence_tombstones", - tombstone_sql(), + tombstone_sql(false), tombstone_page.clone(), )); // The revision-only first pass must be a covering read of the same // index, with no table access at all. for kind in ChangeKind::ALL { - plans.push((kind.table().name, upsert_key_sql(*kind), page.clone())); + plans.push(( + kind.table().name, + upsert_key_sql(*kind, false), + page.clone(), + )); } - plans.push(("evidence_tombstones", tombstone_key_sql(), tombstone_page)); + plans.push(( + "evidence_tombstones", + tombstone_key_sql(false), + tombstone_page, + )); for (table, sql, values) in plans { let details: Vec = conn .prepare(&format!("EXPLAIN QUERY PLAN {sql}")) @@ -2601,6 +2739,63 @@ mod tests { } } + /// Restricted to one session, every page read seeks that session through + /// its table's session index -- never the revision index, never a scan of + /// the table -- and sorts only that session's rows. + #[test] + fn a_session_page_reads_the_session_index() { + let dir = tempfile::tempdir().unwrap(); + let (_store, conn) = store(dir.path()); + let filter = SessionIdentity::new("claude", "s1"); + let mut plans = Vec::new(); + for kind in ChangeKind::ALL { + let stored = stored_columns(&conn, kind.table().name).unwrap(); + let page = page_params(None, Some(&filter), 0, 1_000, 100); + plans.push(( + kind.table().name, + upsert_sql(*kind, &stored, true), + page.clone(), + )); + plans.push((kind.table().name, upsert_key_sql(*kind, true), page)); + } + for sql in [tombstone_sql(true), tombstone_key_sql(true)] { + plans.push(( + "evidence_tombstones", + sql, + page_params(Some(ChangeKind::SessionEvent), Some(&filter), 0, 1_000, 100), + )); + } + for (table, sql, values) in plans { + let plan = conn + .prepare(&format!("EXPLAIN QUERY PLAN {sql}")) + .unwrap() + .query_map(rusqlite::params_from_iter(values), |row| { + row.get::<_, String>(3) + }) + .unwrap() + .collect::>>() + .unwrap() + .join("\n"); + assert!( + plan.contains(&format!("SEARCH {table} USING")), + "{table}: {plan}" + ); + assert!(!plan.contains(&format!("SCAN {table}")), "{table}: {plan}"); + assert!(!plan.contains("_revision"), "{table}: {plan}"); + // The trajectory's session is its primary key; every other + // table's seek binds both the source and the session. + if table == "trajectories" { + assert!(plan.contains("(id=?)"), "{table}: {plan}"); + } else { + assert!( + plan.contains("source=?") && plan.contains("session_id=?"), + "{table}: {plan}" + ); + } + eprintln!("{table}: {}", plan.replace('\n', " | ")); + } + } + /// One page holds exactly `batch` typed rows however many streams feed /// it: the first pass reads revisions only and the second fetches just /// the rows below the cut, so six tables and their tombstones cannot @@ -2655,24 +2850,24 @@ mod tests { // round wrote revisions 7k+1..7k+7, and the marker at 7k+5 is gone // (its tombstone sits at 7k+7), so the fifth smallest is 6. assert_eq!( - page_cut(&conn, ChangeKind::ALL, 0, head, 5).unwrap(), + page_cut(&conn, ChangeKind::ALL, None, 0, head, 5).unwrap(), Some(6) ); assert_eq!( - page_cut(&conn, ChangeKind::ALL, 20, head, 100).unwrap(), + page_cut(&conn, ChangeKind::ALL, None, 20, head, 100).unwrap(), Some(head), "fewer than a page left: the cut is the head" ); assert_eq!( - page_cut(&conn, ChangeKind::ALL, head, head, 5).unwrap(), + page_cut(&conn, ChangeKind::ALL, None, head, head, 5).unwrap(), None ); // Only the rows below the cut are fetched, across every kind. let fetched: usize = ChangeKind::ALL .iter() - .map(|kind| read_upserts(&conn, *kind, 0, 6, 5).unwrap().len()) + .map(|kind| read_upserts(&conn, *kind, None, 0, 6, 5).unwrap().len()) .sum::() - + read_tombstones(&conn, ChangeKind::ALL, 0, 6, 5) + + read_tombstones(&conn, ChangeKind::ALL, None, 0, 6, 5) .unwrap() .len(); assert_eq!(fetched, 5); @@ -2835,6 +3030,7 @@ mod tests { conn: reader, kinds: ChangeKind::ALL.to_vec(), consumer: None, + session: None, batch: 3, head, position: Watermark::START, diff --git a/crates/ai-hist/src/lib.rs b/crates/ai-hist/src/lib.rs index 37c15f18..88b2787a 100644 --- a/crates/ai-hist/src/lib.rs +++ b/crates/ai-hist/src/lib.rs @@ -39,6 +39,7 @@ workspace_mod!(watch); mod change_feed; mod file_lock; mod jsonl_temp; +mod session_identities; mod session_store; mod session_usage; mod usage; @@ -77,6 +78,7 @@ pub use change_feed::{ pub use discover::{declared_evidence_kinds, missing_evidence_kinds, ShallowSession}; pub use ingest::CaptureProgress; pub use relationship_graph::{RelationshipCapabilities, SessionRelationship}; +pub use session_identities::{IdentityQuery, SessionIdentity}; pub use session_store::{ Block, BlockKind, Capability, CatalogIter, CatalogQuery, CatalogSession, ControlKind, Diagnostic, DiscoveryOptions, DiscoveryReport, DiscoveryState, Error, FileEdit, HydrateOptions, diff --git a/crates/ai-hist/src/session_identities.rs b/crates/ai-hist/src/session_identities.rs new file mode 100644 index 00000000..04369c3c --- /dev/null +++ b/crates/ai-hist/src/session_identities.rs @@ -0,0 +1,461 @@ +//! Every session identity the store holds evidence for, keyset-paged. +//! +//! The catalog lists the sessions discovery found, but evidence can arrive +//! for a session the catalog never names: a prompt from a provider's prompt +//! log, a subagent sidechain's events, a connector's observation. A reader +//! deciding what exists -- a consent baseline, a status count -- needs all +//! of them, and needs them without reading a single payload. +//! +//! So the read is a merge of one ordered index per table. Each table offers +//! its next `(source, session_id)` after the cursor through an index that +//! leads with those two columns -- one bounded seek, however many rows the +//! session holds -- and the smallest offer is the next identity. A page costs +//! a seek per identity per table that holds it, never a scan. + +use crate::session_store::{Error, SessionStore, Source}; +use crate::store::{open_db_readonly, schema_is_identity_read_current}; +use anyhow::Result; +use rusqlite::{Connection, OptionalExtension}; + +/// Largest page one [`SessionStore::session_identities`] call returns. +const MAX_IDENTITY_PAGE: usize = 10_000; + +/// Page size when [`IdentityQuery::limit`] is zero. +const DEFAULT_IDENTITY_PAGE: usize = 1_000; + +/// One session as the store names it: the stored source and session id. +/// +/// `source_name` is the stored text, so a session under a source this build +/// does not know is still one identity; [`SessionIdentity::source`] parses +/// it. It is the same pair a [`crate::Change`] carries in +/// [`crate::Change::source_name`] and [`crate::Change::session_id`], and +/// what [`crate::ChangeQuery::session`] restricts a drain to. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +#[non_exhaustive] +pub struct SessionIdentity { + pub source_name: String, + pub session_id: String, +} + +impl SessionIdentity { + pub fn new(source_name: impl Into, session_id: impl Into) -> Self { + Self { + source_name: source_name.into(), + session_id: session_id.into(), + } + } + + /// The source, or `None` for a name this build does not know. + pub fn source(&self) -> Option { + Source::parse(&self.source_name) + } +} + +/// How to page [`SessionStore::session_identities`]. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +#[non_exhaustive] +pub struct IdentityQuery { + /// The last identity of the previous page; `None` starts from the first. + pub after: Option, + /// Identities per page, clamped to `1..=10_000`; zero means 1,000. + pub limit: usize, +} + +impl IdentityQuery { + /// Continue after this identity. + pub fn after(mut self, identity: SessionIdentity) -> Self { + self.after = Some(identity); + self + } + + /// Identities per page; see [`IdentityQuery::limit`]. + pub fn limit(mut self, limit: usize) -> Self { + self.limit = limit; + self + } +} + +impl SessionStore { + /// Every session the store holds anything for, distinct, in + /// `(source_name, session_id)` byte order, one keyset page at a time. + /// + /// A session counts when any evidence table stores a row under it: the + /// catalog, prompts, events, tool calls, file edits, markers, + /// relationships (under their parent), presences, commit links, + /// connector observations and their evidence, and trajectories (under the + /// `trajectory` source, by id). A prompt that names no session is under + /// none, and a child session only an edge names is not one until + /// something is stored under it -- the same pairs the change feed + /// reports in [`crate::Change::session_id`]. + /// + /// Continue with the last identity returned; an empty page is the end. + /// Each page is read on one snapshot, so identities written between + /// pages are seen only if they sort after the cursor. A session id that + /// is empty names no session, and is not an identity. No payload is read: + /// every step is a seek in an index that leads with the source and + /// session. + pub fn session_identities( + &self, + query: IdentityQuery, + ) -> std::result::Result, Error> { + let limit = match query.limit { + 0 => DEFAULT_IDENTITY_PAGE, + limit => limit.min(MAX_IDENTITY_PAGE), + }; + let conn = open_db_readonly(self.db_path()) + .map_err(|error| Error::DatabaseOpen(format!("{error:#}")))?; + // The gate is here rather than at `open`, like the change feed's: a + // read-only store over a database without the catalog's identity + // index keeps every other read, and is told how to get this one. + if !schema_is_identity_read_current(&conn).map_err(Error::query)? { + return Err(Error::stale_schema(self.db_path(), "session-identity")); + } + let page = identities_after( + &conn, + query + .after + .as_ref() + .map(|after| (after.source_name.as_str(), after.session_id.as_str())), + limit, + ) + .map_err(Error::query)?; + Ok(page + .into_iter() + .map(|(source_name, session_id)| SessionIdentity { + source_name, + session_id, + }) + .collect()) + } +} + +/// Every table a session identity can be stored in: its name and the column +/// naming the session. Each has an index leading with `(source, session)`. +const IDENTITY_TABLES: &[(&str, &str)] = &[ + ("sessions", "session_id"), + ("history", "session_id"), + ("session_events", "session_id"), + ("tool_calls", "session_id"), + ("file_edits", "session_id"), + ("session_markers", "session_id"), + ("session_relationships", "parent_session_id"), + ("session_presences", "session_id"), + ("session_commit_links", "session_id"), + ("session_observations", "session_id"), + ("observation_evidence", "session_id"), +]; + +/// A trajectory is a session of its own, under a source it does not store. +const TRAJECTORY_SOURCE: &str = "trajectory"; + +/// The first identity in one table, or the first after a cursor. An empty +/// session id names no session -- `ChangeQuery::session` refuses one -- so +/// it is skipped like NULL. +fn seek_sql(table: &str, session: &str, after: bool) -> String { + let range = if after { + format!(" AND (source, {session}) > (?1, ?2)") + } else { + String::new() + }; + format!( + "SELECT source, {session} FROM {table} \ + WHERE {session} IS NOT NULL AND {session} <> ''{range} \ + ORDER BY source, {session} LIMIT 1" + ) +} + +fn trajectory_sql(after: bool) -> &'static str { + if after { + "SELECT id FROM trajectories WHERE id > ?1 ORDER BY id LIMIT 1" + } else { + "SELECT id FROM trajectories WHERE id > '' ORDER BY id LIMIT 1" + } +} + +type Identity = (String, String); + +/// One table's next identity strictly after `after`. +fn next_in( + conn: &Connection, + arm: Option<(&str, &str)>, + after: Option<(&str, &str)>, +) -> Result> { + let Some((table, session)) = arm else { + // The trajectory arm: its identities all share one source, so the + // cursor either precedes them, falls among them, or follows them. + let found: Option = match after { + Some((source, _)) if source > TRAJECTORY_SOURCE => return Ok(None), + Some((source, id)) if source == TRAJECTORY_SOURCE => conn + .prepare_cached(trajectory_sql(true))? + .query_row([id], |row| row.get(0)) + .optional()?, + _ => conn + .prepare_cached(trajectory_sql(false))? + .query_row([], |row| row.get(0)) + .optional()?, + }; + return Ok(found.map(|id| (TRAJECTORY_SOURCE.to_string(), id))); + }; + let mut statement = conn.prepare_cached(&seek_sql(table, session, after.is_some()))?; + let read = |row: &rusqlite::Row<'_>| Ok((row.get(0)?, row.get(1)?)); + Ok(match after { + Some((source, id)) => statement.query_row([source, id], read).optional()?, + None => statement.query_row([], read).optional()?, + }) +} + +/// Up to `limit` distinct identities after `after`, in order: the merge of +/// every table's ordered identities, each advanced only past what it offered. +/// +/// Every seek of a page reads one snapshot: the caller's transaction when it +/// holds one -- a consent baseline spanning many pages -- and otherwise one +/// deferred read transaction per page. Each seek is its own statement, and on +/// an autocommit connection each would see its own moment: an identity whose +/// only row moves from a table not yet sought to one already sought -- a +/// sidechain's events adopted into the catalog, say -- would be in the store +/// throughout and in no arm's answer. +pub(crate) fn identities_after( + conn: &Connection, + after: Option<(&str, &str)>, + limit: usize, +) -> Result> { + if conn.is_autocommit() { + let snapshot = conn.unchecked_transaction()?; + let page = merge_page(&snapshot, after, limit)?; + snapshot.commit()?; + return Ok(page); + } + merge_page(conn, after, limit) +} + +fn merge_page( + conn: &Connection, + after: Option<(&str, &str)>, + limit: usize, +) -> Result> { + let arms: Vec> = IDENTITY_TABLES + .iter() + .map(|(table, session)| Some((*table, *session))) + .chain([None]) + .collect(); + let mut offers: Vec> = arms + .iter() + .map(|arm| next_in(conn, *arm, after)) + .collect::>()?; + let mut page = Vec::with_capacity(limit.min(DEFAULT_IDENTITY_PAGE)); + while page.len() < limit { + let Some(next) = offers.iter().flatten().min().cloned() else { + break; + }; + for (arm, offer) in arms.iter().zip(offers.iter_mut()) { + if offer.as_ref() == Some(&next) { + *offer = next_in(conn, *arm, Some((next.0.as_str(), next.1.as_str())))?; + } + } + page.push(next); + } + Ok(page) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::session_store::StoreOptions; + use crate::store::open_db; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + fn open(dir: &std::path::Path) -> std::path::PathBuf { + let db = dir.join("ai-history.db"); + SessionStore::open(StoreOptions { + db_path: Some(db.clone()), + ..StoreOptions::default() + }) + .unwrap(); + db + } + + /// A page is one snapshot even on an autocommit connection. An identity + /// whose only row moves from a table the page has not sought yet to one + /// it already has -- committed by another connection between the two + /// seeks -- is still in the page, because every seek reads the store as + /// it stood when the page began. + #[test] + fn a_page_reads_one_snapshot_on_an_autocommit_connection() { + let dir = tempfile::tempdir().unwrap(); + let db = open(dir.path()); + open_db(&db) + .unwrap() + .execute( + "INSERT INTO session_events (source, session_id, message_id, ts_ms, role, kind, \ + text, event_uid) VALUES ('claude', 'moving', 'm', 1, 'user', 'text', 'x', 'e1')", + [], + ) + .unwrap(); + + let reader = open_db_readonly(&db).unwrap(); + assert!(reader.is_autocommit()); + // A page with no hook first, so every statement is prepared and the + // schema parsed: what remains to count is each seek's own work. + assert_eq!(identities_after(&reader, None, 10).unwrap().len(), 1); + // How many progress callbacks the catalog's arm -- the first seek a + // page runs -- takes on its own. The move fires just past them: after + // that seek has read the catalog, before the events arm reads. + let counted = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let counter = Arc::clone(&counted); + reader.progress_handler( + 1, + Some(move || { + counter.fetch_add(1, Ordering::SeqCst); + false + }), + ); + assert_eq!( + next_in(&reader, Some(("sessions", "session_id")), None).unwrap(), + None + ); + let first_arm = counted.load(Ordering::SeqCst); + let fired = Arc::new(AtomicBool::new(false)); + let flag = Arc::clone(&fired); + let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let seen = Arc::clone(&calls); + let db_for_hook = db.clone(); + // The move lands in one transaction, so at every moment the session + // is in exactly one of the two tables. + reader.progress_handler( + 1, + Some(move || { + if seen.fetch_add(1, Ordering::SeqCst) == first_arm && !flag.swap(true, Ordering::SeqCst) { + open_db(&db_for_hook) + .unwrap() + .execute_batch( + "BEGIN IMMEDIATE; \ + DELETE FROM session_events WHERE session_id = 'moving'; \ + INSERT INTO sessions (session_id, source) VALUES ('moving', 'claude'); \ + COMMIT;", + ) + .unwrap(); + } + false + }), + ); + let page = identities_after(&reader, None, 10).unwrap(); + reader.progress_handler(0, None:: bool>); + assert!( + fired.load(Ordering::SeqCst), + "the move must have interleaved" + ); + assert_eq!( + page, + vec![("claude".to_string(), "moving".to_string())], + "the session existed throughout, so the page names it" + ); + assert!(reader.is_autocommit(), "the page's own snapshot is closed"); + } + + /// A store from before the identity index gains it on its first writable + /// open, and the catalog's arm then seeks it; until then a read-only + /// store refuses the listing and names the remedy. + #[test] + fn an_older_store_gains_the_identity_index() { + let dir = tempfile::tempdir().unwrap(); + let db = open(dir.path()); + open_db(&db) + .unwrap() + .execute_batch("DROP INDEX idx_sessions_identity;") + .unwrap(); + let read_only = SessionStore::open(StoreOptions { + db_path: Some(db.clone()), + read_only: true, + ..StoreOptions::default() + }) + .unwrap(); + let error = read_only + .session_identities(IdentityQuery::default()) + .unwrap_err(); + assert!(error.is_stale_schema(), "{error}"); + + let store = SessionStore::open(StoreOptions { + db_path: Some(db.clone()), + ..StoreOptions::default() + }) + .unwrap(); + let conn = open_db(&db).unwrap(); + let present: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name = 'idx_sessions_identity')", + [], + |row| row.get(0), + ) + .unwrap(); + assert!(present, "the writable open migrated the index"); + let plan: String = conn + .query_row( + &format!( + "EXPLAIN QUERY PLAN {}", + seek_sql("sessions", "session_id", true) + ), + ["a", "b"], + |row| row.get(3), + ) + .unwrap(); + assert!( + plan.contains("COVERING INDEX idx_sessions_identity") + || plan.contains("COVERING INDEX delivery_identity_sessions"), + "{plan}" + ); + assert!(store.session_identities(IdentityQuery::default()).is_ok()); + assert!(read_only + .session_identities(IdentityQuery::default()) + .is_ok()); + } + + /// Every seek is an index search on `(source, session)` -- a covering + /// read that never touches a payload -- and none scans its table. + #[test] + fn every_seek_is_an_index_search_on_the_session() { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("ai-history.db"); + SessionStore::open(StoreOptions { + db_path: Some(db.clone()), + ..StoreOptions::default() + }) + .unwrap(); + let conn = crate::store::open_db(&db).unwrap(); + let mut plans = Vec::new(); + for (table, session) in IDENTITY_TABLES { + for after in [false, true] { + plans.push((*table, seek_sql(table, session, after), after, 2)); + } + } + for after in [false, true] { + plans.push(("trajectories", trajectory_sql(after).to_string(), after, 1)); + } + for (table, sql, after, arity) in plans { + let values: Vec<&str> = if after { + ["a", "b"][..arity].to_vec() + } else { + vec![] + }; + let plan = conn + .prepare(&format!("EXPLAIN QUERY PLAN {sql}")) + .unwrap() + .query_map(rusqlite::params_from_iter(values), |row| { + row.get::<_, String>(3) + }) + .unwrap() + .collect::>>() + .unwrap() + .join(" | "); + eprintln!("{table} (after: {after}): {plan}"); + assert!(plan.contains("COVERING INDEX"), "{table}: {plan}"); + assert!(!plan.contains("TEMP B-TREE"), "{table}: {plan}"); + if after { + assert!( + plan.starts_with(&format!("SEARCH {table}")), + "{table}: {plan}" + ); + } + } + } +} diff --git a/crates/ai-hist/src/session_store.rs b/crates/ai-hist/src/session_store.rs index 91c10ed6..a6395c58 100644 --- a/crates/ai-hist/src/session_store.rs +++ b/crates/ai-hist/src/session_store.rs @@ -268,7 +268,7 @@ impl Error { /// A read-only store refusing a database written before the `what` /// schema this version reads. - fn stale_schema(db_path: &Path, what: &str) -> Self { + pub(crate) fn stale_schema(db_path: &Path, what: &str) -> Self { Self::StaleSchema(format!( "{} predates the {what} schema this version reads; \ open it writable once (or run a sync) to migrate it", diff --git a/crates/ai-hist/src/storage.rs b/crates/ai-hist/src/storage.rs index 2ecd541f..96a9107c 100644 --- a/crates/ai-hist/src/storage.rs +++ b/crates/ai-hist/src/storage.rs @@ -219,38 +219,27 @@ pub fn latest_history_for_session( Ok(rows.next().transpose()?) } -/// Bounded, deduplicated identities across catalog, prompts and events. Includes -/// uncatalogued sessions so an export exclusion cannot miss partially read data. -/// Continue with the last returned identity; hold a read transaction when a -/// consistent multi-page baseline is required. +/// Bounded, deduplicated identities across every table that stores a +/// session, the same read as [`crate::SessionStore::session_identities`], so +/// an export exclusion cannot miss partially read data. Each page reads one +/// snapshot -- the caller's transaction, or its own on an autocommit +/// connection. Continue with the last returned identity; hold a read +/// transaction when a consistent multi-page baseline is required. #[cfg(feature = "export")] pub fn session_identities_after( conn: &Connection, after: Option<&crate::export::SessionIdentity>, limit: usize, ) -> Result> { - let mut query = conn.prepare( - "SELECT source, session_id FROM ( - SELECT source, session_id FROM sessions - UNION SELECT source, session_id FROM history WHERE session_id IS NOT NULL - UNION SELECT source, session_id FROM session_events - ) WHERE ?1 IS NULL OR (source, session_id) > (?1, ?2) - ORDER BY source, session_id LIMIT ?3", + let page = crate::session_identities::identities_after( + conn, + after.map(|after| (after.source.as_str(), after.session_id.as_str())), + bounded(limit) as usize, )?; - let rows = query.query_map( - rusqlite::params![ - after.map(|v| v.source.as_str()), - after.map(|v| v.session_id.as_str()), - bounded(limit) - ], - |row| { - Ok(crate::export::SessionIdentity { - source: row.get(0)?, - session_id: row.get(1)?, - }) - }, - )?; - Ok(rows.collect::>>()?) + Ok(page + .into_iter() + .map(|(source, session_id)| crate::export::SessionIdentity { source, session_id }) + .collect()) } #[cfg(all(test, feature = "export"))] diff --git a/crates/ai-hist/src/store.rs b/crates/ai-hist/src/store.rs index fbb38470..f4822d7e 100644 --- a/crates/ai-hist/src/store.rs +++ b/crates/ai-hist/src/store.rs @@ -711,6 +711,9 @@ const REQUIRED_INDEXES: &[&str] = &[ "idx_session_continuity_parent_uuid", "idx_session_continuity_pending", "idx_sessions_project_key", + // The identity listing merges every table in `(source, session_id)` + // order; the catalog's primary key leads with `session_id`. + "idx_sessions_identity", ]; /// Indexes no longer created: nothing queries them, or a replacement covers @@ -753,6 +756,9 @@ const REQUIRED_EVIDENCE_READ_INDEXES: &[&str] = &[ "idx_session_markers_page", ]; const REQUIRED_SCOPE_READ_INDEXES: &[&str] = &["idx_session_presences_location"]; +/// The catalog's arm of the identity listing. Every other table's arm rides +/// an index its own reads already require, or its primary key. +const REQUIRED_IDENTITY_READ_INDEXES: &[&str] = &["idx_sessions_identity"]; const REQUIRED_RELATIONSHIP_READ_INDEXES: &[&str] = &[ "idx_session_relationships_parent", "idx_session_relationships_child", @@ -830,6 +836,12 @@ pub fn schema_is_catalog_read_current(conn: &Connection) -> Result { schema_has_required_indexes(conn, REQUIRED_CATALOG_READ_INDEXES) } +/// Whether the identity listing can seek the catalog in `(source, +/// session_id)` order. +pub fn schema_is_identity_read_current(conn: &Connection) -> Result { + schema_has_required_indexes(conn, REQUIRED_IDENTITY_READ_INDEXES) +} + /// Whether bounded event pagination has both source-scoped and source-less indexes. pub fn schema_is_event_read_current(conn: &Connection) -> Result { schema_has_required_indexes(conn, REQUIRED_EVENT_READ_INDEXES) @@ -1394,6 +1406,13 @@ VALUES ('session_presences_local_backfill_v1'); "CREATE INDEX IF NOT EXISTS idx_sessions_source_recency ON sessions(source, last_activity_ms DESC, session_id)", [], )?; + // The primary key leads with `session_id`; the identity listing merges + // every table's identities in `(source, session_id)` order, and needs the + // catalog in that order too. + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_sessions_identity ON sessions(source, session_id)", + [], + )?; // Shallow discovery keys its "has this file changed?" lookup on the raw // path, because a transcript's session id is not known until it is read. conn.execute( diff --git a/crates/ai-hist/tests/change_feed.rs b/crates/ai-hist/tests/change_feed.rs index 826e9805..4f316cb1 100644 --- a/crates/ai-hist/tests/change_feed.rs +++ b/crates/ai-hist/tests/change_feed.rs @@ -769,3 +769,340 @@ fn a_row_from_an_unknown_source_is_carried() { assert_eq!(changes[0].op, ChangeOp::Delete); assert_eq!(changes[0].source_name, "some-new-agent"); } + +fn session_drain( + store: &SessionStore, + from: Watermark, + source: &str, + session: &str, +) -> Vec { + store + .changes_since(from, ChangeQuery::default().session(source, session)) + .unwrap() + .map(|change| change.unwrap()) + .collect() +} + +/// Every session the unfiltered drain names. +fn sessions_in(changes: &[Change]) -> BTreeSet<(String, String)> { + changes + .iter() + .filter(|change| !change.session_id.is_empty()) + .map(|change| (change.source_name.clone(), change.session_id.clone())) + .collect() +} + +fn assert_session_drains_match(store: &SessionStore, from: Watermark, step: &str) { + let everything = drain(store, from); + let sessions = sessions_in(&everything); + assert!(sessions.len() > 1, "{step}: several sessions to tell apart"); + for (source, session) in &sessions { + let expected: Vec<&Change> = everything + .iter() + .filter(|change| &change.source_name == source && &change.session_id == session) + .collect(); + let filtered = session_drain(store, from, source, session); + assert_eq!( + filtered.iter().collect::>(), + expected, + "{step}: {source} {session} is the whole feed restricted to that session" + ); + assert!(!filtered.is_empty()); + } +} + +/// A drain restricted to one session is the unfiltered drain restricted to +/// that session -- the same changes, rows, keys and revisions, tombstones +/// included -- and names no other session, after syncs, updates and deletes. +#[test] +fn a_session_drain_is_the_feed_restricted_to_that_session() { + let home = Home::new(); + for fixture in [ + "simple-turn.jsonl", + "multi-block-turn.jsonl", + "edit-revert.jsonl", + "compact-boundary.jsonl", + "resume-marker.jsonl", + ] { + home.stage_claude(fixture); + } + home.stage_codex("compaction.jsonl"); + home.stage_codex("with-tool-call.jsonl"); + fs::write( + home.path().join(".claude/history.jsonl"), + "{\"display\":\"a prompt with a session\",\"timestamp\":5,\ + \"sessionId\":\"11111111-1111-1111-1111-111111111111\"}\n\ + {\"display\":\"a prompt without one\",\"timestamp\":6}\n", + ) + .unwrap(); + let store = home.store(); + store.sync(Default::default()).unwrap(); + assert_session_drains_match(&store, Watermark::START, "after a sync"); + + // Updates, a deleted row, and a whole session deleted with its cascade. + let head = store.head_revision().unwrap(); + let writer = home.raw_writer(); + let doomed: String = writer + .query_row( + "SELECT session_id FROM sessions WHERE source = 'claude' \ + AND session_id <> '11111111-1111-1111-1111-111111111111' LIMIT 1", + [], + |row| row.get(0), + ) + .unwrap(); + writer + .execute_batch(&format!( + "UPDATE session_events SET text = 'edited' \ + WHERE rowid = (SELECT MIN(rowid) FROM session_events); \ + DELETE FROM tool_calls WHERE rowid = (SELECT MIN(rowid) FROM tool_calls); \ + UPDATE history SET project = '/elsewhere' WHERE timestamp_ms = 5; \ + DELETE FROM sessions WHERE source = 'claude' AND session_id = '{doomed}';" + )) + .unwrap(); + assert_session_drains_match(&store, Watermark::START, "after updates and deletes"); + assert_session_drains_match(&store, head, "from a watermark"); + let gone = session_drain(&store, head, "claude", &doomed); + assert!( + gone.iter() + .any(|change| change.kind == ChangeKind::Session && change.op == ChangeOp::Delete), + "the deleted session's own tombstones reach its drain: {gone:?}" + ); + + // The drain is bounded to the head at open, like the unfiltered one. + let changes = store + .changes_since( + Watermark::START, + ChangeQuery::default().session("claude", "11111111-1111-1111-1111-111111111111"), + ) + .unwrap(); + assert_eq!(changes.head(), store.head_revision().unwrap()); + + // A prompt that names no session is in the feed, and in no session's + // drain. + let prompts = only(&store, Watermark::START, ChangeKind::History); + assert!(prompts.iter().any(|change| change.session_id.is_empty())); + for (source, session) in sessions_in(&drain(&store, Watermark::START)) { + assert!(session_drain(&store, Watermark::START, &source, &session) + .iter() + .all(|change| !change.session_id.is_empty())); + } +} + +/// A session under a source this build does not know drains like any other, +/// and a trajectory is the session its id names. +#[test] +fn a_session_drain_takes_any_stored_source() { + let home = Home::new(); + let store = home.store(); + home.raw_writer() + .execute_batch( + "INSERT INTO sessions (session_id, source) VALUES ('n1', 'some-new-agent'); \ + INSERT INTO sessions (session_id, source) VALUES ('n2', 'some-new-agent'); \ + INSERT INTO history (source, session_id, prompt, timestamp_ms) \ + VALUES ('some-new-agent', 'n1', 'hi', 7); \ + INSERT INTO trajectories (id, decisions_json, retrospective_json, search_text, \ + updated_ms, timestamp_ms) VALUES ('traj-1', '[]', '{}', 'x', 1, 1);", + ) + .unwrap(); + let changes = session_drain(&store, Watermark::START, "some-new-agent", "n1"); + let kinds: Vec = changes.iter().map(|change| change.kind).collect(); + assert_eq!(kinds, vec![ChangeKind::Session, ChangeKind::History]); + assert!(changes.iter().all(|change| change.source.is_none())); + + let trajectory = session_drain(&store, Watermark::START, "trajectory", "traj-1"); + assert_eq!(trajectory.len(), 1); + assert_eq!(trajectory[0].kind, ChangeKind::Trajectory); + assert!(session_drain(&store, Watermark::START, "claude", "traj-1").is_empty()); +} + +/// A session drain is a one-shot read: it cannot name a consumer, so it +/// can never move one, and an empty identity is refused rather than read. +#[test] +fn a_session_drain_refuses_a_consumer() { + let home = Home::new(); + let store = home.store(); + for (from, query) in [ + ( + Watermark::CONSUMER, + ChangeQuery::default() + .consumer("probe") + .session("claude", "s1"), + ), + ( + Watermark::START, + ChangeQuery::default() + .consumer("probe") + .session("claude", "s1"), + ), + ( + Watermark::START, + ChangeQuery::default().session("claude", ""), + ), + (Watermark::START, ChangeQuery::default().session("", "s1")), + ] { + let error = store.changes_since(from, query).unwrap_err(); + assert_eq!(error.code(), "INVALID_ARGUMENT", "{error}"); + } + let changes = store + .changes_since( + Watermark::START, + ChangeQuery::default().session("claude", "s1"), + ) + .unwrap(); + assert_eq!(changes.commit().unwrap_err().code(), "INVALID_ARGUMENT"); +} + +/// One session's backfill against a store of many: run with `--ignored +/// --nocapture` in release to print the timing. +#[test] +#[ignore] +fn a_session_drain_reads_one_session_of_many() { + const SESSIONS: usize = 2_000; + const EVENTS: usize = 50; + let home = Home::new(); + let store = home.store(); + let mut writer = home.raw_writer(); + let tx = writer.transaction().unwrap(); + for session in 0..SESSIONS { + let id = format!("s{session:05}"); + tx.execute( + "INSERT INTO sessions (session_id, source) VALUES (?, 'claude')", + [&id], + ) + .unwrap(); + for event in 0..EVENTS { + tx.execute( + "INSERT INTO session_events (source, session_id, message_id, ts_ms, role, \ + kind, text, event_uid) VALUES ('claude', ?1, 'm', ?2, 'assistant', 'text', \ + 'some text', ?3)", + rusqlite::params![id, event as i64, format!("e{event}")], + ) + .unwrap(); + } + tx.execute( + "INSERT INTO tool_calls (source, session_id, tool_use_id, name) \ + VALUES ('claude', ?, 't1', 'Bash')", + [&id], + ) + .unwrap(); + } + // And one long session, whose backfill pages many times over. + const LONG: usize = 50_000; + tx.execute( + "INSERT INTO sessions (session_id, source) VALUES ('long', 'claude')", + [], + ) + .unwrap(); + for event in 0..LONG { + tx.execute( + "INSERT INTO session_events (source, session_id, message_id, ts_ms, role, kind, \ + text, event_uid) VALUES ('claude', 'long', 'm', ?1, 'assistant', 'text', \ + 'some text', ?2)", + rusqlite::params![event as i64, format!("e{event}")], + ) + .unwrap(); + } + tx.commit().unwrap(); + let rows = SESSIONS * (EVENTS + 2) + LONG + 1; + + let started = std::time::Instant::now(); + let everything = drain(&store, Watermark::START).len(); + let full = started.elapsed(); + let started = std::time::Instant::now(); + let one = session_drain(&store, Watermark::START, "claude", "s01000").len(); + let filtered = started.elapsed(); + let started = std::time::Instant::now(); + let unfiltered_one = drain(&store, Watermark::START) + .into_iter() + .filter(|change| change.session_id == "s01000") + .count(); + let replay = started.elapsed(); + assert_eq!(everything, rows); + assert_eq!(one, EVENTS + 2); + assert_eq!(unfiltered_one, one); + eprintln!( + "{SESSIONS} sessions, {rows} rows: one-session drain {one} changes in {filtered:?}; \ + full drain {everything} changes in {full:?}; full replay filtered to the session \ + in {replay:?}" + ); + assert!(filtered * 20 < full, "{filtered:?} vs {full:?}"); + + let started = std::time::Instant::now(); + let long = session_drain(&store, Watermark::START, "claude", "long").len(); + let long_elapsed = started.elapsed(); + assert_eq!(long, LONG + 1); + eprintln!("one session of {long} changes, default batch: {long_elapsed:?}"); + let started = std::time::Instant::now(); + let long = store + .changes_since( + Watermark::START, + ChangeQuery::default() + .session("claude", "long") + .batch(ai_hist::MAX_CHANGE_BATCH), + ) + .unwrap() + .count(); + eprintln!( + "one session of {long} changes, batch {}: {:?}", + ai_hist::MAX_CHANGE_BATCH, + started.elapsed() + ); + + // Every identity, paged at the default size. + let started = std::time::Instant::now(); + let mut identities: Vec = Vec::new(); + loop { + let mut query = ai_hist::IdentityQuery::default(); + if let Some(last) = identities.last() { + query = query.after(last.clone()); + } + let page = store.session_identities(query).unwrap(); + if page.is_empty() { + break; + } + identities.extend(page); + } + assert_eq!(identities.len(), SESSIONS + 1); + eprintln!( + "{} identities over {rows} rows, paged at 1,000: {:?}", + identities.len(), + started.elapsed() + ); + + // The same pages as a three-table UNION over the whole tables, for scale. + let started = std::time::Instant::now(); + let reader = home.raw(); + let mut union: Vec<(String, String)> = Vec::new(); + loop { + let last = union.last().cloned(); + let page: Vec<(String, String)> = reader + .prepare( + "SELECT source, session_id FROM ( + SELECT source, session_id FROM sessions + UNION SELECT source, session_id FROM history WHERE session_id IS NOT NULL + UNION SELECT source, session_id FROM session_events + ) WHERE ?1 IS NULL OR (source, session_id) > (?1, ?2) + ORDER BY source, session_id LIMIT 1000", + ) + .unwrap() + .query_map( + rusqlite::params![ + last.as_ref().map(|(source, _)| source.clone()), + last.as_ref().map(|(_, session)| session.clone()) + ], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap() + .collect::>() + .unwrap(); + if page.is_empty() { + break; + } + union.extend(page); + } + assert_eq!(union.len(), identities.len()); + eprintln!( + "the same identities as a three-table UNION: {:?}", + started.elapsed() + ); +} diff --git a/crates/ai-hist/tests/session_identities.rs b/crates/ai-hist/tests/session_identities.rs new file mode 100644 index 00000000..396ef9f9 --- /dev/null +++ b/crates/ai-hist/tests/session_identities.rs @@ -0,0 +1,227 @@ +//! `SessionStore::session_identities` through the public surface: every +//! session the store holds evidence for, catalogued or not, paged without +//! gaps or repeats. +//! +//! Public API only, so this runs in the `--no-default-features` job too. The +//! rows are written with a raw connection, standing in for the writers -- +//! prompt logs, sidechains, connectors -- that store evidence the catalog +//! never names. + +use ai_hist::{ + Change, ChangeQuery, IdentityQuery, SessionIdentity, SessionStore, Source, StoreOptions, + Watermark, +}; +use rusqlite::Connection; +use std::collections::BTreeSet; + +struct Store { + _dir: tempfile::TempDir, + db: std::path::PathBuf, + store: SessionStore, +} + +fn store() -> Store { + let dir = tempfile::tempdir().unwrap(); + let db = dir.path().join("ai-history.db"); + let mut options = StoreOptions::default(); + options.db_path = Some(db.clone()); + options.home = Some(dir.path().to_path_buf()); + let store = SessionStore::open(options).unwrap(); + Store { + _dir: dir, + db, + store, + } +} + +impl Store { + fn write(&self, sql: &str) { + Connection::open(&self.db) + .unwrap() + .execute_batch(sql) + .unwrap(); + } + + fn all(&self, limit: usize) -> Vec { + let mut identities: Vec = Vec::new(); + loop { + let mut query = IdentityQuery::default().limit(limit); + if let Some(last) = identities.last() { + query = query.after(last.clone()); + } + let page = self.store.session_identities(query).unwrap(); + if page.is_empty() { + break; + } + // Zero asks for the default page. + let bound = if limit == 0 { 1_000 } else { limit }; + assert!(page.len() <= bound, "a page is bounded by its limit"); + identities.extend(page); + } + identities + } +} + +fn pair(identity: &SessionIdentity) -> (&str, &str) { + (identity.source_name.as_str(), identity.session_id.as_str()) +} + +/// Every table that can hold a session the catalog does not name. +const EVIDENCE: &str = r#" +INSERT INTO sessions (session_id, source) VALUES ('catalogued', 'claude'); +INSERT INTO session_events (source, session_id, message_id, ts_ms, role, kind, text, event_uid) + VALUES ('claude', 'sidechain', 'm', 1, 'assistant', 'text', 'x', 'e1'), + ('claude', 'sidechain', 'm', 2, 'assistant', 'text', 'y', 'e2'), + ('claude', 'catalogued', 'm', 1, 'user', 'text', 'z', 'e1'); +INSERT INTO history (source, session_id, prompt, timestamp_ms) + VALUES ('codex', 'history-only', 'a prompt', 1), + ('codex', NULL, 'a prompt with no session', 2), + ('claude', '', 'a prompt with an empty session', 3); +INSERT INTO session_events (source, session_id, message_id, ts_ms, role, kind, text, event_uid) + VALUES ('grok', '', 'm', 1, 'user', 'text', 'no session', 'e1'); +INSERT INTO trajectories (id, decisions_json, retrospective_json, search_text, updated_ms, + timestamp_ms) VALUES ('', '[]', '{}', 'x', 1, 1); +INSERT INTO tool_calls (source, session_id, tool_use_id, name) + VALUES ('claude', 'tool-only', 't1', 'Bash'); +INSERT INTO file_edits (source, session_id, tool_use_id, file_path, tool_name) + VALUES ('claude', 'edit-only', 't1', '/a.rs', 'Edit'); +INSERT INTO session_markers (source, session_id, marker_uid, kind) + VALUES ('grok', 'marker-only', 'mk1', 'compaction'); +INSERT INTO session_relationships (source, parent_session_id, relationship_uid, + child_session_id, relationship, identity_status, evidence_kind, created_ms, updated_ms) + VALUES ('claude', 'parent-only', 'r1', 'child-named-by-an-edge', 'delegated', 'observed', + 'sidecar', 1, 1); +INSERT INTO session_presences (source, session_id, location) + VALUES ('opencode', 'presence-only', 'remote'); +INSERT INTO session_commit_links (source, session_id, repo, commit_sha, match_method, + confidence, created_at_ms) + VALUES ('codex', 'commit-only', 'repo', 'abc', 'trailer', 1.0, 1); +INSERT INTO session_observations (source, session_id, location, connector_id, + connector_instance, updated_ms) + VALUES ('some-new-agent', 'observed-only', 'remote', 'conn', 'default', 1); +INSERT INTO observation_evidence (source, session_id, location, connector_id, + connector_instance, evidence_uid, payload_json) + VALUES ('some-new-agent', 'observed-only', 'remote', 'conn', 'default', 'ev1', '{}'); +INSERT INTO trajectories (id, decisions_json, retrospective_json, search_text, updated_ms, + timestamp_ms) VALUES ('traj-1', '[]', '{}', 'x', 1, 1); +"#; + +/// A session counts when any table stores a row under it: an events-only +/// sidechain and a history-only prompt log appear beside the catalog, each +/// once, in order, while a prompt with no session, rows under an empty +/// session id, and a child only an edge names do not. Every identity listed +/// is one `ChangeQuery::session` accepts. +#[test] +fn every_stored_session_is_an_identity_once() { + let store = store(); + store.write(EVIDENCE); + let identities = store.all(1_000); + let pairs: Vec<(&str, &str)> = identities.iter().map(pair).collect(); + assert_eq!( + pairs, + vec![ + ("claude", "catalogued"), + ("claude", "edit-only"), + ("claude", "parent-only"), + ("claude", "sidechain"), + ("claude", "tool-only"), + ("codex", "commit-only"), + ("codex", "history-only"), + ("grok", "marker-only"), + ("opencode", "presence-only"), + ("some-new-agent", "observed-only"), + ("trajectory", "traj-1"), + ] + ); + for identity in &identities { + assert!(store + .store + .changes_since( + Watermark::START, + ChangeQuery::default().session(&identity.source_name, &identity.session_id), + ) + .is_ok()); + } + assert_eq!(identities[0].source(), Some(Source::Claude)); + assert_eq!(identities[9].source(), None, "an unknown source is carried"); + assert_eq!(identities[10].source(), Some(Source::Trajectory)); +} + +/// Every page size walks the same identities, completely and without a +/// repeat, and a cursor that names no identity resumes after where it sorts. +#[test] +fn paging_is_complete_and_never_repeats() { + let store = store(); + store.write(EVIDENCE); + // Many sessions, each in several tables, so page edges fall inside runs + // of duplicates across tables. + let mut bulk = String::new(); + for index in 0..150 { + bulk.push_str(&format!( + "INSERT INTO sessions (session_id, source) VALUES ('s{index:03}', 'claude'); + INSERT INTO session_events (source, session_id, message_id, ts_ms, role, kind, \ + text, event_uid) VALUES ('claude', 's{index:03}', 'm', 1, 'user', 'text', \ + 'x', 'e1'), ('claude', 's{index:03}', 'm', 2, 'user', 'text', 'y', 'e2'); + INSERT INTO history (source, session_id, prompt, timestamp_ms) \ + VALUES ('claude', 's{index:03}', 'p{index}', {index});\n" + )); + } + store.write(&bulk); + let whole = store.all(10_000); + assert_eq!(whole.len(), 11 + 150); + let distinct: BTreeSet<&SessionIdentity> = whole.iter().collect(); + assert_eq!(distinct.len(), whole.len(), "no identity repeats"); + let mut sorted = whole.clone(); + sorted.sort(); + assert_eq!(sorted, whole, "in (source_name, session_id) order"); + for limit in [1, 2, 3, 7, 64, 0] { + assert_eq!(store.all(limit), whole, "limit {limit}"); + } + + let resumed = store + .store + .session_identities(IdentityQuery::default().after(SessionIdentity::new("claude", "s149~"))) + .unwrap(); + assert_eq!(pair(&resumed[0]), ("claude", "sidechain")); + let past_the_end = store + .store + .session_identities(IdentityQuery::default().after(SessionIdentity::new("zzz", ""))) + .unwrap(); + assert!(past_the_end.is_empty()); +} + +/// The identities are the sessions the change feed names, and each one's +/// session drain is its own rows. +#[test] +fn identities_are_the_sessions_the_feed_names() { + let store = store(); + store.write(EVIDENCE); + let feed: Vec = store + .store + .changes_since(Watermark::START, ChangeQuery::default()) + .unwrap() + .map(|change| change.unwrap()) + .collect(); + let named: BTreeSet<(String, String)> = feed + .iter() + .filter(|change| !change.session_id.is_empty()) + .map(|change| (change.source_name.clone(), change.session_id.clone())) + .collect(); + let identities: BTreeSet<(String, String)> = store + .all(1_000) + .into_iter() + .map(|identity| (identity.source_name, identity.session_id)) + .collect(); + assert_eq!(identities, named); + for (source, session) in &identities { + let drained = store + .store + .changes_since( + Watermark::START, + ChangeQuery::default().session(source, session), + ) + .unwrap() + .count(); + assert!(drained > 0, "{source} {session}"); + } +} diff --git a/docs/sourcing-sdk.md b/docs/sourcing-sdk.md index 3a58c9db..0a8bebda 100644 --- a/docs/sourcing-sdk.md +++ b/docs/sourcing-sdk.md @@ -85,7 +85,7 @@ What the facade owes you, and what it asks in return: `Error::UnsupportedOperation`, and a watermark the store cannot serve is `Error::WatermarkAheadOfStore`. -## The ten operations +## The eleven operations | Method | Provider I/O | Database work | Lock | | ------------------------------------------------ | ------------------------------------------- | ------------------------------------------------------------------------- | ------------------------------------------------ | @@ -96,7 +96,8 @@ What the facade owes you, and what it asks in return: | `watch(WatchOptions) -> WatchHandle` | fs-event or polling driven sweeps | the same as `sync`, per tick | `SyncRunLock`, per tick | | `sessions(CatalogQuery) -> CatalogIter` | none | keyset-paged reads over `sessions` | none (WAL reader) | | `session(&SessionRef, SessionQuery)` | none | every table for one session, on one snapshot | none (one deferred read transaction) | -| `changes_since(Watermark, ChangeQuery)` | none | one indexed revision-range read per kind per page, plus tombstones | none (one read snapshot per page) | +| `session_identities(IdentityQuery)` | none | one covering index seek per identity per table that holds it | none (one read snapshot per page) | +| `changes_since(Watermark, ChangeQuery)` | none | one indexed revision-range read per kind per page, plus tombstones; a session drain seeks that session's index instead | none (one read snapshot per page) | | `head_revision() -> Watermark` | none | one read of the feed head | none | | `Source::capabilities() -> SourceCapabilities` | none | none — static | none | @@ -271,6 +272,46 @@ catalog row — `source: Source`, `project_key`, `discovery_state`, the provider-observed metadata — and `session_ref()` turns it into the reference `session` takes. +### `session_identities` + +Every session the store holds evidence for, catalogued or not: +`session_identities(IdentityQuery { after, limit })` returns up to `limit` +distinct `SessionIdentity { source_name, session_id }` after `after`, in +`(source_name, session_id)` byte order; continue with the last one returned, +and an empty page is the end. `limit` is clamped to `1..=10_000`, and zero +means 1,000. A session counts when any evidence table stores a row under it: +`sessions`, `history`, `session_events`, `tool_calls`, `file_edits`, +`session_markers`, `session_relationships` (under the parent), +`session_presences`, `session_commit_links`, `session_observations`, +`observation_evidence`, and `trajectories` (under the `trajectory` source, by +id). The catalog alone misses evidence that arrives without a catalog row — a +subagent sidechain's events, a prompt-log entry, a connector's observation — so +this is the read for anything that decides which sessions exist, such as a +consent baseline. A prompt that names no session is under none, an empty +session id names no session, and a child session that only a relationship +names is not an identity until something is stored under it; every identity +listed is one `ChangeQuery::session` accepts. These are exactly the non-empty `(source_name, session_id)` +pairs the change feed reports. `source_name` is the stored text; +`SessionIdentity::source()` parses it, and is `None` for a source this build +does not know. + +No payload is read. Each table offers its next identity after the cursor +through a covering index that leads with `(source, session)`, one seek +however many rows the session holds, and the smallest offer is the next +identity: + +```text +SEARCH session_events USING COVERING INDEX idx_session_events_session ((source,session_id)>(?,?)) +SEARCH sessions USING COVERING INDEX idx_sessions_identity ((source,session_id)>(?,?)) +SEARCH trajectories USING COVERING INDEX sqlite_autoindex_trajectories_1 (id>?) +``` + +Every seek of a page reads one snapshot, so a page is the store at one +moment; an identity written between pages is seen only if it sorts after the +cursor. The catalog's arm needs `idx_sessions_identity`, which a writable open +adds; a read-only store over a database without it answers +`session_identities` with `StaleSchema`, and keeps every other read. + ### `session` Everything the store holds about one session, on one SQLite snapshot: @@ -392,6 +433,30 @@ this build does not know — a row written by a newer release — and than failing the drain. The feed applies no consent or exclusion rule: an embedder that uploads applies its own selection. +`ChangeQuery::session(source, session_id)` restricts a drain to one session: +exactly the changes whose `source_name` and `session_id` are those, with the +same rows, keys, revisions and tombstones the unfiltered drain reports for it, +bounded to the head at open. `source` is the stored name, so a source this +build does not know works. It covers every kind that stores a session, +relationships under their parent, and a trajectory under the `trajectory` +source by its id. A prompt that names no session is in no session's drain, and +nor is a prompt's delete, whose tombstone carries no session because a +prompt's session is not part of its key. A session drain is a one-shot read, +such as the backfill of a session an embedder has just started following. It +cannot name a consumer, and it has nothing to commit; either is +`Error::InvalidArgument`, as is an empty source or session id. Keep the named +cursor for the whole feed and drain an added session from `Watermark::START`. +Each page reads through the table's session index, never the revision index, +and sorts only that session's rows: + +```text +SEARCH session_events USING INDEX idx_session_events_source_page (source=? AND session_id=?) +USE TEMP B-TREE FOR ORDER BY +``` + +Each page seeks the session again, so a long session drains fastest with a +large `batch`. + `from` is `Watermark::START` to replay everything, an explicit watermark to resume from one a consumer stored itself, or `Watermark::CONSUMER` with `ChangeQuery::consumer` to resume from a named cursor kept in `consumer_cursors` @@ -646,7 +711,7 @@ embedder reads before bumping. | Feature | Default | What it adds | For | | --- | --- | --- | --- | -| *(none)* | ✓ | `SessionStore` and its ten operations, the change feed (`Change`, `ChangeQuery`, `Watermark`, `EvidenceRow`, `StoredRow`), `Source` and `SourceCapabilities`, `Error`, the evidence structs above, `NormalizedUsage` and the usage normalizers, `project_identity`, `declared_evidence_kinds` | Embedders | +| *(none)* | ✓ | `SessionStore` and its eleven operations, the change feed (`Change`, `ChangeQuery`, `Watermark`, `EvidenceRow`, `StoredRow`), `SessionIdentity` and `IdentityQuery`, `Source` and `SourceCapabilities`, `Error`, the evidence structs above, `NormalizedUsage` and the usage normalizers, `project_identity`, `declared_evidence_kinds` | Embedders | | `fs-events` | — | The `notify` backend behind `watch`; without it `watch` polls at `poll_interval_ms`. `WatchOptions::use_fs_events` selects it when it is compiled in | The CLI, and an embedder that wants event-driven ticks | | `delivery` | — | Durable delivery of captured evidence to a destination | The CLI, napi, the relayhistory plugin | | `opencode-backup` | — | Snapshot a live OpenCode SQLite store through `rusqlite`'s backup API before reading it | The CLI, napi |