fix(cache): stop persisting dangling edges/hyperedges in the semantic cache (#1916)

save_semantic_cache groups nodes/edges/hyperedges by their own source_file
and the write loop skips a group whose path is a ghost (not .is_file(),
silently) or out-of-scope per the #1757 guard — but an edge or hyperedge in
an ALLOWED group that references a node id from a skipped group was still
written verbatim, so on replay (check_semantic_cache) it dangled forever.
The #1895 filter does not cover this: it cleans the in-memory merged result,
while the checkpoint writes the cache BEFORE it runs and replay bypasses it
entirely.

Fix at the cache layer (the authoritative write path): before the write
loop, compute the node ids belonging to groups that will be skipped —
mirroring BOTH skip branches — minus ids also defined in a group that will
be written (duplicate-attribution nodes must not be over-pruned). Each
written group then drops edges whose source/target is a skipped id and
hyperedges whose member list intersects them (whole-hyperedge drop,
mirroring #1895). Pruning runs on the incoming result only, so with
merge_existing=True (the llm.py checkpoint path) the prior cached entry's
valid edges survive the union untouched. Everything is gated on
allowed_source_files being provided, so unscoped callers stay
byte-identical.

Complementary hardening in build_from_json: hyperedges were copied into
G.graph["hyperedges"] verbatim without member validation, so a dangling
hyperedge reached graph.json even from a live (non-cache) extraction.
Members are now remapped via the same normalization the pairwise-edge loop
uses and pruned when they still don't resolve; a hyperedge with no
surviving member is dropped whole with a stderr warning. (Single-member
hyperedges are legal in this codebase — per-file flows in the #1574 tests —
so the drop threshold is zero survivors, not two.)

Tests: scoped save with an edge to an out-of-scope real file, the same with
a ghost source_file, whole-hyperedge drop, unscoped save preserved
byte-identically (raw cache entry compared), merge_existing keeping the
prior slice's valid edges, and build_from_json pruning a dangling hyperedge
member / dropping an all-dangling hyperedge.

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

graphify/build.py

@@ -755,10 +755,43 @@ def build_from_json(extraction: dict, *, directed: bool = False, root: str | Pat
         # Relativize hyperedge source_file the same way nodes and edges are
         # (above), so to_json — which has no root and writes G.graph["hyperedges"]
         # verbatim — never leaks an absolute path from a semantic subagent (#1418).
+        kept_hyperedges = []
         for he in hyperedges:
             if isinstance(he, dict) and he.get("source_file"):
                 he["source_file"] = _norm_source_file(he["source_file"], _root)
-        G.graph["hyperedges"] = hyperedges
+            # Validate members against the built node set (#1916): a hyperedge
+            # member absent from the graph used to be copied into
+            # G.graph["hyperedges"] verbatim and reach graph.json dangling,
+            # even from a live (non-cache) extraction. Mirror the pairwise-edge
+            # handling above: remap mismatched ids via normalization first,
+            # then drop members that still don't resolve; drop the hyperedge
+            # itself when no valid member remains (single-member hyperedges
+            # are legal in this codebase, e.g. a per-file flow, so we prune
+            # rather than require two survivors).
+            if isinstance(he, dict) and isinstance(he.get("nodes"), list):
+                valid_members = []
+                for m in he["nodes"]:
+                    try:
+                        hash(m)
+                    except TypeError:
+                        continue
+                    if m not in node_set and isinstance(m, str):
+                        m = norm_to_id.get(_normalize_id(m), m)
+                    if m in node_set:
+                        valid_members.append(m)
+                if not valid_members:
+                    print(
+                        f"[graphify] WARNING: dropping hyperedge "
+                        f"{he.get('id', '?')!r} — none of its members "
+                        f"{he.get('nodes')!r} match built nodes.",
+                        file=sys.stderr,
+                    )
+                    continue
+                if valid_members != he["nodes"]:
+                    he["nodes"] = valid_members
+            kept_hyperedges.append(he)
+        if kept_hyperedges:
+            G.graph["hyperedges"] = kept_hyperedges
     return G
 
 

graphify/cache.py

@@ -615,6 +615,62 @@ def save_semantic_cache(
     if allowed_source_files is not None:
         allowed_paths = {resolved_source_path(path) for path in allowed_source_files}
 
+    def group_skipped(fpath: str) -> bool:
+        """Mirror the write-loop skip condition for one source_file group."""
+        p = resolved_source_path(fpath)
+        return not p.is_file() or (allowed_paths is not None and p not in allowed_paths)
+
+    # Dangling-reference pruning (#1916). A node group is skipped by the write
+    # loop below when its source_file is not a real file (ghost path) or is
+    # out-of-scope per the #1757 guard — but an edge/hyperedge in an ALLOWED
+    # group that references a node id from a skipped group used to be written
+    # verbatim, so on replay (check_semantic_cache) it dangled forever (the
+    # #1895 merged-result filter runs AFTER this checkpoint write and is
+    # bypassed entirely on replay). Compute the node ids that will be skipped
+    # and drop any to-be-written edge whose endpoint — or hyperedge whose
+    # member (whole-hyperedge drop, mirroring #1895) — references one. Gated
+    # on allowed_source_files so unscoped callers stay byte-identical.
+    if allowed_paths is not None:
+        skipped_ids: set = set()
+        written_ids: set = set()
+        for fpath, result in by_file.items():
+            target = skipped_ids if group_skipped(fpath) else written_ids
+            for n in result["nodes"]:
+                nid = n.get("id")
+                if nid is None:
+                    continue
+                try:
+                    hash(nid)
+                except TypeError:
+                    continue
+                target.add(nid)
+        # A duplicate-attribution node (defined in a skipped AND a written
+        # group) still reaches the cache — don't over-prune references to it.
+        skipped_ids -= written_ids
+        if skipped_ids:
+
+            def edge_dangles(e: dict) -> bool:
+                try:
+                    return e.get("source") in skipped_ids or e.get("target") in skipped_ids
+                except TypeError:
+                    # Non-hashable endpoint from an untrusted result; leave it
+                    # to build-time validation rather than fail the save.
+                    return False
+
+            def hyperedge_dangles(h: dict) -> bool:
+                try:
+                    return bool(skipped_ids & set(h.get("nodes") or []))
+                except TypeError:
+                    return False
+
+            for fpath, result in by_file.items():
+                if group_skipped(fpath):
+                    continue
+                result["edges"] = [e for e in result["edges"] if not edge_dangles(e)]
+                result["hyperedges"] = [
+                    h for h in result["hyperedges"] if not hyperedge_dangles(h)
+                ]
+
     saved = 0
     for fpath, result in by_file.items():
         p = resolved_source_path(fpath)

tests/test_build.py

@@ -970,3 +970,27 @@ def test_doc_twin_merge_does_not_touch_code_symbols():
     }
     G = build_from_json(ext, directed=False)
     assert {"m_foo", "m_foo_doc"} <= set(G.nodes())
+
+
+def test_build_from_json_prunes_dangling_hyperedge_members(capsys):
+    """#1916: build_from_json used to copy hyperedges into G.graph["hyperedges"]
+    verbatim without validating members, so a dangling member reached graph.json
+    even from a live (non-cache) extraction. Members absent from the built node
+    set are pruned — matching how dangling pairwise edges are skipped — and a
+    hyperedge with no surviving member is dropped whole."""
+    ext = {
+        "nodes": [
+            {"id": "alpha", "label": "alpha", "file_type": "code", "source_file": "a.py"},
+            {"id": "beta", "label": "beta", "file_type": "code", "source_file": "a.py"},
+        ],
+        "edges": [],
+        "hyperedges": [
+            {"id": "he_partial", "nodes": ["alpha", "beta", "ghost_member"], "source_file": "a.py"},
+            {"id": "he_all_ghost", "nodes": ["ghost1", "ghost2"], "source_file": "a.py"},
+        ],
+    }
+    G = build_from_json(ext)
+    hes = {h["id"]: h for h in G.graph.get("hyperedges", [])}
+    assert set(hes) == {"he_partial"}, "an all-dangling hyperedge must be dropped"
+    assert hes["he_partial"]["nodes"] == ["alpha", "beta"]
+    assert "he_all_ghost" in capsys.readouterr().err

tests/test_cache.py

@@ -586,3 +586,177 @@ def test_save_semantic_cache_merge_existing_unions(tmp_path):
     ids = {n["id"] for n in cached["nodes"]}
     assert ids == {"a", "b"}, "merge_existing must union both chunk slices"
     assert len(cached["edges"]) == 1
+
+
+def test_save_semantic_cache_drops_edges_to_out_of_scope_nodes(tmp_path):
+    """#1916: an edge in an ALLOWED file's group referencing a node grouped
+    under an out-of-scope REAL file used to be written verbatim, so on replay
+    (check_semantic_cache) it dangled forever — the #1895 merged-result filter
+    runs after this checkpoint write and is bypassed entirely on replay. The
+    written entry must carry no reference to the skipped id, while a
+    duplicate-attribution node (also defined in a written group) must not be
+    over-pruned."""
+    from graphify.cache import check_semantic_cache, save_semantic_cache
+
+    allowed = tmp_path / "allowed.md"
+    allowed.write_text("# Allowed\n")
+    outside = tmp_path / "outside.md"
+    outside.write_text("# Outside\n")
+
+    nodes = [
+        {"id": "kept", "source_file": "allowed.md"},
+        {"id": "stray", "source_file": "outside.md"},
+        # duplicate attribution: same id defined in a written AND a skipped group
+        {"id": "dup", "source_file": "allowed.md"},
+        {"id": "dup", "source_file": "outside.md"},
+    ]
+    edges = [
+        {"source": "kept", "target": "stray", "source_file": "allowed.md"},
+        {"source": "stray", "target": "kept", "source_file": "allowed.md"},
+        {"source": "kept", "target": "dup", "source_file": "allowed.md"},
+    ]
+    with pytest.warns(RuntimeWarning, match="out-of-scope source_file"):
+        saved = save_semantic_cache(
+            nodes, edges, root=tmp_path, allowed_source_files=["allowed.md"]
+        )
+    assert saved == 1
+
+    cached_nodes, cached_edges, _, uncached = check_semantic_cache(
+        [str(allowed)], root=tmp_path
+    )
+    assert uncached == []
+    assert {n["id"] for n in cached_nodes} == {"kept", "dup"}
+    pairs = [(e["source"], e["target"]) for e in cached_edges]
+    assert pairs == [("kept", "dup")], "edges touching the skipped id must be dropped"
+
+
+def test_save_semantic_cache_drops_edges_to_ghost_file_nodes(tmp_path):
+    """#1916 (ghost variant): a node group whose source_file does not exist is
+    silently skipped by the write loop; edges in a written group referencing
+    its node ids must not survive into the cache."""
+    from graphify.cache import check_semantic_cache, save_semantic_cache
+
+    real = tmp_path / "real.md"
+    real.write_text("# Real\n")
+
+    nodes = [
+        {"id": "kept", "source_file": "real.md"},
+        {"id": "phantom", "source_file": "ghost.md"},  # no such file on disk
+    ]
+    edges = [
+        {"source": "kept", "target": "phantom", "source_file": "real.md"},
+        {"source": "kept", "target": "kept", "relation": "self", "source_file": "real.md"},
+    ]
+    saved = save_semantic_cache(
+        nodes, edges, root=tmp_path, allowed_source_files=["real.md"]
+    )
+    assert saved == 1
+
+    cached_nodes, cached_edges, _, uncached = check_semantic_cache(
+        [str(real)], root=tmp_path
+    )
+    assert uncached == []
+    assert {n["id"] for n in cached_nodes} == {"kept"}
+    pairs = [(e["source"], e["target"]) for e in cached_edges]
+    assert pairs == [("kept", "kept")]
+
+
+def test_save_semantic_cache_drops_hyperedges_touching_skipped_nodes(tmp_path):
+    """#1916: a hyperedge whose member list intersects the skipped ids is
+    dropped whole (mirroring the #1895 semantics), while hyperedges over
+    surviving nodes are kept."""
+    from graphify.cache import check_semantic_cache, save_semantic_cache
+
+    allowed = tmp_path / "allowed.md"
+    allowed.write_text("# Allowed\n")
+    outside = tmp_path / "outside.md"
+    outside.write_text("# Outside\n")
+
+    nodes = [
+        {"id": "kept", "source_file": "allowed.md"},
+        {"id": "kept2", "source_file": "allowed.md"},
+        {"id": "stray", "source_file": "outside.md"},
+    ]
+    hyperedges = [
+        {"id": "he_bad", "nodes": ["kept", "stray"], "source_file": "allowed.md"},
+        {"id": "he_ok", "nodes": ["kept", "kept2"], "source_file": "allowed.md"},
+    ]
+    with pytest.warns(RuntimeWarning, match="out-of-scope source_file"):
+        save_semantic_cache(
+            nodes, [], hyperedges, root=tmp_path, allowed_source_files=["allowed.md"]
+        )
+
+    _, _, cached_hyperedges, uncached = check_semantic_cache(
+        [str(allowed)], root=tmp_path
+    )
+    assert uncached == []
+    assert {h["id"] for h in cached_hyperedges} == {"he_ok"}
+
+
+def test_save_semantic_cache_unscoped_preserves_dangling_refs_verbatim(tmp_path):
+    """#1916 guard-rail: unscoped callers (allowed_source_files=None) must stay
+    byte-identical — no pruning happens even when an edge or hyperedge
+    references a node grouped under a ghost file."""
+    from graphify.cache import save_semantic_cache
+
+    doc = tmp_path / "doc.md"
+    doc.write_text("# Doc\n")
+
+    nodes = [
+        {"id": "a", "source_file": "doc.md"},
+        {"id": "ghost_n", "source_file": "ghost.md"},  # skipped group (no file)
+    ]
+    edges = [{"source": "a", "target": "ghost_n", "source_file": "doc.md"}]
+    hyperedges = [{"id": "he", "nodes": ["a", "ghost_n"], "source_file": "doc.md"}]
+
+    saved = save_semantic_cache(nodes, edges, hyperedges, root=tmp_path)
+    assert saved == 1
+
+    import json
+    raw = json.loads(
+        (cache_dir(tmp_path, "semantic") / f"{file_hash(doc, tmp_path)}.json").read_text()
+    )
+    assert raw["edges"] == edges
+    assert raw["hyperedges"] == hyperedges
+
+
+def test_save_semantic_cache_merge_existing_prunes_only_incoming(tmp_path):
+    """#1916 + #1715: with merge_existing=True (the llm.py checkpoint path),
+    only the INCOMING slice is pruned before the union — the prior cached
+    entry's valid edges must survive untouched."""
+    from graphify.cache import save_semantic_cache
+
+    big = tmp_path / "big.md"
+    big.write_text("# Big\n")
+    other = tmp_path / "other.md"
+    other.write_text("# Other\n")
+
+    # checkpoint 1: a clean slice
+    save_semantic_cache(
+        [{"id": "a", "source_file": "big.md"}],
+        [{"source": "a", "target": "a", "relation": "self", "source_file": "big.md"}],
+        root=tmp_path,
+        merge_existing=True,
+        allowed_source_files=["big.md"],
+    )
+    # checkpoint 2: incoming slice with a dangling edge to an out-of-scope node
+    nodes2 = [
+        {"id": "b", "source_file": "big.md"},
+        {"id": "stray", "source_file": "other.md"},
+    ]
+    edges2 = [
+        {"source": "b", "target": "stray", "source_file": "big.md"},
+        {"source": "a", "target": "b", "source_file": "big.md"},
+    ]
+    with pytest.warns(RuntimeWarning, match="out-of-scope source_file"):
+        save_semantic_cache(
+            nodes2, edges2, root=tmp_path, merge_existing=True,
+            allowed_source_files=["big.md"],
+        )
+
+    cached = load_cached(big, root=tmp_path, kind="semantic")
+    assert {n["id"] for n in cached["nodes"]} == {"a", "b"}
+    pairs = [(e["source"], e["target"]) for e in cached["edges"]]
+    assert ("a", "a") in pairs, "prior entry's valid edge must survive the union"
+    assert ("a", "b") in pairs, "incoming valid edge must be kept"
+    assert not any("stray" in p for p in pairs)