feat(sdk): SessionStore::has_session, and selection matches the identity listing

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,

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.
 

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<ai_hist::SyncReport, ai_hist::Error>
 pub fn ai_hist::SessionStore::watch(&self, ai_hist::WatchOptions) -> core::result::Result<ai_hist::WatchHandle, ai_hist::Error>
 impl ai_hist::SessionStore
+pub fn ai_hist::SessionStore::has_session(&self, &ai_hist::SessionIdentity) -> core::result::Result<bool, ai_hist::Error>
 pub fn ai_hist::SessionStore::session_identities(&self, ai_hist::IdentityQuery) -> core::result::Result<alloc::vec::Vec<ai_hist::SessionIdentity>, ai_hist::Error>
 pub struct ai_hist::SessionToolCall
 pub ai_hist::SessionToolCall::args_json: core::option::Option<alloc::string::String>

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<bool, Error> {
+        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<bool> {
+    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<bool> {
+    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::<fn() -> 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.

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<bool> {
+    crate::session_identities::identity_exists(conn, &identity.source, &identity.session_id)
+}
+
 #[cfg(all(test, feature = "export"))]
 mod identity_tests {
     use super::*;

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:?}"
+        );
+    }
+}