diff --git a/CHANGELOG.md b/CHANGELOG.md index 9a1dfd71..36d8a5b9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -184,6 +184,13 @@ Notable changes to the native `ai-hist` CLI are documented here. ### Rust API +- `SessionStore::has_session(&SessionIdentity)` answers whether the store + holds anything under one identity, by the same tables and rule as + `session_identities`: true exactly when the listing would name it, false for + an empty source or session id. It is one indexed existence probe per table, + on one snapshot. `storage::session_identity_exists` is the same check, and the + listing skips an empty source as it skips an empty session id, so every + session an embedder counts is one it can select and drain. - `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 @@ -197,8 +204,8 @@ Notable changes to the native `ai-hist` CLI are documented here. 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 + payload is read; each page reads one snapshot, and an empty source or + 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, diff --git a/crates/ai-hist/README.md b/crates/ai-hist/README.md index cf23f247..2d15597e 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`, `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. +Default features expose `SessionStore` — `open`, `discover`, `sync`, `hydrate`, `watch`, `sessions`, `session`, `session_identities`, `has_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. 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 b2201f9d..dce7d959 100644 --- a/crates/ai-hist/public-api.txt +++ b/crates/ai-hist/public-api.txt @@ -650,6 +650,7 @@ pub fn ai_hist::SessionStore::sessions(&self, ai_hist::CatalogQuery) -> ai_hist: 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::has_session(&self, &ai_hist::SessionIdentity) -> core::result::Result 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 diff --git a/crates/ai-hist/src/session_identities.rs b/crates/ai-hist/src/session_identities.rs index 04369c3c..9992ef45 100644 --- a/crates/ai-hist/src/session_identities.rs +++ b/crates/ai-hist/src/session_identities.rs @@ -90,8 +90,8 @@ impl SessionStore { /// /// 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: + /// pages are seen only if they sort after the cursor. An empty source or + /// session id 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( @@ -127,6 +127,18 @@ impl SessionStore { }) .collect()) } + + /// Whether the store holds anything under `identity`: exactly when + /// [`SessionStore::session_identities`] would list it, by the same tables + /// and rule -- so every identity an embedder counts is one it can select, + /// and one [`crate::ChangeQuery::session`] accepts. An empty source or + /// session id is never a session. Each table is one indexed existence + /// probe, all read on one snapshot; no payload is read. + pub fn has_session(&self, identity: &SessionIdentity) -> std::result::Result { + let conn = open_db_readonly(self.db_path()) + .map_err(|error| Error::DatabaseOpen(format!("{error:#}")))?; + identity_exists(&conn, &identity.source_name, &identity.session_id).map_err(Error::query) + } } /// Every table a session identity can be stored in: its name and the column @@ -149,8 +161,8 @@ const IDENTITY_TABLES: &[(&str, &str)] = &[ 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. +/// source or session id names no session -- `ChangeQuery::session` refuses +/// either -- 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)") @@ -159,7 +171,7 @@ fn seek_sql(table: &str, session: &str, after: bool) -> String { }; format!( "SELECT source, {session} FROM {table} \ - WHERE {session} IS NOT NULL AND {session} <> ''{range} \ + WHERE {session} IS NOT NULL AND {session} <> '' AND source <> ''{range} \ ORDER BY source, {session} LIMIT 1" ) } @@ -204,6 +216,48 @@ fn next_in( }) } +/// Whether the store holds anything under one identity: the same tables and +/// the same rule for what names a session as [`identities_after`], each an +/// indexed existence probe. An empty source or session id is never one. +/// +/// The probes read one snapshot, by the same rule as a page: the caller's +/// transaction, or their own on an autocommit connection, so a session whose +/// only row moves between tables mid-check is still found. +pub(crate) fn identity_exists(conn: &Connection, source: &str, session_id: &str) -> Result { + if source.is_empty() || session_id.is_empty() { + return Ok(false); + } + if conn.is_autocommit() { + let snapshot = conn.unchecked_transaction()?; + let found = probe_identity(&snapshot, source, session_id)?; + snapshot.commit()?; + return Ok(found); + } + probe_identity(conn, source, session_id) +} + +fn probe_identity(conn: &Connection, source: &str, session_id: &str) -> Result { + if source == TRAJECTORY_SOURCE { + let found: bool = conn + .prepare_cached("SELECT EXISTS(SELECT 1 FROM trajectories WHERE id = ?1)")? + .query_row([session_id], |row| row.get(0))?; + if found { + return Ok(true); + } + } + for (table, session) in IDENTITY_TABLES { + let found: bool = conn + .prepare_cached(&format!( + "SELECT EXISTS(SELECT 1 FROM {table} WHERE source = ?1 AND {session} = ?2)" + ))? + .query_row([source, session_id], |row| row.get(0))?; + if found { + return Ok(true); + } + } + Ok(false) +} + /// Up to `limit` distinct identities after `after`, in order: the merge of /// every table's ordered identities, each advanced only past what it offered. /// @@ -353,6 +407,137 @@ mod tests { assert!(reader.is_autocommit(), "the page's own snapshot is closed"); } + /// An identity exists exactly when the listing names it, in every table, + /// and never for an empty source or session id; each probe is an index + /// search. + #[test] + fn an_identity_exists_exactly_when_it_is_listed() { + let dir = tempfile::tempdir().unwrap(); + let db = open(dir.path()); + let conn = open_db(&db).unwrap(); + conn.execute_batch( + "INSERT INTO tool_calls (source, session_id, tool_use_id, name) \ + VALUES ('claude', 'tool-only', 't1', 'Bash'), ('', 'blank-source', 't1', 'Bash'); \ + 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 history (source, session_id, prompt, timestamp_ms) \ + VALUES ('codex', '', 'no session', 1); \ + INSERT INTO trajectories (id, decisions_json, retrospective_json, search_text, \ + updated_ms, timestamp_ms) VALUES ('traj-1', '[]', '{}', 'x', 1, 1);", + ) + .unwrap(); + let listed = identities_after(&conn, None, 100).unwrap(); + assert_eq!( + listed, + [ + ("claude", "tool-only"), + ("some-new-agent", "observed-only"), + ("trajectory", "traj-1") + ] + .map(|(source, id)| (source.to_string(), id.to_string())) + ); + for (source, id) in &listed { + assert!(identity_exists(&conn, source, id).unwrap(), "{source} {id}"); + } + for (source, id) in [ + ("", "blank-source"), + ("codex", ""), + ("claude", "missing"), + ("trajectory", "missing"), + ] { + assert!( + !identity_exists(&conn, source, id).unwrap(), + "{source:?} {id:?}" + ); + } + for (table, session) in IDENTITY_TABLES { + let plan: String = conn + .query_row( + &format!( + "EXPLAIN QUERY PLAN SELECT 1 FROM {table} \ + WHERE source = ?1 AND {session} = ?2" + ), + ["a", "b"], + |row| row.get(3), + ) + .unwrap(); + assert!( + plan.starts_with(&format!("SEARCH {table}")), + "{table}: {plan}" + ); + } + } + + /// The existence check reads one snapshot like a page: a session whose + /// only row moves from a table not yet probed to one already probed, + /// committed between the two probes, is still found. + #[test] + fn an_existence_check_reads_one_snapshot() { + 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(); + // Prepare every probe first, so the count below is the catalog + // probe's own work. + assert!(identity_exists(&reader, "claude", "moving").unwrap()); + 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 + }), + ); + reader + .prepare_cached( + "SELECT EXISTS(SELECT 1 FROM sessions WHERE source = ?1 AND session_id = ?2)", + ) + .unwrap() + .query_row(["claude", "moving"], |row| row.get::<_, bool>(0)) + .unwrap(); + let catalog_probe = 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(); + reader.progress_handler( + 1, + Some(move || { + if seen.fetch_add(1, Ordering::SeqCst) == catalog_probe + && !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 found = identity_exists(&reader, "claude", "moving").unwrap(); + reader.progress_handler(0, None:: bool>); + assert!( + fired.load(Ordering::SeqCst), + "the move must have interleaved" + ); + assert!(found, "the session existed throughout"); + } + /// 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. diff --git a/crates/ai-hist/src/storage.rs b/crates/ai-hist/src/storage.rs index 96a9107c..449770fd 100644 --- a/crates/ai-hist/src/storage.rs +++ b/crates/ai-hist/src/storage.rs @@ -242,6 +242,17 @@ pub fn session_identities_after( .collect()) } +/// Whether the store holds anything under one identity, by the same tables +/// and rule as [`session_identities_after`]: a session that listing names is +/// one this accepts. +#[cfg(feature = "export")] +pub fn session_identity_exists( + conn: &Connection, + identity: &crate::export::SessionIdentity, +) -> Result { + crate::session_identities::identity_exists(conn, &identity.source, &identity.session_id) +} + #[cfg(all(test, feature = "export"))] mod identity_tests { use super::*; diff --git a/crates/ai-hist/tests/session_identities.rs b/crates/ai-hist/tests/session_identities.rs index 396ef9f9..feece2ae 100644 --- a/crates/ai-hist/tests/session_identities.rs +++ b/crates/ai-hist/tests/session_identities.rs @@ -225,3 +225,37 @@ fn identities_are_the_sessions_the_feed_names() { assert!(drained > 0, "{source} {session}"); } } + +/// `has_session` answers exactly as the listing does: every listed identity +/// exists, and an empty source or session id, or one nothing is stored +/// under, does not. +#[test] +fn has_session_is_true_exactly_for_listed_identities() { + let store = store(); + store.write(EVIDENCE); + store.write( + "INSERT INTO tool_calls (source, session_id, tool_use_id, name) \ + VALUES ('', 'blank-source', 't1', 'Bash');", + ); + let identities = store.all(1_000); + assert_eq!(identities.len(), 11); + for identity in &identities { + assert!(store.store.has_session(identity).unwrap(), "{identity:?}"); + } + for (source, session) in [ + ("", "blank-source"), + ("claude", ""), + ("grok", ""), + ("codex", "missing"), + ("trajectory", "missing"), + ("claude", "child-named-by-an-edge"), + ] { + assert!( + !store + .store + .has_session(&SessionIdentity::new(source, session)) + .unwrap(), + "{source:?} {session:?}" + ); + } +} diff --git a/docs/sourcing-sdk.md b/docs/sourcing-sdk.md index 0a8bebda..35f80b9f 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 eleven operations +## The twelve operations | Method | Provider I/O | Database work | Lock | | ------------------------------------------------ | ------------------------------------------- | ------------------------------------------------------------------------- | ------------------------------------------------ | @@ -97,6 +97,7 @@ What the facade owes you, and what it asks in return: | `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) | | `session_identities(IdentityQuery)` | none | one covering index seek per identity per table that holds it | none (one read snapshot per page) | +| `has_session(&SessionIdentity)` | none | one indexed existence probe per table | none (one read snapshot) | | `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 | @@ -288,7 +289,7 @@ 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 +source or 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; @@ -306,6 +307,12 @@ SEARCH sessions USING COVERING INDEX idx_sessions_identity ((source,session_id)> SEARCH trajectories USING COVERING INDEX sqlite_autoindex_trajectories_1 (id>?) ``` +`has_session(&SessionIdentity)` answers whether one identity exists, by the +same tables and rule: true exactly when the listing would name it, so every +session an embedder counts is one it can select, and false for an empty source +or session id. It is one indexed existence probe per table, all on one +snapshot. + 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 @@ -711,7 +718,7 @@ embedder reads before bumping. | Feature | Default | What it adds | For | | --- | --- | --- | --- | -| *(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 | +| *(none)* | ✓ | `SessionStore` and its twelve 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 | diff --git a/plugins/relayhistory/rust/src/probe/bridge.rs b/plugins/relayhistory/rust/src/probe/bridge.rs index 74b7a7a2..f2378801 100644 --- a/plugins/relayhistory/rust/src/probe/bridge.rs +++ b/plugins/relayhistory/rust/src/probe/bridge.rs @@ -551,8 +551,9 @@ fn prepare_restart(directory: &Path) -> Result { fn validate_identities(conn: &Connection, identities_requested: &[SessionIdentity]) -> Result<()> { let mut known_requested = true; for id in identities_requested { - let exists: bool=conn.query_row("SELECT EXISTS(SELECT 1 FROM sessions WHERE source=?1 AND session_id=?2) OR EXISTS(SELECT 1 FROM history WHERE source=?1 AND session_id=?2) OR EXISTS(SELECT 1 FROM session_events WHERE source=?1 AND session_id=?2)",rusqlite::params![id.source,id.session_id],|r|r.get(0))?; - known_requested &= exists; + // The identity listing's own rule, so every session status counts is + // one a selection can name. + known_requested &= ai_hist::storage::session_identity_exists(conn, id)?; } ensure!( known_requested, @@ -1133,6 +1134,44 @@ mod tests { ); } + /// A session whose only evidence is a tool call or a connector + /// observation is one the identity listing counts, so a selection can + /// name it. + #[test] + fn a_session_with_only_evidence_rows_can_be_selected() { + let (dir, config) = fixture(SharingMode::Selected); + db(dir.path()) + .unwrap() + .execute_batch( + "INSERT INTO tool_calls (source, session_id, tool_use_id, name) \ + VALUES ('claude', 'tool-only', 't1', 'Bash'); \ + INSERT INTO session_observations (source, session_id, location, \ + connector_id, connector_instance, updated_ms) \ + VALUES ('codex', 'observed-only', 'remote', 'conn', 'default', 1);", + ) + .unwrap(); + let listed = identities(&db(dir.path()).unwrap()).unwrap(); + for key in ["claude:tool-only", "codex:observed-only"] { + let identity = parse_key(key).unwrap(); + assert!(listed.contains(&identity), "{key} is counted"); + } + let config = apply( + dir.path(), + config, + None, + &["claude:tool-only", "codex:observed-only"], + true, + ); + assert_eq!(mode(&config), SharingMode::Selected); + let selected = selected(dir.path()).unwrap(); + for key in ["claude:tool-only", "codex:observed-only"] { + assert!( + selected.contains(&parse_key(key).unwrap()), + "{key} is selected" + ); + } + } + #[test] fn bridge_validation_errors_are_safe_and_actionable() { for key in ["private-input", ":id", "codex:"] {