zeph_memory/store/
acp_sessions.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4use crate::error::MemoryError;
5use crate::store::SqliteStore;
6use crate::types::ConversationId;
7use zeph_db::ActiveDialect;
8#[allow(unused_imports)]
9use zeph_db::sql;
10
11pub struct AcpSessionEvent {
12    pub event_type: String,
13    pub payload: String,
14    pub created_at: String,
15}
16
17pub struct AcpSessionInfo {
18    pub id: String,
19    pub title: Option<String>,
20    pub created_at: String,
21    pub updated_at: String,
22    pub message_count: i64,
23}
24
25/// Snapshot of per-session config fields (#5373), taken on graceful `session/close` so a later
26/// `session/resume` or `session/fork` can inherit these values instead of resetting to
27/// configured defaults.
28pub struct AcpSessionConfigSnapshot {
29    pub current_model: String,
30    pub temperature_preset: String,
31    pub thinking_enabled: bool,
32    pub auto_approve_level: String,
33}
34
35impl SqliteStore {
36    /// Create a new ACP session record.
37    ///
38    /// `owner` stamps `owner_key` (#5868): the authenticated ACP client identity that may
39    /// list/load this session. `None` leaves it unowned — used by non-ACP channels
40    /// (CLI/TUI/Telegram via `zeph_session::SessionStore::create`, which does not call this
41    /// method at all and so is unaffected either way; `None` here is for ACP callers that
42    /// intentionally create an unowned row, e.g. none today, but kept for forward compat).
43    ///
44    /// # Errors
45    ///
46    /// Returns an error if the database write fails.
47    pub async fn create_acp_session(
48        &self,
49        session_id: &str,
50        owner: Option<&str>,
51    ) -> Result<(), MemoryError> {
52        let sql = zeph_db::rewrite_placeholders(&format!(
53            "{} INTO acp_sessions (id, owner_key) VALUES (?, ?){}",
54            <ActiveDialect as zeph_db::dialect::Dialect>::INSERT_IGNORE,
55            <ActiveDialect as zeph_db::dialect::Dialect>::CONFLICT_NOTHING,
56        ));
57        zeph_db::query(sqlx::AssertSqlSafe(sql))
58            .bind(session_id)
59            .bind(owner)
60            .execute(&self.pool)
61            .await?;
62        Ok(())
63    }
64
65    /// Persist a single ACP session event.
66    ///
67    /// # Errors
68    ///
69    /// Returns an error if the database write fails.
70    pub async fn save_acp_event(
71        &self,
72        session_id: &str,
73        event_type: &str,
74        payload: &str,
75    ) -> Result<(), MemoryError> {
76        zeph_db::query(sql!(
77            "INSERT INTO acp_session_events (session_id, event_type, payload) VALUES (?, ?, ?)"
78        ))
79        .bind(session_id)
80        .bind(event_type)
81        .bind(payload)
82        .execute(&self.pool)
83        .await?;
84        Ok(())
85    }
86
87    /// Load all events for an ACP session in insertion order.
88    ///
89    /// # Errors
90    ///
91    /// Returns an error if the database query fails.
92    pub async fn load_acp_events(
93        &self,
94        session_id: &str,
95    ) -> Result<Vec<AcpSessionEvent>, MemoryError> {
96        let rows = zeph_db::query_as::<_, (String, String, String)>(
97            sql!("SELECT event_type, payload, created_at FROM acp_session_events WHERE session_id = ? ORDER BY id"),
98        )
99        .bind(session_id)
100        .fetch_all(&self.pool)
101        .await?;
102
103        Ok(rows
104            .into_iter()
105            .map(|(event_type, payload, created_at)| AcpSessionEvent {
106                event_type,
107                payload,
108                created_at,
109            })
110            .collect())
111    }
112
113    /// Delete an ACP session only if it exists; returns `true` when a row was deleted.
114    ///
115    /// Eliminates the separate exists-check + delete TOCTOU race by relying on a
116    /// single DELETE statement and inspecting affected rows.
117    ///
118    /// # Errors
119    ///
120    /// Returns an error if the database write fails.
121    pub async fn delete_acp_session_checked(&self, session_id: &str) -> Result<bool, MemoryError> {
122        let result = zeph_db::query(sql!("DELETE FROM acp_sessions WHERE id = ?"))
123            .bind(session_id)
124            .execute(&self.pool)
125            .await?;
126        Ok(result.rows_affected() > 0)
127    }
128
129    /// List ACP sessions ordered by last activity descending.
130    ///
131    /// Includes title, `updated_at`, and message count per session.
132    /// Pass `limit = 0` for unlimited results.
133    ///
134    /// # Errors
135    ///
136    /// Returns an error if the database query fails.
137    pub async fn list_acp_sessions(
138        &self,
139        limit: usize,
140    ) -> Result<Vec<AcpSessionInfo>, MemoryError> {
141        // spec-068 §12.3 / D-2: `acp_sessions.event_count` (migration 106, kept current by
142        // `SessionStore::update_seq` on every turn flush per INV-SP-1) replaces the subquery
143        // against `acp_session_events`, which the P1 write cutover leaves permanently empty for
144        // post-cutover sessions.
145        // `created_at`/`updated_at` are `TIMESTAMPTZ` on Postgres (`TEXT` on SQLite); project
146        // both through `Dialect::select_as_text` so they decode into the `String` fields below,
147        // mirroring `agent_sessions.rs::list_agent_sessions`'s fix for the same mismatch.
148        let created_at_sel =
149            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
150        let updated_at_sel =
151            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("updated_at");
152        let (limit_clause, limit_bind) = zeph_db::limit_clause(limit as u64);
153        let raw = format!(
154            "SELECT s.id, s.title, s.{created_at_sel}, s.{updated_at_sel}, \
155             s.event_count AS message_count \
156             FROM acp_sessions s \
157             ORDER BY s.updated_at DESC{limit_clause}"
158        );
159        let query_sql = zeph_db::rewrite_placeholders(&raw);
160        let mut query = zeph_db::query_as::<_, (String, Option<String>, String, String, i64)>(
161            sqlx::AssertSqlSafe(query_sql),
162        );
163        if let Some(lim) = limit_bind {
164            query = query.bind(lim);
165        }
166        let rows = query.fetch_all(&self.pool).await?;
167
168        Ok(rows
169            .into_iter()
170            .map(
171                |(id, title, created_at, updated_at, message_count)| AcpSessionInfo {
172                    id,
173                    title,
174                    created_at,
175                    updated_at,
176                    message_count,
177                },
178            )
179            .collect())
180    }
181
182    /// Fetch metadata for a single ACP session.
183    ///
184    /// Returns `None` if the session does not exist.
185    ///
186    /// # Errors
187    ///
188    /// Returns an error if the database query fails.
189    pub async fn get_acp_session_info(
190        &self,
191        session_id: &str,
192    ) -> Result<Option<AcpSessionInfo>, MemoryError> {
193        // spec-068 §12.3 / D-2: see `list_acp_sessions` — `event_count` replaces the emptied
194        // `acp_session_events` subquery.
195        // `created_at`/`updated_at` are `TIMESTAMPTZ` on Postgres — see `list_acp_sessions`.
196        let created_at_sel =
197            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
198        let updated_at_sel =
199            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("updated_at");
200        let raw = format!(
201            "SELECT s.id, s.title, s.{created_at_sel}, s.{updated_at_sel}, \
202             s.event_count AS message_count \
203             FROM acp_sessions s \
204             WHERE s.id = ?"
205        );
206        let query_sql = zeph_db::rewrite_placeholders(&raw);
207        let row = zeph_db::query_as::<_, (String, Option<String>, String, String, i64)>(
208            sqlx::AssertSqlSafe(query_sql),
209        )
210        .bind(session_id)
211        .fetch_optional(&self.pool)
212        .await?;
213
214        Ok(row.map(
215            |(id, title, created_at, updated_at, message_count)| AcpSessionInfo {
216                id,
217                title,
218                created_at,
219                updated_at,
220                message_count,
221            },
222        ))
223    }
224
225    /// Insert multiple events for a session inside a single transaction.
226    ///
227    /// Atomically writes all events or none. More efficient than individual inserts
228    /// for bulk import use cases.
229    ///
230    /// # Errors
231    ///
232    /// Returns an error if the transaction or any insert fails.
233    pub async fn import_acp_events(
234        &self,
235        session_id: &str,
236        events: &[(&str, &str)],
237    ) -> Result<(), MemoryError> {
238        let mut tx = self.pool.begin().await?;
239        for (event_type, payload) in events {
240            zeph_db::query(sql!(
241                "INSERT INTO acp_session_events (session_id, event_type, payload) VALUES (?, ?, ?)"
242            ))
243            .bind(session_id)
244            .bind(event_type)
245            .bind(payload)
246            .execute(&mut *tx)
247            .await?;
248        }
249        tx.commit().await?;
250        Ok(())
251    }
252
253    /// Update the title of an ACP session.
254    ///
255    /// # Errors
256    ///
257    /// Returns an error if the database write fails.
258    pub async fn update_session_title(
259        &self,
260        session_id: &str,
261        title: &str,
262    ) -> Result<(), MemoryError> {
263        zeph_db::query(sql!("UPDATE acp_sessions SET title = ? WHERE id = ?"))
264            .bind(title)
265            .bind(session_id)
266            .execute(&self.pool)
267            .await?;
268        Ok(())
269    }
270
271    /// Update the title of an ACP session; returns `true` when the row was found and updated.
272    ///
273    /// Eliminates the separate exists-check + update TOCTOU race by relying on a
274    /// single UPDATE statement and inspecting affected rows.
275    ///
276    /// # Errors
277    ///
278    /// Returns an error if the database write fails.
279    pub async fn update_session_title_checked(
280        &self,
281        session_id: &str,
282        title: &str,
283    ) -> Result<bool, MemoryError> {
284        let result = zeph_db::query(sql!("UPDATE acp_sessions SET title = ? WHERE id = ?"))
285            .bind(title)
286            .bind(session_id)
287            .execute(&self.pool)
288            .await?;
289        Ok(result.rows_affected() > 0)
290    }
291
292    /// Persist a snapshot of the session's current config fields (#5373).
293    ///
294    /// Called on graceful `session/close` so a later `session/resume` or `session/fork` of a
295    /// session no longer resident in memory can inherit these values.
296    ///
297    /// # Errors
298    ///
299    /// Returns an error if the database write fails.
300    pub async fn save_session_config(
301        &self,
302        session_id: &str,
303        snapshot: &AcpSessionConfigSnapshot,
304    ) -> Result<(), MemoryError> {
305        zeph_db::query(sql!(
306            "UPDATE acp_sessions SET current_model = ?, temperature_preset = ?, \
307             thinking_enabled = ?, auto_approve_level = ? WHERE id = ?"
308        ))
309        .bind(&snapshot.current_model)
310        .bind(&snapshot.temperature_preset)
311        .bind(snapshot.thinking_enabled)
312        .bind(&snapshot.auto_approve_level)
313        .bind(session_id)
314        .execute(&self.pool)
315        .await?;
316        Ok(())
317    }
318
319    /// Load the persisted config snapshot for a session, if one was saved (#5373).
320    ///
321    /// Returns `None` when the session has no snapshot yet — either it was never closed
322    /// gracefully, or it predates the config-snapshot migration. Callers should fall back to
323    /// configured defaults in that case.
324    ///
325    /// # Errors
326    ///
327    /// Returns an error if the database query fails.
328    pub async fn get_session_config(
329        &self,
330        session_id: &str,
331    ) -> Result<Option<AcpSessionConfigSnapshot>, MemoryError> {
332        let row = zeph_db::query_as::<
333            _,
334            (Option<String>, Option<String>, Option<bool>, Option<String>),
335        >(sql!(
336            "SELECT current_model, temperature_preset, thinking_enabled, auto_approve_level \
337                 FROM acp_sessions WHERE id = ?"
338        ))
339        .bind(session_id)
340        .fetch_optional(&self.pool)
341        .await?;
342
343        Ok(row.and_then(
344            |(current_model, temperature_preset, thinking_enabled, auto_approve_level)| {
345                Some(AcpSessionConfigSnapshot {
346                    current_model: current_model?,
347                    temperature_preset: temperature_preset?,
348                    thinking_enabled: thinking_enabled?,
349                    auto_approve_level: auto_approve_level?,
350                })
351            },
352        ))
353    }
354
355    /// Check whether an ACP session record exists.
356    ///
357    /// # Errors
358    ///
359    /// Returns an error if the database query fails.
360    pub async fn acp_session_exists(&self, session_id: &str) -> Result<bool, MemoryError> {
361        let count: i64 =
362            zeph_db::query_scalar(sql!("SELECT COUNT(*) FROM acp_sessions WHERE id = ?"))
363                .bind(session_id)
364                .fetch_one(&self.pool)
365                .await?;
366        Ok(count > 0)
367    }
368
369    /// List ACP sessions owned by `owner`, ordered by last activity descending (#5868).
370    ///
371    /// Unlike [`Self::list_acp_sessions`], this is strictly scoped: rows owned by another
372    /// `owner_key` or left `NULL` (legacy/non-ACP rows) are never returned. Pass `limit = 0`
373    /// for unlimited results.
374    ///
375    /// # Errors
376    ///
377    /// Returns an error if the database query fails.
378    pub async fn list_acp_sessions_for_owner(
379        &self,
380        limit: usize,
381        owner: &str,
382    ) -> Result<Vec<AcpSessionInfo>, MemoryError> {
383        let created_at_sel =
384            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
385        let updated_at_sel =
386            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("updated_at");
387        let (limit_clause, limit_bind) = zeph_db::limit_clause(limit as u64);
388        let raw = format!(
389            "SELECT s.id, s.title, s.{created_at_sel}, s.{updated_at_sel}, \
390             s.event_count AS message_count \
391             FROM acp_sessions s \
392             WHERE s.owner_key = ? \
393             ORDER BY s.updated_at DESC{limit_clause}"
394        );
395        let query_sql = zeph_db::rewrite_placeholders(&raw);
396        let mut query = zeph_db::query_as::<_, (String, Option<String>, String, String, i64)>(
397            sqlx::AssertSqlSafe(query_sql),
398        )
399        .bind(owner);
400        if let Some(lim) = limit_bind {
401            query = query.bind(lim);
402        }
403        let rows = query.fetch_all(&self.pool).await?;
404
405        Ok(rows
406            .into_iter()
407            .map(
408                |(id, title, created_at, updated_at, message_count)| AcpSessionInfo {
409                    id,
410                    title,
411                    created_at,
412                    updated_at,
413                    message_count,
414                },
415            )
416            .collect())
417    }
418
419    /// Fetch metadata for a session, scoped to `owner` (#5868).
420    ///
421    /// Returns `None` both when the session does not exist and when it is owned by a
422    /// different `owner_key` — the two cases are indistinguishable to the caller by design,
423    /// to avoid leaking existence of another owner's session (mirrors
424    /// [`Self::claim_acp_session_for_owner`]'s uniform not-found semantics). A `NULL`
425    /// (legacy/non-ACP) `owner_key` is treated as accessible.
426    ///
427    /// # Errors
428    ///
429    /// Returns an error if the database query fails.
430    pub async fn get_acp_session_info_for_owner(
431        &self,
432        session_id: &str,
433        owner: &str,
434    ) -> Result<Option<AcpSessionInfo>, MemoryError> {
435        let created_at_sel =
436            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
437        let updated_at_sel =
438            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("updated_at");
439        let raw = format!(
440            "SELECT s.id, s.title, s.{created_at_sel}, s.{updated_at_sel}, \
441             s.event_count AS message_count \
442             FROM acp_sessions s \
443             WHERE s.id = ? AND (s.owner_key = ? OR s.owner_key IS NULL)"
444        );
445        let query_sql = zeph_db::rewrite_placeholders(&raw);
446        let row = zeph_db::query_as::<_, (String, Option<String>, String, String, i64)>(
447            sqlx::AssertSqlSafe(query_sql),
448        )
449        .bind(session_id)
450        .bind(owner)
451        .fetch_optional(&self.pool)
452        .await?;
453
454        Ok(row.map(
455            |(id, title, created_at, updated_at, message_count)| AcpSessionInfo {
456                id,
457                title,
458                created_at,
459                updated_at,
460                message_count,
461            },
462        ))
463    }
464
465    /// Non-mutating accessibility gate: `true` iff `session_id` exists and is owned by
466    /// `owner` or is unowned (`NULL`, legacy/non-ACP row) (#5868).
467    ///
468    /// Use for read-only access gates that must NOT silently claim a legacy row (e.g. the
469    /// HTTP `session_messages_handler`). For load/resume/fork, prefer
470    /// [`Self::claim_acp_session_for_owner`], which additionally claims unowned rows
471    /// atomically.
472    ///
473    /// # Errors
474    ///
475    /// Returns an error if the database query fails.
476    pub async fn acp_session_accessible_for_owner(
477        &self,
478        session_id: &str,
479        owner: &str,
480    ) -> Result<bool, MemoryError> {
481        let count: i64 = zeph_db::query_scalar(sql!(
482            "SELECT COUNT(*) FROM acp_sessions WHERE id = ? AND (owner_key = ? OR owner_key IS NULL)"
483        ))
484        .bind(session_id)
485        .bind(owner)
486        .fetch_one(&self.pool)
487        .await?;
488        Ok(count > 0)
489    }
490
491    /// Atomically grant `owner` access to `session_id`, claiming it if currently unowned
492    /// (`NULL`, legacy/non-ACP row) (#5868).
493    ///
494    /// A single `UPDATE ... RETURNING` statement — no read-then-claim race window. Returns
495    /// `true` iff the row exists AND is accessible: either it was `NULL` and is now claimed
496    /// for `owner`, or it was already owned by `owner` (a redundant, harmless no-op write).
497    /// Returns `false` uniformly for "session does not exist" and "owned by a different
498    /// owner" — callers must map both to the same not-found response so a foreign
499    /// `owner_key` can never be distinguished from a missing session.
500    ///
501    /// Use for load/resume/fork existence checks. For read-only gates that must not claim,
502    /// use [`Self::acp_session_accessible_for_owner`] instead.
503    ///
504    /// # Errors
505    ///
506    /// Returns an error if the database write fails.
507    pub async fn claim_acp_session_for_owner(
508        &self,
509        session_id: &str,
510        owner: &str,
511    ) -> Result<bool, MemoryError> {
512        let row: Option<(String,)> = zeph_db::query_as(sql!(
513            "UPDATE acp_sessions SET owner_key = ? \
514             WHERE id = ? AND (owner_key IS NULL OR owner_key = ?) \
515             RETURNING id"
516        ))
517        .bind(owner)
518        .bind(session_id)
519        .bind(owner)
520        .fetch_optional(&self.pool)
521        .await?;
522        Ok(row.is_some())
523    }
524
525    /// Delete an ACP session scoped to `owner`, returning `true` iff a row was deleted
526    /// (#5868).
527    ///
528    /// Matches [`Self::delete_acp_session_checked`]'s TOCTOU-free single-statement shape,
529    /// additionally restricted to rows owned by `owner` or unowned (`NULL`).
530    ///
531    /// # Errors
532    ///
533    /// Returns an error if the database write fails.
534    pub async fn delete_acp_session_for_owner(
535        &self,
536        session_id: &str,
537        owner: &str,
538    ) -> Result<bool, MemoryError> {
539        let result = zeph_db::query(sql!(
540            "DELETE FROM acp_sessions WHERE id = ? AND (owner_key = ? OR owner_key IS NULL)"
541        ))
542        .bind(session_id)
543        .bind(owner)
544        .execute(&self.pool)
545        .await?;
546        Ok(result.rows_affected() > 0)
547    }
548
549    /// Update the title of a session scoped to `owner`, returning `true` iff a row was
550    /// updated (#5868).
551    ///
552    /// Matches [`Self::update_session_title_checked`]'s TOCTOU-free single-statement shape,
553    /// additionally restricted to rows owned by `owner` or unowned (`NULL`).
554    ///
555    /// # Errors
556    ///
557    /// Returns an error if the database write fails.
558    pub async fn update_session_title_for_owner(
559        &self,
560        session_id: &str,
561        title: &str,
562        owner: &str,
563    ) -> Result<bool, MemoryError> {
564        let result = zeph_db::query(sql!(
565            "UPDATE acp_sessions SET title = ? \
566             WHERE id = ? AND (owner_key = ? OR owner_key IS NULL)"
567        ))
568        .bind(title)
569        .bind(session_id)
570        .bind(owner)
571        .execute(&self.pool)
572        .await?;
573        Ok(result.rows_affected() > 0)
574    }
575
576    /// Create a new ACP session record with an associated conversation.
577    ///
578    /// `owner` stamps `owner_key` (#5868) — see [`Self::create_acp_session`].
579    ///
580    /// # Errors
581    ///
582    /// Returns an error if the database write fails.
583    pub async fn create_acp_session_with_conversation(
584        &self,
585        session_id: &str,
586        conversation_id: ConversationId,
587        owner: Option<&str>,
588    ) -> Result<(), MemoryError> {
589        let sql = zeph_db::rewrite_placeholders(&format!(
590            "{} INTO acp_sessions (id, conversation_id, owner_key) VALUES (?, ?, ?){}",
591            <ActiveDialect as zeph_db::dialect::Dialect>::INSERT_IGNORE,
592            <ActiveDialect as zeph_db::dialect::Dialect>::CONFLICT_NOTHING,
593        ));
594        zeph_db::query(sqlx::AssertSqlSafe(sql))
595            .bind(session_id)
596            .bind(conversation_id)
597            .bind(owner)
598            .execute(&self.pool)
599            .await?;
600        Ok(())
601    }
602
603    /// Get the conversation ID associated with an ACP session.
604    ///
605    /// Returns `None` if the session has no conversation mapping (legacy session).
606    ///
607    /// # Errors
608    ///
609    /// Returns an error if the database query fails.
610    pub async fn get_acp_session_conversation_id(
611        &self,
612        session_id: &str,
613    ) -> Result<Option<ConversationId>, MemoryError> {
614        let row: Option<(Option<ConversationId>,)> = zeph_db::query_as(sql!(
615            "SELECT conversation_id FROM acp_sessions WHERE id = ?"
616        ))
617        .bind(session_id)
618        .fetch_optional(&self.pool)
619        .await?;
620        Ok(row.and_then(|(cid,)| cid))
621    }
622
623    /// Update the conversation mapping for an ACP session.
624    ///
625    /// # Errors
626    ///
627    /// Returns an error if the database write fails.
628    pub async fn set_acp_session_conversation_id(
629        &self,
630        session_id: &str,
631        conversation_id: ConversationId,
632    ) -> Result<(), MemoryError> {
633        zeph_db::query(sql!(
634            "UPDATE acp_sessions SET conversation_id = ? WHERE id = ?"
635        ))
636        .bind(conversation_id)
637        .bind(session_id)
638        .execute(&self.pool)
639        .await?;
640        Ok(())
641    }
642
643    /// Copy all messages from one conversation to another, preserving order.
644    ///
645    /// Summaries are intentionally NOT copied: their `first_message_id`/`last_message_id`
646    /// reference message IDs from the source conversation which differ from the new IDs
647    /// assigned to the copied messages, making the compaction cursor incorrect. The forked
648    /// session inherits the full message history and builds its own compaction state from
649    /// scratch. Other per-conversation state also excluded: embeddings (re-indexed on demand),
650    /// deferred tool summaries (treated as fresh context budget).
651    ///
652    /// # Errors
653    ///
654    /// Returns an error if the database write fails.
655    pub async fn copy_conversation(
656        &self,
657        source: ConversationId,
658        target: ConversationId,
659    ) -> Result<(), MemoryError> {
660        let mut tx = self.pool.begin().await?;
661
662        // Copy messages in order. Only columns present across all migrations are included;
663        // per-message auto-fields (id, created_at, last_accessed, access_count, qdrant_cleaned)
664        // are excluded so they are generated fresh for the target conversation.
665        zeph_db::query(sql!(
666            "INSERT INTO messages \
667                (conversation_id, role, content, parts, visibility, compacted_at, deleted_at) \
668             SELECT ?, role, content, parts, visibility, compacted_at, deleted_at \
669             FROM messages WHERE conversation_id = ? ORDER BY id"
670        ))
671        .bind(target)
672        .bind(source)
673        .execute(&mut *tx)
674        .await?;
675
676        // Summaries are NOT copied — their message ID boundaries reference the source
677        // conversation and would corrupt the compaction cursor in the forked session.
678        // The forked session builds compaction state from its own messages.
679
680        tx.commit().await?;
681        Ok(())
682    }
683}
684
685#[cfg(test)]
686mod tests {
687    use super::*;
688
689    async fn make_store() -> SqliteStore {
690        SqliteStore::new(":memory:")
691            .await
692            .expect("SqliteStore::new")
693    }
694
695    /// Bump `acp_sessions.event_count` and `updated_at`, mirroring the `UPDATE`
696    /// `zeph_session::SessionStore::update_seq` issues in production (spec-068 §12.3 / D-2).
697    /// `list_acp_sessions`/`get_acp_session_info` read `event_count`, not the legacy
698    /// `acp_session_events` table that `save_acp_event` populates — tests asserting on
699    /// `message_count` (or activity ordering, which depends on `updated_at`) must drive both
700    /// through this column directly rather than the retired write path.
701    async fn bump_event_count(store: &SqliteStore, session_id: &str, event_count: i64) {
702        let stmt = zeph_db::rewrite_placeholders(&format!(
703            "UPDATE acp_sessions SET event_count = ?, updated_at = {} WHERE id = ?",
704            <ActiveDialect as zeph_db::dialect::Dialect>::NOW,
705        ));
706        zeph_db::query(sqlx::AssertSqlSafe(stmt))
707            .bind(event_count)
708            .bind(session_id)
709            .execute(store.pool())
710            .await
711            .unwrap();
712    }
713
714    #[tokio::test]
715    async fn create_and_exists() {
716        let store = make_store().await;
717        store.create_acp_session("sess-1", None).await.unwrap();
718        assert!(store.acp_session_exists("sess-1").await.unwrap());
719        assert!(!store.acp_session_exists("sess-2").await.unwrap());
720    }
721
722    #[tokio::test]
723    async fn session_config_round_trips() {
724        let store = make_store().await;
725        store.create_acp_session("sess-1", None).await.unwrap();
726        let snapshot = AcpSessionConfigSnapshot {
727            current_model: "claude:opus".to_owned(),
728            temperature_preset: "creative".to_owned(),
729            thinking_enabled: true,
730            auto_approve_level: "auto-edit".to_owned(),
731        };
732        store
733            .save_session_config("sess-1", &snapshot)
734            .await
735            .unwrap();
736
737        let loaded = store
738            .get_session_config("sess-1")
739            .await
740            .unwrap()
741            .expect("snapshot must be present after save");
742        assert_eq!(loaded.current_model, "claude:opus");
743        assert_eq!(loaded.temperature_preset, "creative");
744        assert!(loaded.thinking_enabled);
745        assert_eq!(loaded.auto_approve_level, "auto-edit");
746    }
747
748    #[tokio::test]
749    async fn session_config_missing_snapshot_returns_none() {
750        let store = make_store().await;
751        store.create_acp_session("sess-1", None).await.unwrap();
752        assert!(store.get_session_config("sess-1").await.unwrap().is_none());
753    }
754
755    #[tokio::test]
756    async fn session_config_unknown_session_returns_none() {
757        let store = make_store().await;
758        assert!(store.get_session_config("no-such").await.unwrap().is_none());
759    }
760
761    #[tokio::test]
762    async fn save_and_load_events() {
763        let store = make_store().await;
764        store.create_acp_session("sess-1", None).await.unwrap();
765        store
766            .save_acp_event("sess-1", "user_message", "hello")
767            .await
768            .unwrap();
769        store
770            .save_acp_event("sess-1", "agent_message", "world")
771            .await
772            .unwrap();
773
774        let events = store.load_acp_events("sess-1").await.unwrap();
775        assert_eq!(events.len(), 2);
776        assert_eq!(events[0].event_type, "user_message");
777        assert_eq!(events[0].payload, "hello");
778        assert_eq!(events[1].event_type, "agent_message");
779        assert_eq!(events[1].payload, "world");
780    }
781
782    #[tokio::test]
783    async fn delete_cascades_events() {
784        let store = make_store().await;
785        store.create_acp_session("sess-1", None).await.unwrap();
786        store
787            .save_acp_event("sess-1", "user_message", "hello")
788            .await
789            .unwrap();
790        store.delete_acp_session_checked("sess-1").await.unwrap();
791
792        assert!(!store.acp_session_exists("sess-1").await.unwrap());
793        let events = store.load_acp_events("sess-1").await.unwrap();
794        assert!(events.is_empty());
795    }
796
797    #[tokio::test]
798    async fn load_events_empty_for_unknown() {
799        let store = make_store().await;
800        let events = store.load_acp_events("no-such").await.unwrap();
801        assert!(events.is_empty());
802    }
803
804    #[tokio::test]
805    async fn list_sessions_includes_title_and_message_count() {
806        let store = make_store().await;
807        store.create_acp_session("sess-b", None).await.unwrap();
808
809        // Sleep so that sess-a's events land in a different second than sess-b's
810        // created_at, making the updated_at DESC ordering deterministic.
811        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
812
813        store.create_acp_session("sess-a", None).await.unwrap();
814        bump_event_count(&store, "sess-a", 2).await;
815        store
816            .update_session_title("sess-a", "My Chat")
817            .await
818            .unwrap();
819
820        let sessions = store.list_acp_sessions(100).await.unwrap();
821        // sess-a has events so updated_at is newer — should be first
822        assert_eq!(sessions[0].id, "sess-a");
823        assert_eq!(sessions[0].title.as_deref(), Some("My Chat"));
824        assert_eq!(sessions[0].message_count, 2);
825
826        // sess-b has no events
827        let b = sessions.iter().find(|s| s.id == "sess-b").unwrap();
828        assert!(b.title.is_none());
829        assert_eq!(b.message_count, 0);
830    }
831
832    #[tokio::test]
833    async fn list_sessions_respects_limit() {
834        let store = make_store().await;
835        for i in 0..5u8 {
836            store
837                .create_acp_session(&format!("sess-{i}"), None)
838                .await
839                .unwrap();
840        }
841        let sessions = store.list_acp_sessions(3).await.unwrap();
842        assert_eq!(sessions.len(), 3);
843    }
844
845    #[tokio::test]
846    async fn list_sessions_limit_one_boundary() {
847        let store = make_store().await;
848        for i in 0..3u8 {
849            store
850                .create_acp_session(&format!("sess-{i}"), None)
851                .await
852                .unwrap();
853        }
854        let sessions = store.list_acp_sessions(1).await.unwrap();
855        assert_eq!(sessions.len(), 1);
856    }
857
858    #[tokio::test]
859    async fn list_sessions_unlimited_when_zero() {
860        let store = make_store().await;
861        for i in 0..5u8 {
862            store
863                .create_acp_session(&format!("sess-{i}"), None)
864                .await
865                .unwrap();
866        }
867        let sessions = store.list_acp_sessions(0).await.unwrap();
868        assert_eq!(sessions.len(), 5);
869    }
870
871    #[tokio::test]
872    async fn get_acp_session_info_returns_none_for_missing() {
873        let store = make_store().await;
874        let info = store.get_acp_session_info("no-such").await.unwrap();
875        assert!(info.is_none());
876    }
877
878    #[tokio::test]
879    async fn get_acp_session_info_returns_data() {
880        let store = make_store().await;
881        store.create_acp_session("sess-x", None).await.unwrap();
882        bump_event_count(&store, "sess-x", 1).await;
883        store.update_session_title("sess-x", "Test").await.unwrap();
884
885        let info = store.get_acp_session_info("sess-x").await.unwrap().unwrap();
886        assert_eq!(info.id, "sess-x");
887        assert_eq!(info.title.as_deref(), Some("Test"));
888        assert_eq!(info.message_count, 1);
889    }
890
891    #[tokio::test]
892    async fn updated_at_trigger_fires_on_event_insert() {
893        let store = make_store().await;
894        store.create_acp_session("sess-t", None).await.unwrap();
895
896        let before = store
897            .get_acp_session_info("sess-t")
898            .await
899            .unwrap()
900            .unwrap()
901            .updated_at
902            .clone();
903
904        // Small sleep so datetime('now') differs
905        tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
906
907        store
908            .save_acp_event("sess-t", "user", "ping")
909            .await
910            .unwrap();
911
912        let after = store
913            .get_acp_session_info("sess-t")
914            .await
915            .unwrap()
916            .unwrap()
917            .updated_at;
918
919        assert!(
920            after > before,
921            "updated_at should increase after event insert: before={before} after={after}"
922        );
923    }
924
925    #[tokio::test]
926    async fn create_session_with_conversation_and_retrieve() {
927        let store = make_store().await;
928        let cid = store.create_conversation().await.unwrap();
929        store
930            .create_acp_session_with_conversation("sess-1", cid, None)
931            .await
932            .unwrap();
933        let retrieved = store
934            .get_acp_session_conversation_id("sess-1")
935            .await
936            .unwrap();
937        assert_eq!(retrieved, Some(cid));
938    }
939
940    #[tokio::test]
941    async fn get_conversation_id_returns_none_for_legacy_session() {
942        let store = make_store().await;
943        store.create_acp_session("legacy", None).await.unwrap();
944        let cid = store
945            .get_acp_session_conversation_id("legacy")
946            .await
947            .unwrap();
948        assert!(cid.is_none());
949    }
950
951    #[tokio::test]
952    async fn get_conversation_id_returns_none_for_missing_session() {
953        let store = make_store().await;
954        let cid = store
955            .get_acp_session_conversation_id("no-such")
956            .await
957            .unwrap();
958        assert!(cid.is_none());
959    }
960
961    #[tokio::test]
962    async fn set_conversation_id_updates_existing_session() {
963        let store = make_store().await;
964        store.create_acp_session("sess-2", None).await.unwrap();
965        let cid = store.create_conversation().await.unwrap();
966        store
967            .set_acp_session_conversation_id("sess-2", cid)
968            .await
969            .unwrap();
970        let retrieved = store
971            .get_acp_session_conversation_id("sess-2")
972            .await
973            .unwrap();
974        assert_eq!(retrieved, Some(cid));
975    }
976
977    #[tokio::test]
978    async fn copy_conversation_copies_messages_in_order() {
979        use zeph_llm::provider::Role;
980        let store = make_store().await;
981        let src = store.create_conversation().await.unwrap();
982        store.save_message(src, "user", "hello").await.unwrap();
983        store.save_message(src, "assistant", "world").await.unwrap();
984
985        let dst = store.create_conversation().await.unwrap();
986        store.copy_conversation(src, dst).await.unwrap();
987
988        let msgs = store.load_history(dst, 100).await.unwrap();
989        assert_eq!(msgs.len(), 2);
990        assert_eq!(msgs[0].role, Role::User);
991        assert_eq!(msgs[0].content, "hello");
992        assert_eq!(msgs[1].role, Role::Assistant);
993        assert_eq!(msgs[1].content, "world");
994    }
995
996    #[tokio::test]
997    async fn copy_conversation_empty_source_is_noop() {
998        let store = make_store().await;
999        let src = store.create_conversation().await.unwrap();
1000        let dst = store.create_conversation().await.unwrap();
1001        store.copy_conversation(src, dst).await.unwrap();
1002        let msgs = store.load_history(dst, 100).await.unwrap();
1003        assert!(msgs.is_empty());
1004    }
1005
1006    #[tokio::test]
1007    async fn copy_conversation_does_not_copy_summaries() {
1008        // Summaries are intentionally excluded because their first/last_message_id
1009        // boundaries would reference source message IDs, corrupting the compaction cursor.
1010        let store = make_store().await;
1011        let src = store.create_conversation().await.unwrap();
1012        store.save_message(src, "user", "hello").await.unwrap();
1013        // Insert a summary directly so we can verify it is not copied.
1014        zeph_db::query(
1015            sql!("INSERT INTO summaries (conversation_id, content, first_message_id, last_message_id, token_estimate) \
1016             VALUES (?, 'summary text', 1, 1, 10)"),
1017        )
1018        .bind(src)
1019        .execute(&store.pool)
1020        .await
1021        .unwrap();
1022
1023        let dst = store.create_conversation().await.unwrap();
1024        store.copy_conversation(src, dst).await.unwrap();
1025
1026        let count: i64 = zeph_db::query_scalar(sql!(
1027            "SELECT COUNT(*) FROM summaries WHERE conversation_id = ?"
1028        ))
1029        .bind(dst)
1030        .fetch_one(&store.pool)
1031        .await
1032        .unwrap();
1033        assert_eq!(
1034            count, 0,
1035            "summaries must not be copied to forked conversation"
1036        );
1037    }
1038
1039    #[tokio::test]
1040    async fn concurrent_sessions_get_distinct_conversation_ids() {
1041        let store = make_store().await;
1042        let cid1 = store.create_conversation().await.unwrap();
1043        let cid2 = store.create_conversation().await.unwrap();
1044        store
1045            .create_acp_session_with_conversation("sess-a", cid1, None)
1046            .await
1047            .unwrap();
1048        store
1049            .create_acp_session_with_conversation("sess-b", cid2, None)
1050            .await
1051            .unwrap();
1052
1053        let retrieved1 = store
1054            .get_acp_session_conversation_id("sess-a")
1055            .await
1056            .unwrap();
1057        let retrieved2 = store
1058            .get_acp_session_conversation_id("sess-b")
1059            .await
1060            .unwrap();
1061
1062        assert!(retrieved1.is_some());
1063        assert!(retrieved2.is_some());
1064        assert_ne!(
1065            retrieved1, retrieved2,
1066            "concurrent sessions must get distinct conversation_ids"
1067        );
1068    }
1069
1070    // ── Owner-scoped access (#5868) ────────────────────────────────────────────
1071
1072    /// Regression guard (#5868): the unscoped CLI-facing methods (`list_acp_sessions`,
1073    /// `acp_session_exists`, `delete_acp_session_checked`) must keep seeing every row
1074    /// regardless of `owner_key` — CLI/operator access is intentionally global, and a future
1075    /// change that accidentally scoped these instead of the `_for_owner` siblings would break
1076    /// `zeph sessions list`/`zeph sessions delete` silently.
1077    #[tokio::test]
1078    async fn cli_facing_unscoped_methods_see_all_owners_and_legacy_rows() {
1079        let store = make_store().await;
1080        store
1081            .create_acp_session("mine", Some("owner-a"))
1082            .await
1083            .unwrap();
1084        store
1085            .create_acp_session("theirs", Some("owner-b"))
1086            .await
1087            .unwrap();
1088        store.create_acp_session("legacy", None).await.unwrap();
1089
1090        let sessions = store.list_acp_sessions(0).await.unwrap();
1091        assert_eq!(sessions.len(), 3);
1092
1093        assert!(store.acp_session_exists("mine").await.unwrap());
1094        assert!(store.acp_session_exists("theirs").await.unwrap());
1095        assert!(store.acp_session_exists("legacy").await.unwrap());
1096
1097        assert!(store.delete_acp_session_checked("theirs").await.unwrap());
1098        assert!(!store.acp_session_exists("theirs").await.unwrap());
1099    }
1100
1101    #[tokio::test]
1102    async fn list_for_owner_excludes_other_owners_and_legacy_rows() {
1103        let store = make_store().await;
1104        store
1105            .create_acp_session("mine", Some("owner-a"))
1106            .await
1107            .unwrap();
1108        store
1109            .create_acp_session("theirs", Some("owner-b"))
1110            .await
1111            .unwrap();
1112        store.create_acp_session("legacy", None).await.unwrap();
1113
1114        let sessions = store
1115            .list_acp_sessions_for_owner(0, "owner-a")
1116            .await
1117            .unwrap();
1118        assert_eq!(sessions.len(), 1);
1119        assert_eq!(sessions[0].id, "mine");
1120    }
1121
1122    #[tokio::test]
1123    async fn get_info_for_owner_returns_none_for_foreign_owner() {
1124        let store = make_store().await;
1125        store
1126            .create_acp_session("theirs", Some("owner-b"))
1127            .await
1128            .unwrap();
1129        let info = store
1130            .get_acp_session_info_for_owner("theirs", "owner-a")
1131            .await
1132            .unwrap();
1133        assert!(info.is_none());
1134    }
1135
1136    #[tokio::test]
1137    async fn get_info_for_owner_accessible_for_legacy_null_row() {
1138        let store = make_store().await;
1139        store.create_acp_session("legacy", None).await.unwrap();
1140        let info = store
1141            .get_acp_session_info_for_owner("legacy", "owner-a")
1142            .await
1143            .unwrap();
1144        assert!(info.is_some());
1145    }
1146
1147    #[tokio::test]
1148    async fn accessible_for_owner_true_for_own_and_legacy_false_for_foreign() {
1149        let store = make_store().await;
1150        store
1151            .create_acp_session("mine", Some("owner-a"))
1152            .await
1153            .unwrap();
1154        store.create_acp_session("legacy", None).await.unwrap();
1155        store
1156            .create_acp_session("theirs", Some("owner-b"))
1157            .await
1158            .unwrap();
1159
1160        assert!(
1161            store
1162                .acp_session_accessible_for_owner("mine", "owner-a")
1163                .await
1164                .unwrap()
1165        );
1166        assert!(
1167            store
1168                .acp_session_accessible_for_owner("legacy", "owner-a")
1169                .await
1170                .unwrap()
1171        );
1172        assert!(
1173            !store
1174                .acp_session_accessible_for_owner("theirs", "owner-a")
1175                .await
1176                .unwrap()
1177        );
1178        assert!(
1179            !store
1180                .acp_session_accessible_for_owner("no-such", "owner-a")
1181                .await
1182                .unwrap()
1183        );
1184    }
1185
1186    #[tokio::test]
1187    async fn claim_for_owner_claims_legacy_null_row() {
1188        let store = make_store().await;
1189        store.create_acp_session("legacy", None).await.unwrap();
1190
1191        assert!(
1192            store
1193                .claim_acp_session_for_owner("legacy", "owner-a")
1194                .await
1195                .unwrap()
1196        );
1197
1198        // Now owned by owner-a: a foreign owner can no longer claim or list it.
1199        assert!(
1200            !store
1201                .claim_acp_session_for_owner("legacy", "owner-b")
1202                .await
1203                .unwrap()
1204        );
1205        let sessions = store
1206            .list_acp_sessions_for_owner(0, "owner-a")
1207            .await
1208            .unwrap();
1209        assert_eq!(sessions.len(), 1);
1210    }
1211
1212    #[tokio::test]
1213    async fn claim_for_owner_is_idempotent_for_the_same_owner() {
1214        let store = make_store().await;
1215        store
1216            .create_acp_session("mine", Some("owner-a"))
1217            .await
1218            .unwrap();
1219
1220        assert!(
1221            store
1222                .claim_acp_session_for_owner("mine", "owner-a")
1223                .await
1224                .unwrap()
1225        );
1226    }
1227
1228    #[tokio::test]
1229    async fn claim_for_owner_returns_false_for_missing_session() {
1230        let store = make_store().await;
1231        assert!(
1232            !store
1233                .claim_acp_session_for_owner("no-such", "owner-a")
1234                .await
1235                .unwrap()
1236        );
1237    }
1238
1239    #[tokio::test]
1240    async fn delete_for_owner_refuses_foreign_owner() {
1241        let store = make_store().await;
1242        store
1243            .create_acp_session("theirs", Some("owner-b"))
1244            .await
1245            .unwrap();
1246
1247        assert!(
1248            !store
1249                .delete_acp_session_for_owner("theirs", "owner-a")
1250                .await
1251                .unwrap()
1252        );
1253        assert!(store.acp_session_exists("theirs").await.unwrap());
1254    }
1255
1256    #[tokio::test]
1257    async fn delete_for_owner_deletes_own_and_legacy_rows() {
1258        let store = make_store().await;
1259        store
1260            .create_acp_session("mine", Some("owner-a"))
1261            .await
1262            .unwrap();
1263        store.create_acp_session("legacy", None).await.unwrap();
1264
1265        assert!(
1266            store
1267                .delete_acp_session_for_owner("mine", "owner-a")
1268                .await
1269                .unwrap()
1270        );
1271        assert!(
1272            store
1273                .delete_acp_session_for_owner("legacy", "owner-a")
1274                .await
1275                .unwrap()
1276        );
1277        assert!(!store.acp_session_exists("mine").await.unwrap());
1278        assert!(!store.acp_session_exists("legacy").await.unwrap());
1279    }
1280
1281    #[tokio::test]
1282    async fn update_title_for_owner_refuses_foreign_owner() {
1283        let store = make_store().await;
1284        store
1285            .create_acp_session("theirs", Some("owner-b"))
1286            .await
1287            .unwrap();
1288
1289        assert!(
1290            !store
1291                .update_session_title_for_owner("theirs", "new title", "owner-a")
1292                .await
1293                .unwrap()
1294        );
1295    }
1296
1297    #[tokio::test]
1298    async fn update_title_for_owner_updates_own_row() {
1299        let store = make_store().await;
1300        store
1301            .create_acp_session("mine", Some("owner-a"))
1302            .await
1303            .unwrap();
1304
1305        assert!(
1306            store
1307                .update_session_title_for_owner("mine", "new title", "owner-a")
1308                .await
1309                .unwrap()
1310        );
1311        let info = store
1312            .get_acp_session_info_for_owner("mine", "owner-a")
1313            .await
1314            .unwrap()
1315            .unwrap();
1316        assert_eq!(info.title.as_deref(), Some("new title"));
1317    }
1318}
zeph_memory/store/acp_sessions.rs

zeph_memory/store/
acp_sessions.rs
