harn_session_store/
store.rs

1//! `SessionStore` trait + the shared types every backend speaks.
2
3use std::collections::BTreeMap;
4use std::sync::Arc;
5
6use async_trait::async_trait;
7use serde::{Deserialize, Serialize};
8
9use super::event::{AppendEvent, EventId, StoredEvent};
10use super::redaction::SharedEventRedactor;
11use super::retention::{RetentionPolicy, SharedArchiveSink, Tombstone};
12use super::search::{default_embedder, Embedder, SearchQuery, SearchResponse};
13use super::signing::SessionSigner;
14
15pub type SessionId = String;
16
17/// Result of a fork operation.
18#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
19pub struct ForkResult {
20    pub child_session_id: SessionId,
21    pub forked_from_event_id: EventId,
22    pub copied_event_count: usize,
23}
24
25#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
26pub struct TruncateResult {
27    pub kept_event_count: usize,
28    pub removed_event_count: usize,
29    pub new_tip_event_id: Option<EventId>,
30}
31
32/// Per-session retention/lifecycle state.
33#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
34#[serde(rename_all = "snake_case")]
35pub enum SessionStatus {
36    Open,
37    Closed,
38    /// Soft-deleted; will become `HardDeleted` once the grace window
39    /// elapses (enforced by retention sweeps; see [`crate::retention`]).
40    SoftDeleted,
41    HardDeleted,
42}
43
44#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
45#[serde(rename_all = "snake_case")]
46pub enum SessionType {
47    User,
48    Subagent,
49    Scheduled,
50}
51
52#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
53pub struct SessionMeta {
54    pub id: SessionId,
55    pub tenant_id: Option<String>,
56    pub persona: Option<String>,
57    pub parent_session_id: Option<SessionId>,
58    #[serde(default)]
59    pub title: Option<String>,
60    /// True when a person chose this title. Derived titles never overwrite it.
61    #[serde(default)]
62    pub title_pinned: bool,
63    #[serde(default)]
64    pub cwd: Option<String>,
65    #[serde(default)]
66    pub model: Option<String>,
67    #[serde(default)]
68    pub session_type: Option<SessionType>,
69    #[serde(default)]
70    pub project_scope: Option<String>,
71    #[serde(default)]
72    pub usage_input: u64,
73    #[serde(default)]
74    pub usage_output: u64,
75    /// Cost in millionths of a US dollar, avoiding floating-point drift.
76    #[serde(default)]
77    pub usage_cost_usd_micros: u64,
78    pub created_at_ms: i64,
79    pub created_at: String,
80    /// Last `append`/`fork` timestamp; refreshed on every mutation.
81    pub updated_at_ms: i64,
82    pub updated_at: String,
83    pub status: SessionStatus,
84    pub event_count: usize,
85    pub last_event_id: Option<EventId>,
86    pub chain_root_hash: Option<String>,
87    pub closed_at_ms: Option<i64>,
88    pub closed_at: Option<String>,
89    pub soft_deleted_at_ms: Option<i64>,
90    pub ttl_seconds: Option<u64>,
91    pub tags: Vec<String>,
92    pub attributes: BTreeMap<String, serde_json::Value>,
93}
94
95#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
96pub struct CreateSession {
97    #[serde(default)]
98    pub id: Option<SessionId>,
99    #[serde(default)]
100    pub tenant_id: Option<String>,
101    #[serde(default)]
102    pub persona: Option<String>,
103    #[serde(default)]
104    pub parent_session_id: Option<SessionId>,
105    #[serde(default)]
106    pub title: Option<String>,
107    /// Seed a session whose title is already a person's choice, so a rename
108    /// against a not-yet-persisted session survives its first derived write.
109    #[serde(default)]
110    pub title_pinned: bool,
111    #[serde(default)]
112    pub cwd: Option<String>,
113    #[serde(default)]
114    pub model: Option<String>,
115    #[serde(default)]
116    pub session_type: Option<SessionType>,
117    #[serde(default)]
118    pub project_scope: Option<String>,
119    #[serde(default)]
120    pub usage_input: u64,
121    #[serde(default)]
122    pub usage_output: u64,
123    #[serde(default)]
124    pub usage_cost_usd_micros: u64,
125    #[serde(default)]
126    pub ttl_seconds: Option<u64>,
127    #[serde(default)]
128    pub tags: Vec<String>,
129    #[serde(default)]
130    pub attributes: BTreeMap<String, serde_json::Value>,
131}
132
133/// Typed mutable facts for an existing session.
134///
135/// Omitted fields are preserved. Clearing metadata is deliberately not part of
136/// this contract; lifecycle operations own destructive state transitions.
137#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
138pub struct UpdateSession {
139    #[serde(default)]
140    pub title: Option<String>,
141    /// Whether `title` is a person's choice, which decides who may overwrite
142    /// whom. Leaving this `None` marks the write as *derived* — a title
143    /// generated from session content — and derived writes never overwrite a
144    /// pinned title. That default is what makes an auto-titling caller
145    /// fail-safe without having to know pinning exists.
146    ///
147    /// | `title` | `title_pinned` | effect |
148    /// |---|---|---|
149    /// | `Some` | `Some(true)` | rename: set the title and pin it |
150    /// | `Some` | `Some(false)` | retitle and release the pin |
151    /// | `None` | `Some(_)` | change the pin state, keep the title |
152    /// | `Some` | `None` | derived: applied only while unpinned |
153    /// | `None` | `None` | no title change |
154    #[serde(default)]
155    pub title_pinned: Option<bool>,
156    #[serde(default)]
157    pub cwd: Option<String>,
158    #[serde(default)]
159    pub model: Option<String>,
160    #[serde(default)]
161    pub parent_session_id: Option<SessionId>,
162    #[serde(default)]
163    pub session_type: Option<SessionType>,
164    #[serde(default)]
165    pub project_scope: Option<String>,
166    #[serde(default)]
167    pub usage_input: Option<u64>,
168    #[serde(default)]
169    pub usage_output: Option<u64>,
170    #[serde(default)]
171    pub usage_cost_usd_micros: Option<u64>,
172}
173
174/// One atomic, idempotent import into a new canonical session.
175///
176/// `source_id` names the external source independently of the target session;
177/// its receipt survives session deletion so retired sources cannot resurrect
178/// data. Reusing a source id with a different digest is a conflict.
179#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
180pub struct ImportSession {
181    pub source_id: String,
182    pub source_digest: String,
183    pub session: CreateSession,
184    #[serde(default)]
185    pub events: Vec<AppendEvent>,
186}
187
188impl ImportSession {
189    /// Validate the backend-independent import contract.
190    pub fn validate(&self) -> StoreResult<()> {
191        if self.source_id.trim().is_empty() || self.source_digest.trim().is_empty() {
192            return Err(StoreError::InvalidInput(
193                "import source_id and source_digest must be non-empty".to_string(),
194            ));
195        }
196        if self
197            .session
198            .id
199            .as_deref()
200            .is_none_or(|session_id| session_id.trim().is_empty())
201        {
202            return Err(StoreError::InvalidInput(
203                "import session id must be explicit and non-empty".to_string(),
204            ));
205        }
206        Ok(())
207    }
208}
209
210#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
211pub struct ImportResult {
212    pub source_id: String,
213    pub source_digest: String,
214    pub session_id: SessionId,
215    pub event_count: usize,
216    /// True only for the call that committed the import.
217    pub imported: bool,
218}
219
220#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
221pub struct ListFilter {
222    #[serde(default)]
223    pub tenant_id: Option<String>,
224    #[serde(default)]
225    pub persona: Option<String>,
226    #[serde(default)]
227    pub status: Option<SessionStatus>,
228    #[serde(default)]
229    pub tag: Option<String>,
230    #[serde(default)]
231    pub parent_session_id: Option<SessionId>,
232    #[serde(default)]
233    pub session_type: Option<SessionType>,
234    #[serde(default)]
235    pub project_scope: Option<String>,
236    /// Inclusive lower bound on `created_at_ms`.
237    #[serde(default)]
238    pub created_after_ms: Option<i64>,
239    /// Inclusive upper bound on `created_at_ms`.
240    #[serde(default)]
241    pub created_before_ms: Option<i64>,
242    #[serde(default)]
243    pub limit: Option<usize>,
244    #[serde(default)]
245    pub cursor: Option<String>,
246    /// Timestamp used to order and paginate matching sessions.
247    #[serde(default)]
248    pub sort_by: ListSortKey,
249    /// Direction for `sort_by`. Session ids remain ascending as the stable
250    /// tie-breaker in both directions.
251    #[serde(default)]
252    pub order: ListOrder,
253}
254
255#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
256#[serde(rename_all = "snake_case")]
257pub enum ListSortKey {
258    #[default]
259    CreatedAt,
260    UpdatedAt,
261}
262
263#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
264#[serde(rename_all = "snake_case")]
265pub enum ListOrder {
266    #[default]
267    Ascending,
268    Descending,
269}
270
271#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
272pub struct ReadRange {
273    /// Inclusive lower bound; `None` means start from the genesis event.
274    #[serde(default)]
275    pub from_event_id: Option<EventId>,
276    /// Inclusive upper bound; `None` means up to the latest event.
277    #[serde(default)]
278    pub to_event_id: Option<EventId>,
279    /// Maximum number of events to return. Capped at
280    /// [`MAX_READ_BATCH`] by the store; callers iterate by advancing
281    /// `from_event_id` on the returned cursor.
282    #[serde(default)]
283    pub limit: Option<usize>,
284}
285
286/// Page of events plus the cursor needed to continue reading.
287#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
288pub struct EventPage {
289    pub events: Vec<StoredEvent>,
290    /// Inclusive `from_event_id` to pass into the next read to resume.
291    /// `None` when the requested range was fully drained.
292    pub next_cursor: Option<EventId>,
293}
294
295#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
296pub struct SnapshotId(pub String);
297
298#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
299pub struct Snapshot {
300    pub id: SnapshotId,
301    pub session: SessionMeta,
302    pub events: Vec<StoredEvent>,
303    pub captured_at_ms: i64,
304    pub captured_at: String,
305}
306
307#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
308pub struct VerifyReport {
309    pub session_id: SessionId,
310    pub chain_root_hash: String,
311    pub event_count: usize,
312    pub signed_event_count: usize,
313    pub failures: Vec<VerifyFailure>,
314}
315
316#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
317pub struct VerifyFailure {
318    pub event_id: EventId,
319    pub reason: String,
320}
321
322/// Typed transient contention returned by a persistent backend.
323#[derive(Clone, Copy, Debug, PartialEq, Eq)]
324pub enum StoreContention {
325    /// Another SQLite connection owns the database write lock.
326    DatabaseBusy,
327    /// A shared-cache SQLite table lock blocks the operation.
328    DatabaseLocked,
329}
330
331impl std::fmt::Display for StoreContention {
332    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
333        match self {
334            Self::DatabaseBusy => f.write_str("database_busy"),
335            Self::DatabaseLocked => f.write_str("database_locked"),
336        }
337    }
338}
339
340/// Errors returned by every backend.
341#[derive(Clone, Debug, PartialEq, Eq)]
342pub enum StoreError {
343    NotFound(String),
344    AlreadyExists(String),
345    Conflict(String),
346    InvalidInput(String),
347    Tenant(String),
348    /// SQLite could not acquire a lock after applying its busy policy.
349    Contention {
350        /// Machine-readable lock-contention reason.
351        kind: StoreContention,
352        /// Backend diagnostic retained for operators.
353        message: String,
354    },
355    /// The persistent store was written by a newer incompatible schema owner.
356    SchemaIncompatible {
357        /// Machine-readable schema owner name.
358        schema: String,
359        /// Version recorded in the persistent store.
360        stored: i64,
361        /// Newest version this runtime can safely open.
362        supported: i64,
363    },
364    Backend(String),
365}
366
367impl std::fmt::Display for StoreError {
368    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
369        match self {
370            Self::NotFound(message) => write!(f, "not found: {message}"),
371            Self::AlreadyExists(message) => write!(f, "already exists: {message}"),
372            Self::Conflict(message) => write!(f, "conflict: {message}"),
373            Self::InvalidInput(message) => write!(f, "invalid input: {message}"),
374            Self::Tenant(message) => write!(f, "tenant: {message}"),
375            Self::Contention { kind, message } => {
376                write!(f, "retryable backend contention ({kind}): {message}")
377            }
378            Self::SchemaIncompatible {
379                schema,
380                stored,
381                supported,
382            } => write!(
383                f,
384                "schema incompatible: {schema} version {stored} is newer than supported version {supported}"
385            ),
386            Self::Backend(message) => write!(f, "backend error: {message}"),
387        }
388    }
389}
390
391impl std::error::Error for StoreError {}
392
393pub type StoreResult<T> = Result<T, StoreError>;
394
395/// Soft cap on a single page of events to keep response sizes bounded.
396/// Memory + sqlite backends apply this; callers iterate via cursors.
397pub const MAX_READ_BATCH: usize = 1_000;
398
399/// Optional processors a host can plug in. Mutation hooks run inline before
400/// persistence; redaction is also reapplied to public retrieval projections
401/// as defense in depth for older stored data.
402#[derive(Clone)]
403pub struct StoreHooks {
404    /// Applied to event payloads and headers before persistence and again
405    /// when events are read, snapshotted, or replayed.
406    pub redaction: Option<SharedEventRedactor>,
407    /// If set, every event is signed at append time. Without a signer
408    /// only the `Receipt` event minted by [`SessionStore::close`] is
409    /// signed (which is enough to verify the chain end-to-end).
410    pub event_signer: Option<SessionSigner>,
411    /// Required to mint receipts on `close`. Without this, `close`
412    /// still finalises the chain root hash but emits an unsigned
413    /// `Receipt` event.
414    pub receipt_signer: Option<SessionSigner>,
415    /// Default retention policy applied to new sessions when their
416    /// meta does not override it.
417    pub retention: RetentionPolicy,
418    /// Optional durable archive destination. The default
419    /// [`SessionStore::sweep_retention`] writes archived sessions and
420    /// tombstones here before the rows leave primary storage; see
421    /// [`super::retention::ArchiveSink`].
422    pub archive_sink: Option<SharedArchiveSink>,
423    /// Embedding implementation used to index and rank canonical redacted
424    /// transcript rows. The default deterministic lexical floor is
425    /// cross-platform and requires no model asset.
426    pub embedder: Arc<dyn Embedder>,
427}
428
429impl Default for StoreHooks {
430    fn default() -> Self {
431        Self {
432            redaction: None,
433            event_signer: None,
434            receipt_signer: None,
435            retention: RetentionPolicy::default(),
436            archive_sink: None,
437            embedder: default_embedder(),
438        }
439    }
440}
441
442#[async_trait]
443pub trait SessionStore: Send + Sync {
444    /// Plug-in processors configured for this store. The default
445    /// [`Self::sweep_retention`] reads `hooks.archive_sink` so the
446    /// retention loop can hand archived sessions to durable storage
447    /// without callers wiring the sink explicitly.
448    fn hooks(&self) -> &StoreHooks;
449
450    async fn create(&self, request: CreateSession) -> StoreResult<SessionMeta>;
451    async fn update(&self, session_id: &str, request: UpdateSession) -> StoreResult<SessionMeta>;
452    async fn describe(&self, session_id: &str) -> StoreResult<SessionMeta>;
453    async fn list(&self, filter: ListFilter) -> StoreResult<Vec<SessionMeta>>;
454    async fn append(&self, session_id: &str, event: AppendEvent) -> StoreResult<StoredEvent>;
455    async fn read(&self, session_id: &str, range: ReadRange) -> StoreResult<EventPage>;
456    async fn fork(
457        &self,
458        session_id: &str,
459        at_event_id: EventId,
460        child_id: Option<SessionId>,
461    ) -> StoreResult<ForkResult>;
462    async fn truncate(&self, session_id: &str, at_event_id: EventId)
463        -> StoreResult<TruncateResult>;
464    async fn snapshot(&self, session_id: &str) -> StoreResult<Snapshot>;
465    async fn replay(&self, snapshot_id: &SnapshotId) -> StoreResult<Snapshot>;
466    async fn close(&self, session_id: &str) -> StoreResult<StoredEvent>;
467    async fn soft_delete(&self, session_id: &str) -> StoreResult<SessionMeta>;
468    async fn hard_delete(&self, session_id: &str) -> StoreResult<()>;
469    async fn verify(&self, session_id: &str) -> StoreResult<VerifyReport>;
470    async fn search(&self, query: SearchQuery) -> StoreResult<SearchResponse>;
471
472    /// Sweep retention. Backends with native scheduling can override
473    /// to skip the default loop; the default sweeps all sessions
474    /// against the configured [`RetentionPolicy`] and routes archived
475    /// sessions + tombstones through `hooks().archive_sink` when set.
476    async fn sweep_retention(
477        &self,
478        policy: &RetentionPolicy,
479        now_ms: i64,
480    ) -> StoreResult<SweepReport> {
481        use tracing::Instrument as _;
482        let span = tracing::info_span!(
483            "harn.session.sweep_retention",
484            harn.session.sweep.archive_sink_configured = self.hooks().archive_sink.is_some(),
485            harn.session.sweep.archived = tracing::field::Empty,
486            harn.session.sweep.soft_deleted = tracing::field::Empty,
487            harn.session.sweep.hard_deleted = tracing::field::Empty,
488        );
489        let span_for_record = span.clone();
490        let sink = self.hooks().archive_sink.clone();
491        let result = async move {
492            let mut report = SweepReport::default();
493            let sessions = self.list(ListFilter::default()).await?;
494            for session in sessions {
495                if policy.should_hard_delete(&session, now_ms) {
496                    if let Some(sink) = sink.as_ref() {
497                        let tombstone = Tombstone {
498                            session_id: session.id.clone(),
499                            tenant_id: session.tenant_id.clone(),
500                            deleted_at_ms: now_ms,
501                            deleted_at: super::event::ms_to_rfc3339(now_ms),
502                            final_chain_root_hash: session.chain_root_hash.clone(),
503                            final_event_id: session.last_event_id,
504                        };
505                        sink.tombstone(&tombstone).await?;
506                        report.tombstoned += 1;
507                    }
508                    self.hard_delete(&session.id).await?;
509                    report.hard_deleted += 1;
510                } else if policy.should_soft_delete(&session, now_ms) {
511                    if policy.should_archive(&session, now_ms) {
512                        if let Some(sink) = sink.as_ref() {
513                            let events = read_all_events(self, &session.id).await?;
514                            sink.archive(&session, &events).await?;
515                            report.archived += 1;
516                        }
517                    }
518                    self.soft_delete(&session.id).await?;
519                    report.soft_deleted += 1;
520                }
521            }
522            Ok::<_, StoreError>(report)
523        }
524        .instrument(span)
525        .await?;
526        span_for_record.record("harn.session.sweep.archived", result.archived as i64);
527        span_for_record.record(
528            "harn.session.sweep.soft_deleted",
529            result.soft_deleted as i64,
530        );
531        span_for_record.record(
532            "harn.session.sweep.hard_deleted",
533            result.hard_deleted as i64,
534        );
535        Ok(result)
536    }
537}
538
539/// Atomic, idempotent ingestion for stores that accept external session data.
540///
541/// This remains separate from [`SessionStore`] so downstream backends do not
542/// need to implement migration semantics unless they expose import support.
543#[async_trait]
544pub trait SessionImporter: SessionStore {
545    async fn import(&self, request: ImportSession) -> StoreResult<ImportResult>;
546}
547
548/// Drain every event for a session via repeated paginated reads. Used
549/// by [`SessionStore::sweep_retention`] when shipping a session to the
550/// [`super::retention::ArchiveSink`].
551async fn read_all_events<S: SessionStore + ?Sized>(
552    store: &S,
553    session_id: &str,
554) -> StoreResult<Vec<StoredEvent>> {
555    let mut all = Vec::new();
556    let mut cursor: Option<EventId> = None;
557    loop {
558        let page = store
559            .read(
560                session_id,
561                ReadRange {
562                    from_event_id: cursor,
563                    to_event_id: None,
564                    limit: Some(MAX_READ_BATCH),
565                },
566            )
567            .await?;
568        let next = page.next_cursor;
569        all.extend(page.events);
570        match next {
571            Some(next_cursor) => cursor = Some(next_cursor),
572            None => break,
573        }
574    }
575    Ok(all)
576}
577
578#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
579pub struct SweepReport {
580    pub soft_deleted: usize,
581    pub hard_deleted: usize,
582    /// Sessions handed to [`super::retention::ArchiveSink::archive`]
583    /// because they crossed `min_age_before_archive_seconds`.
584    pub archived: usize,
585    /// Hard-deleted sessions whose final state was emitted as a
586    /// [`super::retention::Tombstone`] to the archive sink.
587    pub tombstoned: usize,
588}
589
590/// Dyn-dispatch alias so adapters can keep one `Arc<dyn SessionStore>`
591/// in their state without naming the concrete backend everywhere.
592pub type SharedSessionStore = Arc<dyn SessionStore>;
harn_session_store/store.rs

harn_session_store/
store.rs
