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

harn_session_store/
store.rs
