Merge pull request #22 from Marsu6996/fix/consolidation-recall-correctness

fix(memory): consolidation & recall correctness — live crisis graph, score clamp, reflection embedding, crash recovery

Author: CodeAbra

src/iai_mcp/capture.py

@@ -317,32 +317,53 @@ def capture_turn(
     from iai_mcp.embed import embedder_for_store
     from iai_mcp.events import TELEMETRY_EMBED_NATIVE_FAILURE, write_event
 
-    try:
-        # Embed the message content, never the cue. The cue is a provenance
-        # label only (transcript drains and deferred-drain pass a positional
-        # cue such as "session <id> turn <n>"); embedding it collapsed the
-        # stored vector space and broke semantic recall. text is already
-        # validated non-empty above (the MIN_CAPTURE_LEN guard), so embedding
-        # text is safe for every caller.
-        emb = embedder_for_store(store).embed(text)
-    except Exception as exc:
-        write_event(
-            store,
-            TELEMETRY_EMBED_NATIVE_FAILURE,
-            {
-                "op_type": "capture",
-                "backend": "rust",
-                "error_type": type(exc).__name__,
-                "error": str(exc),
-            },
-        )
-        raise NativeError(f"capture encode failed: {exc}") from exc
-    embedding = list(emb)
+    # Embedding is the expensive native (Rust) matmul; the exact-key idem dedup
+    # below is a cheap SQLite lookup. Active sessions re-drain the *whole*
+    # transcript every turn (write_deferred_captures / capture_transcript walk
+    # from line 0 on each call), so embedding eagerly here re-embedded every
+    # already-stored turn only to throw the vector away as a "reinforced"
+    # exact-key re-drain -> chronic daemon CPU. Defer the embed so it runs at
+    # most once and only when actually needed: an already-seen episodic turn
+    # short-circuits on its idem tag and never embeds. Embed the message
+    # content, never the cue (the cue is a positional provenance label;
+    # embedding it collapsed the vector space and broke semantic recall). text
+    # is validated non-empty above (the MIN_CAPTURE_LEN guard).
+    _embed_cache: dict[str, list[float]] = {}
+
+    def _compute_embedding() -> list[float]:
+        if "v" in _embed_cache:
+            return _embed_cache["v"]
+        try:
+            emb = embedder_for_store(store).embed(text)
+        except Exception as exc:
+            write_event(
+                store,
+                TELEMETRY_EMBED_NATIVE_FAILURE,
+                {
+                    "op_type": "capture",
+                    "backend": "rust",
+                    "error_type": type(exc).__name__,
+                    "error": str(exc),
+                },
+            )
+            raise NativeError(f"capture encode failed: {exc}") from exc
+        vec = list(emb)
+        _embed_cache["v"] = vec
+        return vec
 
     with _CAPTURE_DEDUP_LOCK:
         if _is_episodic_conversational(tier, role):
             ts_iso = now.isoformat()
             idem_t = _idem_tag(session_id, role, ts_iso, text, source_uuid=source_uuid)
+            # find_record_by_tag reads SQLite, but a just-inserted record may
+            # still sit in the in-process _record_buffer (not yet flushed to the
+            # records table). Under _CAPTURE_DEDUP_LOCK the insert path is
+            # serialized with this find, so flushing the buffer here makes every
+            # prior committed insert visible to the SQLite-backed find and closes
+            # the check-then-insert race that produced live idem-tag duplicates.
+            from iai_mcp.store import flush_record_buffer
+
+            flush_record_buffer(store)
             existing_id = store.find_record_by_tag(idem_t)
             if existing_id is not None:
                 try:
@@ -362,7 +383,7 @@ def capture_turn(
                 }
         else:
             try:
-                neighbours = store.query_similar(embedding, k=3, tier=tier)
+                neighbours = store.query_similar(_compute_embedding(), k=3, tier=tier)
             except (ValueError, IOError) as exc:
                 log.warning(
                     "capture_dedup_query_failed",
@@ -402,7 +423,7 @@ def capture_turn(
             tier=tier,
             literal_surface=text,
             aaak_index="",
-            embedding=embedding,
+            embedding=_compute_embedding(),
             community_id=None,
             centrality=0.0,
             detail_level=2,

src/iai_mcp/core/_serializers.py

@@ -6,9 +6,18 @@
 def _hit_to_json(h) -> dict:
     _vf = getattr(h, "valid_from", None)
     _vt = getattr(h, "valid_to", None)
+    # Clamp the *displayed* score to [0,1]. Multiplicative boosts (trigram*2,
+    # FTS*3, valence) can drive the internal score past 1.0; that raw value stays
+    # in `sort_score` for ordering, but the client must never see a "confidence"
+    # > 1 (or < 0). Ordering is unaffected: this only touches the serialized
+    # number, never the rank.
+    try:
+        _display_score = max(0.0, min(1.0, float(h.score)))
+    except (TypeError, ValueError):
+        _display_score = 0.0
     return {
         "record_id": str(h.record_id),
-        "score": float(h.score),
+        "score": _display_score,
         "reason": h.reason,
         "literal_surface": h.literal_surface,
         "adjacent_suggestions": [str(x) for x in h.adjacent_suggestions],

src/iai_mcp/daemon/__init__.py

@@ -28,6 +28,7 @@
     DAEMON_SLEEP_CYCLE_STALE,
     DAEMON_WATCHDOG_NEEDS_OPERATOR,
     DAEMON_WEDGE_KILL,
+    emit_best_effort,
     write_event,
 )
 from iai_mcp.identity_audit import continuous_audit
@@ -282,9 +283,47 @@ def _store_is_empty(store: MemoryStore) -> bool:
         # the unknown case as NOT empty so the tick proceeds; a truly empty store
         # just does a little harmless no-op work.
         log.debug("store empty check failed, assuming NOT empty: %s", exc)
+        # e8f3deb fixed the *behavior* (don't park the tick) but left the
+        # condition invisible. A recurring count failure (sqlite left in an error
+        # state by a heavy reader) should surface to the operator, not just
+        # log.debug. emit_best_effort is buffered and never raises, so it is safe
+        # even when the store connection is the thing failing.
+        try:
+            emit_best_effort(
+                store,
+                "store_empty_check_failed",
+                {"error": str(exc), "error_type": type(exc).__name__},
+                severity="warning",
+            )
+        except Exception:  # noqa: BLE001 -- telemetry must never break the tick
+            pass
         return False
 
 
+def _normalize_boot_lifecycle_state(raw: dict) -> tuple[dict, bool]:
+    """Repair a crash-left incoherent lifecycle state at boot.
+
+    A daemon killed mid-SLEEP can leave lifecycle_state.json at
+    current_state=SLEEP with sleep_cycle_progress=None -- incoherent, because a
+    real in-flight sleep cycle always carries a progress dict. Resuming it wedges
+    the daemon (it never advances the pipeline, never reaches the recluster that
+    clears crisis, and recall stays degraded). Reset that one case to a clean
+    WAKE and drop the stale crisis flag. A genuine degeneration re-arms crisis on
+    the next complete sleep cycle. Returns (state, changed).
+    """
+    if (
+        isinstance(raw, dict)
+        and raw.get("current_state") == "SLEEP"
+        and raw.get("sleep_cycle_progress") is None
+    ):
+        out = dict(raw)
+        out["current_state"] = "WAKE"
+        out["crisis_mode"] = False
+        out["crisis_mode_since_ts"] = None
+        return out, True
+    return raw, False
+
+
 def _is_inside_window(
     window: tuple[int, int] | list | None,
     now: datetime,
@@ -1075,6 +1114,38 @@ async def _drain_and_report() -> None:
         _sleep_pipeline = _SleepPipeline(store=store)
 
         from pathlib import Path as _PathS2
+        # Boot normalization for a crash mid-SLEEP: lifecycle_state.json can be
+        # left at current_state=SLEEP with sleep_cycle_progress=None -- an
+        # incoherent state (a real in-flight cycle always carries a progress
+        # dict). Resuming it wedges the daemon: it never advances the sleep
+        # pipeline, never reaches the recluster that clears crisis, and recall
+        # stays degraded (SLEEP + crisis both degrade recall). Reset that one
+        # case to a clean WAKE (and drop the stale crisis flag set before the
+        # crash) so the daemon serves immediately; a genuine degeneration will
+        # simply re-arm crisis on the next complete sleep cycle.
+        try:
+            import json as _json_lc
+            _lc_path = _PathS2.home() / ".iai-mcp" / "lifecycle_state.json"
+            _lc_raw = _json_lc.loads(_lc_path.read_text())
+            _lc_norm, _lc_changed = _normalize_boot_lifecycle_state(_lc_raw)
+            if _lc_changed:
+                _lc_path.write_text(_json_lc.dumps(_lc_norm, indent=2))
+                log.warning(
+                    "lifecycle_boot_normalized: stale SLEEP without "
+                    "sleep_cycle_progress -> WAKE (crisis cleared)"
+                )
+                try:
+                    emit_best_effort(
+                        store,
+                        "lifecycle_boot_normalized",
+                        {"from_state": "SLEEP", "to_state": "WAKE",
+                         "reason": "sleep_without_progress"},
+                        severity="warning",
+                    )
+                except Exception:  # noqa: BLE001 -- telemetry must not block boot
+                    pass
+        except (OSError, ValueError) as _lc_exc:
+            log.debug("lifecycle boot normalization skipped: %s", _lc_exc)
         _s2_config = _load_s2_config()
         _s2_coord = S2Coordinator(
             store=store,

src/iai_mcp/daemon/_watchdog.py

@@ -545,7 +545,11 @@ def _check_sleep_cycle_staleness(
         if not isinstance(progress, dict):
             return (False, {})
         attempt = progress.get("attempt")
-        if not isinstance(attempt, int) or attempt != 1:
+        # A retried-but-still-wedged cycle (attempt >= 2) is exactly the case the
+        # watchdog must catch, not ignore. Gate on attempt < 1 so any genuine
+        # running attempt is monitored; only attempt 0 / negative / non-int (and
+        # bool, since isinstance(True, int) is True) short-circuits.
+        if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
             return (False, {})
         started_at_raw = progress.get("started_at")
         if not isinstance(started_at_raw, str) or not started_at_raw:

src/iai_mcp/dmn_reflection.py

@@ -94,15 +94,30 @@ def synthesize(self, store, window_hours: int) -> MemoryRecord:
             "ts": now.isoformat(),
         }
 
+        # Embed the reflection's literal_surface so the record is retrievable by
+        # vector/cosine recall. Hardcoding a zero vector (the old behaviour) made
+        # every daily-reflection semantic record permanently invisible to recall
+        # and fed zero-norm vectors into the scoring matmul (divide-by-zero
+        # warnings). If the native embed fails, fall back to a zero vector flagged
+        # embedding_pending=1 so the daemon's reembed-pending path fills it later
+        # (never silently leave an unretrievable zero record).
         embed_dim = int(store.embed_dim)
-        embedding = [0.0] * embed_dim
+        embedding_pending = 0
+        try:
+            from iai_mcp.embed import embedder_for_store
+
+            embedding = list(embedder_for_store(store).embed(literal_surface))
+        except Exception:  # noqa: BLE001 -- degrade to deferred reembed, never zero-and-forget
+            embedding = [0.0] * embed_dim
+            embedding_pending = 1
 
         return MemoryRecord(
             id=uuid4(),
             tier="semantic",
             literal_surface=literal_surface,
             aaak_index="",
             embedding=embedding,
+            embedding_pending=embedding_pending,
             community_id=None,
             centrality=0.5,
             detail_level=1,

src/iai_mcp/hippo/_db.py

@@ -616,6 +616,15 @@ def reembed_pending_rows(self, embedder: Any) -> int:
         for row in rows:
             rid = row["id"]
             surface = row["literal_surface"] or ""
+            # On an encrypted store literal_surface is iai:enc:v1: ciphertext; embedding
+            # the ciphertext would produce a garbage vector. Decrypt first (no-op on a
+            # plaintext store or a value that isn't encrypted). A decrypt failure leaves
+            # the row embedding_pending=1 so it is retried rather than poisoned.
+            try:
+                surface = self._decrypt_record_field(rid, "literal_surface", surface)
+            except Exception as exc:  # noqa: BLE001
+                _log.warning("reembed_pending_rows: decrypt failed for id=%s: %s", rid, exc)
+                continue
             try:
                 vec = list(embedder.embed(surface))
             except Exception as exc:  # noqa: BLE001

src/iai_mcp/lilli/cycle/sleep_pipeline/_crisis.py

@@ -66,49 +66,19 @@ def step_crisis_recluster(
 
             if not dry_run:
                 tbl = self._store.db.open_table(RECORDS_TABLE)
-                try:
-                    df2 = tbl.search().to_pandas()
-                except (OSError, ValueError, RuntimeError, StoreError):
-                    df2 = df
-
                 try:
                     from iai_mcp.community import detect_communities
-                    from iai_mcp.graph import MemoryGraph
-                    from iai_mcp.store import EDGES_TABLE
+                    from iai_mcp.lilli.cycle.sleep_pipeline._live_graph import (
+                        build_live_graph,
+                    )
                     import uuid as _uuid
 
-                    g = MemoryGraph()
-                    for _, row in df2.iterrows():
-                        try:
-                            rid = _uuid.UUID(str(row["id"]))
-                            emb = row.get("embedding")
-                            emb_list = (
-                                list(emb) if emb is not None else []
-                            )
-                            g.add_node(rid, None, emb_list)
-                        except (ValueError, TypeError, AttributeError):
-                            continue
-
-                    try:
-                        edges_df = (
-                            self._store.db.open_table(EDGES_TABLE)
-                            .search()
-                            .to_pandas()
-                        )
-                        for _, e in edges_df.iterrows():
-                            try:
-                                src_u = _uuid.UUID(str(e["src"]))
-                                dst_u = _uuid.UUID(str(e["dst"]))
-                                g.add_edge(
-                                    src_u, dst_u,
-                                    weight=float(
-                                        e.get("weight", 1.0) or 1.0
-                                    ),
-                                )
-                            except (ValueError, TypeError, KeyError):
-                                continue
-                    except (OSError, ValueError, RuntimeError, StoreError) as exc:
-                        logger.debug("crisis_recluster edges query failed: %s", exc)
+                    # Recluster on the LIVE graph only (tombstone-filtered,
+                    # live-only edges) -- mirrors retrieve.py 53f04f9. Previously
+                    # this rebuilt communities over ALL records incl. 3000+
+                    # tombstoned, collapsing the partition (it reassigned ~9700
+                    # records into a single community on the real store).
+                    g = build_live_graph(self._store)
 
                     _assignment = detect_communities(
                         g, prior=None, prior_mode="cold"