Repository navigation
[G2] Change feed: revision-stamped SessionStore::changes_since() with named consumer cursors, for incremental downstream ingest #179
Copy link
Copy link
Closed
Description
Part of #160 (group 2 — depends on #162 single crate, #173 incremental hydration; parallel with #178/#180/#181/#182).
Problem
burn's watch loop needs "what changed since my last tick" without rescanning the catalog. relayhistory has the pieces but only for the cloud outbox:
crates/ai-hist-core/src/storage.rs:history_after,trajectories_after,commit_links_after,changed_file_sessions_after,session_identities_after,pending_event_sessions,latest_history_for_session— watermark readers over monotonic ids.plugins/relayhistory/rust/src/outbox.rs:9: "Cursor model (mirrors burn'sarchive_statewatermark): monotonichistory.id…" — the two repos already share the cursor idea, but relayhistory's is plugin-private and covershistory, notsession_events/tool_calls/file_edits/markers/relationships.- The delivery coordinator (
delivery*,HistoryExportRecord { record_id, revision_id, revision, kind, operation: upsert|delete, payload }) is a full at-least-once queue with journal, leases and retention caps (delivery/schema.rsRAISE(ABORT, 'delivery retention limit exceeded…')). It is heavier than a pull cursor and its retention cap would block ingest if a consumer fell behind — wrong shape for an in-process Rust consumer (and it is behind thedeliveryfeature after [G1] Merge ai-hist-core and ai-hist-engine into one publishedai-histcrate with feature gates, a release pipeline and a semver policy #162). - Session-level "dirty" tracking exists (
changed_file_sessions_after, hydration checkpointsupdated_ms) but no per-kind revision onsession_eventsetc.
Scope
- Revision column: add
revision INTEGER NOT NULL(monotonic, fromobservation_clock,observations.rs:118) tosession_events,tool_calls,file_edits,session_markers,session_relationships,sessions,session_requests; every upsert stamps the current revision; deletes (e.g. sidechain rows removed on re-parse,lib.rs:3026-3031) write a tombstone row toevidence_tombstones(kind, source, session_id, record_key, revision). Index(revision)per table. - API — this is the seventh facade method in [G2]
SessionStorefacade: the single public entry point (seven methods, typed evidence, no raw connections, no contract constant) #178, no others:Named consumer cursors live in a smallpub struct Watermark { pub revision: u64 } pub struct Change { pub kind: EvidenceKind, pub source: Source, pub session_id: String, pub record_key: String, pub revision: u64, pub op: ChangeOp /* Upsert(EvidenceRow) | Delete */ } pub struct ChangeQuery { pub kinds: Option<Vec<EvidenceKind>>, pub consumer: Option<String> /* named cursor: read start from it, and advance it as the iterator is drained */, pub batch: usize /* ≤ 10,000 */ } impl SessionStore { pub fn changes_since(&self, from: Watermark, q: ChangeQuery) -> impl Iterator<Item = Result<Change, Error>>; // The iterator exposes .head() -> Watermark (store head at open) and .position() -> Watermark (last yielded). }
consumer_cursors(name, revision, updated_ms)table so an embedder's progress survives its own ledger reset; whenq.consumeris set,frommay beWatermark::CONSUMERto mean "wherever that consumer left off", and the cursor advances only when the consumer calls.commit()on the iterator (so a consumer that fails mid-batch resumes from its last commit, not from the head). Ordered by(revision, kind, record_key);Upsertcarries the typedEvidenceRowso no second round trip is needed. - In-progress safety: rows for a held (in-progress) message ([G1] Incremental transcript hydration: byte-offset cursors, prefix validation, in-progress message handling, bounded memory #173) are never visible in the feed until committed, so a consumer that materialises turns sees each message exactly once with its final usage. Test: append a partial Claude message, tick, assert no change; complete it, tick, assert one upsert per block + one
session_requestsupsert. - Re-hydration semantics: a parser-version re-parse (
HYDRATION_PARSER_VERSIONbump) re-stamps every row of that session with a new revision — document that consumers must treat a re-seenrecord_keyas a replace, not a duplicate (burn'sid_fingerprintupserts already do). SyncReport/statsexposehead_revision; a consumer whose stored cursor exceeds it getsError::WatermarkAheadOfStore(DB was reset → full resync).- TS/napi:
getChangesSince, consumer cursor commit — rides [G2] Expose the new evidence through napi, the TypeScript SDK and MCP (markers, tool-result facts, usage/requests, control kinds, continuity, change feed) with contract bumps #181; the Rust side lands here.
Acceptance
- Property test: for any sequence of hydrations over the corpus, replaying
changes_since(Watermark{0})reconstructs the same table contents as a direct read (modulo tombstoned rows). - Two named consumers advance independently; an uncommitted drain does not move the cursor.
- A consumer whose stored watermark exceeds
head_revisiongetsWatermarkAheadOfStore. - Batches of 10,000 rows over a 1 GB synthetic store (from [G1] Full-sync and hydration throughput baseline, memory ceiling, and a CI regression gate #176) complete in bounded time with the index (query-plan assertion, no full scan).
Out of scope
Push/subscription callbacks; cloud delivery changes.
Reactions are currently unavailable
Activity
[-][G2] Change feed: watermark-cursored `evidence_after()` across kinds with named consumer cursors, for incremental downstream ingest[/-][+][G2] Change feed: revision-stamped `SessionStore::changes_since()` with named consumer cursors, for incremental downstream ingest[/+]Digest the catalog under the sync lock; refuse path refs that name no…