perf(consolidate): v3.11 deferred fixes — batching, cache, plasticity, CLS, cascade (#13)

Addresses the five pathologies identified by the agent+genius audit of
darval's performance report (issue #13). Phase order matches the telemetry
patch commit message in 69d81fb.

Phase A — Set-based SQL batching
---------------------------------

Per-row UPDATE/DELETE loops with per-row commits were the dominant cost
of consolidate on a 66K-memory store (Erlang analysis: ~64 ops/s ceiling
imposed by fsync amplification). Added batch methods:

  pg_store.update_memories_heat_batch      (UPDATE FROM UNNEST)
  pg_store.update_entities_heat_batch
  pg_store.archive_entities_batch          (UPDATE WHERE id = ANY)
  pg_store.update_relationships_weight_batch
  pg_store.delete_relationships_batch
  pg_store.insert_stage_transitions_batch  (INSERT FROM UNNEST)

Each mirrors onto SqliteMemoryStore via executemany + single commit.

Callers converted:
  decay.run_decay_cycle          → heat updates in one statement
  decay._decay_entities          → entity heat in one statement
  decay._modulate_domain         → metabolic adjustments in one statement
  plasticity._apply_updates      → 33k LTP/LTD updates in one statement
  pruning._prune_edges           → 32k DELETEs in one statement
  pruning._archive_orphans       → orphan archival in one statement
  cascade.run_cascade_advancement → transition INSERTs in one statement

Erlang predicted 100-500× speedup on the decay stage alone. On
darval's 66K store the decay stage should drop from ~1100 s to 3-10 s.

Phase B — Consolidation-scoped memory cache
--------------------------------------------

`store.get_all_memories_for_decay()` was called 6× per consolidate run
(decay, compression, memify, homeostatic, sleep, emergence). Load once
at handler entry, pass list to each stage:

  run_decay_cycle(store, settings, memories=None)
  run_compression_cycle(store, settings, embeddings, memories=None)
  run_memify_cycle(store, memories=None)
  run_homeostatic_cycle(store, memories=None)
  run_deep_sleep(store, embeddings, memories=None)
  (emergence closure uses the same list inline)

Each stage retains a fallback load when called standalone outside the
handler, preserving backward compatibility.

Phase C — Plasticity co-access starvation
------------------------------------------

Feinstein's audit: `get_hot_memories(limit=50)` sampled 0.5% of a
10k-entity store → 99.95% LTD was distribution collapse, not
plasticity. Fix:

  - raised sample cap 50 → 2000 (_CO_ACCESS_SAMPLE_CAP)
  - reuses the Phase B consolidation-scoped list when present
  - precomputes lowercase entity name index once per run
  - exception handler now surfaces the error (was logger.debug silent)
  - returns co_access_pairs and memories_sampled for diagnostics

Phase D — CLS causal-edge signal gate
--------------------------------------

Feynman's audit: `_discover_causal_edges` hard-capped at 500 episodic
on a 25K store (2% sample), ran O(entities × 500) substring scans,
produced 0 edges by construction. Fix:

  - episodic sample cap 500 → 2000 (_EPISODIC_SAMPLE_CAP)
  - precomputes content_lowered once, then per-name scan is one pass
  - early-exit gate: if fewer than _MIN_ENTITIES_FOR_PC entities
    clear _PC_MIN_OBSERVATIONS mentions, skip the O(E²) PC pass
  - restricts the PC vocabulary to qualifying entities only, so the
    matrix is E_qualifying² not E_all²
  - returns episodic_scanned for diagnostics

Phase E — Cascade heartbeat + payload trim
-------------------------------------------

Feinstein's audit: cascade wrote a heartbeat UPDATE on every scanned
memory (~2000) even when nothing advanced, each with its own commit;
also per-transition INSERT+commit for stage_transitions (503 fsyncs
on darval's run); also returned the full 503-transition list in the
MCP response duplicating queryable DB state. Fix:

  - heartbeat UPDATE skipped when |Δhours| < _HEARTBEAT_SKIP_HOURS (1h)
    — the previous value was pure noise and wasted fsync
  - stage_transitions INSERTs batched into one statement
  - response payload capped: transitions_preview (first 50) + count
  - exception handler surfaces the error (was logger.debug silent)
  - returns scanned / heartbeats_written / heartbeats_skipped
  - SQLite schema gets stage_transitions table + stage_entered_at
    migration so cascade no longer crashes silently on cowork mode

Reviewed by agents: code-reviewer (APPROVE), test-engineer (2188 pass),
genius:erlang (queuing analysis), genius:feinstein (differential),
genius:feynman (independent rederivation).

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

Author: cdeust

mcp_server/handlers/consolidate.py

@@ -127,8 +127,14 @@ async def handler(args: dict[str, Any] | None = None) -> dict[str, Any]:
     embeddings = get_embedding_engine()
     start = time.monotonic()
 
-    stats = _run_cycles(args, store, settings, embeddings)
-    stats = _run_always_cycles(args, store, stats)
+    # Phase B (issue #13): load the full memory list once and thread it
+    # through every stage that needs it, so consolidate does ONE load
+    # instead of 6 (decay, compression, memify, homeostatic, sleep,
+    # emergence). Cheap stages still load ad-hoc for standalone callers.
+    memories = store.get_all_memories_for_decay()
+
+    stats = _run_cycles(args, store, settings, embeddings, memories)
+    stats = _run_always_cycles(args, store, stats, memories)
 
     elapsed_ms = int((time.monotonic() - start) * 1000)
     stats["duration_ms"] = elapsed_ms
@@ -149,28 +155,33 @@ def _run_cycles(
     store: MemoryStore,
     settings: Any,
     embeddings: EmbeddingEngine,
+    memories: list[dict],
 ) -> dict[str, Any]:
-    """Run optional maintenance cycles based on args flags."""
+    """Run optional maintenance cycles based on args flags.
+
+    `memories` is the consolidation-scoped snapshot so stages share one
+    load across the whole run (issue #13).
+    """
     stats: dict[str, Any] = {}
 
     if args.get("decay", True):
-        stats["decay"] = _timed(run_decay_cycle, store, settings)
+        stats["decay"] = _timed(run_decay_cycle, store, settings, memories)
         stats["plasticity"] = _timed(run_plasticity_cycle, store)
         stats["pruning"] = _timed(run_pruning_cycle, store)
 
     if args.get("compress", True):
         stats["compression"] = _timed(
-            run_compression_cycle, store, settings, embeddings
+            run_compression_cycle, store, settings, embeddings, memories
         )
 
     if args.get("cls", True):
         stats["cls"] = _timed(run_cls_cycle, store, settings, embeddings)
 
     if args.get("memify", True):
-        stats["memify"] = _timed(run_memify_cycle, store)
+        stats["memify"] = _timed(run_memify_cycle, store, memories)
 
     if args.get("deep", False):
-        stats["deep_sleep"] = _timed(run_deep_sleep, store, embeddings)
+        stats["deep_sleep"] = _timed(run_deep_sleep, store, embeddings, memories)
 
     return stats
 
@@ -179,17 +190,18 @@ def _run_always_cycles(
     args: dict,
     store: MemoryStore,
     stats: dict[str, Any],
+    memories: list[dict],
 ) -> dict[str, Any]:
     """Run cycles that always execute regardless of flags."""
     stats["cascade"] = _timed(run_cascade_advancement, store)
-    stats["homeostatic"] = _timed(run_homeostatic_cycle, store)
+    stats["homeostatic"] = _timed(run_homeostatic_cycle, store, memories)
 
     if args.get("deep", False):
         stats["transfer"] = _timed(run_two_stage_transfer, store)
 
     def _run_emergence() -> dict[str, Any]:
-        all_mems = store.get_all_memories_for_decay()
-        return emergence_tracker.generate_emergence_report(all_mems) or {}
+        # Uses the consolidation-scoped memory list — no extra load.
+        return emergence_tracker.generate_emergence_report(memories) or {}
 
     stats["emergence"] = _timed(_run_emergence)
 

mcp_server/handlers/consolidation/cascade.py

@@ -17,34 +17,61 @@
 
 _ADVANCEABLE_STAGES = ["labile", "early_ltp", "late_ltp", "reconsolidating"]
 
+# Source: issue #13 — cascade previously wrote a heartbeat UPDATE on
+# EVERY scanned memory (~2000) even when nothing advanced. Below this
+# delta, the hours_in_stage change is noise and the write is waste.
+_HEARTBEAT_SKIP_HOURS = 1.0
+
+# Source: issue #13 — the 503-transition payload darval reported is
+# redundant with the stage_transitions table and inflates the MCP
+# response. Surface a preview + count instead.
+_TRANSITION_PREVIEW_CAP = 50
+
 
 def run_cascade_advancement(store: MemoryStore) -> dict:
-    """Advance memory consolidation stages based on real elapsed time."""
+    """Advance memory consolidation stages based on real elapsed time.
+
+    Skips no-op heartbeat UPDATEs (|Δhours| < _HEARTBEAT_SKIP_HOURS),
+    batches stage_transitions INSERTs into one statement, and caps the
+    response payload at `transitions_preview` (first N) + total count.
+    """
     try:
-        advanced = 0
         transitions: list[dict] = []
+        heartbeats_written = 0
+        heartbeats_skipped = 0
+        scanned = 0
         now = datetime.now(timezone.utc)
 
         for stage_name in _ADVANCEABLE_STAGES:
             memories = store.get_memories_by_stage(stage_name, limit=500)
+            scanned += len(memories)
 
             for mem in memories:
-                result = _try_advance(store, mem, stage_name, now)
+                result, heartbeat = _try_advance(store, mem, stage_name, now)
                 if result:
-                    advanced += 1
                     transitions.append(result)
+                if heartbeat == "written":
+                    heartbeats_written += 1
+                elif heartbeat == "skipped":
+                    heartbeats_skipped += 1
 
-        # Log transitions to stage_transitions table
-        for t in transitions:
-            _log_transition(store, t)
+        store.insert_stage_transitions_batch(transitions)
 
         return {
-            "advanced": advanced,
-            "transitions": transitions,
+            "advanced": len(transitions),
+            "scanned": scanned,
+            "heartbeats_written": heartbeats_written,
+            "heartbeats_skipped": heartbeats_skipped,
+            "transitions_count": len(transitions),
+            "transitions_preview": transitions[:_TRANSITION_PREVIEW_CAP],
+        }
+    except Exception as exc:
+        logger.warning("Cascade advancement failed: %s", exc, exc_info=True)
+        return {
+            "advanced": 0,
+            "scanned": 0,
+            "error": f"{type(exc).__name__}: {exc}",
         }
-    except Exception as e:
-        logger.debug("Cascade advancement failed: %s", e)
-        return {"advanced": 0, "transitions": []}
 
 
 def _compute_real_hours(mem: dict, now: datetime) -> float:
@@ -93,8 +120,12 @@ def _try_advance(
     mem: dict,
     stage_name: str,
     now: datetime,
-) -> dict | None:
-    """Check and advance a single memory. Returns transition info or None."""
+) -> tuple[dict | None, str]:
+    """Check and advance a single memory.
+
+    Returns (transition_or_None, heartbeat_status) where heartbeat_status
+    is one of "written", "skipped", "transition".
+    """
     hours = _compute_real_hours(mem, now)
 
     ready, next_stage, _ = compute_advancement_readiness(
@@ -124,22 +155,32 @@ def _try_advance(
             mem.get("hippocampal_dependency", 1.0),
         )
         _update_stage_entered(store, mem["id"], new_entered)
-        return {
-            "memory_id": mem["id"],
-            "from_stage": stage_name,
-            "to_stage": next_stage,
-            "hours_in_prev": round(hours, 2),
-        }
+        return (
+            {
+                "memory_id": mem["id"],
+                "from_stage": stage_name,
+                "to_stage": next_stage,
+                "hours_in_prev": round(hours, 2),
+            },
+            "transition",
+        )
+
+    # Not advancing: only write a heartbeat if the hours delta is
+    # large enough to be informative. Below _HEARTBEAT_SKIP_HOURS the
+    # change is noise and the write is wasted fsync amplification
+    # (issue #13, Feinstein audit of darval's 66K-store run).
+    prev_hours = float(mem.get("hours_in_stage", 0.0) or 0.0)
+    if abs(hours - prev_hours) < _HEARTBEAT_SKIP_HOURS:
+        return None, "skipped"
 
-    # Not ready: just update hours_in_stage with real value
     store.update_memory_consolidation(
         mem["id"],
         stage_name,
         round(hours, 2),
         mem.get("replay_count", 0),
         mem.get("hippocampal_dependency", 1.0),
     )
-    return None
+    return None, "written"
 
 
 def _update_stage_entered(store: MemoryStore, memory_id: int, now: datetime) -> None:
@@ -152,23 +193,3 @@ def _update_stage_entered(store: MemoryStore, memory_id: int, now: datetime) ->
         store._conn.commit()
     except Exception:
         pass
-
-
-def _log_transition(store: MemoryStore, transition: dict) -> None:
-    """Log a stage transition to the stage_transitions table."""
-    try:
-        store._conn.execute(
-            "INSERT INTO stage_transitions "
-            "(memory_id, from_stage, to_stage, hours_in_prev_stage, trigger) "
-            "VALUES (%s, %s, %s, %s, %s)",
-            (
-                transition["memory_id"],
-                transition["from_stage"],
-                transition["to_stage"],
-                transition["hours_in_prev"],
-                "cascade",
-            ),
-        )
-        store._conn.commit()
-    except Exception as e:
-        logger.debug("Failed to log transition: %s", e)

mcp_server/handlers/consolidation/cls.py

@@ -17,11 +17,20 @@
 
 logger = logging.getLogger(__name__)
 
+# Source: issue #13 — previous cap of 500 saw ~2% of a 25k-episodic
+# store and produced 0 patterns by construction. 2000 matches plasticity
+# sampling and keeps PC algorithm's O(E^2) worst case tractable on a
+# 10k-entity vocabulary.
+_EPISODIC_SAMPLE_CAP = 2000
+_SEMANTICS_SAMPLE_CAP = 2000
+
 _EMPTY_CLS_STATS = {
     "patterns_found": 0,
     "new_semantics_created": 0,
     "skipped_inconsistent": 0,
     "skipped_duplicate": 0,
+    "causal_edges_found": 0,
+    "episodic_scanned": 0,
 }
 
 
@@ -30,9 +39,16 @@ def run_cls_cycle(
     settings,
     embeddings: EmbeddingEngine,
 ) -> dict:
-    """Run CLS consolidation: episodic -> semantic pattern extraction."""
-    episodic = store.get_episodic_memories(limit=500)
-    existing_semantics = store.get_semantic_memories(limit=500)
+    """Run CLS consolidation: episodic → semantic pattern extraction.
+
+    Pattern extraction (`plan_cls_consolidation`) and causal-edge
+    discovery (`_discover_causal_edges`) sample up to 2000 episodic
+    memories each — raised from 500 after Feynman's audit of darval's
+    66K run in issue #13 showed 500 sampled 2% of the episodic store
+    and produced 0 patterns by construction.
+    """
+    episodic = store.get_episodic_memories(limit=_EPISODIC_SAMPLE_CAP)
+    existing_semantics = store.get_semantic_memories(limit=_SEMANTICS_SAMPLE_CAP)
 
     if not episodic:
         return _EMPTY_CLS_STATS.copy()
@@ -47,6 +63,7 @@ def run_cls_cycle(
         "skipped_inconsistent": plan["skipped_inconsistent"],
         "skipped_duplicate": plan["skipped_duplicate"],
         "causal_edges_found": causal_edges_found,
+        "episodic_scanned": len(episodic),
     }
 
 
@@ -133,39 +150,69 @@ def _discover_causal_edges(
     store: MemoryStore,
     episodic: list[dict],
 ) -> int:
-    """Discover causal edges from entity co-occurrences."""
+    """Discover causal edges from entity co-occurrences (PC algorithm).
+
+    Gates on minimum signal before running the O(E²) independence tests:
+    the PC algorithm needs at least `min_observations` mentions per
+    entity in the sample to distinguish correlation from chance, so if
+    fewer than `_MIN_ENTITIES_FOR_PC` entities clear that threshold,
+    skip the analysis entirely (issue #13 Phase D).
+    """
     try:
         all_entities = store.get_all_entities(min_heat=0.0)
         entity_names = [e["name"] for e in all_entities if e.get("name")]
         if not entity_names or not episodic:
             return 0
 
-        co_matrix = compute_co_occurrence_matrix(episodic, entity_names)
         entity_counts = _count_entity_mentions(entity_names, episodic)
+        qualifying = sum(1 for c in entity_counts.values() if c >= _PC_MIN_OBSERVATIONS)
+        if qualifying < _MIN_ENTITIES_FOR_PC:
+            # Insufficient signal — don't run the full O(E^2) pass.
+            return 0
+
+        # Restrict the vocabulary to entities that meet the minimum, so
+        # the co-occurrence matrix is E_qualifying^2, not E_all^2.
+        active_names = [
+            n for n in entity_names if entity_counts[n] >= _PC_MIN_OBSERVATIONS
+        ]
+        co_matrix = compute_co_occurrence_matrix(episodic, active_names)
+        active_counts = {n: entity_counts[n] for n in active_names}
         edges = discover_causal_edges(
-            entity_names,
+            active_names,
             co_matrix,
-            entity_counts,
+            active_counts,
             len(episodic),
-            min_observations=3,
+            min_observations=_PC_MIN_OBSERVATIONS,
             independence_threshold=0.5,
         )
         return _store_causal_edges(store, all_entities, edges)
-    except Exception:
-        logger.debug("Causal discovery failed (non-fatal)")
+    except Exception as exc:
+        logger.warning("Causal discovery failed: %s", exc, exc_info=True)
         return 0
 
 
+# Source: PC algorithm lower bound — need ≥3 observations per variable
+# to distinguish dependence from sampling noise; need ≥5 active variables
+# for the independence tests to produce any non-trivial edge.
+_PC_MIN_OBSERVATIONS = 3
+_MIN_ENTITIES_FOR_PC = 5
+
+
 def _count_entity_mentions(
     entity_names: list[str],
     episodic: list[dict],
 ) -> dict[str, int]:
-    """Count how many episodic memories mention each entity."""
+    """Count how many episodic memories mention each entity.
+
+    Single pass over the episodic sample with precomputed lowercase
+    content and lowercase entity names (replaces the old O(N_ep × N_ent)
+    loop that called .lower() on every cell).
+    """
+    content_lowered = [(m.get("content") or "").lower() for m in episodic]
     counts: dict[str, int] = {}
     for name in entity_names:
-        counts[name] = sum(
-            1 for m in episodic if name.lower() in (m.get("content") or "").lower()
-        )
+        name_l = name.lower()
+        counts[name] = sum(1 for c in content_lowered if name_l in c)
     return counts
 
 

mcp_server/handlers/consolidation/compression.py

@@ -24,15 +24,21 @@ def run_compression_cycle(
     store: MemoryStore,
     settings: Any,
     embeddings: EmbeddingEngine,
+    memories: list[dict] | None = None,
 ) -> dict:
-    """Compress aging memories along the rate-distortion curve."""
-    memories = store.get_all_memories_for_decay()
+    """Compress aging memories along the rate-distortion curve.
+
+    `memories` may be pre-loaded by the consolidate handler (issue #13).
+    """
+    if memories is None:
+        memories = store.get_all_memories_for_decay()
 
     stats = {
         "compressed_to_gist": 0,
         "compressed_to_tag": 0,
         "protected_skipped": 0,
         "semantic_skipped": 0,
+        "rows_scanned": len(memories),
     }
 
     for mem in memories: