fix(verif): consolidation cadence uses ingested_at instead of wall-clock created_at

Production-relevant bug surfaced by LoCoMo smoke (tasks/e1-v3-locomo-smoke-finding.md):
backfilled memories with backdated created_at (e.g. a 2023 conversation
imported in 2026 wall-clock) trigger immediate gist/tag compression on
the first consolidation pass because (now - created_at) >> 7-day gist
gate. The intended semantics is "memory has had time to be revisited
in MY system" — elapsed since ingest, not elapsed since the original
event.

Changes:
- pg_schema.py: add memories.ingested_at TIMESTAMPTZ NOT NULL DEFAULT
  NOW() in MEMORIES_DDL (fresh DBs) plus an idempotent migration block
  that ALTERs existing tables and backfills ingested_at = created_at
  for legacy rows. Comment kept semicolon-free (df14e16, 9f94bd3).
- pg_store.py: row normalizer surfaces ingested_at as ISO string. INSERT
  inherits DEFAULT NOW() automatically.
- compression.py: cadence gate reads ingested_at (fallback to created_at
  for in-memory dicts that never round-tripped through PG).
- decay_cycle.py: ACT-R lifetime L is ingest-relative; _hours_since_access
  fallback chain now last_accessed -> ingested_at -> created_at.
- write_post_store.py, write_gate.py: synaptic-tagging window and
  temporal-novelty signal both ask "in MY system" — also use ingested_at.
- Tests: regression in test_compression.py (backdated created_at + fresh
  ingested_at must stay level 0), in test_decay_cycle.py (ACT-R does not
  collapse on backfill), and a schema-shape test test_pg_ingested_at.py
  asserting the column declaration, migration guard, backfill statement,
  and semicolon-free comments.

Audit: synaptic_*, microglial_pruning, cascade_* contain no wall-clock
cadence reads (none of them use datetime.now() / created_at). The
PL/pgSQL effective_heat() and recall_memories() functions correctly use
heat_base_set_at / last_accessed (last_accessed -> created_at fallback
is for retrieval recency ranking, which IS event-time semantics — not
cadence — and is left unchanged). replay_execution.py and sleep_compute.py
sort by created_at for narrative/temporal ordering — correct, unchanged.

Verification:
- Full suite: 2656 passed.
- LoCoMo smoke (--with-consolidation, --limit 1): MRR 0.866 (matches
  baseline non-consolidation MRR exactly; pre-fix this collapsed to
  0.222 because immediate compression destroyed retrievable content).

Sources: Anderson & Lebiere (1998) ACT-R Eq. 4.4 — lifetime L is
elapsed time in the learner's system since acquisition, not elapsed
time since the original-source event. Tse et al. (2007) and McClelland
et al. (1995) describe consolidation timescales relative to acquisition
in the system, not relative to the original-event date.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: cdeust

mcp_server/core/compression.py

@@ -31,18 +31,34 @@
 _CAMELCASE_RE = re.compile(r"\b[A-Z][a-z]+(?:[A-Z][a-z]+)+\b")
 
 
-def _parse_created_at(memory: dict) -> datetime | None:
-    """Parse created_at from memory, returning None on failure."""
-    created_at_str = memory.get("created_at", "")
-    if not created_at_str:
-        return None
-    try:
-        created_at = datetime.fromisoformat(created_at_str)
-    except (ValueError, TypeError):
+def _parse_ingested_at(memory: dict) -> datetime | None:
+    """Parse ingest timestamp for cadence reasoning.
+
+    Compression cadence asks "has this memory had time to be revisited
+    in MY system" — that is elapsed time since ingest, NOT elapsed time
+    since the original event. Backfilled / imported memories carry a
+    backdated created_at (e.g. a 2023 conversation imported in 2026);
+    using created_at would compress them on the first consolidation
+    pass, before retrieval ever runs (see tasks/e1-v3-locomo-smoke-finding.md).
+
+    Falls back to created_at for legacy rows that predate the
+    ingested_at column (the schema migration in pg_schema.py backfills
+    ingested_at = created_at in that case anyway, so the fallback only
+    matters for in-memory dicts that never round-tripped through PG).
+    """
+    raw = memory.get("ingested_at") or memory.get("created_at", "")
+    if not raw:
         return None
-    if created_at.tzinfo is None:
-        created_at = created_at.replace(tzinfo=timezone.utc)
-    return created_at
+    if isinstance(raw, datetime):
+        dt = raw
+    else:
+        try:
+            dt = datetime.fromisoformat(raw)
+        except (ValueError, TypeError):
+            return None
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt
 
 
 def _compute_resistance(memory: dict) -> float:
@@ -74,11 +90,13 @@ def get_compression_schedule(
     if memory.get("store_type", "episodic") == "semantic":
         return 0
 
-    created_at = _parse_created_at(memory)
-    if created_at is None:
+    ingested_at = _parse_ingested_at(memory)
+    if ingested_at is None:
         return 0
 
-    hours_elapsed = (datetime.now(timezone.utc) - created_at).total_seconds() / 3600.0
+    # Cadence is measured from ingest, not from the original event.
+    # Source: tasks/e1-v3-locomo-smoke-finding.md.
+    hours_elapsed = (datetime.now(timezone.utc) - ingested_at).total_seconds() / 3600.0
     resistance = _compute_resistance(memory)
 
     if hours_elapsed < gist_age_hours * resistance:

mcp_server/core/decay_cycle.py

@@ -75,8 +75,19 @@ def _parse_datetime(value) -> datetime | None:
 
 
 def _hours_since_access(mem: dict, now: datetime) -> float | None:
-    """Return hours elapsed since last access, or None if unparseable."""
-    last_accessed = mem.get("last_accessed", mem.get("created_at", ""))
+    """Return hours elapsed since last access, or None if unparseable.
+
+    Fallback chain: last_accessed → ingested_at → created_at. The
+    ingested_at fallback (added for the consolidation-cadence fix —
+    tasks/e1-v3-locomo-smoke-finding.md) ensures backfilled memories
+    with a backdated created_at do not falsely register as
+    "last-accessed years ago" when last_accessed is missing.
+    """
+    last_accessed = (
+        mem.get("last_accessed")
+        or mem.get("ingested_at")
+        or mem.get("created_at", "")
+    )
     last_dt = _parse_datetime(last_accessed)
     if last_dt is None:
         return None
@@ -85,12 +96,25 @@ def _hours_since_access(mem: dict, now: datetime) -> float | None:
 
 
 def _hours_since_creation(mem: dict, now: datetime) -> float | None:
-    """Return hours elapsed since creation."""
-    created = mem.get("created_at", "")
-    created_dt = _parse_datetime(created)
-    if created_dt is None:
+    """Return hours elapsed since the memory entered THIS system.
+
+    ACT-R "lifetime L" in B_i = ln(n) − d·ln(L) is lifetime in the
+    learner's system since acquisition (Anderson & Lebiere 1998), not
+    elapsed time since the original-source event. For Cortex this is
+    ``ingested_at``: backfilled / imported memories with backdated
+    ``created_at`` would otherwise return a wrongly-large L on first
+    decay pass and collapse heat to near-zero before any access.
+    Source: tasks/e1-v3-locomo-smoke-finding.md.
+
+    Falls back to ``created_at`` only for in-memory dicts that never
+    round-tripped through PG (schema backfills ingested_at=created_at
+    for legacy rows, so PG-sourced dicts always have the field).
+    """
+    raw = mem.get("ingested_at") or mem.get("created_at", "")
+    ingested_dt = _parse_datetime(raw)
+    if ingested_dt is None:
         return None
-    hours = (now - created_dt).total_seconds() / 3600.0
+    hours = (now - ingested_dt).total_seconds() / 3600.0
     return max(_MIN_LIFETIME_HOURS, hours)
 
 
@@ -263,8 +287,18 @@ def compute_decay_updates(
 
 
 def _parse_hours_since_access(record: dict, now: datetime) -> float | None:
-    """Parse hours since last access from a memory or entity record."""
-    last_accessed = record.get("last_accessed", record.get("created_at", ""))
+    """Parse hours since last access from a memory or entity record.
+
+    Fallback chain mirrors _hours_since_access: last_accessed →
+    ingested_at → created_at. Entities do not currently carry
+    ingested_at, so this is a no-op for entity records but defensive
+    for memory dicts that take this path.
+    """
+    last_accessed = (
+        record.get("last_accessed")
+        or record.get("ingested_at")
+        or record.get("created_at", "")
+    )
     last_dt = _parse_datetime(last_accessed)
     if last_dt is None:
         return None

mcp_server/core/write_gate.py

@@ -78,8 +78,14 @@ def compute_temporal_novelty(
         best_idx = similarities.index(max(similarities))
         if best_idx < len(vec_hits):
             best_mem = get_memory(vec_hits[best_idx][0])
-            if best_mem and best_mem.get("created_at"):
-                hours = _parse_hours_since(best_mem["created_at"])
+            # Temporal novelty asks "have we seen this in MY system
+            # recently?" — that is elapsed-since-ingest, not
+            # elapsed-since-the-original-event. Use ingested_at and fall
+            # back to created_at for legacy rows.
+            # Source: tasks/e1-v3-locomo-smoke-finding.md.
+            if best_mem and (best_mem.get("ingested_at") or best_mem.get("created_at")):
+                ts = best_mem.get("ingested_at") or best_mem["created_at"]
+                hours = _parse_hours_since(ts)
     return _compute_temporal_novelty(hours)
 
 

mcp_server/core/write_post_store.py

@@ -131,7 +131,14 @@ def _build_tagging_candidates(
         if mem["id"] == exclude_id:
             continue
         mem_ents = _find_shared_entities(mem["id"], entity_names, store)
-        hours_ago = _hours_since_creation(mem.get("created_at", ""))
+        # Synaptic-tagging window is cadence-relative to ingest time, not
+        # original-event time. For backfilled memories with a backdated
+        # created_at this prevents the tagging window from collapsing
+        # immediately on first consolidation pass.
+        # Source: tasks/e1-v3-locomo-smoke-finding.md.
+        hours_ago = _hours_since_creation(
+            mem.get("ingested_at") or mem.get("created_at", "")
+        )
         candidates.append(
             {
                 "id": mem["id"],

mcp_server/infrastructure/pg_schema.py

@@ -28,6 +28,7 @@
     domain          TEXT DEFAULT '',
     directory_context TEXT DEFAULT '',
     created_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    ingested_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
     last_accessed   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
     heat_base       REAL NOT NULL DEFAULT 1.0
                     CHECK (heat_base >= 0.0 AND heat_base <= 1.0),
@@ -1283,6 +1284,33 @@
     END IF;
 END $$;
 
+-- Migration: add ingested_at for consolidation cadence reasoning.
+-- Source: tasks/e1-v3-locomo-smoke-finding.md.
+-- created_at = original event/utterance time (may be backdated on import).
+-- ingested_at = when the row entered THIS Cortex DB (always NOW at insert).
+-- Compression and decay cadence MUST use ingested_at: the mechanism asks
+-- "has this memory had time to be revisited in MY system" not "when did
+-- the original event happen". Backfill existing rows from created_at to
+-- preserve idempotency and pre-existing semantics.
+-- NOTE keep this comment free of semicolons (DDL is split on the literal
+-- character per _split_statements — df14e16 and 9f94bd3 are prior
+-- incidents of that class).
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_name = 'memories' AND column_name = 'ingested_at'
+    ) THEN
+        ALTER TABLE memories ADD COLUMN ingested_at TIMESTAMPTZ NOT NULL DEFAULT NOW();
+        -- Backfill rows that pre-existed this column. They were created
+        -- before ingested_at was tracked, so the safest assumption is
+        -- ingested_at = created_at (i.e., they entered the system at the
+        -- time their content was authored). This block runs only inside
+        -- the IF NOT EXISTS guard, so it is naturally idempotent.
+        UPDATE memories SET ingested_at = created_at;
+    END IF;
+END $$;
+
 -- Migration: persist arousal and dominant_emotion from emotional tagging
 DO $$ BEGIN
     IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name='memories' AND column_name='arousal')

mcp_server/infrastructure/pg_store.py

@@ -768,7 +768,7 @@ def _normalize_memory_row(self, row: dict[str, Any]) -> dict[str, Any]:
             except (json.JSONDecodeError, TypeError):
                 d["tags"] = []
         # Convert datetime to ISO string for compatibility
-        for field in ("created_at", "last_accessed", "last_reconsolidated"):
+        for field in ("created_at", "ingested_at", "last_accessed", "last_reconsolidated"):
             if isinstance(d.get(field), datetime):
                 d[field] = d[field].isoformat()
         return d

tests_py/core/test_compression.py

@@ -60,6 +60,56 @@ def test_no_created_at(self):
     def test_invalid_timestamp(self):
         assert get_compression_schedule({"created_at": "invalid"}) == 0
 
+    # ── Cadence relative to ingest, not original event ──────────────────
+    # Source: tasks/e1-v3-locomo-smoke-finding.md.
+
+    def test_backdated_created_at_with_fresh_ingest_stays_level_zero(self):
+        """Regression: backfilled memories must NOT compress immediately.
+
+        A user imports a 2023 conversation in 2026. created_at is the
+        original event time (years ago). ingested_at is now. The cadence
+        gate must read ingested_at — otherwise the memory gist-compresses
+        on the first consolidation pass before retrieval ever runs
+        (LoCoMo MRR collapse from 0.866 to 0.222 surfaced this bug).
+        """
+        years_ago = (datetime.now(timezone.utc) - timedelta(days=365 * 3)).isoformat()
+        now = datetime.now(timezone.utc).isoformat()
+        mem = {
+            "created_at": years_ago,
+            "ingested_at": now,
+            "importance": 0.3,
+            "store_type": "episodic",
+        }
+        # Fresh ingest → level 0 even though created_at says 3 years old.
+        assert get_compression_schedule(mem) == 0
+
+    def test_fresh_created_at_old_ingest_compresses(self):
+        """Inverse: an old ingest with a recent created_at still compresses.
+
+        Demonstrates that ingested_at — not created_at — is the cadence
+        signal. (Pathological dict; real INSERTs always set ingested_at
+        to NOW(), but the contract must hold.)
+        """
+        now = datetime.now(timezone.utc).isoformat()
+        sixty_days_ago = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat()
+        mem = {
+            "created_at": now,
+            "ingested_at": sixty_days_ago,
+            "importance": 0.3,
+            "store_type": "episodic",
+        }
+        assert get_compression_schedule(mem) == 2
+
+    def test_missing_ingested_at_falls_back_to_created_at(self):
+        """Legacy rows (pre-migration) lack ingested_at → fall back."""
+        sixty_days_ago = (datetime.now(timezone.utc) - timedelta(days=60)).isoformat()
+        mem = {
+            "created_at": sixty_days_ago,
+            "importance": 0.3,
+            "store_type": "episodic",
+        }
+        assert get_compression_schedule(mem) == 2
+
 
 class TestExtractGist:
     def test_short_content_preserved(self):

tests_py/core/test_decay_cycle.py

@@ -140,3 +140,66 @@ def test_cold_entities_skipped(self):
         entities = [{"id": 1, "heat": 0.01, "last_accessed": now.isoformat()}]
         updates = compute_entity_decay(entities, now=now, cold_threshold=0.05)
         assert len(updates) == 0
+
+
+class TestIngestRelativeCadence:
+    """Regression: cadence reasoning uses ingested_at, not created_at.
+
+    Source: tasks/e1-v3-locomo-smoke-finding.md.
+    """
+
+    def test_adaptive_decay_uses_ingested_at_not_created_at(self):
+        """Backfilled memory with backdated created_at must NOT collapse.
+
+        ACT-R lifetime L = elapsed time since acquisition by THIS system.
+        For a 3-year-old created_at but a fresh ingested_at, L should be
+        ~hours (just-ingested), not ~3-years. Otherwise the ACT-R base
+        level B = ln(n) - 0.5*ln(L) sinks far below threshold and heat
+        collapses to near-zero on first decay pass.
+        """
+        now = datetime.now(timezone.utc)
+        years_ago = (now - timedelta(days=365 * 3)).isoformat()
+        # Just-ingested 5 minutes ago.
+        fresh_ingest = (now - timedelta(minutes=5)).isoformat()
+        mems = [
+            {
+                "id": 1,
+                "heat": 0.8,
+                "created_at": years_ago,
+                "ingested_at": fresh_ingest,
+                "last_accessed": fresh_ingest,
+                "access_count": 1,
+                "importance": 0.5,
+                "emotional_valence": 0.0,
+                "confidence": 1.0,
+            }
+        ]
+        updates = compute_decay_updates(mems, now=now, adaptive_decay=True)
+        # If cadence used created_at, lifetime would be ~26 280 hours and
+        # heat would collapse to near-zero (<<0.1). With ingested_at it
+        # stays close to its initial 0.8.
+        if updates:
+            new_heat = updates[0][1]
+            assert new_heat > 0.4, (
+                f"freshly-ingested backdated memory collapsed to {new_heat}"
+            )
+
+    def test_legacy_dict_falls_back_to_created_at(self):
+        """Pre-migration rows (no ingested_at) keep the old behaviour."""
+        now = datetime.now(timezone.utc)
+        old = (now - timedelta(hours=48)).isoformat()
+        mems = [
+            {
+                "id": 1,
+                "heat": 1.0,
+                "last_accessed": old,
+                "created_at": old,
+                "importance": 0.3,
+                "emotional_valence": 0.0,
+                "confidence": 1.0,
+            }
+        ]
+        updates = compute_decay_updates(mems, now=now)
+        # Decays normally — fallback chain kept intact.
+        assert len(updates) == 1
+        assert updates[0][1] < 1.0