fix(graph): honor historical visibility anchors

Author: Coding-Dev-Tools

engraphis/service.py

@@ -8530,8 +8530,8 @@ def temporal_ghost(row: Any) -> bool:
             "ON visibility_support.edge_id=visibility_edge.id "
         )
         # A live scene must count only the same temporal edge/support/memory rows
-        # that the later edge query can render. History intentionally keeps the
-        # broader public relation set so closed rows remain available as ghosts.
+        # that the later edge query can render. History keeps closed rows available
+        # as ghosts, but still applies the selected world/system-time anchors.
         if not include_history:
             visibility_sql += (
                 "AND (visibility_support.valid_from IS NULL "
@@ -8546,6 +8546,16 @@ def temporal_ghost(row: Any) -> bool:
                 "OR ?<visibility_support.expired_at) "
             )
             visibility_params.extend((t, t, t, known_t, known_t))
+        else:
+            visibility_sql += (
+                "AND (visibility_support.valid_from IS NULL "
+                "OR visibility_support.valid_from<=?) "
+                "AND (visibility_support.ingested_at IS NULL "
+                "OR visibility_support.ingested_at<=?) "
+                "AND (visibility_support.expired_at IS NULL "
+                "OR ?<visibility_support.expired_at) "
+            )
+            visibility_params.extend((t, known_t, known_t))
         visibility_sql += (
             "LEFT JOIN memories visibility_memory "
             "ON visibility_memory.id=visibility_support.memory_id "
@@ -8565,6 +8575,17 @@ def temporal_ghost(row: Any) -> bool:
                 "OR ?<visibility_memory.expired_at) "
             )
             visibility_params.extend((wid, t, t, t, known_t, known_t))
+        else:
+            visibility_sql += (
+                "AND visibility_memory.workspace_id=? "
+                "AND (visibility_memory.valid_from IS NULL "
+                "OR visibility_memory.valid_from<=?) "
+                "AND (visibility_memory.ingested_at IS NULL "
+                "OR visibility_memory.ingested_at<=?) "
+                "AND (visibility_memory.expired_at IS NULL "
+                "OR ?<visibility_memory.expired_at) "
+            )
+            visibility_params.extend((wid, t, known_t, known_t))
         visibility_sql += "WHERE visibility_edge.workspace_id=? "
         visibility_params.append(wid)
         if repo_id:
@@ -8584,6 +8605,16 @@ def temporal_ghost(row: Any) -> bool:
                 "OR ?<visibility_edge.expired_at) "
             )
             visibility_params.extend((t, t, t, known_t, known_t))
+        else:
+            visibility_sql += (
+                "AND (visibility_edge.valid_from IS NULL "
+                "OR visibility_edge.valid_from<=?) "
+                "AND (visibility_edge.ingested_at IS NULL "
+                "OR visibility_edge.ingested_at<=?) "
+                "AND (visibility_edge.expired_at IS NULL "
+                "OR ?<visibility_edge.expired_at) "
+            )
+            visibility_params.extend((t, known_t, known_t))
         visibility_sql += (
             "GROUP BY visibility_edge.id, visibility_edge.repo_id, "
             "visibility_edge.src, visibility_edge.dst) "
@@ -8779,10 +8810,11 @@ def temporal_ghost(row: Any) -> bool:
         entity_sql += " AND (NOT EXISTS (SELECT 1 FROM edges hidden_edge "
         entity_sql += (
             "WHERE hidden_edge.workspace_id=? "
-            "AND (hidden_edge.src=entity.id OR hidden_edge.dst=entity.id)) "
+            "AND (hidden_edge.src=entity.id OR hidden_edge.dst=entity.id) "
+            "AND (hidden_edge.ingested_at IS NULL OR hidden_edge.ingested_at<=?)) "
             "OR EXISTS (SELECT 1 FROM edges public_edge "
         )
-        entity_params.extend((wid,))
+        entity_params.extend((wid, known_t))
         if not include_history:
             entity_sql += (
                 "WHERE public_edge.workspace_id=? "

tests/test_graph_explorer_v2.py

@@ -3884,6 +3884,78 @@ def test_live_visibility_cap_ignores_closed_relations(monkeypatch):
     assert {edge["id"] for edge in scene["edges"]} == {"edge_bg"}
 
 
+def test_history_visibility_cap_ignores_edges_learned_after_known_at(monkeypatch):
+    """Future system-time rows must not consume a time-travel visibility budget."""
+    service, alpha, beta, _gamma = _seed_service()
+    workspace_id = service.store.get_or_create_workspace("acme")
+    old_target = service.store.upsert_entity(Node(
+        id="history-old-target", name="History Old Target", ntype="concept",
+        workspace_id=workspace_id,
+    ))
+    future_target = service.store.upsert_entity(Node(
+        id="history-future-target", name="History Future Target", ntype="concept",
+        workspace_id=workspace_id,
+    ))
+    service.store.upsert_edge(Edge(
+        id="history-old-edge", src=alpha, dst=old_target, relation="old",
+        workspace_id=workspace_id,
+    ))
+    service.store.upsert_edge(Edge(
+        id="history-future-edge", src=beta, dst=future_target, relation="future",
+        workspace_id=workspace_id,
+    ))
+    service.store.conn.execute("UPDATE entities SET created_at=0")
+    service.store.conn.execute(
+        "UPDATE edges SET valid_from=0, ingested_at=0 WHERE id='history-old-edge'"
+    )
+    service.store.conn.execute(
+        "UPDATE edges SET valid_from=0, ingested_at=200 WHERE id='history-future-edge'"
+    )
+    service.store.conn.commit()
+    monkeypatch.setattr(service_module, "MAX_GRAPH_ANALYSIS_EDGES", 1)
+
+    scene = service.graph_scene(
+        workspace="acme", level="complete", include_memory_nodes=False,
+        include_history=True, valid_at=100, known_at=100,
+    )
+
+    edge_ids = {edge["id"] for edge in scene["edges"]}
+    assert "history-old-edge" in edge_ids
+    assert "history-future-edge" not in edge_ids
+
+
+def test_live_scene_keeps_entities_before_future_edge_known_at():
+    """An edge learned after known_at must not hide its otherwise unlinked nodes."""
+    service = MemoryService.create(":memory:", graph_extractor="none")
+    workspace_id = service.store.get_or_create_workspace("acme")
+    source = service.store.upsert_entity(Node(
+        id="known-source", name="Known Source", ntype="concept",
+        workspace_id=workspace_id,
+    ))
+    target = service.store.upsert_entity(Node(
+        id="known-target", name="Known Target", ntype="concept",
+        workspace_id=workspace_id,
+    ))
+    service.store.upsert_edge(Edge(
+        id="future-edge", src=source, dst=target, relation="future",
+        workspace_id=workspace_id,
+    ))
+    service.store.conn.execute("UPDATE entities SET created_at=0")
+    service.store.conn.execute(
+        "UPDATE edges SET valid_from=0, ingested_at=200 WHERE id='future-edge'"
+    )
+    service.store.conn.commit()
+
+    scene = service.graph_scene(
+        workspace="acme", level="complete", include_memory_nodes=False,
+        valid_at=100, known_at=100,
+    )
+
+    node_ids = {node["id"] for node in scene["nodes"]}
+    assert {source, target} <= node_ids
+    assert "future-edge" not in {edge["id"] for edge in scene["edges"]}
+
+
 def test_live_entity_cap_ignores_closed_edge_only_entities(monkeypatch):
     """Closed-only endpoints must not consume the live entity candidate cap."""
     service, _alpha, _beta, _gamma = _seed_service()