fix(forgetting): a restored episode keeps the time it happened

alembic/versions/20261103_1200_a3_archived_original_created_at.py

@@ -0,0 +1,44 @@
+"""a3 — the bin remembers when an archived row happened.
+
+`archived_memories` kept only `archived_at`, the moment a row was removed, so a
+restored episode was stamped with the moment of its return: it never aged, never
+met the purge, and sorted to the end of the L4 queue (`created_at` order).
+
+`ArchivedMemories.ensure()` adds this column on bases that never run migrations —
+it is called from constructors that do not migrate. This revision puts the same
+column in the chain so `alembic upgrade head` alone yields the schema the code
+expects.
+
+SQLite's ADD COLUMN is a schema-only change: no existing row is rewritten, and
+every old row reads NULL (the time was never recorded, so NULL is the truth, not
+a gap). The existence guard makes it a no-op when `ensure()` reached the base
+first — a script or autohook can have run it before the server migrated.
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "20261103_1200_a3"
+down_revision: str | Sequence[str] | None = "20261004_2200_g27"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _column_exists(conn: sqlalchemy.Connection, table: str, column: str) -> bool:
+    rows = conn.execute(sqlalchemy.text(f"PRAGMA table_info({table})")).fetchall()
+    return any(row[1] == column for row in rows)
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if not _column_exists(conn, "archived_memories", "original_created_at"):
+        op.execute("ALTER TABLE archived_memories ADD COLUMN original_created_at REAL")
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if _column_exists(conn, "archived_memories", "original_created_at"):
+        op.execute("ALTER TABLE archived_memories DROP COLUMN original_created_at")

core/episodic.py

@@ -201,6 +201,10 @@ async def _run(conn: Any) -> int:
                     importance=float(row["emotional_weight"]),
                     original_id=int(row["episode_id"]),
                     reason=f"inactive_{days}d",
+                    # The row is already in hand and carries the moment it happened.
+                    # Without it the bin keeps only `archived_at`, and a restore
+                    # could not tell when this episode belonged to.
+                    original_created_at=float(row["created_at"]),
                 )
                 archived_count += 1
 

features/compression.py

@@ -59,7 +59,7 @@ async def compress_episodes(self, user_id: str, min_weight: float = 0.3, layer:
         params = (user_id, layer, min_weight, cutoff)
         doomed = await self._dispatch.query(
             DB_NAME,
-            "SELECT episode_id, summary, emotional_weight FROM episodes WHERE user_id=? AND layer=? AND emotional_weight < ? AND created_at < ?",
+            "SELECT episode_id, summary, emotional_weight, created_at FROM episodes WHERE user_id=? AND layer=? AND emotional_weight < ? AND created_at < ?",
             params,
         )
         if not doomed:
@@ -78,6 +78,7 @@ async def compress_episodes(self, user_id: str, min_weight: float = 0.3, layer:
                 importance=float(r["emotional_weight"]),
                 original_id=int(r["episode_id"]),
                 reason="compression_low_weight",
+                original_created_at=float(r["created_at"]),
             )
             with contextlib.suppress(Exception):
                 from lifecycle.transitions import record_transition

lifecycle/forgetting.py

@@ -334,11 +334,13 @@ async def restore_entries(self, ids: list[int], user_id: str, memory_type: str |
         is the argument that resolves it rather than a contradiction of it.
 
         WHAT IS LOST. `layer` is not in the archive and neither are an episode's
-        `tags`, `created_at` or its layer, so a restored episode lands in this
-        system's own layer and carries the moment of its return. Data the archive
-        never kept cannot be restored, and this method does not pretend it can.
-        The rule is settled rather than pending: separating that metadata is
-        separate work, deliberately not done here.
+        `tags`, so a restored episode lands in this system's own layer. Its
+        `created_at` is now kept — in `original_created_at` — but only for rows
+        archived after the bin started recording it; an older row still carries
+        the moment of its return, and `archived_at` is not used as a stand-in for a
+        time nobody recorded. Data the archive never kept cannot be restored, and
+        this method does not pretend it can. The rule is settled rather than
+        pending: separating the rest is separate work, deliberately not done here.
         """
         from shared.archived_memories import (
             REFUSAL_ALREADY_PRESENT,
@@ -520,6 +522,10 @@ async def _restore_episode(self, row: Any, user_id: str) -> int:
         whose summary was re-added is a no-op on the row count. That is counted as
         restored — the text is in the table, which is what the caller asked for —
         and it is a deliberate reading rather than an accident of the dedup.
+
+        The episode is given back with the time it happened when the bin recorded
+        it (`original_created_at`). A row archived before that column existed has
+        no time to give back, so it is stamped with the moment of its return.
         """
         from core.episodic import EpisodicMemory
 
@@ -533,7 +539,22 @@ async def _restore_episode(self, row: Any, user_id: str) -> int:
         # `forget_primitive_fuzzy_l3`, so the old form would have restored that episode
         # at 0.5. There is no single base to measure — the archive is per instance.
         weight = 0.5 if row["importance"] is None else float(row["importance"])
-        return await mem.save(user_id, str(row["content"]), emotional_weight=weight)
+        # The bin now carries the moment the episode happened. It is absent on rows
+        # archived before the column existed (3159 of them on one live base,
+        # measured 2026-11-03), and `archived_at` is not a substitute: it is when
+        # the row was removed, not when it happened. So a missing value keeps the
+        # old behaviour and the episode is stamped with its return.
+        #
+        # The key check is for a base whose bin predates the column and where
+        # `ensure()` has not run in this process yet — `restore_entries` reads the
+        # table directly and must not crash on a schema it can still serve.
+        created_at = None
+        # `in row` would test VALUES, not column names on sqlite3.Row, so the key
+        # check has to go through `.keys()` (SIM118 does not apply to this type).
+        if "original_created_at" in row.keys():  # noqa: SIM118
+            recorded = row["original_created_at"]
+            created_at = None if recorded is None else float(recorded)
+        return await mem.save(user_id, str(row["content"]), emotional_weight=weight, created_at=created_at)
 
     async def _restore_core(self, conn: Any, row: Any, original_id: int, now: float) -> None:
         """Put an archived L4 row back into `core_memory`.

mcp_server/tools/primitives/forget.py

@@ -274,6 +274,7 @@ async def _forget_episode_ids(app: AppContext, episode_ids: list[int], user_id:
                 importance=episode.emotional_weight,
                 original_id=episode.episode_id,
                 reason="forget_primitive_episode_ids",
+                original_created_at=episode.created_at,
             )
     outcome.deleted_l3 = int(await mem.l3.delete_by_ids([e.episode_id for e in found], user_id=user_id))
     return outcome
@@ -296,6 +297,7 @@ async def _delete_matching_episodes(
                 importance=episode.emotional_weight,
                 original_id=episode.episode_id,
                 reason="forget_primitive_fuzzy_l3",
+                original_created_at=episode.created_at,
             )
     return int(await mem.l3.delete_by_ids([e.episode_id for e in episodes], user_id=user_id)), hits
 

scripts/dedup_episode_twins.py

@@ -134,10 +134,13 @@ async def run(apply: bool) -> int:
         drop: list[dict] = []
         klass = Counter()
         for members in groups.values():
-            k, d, name = classify(members)
+            # Named apart from `keep`/`drop` and from the loop variables below:
+            # unpacking into `k, d` bound `d` to `list[dict]`, so every later
+            # `for d in drop` looked like a dict assigned to a list.
+            group_keep, group_drop, name = classify(members)
             klass[name] += 1
-            keep.extend(k)
-            drop.extend(d)
+            keep.extend(group_keep)
+            drop.extend(group_drop)
 
         before_total = await dispatch.scalar(DB_NAME, "SELECT COUNT(*) FROM episodes") or 0
         orphans_sql = (
@@ -206,6 +209,7 @@ async def run(apply: bool) -> int:
                 importance=float(d["emotional_weight"] or 0.0),
                 original_id=int(d["episode_id"]),
                 reason="duplicate_episode_replay",
+                original_created_at=float(d["created_at"]),
             )
             with contextlib.suppress(Exception):
                 from lifecycle.transitions import record_transition

scripts/purge_foreign_rows.py

@@ -117,7 +117,7 @@ def plan(db: Path) -> dict[str, Any]:
         # --- episodes: via the raw:<l0_id> provenance tag ---
         foreign_eps: list[sqlite3.Row] = []
         keep_summaries: set[str] = set()
-        for row in conn.execute("SELECT episode_id, summary, layer, user_id, emotional_weight, tags FROM episodes"):
+        for row in conn.execute("SELECT episode_id, summary, layer, user_id, emotional_weight, tags, created_at FROM episodes"):
             raws = _raw_tags(row["tags"])
             if raws and any(r in marked for r in raws):
                 foreign_eps.append(row)
@@ -220,6 +220,7 @@ async def apply(p: dict[str, Any], db: Path) -> None:
             importance=float(r["emotional_weight"] or 0.0),
             original_id=int(r["episode_id"]),
             reason=ARCHIVE_REASON,
+            original_created_at=float(r["created_at"]),
         )
         archived += 1
     for r in p["facts"]:

scripts/purge_speaker_layer_leaks.py

@@ -102,7 +102,7 @@ def plan(db: Path) -> dict[str, Any]:
         # every legitimate agent episode (6030 of them) and would delete the whole
         # agent layer instead of the leak.
         doomed_eps: list[sqlite3.Row] = []
-        for row in conn.execute("SELECT episode_id, summary, tags, user_id, emotional_weight FROM episodes WHERE layer='user'"):
+        for row in conn.execute("SELECT episode_id, summary, tags, user_id, emotional_weight, created_at FROM episodes WHERE layer='user'"):
             try:
                 tags = json.loads(row["tags"]) if row["tags"] else []
             except (TypeError, ValueError):
@@ -227,6 +227,7 @@ async def _audit(target_id: str, details: dict[str, Any]) -> None:
                     importance=float(r["emotional_weight"] or 0.0),
                     original_id=int(r["episode_id"]),
                     reason=ARCHIVE_REASON,
+                    original_created_at=float(r["created_at"]),
                 )
             ph = ",".join("?" * len(ep_ids))
             await conn.execute(f"DELETE FROM episodes WHERE episode_id IN ({ph})", tuple(ep_ids))

shared/archived_memories.py

@@ -1,6 +1,8 @@
 """ArchivedMemories — async archived memory storage."""
 
 from __future__ import annotations
+
+import sqlite3
 from typing import Any
 
 from shared.connection import AsyncConnectionManager, connection_manager
@@ -163,14 +165,48 @@ async def ensure(self) -> None:
         first. Run alone, or with `-p no:randomly`, it failed with
         `no such table: archived_memories`. A test that needs a neighbour to
         pass is not testing the class.
+
+        `original_created_at` is added here and not only in the migration because
+        this method is what a base without migrations gets: constructors call it
+        on bases that never ran `alembic upgrade`. A bin written before the column
+        existed holds 3159 archived episodes on one live base (measured 2026-11-03),
+        and they must survive the ALTER — it is a schema-only change, so no row is
+        rewritten and every old row reads NULL.
         """
         if self._ready:
             return
-        cols = [r["name"] for r in await self._dispatch.query(DB_NAME, "PRAGMA table_info(archived_memories)")]
+        cols = await self._columns()
         if not cols:
             await self._init_db()
+            # Re-read instead of trusting the CREATE: another process may have won
+            # the race and created the table with an older shape. The ALTER below
+            # is what actually guarantees the column.
+            cols = await self._columns()
+        if cols and "original_created_at" not in cols:
+            await self._add_original_created_at()
         self._ready = True
 
+    async def _columns(self) -> list[str]:
+        rows = await self._dispatch.query(DB_NAME, "PRAGMA table_info(archived_memories)")
+        return [r["name"] for r in rows]
+
+    async def _add_original_created_at(self) -> None:
+        """Add the time an archived row happened, once, without touching a row.
+
+        WHY THE EXCEPTION IS TOLERATED AND NOT MATCHED BY TEXT. Two processes can
+        reach `ensure()` on one base at the same moment: both read the old schema,
+        then serialize on SQLite's write lock, and the loser's ALTER finds the
+        column already there. That is the operation having succeeded, so success is
+        decided by re-reading the schema — not by the error message, which would
+        also swallow a real failure that happens to read alike. A genuine failure
+        leaves the column missing and is re-raised.
+        """
+        try:
+            await self._dispatch.write(DB_NAME, "ALTER TABLE archived_memories ADD COLUMN original_created_at REAL")
+        except sqlite3.OperationalError:
+            if "original_created_at" not in await self._columns():
+                raise
+
     async def _init_db(self) -> None:
         await self._dispatch.execute_script(
             DB_NAME,
@@ -181,7 +217,8 @@ async def _init_db(self) -> None:
                 original_id INTEGER, content TEXT NOT NULL,
                 memory_type TEXT, importance REAL,
                 archive_reason TEXT NOT NULL,
-                archived_at DATETIME DEFAULT CURRENT_TIMESTAMP
+                archived_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+                original_created_at REAL
             );
             CREATE INDEX IF NOT EXISTS idx_archived_user ON archived_memories(user_id, archived_at DESC);
         """,
@@ -195,6 +232,7 @@ async def archive(
         importance: float | None = None,
         original_id: int | None = None,
         reason: str = "manual",
+        original_created_at: float | None = None,
     ) -> int:
         """Record a row in the bin.
 
@@ -214,12 +252,20 @@ async def archive(
         repository — `core/memory.py`, `hooks/shared.py`, `importance_scheduler.py` —
         but those are not calls to this method and were counted as though they were
         in an earlier version of this docstring.)
+
+        `original_created_at` is the moment the row *happened*, which only the
+        caller knows: an episode carries it in its own `created_at`. Recording it
+        here is what lets a restore give the row back with its own age instead of
+        stamping it with the moment of its return. `None` is honest for rows that
+        never had a time, and for core rows, whose restore path does not use it.
         """
         await self.ensure()
         return await self._dispatch.insert(
             DB_NAME,
-            "INSERT INTO archived_memories (user_id, original_id, content, memory_type, importance, archive_reason) VALUES (?, ?, ?, ?, ?, ?)",
-            (user_id, original_id, content, memory_type, importance, reason),
+            "INSERT INTO archived_memories "
+            "(user_id, original_id, content, memory_type, importance, archive_reason, original_created_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (user_id, original_id, content, memory_type, importance, reason, original_created_at),
         )
 
     async def get_archived(self, user_id: str = "default", limit: int = 50) -> dict[str, Any]: