fix(llm): drop out-of-scope nodes from the merged extraction result (#1895)

The #1757 cache guard refuses to write a cache entry for a node whose
source_file resolves to a real corpus file that was not dispatched, but
the node itself still flowed into merged["nodes"] and landed in
graph.json (plus any edges/hyperedges built on it). extract_corpus_
parallel now filters the merged result right before the #1890
dispatched-vs-returned reconciliation: a node is dropped when its
source_file resolves (against root, same normalization as #1890) to an
existing file (.is_file(), mirroring the #1757 condition) outside the
dispatched set. Non-file source_files (concepts, model-invented anchors)
pass through untouched. Edges whose endpoint and hyperedges whose member
is a dropped node id go with it, plus any edge/hyperedge itself
attributed to an undispatched real file. One summary warning names the
offending files and merged["out_of_scope_dropped"] records the count.
Running before the reconciliation keeps covered/uncovered reflecting the
post-filter graph.

Regression tests: a chunk over A.md+C.md returning a stray B.py node
loses the stray (and its edge/hyperedge) while sibling and concept
attributions survive; a clean run records a zero count and no warning.

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

graphify/llm.py

@@ -1994,6 +1994,72 @@ def extract_corpus_parallel(
     # they are silently re-dispatched (and re-omitted) forever. Diff the files we
     # dispatched against the source_files that actually came back and surface the gap.
     dispatched = {unit_path(f) for chunk in chunks for f in chunk}
+
+    # Out-of-scope node filter (#1895). The #1757 cache guard already refuses
+    # to WRITE a cache entry for a node whose source_file is a real file that
+    # was not dispatched, but the node itself still flowed into the merged
+    # result and landed in graph.json. Mirror the #1757 condition here: resolve
+    # each source_file against root and drop the node only when it resolves to
+    # an existing file (.is_file()) outside the dispatched set — non-file
+    # source_files (concepts, model-invented anchors) pass through untouched.
+    # Runs BEFORE the #1890 covered/uncovered reconciliation so that diff
+    # reflects the post-filter graph.
+    def _resolve_against_root(value: "str | Path") -> Path:
+        p = Path(value)
+        if not p.is_absolute():
+            p = root / p
+        try:
+            return p.resolve()
+        except (OSError, RuntimeError):
+            return p
+
+    _dispatched_resolved = {_resolve_against_root(p) for p in dispatched}
+
+    def _out_of_scope(item: dict) -> bool:
+        sf = item.get("source_file")
+        if not sf:
+            return False
+        p = _resolve_against_root(sf)
+        return p.is_file() and p not in _dispatched_resolved
+
+    dropped_ids: set = set()
+    dropped_files: set[str] = set()
+    kept_nodes: list[dict] = []
+    for n in merged.get("nodes", []):
+        if _out_of_scope(n):
+            if n.get("id") is not None:
+                dropped_ids.add(n.get("id"))
+            dropped_files.add(str(n.get("source_file")))
+            continue
+        kept_nodes.append(n)
+    dropped_node_count = len(merged.get("nodes", [])) - len(kept_nodes)
+    merged["out_of_scope_dropped"] = dropped_node_count
+    if dropped_node_count:
+        merged["nodes"] = kept_nodes
+        # Keep the graph consistent: an edge or hyperedge referencing a
+        # dropped node's id (or itself attributed to an undispatched real
+        # file) must not survive its endpoint.
+        merged["edges"] = [
+            e for e in merged.get("edges", [])
+            if not _out_of_scope(e)
+            and e.get("source") not in dropped_ids
+            and e.get("target") not in dropped_ids
+        ]
+        merged["hyperedges"] = [
+            h for h in merged.get("hyperedges", [])
+            if not _out_of_scope(h)
+            and not (dropped_ids & set(h.get("nodes", []) or []))
+        ]
+        shown = ", ".join(sorted(Path(f).name for f in dropped_files)[:5])
+        more = f" (+{len(dropped_files) - 5} more)" if len(dropped_files) > 5 else ""
+        print(
+            f"[graphify] WARNING: dropped {dropped_node_count} out-of-scope node(s) "
+            f"attributed to file(s) not dispatched for extraction: {shown}{more}. "
+            "The model mis-attributed them to another corpus file; they were "
+            "excluded from the graph (#1895).",
+            file=sys.stderr,
+        )
+
     covered: set[Path] = set()
     for n in merged.get("nodes", []):
         sf = n.get("source_file")

tests/test_chunking.py

@@ -360,6 +360,92 @@ def test_omitted_documents_are_reconciled_and_warned(tmp_path, capsys):
     assert "produced no nodes" in err and "doc1.md" in err
 
 
+def test_out_of_scope_nodes_are_dropped_from_merged_result(tmp_path, capsys):
+    """#1895: the #1757 cache guard skips the CACHE write for a node attributed
+    to a real corpus file that was not dispatched, but the node itself still
+    flowed into merged["nodes"] and landed in graph.json. The merged result must
+    drop such nodes (and edges/hyperedges touching them), warn once, and record
+    the count — while keeping in-scope sibling attributions (a node attributed
+    to a different dispatched file in the same chunk) and non-file concept
+    source_files, mirroring the #1757 `.is_file()` condition."""
+    from graphify.llm import extract_corpus_parallel
+
+    a = tmp_path / "A.md"; a.write_text("# a\n")
+    c = tmp_path / "C.md"; c.write_text("# c\n")
+    # B.py exists on disk but is NOT dispatched — the #1895 out-of-scope case.
+    b = tmp_path / "B.py"; b.write_text("def b(): pass\n")
+
+    def stray(chunk, **kwargs):
+        return {
+            "nodes": [
+                {"id": "a_ok", "source_file": "A.md", "file_type": "document"},
+                # sibling attribution: a different dispatched file in the same chunk
+                {"id": "c_sibling", "source_file": "C.md", "file_type": "document"},
+                # out-of-scope: real file on disk, never dispatched
+                {"id": "b_stray", "source_file": "B.py", "file_type": "code"},
+                # concept node: source_file is not a file — must survive
+                {"id": "auth_flow", "source_file": "auth flow", "file_type": "concept"},
+            ],
+            "edges": [
+                {"source": "a_ok", "target": "c_sibling", "source_file": "A.md"},
+                {"source": "a_ok", "target": "b_stray", "source_file": "A.md"},
+            ],
+            "hyperedges": [
+                {"id": "h_bad", "nodes": ["a_ok", "c_sibling", "b_stray"], "source_file": "A.md"},
+                {"id": "h_ok", "nodes": ["a_ok", "c_sibling", "auth_flow"], "source_file": "A.md"},
+            ],
+            "input_tokens": 1, "output_tokens": 1,
+        }
+
+    with patch("graphify.llm.extract_files_direct", side_effect=stray):
+        result = extract_corpus_parallel(
+            [a, c], backend="kimi", root=tmp_path,
+            token_budget=None, chunk_size=2, max_concurrency=1,
+        )
+
+    ids = {n["id"] for n in result["nodes"]}
+    assert "b_stray" not in ids, "out-of-scope node leaked into the merged graph (#1895)"
+    assert {"a_ok", "c_sibling", "auth_flow"} <= ids, (
+        f"in-scope sibling/concept attributions must be kept: {ids}"
+    )
+    assert result["out_of_scope_dropped"] == 1
+    # Edges/hyperedges referencing the dropped node id are gone; in-scope ones stay.
+    assert all(
+        "b_stray" not in (e.get("source"), e.get("target")) for e in result["edges"]
+    ), f"edge to dropped node survived: {result['edges']}"
+    assert any(
+        e["source"] == "a_ok" and e["target"] == "c_sibling" for e in result["edges"]
+    )
+    assert [h["id"] for h in result["hyperedges"]] == ["h_ok"]
+    err = capsys.readouterr().err
+    assert "out-of-scope" in err and "B.py" in err
+    # The dispatched files all produced nodes — reconciliation sees no gaps.
+    assert result["uncovered_files"] == []
+
+
+def test_out_of_scope_drop_count_is_zero_when_all_in_scope(tmp_path, capsys):
+    """Counter-test: a clean run records out_of_scope_dropped == 0 and no warning."""
+    from graphify.llm import extract_corpus_parallel
+
+    a = tmp_path / "A.md"; a.write_text("# a\n")
+
+    def clean(chunk, **kwargs):
+        return {
+            "nodes": [{"id": "a_ok", "source_file": "A.md", "file_type": "document"}],
+            "edges": [], "hyperedges": [], "input_tokens": 1, "output_tokens": 1,
+        }
+
+    with patch("graphify.llm.extract_files_direct", side_effect=clean):
+        result = extract_corpus_parallel(
+            [a], backend="kimi", root=tmp_path,
+            token_budget=None, chunk_size=1, max_concurrency=1,
+        )
+
+    assert result["out_of_scope_dropped"] == 0
+    assert [n["id"] for n in result["nodes"]] == ["a_ok"]
+    assert "out-of-scope" not in capsys.readouterr().err
+
+
 def test_checkpoint_caches_sliced_document_chunks(tmp_path, capsys):
     """#1870: the checkpoint's allowlist must resolve a FileSlice to its parent
     path (via unit_path), not read a non-existent `.rel`. An oversized doc is