feat(storage): preserve retention evidence for later evaluation

README.md

@@ -294,6 +294,22 @@ client.set_config(reflexio.SetConfigRequest(
 ))
 ```
 
+#### Optional retention archive
+
+SQLite deployments can preserve rows removed by row-count retention in local
+JSONL files. The feature is off by default.
+
+| Environment variable | Meaning |
+|---|---|
+| `REFLEXIO_RETENTION_ARCHIVE=1` | Archive rows before retention deletes them. |
+| `REFLEXIO_RETENTION_ARCHIVE_DIR=/path` | Override the default `<database-directory>/archive` location. |
+| `REFLEXIO_RETENTION_ARCHIVE_WARN_BYTES=10737418240` | Log an error after total JSONL size exceeds this warning threshold; retention continues. |
+
+Archive-enabled cleanup processes at most 1,000 parent rows per pass and holds
+one SQLite write transaction while copying and deleting that batch. JSONL files
+do not have a hard size limit; operators should monitor and rotate them after a
+warning. Postgres and Supabase do not support this local archive.
+
 ## Integrations
 
 Reflexio integrates with popular AI agent frameworks out of the box:

reflexio/models/api_schema/domain/entities.py

@@ -36,6 +36,7 @@
     "CitationKind",
     "Citation",
     "RetrievedLearningKind",
+    "RetrievedLearningSnapshot",
     "RetrievedLearning",
     "LearningImpact",
     "Interaction",
@@ -174,24 +175,37 @@ class Citation(BaseModel):
 
 type LearningImpact = Literal["positive", "negative", "neutral"]
 
+MAX_RETRIEVED_LEARNING_SNAPSHOT_BYTES = 10 * 1024**2
+
+
+class RetrievedLearningSnapshot(BaseModel):
+    """Source learning fields captured at injection time for later evaluation."""
+
+    title: str = Field(default="", max_length=1_000)
+    content: str = Field(max_length=100_000)
+    trigger: str = Field(default="", max_length=10_000)
+    rationale: str = Field(default="", max_length=10_000)
+
 
 class RetrievedLearning(BaseModel):
     """A learning the caller retrieved and injected into the agent context.
 
-    Deliberately minimal — just the identity pair. It does NOT reuse
-    ``Citation``: citations carry injection-time debug fields (``tag``,
-    ``title``) that callers should not need to supply (or see) when declaring
-    what was retrieved.
+    The identity pair remains sufficient for compatibility. New callers may
+    include the injection-time snapshot needed for historically
+    accurate evaluation.
 
     Attributes:
         kind (RetrievedLearningKind): Which kind of learning this references.
         learning_id (str): Stable storage id — ``profile_id`` for profiles,
             ``user_playbook_id`` for user playbooks, ``agent_playbook_id``
             for agent playbooks (numeric ids as decimal strings).
+        snapshot (RetrievedLearningSnapshot | None): Source learning fields
+            captured at injection time. Optional for older callers.
     """
 
     kind: RetrievedLearningKind
     learning_id: str = Field(min_length=1, max_length=1_000)
+    snapshot: RetrievedLearningSnapshot | None = None
 
 
 # information about the user interaction sent by the client
@@ -824,6 +838,23 @@ def validate_retrieved_learnings_total(self) -> Self:
                 "a publish request may carry at most 1000 retrieved_learnings"
                 f" across all interactions (got {total})"
             )
+        snapshot_bytes = sum(
+            len(value.encode("utf-8"))
+            for interaction in self.interaction_data_list
+            for learning in interaction.retrieved_learnings
+            if learning.snapshot is not None
+            for value in (
+                learning.snapshot.title,
+                learning.snapshot.content,
+                learning.snapshot.trigger,
+                learning.snapshot.rationale,
+            )
+        )
+        if snapshot_bytes > MAX_RETRIEVED_LEARNING_SNAPSHOT_BYTES:
+            raise ValueError(
+                "a publish request may carry at most 10 MiB of retrieved-learning "
+                f"snapshot text (got {snapshot_bytes} bytes)"
+            )
         return self
 
 

reflexio/server/services/generation_service.py

@@ -57,6 +57,10 @@
     delete_count_for_retention,
     get_row_retention_limits,
 )
+from reflexio.server.services.storage.retention_archive import (
+    RETENTION_ARCHIVE_DELETE_BATCH,
+    retention_archive_enabled,
+)
 from reflexio.server.services.tagging.tagging_scheduler import schedule_tagging
 from reflexio.server.tracing import sentry_tags
 from reflexio.server.usage_metrics import record_usage_event
@@ -1327,10 +1331,20 @@ def _cleanup_retention_target(self, target_name: str, limit: int) -> None:
         if total_count < limit:
             return
         delete_count = delete_count_for_retention(total_count)
-        deleted = self.storage.delete_oldest_retention_target_rows(  # type: ignore[reportOptionalMemberAccess]
-            target_name,
-            delete_count,
+        batch_size = (
+            RETENTION_ARCHIVE_DELETE_BATCH
+            if retention_archive_enabled()
+            else delete_count
         )
+        deleted = 0
+        while deleted < delete_count:
+            batch_deleted = self.storage.delete_oldest_retention_target_rows(  # type: ignore[reportOptionalMemberAccess]
+                target_name,
+                min(batch_size, delete_count - deleted),
+            )
+            deleted += batch_deleted
+            if batch_deleted == 0:
+                break
         logger.info(
             "Cleaned up %d oldest %s row(s) (total was %d, limit %d)",
             deleted,

reflexio/server/services/storage/retention_archive.py

@@ -0,0 +1,136 @@
+"""Append-only JSONL sink for rows removed by row-count retention."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+DEFAULT_RETENTION_ARCHIVE_WARN_BYTES = 10 * 1024**3
+RETENTION_ARCHIVE_DELETE_BATCH = 1_000
+logger = logging.getLogger(__name__)
+_warned_archive_dirs: set[Path] = set()
+
+
+def retention_archive_enabled() -> bool:
+    """Return whether archive-before-delete retention is enabled."""
+    return os.environ.get("REFLEXIO_RETENTION_ARCHIVE", "").lower() in {
+        "1",
+        "true",
+    }
+
+
+def resolve_archive_directory(database_path: str) -> Path:
+    """Resolve the archive directory from the override or SQLite DB path.
+
+    Args:
+        database_path: Path to the active SQLite database.
+
+    Returns:
+        Directory that owns the per-table JSONL archive files.
+    """
+    override = os.environ.get("REFLEXIO_RETENTION_ARCHIVE_DIR")
+    if override:
+        return Path(override).expanduser()
+    return Path(database_path).expanduser().parent / "archive"
+
+
+def retention_archive_warning_bytes() -> int:
+    """Return the total archive-directory warning threshold in bytes.
+
+    Returns:
+        Positive warning threshold. Defaults to 10 GiB.
+
+    Invalid values log an error and fall back to 10 GiB so observability
+    configuration can never stop live-row retention.
+    """
+    raw = os.environ.get("REFLEXIO_RETENTION_ARCHIVE_WARN_BYTES")
+    if raw is None or not raw.strip():
+        return DEFAULT_RETENTION_ARCHIVE_WARN_BYTES
+    try:
+        value = int(raw)
+    except ValueError:
+        logger.error(
+            "Invalid REFLEXIO_RETENTION_ARCHIVE_WARN_BYTES=%r; using %d",
+            raw,
+            DEFAULT_RETENTION_ARCHIVE_WARN_BYTES,
+        )
+        return DEFAULT_RETENTION_ARCHIVE_WARN_BYTES
+    if value <= 0:
+        logger.error(
+            "Invalid REFLEXIO_RETENTION_ARCHIVE_WARN_BYTES=%r; using %d",
+            raw,
+            DEFAULT_RETENTION_ARCHIVE_WARN_BYTES,
+        )
+        return DEFAULT_RETENTION_ARCHIVE_WARN_BYTES
+    return value
+
+
+def archive_size_bytes(archive_dir: Path) -> int:
+    """Return total bytes currently used by JSONL archive files.
+
+    Args:
+        archive_dir: Directory containing per-table JSONL files.
+
+    Returns:
+        Sum of all ``*.jsonl`` file sizes, or zero before first write.
+    """
+    if not archive_dir.is_dir():
+        return 0
+    return sum(path.stat().st_size for path in archive_dir.glob("*.jsonl"))
+
+
+def append_archive_rows(
+    archive_dir: Path, table_name: str, rows: list[dict[str, Any]]
+) -> None:
+    """Append sanitized rows to ``table_name``'s archive file.
+
+    Each JSON line is emitted with one ``write`` call. Values unsupported by
+    the JSON encoder fall back to their ``repr`` while embeddings are omitted.
+
+    Args:
+        archive_dir: Directory containing per-table JSONL files.
+        table_name: Physical table being archived.
+        rows: Complete database rows selected for retention removal.
+
+    The warning threshold is observability only. Crossing it never blocks
+    live-row retention.
+    """
+    if not rows:
+        return
+    archive_dir.mkdir(parents=True, exist_ok=True)
+    archived_at = int(time.time())
+    warning_bytes = retention_archive_warning_bytes()
+    projected_bytes = archive_size_bytes(archive_dir)
+    resolved_archive_dir = archive_dir.resolve()
+    if projected_bytes <= warning_bytes:
+        _warned_archive_dirs.discard(resolved_archive_dir)
+    path = archive_dir / f"{table_name}.jsonl"
+    with path.open("ab") as archive_file:
+        for row in rows:
+            sanitized = {key: value for key, value in row.items() if key != "embedding"}
+            record = {
+                "table": table_name,
+                "archived_at": archived_at,
+                "row": sanitized,
+            }
+            encoded_line = (
+                json.dumps(record, default=repr, separators=(",", ":")) + "\n"
+            ).encode("utf-8")
+            archive_file.write(encoded_line)
+            projected_bytes += len(encoded_line)
+    if (
+        projected_bytes > warning_bytes
+        and resolved_archive_dir not in _warned_archive_dirs
+    ):
+        logger.error(
+            "Retention archive exceeded its warning threshold: directory=%s "
+            "size_bytes=%d warning_bytes=%d; live-row retention continues",
+            archive_dir,
+            projected_bytes,
+            warning_bytes,
+        )
+        _warned_archive_dirs.add(resolved_archive_dir)

reflexio/server/services/storage/retention_mixin.py

@@ -9,14 +9,24 @@
 
 from __future__ import annotations
 
+import logging
 from abc import ABC, abstractmethod
 from collections.abc import Iterator, Sequence
+from contextlib import AbstractContextManager, nullcontext
+from pathlib import Path
 from typing import Any
 
 from reflexio.server.services.storage.retention import (
+    RETENTION_CASCADES,
     RETENTION_TARGETS_BY_NAME,
     RetentionTarget,
 )
+from reflexio.server.services.storage.retention_archive import (
+    append_archive_rows,
+    retention_archive_enabled,
+)
+
+logger = logging.getLogger(__name__)
 
 # Conservative chunk size for IN-list deletes. Picked to stay well under:
 #   - SQLite's SQLITE_MAX_VARIABLE_NUMBER (999 on builds before 3.32; 32766 after).
@@ -98,11 +108,41 @@ def delete_oldest_retention_target_rows(self, target_name: str, count: int) -> i
         target = get_retention_target(target_name)
         if not self._retention_table_exists(target.table_name):
             return 0
-        keys = self._retention_select_oldest_keys(target, count)
-        if not keys:
-            return 0
-        self._retention_perform_delete(target, keys)
-        return len(keys)
+        archive_enabled = retention_archive_enabled()
+        guard = self._retention_guard() if archive_enabled else nullcontext()
+        with guard:
+            keys = self._retention_select_oldest_keys(target, count)
+            if not keys:
+                return 0
+            if archive_enabled:
+                try:
+                    archive_dir = self._retention_archive_directory()
+                    parent_ids = [(key[0],) for key in keys]
+                    cascades = RETENTION_CASCADES.get(target.name, ())
+                    if cascades and len(target.id_columns) != 1:
+                        raise AssertionError(
+                            f"Cascade archive target {target.name} must have one key"
+                        )
+                    for cascade in cascades:
+                        rows = self._retention_fetch_rows(
+                            cascade.table_name,
+                            (cascade.fk_column,),
+                            parent_ids,
+                        )
+                        append_archive_rows(archive_dir, cascade.table_name, rows)
+                    rows = self._retention_fetch_rows(
+                        target.table_name, target.id_columns, keys
+                    )
+                    append_archive_rows(archive_dir, target.table_name, rows)
+                except Exception:  # noqa: BLE001 - any archive failure must stop deletion
+                    logger.error(  # noqa: G201 - contract requires an explicit error log
+                        "Failed to archive retention rows for target %s; deletion aborted",
+                        target.name,
+                        exc_info=True,
+                    )
+                    raise
+            self._retention_perform_delete(target, keys)
+            return len(keys)
 
     def _retention_perform_delete(
         self, target: RetentionTarget, keys: list[tuple[Any, ...]]
@@ -118,6 +158,23 @@ def _retention_perform_delete(
 
     # -- Backend hooks --
 
+    def _retention_guard(self) -> AbstractContextManager[Any]:
+        """Guard selection, optional archive writes, and deletion as one unit."""
+        return nullcontext()
+
+    def _retention_archive_directory(self) -> Path:
+        """Return the archive sink directory for this backend."""
+        raise NotImplementedError("Retention archive is unavailable for this backend")
+
+    def _retention_fetch_rows(
+        self,
+        table_name: str,
+        key_columns: tuple[str, ...],
+        keys: list[tuple[Any, ...]],
+    ) -> list[dict[str, Any]]:
+        """Fetch complete rows selected by ``key_columns`` and ``keys``."""
+        raise NotImplementedError("Retention archive is unavailable for this backend")
+
     @abstractmethod
     def _retention_table_exists(self, table_name: str) -> bool:
         """Return whether ``table_name`` exists in the backing store."""