feat(storage): preserve retention evidence for later evaluation

README.md

@@ -294,6 +294,24 @@ 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_MAX_BYTES=10737418240` | Bound completed JSONL segments to this total size (10 GiB by default). |
+
+Each archive-enabled SQLite transaction processes at most 1,000 parent rows.
+One JSONL segment contains that target batch and its cascade-deleted rows. After
+the newest segment is complete, Reflexio removes the oldest whole segments until
+the archive is back under its ceiling (FIFO). A single batch larger than the
+ceiling, or any archive I/O failure, is logged as evidence loss; live retention
+still continues. The local archive is implemented for SQLite.
+
 ## 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,133 @@
+"""Bounded FIFO JSONL archive for rows removed by row-count retention."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+DEFAULT_RETENTION_ARCHIVE_MAX_BYTES = 10 * 1024**3
+RETENTION_ARCHIVE_DELETE_BATCH = 1_000
+logger = logging.getLogger(__name__)
+
+
+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:
+    """Return the configured archive directory or the SQLite-local default."""
+    override = os.environ.get("REFLEXIO_RETENTION_ARCHIVE_DIR")
+    if override:
+        return Path(override).expanduser()
+    return Path(database_path).expanduser().parent / "archive"
+
+
+def retention_archive_max_bytes() -> int:
+    """Return the positive archive ceiling, defaulting to 10 GiB."""
+    raw = os.environ.get("REFLEXIO_RETENTION_ARCHIVE_MAX_BYTES")
+    if raw is None or not raw.strip():
+        return DEFAULT_RETENTION_ARCHIVE_MAX_BYTES
+    try:
+        value = int(raw)
+    except ValueError:
+        value = 0
+    if value <= 0:
+        logger.error(
+            "Invalid REFLEXIO_RETENTION_ARCHIVE_MAX_BYTES=%r; using %d",
+            raw,
+            DEFAULT_RETENTION_ARCHIVE_MAX_BYTES,
+        )
+        return DEFAULT_RETENTION_ARCHIVE_MAX_BYTES
+    return value
+
+
+def _encode_records(rows_by_table: Mapping[str, list[dict[str, Any]]]) -> bytes:
+    archived_at = int(time.time())
+    lines = []
+    for table_name, rows in rows_by_table.items():
+        for row in rows:
+            record = {
+                "table": table_name,
+                "archived_at": archived_at,
+                "row": {key: value for key, value in row.items() if key != "embedding"},
+            }
+            lines.append(json.dumps(record, default=repr, separators=(",", ":")))
+    return ("\n".join(lines) + "\n").encode("utf-8") if lines else b""
+
+
+def append_archive_batch(
+    archive_dir: Path, rows_by_table: Mapping[str, list[dict[str, Any]]]
+) -> bool:
+    """Append one retention batch and evict oldest segments to stay bounded.
+
+    One segment contains the target rows and any cascade-deleted rows, so FIFO
+    eviction never splits a retention batch. The completed newest segment is
+    installed before old segments are removed; a crash can temporarily exceed
+    the ceiling but cannot replace newest evidence with older evidence.
+
+    Returns:
+        True when the batch was archived. False only when one batch is itself
+        larger than the configured ceiling.
+    """
+    encoded = _encode_records(rows_by_table)
+    if not encoded:
+        return True
+    max_bytes = retention_archive_max_bytes()
+    row_count = sum(len(rows) for rows in rows_by_table.values())
+    if len(encoded) > max_bytes:
+        logger.error(
+            "Retention archive batch exceeds the archive ceiling; skipping evidence "
+            "while live-row retention continues: rows_skipped=%d batch_bytes=%d "
+            "ceiling_bytes=%d",
+            row_count,
+            len(encoded),
+            max_bytes,
+        )
+        return False
+
+    archive_dir.mkdir(parents=True, exist_ok=True)
+    segment = archive_dir / f"{time.time_ns():020d}-{uuid4().hex}.jsonl"
+    temporary = segment.with_suffix(".tmp")
+    try:
+        temporary.write_bytes(encoded)
+        temporary.replace(segment)
+    finally:
+        temporary.unlink(missing_ok=True)
+
+    segments = sorted(
+        archive_dir.glob("*.jsonl"),
+        key=lambda path: (path.stat().st_mtime_ns, path.name),
+    )
+    total_bytes = sum(path.stat().st_size for path in segments)
+    evicted_files = 0
+    evicted_bytes = 0
+    for oldest in segments:
+        if total_bytes <= max_bytes:
+            break
+        if oldest == segment:
+            continue
+        size = oldest.stat().st_size
+        oldest.unlink()
+        total_bytes -= size
+        evicted_files += 1
+        evicted_bytes += size
+    if evicted_files:
+        logger.info(
+            "Retention archive FIFO evicted oldest evidence: files=%d bytes=%d "
+            "size_bytes=%d ceiling_bytes=%d",
+            evicted_files,
+            evicted_bytes,
+            total_bytes,
+            max_bytes,
+        )
+    return True

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_batch,
+    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,42 @@ 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, ())
+                    rows_by_table: dict[str, list[dict[str, Any]]] = {}
+                    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,
+                        )
+                        rows_by_table[cascade.table_name] = rows
+                    rows_by_table[target.table_name] = self._retention_fetch_rows(
+                        target.table_name, target.id_columns, keys
+                    )
+                    append_archive_batch(archive_dir, rows_by_table)
+                except Exception:  # noqa: BLE001 - archiving must not stop retention
+                    logger.error(  # noqa: G201 - contract requires an explicit error log
+                        "Failed to archive retention rows for target %s; evidence was "
+                        "not preserved, but live-row retention continues",
+                        target.name,
+                        exc_info=True,
+                    )
+            self._retention_perform_delete(target, keys)
+            return len(keys)
 
     def _retention_perform_delete(
         self, target: RetentionTarget, keys: list[tuple[Any, ...]]
@@ -118,6 +159,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."""