zeph_memory/store/
compression_guidelines.rs

1// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
2// SPDX-License-Identifier: MIT OR Apache-2.0
3
4//! SQLite-backed store for ACON compression guidelines and failure pairs.
5
6use std::borrow::Cow;
7use std::sync::LazyLock;
8#[allow(unused_imports)]
9use zeph_db::sql;
10
11use regex::Regex;
12use zeph_common::text::truncate_to_bytes_ref;
13
14use crate::error::MemoryError;
15use crate::store::SqliteStore;
16use crate::types::ConversationId;
17
18static SECRET_RE: LazyLock<Regex> = LazyLock::new(|| {
19    Regex::new(
20        r#"(?:sk-|sk_live_|sk_test_|AKIA|ghp_|gho_|-----BEGIN|xoxb-|xoxp-|AIza|ya29\.|glpat-|hf_|npm_|dckr_pat_)[^\s"'`,;\{\}\[\]]*"#,
21    )
22    .expect("secret regex")
23});
24
25static PATH_RE: LazyLock<Regex> = LazyLock::new(|| {
26    Regex::new(r#"(?:/home/|/Users/|/root/|/tmp/|/var/)[^\s"'`,;\{\}\[\]]*"#).expect("path regex")
27});
28
29/// Matches `Authorization: Bearer <token>` headers; captures the token value for redaction.
30static BEARER_RE: LazyLock<Regex> =
31    LazyLock::new(|| Regex::new(r"(?i)(Authorization:\s*Bearer\s+)\S+").expect("bearer regex"));
32
33/// Matches standalone JWT tokens (three Base64url-encoded parts separated by dots).
34/// The signature segment uses `*` to handle `alg=none` JWTs with an empty signature.
35static JWT_RE: LazyLock<Regex> = LazyLock::new(|| {
36    Regex::new(r"eyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]*").expect("jwt regex")
37});
38
39/// Redact secrets and filesystem paths from text before persistent storage.
40///
41/// Returns `Cow::Borrowed` when no sensitive content is found (zero-alloc fast path).
42pub(crate) fn redact_sensitive(text: &str) -> Cow<'_, str> {
43    // Each replace_all may return Cow::Borrowed (no match) or Cow::Owned (replaced).
44    // We materialise intermediate Owned values into String so that subsequent steps
45    // do not hold a borrow of a local.
46    let s0: Cow<'_, str> = SECRET_RE.replace_all(text, "[REDACTED]");
47    let s1: Cow<'_, str> = match PATH_RE.replace_all(s0.as_ref(), "[PATH]") {
48        Cow::Borrowed(_) => s0,
49        Cow::Owned(o) => Cow::Owned(o),
50    };
51    // Replace only the token value in Bearer headers, keeping the header name intact.
52    let s2: Cow<'_, str> = match BEARER_RE.replace_all(s1.as_ref(), "${1}[REDACTED]") {
53        Cow::Borrowed(_) => s1,
54        Cow::Owned(o) => Cow::Owned(o),
55    };
56    match JWT_RE.replace_all(s2.as_ref(), "[REDACTED_JWT]") {
57        Cow::Borrowed(_) => s2,
58        Cow::Owned(o) => Cow::Owned(o),
59    }
60}
61
62/// A recorded compression failure pair: the compressed context and the response
63/// that indicated context was lost.
64#[derive(Debug, Clone)]
65pub struct CompressionFailurePair {
66    pub id: i64,
67    pub conversation_id: ConversationId,
68    pub compressed_context: String,
69    pub failure_reason: String,
70    pub category: String,
71    pub created_at: String,
72}
73
74/// Maximum characters stored per `compressed_context` or `failure_reason` field.
75const MAX_FIELD_CHARS: usize = 4096;
76
77fn truncate_field(s: &str) -> &str {
78    truncate_to_bytes_ref(s, MAX_FIELD_CHARS)
79}
80
81impl SqliteStore {
82    /// Load the latest active compression guidelines.
83    ///
84    /// When `conversation_id` is `Some`, returns conversation-specific guidelines
85    /// preferred over global (NULL) ones. When `None`, returns only global guidelines.
86    ///
87    /// Returns `(version, guidelines_text)`. Returns `(0, "")` if no guidelines exist yet.
88    ///
89    /// # Errors
90    ///
91    /// Returns an error if the database query fails.
92    pub async fn load_compression_guidelines(
93        &self,
94        conversation_id: Option<ConversationId>,
95    ) -> Result<(i64, String), MemoryError> {
96        // `version` is `INTEGER` (`INT4`) on Postgres, so it decodes as `i32`, not `i64`;
97        // widened back to `i64` below to keep this function's public return type unchanged.
98        let row = zeph_db::query_as::<_, (i32, String)>(sql!(
99            // When conversation_id is Some(cid): `conversation_id = cid` matches
100            // conversation-specific rows; `conversation_id IS NULL` matches global rows.
101            // The CASE ensures conversation-specific rows sort before global ones.
102            // When conversation_id is None: `conversation_id = NULL` is always false in SQL,
103            // so only `conversation_id IS NULL` rows match — correct global-only behavior.
104            "SELECT version, guidelines FROM compression_guidelines \
105             WHERE conversation_id = ? OR conversation_id IS NULL \
106             ORDER BY CASE WHEN conversation_id IS NOT NULL THEN 0 ELSE 1 END, \
107                      version DESC \
108             LIMIT 1"
109        ))
110        .bind(conversation_id.map(|c| c.0))
111        .fetch_optional(&self.pool)
112        .await?;
113
114        Ok(row.map_or((0, String::new()), |(version, guidelines)| {
115            (i64::from(version), guidelines)
116        }))
117    }
118
119    /// Load only the version and creation timestamp of the latest active compression guidelines.
120    ///
121    /// Same scoping rules as [`Self::load_compression_guidelines`]: conversation-specific rows are
122    /// preferred over global ones.  Returns `(0, "")` if no guidelines exist yet.
123    ///
124    /// Use this in hot paths where the full text is not needed (e.g. metrics sync).
125    ///
126    /// # Errors
127    ///
128    /// Returns an error if the database query fails.
129    pub async fn load_compression_guidelines_meta(
130        &self,
131        conversation_id: Option<ConversationId>,
132    ) -> Result<(i64, String), MemoryError> {
133        // `created_at` is `TIMESTAMPTZ` on Postgres (`TEXT` on SQLite); project through
134        // `Dialect::select_as_text` so it decodes into the `String` tuple field below.
135        // `version` is `INTEGER` (`INT4`) on Postgres, so it decodes as `i32`, not `i64`;
136        // widened back to `i64` below to keep this function's public return type unchanged.
137        let created_at_sel =
138            <zeph_db::ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
139        let raw = format!(
140            "SELECT version, {created_at_sel} FROM compression_guidelines \
141             WHERE conversation_id = ? OR conversation_id IS NULL \
142             ORDER BY CASE WHEN conversation_id IS NOT NULL THEN 0 ELSE 1 END, \
143                      version DESC \
144             LIMIT 1"
145        );
146        let query_sql = zeph_db::rewrite_placeholders(&raw);
147        let row = zeph_db::query_as::<_, (i32, String)>(sqlx::AssertSqlSafe(query_sql))
148            .bind(conversation_id.map(|c| c.0)) // lgtm[rust/cleartext-logging]
149            .fetch_optional(&self.pool)
150            .await?;
151
152        Ok(row.map_or((0, String::new()), |(version, created_at)| {
153            (i64::from(version), created_at)
154        }))
155    }
156
157    /// Save a new version of the compression guidelines.
158    ///
159    /// When `conversation_id` is `Some`, the guidelines are scoped to that conversation.
160    /// When `None`, the guidelines are global (apply as fallback for all conversations).
161    ///
162    /// Inserts a new row; older versions are retained for audit.
163    /// Returns the new version number.
164    ///
165    /// Note: version numbers are globally sequential across all conversation scopes —
166    /// they are not per-conversation counters. The UNIQUE(version) constraint from
167    /// migration 033 is preserved.
168    ///
169    /// # Errors
170    ///
171    /// Returns an error if the database insert fails (including FK violation if
172    /// `conversation_id` does not reference a valid conversation row).
173    pub async fn save_compression_guidelines(
174        &self,
175        guidelines: &str,
176        token_count: i64,
177        conversation_id: Option<ConversationId>,
178    ) -> Result<i64, MemoryError> {
179        // The INSERT...SELECT computes MAX(version)+1 across all rows (global + per-conversation)
180        // and inserts it in a single statement. SQLite's single-writer WAL guarantee makes this
181        // atomic — no concurrent writer can observe the same MAX and produce a duplicate version.
182        // `version` is `INTEGER` (`INT4`) on Postgres, so `RETURNING version` decodes as `i32`,
183        // not `i64`; widened back to `i64` below to keep this function's public return type
184        // unchanged.
185        let new_version: i32 = zeph_db::query_scalar(
186            sql!("INSERT INTO compression_guidelines (version, guidelines, token_count, conversation_id) \
187             SELECT COALESCE(MAX(version), 0) + 1, ?, ?, ? \
188             FROM compression_guidelines \
189             RETURNING version"),
190        )
191        .bind(guidelines)
192        .bind(token_count)
193        .bind(conversation_id.map(|c| c.0))
194        .fetch_one(&self.pool)
195        .await?;
196        Ok(i64::from(new_version))
197    }
198
199    /// Log a compression failure pair.
200    ///
201    /// Both `compressed_context` and `failure_reason` are truncated to 4096 chars.
202    /// `category` should be one of: `tool_output`, `assistant_reasoning`, `user_context`, `unknown`.
203    /// Returns the inserted row id.
204    ///
205    /// # Errors
206    ///
207    /// Returns an error if the database insert fails.
208    pub async fn log_compression_failure(
209        &self,
210        conversation_id: ConversationId,
211        compressed_context: &str,
212        failure_reason: &str,
213        category: &str,
214    ) -> Result<i64, MemoryError> {
215        let ctx = redact_sensitive(compressed_context);
216        let ctx = truncate_field(&ctx);
217        let reason = redact_sensitive(failure_reason);
218        let reason = truncate_field(&reason);
219        let id = zeph_db::query_scalar(sql!(
220            "INSERT INTO compression_failure_pairs \
221             (conversation_id, compressed_context, failure_reason, category) \
222             VALUES (?, ?, ?, ?) RETURNING id"
223        ))
224        .bind(conversation_id.0)
225        .bind(ctx)
226        .bind(reason)
227        .bind(category)
228        .fetch_one(&self.pool)
229        .await?;
230        Ok(id)
231    }
232
233    /// Get unused failure pairs (oldest first), up to `limit`.
234    ///
235    /// # Errors
236    ///
237    /// Returns an error if the database query fails.
238    pub async fn get_unused_failure_pairs(
239        &self,
240        limit: usize,
241    ) -> Result<Vec<CompressionFailurePair>, MemoryError> {
242        let limit = i64::try_from(limit).unwrap_or(i64::MAX);
243        // `created_at` is `TIMESTAMPTZ` on Postgres (`TEXT` on SQLite); project through
244        // `Dialect::select_as_text` so it decodes into the `String` tuple field below.
245        // `ORDER BY` is table-qualified so it sorts on the native timestamp, not the cast text.
246        let created_at_sel =
247            <zeph_db::ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
248        let raw = format!(
249            "SELECT id, conversation_id, compressed_context, failure_reason, category, {created_at_sel} \
250             FROM compression_failure_pairs \
251             WHERE used_in_update = FALSE \
252             ORDER BY compression_failure_pairs.created_at ASC \
253             LIMIT ?"
254        );
255        let query_sql = zeph_db::rewrite_placeholders(&raw);
256        let rows = zeph_db::query_as::<_, (i64, i64, String, String, String, String)>(
257            sqlx::AssertSqlSafe(query_sql),
258        )
259        .bind(limit)
260        .fetch_all(&self.pool)
261        .await?;
262
263        Ok(rows
264            .into_iter()
265            .map(
266                |(id, cid, ctx, reason, category, created_at)| CompressionFailurePair {
267                    id,
268                    conversation_id: ConversationId(cid),
269                    compressed_context: ctx,
270                    failure_reason: reason,
271                    category,
272                    created_at,
273                },
274            )
275            .collect())
276    }
277
278    /// Get unused failure pairs for a specific category (oldest first), up to `limit`.
279    ///
280    /// Used by the categorized ACON updater to run per-category update cycles.
281    ///
282    /// # Errors
283    ///
284    /// Returns an error if the database query fails.
285    pub async fn get_unused_failure_pairs_by_category(
286        &self,
287        category: &str,
288        limit: usize,
289    ) -> Result<Vec<CompressionFailurePair>, MemoryError> {
290        let limit = i64::try_from(limit).unwrap_or(i64::MAX);
291        // `created_at` is `TIMESTAMPTZ` on Postgres — see `get_unused_failure_pairs`.
292        let created_at_sel =
293            <zeph_db::ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
294        let raw = format!(
295            "SELECT id, conversation_id, compressed_context, failure_reason, category, {created_at_sel} \
296             FROM compression_failure_pairs \
297             WHERE used_in_update = FALSE AND category = ? \
298             ORDER BY compression_failure_pairs.created_at ASC \
299             LIMIT ?"
300        );
301        let query_sql = zeph_db::rewrite_placeholders(&raw);
302        let rows = zeph_db::query_as::<_, (i64, i64, String, String, String, String)>(
303            sqlx::AssertSqlSafe(query_sql),
304        )
305        .bind(category)
306        .bind(limit)
307        .fetch_all(&self.pool)
308        .await?;
309
310        Ok(rows
311            .into_iter()
312            .map(
313                |(id, cid, ctx, reason, cat, created_at)| CompressionFailurePair {
314                    id,
315                    conversation_id: ConversationId(cid),
316                    compressed_context: ctx,
317                    failure_reason: reason,
318                    category: cat,
319                    created_at,
320                },
321            )
322            .collect())
323    }
324
325    /// Count unused failure pairs for a specific category.
326    ///
327    /// # Errors
328    ///
329    /// Returns an error if the database query fails.
330    pub async fn count_unused_failure_pairs_by_category(
331        &self,
332        category: &str,
333    ) -> Result<i64, MemoryError> {
334        let count = zeph_db::query_scalar(sql!(
335            "SELECT COUNT(*) FROM compression_failure_pairs \
336             WHERE used_in_update = FALSE AND category = ?"
337        ))
338        .bind(category)
339        .fetch_one(&self.pool)
340        .await?;
341        Ok(count)
342    }
343
344    /// Load the latest compression guidelines for a specific category.
345    ///
346    /// When `conversation_id` is `Some`, prefers conversation-specific rows.
347    /// Returns `(0, "")` if no guidelines exist for this category.
348    ///
349    /// # Errors
350    ///
351    /// Returns an error if the database query fails.
352    pub async fn load_compression_guidelines_by_category(
353        &self,
354        category: &str,
355        conversation_id: Option<ConversationId>,
356    ) -> Result<(i64, String), MemoryError> {
357        // `version` is `INTEGER` (`INT4`) on Postgres, so it decodes as `i32`, not `i64`;
358        // widened back to `i64` below to keep this function's public return type unchanged.
359        let row = zeph_db::query_as::<_, (i32, String)>(sql!(
360            "SELECT version, guidelines FROM compression_guidelines \
361             WHERE category = ? \
362             AND (conversation_id = ? OR conversation_id IS NULL) \
363             ORDER BY CASE WHEN conversation_id IS NOT NULL THEN 0 ELSE 1 END, \
364                      version DESC \
365             LIMIT 1"
366        ))
367        .bind(category)
368        .bind(conversation_id.map(|c| c.0))
369        .fetch_optional(&self.pool)
370        .await?;
371
372        Ok(row.map_or((0, String::new()), |(version, guidelines)| {
373            (i64::from(version), guidelines)
374        }))
375    }
376
377    /// Save a new version of compression guidelines for a specific category.
378    ///
379    /// # Errors
380    ///
381    /// Returns an error if the database insert fails.
382    pub async fn save_compression_guidelines_with_category(
383        &self,
384        guidelines: &str,
385        token_count: i64,
386        category: &str,
387        conversation_id: Option<ConversationId>,
388    ) -> Result<i64, MemoryError> {
389        // `version` is `INTEGER` (`INT4`) on Postgres, so `RETURNING version` decodes as `i32`,
390        // not `i64`; widened back to `i64` below to keep this function's public return type
391        // unchanged.
392        let new_version: i32 = zeph_db::query_scalar(sql!(
393            "INSERT INTO compression_guidelines \
394             (version, category, guidelines, token_count, conversation_id) \
395             SELECT COALESCE(MAX(version), 0) + 1, ?, ?, ?, ? \
396             FROM compression_guidelines \
397             RETURNING version"
398        ))
399        .bind(category)
400        .bind(guidelines)
401        .bind(token_count)
402        .bind(conversation_id.map(|c| c.0))
403        .fetch_one(&self.pool)
404        .await?;
405        Ok(i64::from(new_version))
406    }
407
408    /// Mark failure pairs as consumed by the updater.
409    ///
410    /// # Errors
411    ///
412    /// Returns an error if the database update fails.
413    pub async fn mark_failure_pairs_used(&self, ids: &[i64]) -> Result<(), MemoryError> {
414        if ids.is_empty() {
415            return Ok(());
416        }
417        let placeholders = zeph_db::placeholder_list(1, ids.len());
418        let query = format!(
419            "UPDATE compression_failure_pairs SET used_in_update = TRUE WHERE id IN ({placeholders})"
420        );
421        let mut q = zeph_db::query(sqlx::AssertSqlSafe(query));
422        for id in ids {
423            q = q.bind(id);
424        }
425        q.execute(&self.pool).await?;
426        Ok(())
427    }
428
429    /// Count unused failure pairs.
430    ///
431    /// # Errors
432    ///
433    /// Returns an error if the database query fails.
434    pub async fn count_unused_failure_pairs(&self) -> Result<i64, MemoryError> {
435        let count = zeph_db::query_scalar(sql!(
436            "SELECT COUNT(*) FROM compression_failure_pairs WHERE used_in_update = FALSE"
437        ))
438        .fetch_one(&self.pool)
439        .await?;
440        Ok(count)
441    }
442
443    /// Delete old used failure pairs, keeping the most recent `keep_recent` unused pairs.
444    ///
445    /// Removes all rows where `used_in_update = TRUE`. Unused rows are managed by the
446    /// `max_stored_pairs` enforcement below: if there are more than `keep_recent` unused pairs,
447    /// the oldest excess rows are deleted.
448    ///
449    /// # Errors
450    ///
451    /// Returns an error if the database query fails.
452    pub async fn cleanup_old_failure_pairs(&self, keep_recent: usize) -> Result<(), MemoryError> {
453        // Delete all used pairs (they've already been processed).
454        zeph_db::query(sql!(
455            "DELETE FROM compression_failure_pairs WHERE used_in_update = TRUE"
456        ))
457        .execute(&self.pool)
458        .await?;
459
460        // Keep only the most recent `keep_recent` unused pairs.
461        let keep = i64::try_from(keep_recent).unwrap_or(i64::MAX);
462        zeph_db::query(sql!(
463            "DELETE FROM compression_failure_pairs \
464             WHERE used_in_update = FALSE \
465             AND id NOT IN ( \
466                 SELECT id FROM compression_failure_pairs \
467                 WHERE used_in_update = FALSE \
468                 ORDER BY created_at DESC \
469                 LIMIT ? \
470             )"
471        ))
472        .bind(keep)
473        .execute(&self.pool)
474        .await?;
475
476        Ok(())
477    }
478}
479
480#[cfg(test)]
481mod tests {
482    use super::*;
483
484    // pool_size=1 is required: SQLite :memory: creates an isolated database per
485    // connection, so multiple connections would each see an empty schema.
486    async fn make_store() -> SqliteStore {
487        SqliteStore::with_pool_size(":memory:", 1)
488            .await
489            .expect("in-memory SqliteStore")
490    }
491
492    #[tokio::test]
493    async fn load_guidelines_meta_returns_defaults_when_empty() {
494        let store = make_store().await;
495        let (version, created_at) = store.load_compression_guidelines_meta(None).await.unwrap();
496        assert_eq!(version, 0);
497        assert!(created_at.is_empty());
498    }
499
500    #[tokio::test]
501    async fn load_guidelines_meta_returns_version_and_created_at() {
502        let store = make_store().await;
503        store
504            .save_compression_guidelines("keep file paths", 4, None)
505            .await
506            .unwrap();
507        let (version, created_at) = store.load_compression_guidelines_meta(None).await.unwrap();
508        assert_eq!(version, 1);
509        assert!(!created_at.is_empty(), "created_at should be populated");
510    }
511
512    #[tokio::test]
513    async fn load_guidelines_returns_defaults_when_empty() {
514        let store = make_store().await;
515        let (version, text) = store.load_compression_guidelines(None).await.unwrap();
516        assert_eq!(version, 0);
517        assert!(text.is_empty());
518    }
519
520    #[tokio::test]
521    async fn save_and_load_guidelines() {
522        let store = make_store().await;
523        let v1 = store
524            .save_compression_guidelines("always preserve file paths", 4, None)
525            .await
526            .unwrap();
527        assert_eq!(v1, 1);
528        let v2 = store
529            .save_compression_guidelines(
530                "always preserve file paths\nalways preserve errors",
531                8,
532                None,
533            )
534            .await
535            .unwrap();
536        assert_eq!(v2, 2);
537        // Loading should return the latest version.
538        let (v, text) = store.load_compression_guidelines(None).await.unwrap();
539        assert_eq!(v, 2);
540        assert!(text.contains("errors"));
541    }
542
543    #[tokio::test]
544    async fn load_guidelines_prefers_conversation_specific() {
545        let store = make_store().await;
546        let cid = ConversationId(store.create_conversation().await.unwrap().0);
547        store
548            .save_compression_guidelines("global rule", 2, None)
549            .await
550            .unwrap();
551        store
552            .save_compression_guidelines("conversation rule", 2, Some(cid))
553            .await
554            .unwrap();
555        let (_, text) = store.load_compression_guidelines(Some(cid)).await.unwrap();
556        assert_eq!(text, "conversation rule");
557    }
558
559    #[tokio::test]
560    async fn load_guidelines_falls_back_to_global() {
561        let store = make_store().await;
562        let cid = ConversationId(store.create_conversation().await.unwrap().0);
563        store
564            .save_compression_guidelines("global rule", 2, None)
565            .await
566            .unwrap();
567        // No conversation-specific guidelines; should fall back to global.
568        let (_, text) = store.load_compression_guidelines(Some(cid)).await.unwrap();
569        assert_eq!(text, "global rule");
570    }
571
572    #[tokio::test]
573    async fn load_guidelines_none_returns_global_only() {
574        let store = make_store().await;
575        let cid = ConversationId(store.create_conversation().await.unwrap().0);
576        store
577            .save_compression_guidelines("conversation rule", 2, Some(cid))
578            .await
579            .unwrap();
580        // None should not return conversation-scoped guidelines.
581        let (version, text) = store.load_compression_guidelines(None).await.unwrap();
582        assert_eq!(version, 0);
583        assert!(text.is_empty());
584    }
585
586    #[tokio::test]
587    async fn load_guidelines_scope_isolation() {
588        let store = make_store().await;
589        let cid_a = ConversationId(store.create_conversation().await.unwrap().0);
590        let cid_b = ConversationId(store.create_conversation().await.unwrap().0);
591
592        // Global guideline (conversation_id = None) — visible to all conversations.
593        store
594            .save_compression_guidelines("Use bullet points", 1, None)
595            .await
596            .unwrap();
597        // Conversation-A-specific guideline — must NOT be visible to B.
598        store
599            .save_compression_guidelines("Be concise", 2, Some(cid_a))
600            .await
601            .unwrap();
602
603        // Conversation B: gets only the global guideline, not A's.
604        let (_, text_b) = store
605            .load_compression_guidelines(Some(cid_b))
606            .await
607            .unwrap();
608        assert_eq!(
609            text_b, "Use bullet points",
610            "conversation B must see global guideline"
611        );
612
613        // Conversation A: gets its own guideline (preferred over global).
614        let (_, text_a) = store
615            .load_compression_guidelines(Some(cid_a))
616            .await
617            .unwrap();
618        assert_eq!(
619            text_a, "Be concise",
620            "conversation A must prefer its own guideline over global"
621        );
622
623        // None scope: gets only the global guideline.
624        let (_, text_global) = store.load_compression_guidelines(None).await.unwrap();
625        assert_eq!(
626            text_global, "Use bullet points",
627            "None scope must see only the global guideline"
628        );
629    }
630
631    #[tokio::test]
632    async fn save_with_nonexistent_conversation_id_fails() {
633        let store = make_store().await;
634        let nonexistent = ConversationId(99999);
635        let result = store
636            .save_compression_guidelines("rule", 1, Some(nonexistent))
637            .await;
638        assert!(
639            result.is_err(),
640            "FK violation expected for nonexistent conversation_id"
641        );
642    }
643
644    #[tokio::test]
645    async fn cascade_delete_removes_conversation_guidelines() {
646        let store = make_store().await;
647        let cid = ConversationId(store.create_conversation().await.unwrap().0);
648        store
649            .save_compression_guidelines("rule", 1, Some(cid))
650            .await
651            .unwrap();
652        // Delete the conversation row directly — should cascade-delete the guideline.
653        zeph_db::query(sql!("DELETE FROM conversations WHERE id = ?"))
654            .bind(cid.0)
655            .execute(store.pool())
656            .await
657            .unwrap();
658        let (version, text) = store.load_compression_guidelines(Some(cid)).await.unwrap();
659        assert_eq!(version, 0);
660        assert!(text.is_empty());
661    }
662
663    #[tokio::test]
664    async fn log_and_count_failure_pairs() {
665        let store = make_store().await;
666        let cid = ConversationId(store.create_conversation().await.unwrap().0);
667        store
668            .log_compression_failure(cid, "compressed ctx", "i don't recall that", "unknown")
669            .await
670            .unwrap();
671        let count = store.count_unused_failure_pairs().await.unwrap();
672        assert_eq!(count, 1);
673    }
674
675    #[tokio::test]
676    async fn get_unused_pairs_sorted_oldest_first() {
677        let store = make_store().await;
678        let cid = ConversationId(store.create_conversation().await.unwrap().0);
679        store
680            .log_compression_failure(cid, "ctx A", "reason A", "unknown")
681            .await
682            .unwrap();
683        store
684            .log_compression_failure(cid, "ctx B", "reason B", "unknown")
685            .await
686            .unwrap();
687        let pairs = store.get_unused_failure_pairs(10).await.unwrap();
688        assert_eq!(pairs.len(), 2);
689        assert_eq!(pairs[0].compressed_context, "ctx A");
690    }
691
692    #[tokio::test]
693    async fn mark_pairs_used_reduces_count() {
694        let store = make_store().await;
695        let cid = ConversationId(store.create_conversation().await.unwrap().0);
696        let id = store
697            .log_compression_failure(cid, "ctx", "reason", "unknown")
698            .await
699            .unwrap();
700        store.mark_failure_pairs_used(&[id]).await.unwrap();
701        let count = store.count_unused_failure_pairs().await.unwrap();
702        assert_eq!(count, 0);
703    }
704
705    #[tokio::test]
706    async fn cleanup_deletes_used_and_trims_unused() {
707        let store = make_store().await;
708        let cid = ConversationId(store.create_conversation().await.unwrap().0);
709        // Add 3 pairs and mark 1 used.
710        let id1 = store
711            .log_compression_failure(cid, "ctx1", "r1", "tool_output")
712            .await
713            .unwrap();
714        store
715            .log_compression_failure(cid, "ctx2", "r2", "tool_output")
716            .await
717            .unwrap();
718        store
719            .log_compression_failure(cid, "ctx3", "r3", "unknown")
720            .await
721            .unwrap();
722        store.mark_failure_pairs_used(&[id1]).await.unwrap();
723        // Cleanup: keep at most 1 unused.
724        store.cleanup_old_failure_pairs(1).await.unwrap();
725        let count = store.count_unused_failure_pairs().await.unwrap();
726        assert_eq!(count, 1, "only 1 unused pair should remain");
727    }
728
729    #[test]
730    fn redact_sensitive_api_key_is_redacted() {
731        let result = redact_sensitive("token sk-abc123def456 used for auth");
732        assert!(result.contains("[REDACTED]"), "API key must be redacted");
733        assert!(
734            !result.contains("sk-abc123"),
735            "original key must not appear"
736        );
737    }
738
739    #[test]
740    fn redact_sensitive_plain_text_borrows() {
741        let text = "safe text, no secrets here";
742        let result = redact_sensitive(text);
743        assert!(
744            matches!(result, Cow::Borrowed(_)),
745            "plain text must return Cow::Borrowed (zero-alloc)"
746        );
747    }
748
749    #[test]
750    fn redact_sensitive_filesystem_path_is_redacted() {
751        let result = redact_sensitive("config loaded from /Users/dev/project/config.toml");
752        assert!(
753            result.contains("[PATH]"),
754            "filesystem path must be redacted"
755        );
756        assert!(
757            !result.contains("/Users/dev/"),
758            "original path must not appear"
759        );
760    }
761
762    #[test]
763    fn redact_sensitive_combined_secret_and_path() {
764        let result = redact_sensitive("key sk-abc at /home/user/file");
765        assert!(result.contains("[REDACTED]"), "secret must be redacted");
766        assert!(result.contains("[PATH]"), "path must be redacted");
767    }
768
769    #[tokio::test]
770    async fn log_compression_failure_redacts_secrets() {
771        let store = make_store().await;
772        let cid = ConversationId(store.create_conversation().await.unwrap().0);
773        store
774            .log_compression_failure(
775                cid,
776                "token sk-abc123def456 used for auth",
777                "context lost",
778                "unknown",
779            )
780            .await
781            .unwrap();
782        let pairs = store.get_unused_failure_pairs(10).await.unwrap();
783        assert_eq!(pairs.len(), 1);
784        assert!(
785            pairs[0].compressed_context.contains("[REDACTED]"),
786            "stored context must have redacted secret"
787        );
788        assert!(
789            !pairs[0].compressed_context.contains("sk-abc123"),
790            "stored context must not contain raw secret"
791        );
792    }
793
794    #[tokio::test]
795    async fn log_compression_failure_redacts_paths() {
796        let store = make_store().await;
797        let cid = ConversationId(store.create_conversation().await.unwrap().0);
798        store
799            .log_compression_failure(
800                cid,
801                "/Users/dev/project/config.toml was loaded",
802                "lost",
803                "unknown",
804            )
805            .await
806            .unwrap();
807        let pairs = store.get_unused_failure_pairs(10).await.unwrap();
808        assert!(
809            pairs[0].compressed_context.contains("[PATH]"),
810            "stored context must have redacted path"
811        );
812        assert!(
813            !pairs[0].compressed_context.contains("/Users/dev/"),
814            "stored context must not contain raw path"
815        );
816    }
817
818    #[tokio::test]
819    async fn log_compression_failure_reason_also_redacted() {
820        let store = make_store().await;
821        let cid = ConversationId(store.create_conversation().await.unwrap().0);
822        store
823            .log_compression_failure(
824                cid,
825                "some context",
826                "secret ghp_abc123xyz was leaked",
827                "unknown",
828            )
829            .await
830            .unwrap();
831        let pairs = store.get_unused_failure_pairs(10).await.unwrap();
832        assert!(
833            pairs[0].failure_reason.contains("[REDACTED]"),
834            "failure_reason must also be redacted"
835        );
836        assert!(
837            !pairs[0].failure_reason.contains("ghp_abc123xyz"),
838            "raw secret must not appear in failure_reason"
839        );
840    }
841
842    #[tokio::test]
843    async fn truncate_field_respects_char_boundary() {
844        let s = "а".repeat(5000); // Cyrillic 'а', 2 bytes each
845        let truncated = truncate_field(&s);
846        assert!(truncated.len() <= MAX_FIELD_CHARS);
847        assert!(s.is_char_boundary(truncated.len()));
848    }
849
850    #[tokio::test]
851    async fn unique_constraint_prevents_duplicate_version() {
852        let store = make_store().await;
853        // Insert version 1 via the public API.
854        store
855            .save_compression_guidelines("first", 1, None)
856            .await
857            .unwrap();
858        // store.pool() access is intentional: we need direct pool access to bypass
859        // the public API and test the UNIQUE constraint at the SQL level.
860        let result = zeph_db::query(
861            sql!("INSERT INTO compression_guidelines (version, guidelines, token_count) VALUES (1, 'dup', 0)"),
862        )
863        .execute(store.pool())
864        .await;
865        assert!(
866            result.is_err(),
867            "duplicate version insert should violate UNIQUE constraint"
868        );
869    }
870
871    #[test]
872    fn redact_sensitive_bearer_token_is_redacted() {
873        let result =
874            redact_sensitive("Authorization: Bearer eyJhbGciOiJSUzI1NiJ9.payload.signature");
875        assert!(
876            result.contains("[REDACTED]"),
877            "Bearer token must be redacted: {result}"
878        );
879        assert!(
880            !result.contains("eyJhbGciOiJSUzI1NiJ9"),
881            "raw JWT header must not appear: {result}"
882        );
883        assert!(
884            result.contains("Authorization:"),
885            "header name must be preserved: {result}"
886        );
887    }
888
889    #[test]
890    fn redact_sensitive_bearer_token_case_insensitive() {
891        let result =
892            redact_sensitive("authorization: bearer eyJhbGciOiJSUzI1NiJ9.payload.signature");
893        assert!(
894            result.contains("[REDACTED]"),
895            "Bearer header match must be case-insensitive: {result}"
896        );
897    }
898
899    #[test]
900    fn redact_sensitive_standalone_jwt_is_redacted() {
901        let jwt = "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJ1c2VyMTIzIn0.SflKxwRJSMeKKF2";
902        let input = format!("token value: {jwt} was found in logs");
903        let result = redact_sensitive(&input);
904        assert!(
905            result.contains("[REDACTED_JWT]"),
906            "standalone JWT must be replaced with [REDACTED_JWT]: {result}"
907        );
908        assert!(
909            !result.contains("eyJhbGci"),
910            "raw JWT must not appear: {result}"
911        );
912    }
913
914    #[test]
915    fn redact_sensitive_mixed_content_all_redacted() {
916        let input =
917            "key sk-abc123 at /home/user/f with Authorization: Bearer eyJhbG.pay.sig and eyJx.b.c";
918        let result = redact_sensitive(input);
919        assert!(result.contains("[REDACTED]"), "API key must be redacted");
920        assert!(result.contains("[PATH]"), "path must be redacted");
921        assert!(!result.contains("sk-abc123"), "raw API key must not appear");
922        assert!(!result.contains("eyJhbG"), "raw JWT must not appear");
923    }
924
925    #[test]
926    fn redact_sensitive_partial_jwt_not_redacted() {
927        // A string starting with eyJ but missing the third segment is not a valid JWT.
928        let input = "eyJhbGciOiJSUzI1NiJ9.onlytwoparts";
929        let result = redact_sensitive(input);
930        // Should not be replaced by the JWT regex (only two dot-separated parts).
931        assert!(
932            !result.contains("[REDACTED_JWT]"),
933            "two-part eyJ string must not be treated as JWT: {result}"
934        );
935        // No substitution occurred — must be zero-alloc Cow::Borrowed.
936        assert!(
937            matches!(result, Cow::Borrowed(_)),
938            "no-match input must return Cow::Borrowed: {result}"
939        );
940    }
941
942    #[test]
943    fn redact_sensitive_alg_none_jwt_empty_signature_redacted() {
944        // alg=none JWTs have an empty third segment: <header>.<payload>.
945        let input =
946            "token: eyJhbGciOiJub25lIn0.eyJzdWIiOiJ1c2VyIn0. was submitted without signature";
947        let result = redact_sensitive(input);
948        assert!(
949            result.contains("[REDACTED_JWT]"),
950            "alg=none JWT with empty signature must be redacted: {result}"
951        );
952        assert!(
953            !result.contains("eyJhbGciOiJub25lIn0"),
954            "raw alg=none JWT header must not appear: {result}"
955        );
956    }
957
958    // ── Category-aware store methods (MF-4) ──────────────────────────────────
959
960    #[tokio::test]
961    async fn get_unused_pairs_by_category_filters_correctly() {
962        let store = make_store().await;
963        let cid = ConversationId(store.create_conversation().await.unwrap().0);
964        store
965            .log_compression_failure(cid, "tool ctx", "lost tool output", "tool_output")
966            .await
967            .unwrap();
968        store
969            .log_compression_failure(cid, "user ctx", "lost user context", "user_context")
970            .await
971            .unwrap();
972
973        let tool_pairs = store
974            .get_unused_failure_pairs_by_category("tool_output", 10)
975            .await
976            .unwrap();
977        assert_eq!(tool_pairs.len(), 1);
978        assert_eq!(tool_pairs[0].category, "tool_output");
979        assert_eq!(tool_pairs[0].compressed_context, "tool ctx");
980
981        let user_pairs = store
982            .get_unused_failure_pairs_by_category("user_context", 10)
983            .await
984            .unwrap();
985        assert_eq!(user_pairs.len(), 1);
986        assert_eq!(user_pairs[0].category, "user_context");
987
988        // Unknown category returns nothing.
989        let unknown_pairs = store
990            .get_unused_failure_pairs_by_category("assistant_reasoning", 10)
991            .await
992            .unwrap();
993        assert!(unknown_pairs.is_empty());
994    }
995
996    #[tokio::test]
997    async fn count_unused_pairs_by_category_returns_correct_count() {
998        let store = make_store().await;
999        let cid = ConversationId(store.create_conversation().await.unwrap().0);
1000        store
1001            .log_compression_failure(cid, "ctx A", "reason", "tool_output")
1002            .await
1003            .unwrap();
1004        store
1005            .log_compression_failure(cid, "ctx B", "reason", "tool_output")
1006            .await
1007            .unwrap();
1008        store
1009            .log_compression_failure(cid, "ctx C", "reason", "user_context")
1010            .await
1011            .unwrap();
1012
1013        let tool_count = store
1014            .count_unused_failure_pairs_by_category("tool_output")
1015            .await
1016            .unwrap();
1017        assert_eq!(tool_count, 2);
1018
1019        let user_count = store
1020            .count_unused_failure_pairs_by_category("user_context")
1021            .await
1022            .unwrap();
1023        assert_eq!(user_count, 1);
1024
1025        let unknown_count = store
1026            .count_unused_failure_pairs_by_category("assistant_reasoning")
1027            .await
1028            .unwrap();
1029        assert_eq!(unknown_count, 0);
1030    }
1031
1032    #[tokio::test]
1033    async fn save_and_load_guidelines_by_category() {
1034        let store = make_store().await;
1035        store
1036            .save_compression_guidelines_with_category(
1037                "preserve tool names",
1038                3,
1039                "tool_output",
1040                None,
1041            )
1042            .await
1043            .unwrap();
1044        store
1045            .save_compression_guidelines_with_category("keep user intent", 3, "user_context", None)
1046            .await
1047            .unwrap();
1048
1049        let (_, tool_text) = store
1050            .load_compression_guidelines_by_category("tool_output", None)
1051            .await
1052            .unwrap();
1053        assert_eq!(tool_text, "preserve tool names");
1054
1055        let (_, user_text) = store
1056            .load_compression_guidelines_by_category("user_context", None)
1057            .await
1058            .unwrap();
1059        assert_eq!(user_text, "keep user intent");
1060    }
1061
1062    #[tokio::test]
1063    async fn load_guidelines_by_category_returns_defaults_when_empty() {
1064        let store = make_store().await;
1065        // No guidelines saved for this category.
1066        let (version, text) = store
1067            .load_compression_guidelines_by_category("tool_output", None)
1068            .await
1069            .unwrap();
1070        assert_eq!(version, 0, "version must be 0 when no entries exist");
1071        assert!(text.is_empty(), "text must be empty when no entries exist");
1072    }
1073
1074    /// Concurrent saves must produce strictly unique versions with no collisions.
1075    ///
1076    /// Uses a file-backed database because `SQLite` `:memory:` creates an isolated
1077    /// database per connection — a multi-connection pool over `:memory:` would give
1078    /// each writer its own empty schema and cannot test shared-state atomicity.
1079    #[tokio::test]
1080    async fn concurrent_saves_produce_unique_versions() {
1081        use std::collections::HashSet;
1082        use std::sync::Arc;
1083
1084        let dir = tempfile::tempdir().expect("tempdir");
1085        let db_path = dir.path().join("test.db");
1086        let store = Arc::new(
1087            SqliteStore::with_pool_size(db_path.to_str().expect("utf8 path"), 4)
1088                .await
1089                .expect("file-backed SqliteStore"),
1090        );
1091
1092        let tasks: Vec<_> = (0..8_i64)
1093            .map(|i| {
1094                let s = Arc::clone(&store);
1095                let fut = async move {
1096                    s.save_compression_guidelines(&format!("guideline {i}"), i, None)
1097                        .await
1098                        .expect("concurrent save must succeed")
1099                };
1100                tokio::spawn(fut) // EXEMPT: test-only concurrent writers for UNIQUE version constraint test
1101            })
1102            .collect();
1103
1104        let mut versions = HashSet::new();
1105        for task in tasks {
1106            let v = task.await.expect("task must not panic");
1107            assert!(versions.insert(v), "version {v} appeared more than once");
1108        }
1109        assert_eq!(
1110            versions.len(),
1111            8,
1112            "all 8 saves must produce distinct versions"
1113        );
1114    }
1115}
zeph_memory/store/compression_guidelines.rs

zeph_memory/store/
compression_guidelines.rs
