zeph_session/
store.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! [`SessionStore`]: the `acp_sessions` metadata index.
5//!
6//! Promotes the existing `acp_sessions` table (migration 013, `crates/zeph-memory`) to a
7//! channel-agnostic conversation-session index (spec-068 §2 Decision D1 — no new `sessions`
8//! table is introduced). `zeph-session` talks to the table directly via [`zeph_db::DbPool`]
9//! rather than depending on `zeph-memory`, keeping the crate boundary intact.
10//!
11//! The event log ([`crate::log::SessionEventLog`]) is the source of truth for conversation
12//! content; this store only tracks lightweight, queryable metadata (`last_seq`, `status`,
13//! fork provenance) used to reconcile the projection on open (INV-SP-3) and to answer
14//! `sessions list` without replaying every log.
15
16use zeph_db::{ActiveDialect, DbPool, dialect::Dialect, sql};
17
18use crate::error::SessionError;
19
20/// Lifecycle status of a conversation-session, mirroring the `acp_sessions.status` CHECK
21/// constraint added in migration 106.
22#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
23#[serde(rename_all = "snake_case")]
24pub enum SessionStatus {
25    /// Actively attached to a live agent/actor.
26    Active,
27    /// Persisted but not currently attached.
28    Idle,
29    /// Explicitly archived; excluded from default `list` results.
30    Archived,
31}
32
33impl SessionStatus {
34    /// The `TEXT` representation stored in the `status` column.
35    #[must_use]
36    pub fn as_str(self) -> &'static str {
37        match self {
38            Self::Active => "active",
39            Self::Idle => "idle",
40            Self::Archived => "archived",
41        }
42    }
43}
44
45impl std::str::FromStr for SessionStatus {
46    type Err = SessionError;
47
48    fn from_str(s: &str) -> Result<Self, Self::Err> {
49        match s {
50            "active" => Ok(Self::Active),
51            "idle" => Ok(Self::Idle),
52            "archived" => Ok(Self::Archived),
53            other => Err(SessionError::NotFound(format!(
54                "unknown session status: {other}"
55            ))),
56        }
57    }
58}
59
60/// A conversation-session's metadata row, as tracked in `acp_sessions`.
61#[derive(Debug, Clone, serde::Serialize)]
62pub struct SessionMetadata {
63    pub session_id: String,
64    pub title: Option<String>,
65    pub created_at: String,
66    pub updated_at: String,
67    pub conversation_id: Option<i64>,
68    pub last_seq: u64,
69    pub event_count: u64,
70    pub forked_from: Option<String>,
71    pub forked_at_seq: Option<u64>,
72    pub status: SessionStatus,
73    pub last_condensed_seq: u64,
74}
75
76/// Filter parameters for [`SessionStore::list`].
77#[derive(Debug, Clone, Default)]
78pub struct SessionFilter {
79    /// Restrict to a single status; `None` returns all statuses.
80    pub status: Option<SessionStatus>,
81    /// Maximum rows returned; `0` means unlimited.
82    pub limit: usize,
83}
84
85/// CRUD access to the `acp_sessions` metadata index.
86pub struct SessionStore {
87    pool: DbPool,
88}
89
90impl SessionStore {
91    /// Wrap an existing [`DbPool`]. `zeph-session` does not own a dedicated database file —
92    /// it shares the pool that already owns `acp_sessions` (migration 013).
93    #[must_use]
94    pub fn new(pool: DbPool) -> Self {
95        Self { pool }
96    }
97
98    /// Insert a new session row with `status = 'active'`, ignoring the call if the row already
99    /// exists (idempotent, mirrors the existing `create_acp_session` pattern).
100    ///
101    /// # Errors
102    ///
103    /// Returns [`SessionError::Db`] if the write fails.
104    #[tracing::instrument(name = "session.store.create", skip_all, level = "debug")]
105    pub async fn create(&self, session_id: &str) -> Result<(), SessionError> {
106        let stmt = zeph_db::rewrite_placeholders(&format!(
107            "{} INTO acp_sessions (id, status) VALUES (?, 'active'){}",
108            <ActiveDialect as Dialect>::INSERT_IGNORE,
109            <ActiveDialect as Dialect>::CONFLICT_NOTHING,
110        ));
111        zeph_db::query(sqlx::AssertSqlSafe(stmt))
112            .bind(session_id)
113            .execute(&self.pool)
114            .await?;
115        Ok(())
116    }
117
118    /// Update `last_seq`, `event_count`, and `updated_at` after a turn's events are flushed to
119    /// the log (INV-SP-1: called only after the log append is durable).
120    ///
121    /// Explicitly bumps `updated_at` here because the pre-cutover `AFTER INSERT ON
122    /// acp_session_events` trigger (migration 017) that used to drive it never fires for
123    /// post-cutover sessions (spec-068 §12.3 / D-2: `acp_session_events` is a write target only
124    /// for legacy pre-cutover sessions) — without this, `list_acp_sessions`' "ordered by last
125    /// activity descending" would silently degrade to "ordered by creation time" for every
126    /// session created after the cutover.
127    ///
128    /// # Errors
129    ///
130    /// Returns [`SessionError::Db`] if the write fails.
131    #[allow(clippy::cast_possible_wrap)]
132    #[tracing::instrument(name = "session.store.update_seq", skip_all, level = "debug")]
133    pub async fn update_seq(
134        &self,
135        session_id: &str,
136        last_seq: u64,
137        event_count: u64,
138    ) -> Result<(), SessionError> {
139        let stmt = zeph_db::rewrite_placeholders(&format!(
140            "UPDATE acp_sessions SET last_seq = ?, event_count = ?, updated_at = {} WHERE id = ?",
141            <ActiveDialect as Dialect>::NOW,
142        ));
143        zeph_db::query(sqlx::AssertSqlSafe(stmt))
144            .bind(last_seq as i64)
145            .bind(event_count as i64)
146            .bind(session_id)
147            .execute(&self.pool)
148            .await?;
149        Ok(())
150    }
151
152    /// Update the session's lifecycle status.
153    ///
154    /// # Errors
155    ///
156    /// Returns [`SessionError::Db`] if the write fails.
157    #[tracing::instrument(name = "session.store.set_status", skip_all, level = "debug")]
158    pub async fn set_status(
159        &self,
160        session_id: &str,
161        status: SessionStatus,
162    ) -> Result<(), SessionError> {
163        zeph_db::query(sql!("UPDATE acp_sessions SET status = ? WHERE id = ?"))
164            .bind(status.as_str())
165            .bind(session_id)
166            .execute(&self.pool)
167            .await?;
168        Ok(())
169    }
170
171    /// Update the high-water condensation mark (INV-SP-4 non-overlap tracking).
172    ///
173    /// # Errors
174    ///
175    /// Returns [`SessionError::Db`] if the write fails.
176    #[allow(clippy::cast_possible_wrap)]
177    #[tracing::instrument(name = "session.store.set_condensed_seq", skip_all, level = "debug")]
178    pub async fn set_condensed_seq(
179        &self,
180        session_id: &str,
181        last_condensed_seq: u64,
182    ) -> Result<(), SessionError> {
183        zeph_db::query(sql!(
184            "UPDATE acp_sessions SET last_condensed_seq = ? WHERE id = ?"
185        ))
186        .bind(last_condensed_seq as i64)
187        .bind(session_id)
188        .execute(&self.pool)
189        .await?;
190        Ok(())
191    }
192
193    /// Fetch a single session's metadata.
194    ///
195    /// # Errors
196    ///
197    /// Returns [`SessionError::Db`] if the query fails.
198    #[tracing::instrument(name = "session.store.get", skip_all, level = "debug")]
199    pub async fn get(&self, session_id: &str) -> Result<Option<SessionMetadata>, SessionError> {
200        let row = zeph_db::query_as::<_, SessionRow>(sql!(
201            "SELECT id, title, created_at, updated_at, conversation_id, last_seq, event_count, \
202             forked_from, forked_at_seq, status, last_condensed_seq \
203             FROM acp_sessions WHERE id = ?"
204        ))
205        .bind(session_id)
206        .fetch_optional(&self.pool)
207        .await?;
208        row.map(TryInto::try_into).transpose()
209    }
210
211    /// List sessions, most recently updated first.
212    ///
213    /// # Errors
214    ///
215    /// Returns [`SessionError::Db`] if the query fails.
216    #[tracing::instrument(name = "session.store.list", skip_all, level = "debug")]
217    pub async fn list(&self, filter: &SessionFilter) -> Result<Vec<SessionMetadata>, SessionError> {
218        #[allow(clippy::cast_possible_wrap)]
219        let sql_limit: i64 = if filter.limit == 0 {
220            -1
221        } else {
222            filter.limit as i64
223        };
224        let status_filter = filter.status.map(SessionStatus::as_str);
225
226        let rows = zeph_db::query_as::<_, SessionRow>(sql!(
227            "SELECT id, title, created_at, updated_at, conversation_id, last_seq, event_count, \
228             forked_from, forked_at_seq, status, last_condensed_seq \
229             FROM acp_sessions \
230             WHERE (? IS NULL OR status = ?) \
231             ORDER BY updated_at DESC LIMIT ?"
232        ))
233        .bind(status_filter)
234        .bind(status_filter)
235        .bind(sql_limit)
236        .fetch_all(&self.pool)
237        .await?;
238
239        rows.into_iter().map(TryInto::try_into).collect()
240    }
241
242    /// Link this session to a `ConversationId` (raw `i64` — `zeph-session` does not depend on
243    /// `zeph-memory`'s newtype), enforcing the `SessionId`<->`ConversationId` bijection (spec
244    /// §5.2) via the unique partial index added in migration 106.
245    ///
246    /// # Errors
247    ///
248    /// Returns [`SessionError::Db`] if the write fails (including a unique-constraint violation
249    /// when `conversation_id` is already linked to a different session).
250    #[tracing::instrument(name = "session.store.link_conversation", skip_all, level = "debug")]
251    pub async fn link_conversation(
252        &self,
253        session_id: &str,
254        conversation_id: i64,
255    ) -> Result<(), SessionError> {
256        zeph_db::query(sql!(
257            "UPDATE acp_sessions SET conversation_id = ? WHERE id = ?"
258        ))
259        .bind(conversation_id)
260        .bind(session_id)
261        .execute(&self.pool)
262        .await?;
263        Ok(())
264    }
265
266    /// Look up the session already linked to a `ConversationId`, if any.
267    ///
268    /// Used at non-ACP channel startup (CLI/TUI/Telegram) to resume the same conversation's
269    /// existing session (and its event log) across process restarts, rather than minting a new
270    /// `SessionId` every launch (spec §12.2).
271    ///
272    /// # Errors
273    ///
274    /// Returns [`SessionError::Db`] if the query fails.
275    #[tracing::instrument(
276        name = "session.store.get_by_conversation_id",
277        skip_all,
278        level = "debug"
279    )]
280    pub async fn get_by_conversation_id(
281        &self,
282        conversation_id: i64,
283    ) -> Result<Option<SessionMetadata>, SessionError> {
284        let row = zeph_db::query_as::<_, SessionRow>(sql!(
285            "SELECT id, title, created_at, updated_at, conversation_id, last_seq, event_count, \
286             forked_from, forked_at_seq, status, last_condensed_seq \
287             FROM acp_sessions WHERE conversation_id = ?"
288        ))
289        .bind(conversation_id)
290        .fetch_optional(&self.pool)
291        .await?;
292        row.map(TryInto::try_into).transpose()
293    }
294
295    /// Record a fork: sets `forked_from`/`forked_at_seq` on the child row.
296    ///
297    /// Does not touch the parent's log (the `ForkPoint` provenance event is appended by
298    /// [`crate::replay`]'s `ForkEngine`, which owns the parent's [`crate::log::SessionEventLog`]).
299    ///
300    /// # Errors
301    ///
302    /// Returns [`SessionError::Db`] if either write fails.
303    #[allow(clippy::cast_possible_wrap)]
304    #[tracing::instrument(name = "session.store.record_fork", skip_all, level = "debug")]
305    pub async fn record_fork(
306        &self,
307        new_session_id: &str,
308        src_session_id: &str,
309        forked_at_seq: u64,
310    ) -> Result<(), SessionError> {
311        let stmt = zeph_db::rewrite_placeholders(&format!(
312            "{} INTO acp_sessions (id, status, forked_from, forked_at_seq) VALUES (?, 'active', ?, ?){}",
313            <ActiveDialect as Dialect>::INSERT_IGNORE,
314            <ActiveDialect as Dialect>::CONFLICT_NOTHING,
315        ));
316        zeph_db::query(sqlx::AssertSqlSafe(stmt))
317            .bind(new_session_id)
318            .bind(src_session_id)
319            .bind(forked_at_seq as i64)
320            .execute(&self.pool)
321            .await?;
322        Ok(())
323    }
324
325    /// Delete a session's metadata row. Returns `true` if a row was deleted.
326    ///
327    /// Does not remove the on-disk event log directory or blobs — callers with access to
328    /// `[session] data_dir` are responsible for that (mirrors the separation of concerns between
329    /// [`SessionStore`] and [`crate::log::SessionEventLog`]).
330    ///
331    /// # Errors
332    ///
333    /// Returns [`SessionError::Db`] if the write fails.
334    #[tracing::instrument(name = "session.store.delete", skip_all, level = "debug")]
335    pub async fn delete(&self, session_id: &str) -> Result<bool, SessionError> {
336        let result = zeph_db::query(sql!("DELETE FROM acp_sessions WHERE id = ?"))
337            .bind(session_id)
338            .execute(&self.pool)
339            .await?;
340        Ok(result.rows_affected() > 0)
341    }
342}
343
344#[derive(sqlx::FromRow)]
345struct SessionRow {
346    id: String,
347    title: Option<String>,
348    created_at: String,
349    updated_at: String,
350    conversation_id: Option<i64>,
351    last_seq: i64,
352    event_count: i64,
353    forked_from: Option<String>,
354    forked_at_seq: Option<i64>,
355    status: String,
356    last_condensed_seq: i64,
357}
358
359impl TryFrom<SessionRow> for SessionMetadata {
360    type Error = SessionError;
361
362    fn try_from(row: SessionRow) -> Result<Self, Self::Error> {
363        Ok(Self {
364            session_id: row.id,
365            title: row.title,
366            created_at: row.created_at,
367            updated_at: row.updated_at,
368            conversation_id: row.conversation_id,
369            last_seq: u64::try_from(row.last_seq).unwrap_or(0),
370            event_count: u64::try_from(row.event_count).unwrap_or(0),
371            forked_from: row.forked_from,
372            forked_at_seq: row.forked_at_seq.map(|v| u64::try_from(v).unwrap_or(0)),
373            status: row.status.parse()?,
374            last_condensed_seq: u64::try_from(row.last_condensed_seq).unwrap_or(0),
375        })
376    }
377}
378
379#[cfg(test)]
380mod tests {
381    use super::*;
382
383    async fn make_pool() -> DbPool {
384        let config = zeph_db::DbConfig {
385            url: ":memory:".to_owned(),
386            ..Default::default()
387        };
388        let pool = config
389            .connect()
390            .await
391            .expect("connect in-memory sqlite pool");
392        zeph_db::run_migrations(&pool)
393            .await
394            .expect("run migrations");
395        pool
396    }
397
398    #[tokio::test]
399    async fn test_migration_106_idempotent() {
400        let pool = make_pool().await;
401        zeph_db::run_migrations(&pool)
402            .await
403            .expect("second run is a no-op");
404    }
405
406    #[tokio::test]
407    async fn create_and_get_defaults() {
408        let store = SessionStore::new(make_pool().await);
409        store.create("s1").await.unwrap();
410        let meta = store.get("s1").await.unwrap().expect("row exists");
411        assert_eq!(meta.session_id, "s1");
412        assert_eq!(meta.last_seq, 0);
413        assert_eq!(meta.event_count, 0);
414        assert_eq!(meta.status, SessionStatus::Active);
415        assert!(meta.forked_from.is_none());
416    }
417
418    #[tokio::test]
419    async fn update_seq_persists() {
420        let store = SessionStore::new(make_pool().await);
421        store.create("s1").await.unwrap();
422        store.update_seq("s1", 41, 20).await.unwrap();
423        let meta = store.get("s1").await.unwrap().unwrap();
424        assert_eq!(meta.last_seq, 41);
425        assert_eq!(meta.event_count, 20);
426    }
427
428    #[tokio::test]
429    async fn set_status_persists() {
430        let store = SessionStore::new(make_pool().await);
431        store.create("s1").await.unwrap();
432        store.set_status("s1", SessionStatus::Idle).await.unwrap();
433        let meta = store.get("s1").await.unwrap().unwrap();
434        assert_eq!(meta.status, SessionStatus::Idle);
435    }
436
437    #[tokio::test]
438    async fn record_fork_sets_provenance() {
439        let store = SessionStore::new(make_pool().await);
440        store.create("parent").await.unwrap();
441        store.record_fork("child", "parent", 12).await.unwrap();
442        let meta = store.get("child").await.unwrap().unwrap();
443        assert_eq!(meta.forked_from.as_deref(), Some("parent"));
444        assert_eq!(meta.forked_at_seq, Some(12));
445    }
446
447    #[tokio::test]
448    async fn list_filters_by_status() {
449        let store = SessionStore::new(make_pool().await);
450        store.create("s1").await.unwrap();
451        store.create("s2").await.unwrap();
452        store
453            .set_status("s2", SessionStatus::Archived)
454            .await
455            .unwrap();
456
457        let active = store
458            .list(&SessionFilter {
459                status: Some(SessionStatus::Active),
460                limit: 0,
461            })
462            .await
463            .unwrap();
464        assert_eq!(active.len(), 1);
465        assert_eq!(active[0].session_id, "s1");
466
467        let all = store.list(&SessionFilter::default()).await.unwrap();
468        assert_eq!(all.len(), 2);
469    }
470
471    #[tokio::test]
472    async fn delete_removes_row() {
473        let store = SessionStore::new(make_pool().await);
474        store.create("s1").await.unwrap();
475        assert!(store.delete("s1").await.unwrap());
476        assert!(store.get("s1").await.unwrap().is_none());
477        assert!(!store.delete("s1").await.unwrap());
478    }
479
480    #[tokio::test]
481    async fn get_missing_returns_none() {
482        let store = SessionStore::new(make_pool().await);
483        assert!(store.get("no-such").await.unwrap().is_none());
484    }
485
486    #[tokio::test]
487    async fn link_conversation_and_lookup_round_trips() {
488        let pool = make_pool().await;
489        let store = SessionStore::new(pool.clone());
490        store.create("s1").await.unwrap();
491
492        // `conversation_id` carries an FK to `conversations(id)` (migration 001); insert a row
493        // directly since creating conversations is zeph-memory's domain, out of scope here.
494        let (cid,): (i64,) =
495            zeph_db::query_as("INSERT INTO conversations DEFAULT VALUES RETURNING id")
496                .fetch_one(&pool)
497                .await
498                .unwrap();
499
500        store.link_conversation("s1", cid).await.unwrap();
501
502        let meta = store.get("s1").await.unwrap().unwrap();
503        assert_eq!(meta.conversation_id, Some(cid));
504
505        let found = store.get_by_conversation_id(cid).await.unwrap().unwrap();
506        assert_eq!(found.session_id, "s1");
507    }
508
509    #[tokio::test]
510    async fn get_by_conversation_id_returns_none_when_unlinked() {
511        let store = SessionStore::new(make_pool().await);
512        store.create("s1").await.unwrap();
513        assert!(store.get_by_conversation_id(99).await.unwrap().is_none());
514    }
515}
zeph_session/store.rs

zeph_session/
store.rs
