fix(recall): retain strongest graph traversal

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [SemVer](https://semver.org/) starting from v3.1.2.
 - **Advisory offline linear BEAM throughput benchmark (#901).** A fixed synthetic workload on a temporary database reports warm per-call latency, sequential throughput, raw samples and environment/revision metadata as JSON, with compatible same-machine baseline comparisons and a documented retention/interpretation policy. No runtime recall changes or CI timing threshold.
 
 ### Fixed
+- **Polyphonic graph recall retains the strongest eligible traversal to a memory (#1303).** When multiple entity seeds reach the same memory, a later shorter path can replace an earlier weaker one. Scoped recall checks gist/fact seeds against a visible, unsuperseded and unexpired backing row satisfying the caller's date, source and topic filters before traversal; unscoped callers keep their prior behavior. The graph voice still emits each traversed memory once and excludes seed IDs. Enhanced recall cache version 17 invalidates results from the actual Main predecessor, generation 16.
 - **Recency decay no longer rewards a row for being misdated.** `_recency_decay` now clamps the *age*, so a timestamp in the future saturates at exactly 1.0 instead of returning a factor above it, and a far-future timestamp can no longer overflow `math.exp` into the neutral 0.5 fallback. Legacy local-naive rows read as if they were older or newer depending on the host's UTC offset: on a host with a positive offset they land that offset in the future, where the unclamped factor reached 1.055 at UTC+9 — and because decay enters the score as a multiplier, that bonus silently promoted misdated rows. The enhanced-recall cache generation advances to 16 so entries holding the inflated score are recomputed rather than served. Timestamp provenance and any legacy-data migration remain separate work.
 - **Legacy Hermes per-file plugin installation instructions no longer create an incomplete package on upgrade (#1095).** The README points to the supported directory installer, and the guide replaces its per-file fallback with an ownership-checked backup and installer recovery for a selected Hermes home. Custom and profile artifacts are not overwritten. The real Hermes loader regression covers published 0.5.0 to 0.7.0 provider wheels with a compatible core, plus recovery. PM-managed and native macOS update survival remain separate.
 - **Graph recall no longer scans every edge for each walked node.** `find_related_memories` with an `edge_type` used `idx_edges_type`, which matches nearly every edge when one type dominates, as `ctx` does. It now uses the source and target indexes. On a database with no `ANALYZE` statistics, which Mnemosyne doesn't create, it returns the same results in the same order. A rare type on a node with very many edges costs more, because it reads all of that node's edges.

mnemosyne/core/beam.py

@@ -1189,15 +1189,20 @@ def _polyphonic_scope_where(
     channel_id: Optional[str] = None,
     veracity: Optional[str] = None,
     memory_type: Optional[str] = None,
+    from_date: Optional[str] = None,
+    to_date: Optional[str] = None,
+    source: Optional[str] = None,
+    topic: Optional[str] = None,
 ) -> Tuple[str, List[Any]]:
     """Scope predicate the polyphonic voices apply before their own limits.
 
     It must never be stricter than _polyphonic_row_passes_filters, which
     keeps rows whose session_id is NULL (import writes NULL when an export
     carries one), so those rows are let through here too. The exact author,
     channel, veracity and memory_type filters match the post-filter's
-    equality checks. The date filters stay post-filter only: the post-filter
-    and SQL disagree on date-only timestamps.
+    equality checks. Date comparisons use the post-filter's string bounds,
+    including date-only timestamps and its empty-string treatment of NULL.
+    Source and topic are separate exact matches, just like the post-filter.
     """
     clause, params = _recall_scope_where(
         session_id=session_id, cross_session=cross_session,
@@ -1208,10 +1213,17 @@ def _polyphonic_scope_where(
     clauses = [clause]
     for column, value in (("author_id", author_id), ("author_type", author_type),
                           ("channel_id", channel_id), ("veracity", veracity),
-                          ("memory_type", memory_type)):
+                          ("memory_type", memory_type), ("source", source),
+                          ("source", topic)):
         if value:
             clauses.append(f"{column} = ?")
             params.append(value)
+    if from_date:
+        clauses.append("COALESCE(timestamp, '') >= ?")
+        params.append(from_date)
+    if to_date:
+        clauses.append("COALESCE(timestamp, '') <= ?")
+        params.append(f"{to_date}T23:59:59")
     return " AND ".join(clauses), params
 
 
@@ -11197,15 +11209,11 @@ def _get_query_embedding():
     # so entries cached under an older digest are not reused. Part of the
     # hashed payload; the opaque key keeps the "v2:" prefix because QueryCache's
     # opaque-path recognition (_OPAQUE_V2_KEY_RE) keys off that prefix.
-    #
-    # 16: _recency_decay clamps the AGE, so a legacy local-naive row that reads
-    # as future-dated no longer scores above 1.0. Generation 15 entries hold the
-    # inflated score for exactly those rows, so they must not survive the change.
-    # Generation 16 is also claimed by #1275, #1308 and #1315. Whichever change
-    # lands first takes 16; each later independently merged semantic change takes
-    # the next generation that is actually free in the integration history at that
-    # point — not a number reserved here, since another change may consume it first.
-    _ENHANCED_RECALL_CACHE_VERSION = 16
+    # 16: clamp future-dated recency ages before exponentiation.
+    # 17: strongest eligible graph traversal and scoped active-seed selection.
+    # Main generation 16 can cache the pre-graph-fix ranking and paths.
+    # Recheck the actual predecessor before each later semantic integration.
+    _ENHANCED_RECALL_CACHE_VERSION = 17
     # NOTE: the key carries the env MNEMOSYNE_VEC_TYPE, not the table's
     # live DDL type — a reindex under a different type without a version
     # bump serves stale admission/ranking. Reindex flows must bump.
@@ -11880,6 +11888,7 @@ def _recall_polyphonic(self, query: str, top_k: int,
             session_id=self.session_id, cross_session=cross_session,
             author_id=author_id, author_type=author_type, channel_id=channel_id,
             veracity=veracity, memory_type=memory_type,
+            from_date=from_date, to_date=to_date, source=source, topic=topic,
         )
         try:
             polyphonic_results = engine.recall(

mnemosyne/core/polyphonic_recall.py

@@ -194,7 +194,10 @@ def recall(self, query: str, query_embedding: np.ndarray = None,
                     excluded = resolve_exclusions(conn, exclude_captures)
                 finally:
                     conn.close()
-        graph_results = self._graph_voice(query)
+        graph_results = (
+            self._graph_voice(query) if scope_where is None else
+            self._graph_voice(query, scope_where=scope_where, scope_params=scope_params)
+        )
         fact_results = self._fact_voice(query)
         if excluded:
             # These voices carry no producing-tier evidence. An ambiguous
@@ -826,7 +829,8 @@ def knn_excludes_unseen_admitted_rows(
             if own_conn:
                 conn.close()
 
-    def _graph_voice(self, query: str) -> List[RecallResult]:
+    def _graph_voice(self, query: str, scope_where: Optional[str] = None,
+                     scope_params: Sequence[Any] = ()) -> List[RecallResult]:
         """
         Voice 2: Episodic graph traversal.
 
@@ -874,30 +878,47 @@ def _graph_voice(self, query: str) -> List[RecallResult]:
                     metadata={"entity": entity, "fact": f"{fact.subject} {fact.predicate} {fact.object}"}
                 ))
 
+        # Pruned seeds cannot become output rows through another seed's walk.
+        seed_exclusion_ids = set(seed_ids)
+        if scope_where is not None and seed_ids:
+            # Gists/facts have no shared scope column. Resolve their derived
+            # memory IDs against both backing tiers before they can influence
+            # traversal or fusion. Visibility and active validity must hold
+            # on the same backing row; two ineligible tier copies cannot
+            # supply one half each. Unscoped standalone calls are unchanged.
+            eligible = {mid: PolyphonicResult(mid, 0.0, {}, {}) for mid in seed_ids}
+            self._drop_out_of_scope(eligible, scope_where, scope_params, require_active=True)
+            seed_ids = {mid: None for mid in seed_ids if mid in eligible}
+            results = [result for result in results if result.memory_id in seed_ids]
+
         # Graph traversal: from seed memory IDs, walk edges to
         # discover indirectly related memories (ctx edges only,
         # moderate weight threshold to avoid noise).
-        traversed_ids = set()
+        traversed = {}
         for seed_id in seed_ids:
             related = self.graph.find_related_memories(
                 seed_id, depth=2, edge_type="ctx", min_weight=0.3
             )
             for rel in related:
                 mid = rel["memory_id"]
-                if mid not in traversed_ids and mid not in seed_ids:
-                    traversed_ids.add(mid)
-                    results.append(RecallResult(
+                if mid in seed_exclusion_ids:
+                    continue
+                score = 0.4 / rel["depth"]  # Score decays with hop distance
+                previous = traversed.get(mid)
+                if previous is None or score > previous.score:
+                    traversed[mid] = RecallResult(
                         memory_id=mid,
-                        score=0.4 / rel["depth"],  # Score decays with hop distance
+                        score=score,
                         voice="graph_traversal",
                         metadata={
                             "seed": seed_id,
                             "edge_type": rel["edge_type"],
                             "depth": rel["depth"],
                             "weight": rel["weight"],
                         }
-                    ))
+                    )
 
+        results.extend(traversed.values())
         return results
 
     def _fact_voice(self, query: str) -> List[RecallResult]:
@@ -1066,14 +1087,18 @@ def _extract_entities(self, text: str) -> List[str]:
         return list(dict.fromkeys(entities))
 
     def _drop_out_of_scope(self, results: Dict[str, PolyphonicResult],
-                           scope_where: str, scope_params: Sequence[Any]) -> None:
+                           scope_where: str, scope_params: Sequence[Any], *,
+                           require_active: bool = False) -> None:
         """Remove memories outside the caller's scope before the fused cut.
 
         Graph and fact hits come from stores with no session column, so they
         are checked here against the memory rows they name. An id stays when
         any tier holding it is in scope. Ids with no memory row in either tier
         (consolidated facts, gists) are removed too: BeamMemory can only
         resolve memory rows, so they would hold a slot and then be discarded.
+        Graph seeds additionally require an unsuperseded, unexpired backing
+        row in that same scope. Minimal standalone schemas without validity
+        columns retain their existing visibility-only behavior.
         """
         if not results:
             return
@@ -1089,6 +1114,7 @@ def _drop_out_of_scope(self, results: Dict[str, PolyphonicResult],
         try:
             memory_ids = tuple(results)
             in_scope, out_of_scope = set(), set()
+            now_iso = datetime.now(timezone.utc).isoformat()
             for table in ("working_memory", "episodic_memory"):
                 # A standalone engine may point at a store without one of the
                 # tiers; any other failure must surface, or foreign rows would
@@ -1097,14 +1123,23 @@ def _drop_out_of_scope(self, results: Dict[str, PolyphonicResult],
                     "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)
                 ).fetchone() is None:
                     continue
+                visible = f"({scope_where})"
+                visible_params = list(scope_params)
+                if require_active:
+                    columns = {row[1] for row in conn.execute(f"PRAGMA table_info({table})")}
+                    if "superseded_by" in columns:
+                        visible += " AND superseded_by IS NULL"
+                    if "valid_until" in columns:
+                        visible += " AND (valid_until IS NULL OR julianday(valid_until) > julianday(?))"
+                        visible_params.append(now_iso)
                 # Chunked under SQLite's bound-variable limit, as the vec scans do.
                 for offset in range(0, len(memory_ids), 900):
                     chunk = memory_ids[offset : offset + 900]
                     placeholders = ", ".join("?" for _ in chunk)
                     rows = conn.execute(
-                        f"SELECT id, ({scope_where}) AS visible FROM {table} "
+                        f"SELECT id, ({visible}) AS visible FROM {table} "
                         f"WHERE id IN ({placeholders})",
-                        (*scope_params, *chunk),
+                        (*visible_params, *chunk),
                     ).fetchall()
                     for row in rows:
                         (in_scope if row["visible"] else out_of_scope).add(row["id"])

tests/test_enhanced_recall_cache.py

@@ -130,7 +130,7 @@ def execute_without_user_version(conn, sql, parameters=(), *args, **kwargs):
     )
 
 
-@pytest.mark.parametrize("previous_version", [6, 7, 8, 9, 10, 11, 12, 13, 14, 15])
+@pytest.mark.parametrize("previous_version", [6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16])
 def test_cache_version_bump_invalidates_prior_candidate_and_cap_entries(
     enhanced, monkeypatch, tmp_path: Path, previous_version: int
 ):
@@ -143,9 +143,9 @@ def test_cache_version_bump_invalidates_prior_candidate_and_cap_entries(
     hash-independent polyphonic order. Version 12 predates veracity and
     memory_type filters before voice limits. Version 13 predates hydration of
     the first eligible tier. Version 14 predates bind-budget-independent
-    linear candidate ordering. Version 15 predates the recency-decay age clamp,
-    so its entries hold an inflated score for legacy future-dated rows. The
-    ``v2:`` key prefix stays fixed
+    linear candidate ordering. Version 15 predates the recency-decay age clamp.
+    Version 16 includes that clamp but predates strongest-path selection
+    and scoped active-seed eligibility. The ``v2:`` key prefix stays fixed
     for QueryCache's opaque-key path.
     """
     memory, calls = enhanced
@@ -1792,6 +1792,62 @@ def test_sleep_invalidates_warmed_v3_cache(monkeypatch, tmp_path):
         _close_memory(memory)
 
 
+def test_persistent_cache_does_not_reuse_main_16_graph_ranking(monkeypatch, tmp_path: Path):
+    """Restart invalidates the actual predecessor's cached weaker graph path.
+
+    The real graph-selection regressions live in test_polyphonic_graph_strongest_path;
+    this test isolates persistence and recomputation at the enhanced-cache boundary.
+    """
+    monkeypatch.setenv("MNEMOSYNE_ENHANCED_RECALL", "1")
+    monkeypatch.setattr(
+        beam_module, "resolve_beam_runtime", lambda: SimpleNamespace(cross_session=False)
+    )
+    db_path = tmp_path / "memories.db"
+    cache_path = tmp_path / "query_cache.db"
+    calls = []
+
+    def build_memory(score):
+        memory = BeamMemory(session_id="session-a", db_path=db_path)
+        memory._query_cache = QueryCache(db_path=cache_path)
+
+        def recall(query, top_k=40, **kwargs):
+            calls.append(query)
+            return [{"id": "target", "content": "graph cache sentinel", "score": score}]
+
+        memory.recall = recall
+        return memory
+
+    with monkeypatch.context() as ctx:
+        ctx.setattr(BeamMemory, "_ENHANCED_RECALL_CACHE_VERSION", 16)
+        old = build_memory(0.2)
+        try:
+            assert _call(old)[0]["score"] == 0.2
+            assert _call(old)[0]["score"] == 0.2
+            assert len(calls) == 1
+        finally:
+            _close_memory(old)
+
+    probe = QueryCache(db_path=cache_path)
+    try:
+        keys = [row[0] for row in probe._conn.execute("SELECT normalized FROM query_cache")]
+        assert len(keys) == 1
+        assert probe.get_opaque(keys[0])[0]["score"] == 0.2
+    finally:
+        probe.close()
+
+    fresh = build_memory(0.4)
+    try:
+        misses = fresh._query_cache.stats()["misses"]
+        assert _call(fresh)[0]["score"] == 0.4
+        assert len(calls) == 2
+        assert fresh._query_cache.stats()["misses"] == misses + 1
+        assert _call(fresh)[0]["score"] == 0.4
+        assert len(calls) == 2
+        assert fresh._query_cache.stats()["misses"] == misses + 1
+    finally:
+        _close_memory(fresh)
+
+
 def test_warm_persistent_cache_does_not_serve_a_pre_clamp_score(monkeypatch, tmp_path: Path):
     """A generation-15 entry holding an inflated score must be recomputed, not served.
 
@@ -1903,11 +1959,11 @@ def pre_clamp_decay(timestamp_str, halflife_hours=beam_module.RECENCY_HALFLIFE_H
         "the cache key did not change between generations, so the stale entry could still be served"
     )
     assert [hit for _key, hit in fresh_keys] == [False, True], (
-        "expected the first generation-16 request to miss and the second to hit; got "
+        "expected the first current-generation request to miss and the second to hit; got "
         f"{[hit for _key, hit in fresh_keys]}"
     )
     assert corrected == pytest.approx(1.0), (
-        f"the warm cache served {corrected!r} instead of recomputing; generation 16 must "
+        f"the warm cache served {corrected!r} instead of recomputing; the current generation must "
         "invalidate entries written under the pre-clamp ranking"
     )
     assert repeats == pytest.approx(corrected)