fix(extract): scope fresh semantic results to dispatched files (#2926)

graphify/cache.py

@@ -9,7 +9,7 @@
 import tempfile
 import time
 import warnings
-from collections.abc import Iterable
+from collections.abc import Callable, Iterable
 from pathlib import Path
 
 # Output directory name — override with GRAPHIFY_OUT env var for worktrees or
@@ -1286,6 +1286,50 @@ def _group_has_partial_marker(group: dict) -> bool:
     return False
 
 
+def _semantic_source_matcher(
+    root: Path,
+) -> tuple[Callable[[str | Path], Path], Callable[[str | Path], str]]:
+    """Shared path-identity machinery for the semantic-scope guards (#1757/#2926).
+
+    ``save_semantic_cache``'s write allowlist and ``scope_semantic_result``'s
+    graph-feed filter must agree on exactly which ``source_file`` values are in
+    scope, so both derive their matching from this one implementation rather
+    than parallel copies that could drift apart.
+
+    Returns ``(source_identity, normalize_value)`` closed over ``root``:
+
+    - ``normalize_value(src)`` maps a raw ``source_file`` to its portable
+      relative forward-slash form (#2197),
+    - ``source_identity(value)`` maps any form (relative or absolute, against
+      the walked or the resolved root) to the single canonical walked path
+      that identities are compared against.
+    """
+    root_walked = _normalize_path(Path(os.path.abspath(root)))
+    root_resolved = _normalize_path(Path(root).resolve())
+
+    def normalize_value(src: str | Path) -> str:
+        norm = _normalize_source_file_value(src, root_walked)
+        if Path(norm).is_absolute() and root_walked != root_resolved:
+            norm = _normalize_source_file_value(src, root_resolved)
+        return norm
+
+    def source_identity(value: str | Path) -> Path:
+        path = Path(value)
+        if not path.is_absolute():
+            path = root_walked / path
+        elif root_walked != root_resolved:
+            normalized = _normalize_path(Path(os.path.abspath(path)))
+            try:
+                relative = normalized.relative_to(root_resolved)
+            except ValueError:
+                pass
+            else:
+                path = root_walked / relative
+        return _normalize_path(Path(os.path.abspath(path)))
+
+    return source_identity, normalize_value
+
+
 def save_semantic_cache(
     nodes: list[dict],
     edges: list[dict],
@@ -1350,8 +1394,7 @@ def save_semantic_cache(
     from collections import defaultdict
 
     kind = "semantic" if mode is None else f"semantic-{mode}"
-    root_walked = _normalize_path(Path(os.path.abspath(root)))
-    root_resolved = _normalize_path(Path(root).resolve())
+    source_path, _normalize_value = _semantic_source_matcher(root)
 
     def _normalized(item: dict) -> dict:
         """Copy of ``item`` with a portable ``source_file`` (#2197).
@@ -1366,9 +1409,7 @@ def _normalized(item: dict) -> dict:
         src = item.get("source_file")
         if not src:
             return item
-        norm = _normalize_source_file_value(src, root_walked)
-        if Path(norm).is_absolute() and root_walked != root_resolved:
-            norm = _normalize_source_file_value(src, root_resolved)
+        norm = _normalize_value(src)
         if norm != src:
             item = {**item, "source_file": norm}
         return item
@@ -1390,21 +1431,6 @@ def _normalized(item: dict) -> dict:
         if src:
             by_file[src]["hyperedges"].append(h)
 
-    def source_path(value: str | Path) -> Path:
-        """Return the normalized walked identity for a semantic group."""
-        path = Path(value)
-        if not path.is_absolute():
-            path = root_walked / path
-        elif root_walked != root_resolved:
-            normalized = _normalize_path(Path(os.path.abspath(path)))
-            try:
-                relative = normalized.relative_to(root_resolved)
-            except ValueError:
-                pass
-            else:
-                path = root_walked / relative
-        return _normalize_path(Path(os.path.abspath(path)))
-
     def resolved_source_path(value: str | Path) -> Path:
         path = source_path(value)
         try:
@@ -1559,3 +1585,105 @@ def hyperedge_dangles(h: dict) -> bool:
             stacklevel=2,
         )
     return saved
+
+
+def scope_semantic_result(
+    result: dict,
+    root: Path = Path("."),
+    allowed_source_files: "Iterable[str | Path] | None" = None,
+) -> tuple[set[str], int]:
+    """Scope an extraction result in place to the files actually dispatched (#2926).
+
+    Graph-side mirror of the ``allowed_source_files`` write-guard in
+    :func:`save_semantic_cache` (#1757). A model can attribute stray
+    nodes/edges to a corpus file that was not part of the current extraction
+    batch; :func:`build_merge` derives its replace-set from the source_files
+    present in the new chunks, so such a stray fragment would REPLACE that
+    file's entire prior contribution in graph.json — while its manifest entry
+    still says unchanged, so no later incremental run re-dispatches it and the
+    loss is permanent until a full rebuild.
+
+    Items whose ``source_file`` resolves outside ``allowed_source_files`` are
+    dropped from ``result``'s ``nodes`` / ``edges`` / ``hyperedges`` lists
+    (mutated in place); items without a ``source_file`` pass through. An edge
+    or hyperedge that survives the scope filter but references a dropped node
+    id is dropped too (#1916 mirror), unless that id is also defined by a kept
+    node (duplicate attribution must not be over-pruned).
+
+    Path matching shares :func:`_semantic_source_matcher` with
+    :func:`save_semantic_cache` (relative against ``root``, walked-path
+    identity), so an item this function keeps can never still hit the save's
+    out-of-scope skip, and vice versa.
+
+    Returns ``(dropped_source_files, dropped_item_count)`` for logging;
+    ``dropped_source_files`` holds the normalized ``source_file`` strings of
+    every group that had at least one item removed.
+    """
+    if allowed_source_files is None:
+        return set(), 0
+
+    source_identity, normalize_value = _semantic_source_matcher(root)
+
+    def _item_identity(item: dict) -> tuple[str | None, Path | None]:
+        """(display form, walked identity) of an item's source_file."""
+        src = item.get("source_file")
+        if not src:
+            return None, None
+        norm = normalize_value(src)
+        return norm, source_identity(norm)
+
+    allowed_paths = {source_identity(str(path)) for path in allowed_source_files}
+
+    def _hashable(value) -> bool:
+        try:
+            hash(value)
+        except TypeError:
+            return False
+        return True
+
+    dropped_files: set[str] = set()
+    dropped_items = 0
+    dropped_ids: set = set()
+    kept_ids: set = set()
+    for bucket in ("nodes", "edges", "hyperedges"):
+        kept: list[dict] = []
+        for item in result.get(bucket) or []:
+            display, ident = _item_identity(item)
+            if ident is not None and ident not in allowed_paths:
+                dropped_files.add(display)
+                dropped_items += 1
+                if bucket == "nodes" and item.get("id") is not None:
+                    nid = item["id"]
+                    if _hashable(nid):
+                        dropped_ids.add(nid)
+                continue
+            if bucket == "nodes" and item.get("id") is not None and _hashable(item["id"]):
+                kept_ids.add(item["id"])
+            kept.append(item)
+        result[bucket] = kept
+
+    # A duplicate-attribution node (defined in a dropped AND a kept group)
+    # survives the filter — don't prune references to it.
+    dropped_ids -= kept_ids
+    if dropped_ids:
+
+        def edge_dangles(e: dict) -> bool:
+            try:
+                return e.get("source") in dropped_ids or e.get("target") in dropped_ids
+            except TypeError:
+                # Non-hashable endpoint from an untrusted result; leave it
+                # to build-time validation rather than fail here.
+                return False
+
+        def hyperedge_dangles(h: dict) -> bool:
+            try:
+                return bool(dropped_ids & set(h.get("nodes") or []))
+            except TypeError:
+                return False
+
+        result["edges"] = [e for e in result.get("edges") or [] if not edge_dangles(e)]
+        result["hyperedges"] = [
+            h for h in result.get("hyperedges") or [] if not hyperedge_dangles(h)
+        ]
+
+    return dropped_files, dropped_items

graphify/cli.py

@@ -3515,6 +3515,7 @@ def _ctx_identity(source_file) -> str | None:
             check_semantic_cache as _check_semantic_cache,
             prune_semantic_cache as _prune_semantic_cache,
             save_semantic_cache as _save_semantic_cache,
+            scope_semantic_result as _scope_semantic_result,
         )
         sem_result: dict = {
             "nodes": [], "edges": [], "hyperedges": [],
@@ -3619,6 +3620,24 @@ def _progress(idx: int, total: int, _result: dict) -> None:
                 # graph without an explicit --allow-partial override.
                 if _chunk_stats["total"] and _chunk_stats["succeeded"] < _chunk_stats["total"]:
                     _extraction_incomplete = True
+                # #2926: scope the fresh result to the files actually dispatched,
+                # mirroring the allowed_source_files guard the cache write below
+                # applies. A model can attribute stray nodes/edges to a corpus
+                # file that was not dispatched this run; build_merge() derives
+                # its replace-set from the source_files present in new chunks,
+                # so such a fragment would REPLACE that file's entire prior
+                # contribution in graph.json while its manifest entry still says
+                # unchanged — no later incremental run re-dispatches it and the
+                # loss is permanent until a full rebuild.
+                _dropped_files, _dropped_items = _scope_semantic_result(
+                    fresh, target, uncached_paths,
+                )
+                if _dropped_files:
+                    print(
+                        f"[graphify extract] dropped {_dropped_items} out-of-scope "
+                        f"item(s) attributed to {len(_dropped_files)} file(s) not "
+                        f"dispatched this run: {', '.join(sorted(_dropped_files))}"
+                    )
                 # Which files truncated this run (item markers + the empty-parse
                 # _partial_files set). Computed BEFORE the save so it can be passed
                 # as partial_source_files: without it, a file whose only truncated

tests/test_cache.py

@@ -1749,3 +1749,116 @@ def test_corrupt_semantic_entry_warns_and_is_a_miss(tmp_path):
     # The corrupt entry is a miss, so the file is re-dispatched for extraction.
     assert nodes == []
     assert uncached == [str(f)]
+
+
+# --- #2926: graph-side scope filter -------------------------------------------
+# The #1757 guard protects the cache write, but the unfiltered fresh result
+# also feeds build_merge(), whose replace-set logic swaps a non-dispatched
+# file's entire prior contribution for a stray fragment. scope_semantic_result
+# applies the same allowlist to the result dict before it reaches the merge.
+
+def test_scope_semantic_result_drops_out_of_scope_groups(tmp_path):
+    """Stray items attributed to a non-dispatched file are removed; allowed
+    and source-less items pass through."""
+    from graphify.cache import scope_semantic_result
+
+    result = {
+        "nodes": [
+            {"id": "kept", "source_file": "intended.md"},
+            {"id": "stray", "source_file": "protected.md"},
+            {"id": "phantom", "source_file": "src/foo.ts"},  # nonexistent path
+            {"id": "no_source"},  # no source_file: passes through
+        ],
+        "edges": [
+            {"source": "kept", "target": "other", "source_file": "intended.md"},
+            {"source": "stray", "target": "kept", "source_file": "protected.md"},
+        ],
+        "hyperedges": [
+            {"id": "h_kept", "nodes": ["kept"], "source_file": "intended.md"},
+            {"id": "h_stray", "nodes": ["stray"], "source_file": "protected.md"},
+        ],
+    }
+
+    dropped_files, dropped_items = scope_semantic_result(
+        result, root=tmp_path, allowed_source_files=["intended.md"],
+    )
+
+    assert [n["id"] for n in result["nodes"]] == ["kept", "no_source"]
+    assert [e["source"] for e in result["edges"]] == ["kept"]
+    assert [h["id"] for h in result["hyperedges"]] == ["h_kept"]
+    assert dropped_files == {"protected.md", "src/foo.ts"}
+    assert dropped_items == 4  # 2 stray nodes + 1 stray edge + 1 stray hyperedge
+
+
+def test_scope_semantic_result_matches_absolute_and_relative_forms(tmp_path):
+    """An absolute in-root source_file and its relative form are the same
+    identity — both must match the allowlist entry (#2197 normalization)."""
+    from graphify.cache import scope_semantic_result
+
+    result = {
+        "nodes": [
+            {"id": "abs", "source_file": str(tmp_path / "doc.md")},
+            {"id": "rel", "source_file": "doc.md"},
+        ],
+        "edges": [],
+        "hyperedges": [],
+    }
+
+    dropped_files, dropped_items = scope_semantic_result(
+        result, root=tmp_path, allowed_source_files=["doc.md"],
+    )
+
+    assert [n["id"] for n in result["nodes"]] == ["abs", "rel"]
+    assert dropped_files == set()
+    assert dropped_items == 0
+
+
+def test_scope_semantic_result_prunes_edges_referencing_dropped_ids(tmp_path):
+    """#1916 mirror: an edge attributed to an ALLOWED file that references a
+    node id only defined by a DROPPED group would materialize that id as a
+    phantom node at build time; it must be dropped too. A duplicate-attribution
+    id (defined by both a kept and a dropped group) keeps its edges."""
+    from graphify.cache import scope_semantic_result
+
+    result = {
+        "nodes": [
+            {"id": "kept", "source_file": "a.md"},
+            {"id": "shared", "source_file": "a.md"},   # also defined by b.md
+            {"id": "shared", "source_file": "b.md"},   # duplicate attribution
+            {"id": "gone", "source_file": "c.md"},
+        ],
+        "edges": [
+            {"source": "kept", "target": "shared", "source_file": "a.md"},
+            {"source": "kept", "target": "gone", "source_file": "a.md"},
+        ],
+        "hyperedges": [
+            {"id": "h1", "nodes": ["kept", "gone"], "source_file": "a.md"},
+            {"id": "h2", "nodes": ["kept", "shared"], "source_file": "a.md"},
+        ],
+    }
+
+    scope_semantic_result(result, root=tmp_path,
+                          allowed_source_files=["a.md"])
+
+    # The b.md COPY of "shared" is dropped as out-of-scope, but because a kept
+    # node also defines that id, references to it must NOT be pruned.
+    assert [n["id"] for n in result["nodes"]] == ["kept", "shared"]
+    assert [e["target"] for e in result["edges"]] == ["shared"]
+    assert [h["id"] for h in result["hyperedges"]] == ["h2"]
+
+
+def test_scope_semantic_result_unscoped_is_a_no_op():
+    """allowed_source_files=None must leave the result untouched (same contract
+    as save_semantic_cache's unscoped callers)."""
+    from graphify.cache import scope_semantic_result
+
+    result = {
+        "nodes": [{"id": "n", "source_file": "anywhere.md"}],
+        "edges": [],
+        "hyperedges": [],
+    }
+
+    dropped_files, dropped_items = scope_semantic_result(result, root=Path("."))
+
+    assert (dropped_files, dropped_items) == (set(), 0)
+    assert [n["id"] for n in result["nodes"]] == ["n"]