fix(core): apply system-time snapshot to history when known_at is omitted

conftest.py

@@ -12,12 +12,13 @@
 # explicitly via monkeypatch (see tests/test_update_check.py).
 os.environ.setdefault("ENGRAPHIS_UPDATE_CHECK", "0")
 
-# Owner machines may configure ENGRAPHIS_EXTRACTOR=llm (via ~/.engraphis/config.env),
-# which would make every ingest-path test block on live LLM extraction calls. The unit
-# suite is offline-inert by contract (AGENTS.md §1 "primary offline gate"); tests that
-# exercise extraction opt back in explicitly via monkeypatch.setenv. setdefault keeps a
-# real shell override working, matching how config.env itself defers to the environment.
-os.environ.setdefault("ENGRAPHIS_EXTRACTOR", "none")
+# Owner machines may configure ENGRAPHIS_EXTRACTOR=llm_structured (via ~/.engraphis/config.env
+# or an exported shell variable), which would make every ingest-path test block on live LLM
+# extraction calls — and under tests/conftest.py's DNS stub they hang instead of failing
+# fast. The unit suite is offline-inert by contract (AGENTS.md §1 "primary offline gate");
+# tests that exercise extraction opt back in explicitly via monkeypatch.setenv, so this is
+# forced rather than setdefault: no shell export can leak a live LLM into the gate.
+os.environ["ENGRAPHIS_EXTRACTOR"] = "none"
 
 # The legacy scripts/test_*.py files are HTTP smoke tests (need a running server +
 # httpx), not unit tests. Keep pytest focused on the tests/ suite.

engraphis/core/engine.py

@@ -2783,13 +2783,16 @@ def _relatedness(self, query: str, flt: SearchFilter, *,
         records = self.store.list_memories(
             flt, include_invalid=include_invalid, limit=500, prompt_only=prompt_only,
         )
-        if include_invalid and flt.known_at is not None:
+        if include_invalid:
             # History must retain closed valid-time intervals, but cannot expose a
             # record that was not known at the requested system-time snapshot.
+            # Default to the current snapshot when the caller omitted known_at —
+            # same normalization RecallEngine.recall applies (core/recall.py).
+            system_time = flt.known_at if flt.known_at is not None else now_ts()
             records = [
                 rec for rec in records
-                if (rec.ingested_at is None or rec.ingested_at <= flt.known_at)
-                and (rec.expired_at is None or flt.known_at < rec.expired_at)
+                if (rec.ingested_at is None or rec.ingested_at <= system_time)
+                and (rec.expired_at is None or system_time < rec.expired_at)
             ]
         for rec in records:
             # Public history is model-adjacent just like ordinary recall: tool output

tests/test_engine.py

@@ -1110,6 +1110,42 @@ def test_why_and_timeline_history_respect_known_time_but_keep_closed_records():
     ]
 
 
+def test_why_and_timeline_default_snapshot_hides_expired_and_future_records():
+    eng = MemoryEngine.create(":memory:")
+    wid = eng.store.get_or_create_workspace("w")
+    rid = eng.store.get_or_create_repo(wid, "r")
+    far_past = time.time() - 10 * 86_400
+    future = time.time() + 10 * 86_400
+    records = (
+        MemoryRecord(
+            id="", workspace_id=wid, repo_id=rid, scope=Scope.REPO,
+            content="Default snapshot retention-expired record",
+            valid_from=far_past - 1.0, valid_to=far_past + 1.0,
+            ingested_at=far_past, expired_at=far_past + 2.0,
+        ),
+        MemoryRecord(
+            id="", workspace_id=wid, repo_id=rid, scope=Scope.REPO,
+            content="Default snapshot future-dated record",
+            valid_from=future, ingested_at=future,
+        ),
+        MemoryRecord(
+            id="", workspace_id=wid, repo_id=rid, scope=Scope.REPO,
+            content="Default snapshot live closed-interval record",
+            valid_from=1.0, valid_to=time.time() + 3_600.0, ingested_at=1.0,
+        ),
+    )
+    for record in records:
+        eng.store.add_memory(record)
+
+    timeline = eng.timeline("default snapshot", workspace_id=wid, repo_id=rid)
+    why = eng.why("default snapshot", workspace_id=wid, repo_id=rid)
+
+    assert [record.content for record in timeline] == [
+        "Default snapshot live closed-interval record",
+    ]
+    assert why["supersedes"] == []
+
+
 def test_temporal_supersession_closes_at_effective_time_and_keeps_vectors():
     eng = MemoryEngine.create(":memory:")
     wid = eng.store.get_or_create_workspace("w")