graphify/tests/test_watch.py

"""Tests for watch.py - file watcher helpers (no watchdog required)."""
import json
import os
import subprocess
import sys
import time
from pathlib import Path
import pytest

from graphify.watch import _notify_only, _WATCHED_EXTENSIONS, _rebuild_lock, _check_shrink


# --- _notify_only ---

def test_notify_only_creates_flag(tmp_path):
    _notify_only(tmp_path)
    flag = tmp_path / "graphify-out" / "needs_update"
    assert flag.exists()
    assert flag.read_text() == "1"

def test_notify_only_creates_flag_dir(tmp_path):
    # graphify-out dir does not exist yet
    assert not (tmp_path / "graphify-out").exists()
    _notify_only(tmp_path)
    assert (tmp_path / "graphify-out").is_dir()

def test_notify_only_idempotent(tmp_path):
    _notify_only(tmp_path)
    _notify_only(tmp_path)
    flag = tmp_path / "graphify-out" / "needs_update"
    assert flag.read_text() == "1"


# --- _WATCHED_EXTENSIONS ---

def test_watched_extensions_includes_code():
    assert ".py" in _WATCHED_EXTENSIONS
    assert ".ts" in _WATCHED_EXTENSIONS
    assert ".go" in _WATCHED_EXTENSIONS
    assert ".rs" in _WATCHED_EXTENSIONS

def test_watched_extensions_includes_docs():
    assert ".md" in _WATCHED_EXTENSIONS
    assert ".txt" in _WATCHED_EXTENSIONS
    assert ".pdf" in _WATCHED_EXTENSIONS

def test_watched_extensions_includes_images():
    assert ".png" in _WATCHED_EXTENSIONS
    assert ".jpg" in _WATCHED_EXTENSIONS

def test_watched_extensions_excludes_noise():
    # .json is now indexed (bash/JSON extractors added in #866)
    assert ".json" in _WATCHED_EXTENSIONS
    assert ".sh" in _WATCHED_EXTENSIONS
    assert ".pyc" not in _WATCHED_EXTENSIONS
    assert ".log" not in _WATCHED_EXTENSIONS


# --- watch() import error without watchdog ---

def test_check_update_no_flag_returns_true(tmp_path):
    """check_update returns True and is silent when needs_update flag is absent."""
    from graphify.watch import check_update
    assert check_update(tmp_path) is True


def test_check_update_with_flag_returns_true_and_prints(tmp_path, capsys):
    """check_update returns True and prints notification when flag exists."""
    from graphify.watch import check_update
    flag = tmp_path / "graphify-out" / "needs_update"
    flag.parent.mkdir(parents=True, exist_ok=True)
    flag.write_text("1")
    result = check_update(tmp_path)
    assert result is True
    out = capsys.readouterr().out
    assert "graphify --update" in out


def test_check_update_does_not_clear_flag(tmp_path):
    """check_update never removes the needs_update flag (clearing is LLM's job)."""
    from graphify.watch import check_update
    flag = tmp_path / "graphify-out" / "needs_update"
    flag.parent.mkdir(parents=True, exist_ok=True)
    flag.write_text("1")
    check_update(tmp_path)
    assert flag.exists()


def test_watch_raises_without_watchdog(tmp_path, monkeypatch):
    import builtins
    real_import = builtins.__import__

    def mock_import(name, *args, **kwargs):
        if name == "watchdog.observers" or name == "watchdog.events":
            raise ImportError("mocked missing watchdog")
        return real_import(name, *args, **kwargs)

    monkeypatch.setattr(builtins, "__import__", mock_import)

    from graphify.watch import watch
    with pytest.raises(ImportError, match="watchdog not installed"):
        watch(tmp_path)


# --- _rebuild_lock (GH-858) ---


@pytest.mark.skipif(sys.platform == "win32", reason="fcntl-only (POSIX)")
def test_rebuild_lock_writes_pid_with_newline(tmp_path):
    out = tmp_path / "graphify-out"
    lock_path = out / ".rebuild.lock"
    with _rebuild_lock(out) as got:
        assert got is True
        assert lock_path.exists()
        contents = lock_path.read_text(encoding="utf-8")
        assert contents == f"{os.getpid()}\n", contents


@pytest.mark.skipif(sys.platform == "win32", reason="fcntl-only (POSIX)")
def test_rebuild_lock_removed_after_release(tmp_path):
    """GH-858: lock file must be unlinked once the rebuild completes so
    downstream waiters that poll for its absence unblock promptly."""
    out = tmp_path / "graphify-out"
    lock_path = out / ".rebuild.lock"
    with _rebuild_lock(out) as got:
        assert got is True
    assert not lock_path.exists(), "lock file should be unlinked after release"


@pytest.mark.skipif(sys.platform == "win32", reason="fcntl-only (POSIX)")
def test_rebuild_lock_does_not_accumulate_pids_across_runs(tmp_path):
    """GH-858: each acquisition truncates and rewrites the PID line rather
    than appending, so the file never grows into a digit-concatenation."""
    out = tmp_path / "graphify-out"
    lock_path = out / ".rebuild.lock"
    expected = f"{os.getpid()}\n"
    for _ in range(5):
        with _rebuild_lock(out) as got:
            assert got is True
            assert lock_path.read_text(encoding="utf-8") == expected
        assert not lock_path.exists()


def test_graphify_root_preserves_relative_when_invoked_with_relative_path(tmp_path, monkeypatch):
    """#777: ``.graphify_root`` stores the user-supplied path (``.``), not the
    resolved absolute, so a committed ``graphify-out/.graphify_root`` is
    portable across clones and CI runners."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "lib.py").write_text("def f(): pass\n", encoding="utf-8")

    monkeypatch.chdir(corpus)
    assert _rebuild_code(Path("."), acquire_lock=False) is True

    saved = (corpus / "graphify-out" / ".graphify_root").read_text(encoding="utf-8")
    assert saved == ".", (
        f".graphify_root must preserve the user-supplied path; got {saved!r}"
    )


def test_rebuild_code_writes_community_name(tmp_path):
    """#1808: `graphify update` / _rebuild_code must forward community_labels to
    to_json, so graph.json nodes carry a human-readable community_name (hub-derived
    for a code-only rebuild) — not just a numeric community id. Before the fix,
    _rebuild_code called to_json without community_labels, so the labels a
    cluster-only pass writes were stripped again on every incremental rebuild."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text(
        "def alpha():\n    return beta()\n\ndef beta():\n    return 1\n", encoding="utf-8"
    )
    (corpus / "b.py").write_text(
        "import a\n\ndef gamma():\n    return a.alpha()\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, acquire_lock=False) is True

    graph = json.loads((corpus / "graphify-out" / "graph.json").read_text(encoding="utf-8"))
    clustered = [n for n in graph["nodes"] if n.get("community") is not None]
    assert clustered, "expected clustered nodes in the rebuilt graph"
    assert all(n.get("community_name") for n in clustered), (
        "clustered nodes missing community_name — the update rebuild stripped the "
        "labels that cluster-only writes (#1808)"
    )


def test_rebuild_code_drops_labels_whose_community_changed(tmp_path):
    """An incremental rebuild must not reuse a saved label for a community whose
    membership changed. Labels are keyed by cid, but re-clustering reassigns cids,
    so after new files land cid N can cover a different community and its old name
    is then simply wrong. cluster-only guards this with the `.sig` membership
    fingerprints; _rebuild_code ignored them and hub-filled only *missing* labels,
    so stale names survived and were written back to labels.json as if current."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text(
        "def alpha():\n    return beta()\n\ndef beta():\n    return 1\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, acquire_lock=False) is True

    out = corpus / "graphify-out"
    labels_file = out / ".graphify_labels.json"
    sig_file = out / ".graphify_labels.json.sig"
    assert sig_file.exists(), "rebuild must persist membership signatures beside labels"

    # Stand in for an LLM naming pass: give every community a distinctive name,
    # leaving the signatures untouched so they still describe THIS clustering.
    labels = json.loads(labels_file.read_text(encoding="utf-8"))
    assert labels, "expected the first rebuild to write community labels"
    labels_file.write_text(
        json.dumps({cid: f"Named-{cid}" for cid in labels}), encoding="utf-8"
    )

    # Grow the corpus so clustering changes, then rebuild incrementally.
    for name in ("b.py", "c.py", "d.py"):
        (corpus / name).write_text(
            f"def {name[0]}_one():\n    return {name[0]}_two()\n\n"
            f"def {name[0]}_two():\n    return 2\n",
            encoding="utf-8",
        )
    assert _rebuild_code(corpus, acquire_lock=False) is True

    graph = json.loads((out / "graph.json").read_text(encoding="utf-8"))
    after = json.loads(labels_file.read_text(encoding="utf-8"))
    sigs = json.loads(sig_file.read_text(encoding="utf-8"))

    communities = {}
    for node in graph["nodes"]:
        cid = node.get("community")
        if cid is not None:
            communities.setdefault(str(cid), []).append(node["id"])

    from graphify.cluster import community_member_sigs
    expected = {
        str(cid): sig
        for cid, sig in community_member_sigs(
            {int(c): m for c, m in communities.items()}
        ).items()
    }
    assert sigs == expected, (
        "signatures must be rewritten in step with the labels; a drifting sidecar "
        "leaves the staleness guard nothing accurate to check against"
    )

    # Any surviving "Named-N" must sit on a community that genuinely did not change.
    for cid, name in after.items():
        if name.startswith("Named-"):
            assert cid in communities, f"label kept for vanished community {cid}"
            assert sigs[cid] == expected[cid], (
                f"community {cid} kept the stale label {name!r} after its "
                f"membership changed"
            )

    for node in graph["nodes"]:
        cid = node.get("community")
        if cid is not None and node.get("community_name", "").startswith("Named-"):
            assert sigs[str(cid)] == expected[str(cid)], (
                f"node {node['id']} carries stale community_name "
                f"{node['community_name']!r}"
            )


def test_rebuild_code_keeps_a_visualization_when_over_the_viz_cap(tmp_path, monkeypatch):
    """Crossing the viz node limit must not leave the project with no graph.html.
    _rebuild_code used to unlink the existing file and write nothing, so a repo
    that grew past the cap silently lost its visualization — and the file was
    already gone by the time the user read the message. The export path falls
    back to the community-aggregation view in exactly this case; the incremental
    path should too, so the artifact stays both current and present."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    # Several *disconnected* clusters: the aggregator declines to render a
    # single-community meta-graph, so a ring of mutually-importing modules
    # would collapse to one community and exercise the wrong path.
    for g in range(4):
        for i in range(3):
            other = (i + 1) % 3
            (corpus / f"g{g}_m{i}.py").write_text(
                f"import g{g}_m{other}\n\n"
                + "".join(f"def g{g}_f{i}_{j}():\n    return {j}\n\n" for j in range(4)),
                encoding="utf-8",
            )
    assert _rebuild_code(corpus, acquire_lock=False) is True
    html = corpus / "graphify-out" / "graph.html"
    assert html.exists(), "expected a normal (under-cap) rebuild to write graph.html"
    before = html.read_text(encoding="utf-8")

    # Drop the cap below the graph's size but above its community count — the
    # real shape of this bug. (A cap under the community count would make the
    # aggregated meta-graph breach it too, which is a different situation.)
    graph = json.loads((corpus / "graphify-out" / "graph.json").read_text(encoding="utf-8"))
    communities = {n.get("community") for n in graph["nodes"] if n.get("community") is not None}
    cap = (len(communities) + len(graph["nodes"])) // 2
    assert len(communities) < cap < len(graph["nodes"]), "test corpus cannot exercise the cap"
    monkeypatch.setenv("GRAPHIFY_VIZ_NODE_LIMIT", str(cap))
    (corpus / "g9_extra.py").write_text("def extra():\n    return 1\n", encoding="utf-8")
    assert _rebuild_code(corpus, acquire_lock=False) is True

    assert html.exists(), (
        "graph.html was deleted when the graph exceeded the viz cap — the "
        "incremental rebuild must fall back to the aggregated view like export does"
    )
    after = html.read_text(encoding="utf-8")
    assert after != before, "graph.html must be re-rendered, not left stale"

    # And the documented kill switch still means "no viz", not "aggregate".
    monkeypatch.setenv("GRAPHIFY_VIZ_NODE_LIMIT", "0")
    (corpus / "g9_extra2.py").write_text("def extra2():\n    return 2\n", encoding="utf-8")
    assert _rebuild_code(corpus, acquire_lock=False) is True
    assert not html.exists(), "GRAPHIFY_VIZ_NODE_LIMIT=0 must disable the HTML viz outright"


def test_update_rebuilds_with_nested_star_gitignore(tmp_path):
    """#1880: `graphify update` must not emit 0 nodes (and then refuse to
    overwrite) just because the source tree has a nested `.gitignore` with a
    broad pattern. This was the 0.9.15 symptom of the #1847/#1873 subtree-scoping
    bug: a nested bare `*` zeroed the re-scan, update built 0 nodes, and the
    shrink-guard refused. With scoping fixed the rebuild sees the real files."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    (corpus / "src").mkdir(parents=True)
    (corpus / "src" / "a.py").write_text(
        "from src.b import Base\nclass App(Base):\n    def run(self): return 1\n", encoding="utf-8"
    )
    (corpus / "src" / "b.py").write_text("class Base: pass\n", encoding="utf-8")
    (corpus / "main.py").write_text("def top(): return 2\n", encoding="utf-8")
    # a common scratch-dir idiom deeper in the tree: ignore everything HERE only
    (corpus / "scratch").mkdir()
    (corpus / "scratch" / ".gitignore").write_text("*\n", encoding="utf-8")
    (corpus / "scratch" / "junk.py").write_text("x = 1\n", encoding="utf-8")

    assert _rebuild_code(corpus, acquire_lock=False) is True

    graph = json.loads((corpus / "graphify-out" / "graph.json").read_text(encoding="utf-8"))
    sources = {n.get("source_file", "") for n in graph["nodes"]}
    assert graph["nodes"], "update produced 0 nodes on a tree with a nested '*' gitignore (#1880)"
    assert any("src/a.py" in s for s in sources) and any("main.py" in s for s in sources)
    # the nested-ignored scratch file stays out (scoped correctly, not tree-wide)
    assert not any("scratch/junk.py" in s for s in sources)


def test_update_discovers_newly_added_files_and_dirs(tmp_path):
    """#1837: after an initial build, a plain `graphify update` (full re-scan, no
    change-list) must discover brand-new files AND new directories. The reported
    silent no-op was the #1873 nested-gitignore scoping bug zeroing the re-scan;
    this pins the build -> add -> update -> discovered sequence the earlier test
    (single build) did not cover, with a nested `*` scratch dir as a guard."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    (corpus / "src").mkdir(parents=True)
    (corpus / "src" / "a.py").write_text("def alpha(): return 1\n", encoding="utf-8")
    assert _rebuild_code(corpus, acquire_lock=False) is True

    # Add a brand-new file and a brand-new nested directory after the first build,
    # plus a scratch dir that ignores only itself.
    (corpus / "src" / "new.py").write_text("def added(): return 2\n", encoding="utf-8")
    (corpus / "monitor").mkdir()
    (corpus / "monitor" / "dash.py").write_text("def board(): return 3\n", encoding="utf-8")
    (corpus / "scratch").mkdir()
    (corpus / "scratch" / ".gitignore").write_text("*\n", encoding="utf-8")
    (corpus / "scratch" / "junk.py").write_text("x = 1\n", encoding="utf-8")

    assert _rebuild_code(corpus, acquire_lock=False) is True

    sources = {n.get("source_file", "") for n in
               json.loads((corpus / "graphify-out" / "graph.json").read_text())["nodes"]}
    assert any("src/new.py" in s for s in sources), "new file not discovered by update (#1837)"
    assert any("monitor/dash.py" in s for s in sources), "new directory not discovered (#1837)"
    assert not any("scratch/junk.py" in s for s in sources)


def test_rebuild_honors_persisted_excludes(tmp_path):
    """#1886: `--exclude` recorded at extract time must survive into update/watch/
    hook rebuilds. Before the fix only the initial scan applied the excludes, so
    the first rebuild silently re-indexed the excluded paths. _rebuild_code now
    reads the persisted build config and re-applies them."""
    import json
    from graphify.watch import _rebuild_code, _write_build_config

    corpus = tmp_path / "corpus"
    (corpus / "src").mkdir(parents=True)
    (corpus / "vendor").mkdir()
    (corpus / "src" / "app.py").write_text("def keep(): return 1\n", encoding="utf-8")
    (corpus / "main.py").write_text("def top(): return 2\n", encoding="utf-8")
    (corpus / "vendor" / "lib.py").write_text("def vendored(): pass\n", encoding="utf-8")
    _write_build_config(corpus / "graphify-out", excludes=["vendor"])

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    graph = json.loads((corpus / "graphify-out" / "graph.json").read_text(encoding="utf-8"))
    sources = {n.get("source_file", "") for n in graph["nodes"]}
    assert any("src/app.py" in s for s in sources) and any("main.py" in s for s in sources)
    assert not any("vendor/lib.py" in s for s in sources), (
        "rebuild silently re-included an excluded path (#1886)"
    )


def test_rebuild_honors_persisted_no_gitignore(tmp_path):
    import json
    from graphify.watch import _rebuild_code, _write_build_config

    corpus = tmp_path / "corpus"
    generated = corpus / "generated"
    generated.mkdir(parents=True)
    (corpus / ".gitignore").write_text("generated/\n")
    (generated / "gen.py").write_text("def generated(): return 1\n")
    _write_build_config(
        corpus / "graphify-out", excludes=None, gitignore=False
    )

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    graph = json.loads((corpus / "graphify-out" / "graph.json").read_text())
    sources = {Path(str(node.get("source_file", ""))).as_posix() for node in graph["nodes"]}
    assert any(source.endswith("generated/gen.py") for source in sources)


def test_graphify_root_preserves_absolute_when_user_supplied(tmp_path):
    """When the caller supplies an absolute path, ``.graphify_root`` stores
    that absolute form verbatim — preserving explicit-absolute intent."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "lib.py").write_text("def f(): pass\n", encoding="utf-8")
    assert _rebuild_code(corpus, acquire_lock=False) is True

    saved = (corpus / "graphify-out" / ".graphify_root").read_text(encoding="utf-8")
    assert saved == str(corpus), (
        f"absolute caller path must be preserved as-is; got {saved!r}"
    )


def test_rebuild_code_deleted_cwd_without_repo_root_returns_false(tmp_path, monkeypatch, capsys):
    """Detached hooks can inherit a CWD that no longer exists.

    Without GRAPHIFY_REPO_ROOT, the rebuild should fail cleanly before creating
    relative graphify-out queue/lock files.
    """
    from graphify.watch import _rebuild_code

    old_cwd = Path.cwd()
    gone = tmp_path / "gone"
    gone.mkdir()
    monkeypatch.delenv("GRAPHIFY_REPO_ROOT", raising=False)

    os.chdir(gone)
    gone.rmdir()
    try:
        assert _rebuild_code(Path("."), changed_paths=[Path("lib.py")]) is False
    finally:
        os.chdir(old_cwd)

    out = capsys.readouterr().out
    assert "current working directory no longer exists" in out


def test_rebuild_code_deleted_cwd_uses_graphify_repo_root(tmp_path, monkeypatch):
    """GRAPHIFY_REPO_ROOT lets detached hook rebuilds recover from a deleted CWD."""
    from graphify.watch import _rebuild_code

    old_cwd = Path.cwd()
    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "lib.py").write_text("def f(): pass\n", encoding="utf-8")
    gone = tmp_path / "gone"
    gone.mkdir()
    monkeypatch.setenv("GRAPHIFY_REPO_ROOT", str(corpus))

    os.chdir(gone)
    gone.rmdir()
    try:
        assert _rebuild_code(
            Path("."),
            changed_paths=[Path("lib.py")],
            no_cluster=True,
        ) is True
        assert Path.cwd().resolve() == corpus.resolve()
        assert (corpus / "graphify-out" / "graph.json").exists()
    finally:
        os.chdir(old_cwd)


def test_rebuild_code_evicts_nodes_from_deleted_files(tmp_path):
    """#1007: graphify update (_rebuild_code with no changed_paths) must remove
    nodes and edges from files deleted since the last run."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()

    (corpus / "auth.py").write_text(
        "def login(): pass\ndef logout(): pass\n", encoding="utf-8"
    )
    (corpus / "utils.py").write_text(
        "def format_date(): pass\n", encoding="utf-8"
    )

    assert _rebuild_code(corpus, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    node_labels_before = {n["label"] for n in data.get("nodes", [])}
    assert "format_date()" in node_labels_before

    (corpus / "utils.py").unlink()

    assert _rebuild_code(corpus, acquire_lock=False) is True
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    node_labels_after = {n["label"] for n in data.get("nodes", [])}
    assert "format_date()" not in node_labels_after, "stale function node from deleted file must be evicted"
    assert "login()" in node_labels_after, "nodes from surviving file must be kept"


def _add_unrelated_semantic_pair(graph_path):
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    data["nodes"].extend([
        {"id": "docs_topic", "label": "DocsTopic", "file_type": "concept"},
        {"id": "shared_concept", "label": "SharedConcept", "file_type": "concept"},
    ])
    data["links"].append({
        "source": "docs_topic",
        "target": "shared_concept",
        "relation": "related_to",
    })
    data["hyperedges"] = [{
        "id": "semantic_context",
        "label": "Semantic context",
        "nodes": ["docs_topic", "shared_concept"],
    }]
    graph_path.write_text(json.dumps(data), encoding="utf-8")


@pytest.mark.parametrize(
    "changed_paths",
    [None, [Path("doc.md")]],
    ids=["full-update", "incremental-doc-update"],
)
def test_rebuild_code_preserves_hyperedges_for_rebuilt_surviving_source(
    tmp_path, changed_paths
):
    """#1755: AST-only updates must not drop semantic hyperedges whose members survive."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "doc.md").write_text(
        "# Design\n\n## Flow\n\nDetails.\n", encoding="utf-8"
    )

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    assert {"doc", "doc_design"} <= {node["id"] for node in data["nodes"]}
    data["hyperedges"] = [{
        "id": "doc_flow_group",
        "label": "Doc flow group",
        "nodes": ["doc", "doc_design"],
        "relation": "implements",
        "confidence": "EXTRACTED",
        "confidence_score": 1.0,
        "source_file": "doc.md",
    }]
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    assert _rebuild_code(
        corpus,
        changed_paths=changed_paths,
        no_cluster=True,
        acquire_lock=False,
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    assert after["hyperedges"] == [{
        "id": "doc_flow_group",
        "label": "Doc flow group",
        "nodes": ["doc", "doc_design"],
        "relation": "implements",
        "confidence": "EXTRACTED",
        "confidence_score": 1.0,
        "source_file": "doc.md",
    }]


@pytest.mark.parametrize(
    "changed_paths",
    [None, [Path("auth.md")]],
    ids=["full-update", "incremental-doc-update"],
)
def test_rebuild_code_preserves_semantic_edges_from_reextracted_doc(
    tmp_path, changed_paths
):
    """#1865: AST-only updates must not evict semantic edges whose source_file
    is a re-extracted document; only that source's AST-tier edges are replaced."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "auth.md").write_text(
        "# Token Validation\n\nVerifies bearer tokens.\n", encoding="utf-8"
    )
    (corpus / "login.md").write_text(
        "# Session Verification\n\nVerifies login sessions.\n", encoding="utf-8"
    )

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    node_ids = {n["id"] for n in data["nodes"]}
    assert {"auth_token_validation", "login_session_verification"} <= node_ids

    data["links"].extend([
        {
            "source": "auth_token_validation",
            "target": "login_session_verification",
            "relation": "semantically_similar_to",
            "confidence": "INFERRED",
            "source_file": "auth.md",
        },
        # A stale AST-tier edge of the same source must still be evicted.
        {
            "source": "auth_token_validation",
            "target": "login_session_verification",
            "relation": "references",
            "_origin": "ast",
            "source_file": "auth.md",
        },
    ])
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    assert _rebuild_code(
        corpus,
        changed_paths=changed_paths,
        no_cluster=True,
        acquire_lock=False,
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    relations = {
        (e.get("source"), e.get("target"), e.get("relation"))
        for e in after["links"]
    }
    assert (
        "auth_token_validation", "login_session_verification", "semantically_similar_to"
    ) in relations, "semantic edge from a re-extracted doc must survive an AST-only update"
    assert (
        "auth_token_validation", "login_session_verification", "references"
    ) not in relations, "stale AST-tier edge of a re-extracted source must be evicted"


@pytest.mark.parametrize(
    "changed_paths",
    [None, [Path("only.py")]],
    ids=["full-update", "incremental-update"],
)
def test_rebuild_code_prunes_final_deleted_file(tmp_path, changed_paths):
    """Deleting the final code file must reconcile the existing graph."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    only = corpus / "only.py"
    only.write_text("def only_fn():\n    return 1\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    _add_unrelated_semantic_pair(graph_path)
    before = json.loads(graph_path.read_text(encoding="utf-8"))
    code_node_id = next(n["id"] for n in before["nodes"] if n.get("source_file") == "only.py")
    before["hyperedges"].append({
        "id": "code_context",
        "label": "Code context",
        "nodes": [code_node_id],
        "source_file": "only.py",
    })
    before["nodes"].append({
        "id": "sourceless_ast_stub",
        "label": "ExternalType",
        "file_type": "class",
        "_origin": "ast",
    })
    graph_path.write_text(json.dumps(before), encoding="utf-8")

    only.unlink()
    assert _rebuild_code(
        corpus,
        changed_paths=changed_paths,
        no_cluster=True,
        acquire_lock=False,
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    assert not any(n.get("source_file") == "only.py" for n in after["nodes"])
    assert {"docs_topic", "shared_concept"} <= {n["id"] for n in after["nodes"]}
    assert any(
        e.get("source") == "docs_topic" and e.get("target") == "shared_concept"
        for e in after["links"]
    )
    assert {he["id"] for he in after["hyperedges"]} == {"semantic_context"}
    assert "sourceless_ast_stub" not in {n["id"] for n in after["nodes"]}


def test_rebuild_code_prunes_renamed_source_not_listed_by_hook(tmp_path):
    """A hook-style rename list may contain only the destination path."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    old = corpus / "old.py"
    old.write_text("def old_fn():\n    return 1\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    _add_unrelated_semantic_pair(graph_path)

    renamed = corpus / "renamed.py"
    old.rename(renamed)
    assert _rebuild_code(
        corpus,
        changed_paths=[Path("renamed.py")],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    sources = {n.get("source_file") for n in after["nodes"]}
    assert "old.py" not in sources
    assert "renamed.py" in sources
    assert {"docs_topic", "shared_concept"} <= {n["id"] for n in after["nodes"]}
    assert any(
        e.get("source") == "docs_topic" and e.get("target") == "shared_concept"
        for e in after["links"]
    )


def test_rebuild_code_normalizes_preserved_source_paths(tmp_path):
    """An incremental rebuild must not treat ./foo.py as a deleted live source."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    foo = corpus / "foo.py"
    bar = corpus / "bar.py"
    foo.write_text("def foo_fn():\n    return 1\n", encoding="utf-8")
    bar.write_text("def bar_fn():\n    return 1\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    for item in data["nodes"] + data["links"]:
        if item.get("source_file") == "foo.py":
            item["source_file"] = "./foo.py"
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    bar.write_text("def updated_bar_fn():\n    return 2\n", encoding="utf-8")
    assert _rebuild_code(
        corpus,
        changed_paths=[Path("bar.py")],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    assert "foo_fn()" in {n.get("label") for n in after["nodes"]}


def test_rebuild_code_prunes_renamed_ast_backed_document(tmp_path):
    """Destination-only rename reconciliation also covers AST-backed docs."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    old = corpus / "old.md"
    old.write_text("# Old heading\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    renamed = corpus / "renamed.md"
    old.rename(renamed)
    assert _rebuild_code(
        corpus,
        changed_paths=[Path("renamed.md")],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    sources = {n.get("source_file") for n in after["nodes"]}
    assert "old.md" not in sources
    assert "renamed.md" in sources


def test_rebuild_code_evicts_removed_symbol_from_surviving_file(tmp_path):
    """#1116: graphify update (_rebuild_code with no changed_paths) must prune a
    symbol removed from a file that still exists — and its inbound call edge —
    without dropping genuine semantic nodes that share the surviving file."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()

    (corpus / "a.py").write_text(
        "def foo(): pass\ndef bar(): pass\n", encoding="utf-8"
    )
    (corpus / "b.py").write_text(
        "from a import foo\n\ndef caller():\n    foo()\n", encoding="utf-8"
    )

    assert _rebuild_code(corpus, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))

    def labels(d):
        return {n["label"] for n in d.get("nodes", [])}

    def id_for(d, label):
        return next(n["id"] for n in d.get("nodes", []) if n["label"] == label)

    def edges(d):
        return d.get("links", d.get("edges", []))

    before = labels(data)
    assert {"foo()", "bar()", "caller()"} <= before
    foo_id = id_for(data, "foo()")
    caller_id = id_for(data, "caller()")
    assert any(
        {e.get("source"), e.get("target")} == {caller_id, foo_id}
        for e in edges(data)
    ), "cross-file caller->foo call edge must exist before removal"

    # Pre-seed a semantic node on the surviving a.py (no AST id, no _origin
    # marker). A naive "evict every re-extracted file's nodes by source_file"
    # fix would wrongly delete this; the identity-based fix must keep it.
    data["nodes"].append({
        "id": "a_authconcept",
        "label": "AuthConcept",
        "file_type": "concept",
        "source_file": "a.py",
    })
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # Remove foo() from a.py (keep bar); leave b.py untouched.
    (corpus / "a.py").write_text("def bar(): pass\n", encoding="utf-8")

    # No force=True: a symbol removed from a re-extracted file is a legitimate
    # shrink, so the shrink-guard must let `graphify update` refresh the graph
    # without --force (the lost node belongs to a rebuilt source).
    assert _rebuild_code(corpus, acquire_lock=False) is True
    after_data = json.loads(graph_path.read_text(encoding="utf-8"))
    after = labels(after_data)

    assert "foo()" not in after, "removed symbol must be pruned from surviving file"
    assert not any(
        e.get("source") == foo_id or e.get("target") == foo_id
        for e in edges(after_data)
    ), "dangling edge to the removed symbol must be dropped"
    assert "bar()" in after, "surviving symbol in the same file must be kept"
    assert "caller()" in after, "unchanged file's nodes must be kept"
    assert "AuthConcept" in after, "semantic node on a surviving file must not be evicted"


def test_rebuild_code_preupgrade_marker_less_node_one_cycle_lag(tmp_path):
    """#1118 backward-compat: a graph.json built before #1116 has no `_origin`
    markers. On the first `graphify update` after upgrading, a symbol removed
    from a surviving file is NOT pruned that cycle — its old node carries no
    marker, so the new drop-rule skips it. This is a deliberate one-cycle lag
    (no data loss); it self-heals once the node has been stamped `_origin="ast"`
    (which a full re-extraction does for every surviving symbol)."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text("def bar(): pass\n", encoding="utf-8")

    assert _rebuild_code(corpus, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))

    def labels(d):
        return {n["label"] for n in d.get("nodes", [])}

    # Simulate a pre-#1116 graph: strip every `_origin` marker, then inject a
    # stale AST node for a symbol no longer present in a.py's source — also
    # marker-less, exactly as a pre-upgrade graph would carry it.
    for n in data["nodes"]:
        n.pop("_origin", None)
    data["nodes"].append({
        "id": "a_foo",
        "label": "foo()",
        "file_type": "function",
        "source_file": "a.py",
    })
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # First update after "upgrade" (full rebuild, no changed_paths): the stale
    # node has no marker, so the drop-rule skips it and it survives this cycle.
    assert _rebuild_code(corpus, acquire_lock=False, force=True) is True
    after = json.loads(graph_path.read_text(encoding="utf-8"))
    assert "foo()" in labels(after), (
        "pre-upgrade marker-less stale node must survive the first update — "
        "documented one-cycle backward-compat lag (#1118)"
    )

    # Once stamped (a full re-extraction stamps every surviving symbol), the
    # drop-rule applies on the next update and the stale node self-heals away.
    for n in after["nodes"]:
        if n["label"] == "foo()":
            n["_origin"] = "ast"
    graph_path.write_text(json.dumps(after), encoding="utf-8")

    assert _rebuild_code(corpus, acquire_lock=False, force=True) is True
    healed = json.loads(graph_path.read_text(encoding="utf-8"))
    assert "foo()" not in labels(healed), (
        "once carrying _origin=ast, the stale node is pruned on the next "
        "update (self-heal)"
    )
    assert "bar()" in labels(healed), "surviving symbol must be kept throughout"


@pytest.mark.skipif(sys.platform == "win32", reason="fcntl-only (POSIX)")
def test_rebuild_lock_non_blocking_does_not_clobber_holder(tmp_path):
    """GH-858: a non-blocking caller that fails to acquire the lock must not
    truncate the holder's PID payload."""
    out = tmp_path / "graphify-out"
    lock_path = out / ".rebuild.lock"
    with _rebuild_lock(out) as outer:
        assert outer is True
        held_contents = lock_path.read_text(encoding="utf-8")
        with _rebuild_lock(out, blocking=False) as inner:
            assert inner is False
            # Holder's PID line must still be intact.
            assert lock_path.read_text(encoding="utf-8") == held_contents


def test_rebuild_code_is_idempotent_when_cluster_ids_flap(tmp_path, monkeypatch):
    from graphify import cluster as cluster_mod
    from graphify.watch import _rebuild_code

    src = tmp_path / "app.py"
    src.write_text("def alpha():\n    return 1\n\ndef beta():\n    return alpha()\n", encoding="utf-8")

    calls = {"n": 0}

    def flaky_cluster(G):
        calls["n"] += 1
        nodes = sorted(G.nodes())
        if calls["n"] % 2 == 1:
            return {100: nodes}
        return {7: nodes}

    monkeypatch.setattr(cluster_mod, "cluster", flaky_cluster)
    monkeypatch.setattr(cluster_mod, "score_all", lambda _G, comm: {cid: 1.0 for cid in comm})

    assert _rebuild_code(tmp_path)
    graph_path = tmp_path / "graphify-out" / "graph.json"
    report_path = tmp_path / "graphify-out" / "GRAPH_REPORT.md"
    first_graph = graph_path.read_text(encoding="utf-8")
    first_report = report_path.read_text(encoding="utf-8")

    assert _rebuild_code(tmp_path)
    second_graph = graph_path.read_text(encoding="utf-8")
    second_report = report_path.read_text(encoding="utf-8")

    assert first_graph == second_graph
    assert first_report == second_report


def test_rebuild_code_skips_cluster_when_topology_unchanged(tmp_path, monkeypatch):
    from graphify import cluster as cluster_mod
    from graphify.watch import _rebuild_code

    src = tmp_path / "app.py"
    src.write_text("def alpha():\n    return 1\n\ndef beta():\n    return alpha()\n", encoding="utf-8")

    calls = {"n": 0}

    def cluster_once(G):
        calls["n"] += 1
        if calls["n"] > 1:
            raise AssertionError("cluster() should be skipped when topology is unchanged")
        return {0: sorted(G.nodes())}

    monkeypatch.setattr(cluster_mod, "cluster", cluster_once)
    monkeypatch.setattr(cluster_mod, "score_all", lambda _G, comm: {cid: 1.0 for cid in comm})

    assert _rebuild_code(tmp_path)
    assert _rebuild_code(tmp_path)
    assert calls["n"] == 1


# --- .graphifyignore honored in watch handler (gh-928) ---


def _watchdog_available() -> bool:
    try:
        import watchdog  # noqa: F401
        return True
    except ImportError:
        return False


@pytest.mark.skipif(not _watchdog_available(), reason="watchdog not installed")
def test_watch_handler_honors_graphifyignore(tmp_path, monkeypatch):
    """gh-928: the watch Handler must short-circuit paths matching
    .graphifyignore so busy volumes (node_modules churn, build artefacts,
    Time Machine writes, …) don't wake the rebuild pipeline.
    """
    import threading
    from graphify import watch as watch_mod

    watch_root = tmp_path / ".hidden-parent" / "corpus"
    watch_root.mkdir(parents=True)
    (watch_root / ".graphifyignore").write_text("node_modules/\nbuild/\n", encoding="utf-8")
    (watch_root / "node_modules").mkdir()
    (watch_root / "build").mkdir()

    rebuild_calls: list[Path] = []
    notify_calls: list[Path] = []
    monkeypatch.setattr(watch_mod, "_rebuild_code", lambda p, **kw: rebuild_calls.append(p) or True)
    monkeypatch.setattr(watch_mod, "_notify_only", lambda p: notify_calls.append(p))

    # Run watch() in a thread with a short debounce so we can verify the
    # post-debounce dispatch path actually runs on real events.
    t = threading.Thread(
        target=watch_mod.watch,
        args=(watch_root,),
        kwargs={"debounce": 0.2},
        daemon=True,
    )
    t.start()
    time.sleep(0.5)  # let observer.start() settle

    # Ignored writes — handler must drop these.
    (watch_root / "node_modules" / "junk.js").write_text("// noise\n", encoding="utf-8")
    (watch_root / "build" / "out.py").write_text("x = 1\n", encoding="utf-8")
    time.sleep(1.0)
    assert rebuild_calls == [], "ignored writes triggered a rebuild"
    assert notify_calls == [], "ignored writes triggered a notify"

    # Non-ignored write — handler must accept and (after debounce) dispatch.
    (watch_root / "app.py").write_text("def f():\n    return 1\n", encoding="utf-8")
    deadline = time.monotonic() + 5.0
    while time.monotonic() < deadline and not rebuild_calls:
        time.sleep(0.1)
    assert rebuild_calls, "non-ignored .py write should have triggered _rebuild_code"


@pytest.mark.skipif(not _watchdog_available(), reason="watchdog not installed")
def test_watch_loads_graphifyignore_once(tmp_path, monkeypatch):
    """gh-928: .graphifyignore must be parsed exactly once at watch() startup,
    not per filesystem event. Otherwise busy volumes re-read the file
    thousands of times per second.
    """
    import threading
    from graphify import watch as watch_mod
    from graphify import detect as detect_mod

    (tmp_path / ".graphifyignore").write_text("ignored/\n", encoding="utf-8")
    (tmp_path / "ignored").mkdir()

    calls = {"n": 0}
    real_loader = detect_mod._load_graphifyignore

    def counting_loader(root, **kwargs):
        calls["n"] += 1
        return real_loader(root, **kwargs)

    # Patch the symbol the watch module imported at module-load time.
    monkeypatch.setattr(watch_mod, "_load_graphifyignore", counting_loader)
    monkeypatch.setattr(watch_mod, "_rebuild_code", lambda p, **kw: True)
    monkeypatch.setattr(watch_mod, "_notify_only", lambda p: None)

    t = threading.Thread(target=watch_mod.watch, args=(tmp_path,), kwargs={"debounce": 0.2}, daemon=True)
    t.start()
    time.sleep(0.5)

    # Generate many events; loader must not be called again.
    for i in range(50):
        (tmp_path / "ignored" / f"f{i}.py").write_text("x\n", encoding="utf-8")
    time.sleep(0.7)
    assert calls["n"] == 1, f"_load_graphifyignore called {calls['n']} times; expected 1"


# --- _check_shrink: silent-corruption guard with explicit-deletion bypass ---

def _shrink_payload(n: int) -> dict:
    """Build a minimal graph-data dict with *n* placeholder nodes."""
    return {"nodes": [{"id": f"n{i}"} for i in range(n)], "links": []}


def test_check_shrink_blocks_silent_shrink(capsys):
    """Default case: smaller new graph + no force + no declared deletions = refuse."""
    ok = _check_shrink(
        force=False,
        existing_data=_shrink_payload(100),
        new_data=_shrink_payload(80),
    )
    assert ok is False
    captured = capsys.readouterr()
    assert "Refusing to overwrite" in captured.err
    assert "80 nodes" in captured.err and "100" in captured.err


def test_check_shrink_allows_force_override():
    """force=True bypasses the guard regardless of node delta."""
    ok = _check_shrink(
        force=True,
        existing_data=_shrink_payload(100),
        new_data=_shrink_payload(1),
    )
    assert ok is True


def test_check_shrink_allows_explicit_deletions(capsys):
    """Caller declared deletions → shrink is expected → guard skipped silently."""
    ok = _check_shrink(
        force=False,
        existing_data=_shrink_payload(100),
        new_data=_shrink_payload(80),
        had_explicit_deletions=True,
    )
    assert ok is True
    # And critically, no scary warning is printed when the shrink is intentional.
    assert "Refusing to overwrite" not in capsys.readouterr().err


def test_check_shrink_allows_no_existing_data():
    """First-run case: no existing graph → guard inert."""
    ok = _check_shrink(
        force=False,
        existing_data={},
        new_data=_shrink_payload(50),
    )
    assert ok is True


def test_check_shrink_allows_shrink_within_rebuilt_sources(capsys):
    """#1116: a symbol removed from a re-extracted file is a legitimate shrink —
    every lost node belongs to a rebuilt source, so the write proceeds (no --force)."""
    existing = {"nodes": [
        {"id": "a", "source_file": "m.py"},
        {"id": "b", "source_file": "m.py"},
        {"id": "c", "source_file": "other.py"},
    ], "links": []}
    new = {"nodes": [
        {"id": "a", "source_file": "m.py"},
        {"id": "c", "source_file": "other.py"},
    ], "links": []}
    ok = _check_shrink(False, existing, new, rebuilt_sources={"m.py"})
    assert ok is True
    assert "Refusing to overwrite" not in capsys.readouterr().err


def test_check_shrink_blocks_shrink_outside_rebuilt_sources(capsys):
    """The guard's real job is intact: a node lost from a file we did NOT re-extract
    (the failed-chunk signal) is still refused even with rebuilt_sources set."""
    existing = {"nodes": [
        {"id": "a", "source_file": "m.py"},
        {"id": "z", "source_file": "untouched.py"},
    ], "links": []}
    new = {"nodes": [{"id": "a", "source_file": "m.py"}], "links": []}
    ok = _check_shrink(False, existing, new, rebuilt_sources={"m.py"})
    assert ok is False
    assert "Refusing to overwrite" in capsys.readouterr().err


def test_check_shrink_allows_growth():
    """new > existing is always fine."""
    ok = _check_shrink(
        force=False,
        existing_data=_shrink_payload(50),
        new_data=_shrink_payload(60),
    )
    assert ok is True


def test_check_shrink_unlinks_tmp_on_refuse(tmp_path):
    """When refusing, the temp graph file gets cleaned up so it can't leak across runs."""
    tmp = tmp_path / "graph.tmp.json"
    tmp.write_text("{}", encoding="utf-8")
    ok = _check_shrink(
        force=False,
        existing_data=_shrink_payload(100),
        new_data=_shrink_payload(80),
        tmp=tmp,
    )
    assert ok is False
    assert not tmp.exists()


def test_check_shrink_keeps_tmp_when_deletions_declared(tmp_path):
    """Mirror of the above: if the caller declared deletions, the tmp file is NOT unlinked
    because the caller is going to swap it into place. Regression guard against a future
    bug where the tmp cleanup leaks out of the refuse branch.
    """
    tmp = tmp_path / "graph.tmp.json"
    tmp.write_text("{}", encoding="utf-8")
    ok = _check_shrink(
        force=False,
        existing_data=_shrink_payload(100),
        new_data=_shrink_payload(80),
        tmp=tmp,
        had_explicit_deletions=True,
    )
    assert ok is True
    assert tmp.exists()


# --- _rebuild_code integration: post-commit delete scenario ---

@pytest.mark.skipif(sys.platform == "win32", reason="git CLI behaviour varies on Windows runners")
def test_rebuild_code_prunes_deleted_file_nodes(tmp_path):
    """End-to-end probe of the post-commit-delete bug fix.

    Build a tiny graph, delete one of its source files, then call _rebuild_code
    with the deleted path in changed_paths. Without the fix this raises the
    shrink guard and refuses to write; with the fix the deleted file's nodes
    are pruned and graph.json is rewritten.
    """
    from graphify.watch import _rebuild_code

    # Set up a minimal "project" with two Python files in a git repo so detect
    # treats it as a real corpus.
    subprocess.run(["git", "init", "-q", str(tmp_path)], check=True)
    subprocess.run(
        ["git", "-C", str(tmp_path), "config", "user.email", "test@example.com"],
        check=True,
    )
    subprocess.run(
        ["git", "-C", str(tmp_path), "config", "user.name", "Test"],
        check=True,
    )

    keep = tmp_path / "keep.py"
    drop = tmp_path / "drop.py"
    keep.write_text("def keep_fn():\n    return 1\n", encoding="utf-8")
    drop.write_text("def drop_fn():\n    return 2\n", encoding="utf-8")

    # Initial build covers both files.
    cwd = os.getcwd()
    try:
        os.chdir(tmp_path)
        ok = _rebuild_code(tmp_path, no_cluster=True)
        assert ok is True
        graph_path = tmp_path / "graphify-out" / "graph.json"
        assert graph_path.exists()
        before = json.loads(graph_path.read_text(encoding="utf-8"))
        before_sources = {n.get("source_file") for n in before.get("nodes", [])}
        assert "drop.py" in before_sources

        # Now delete drop.py and re-run with it in the change list. This is what
        # the post-commit hook does when git diff --name-only HEAD~1 HEAD includes
        # a deletion: the path is passed to _rebuild_code even though it no
        # longer exists on disk.
        drop.unlink()
        ok = _rebuild_code(
            tmp_path,
            changed_paths=[Path("drop.py")],
            no_cluster=True,
        )
        assert ok is True, "rebuild should succeed even though the graph shrinks"

        after = json.loads(graph_path.read_text(encoding="utf-8"))
        after_sources = {n.get("source_file") for n in after.get("nodes", [])}
        assert "drop.py" not in after_sources, "deleted file's nodes should be pruned"
        assert "keep.py" in after_sources, "untouched file's nodes should survive"
    finally:
        os.chdir(cwd)


def test_rebuild_code_accepts_repo_relative_changed_path_for_subdir_root(tmp_path):
    """#1348: git-hook paths are repo-root-relative even when the graph root is a subdir."""
    from graphify.watch import _rebuild_code

    src = tmp_path / "src"
    src.mkdir()
    app = src / "app.py"
    app.write_text("def old_name():\n    return 1\n", encoding="utf-8")

    cwd = os.getcwd()
    try:
        os.chdir(tmp_path)
        assert _rebuild_code(Path("src"), no_cluster=True, acquire_lock=False) is True
        graph_path = src / "graphify-out" / "graph.json"
        before = json.loads(graph_path.read_text(encoding="utf-8"))
        assert "old_name()" in {n.get("label") for n in before.get("nodes", [])}

        app.write_text("def new_name():\n    return 2\n", encoding="utf-8")
        assert _rebuild_code(
            Path("src"),
            changed_paths=[Path("src/app.py")],
            no_cluster=True,
            acquire_lock=False,
            force=True,
        ) is True

        after = json.loads(graph_path.read_text(encoding="utf-8"))
        labels = {n.get("label") for n in after.get("nodes", [])}
        assert "old_name()" not in labels
        assert "new_name()" in labels
    finally:
        os.chdir(cwd)


@pytest.mark.parametrize(
    "changed_paths",
    [None, [Path("src/app.py")]],
    ids=["full-update", "incremental-update"],
)
def test_rebuild_code_subdir_preserves_outside_ast_nodes(tmp_path, changed_paths):
    """A full rebuild of a subdirectory must not prune graph data outside it."""
    from graphify.watch import _rebuild_code

    src = tmp_path / "src"
    src.mkdir()
    (tmp_path / "app.py").write_text("def outside_fn():\n    return 2\n", encoding="utf-8")
    (src / "app.py").write_text("def inside_fn():\n    return 1\n", encoding="utf-8")

    cwd = os.getcwd()
    try:
        os.chdir(tmp_path)
        assert _rebuild_code(Path("src"), no_cluster=True, acquire_lock=False) is True
        graph_path = src / "graphify-out" / "graph.json"
        data = json.loads(graph_path.read_text(encoding="utf-8"))
        inside_id = next(n["id"] for n in data["nodes"] if n.get("label") == "inside_fn()")
        outside_source = "app.py"
        data["nodes"].extend([
            {
                "id": "outside_ast",
                "label": "outside_fn()",
                "file_type": "function",
                "source_file": outside_source,
                "_origin": "ast",
            },
            {
                "id": "stale_inside_ast",
                "label": "stale_inside_fn()",
                "file_type": "function",
                "source_file": "src/deleted.py",
                "_origin": "ast",
            },
        ])
        data["links"].append({
            "source": "outside_ast",
            "target": inside_id,
            "relation": "calls",
            "source_file": outside_source,
        })
        graph_path.write_text(json.dumps(data), encoding="utf-8")

        assert _rebuild_code(
            Path("src"),
            changed_paths=changed_paths,
            no_cluster=True,
            acquire_lock=False,
        ) is True
        after = json.loads(graph_path.read_text(encoding="utf-8"))
        node_ids = {n["id"] for n in after["nodes"]}
        assert "outside_ast" in node_ids
        assert "stale_inside_ast" not in node_ids
        outside_node = next(n for n in after["nodes"] if n["id"] == "outside_ast")
        assert outside_node["source_file"] == outside_source
        outside_edge = next(
            e
            for e in after["links"]
            if e.get("source") == "outside_ast" and e.get("target") == inside_id
        )
        assert outside_edge["source_file"] == outside_source
    finally:
        os.chdir(cwd)


def test_rebuild_code_subdir_survives_absolute_to_relative_invocation(tmp_path):
    """Persisted source paths keep their meaning when invocation style changes."""
    from graphify.watch import _rebuild_code

    src = tmp_path / "src"
    src.mkdir()
    old = src / "old.py"
    old.write_text("def old_fn():\n    return 1\n", encoding="utf-8")

    cwd = os.getcwd()
    try:
        os.chdir(tmp_path)
        assert _rebuild_code(src, no_cluster=True, acquire_lock=False) is True
        graph_path = src / "graphify-out" / "graph.json"
        data = json.loads(graph_path.read_text(encoding="utf-8"))
        data["nodes"].append({
            "id": "local_semantic",
            "label": "LocalSemantic",
            "file_type": "concept",
            "source_file": "old.py",
        })
        graph_path.write_text(json.dumps(data), encoding="utf-8")

        assert _rebuild_code(Path("src"), no_cluster=True, acquire_lock=False) is True
        rebased = json.loads(graph_path.read_text(encoding="utf-8"))
        semantic = next(n for n in rebased["nodes"] if n["id"] == "local_semantic")
        assert semantic["source_file"] == "src/old.py"

        old.rename(src / "renamed.py")

        assert _rebuild_code(Path("src"), no_cluster=True, acquire_lock=False) is True
        after = json.loads(graph_path.read_text(encoding="utf-8"))
        sources = {n.get("source_file") for n in after["nodes"]}
        assert "old.py" not in sources
        assert "src/renamed.py" in sources
    finally:
        os.chdir(cwd)


def test_rebuild_code_prunes_legacy_watch_relative_subdir_source(tmp_path):
    """Pre-rebase subdirectory graphs stored source_file relative to watch_root."""
    from graphify.watch import _rebuild_code

    src = tmp_path / "src"
    src.mkdir()
    old = src / "old.py"
    old.write_text("def old_fn():\n    return 1\n", encoding="utf-8")

    cwd = os.getcwd()
    try:
        os.chdir(tmp_path)
        assert _rebuild_code(Path("src"), no_cluster=True, acquire_lock=False) is True
        graph_path = src / "graphify-out" / "graph.json"
        data = json.loads(graph_path.read_text(encoding="utf-8"))
        for item in data["nodes"] + data["links"]:
            source = item.get("source_file")
            if source and source.startswith("src/"):
                item["source_file"] = source.removeprefix("src/")
        graph_path.write_text(json.dumps(data), encoding="utf-8")

        old.rename(src / "renamed.py")
        assert _rebuild_code(
            Path("src"),
            changed_paths=[Path("src/renamed.py")],
            no_cluster=True,
            acquire_lock=False,
        ) is True

        after = json.loads(graph_path.read_text(encoding="utf-8"))
        sources = {n.get("source_file") for n in after["nodes"]}
        assert "old.py" not in sources
        assert "src/renamed.py" in sources
    finally:
        os.chdir(cwd)


def test_rebuild_code_does_not_update_root_marker_when_write_is_refused(tmp_path, monkeypatch):
    """A rejected candidate keeps the marker paired with the existing graph."""
    from graphify import watch as watch_mod

    src = tmp_path / "src"
    src.mkdir()
    app = src / "app.py"
    app.write_text("def before():\n    return 1\n", encoding="utf-8")

    cwd = os.getcwd()
    try:
        os.chdir(tmp_path)
        assert watch_mod._rebuild_code(src, no_cluster=True, acquire_lock=False) is True
        marker = src / "graphify-out" / ".graphify_root"
        assert marker.read_text(encoding="utf-8") == str(src)

        app.write_text("def after():\n    return 2\n", encoding="utf-8")
        monkeypatch.setattr(watch_mod, "_check_shrink", lambda *args, **kwargs: False)
        assert watch_mod._rebuild_code(
            Path("src"), no_cluster=True, acquire_lock=False
        ) is False
        assert marker.read_text(encoding="utf-8") == str(src)
    finally:
        os.chdir(cwd)


@pytest.mark.skipif(sys.platform == "win32", reason="symlink setup differs on Windows")
def test_rebuild_code_incremental_rename_preserves_symlink_source_path(tmp_path):
    """Changed files under followed symlinks retain their watched lexical path."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    real = corpus / "real"
    real.mkdir()
    (corpus / ".graphifyignore").write_text("real/\n", encoding="utf-8")
    old = real / "old.py"
    old.write_text("def linked_fn():\n    return 1\n", encoding="utf-8")
    (corpus / "linked").symlink_to(real, target_is_directory=True)

    assert _rebuild_code(
        corpus,
        follow_symlinks=True,
        no_cluster=True,
        acquire_lock=False,
    ) is True
    graph_path = corpus / "graphify-out" / "graph.json"

    first = real / "first.py"
    old.rename(first)
    assert _rebuild_code(
        corpus,
        changed_paths=[Path("linked/first.py")],
        follow_symlinks=True,
        no_cluster=True,
        acquire_lock=False,
    ) is True

    second = real / "second.py"
    first.rename(second)
    assert _rebuild_code(
        corpus,
        changed_paths=[Path("linked/second.py")],
        follow_symlinks=True,
        no_cluster=True,
        acquire_lock=False,
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    sources = {n.get("source_file") for n in after["nodes"]}
    assert "linked/old.py" not in sources
    assert "linked/first.py" not in sources
    assert "linked/second.py" in sources


# --- #1059: pending-changes queue prevents commit drops under lock contention ---


def test_queue_and_drain_pending_round_trip(tmp_path):
    """_queue_pending writes one path per line; _drain_pending reads + unlinks
    and returns the same set of paths."""
    from graphify.watch import _queue_pending, _drain_pending, _PENDING_FILENAME

    out = tmp_path / "graphify-out"
    paths = [Path("a.py"), Path("sub/b.py"), Path("c.md")]
    _queue_pending(out, paths)

    pending_file = out / _PENDING_FILENAME
    assert pending_file.exists()
    # Each path written on its own line.
    assert pending_file.read_text(encoding="utf-8").splitlines() == [
        "a.py", "sub/b.py", "c.md",
    ]

    drained = _drain_pending(out)
    assert drained == paths
    # Drain unlinks so subsequent callers see an empty queue.
    assert not pending_file.exists()
    assert _drain_pending(out) == []


def test_drain_pending_dedupes_and_skips_blank_lines(tmp_path):
    """Repeated appends across concurrent contenders must dedupe; partial
    writes leaving blank lines must not poison the merge."""
    from graphify.watch import _queue_pending, _drain_pending

    out = tmp_path / "graphify-out"
    _queue_pending(out, [Path("a.py"), Path("b.py")])
    _queue_pending(out, [Path("b.py"), Path("c.py")])
    # Simulate a torn write leaving an empty line.
    with open(out / ".pending_changes", "a", encoding="utf-8") as fh:
        fh.write("\n   \n")

    drained = _drain_pending(out)
    assert drained == [Path("a.py"), Path("b.py"), Path("c.py")]


def test_queue_pending_noop_on_empty_list(tmp_path):
    """Empty change set must not create an empty .pending_changes file."""
    from graphify.watch import _queue_pending, _PENDING_FILENAME

    out = tmp_path / "graphify-out"
    _queue_pending(out, [])
    assert not (out / _PENDING_FILENAME).exists()


@pytest.mark.skipif(sys.platform == "win32", reason="fcntl-only (POSIX)")
def test_rebuild_code_queues_on_lock_contention(tmp_path, monkeypatch, capsys):
    """#1059: when the rebuild lock is held, an incremental hook must queue
    its changed_paths to .pending_changes and print 'queued' instead of
    silently dropping the change set."""
    from graphify.watch import _rebuild_code, _rebuild_lock, _PENDING_FILENAME

    out = tmp_path / "graphify-out"
    out.mkdir()

    # Hold the lock so the next non-blocking attempt fails. Use a real
    # _rebuild_lock context manager in this same process — flock on the same
    # file descriptor would otherwise be re-entrant on Linux, so we open
    # the file ourselves via the lock helper.
    with _rebuild_lock(out, blocking=False) as outer_got:
        assert outer_got is True

        ok = _rebuild_code(
            tmp_path,
            changed_paths=[Path("a.py"), Path("b.py")],
        )
        assert ok is False

        # Output should say "queued", not "skipping".
        captured = capsys.readouterr().out
        assert "queued" in captured.lower()
        assert "skipping" not in captured.lower()

        # And the paths must have been written to the pending file so the
        # eventual lock-holder can drain them.
        pending = out / _PENDING_FILENAME
        assert pending.exists()
        assert pending.read_text(encoding="utf-8").splitlines() == ["a.py", "b.py"]


@pytest.mark.skipif(sys.platform == "win32", reason="fcntl-only (POSIX)")
def test_rebuild_code_merges_pending_on_acquire(tmp_path, monkeypatch):
    """#1059: the process that acquires the lock must drain .pending_changes
    and pass the merged change set to the inner rebuild call."""
    from graphify import watch as watch_mod

    out = tmp_path / "graphify-out"
    out.mkdir()
    # Pre-populate the queue as if an earlier contender had dropped its paths.
    watch_mod._queue_pending(out, [Path("queued1.py"), Path("queued2.py")])

    # Snapshot the original BEFORE monkeypatching so we can drive the outer
    # dispatch path while the inner recursive call resolves to our spy.
    orig_rebuild = watch_mod._rebuild_code
    inner_calls: list[list[str]] = []

    def recording_inner(watch_path, **kwargs):
        if kwargs.get("acquire_lock") is False:
            paths = kwargs.get("changed_paths") or []
            inner_calls.append([p.as_posix() for p in paths])
        return True

    monkeypatch.setattr(watch_mod, "_rebuild_code", recording_inner)

    ok = orig_rebuild(
        tmp_path,
        changed_paths=[Path("own.py"), Path("queued1.py")],
    )
    assert ok is True

    # The first inner call must have received the merged + deduped set:
    # own.py first (caller's order preserved), then drained queued1/queued2,
    # with queued1.py deduped against own's prior occurrence.
    assert inner_calls, "inner _rebuild_code should have been called"
    assert inner_calls[0] == ["own.py", "queued1.py", "queued2.py"]

    # And .pending_changes was drained.
    assert not (out / watch_mod._PENDING_FILENAME).exists()


@pytest.mark.skipif(sys.platform == "win32", reason="fcntl-only (POSIX)")
def test_rebuild_code_drains_late_arrivals(tmp_path, monkeypatch):
    """#1059: after the primary rebuild, the lock-holder must loop and drain
    any paths queued by hooks that arrived mid-rebuild."""
    from graphify import watch as watch_mod
    from graphify.watch import _rebuild_code as orig_rebuild

    out = tmp_path / "graphify-out"
    out.mkdir()

    inner_calls: list[list[str]] = []
    call_state = {"i": 0}

    def fake_inner(watch_path, **kwargs):
        if kwargs.get("acquire_lock") is False:
            paths = [p.as_posix() for p in (kwargs.get("changed_paths") or [])]
            inner_calls.append(paths)
            # Simulate a late-arriving hook that queues during the FIRST
            # inner rebuild only. The outer drain loop must see it.
            call_state["i"] += 1
            if call_state["i"] == 1:
                watch_mod._queue_pending(out, [Path("late.py")])
        return True

    monkeypatch.setattr(watch_mod, "_rebuild_code", fake_inner)

    ok = orig_rebuild(tmp_path, changed_paths=[Path("own.py")])
    assert ok is True

    # First inner call covers our own change set; second is the late-drain
    # pass that picks up "late.py".
    assert len(inner_calls) >= 2
    assert inner_calls[0] == ["own.py"]
    assert inner_calls[1] == ["late.py"]
    # And the queue is now empty (no further late drains).
    assert not (out / watch_mod._PENDING_FILENAME).exists()


def test_rebuild_code_full_corpus_skips_pending_queue(tmp_path, monkeypatch):
    """#1059: changed_paths=None means a full-corpus rebuild — the queue
    must not be touched on the failure path because there is nothing
    incremental to preserve."""
    from graphify import watch as watch_mod
    from graphify.watch import _rebuild_code as orig_rebuild

    out = tmp_path / "graphify-out"
    out.mkdir()

    # Pre-existing queued paths from an earlier incremental hook.
    watch_mod._queue_pending(out, [Path("earlier.py")])

    # Force the inner call to record what it saw.
    seen: list = []

    def fake_inner(watch_path, **kwargs):
        if kwargs.get("acquire_lock") is False:
            seen.append(kwargs.get("changed_paths"))
        return True

    monkeypatch.setattr(watch_mod, "_rebuild_code", fake_inner)

    ok = orig_rebuild(tmp_path, changed_paths=None)
    assert ok is True
    # Full-corpus rebuild passes None to the inner call (does not merge in
    # the queued paths — a full rebuild already covers them).
    assert seen == [None]
    # The queue still gets drained on entry so stale entries don't leak,
    # but no late-arrival loop runs for the full-corpus path.
    assert not (out / watch_mod._PENDING_FILENAME).exists()


def test_merge_changed_paths_dedupes_in_order():
    """_merge_changed_paths preserves first-seen order and drops dupes."""
    from graphify.watch import _merge_changed_paths

    merged = _merge_changed_paths(
        [Path("a.py"), Path("b.py")],
        None,
        [Path("b.py"), Path("c.py")],
        [Path("a.py")],
    )
    assert [p.as_posix() for p in merged] == ["a.py", "b.py", "c.py"]


def test_rebuild_code_preserves_nodes_from_excluded_but_alive_file(tmp_path, capsys):
    """Fail-closed eviction: a file that leaves the scan corpus but still exists
    on disk was EXCLUDED, not deleted — on an INCREMENTAL rebuild its nodes must
    survive a newly-honored .gitignore, with a loud message, instead of being
    silently mass-evicted as stale sources (the docs/brainstorms incident: an
    upgrade started honoring .gitignore and evicted 655 nodes whose files were
    present). .gitignore-driven eviction is deliberately gated on a full rebuild
    (#2495): only .graphifyignore/--exclude matches evict on the hook path.
    """
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    (corpus / "notes").mkdir(parents=True)
    (corpus / "auth.py").write_text("def login(): pass\n", encoding="utf-8")
    (corpus / "notes" / "brainstorm.md").write_text(
        "# Brainstorm\n\nA local-only design note.\n", encoding="utf-8"
    )

    assert _rebuild_code(corpus, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    labels = {n["label"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "brainstorm.md" in labels

    # The file becomes gitignored (leaves the corpus) but stays on disk.
    (corpus / ".gitignore").write_text("notes/\n", encoding="utf-8")
    capsys.readouterr()

    assert _rebuild_code(corpus, changed_paths=[Path("auth.py")], acquire_lock=False) is True
    labels = {n["label"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "brainstorm.md" in labels, (
        "nodes from a gitignored-but-alive file must be preserved on an incremental rebuild"
    )
    assert "fail-closed: kept" in capsys.readouterr().out


def test_rebuild_code_still_evicts_when_excluded_file_is_also_deleted(tmp_path):
    """The fail-closed preserve must not weaken true-deletion eviction: once the
    excluded file is actually gone from disk, its nodes are evicted as before."""
    import json
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    (corpus / "notes").mkdir(parents=True)
    (corpus / "auth.py").write_text("def login(): pass\n", encoding="utf-8")
    (corpus / "notes" / "brainstorm.md").write_text("# Brainstorm\n", encoding="utf-8")

    assert _rebuild_code(corpus, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"

    (corpus / "notes" / "brainstorm.md").unlink()

    assert _rebuild_code(corpus, changed_paths=[Path("auth.py")], acquire_lock=False) is True
    labels = {n["label"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "brainstorm.md" not in labels, "deleted file's nodes must still be evicted"
    assert "login()" in labels


# ── #2495: `graphify update` purges newly-ignored files that still exist ───────


def test_update_prunes_newly_graphifyignored_file(tmp_path, monkeypatch, capsys):
    """#2495: a file added to .graphifyignore while still on disk must be purged
    from the graph by a full `graphify update` — nodes, edges, and hyperedges —
    instead of being preserved forever by the fail-closed keep. The legitimate
    shrink passes without --force/GRAPHIFY_FORCE, unchanged in-corpus files stay
    byte-identical, and the prune is reported distinctly from the fail-closed
    keep line."""
    from graphify.watch import _rebuild_code

    monkeypatch.delenv("GRAPHIFY_FORCE", raising=False)
    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text("def alpha():\n    return 1\n", encoding="utf-8")
    (corpus / "b.py").write_text("def helper():\n    return 2\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    assert any(n.get("source_file") == "b.py" for n in data["nodes"])
    # Inject a hyperedge OWNED by b.py whose members all live in a.py, so only
    # the source-identity eviction (not member loss) can remove it.
    a_ids = [n["id"] for n in data["nodes"] if n.get("source_file") == "a.py"]
    assert a_ids
    data.setdefault("hyperedges", []).append(
        {"id": "h_b", "nodes": a_ids, "source_file": "b.py"}
    )
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    def _a_items(payload):
        return sorted(
            (
                json.dumps(item, sort_keys=True)
                for bucket in ("nodes", "links")
                for item in payload.get(bucket, [])
                if item.get("source_file") == "a.py"
            ),
        )

    a_before = _a_items(data)
    (corpus / ".graphifyignore").write_text("b.py\n", encoding="utf-8")
    capsys.readouterr()

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    for bucket in ("nodes", "links", "hyperedges"):
        assert not any(
            item.get("source_file") == "b.py" for item in after.get(bucket, [])
        ), f"{bucket} still reference newly-ignored b.py (#2495)"
    assert "h_b" not in {h.get("id") for h in after.get("hyperedges", [])}
    assert _a_items(after) == a_before, (
        "unchanged in-corpus file's items must survive the prune byte-identical"
    )
    out = capsys.readouterr().out
    assert "newly-ignored file(s)" in out
    assert "fail-closed: kept" not in out


def test_update_still_prunes_deleted_file_with_ignore_rules_present(tmp_path):
    """Regression guard for #2495: routing ignore-rule evidence through the
    corpus sweep must not weaken plain deletion eviction while unrelated ignore
    rules exist."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / ".graphifyignore").write_text("unrelated/\n", encoding="utf-8")
    (corpus / "a.py").write_text("def alpha():\n    return 1\n", encoding="utf-8")
    (corpus / "b.py").write_text("def helper():\n    return 2\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"

    (corpus / "b.py").unlink()

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    sources = {n.get("source_file") for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "b.py" not in sources, "deleted file's nodes must still be evicted"
    assert "a.py" in sources


def test_update_keeps_alive_file_dropped_from_corpus_without_ignore_rule(
    tmp_path, monkeypatch, capsys
):
    """#1795 retained under #2495: evicting an alive file now requires POSITIVE
    evidence — a live ignore rule matching it. A file that leaves the corpus for
    any other reason (filter regression, extractor loss) stays preserved, with
    the fail-closed message."""
    from graphify import detect as detect_mod
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text("def alpha():\n    return 1\n", encoding="utf-8")
    (corpus / "b.py").write_text("def helper():\n    return 2\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"

    real_detect = detect_mod.detect

    def narrowed(root, **kwargs):
        res = real_detect(root, **kwargs)
        res["files"]["code"] = [
            f for f in res["files"]["code"] if not str(f).endswith("b.py")
        ]
        return res

    monkeypatch.setattr(detect_mod, "detect", narrowed)
    capsys.readouterr()

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    sources = {n.get("source_file") for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "b.py" in sources, (
        "alive file dropped from the corpus with NO matching ignore rule must be kept"
    )
    out = capsys.readouterr().out
    assert "fail-closed: kept" in out
    assert "newly-ignored" not in out


def test_update_evicts_semantic_nodes_from_newly_ignored_non_ast_source(tmp_path, capsys):
    """#2495 extends #2051: a non-AST source (.txt) whose semantic nodes evict on
    disk absence must also evict when the file is alive but newly matched by a
    .graphifyignore rule."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text("def handle():\n    return 1\n", encoding="utf-8")
    (corpus / "notes.txt").write_text("Rationale to be ignored.\n", encoding="utf-8")
    (corpus / "kept.txt").write_text("Rationale that stays.\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    # No LLM in tests, so inject the semantic layer these .txt files would carry.
    data["nodes"].extend([
        {"id": "notes_concept", "label": "Notes Concept", "file_type": "concept",
         "source_file": "notes.txt"},
        {"id": "kept_concept", "label": "Kept Concept", "file_type": "concept",
         "source_file": "kept.txt"},
    ])
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    (corpus / ".graphifyignore").write_text("notes.txt\n", encoding="utf-8")
    capsys.readouterr()

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    after_ids = {n["id"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "notes_concept" not in after_ids, (
        "semantic node from a newly-ignored alive non-AST source must be evicted (#2495)"
    )
    assert "kept_concept" in after_ids
    assert "newly-ignored file(s)" in capsys.readouterr().out


def test_gitignore_eviction_gated_on_full_rebuild(tmp_path, capsys):
    """#2495 policy split: a .gitignore-only match is preserved fail-closed on an
    incremental rebuild (#1795's motivating case — a deliberately-graphed
    .gitignore'd tree survives the hook path), but an explicit full
    `graphify update` honors it and purges the file."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    generated = corpus / "generated"
    generated.mkdir(parents=True)
    (corpus / "app.py").write_text("def handle():\n    return 1\n", encoding="utf-8")
    (generated / "gen.py").write_text("def generated():\n    return 2\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    sources = {n.get("source_file") for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "generated/gen.py" in sources

    (corpus / ".gitignore").write_text("generated/\n", encoding="utf-8")
    capsys.readouterr()

    # Incremental (hook) rebuild: .gitignore-driven eviction is NOT honored.
    assert _rebuild_code(
        corpus, changed_paths=[Path("app.py")], no_cluster=True, acquire_lock=False
    ) is True
    sources = {n.get("source_file") for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "generated/gen.py" in sources, (
        ".gitignore-only match must be preserved on the incremental path"
    )
    assert "fail-closed: kept" in capsys.readouterr().out

    # Explicit full update: the .gitignore match is honored and purged.
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    sources = {n.get("source_file") for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "generated/gen.py" not in sources, (
        ".gitignore match must be purged by an explicit full update (#2495)"
    )
    assert "newly-ignored file(s)" in capsys.readouterr().out


def test_graphifyignore_eviction_applies_to_incremental_rebuilds(tmp_path, capsys):
    """#2495 policy split, other half: a .graphifyignore match is unambiguous
    graph-level intent and evicts on the incremental (hook) path too."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text("def handle():\n    return 1\n", encoding="utf-8")
    (corpus / "vendor.py").write_text("def vendored():\n    return 2\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"

    (corpus / ".graphifyignore").write_text("vendor.py\n", encoding="utf-8")
    capsys.readouterr()

    assert _rebuild_code(
        corpus, changed_paths=[Path("app.py")], no_cluster=True, acquire_lock=False
    ) is True
    sources = {n.get("source_file") for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "vendor.py" not in sources, (
        ".graphifyignore match must evict on an incremental rebuild too (#2495)"
    )
    assert "app.py" in sources
    assert "newly-ignored file(s)" in capsys.readouterr().out


# --- #1915: semantic-backed docs must not be double-represented by the AST quick-scan ---


_SEMANTIC_GUIDE_IDS = {"guide_doc", "auth_flow", "session_model"}
_AST_GUIDE_IDS = {"guide", "guide_overview", "guide_setup", "guide_usage"}


def _seed_semantic_doc_graph(corpus):
    """Build a code-only graph, then add guide.md represented ONLY semantically.

    Mimics a graph produced by the CLI ``graphify . --update`` path: code AST
    nodes plus a semantic (LLM) layer for the document — a ``<slug>_doc`` node
    and concept nodes, none carrying the ``_origin`` marker — and NO AST
    heading nodes for the doc.
    """
    from graphify.watch import _rebuild_code

    corpus.mkdir()
    (corpus / "app.py").write_text(
        "def handle_login():\n    return 1\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    (corpus / "guide.md").write_text(
        "# Overview\n\nIntro.\n\n## Setup\n\nSteps.\n\n## Usage\n\nMore.\n",
        encoding="utf-8",
    )
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    code_node_id = next(
        n["id"] for n in data["nodes"] if n.get("source_file") == "app.py"
    )
    data["nodes"].extend([
        {"id": "guide_doc", "label": "Guide", "file_type": "document",
         "source_file": "guide.md"},
        {"id": "auth_flow", "label": "Auth Flow", "file_type": "concept",
         "source_file": "guide.md"},
        {"id": "session_model", "label": "Session Model", "file_type": "concept",
         "source_file": "guide.md"},
    ])
    data["links"].extend([
        {"source": "guide_doc", "target": "auth_flow", "relation": "explains",
         "confidence": "INFERRED", "source_file": "guide.md"},
        {"source": "auth_flow", "target": code_node_id,
         "relation": "implemented_by", "confidence": "INFERRED",
         "source_file": "guide.md"},
    ])
    graph_path.write_text(json.dumps(data), encoding="utf-8")
    return graph_path


_CONCEPT_ONLY_GUIDE_IDS = {"auth_flow", "session_model"}


def _seed_semantic_doc_graph_concept_only(corpus):
    """Like ``_seed_semantic_doc_graph``, but guide.md's semantic layer is
    ONLY concept/rationale nodes (no ``file_type=="document"`` node) — the
    extraction spec's preferred shape for a doc full of named concepts (#1954).
    """
    from graphify.watch import _rebuild_code

    corpus.mkdir()
    (corpus / "app.py").write_text(
        "def handle_login():\n    return 1\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    (corpus / "guide.md").write_text(
        "# Overview\n\nIntro.\n\n## Setup\n\nSteps.\n\n## Usage\n\nMore.\n",
        encoding="utf-8",
    )
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    code_node_id = next(
        n["id"] for n in data["nodes"] if n.get("source_file") == "app.py"
    )
    data["nodes"].extend([
        {"id": "auth_flow", "label": "Auth Flow", "file_type": "concept",
         "source_file": "guide.md"},
        {"id": "session_model", "label": "Session Model", "file_type": "rationale",
         "source_file": "guide.md"},
    ])
    data["links"].extend([
        {"source": "auth_flow", "target": "session_model", "relation": "explains",
         "confidence": "INFERRED", "source_file": "guide.md"},
        {"source": "auth_flow", "target": code_node_id,
         "relation": "implemented_by", "confidence": "INFERRED",
         "source_file": "guide.md"},
    ])
    graph_path.write_text(json.dumps(data), encoding="utf-8")
    return graph_path


def test_rebuild_code_semantic_doc_not_double_represented_on_full_rebuild(tmp_path):
    """#1915: a full _rebuild_code must not AST-quick-scan a doc whose semantic
    (LLM) nodes already represent it. Before the fix the quick-scan minted
    heading nodes ON TOP of the preserved semantic nodes, representing every
    doc twice (~4x bloated graph vs the CLI update path)."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_semantic_doc_graph(corpus)
    before = json.loads(graph_path.read_text(encoding="utf-8"))

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_ids = {n["id"] for n in after["nodes"]}
    assert _SEMANTIC_GUIDE_IDS <= after_ids, "semantic doc nodes must be preserved"
    assert not (_AST_GUIDE_IDS & after_ids), (
        "AST heading nodes minted for a semantic-backed doc (#1915)"
    )
    assert len(after["nodes"]) == len(before["nodes"]), (
        f"node count inflated {len(before['nodes'])} -> {len(after['nodes'])} (#1915)"
    )


def test_rebuild_code_concept_only_semantic_doc_not_double_represented_on_full_rebuild(
    tmp_path,
):
    """#1954: a doc represented ONLY by concept/rationale nodes (no
    file_type=="document" node) must also be recognized as semantic-backed
    and skipped by the AST quick-scan — not just docs with a "document" node."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_semantic_doc_graph_concept_only(corpus)
    before = json.loads(graph_path.read_text(encoding="utf-8"))

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_ids = {n["id"] for n in after["nodes"]}
    assert _CONCEPT_ONLY_GUIDE_IDS <= after_ids, "semantic doc nodes must be preserved"
    assert not (_AST_GUIDE_IDS & after_ids), (
        "AST heading nodes minted for a concept-only semantic-backed doc (#1954)"
    )
    assert len(after["nodes"]) == len(before["nodes"]), (
        f"node count inflated {len(before['nodes'])} -> {len(after['nodes'])} (#1954)"
    )


@pytest.mark.parametrize(
    "changed",
    [[Path("guide.md")], [Path("guide.md"), Path("app.py")]],
    ids=["doc-only", "doc-plus-code"],
)
def test_rebuild_code_incremental_preserves_semantic_doc_nodes_and_edges(
    tmp_path, changed
):
    """#1915: an incremental rebuild whose change set includes a semantic-backed
    doc must not wipe the doc's semantic nodes or their edges — re-extraction
    owns only a source's AST tier (node-level mirror of #1865's edge rule)."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_semantic_doc_graph(corpus)

    assert _rebuild_code(
        corpus, changed_paths=changed, no_cluster=True, acquire_lock=False
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_ids = {n["id"] for n in after["nodes"]}
    assert _SEMANTIC_GUIDE_IDS <= after_ids, (
        "semantic doc nodes wiped by an incremental rebuild"
    )
    relations = {
        (e.get("source"), e.get("target"), e.get("relation"))
        for e in after["links"]
    }
    assert ("guide_doc", "auth_flow", "explains") in relations, (
        "semantic doc edge dropped by an incremental rebuild"
    )
    assert any(
        src == "auth_flow" and rel == "implemented_by"
        for src, _tgt, rel in relations
    ), "doc-to-code semantic edge dropped by an incremental rebuild"
    assert not (_AST_GUIDE_IDS & after_ids), (
        "incremental rebuild AST-quick-scanned a semantic-backed doc (#1915)"
    )


@pytest.mark.parametrize(
    "changed",
    [[Path("guide.md")], [Path("guide.md"), Path("app.py")]],
    ids=["doc-only", "doc-plus-code"],
)
def test_rebuild_code_incremental_preserves_concept_only_semantic_doc_nodes_and_edges(
    tmp_path, changed
):
    """#1954: incremental analogue — a concept/rationale-only semantic doc
    must not lose its nodes/edges nor get AST-quick-scanned on an incremental
    rebuild, mirroring the #1915 doc-node case above."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_semantic_doc_graph_concept_only(corpus)

    assert _rebuild_code(
        corpus, changed_paths=changed, no_cluster=True, acquire_lock=False
    ) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_ids = {n["id"] for n in after["nodes"]}
    assert _CONCEPT_ONLY_GUIDE_IDS <= after_ids, (
        "concept-only semantic doc nodes wiped by an incremental rebuild"
    )
    relations = {
        (e.get("source"), e.get("target"), e.get("relation"))
        for e in after["links"]
    }
    assert ("auth_flow", "session_model", "explains") in relations, (
        "concept-only semantic doc edge dropped by an incremental rebuild"
    )
    assert any(
        src == "auth_flow" and rel == "implemented_by"
        for src, _tgt, rel in relations
    ), "doc-to-code semantic edge dropped by an incremental rebuild"
    assert not (_AST_GUIDE_IDS & after_ids), (
        "incremental rebuild AST-quick-scanned a concept-only semantic-backed doc (#1954)"
    )


def test_rebuild_code_quick_scans_doc_without_semantic_nodes(tmp_path):
    """#09b33b7 guard: a doc with NO semantic layer still gets the AST
    quick-scan so no-LLM corpora keep their heading structure — #1915's
    semantic-supersedes-AST rule must not regress the fallback."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text("def f():\n    return 1\n", encoding="utf-8")
    (corpus / "notes.md").write_text("# Alpha\n\n## Beta\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    ids = {n["id"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert {"notes", "notes_alpha", "notes_beta"} <= ids

    # A rebuild over the existing graph (still no semantic nodes for the doc)
    # keeps quick-scanning it rather than dropping its structure.
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    ids = {n["id"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert {"notes", "notes_alpha", "notes_beta"} <= ids


def test_full_rebuild_preserves_semantic_backed_doc_ast_layer(tmp_path):
    """#2333 (COEXIST): a doc that carries BOTH an AST heading layer and a
    semantic layer keeps both across a full rebuild. The doc is excluded from
    extract_targets (#1915, no re-quick-scan), so its AST nodes are NOT
    regenerated this run — the full-rebuild ownership rule must therefore not
    drop them (it owns only rebuilt sources, not everything in watch_root).
    Supersedes the pre-COEXIST #1915 self-heal, which deleted the AST layer."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text(
        "def handle_login():\n    return 1\n", encoding="utf-8"
    )
    (corpus / "guide.md").write_text(
        "# Overview\n\n## Setup\n\n## Usage\n", encoding="utf-8"
    )
    # Initial build quick-scans guide.md (no semantic layer yet): AST nodes.
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    assert _AST_GUIDE_IDS <= {n["id"] for n in data["nodes"]}
    doc_nodes_before = sum(
        1 for n in data["nodes"] if n.get("source_file") == "guide.md"
    )

    # Layer the semantic representation on top — both tiers now coexist.
    data["nodes"].append(
        {"id": "auth_flow", "label": "Auth Flow", "file_type": "concept",
         "source_file": "guide.md"},
    )
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # Full rebuild WITHOUT force: guide.md is now semantic-backed, so it is
    # excluded from extract_targets — its existing AST layer must survive.
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_ids = {n["id"] for n in after["nodes"]}
    assert "auth_flow" in after_ids, "semantic layer lost on full rebuild"
    assert _AST_GUIDE_IDS <= after_ids, (
        "AST heading layer of a semantic-backed doc dropped by a full "
        "rebuild (#2333 COEXIST)"
    )
    doc_nodes_after = sum(
        1 for n in after["nodes"]
        if n.get("source_file") == "guide.md" and n["id"] != "auth_flow"
    )
    assert doc_nodes_after == doc_nodes_before, (
        "document-node count changed across a full rebuild of a "
        f"semantic-backed doc: {doc_nodes_before} -> {doc_nodes_after}"
    )


def test_full_rebuild_regenerates_docs_with_legacy_unstamped_nodes(tmp_path):
    """#2334: a legacy heading node without the _origin stamp (pre-0.9.16
    graph) must not fake a semantic layer — _is_ast_tier's shape fallback
    (source_location "L<n>") classifies it as AST, so the doc stays in
    extract_targets, is re-quick-scanned, and comes back fully re-stamped."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text(
        "def handle_login():\n    return 1\n", encoding="utf-8"
    )
    (corpus / "guide.md").write_text(
        "# Overview\n\n## Setup\n\n## Usage\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    assert _AST_GUIDE_IDS <= {n["id"] for n in data["nodes"]}

    # Simulate a legacy graph: strip the _origin stamp from one heading node.
    stripped = next(
        n for n in data["nodes"] if n["id"] == "guide_overview"
    )
    stripped.pop("_origin", None)
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_by_id = {n["id"]: n for n in after["nodes"]}
    assert _AST_GUIDE_IDS <= set(after_by_id), (
        "legacy unstamped heading node made the doc look semantic-backed "
        "and its AST structure was dropped (#2334)"
    )
    for nid in _AST_GUIDE_IDS:
        assert after_by_id[nid].get("_origin") == "ast", (
            f"{nid} not re-stamped with _origin=ast after the full rebuild"
        )


def test_full_rebuild_drops_stale_ast_for_reextracted_code(tmp_path):
    """#1116 guard: tier-scoping the full-rebuild ownership rule (#2333) must
    not stop a genuinely re-extracted code file from shedding its stale AST
    nodes — a renamed function's old symbol node still disappears."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text(
        "def old_name():\n    return 1\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    ids = {n["id"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "app_old_name" in ids

    (corpus / "app.py").write_text(
        "def new_name():\n    return 1\n", encoding="utf-8"
    )
    # Full rebuild (no changed_paths): app.py is re-extracted, so its stale
    # AST symbol node is owned by this run and must be dropped.
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    ids = {n["id"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "app_new_name" in ids
    assert "app_old_name" not in ids, (
        "stale AST node for a re-extracted code file survived (#1116)"
    )


# ── #2014: code-typed semantic nodes count as a doc's semantic layer ───────────

_CODE_ONLY_GUIDE_IDS = {"parse_config", "load_settings"}


def _seed_semantic_doc_graph_code_only(corpus):
    """Like ``_seed_semantic_doc_graph``, but guide.md's semantic layer is ONLY
    code-typed nodes — symbols the LLM surfaced from WITHIN the doc (llm.py
    ``_bind_node_evidence``), with no document/concept node at all (#2014)."""
    from graphify.watch import _rebuild_code

    corpus.mkdir()
    (corpus / "app.py").write_text(
        "def handle_login():\n    return 1\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    (corpus / "guide.md").write_text(
        "# Overview\n\nIntro.\n\n## Setup\n\nSteps.\n\n## Usage\n\nMore.\n",
        encoding="utf-8",
    )
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    code_node_id = next(
        n["id"] for n in data["nodes"] if n.get("source_file") == "app.py"
    )
    data["nodes"].extend([
        {"id": "parse_config", "label": "parse_config()", "file_type": "code",
         "source_file": "guide.md"},
        {"id": "load_settings", "label": "load_settings()", "file_type": "code",
         "source_file": "guide.md"},
    ])
    data["links"].append({
        "source": "parse_config", "target": code_node_id,
        "relation": "implemented_by", "confidence": "INFERRED",
        "source_file": "guide.md",
    })
    graph_path.write_text(json.dumps(data), encoding="utf-8")
    return graph_path


def test_rebuild_code_code_only_semantic_doc_not_double_represented_on_full_rebuild(
    tmp_path,
):
    """#2014: a doc represented ONLY by code-typed semantic nodes (symbols
    surfaced from within it) must be recognized as semantic-backed and skipped
    by the AST quick-scan. Before the fix "code" was absent from the semantic
    file_type gate, so the doc was re-AST-scanned — minting heading nodes AND
    dropping the code-typed semantic nodes (they belonged to a now-rebuilt
    source), silently losing them."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_semantic_doc_graph_code_only(corpus)

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_ids = {n["id"] for n in after["nodes"]}
    assert _CODE_ONLY_GUIDE_IDS <= after_ids, (
        "code-typed semantic doc nodes dropped by a full rebuild (#2014)"
    )
    assert not (_AST_GUIDE_IDS & after_ids), (
        "AST heading nodes minted for a code-only semantic-backed doc (#2014)"
    )


# ── #2051: deleted non-AST sources (docs/papers/images) get evicted ────────────

def test_rebuild_code_evicts_semantic_nodes_from_deleted_non_ast_source(tmp_path):
    """#2051: a full `graphify update` must evict semantic nodes whose non-AST
    source file (a .txt/.pdf/.png with no code extractor) was deleted from disk.
    The corpus sweep used to skip every sourceless-of-extractor node, so those
    nodes survived forever and were served as authoritative long after the file
    was gone. Disk absence is the only deletion evidence for such sources."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text("def handle():\n    return 1\n", encoding="utf-8")
    # Two non-AST semantic sources: one stays on disk, one gets deleted.
    (corpus / "kept.txt").write_text("Design rationale that stays.\n", encoding="utf-8")
    (corpus / "gone.txt").write_text("Rationale that will be deleted.\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    # No LLM in tests, so inject the semantic layer these .txt files would carry.
    data["nodes"].extend([
        {"id": "kept_concept", "label": "Kept Concept", "file_type": "concept",
         "source_file": "kept.txt"},
        {"id": "gone_concept", "label": "Gone Concept", "file_type": "concept",
         "source_file": "gone.txt"},
    ])
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # Delete one non-AST source; the other stays.
    (corpus / "gone.txt").unlink()

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    after_ids = {n["id"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "gone_concept" not in after_ids, (
        "semantic node from a deleted non-AST source must be evicted (#2051)"
    )
    assert "kept_concept" in after_ids, (
        "semantic node from a surviving non-AST source must be preserved"
    )


def test_rebuild_code_preserves_remote_source_across_repeated_updates(tmp_path):
    """#2051 follow-up: a node whose source_file is a URL/virtual scheme
    (gdoc://, s3://, http://) must survive REPEATED `graphify update`s. Path
    normalization on the write side collapses the double slash (`gdoc://x` ->
    `gdoc:/x`), so a literal `"://"` guard matched on the first update but missed
    on the second, dropping the node into the disk-absence eviction branch
    (Path('gdoc:/x').exists() is False) — a data-loss regression from the #2051
    disk-absence sweep. The scheme is now matched with a regex tolerant of the
    collapse."""
    from graphify.watch import _rebuild_code, _is_remote_source

    # unit-level: the guard tolerates the slash collapse and rejects local paths
    assert _is_remote_source("gdoc://abc")
    assert _is_remote_source("gdoc:/abc")        # collapsed form
    assert _is_remote_source("s3://bucket/key")
    assert _is_remote_source("https://example.com/doc")
    assert not _is_remote_source("src/app.py")
    assert not _is_remote_source("notes.txt")
    assert not _is_remote_source("C:/Users/x/a.py")  # Windows drive != scheme

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text("def handle():\n    return 1\n", encoding="utf-8")
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    data["nodes"].append(
        {"id": "remote_doc", "label": "Remote Spec", "file_type": "document",
         "source_file": "gdoc://team/spec"}
    )
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # Three consecutive full updates: the remote node must persist through every
    # one, even after its stored source_file is normalized to the collapsed form.
    for i in range(3):
        assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
        after = json.loads(graph_path.read_text(encoding="utf-8"))
        ids = {n["id"] for n in after["nodes"]}
        assert "remote_doc" in ids, f"remote-source node evicted on update #{i + 1} (#2051 follow-up)"


# ── #2056: present-but-unextractable files in a change set are not deletions ───

def test_rebuild_code_incremental_preserves_present_non_ast_source(tmp_path):
    """#2056: an incremental rebuild whose change set names a file that exists but
    has no AST extractor (a doc/paper/image, or an excluded path) must NOT treat
    it as deleted. The old change-set loop routed any present-but-untracked file
    to _add_deleted_source, evicting its semantic nodes AND flipping
    had_explicit_deletions so the shrink guard waved the loss through."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "app.py").write_text("def handle():\n    return 1\n", encoding="utf-8")
    (corpus / "spec.txt").write_text("A spec with a semantic layer.\n", encoding="utf-8")

    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    data["nodes"].append(
        {"id": "spec_concept", "label": "Spec Concept", "file_type": "concept",
         "source_file": "spec.txt"}
    )
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # spec.txt is present but not AST-extractable; app.py is a real code change.
    assert _rebuild_code(
        corpus, changed_paths=[Path("spec.txt"), Path("app.py")],
        no_cluster=True, acquire_lock=False,
    ) is True

    after_ids = {n["id"] for n in json.loads(graph_path.read_text(encoding="utf-8"))["nodes"]}
    assert "spec_concept" in after_ids, (
        "present-but-unextractable file in change set wrongly evicted as deleted (#2056)"
    )


# --- #2251: fail-closed load of the existing graph ---


def _seed_graph_with_semantic_layer(corpus: Path) -> Path:
    """Build a real graph for one code file, then inject two semantic nodes
    (no _origin marker) sourced from a non-AST notes.txt, plus a semantic link.
    Returns the graph.json path."""
    from graphify.watch import _rebuild_code

    corpus.mkdir()
    (corpus / "a.py").write_text("def alpha():\n    return 1\n", encoding="utf-8")
    (corpus / "notes.txt").write_text("Design notes with a semantic layer.\n",
                                      encoding="utf-8")
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    assert data["nodes"], "seed rebuild must produce AST code nodes"
    data["nodes"].extend([
        {"id": "notes_doc", "label": "Design Notes", "file_type": "document",
         "source_file": "notes.txt"},
        {"id": "notes_concept", "label": "Design Concept", "file_type": "concept",
         "source_file": "notes.txt"},
    ])
    data["links"].append({
        "source": "notes_concept", "target": "notes_doc",
        "relation": "described_in", "confidence": "INFERRED",
        "source_file": "notes.txt",
    })
    graph_path.write_text(json.dumps(data), encoding="utf-8")
    return graph_path


def test_rebuild_refuses_overwrite_when_existing_graph_over_size_cap(
    tmp_path, monkeypatch, capsys
):
    """#2251: an existing graph.json over GRAPHIFY_MAX_GRAPH_BYTES could not be
    READ, which is not license to overwrite it. The old code swallowed the cap
    ValueError, treated the baseline as empty, and collapsed the graph to
    code-only output."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_graph_with_semantic_layer(corpus)
    before = graph_path.read_bytes()
    assert len(before) > 100

    monkeypatch.setenv("GRAPHIFY_MAX_GRAPH_BYTES", "100")
    assert _rebuild_code(
        corpus, changed_paths=[Path("a.py")], no_cluster=True, acquire_lock=False,
    ) is False
    assert graph_path.read_bytes() == before, (
        "graph.json must be byte-identical after a refused over-cap rebuild"
    )
    assert "error:" in capsys.readouterr().err


@pytest.mark.parametrize("no_cluster", [True, False],
                         ids=["no-cluster", "clustered"])
def test_rebuild_refuses_overwrite_when_existing_graph_corrupt(
    tmp_path, capsys, no_cluster
):
    """#2251: unparseable graph.json (e.g. truncated by a crash) must fail the
    rebuild closed on both write paths, not be overwritten as if absent."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_graph_with_semantic_layer(corpus)
    truncated = graph_path.read_text(encoding="utf-8")[:40]
    graph_path.write_text(truncated, encoding="utf-8")

    assert _rebuild_code(
        corpus, changed_paths=[Path("a.py")],
        no_cluster=no_cluster, acquire_lock=False,
    ) is False
    assert graph_path.read_text(encoding="utf-8") == truncated, (
        "corrupt graph.json must be left untouched for manual recovery"
    )
    assert "error:" in capsys.readouterr().err


def test_rebuild_force_does_not_clobber_unreadable_graph(tmp_path, capsys):
    """#2251: --force means \"accept a shrink\", not \"overwrite a graph that
    could not be read\"."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_graph_with_semantic_layer(corpus)
    truncated = graph_path.read_text(encoding="utf-8")[:40]
    graph_path.write_text(truncated, encoding="utf-8")

    assert _rebuild_code(
        corpus, changed_paths=[Path("a.py")], force=True,
        no_cluster=True, acquire_lock=False,
    ) is False
    assert graph_path.read_text(encoding="utf-8") == truncated
    assert "error:" in capsys.readouterr().err


def test_rebuild_readable_graph_still_preserves_semantic_nodes(tmp_path):
    """Happy-path regression for the #2251 fix: a valid existing graph under the
    default cap still reconciles — the rebuild succeeds and the semantic layer
    survives an incremental code-only update."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    graph_path = _seed_graph_with_semantic_layer(corpus)

    assert _rebuild_code(
        corpus, changed_paths=[Path("a.py")], no_cluster=True, acquire_lock=False,
    ) is True
    after = json.loads(graph_path.read_text(encoding="utf-8"))
    after_ids = {n["id"] for n in after["nodes"]}
    assert {"notes_doc", "notes_concept"} <= after_ids, (
        "semantic nodes must survive an incremental rebuild with a readable graph"
    )
    assert any(
        e.get("source") == "notes_concept" and e.get("target") == "notes_doc"
        for e in after["links"]
    ), "semantic link must survive an incremental rebuild"


# --- #2342: `graphify update` rebuild must inherit the on-disk directed flag ---

def test_rebuild_code_inherits_directed_flag_clustered(tmp_path):
    """#2342: the clustered `graphify update` rebuild path built the graph via
    build_from_json(result) with no directed= argument, so it always took the
    directed=False default and silently downgraded an existing directed graph.
    """
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text(
        "def alpha():\n    return beta()\n\ndef beta():\n    return 1\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, acquire_lock=False) is True

    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    data["directed"] = True
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # Add a NEW function and a call to it, so the rebuild genuinely changes
    # topology. Editing only a function BODY leaves the node/edge set identical,
    # so the same_topology check short-circuits before graph.json is rewritten
    # and the seeded flag survives untouched — the assertion below would then
    # pass without the fix ever running.
    n_before = len(json.loads(graph_path.read_text(encoding="utf-8"))["nodes"])
    (corpus / "a.py").write_text(
        "def alpha():\n    return beta() + gamma()\n\n"
        "def beta():\n    return 1\n\ndef gamma():\n    return 2\n",
        encoding="utf-8",
    )
    assert _rebuild_code(corpus, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    assert len(after["nodes"]) > n_before, (
        "guard: graph.json must actually have been rewritten, otherwise the "
        "directed assertion below is vacuous"
    )
    assert after.get("directed") is True, (
        "graphify update (clustered) must preserve an existing directed=True graph"
    )
    edge = next(
        e for e in after["links"]
        if e.get("relation") == "calls"
    )
    assert edge["source"] == "a_alpha" and edge["target"] == "a_beta", (
        "directed calls edge must still read alpha -> beta, not reversed"
    )


def test_rebuild_code_inherits_directed_flag_no_cluster(tmp_path):
    """#2342, --no-cluster path: candidate_graph_data was built straight from the
    raw merged extraction (`result`), which never carries a directed key, so the
    written graph.json silently lost the flag on every no-cluster update."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text("def f(): pass\n", encoding="utf-8")
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    data["directed"] = True
    graph_path.write_text(json.dumps(data), encoding="utf-8")

    # A new top-level function, not just a body edit, so the candidate graph
    # really differs and the same_graph check cannot short-circuit the write
    # (which would leave the seeded flag in place and make this vacuous).
    n_before = len(json.loads(graph_path.read_text(encoding="utf-8"))["nodes"])
    (corpus / "a.py").write_text("def f(): pass\n\ndef g(): pass\n", encoding="utf-8")
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    assert len(after["nodes"]) > n_before, (
        "guard: graph.json must actually have been rewritten, otherwise the "
        "directed assertion below is vacuous"
    )
    assert after.get("directed") is True, (
        "graphify update --no-cluster must preserve an existing directed=True graph"
    )


def test_rebuild_code_keeps_undirected_graph_undirected(tmp_path):
    """An existing undirected graph (no directed key, the on-disk default) must
    not be spuriously flipped to directed=True by an update rebuild."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text("def f(): pass\n", encoding="utf-8")
    assert _rebuild_code(corpus, acquire_lock=False) is True

    graph_path = corpus / "graphify-out" / "graph.json"
    before = json.loads(graph_path.read_text(encoding="utf-8"))
    assert before.get("directed", False) is False

    (corpus / "a.py").write_text("def f(): return 1\n", encoding="utf-8")
    assert _rebuild_code(corpus, acquire_lock=False) is True

    after = json.loads(graph_path.read_text(encoding="utf-8"))
    assert after.get("directed", False) is False, (
        "an undirected graph must not be spuriously flipped to directed by update"
    )


def test_rebuild_code_fresh_build_defaults_undirected(tmp_path):
    """No existing graph at all (first build via _rebuild_code) must still
    default to directed=False - #2342's fix only inherits, never invents True."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "a.py").write_text("def f(): pass\n", encoding="utf-8")
    assert _rebuild_code(corpus, acquire_lock=False) is True

    graph_path = corpus / "graphify-out" / "graph.json"
    data = json.loads(graph_path.read_text(encoding="utf-8"))
    assert data.get("directed", False) is False


def test_incremental_rebuild_preserves_call_to_unchanged_typescript_target(tmp_path):
    """#2406: rebuilding a changed caller retains its SHARED DIRECT calls into unchanged files.

    Scope here is the shared direct-call pass (a plain `shared()`); member calls
    (#2437) and indirect_call (#2438) are covered by the tests at the end of
    this file.

    A later edit that removes the call must still remove the edge, preventing a
    stale-edge-preservation workaround.
    """
    import json

    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()

    target = corpus / "B.ts"
    caller = corpus / "A.ts"

    target.write_text(
        """
export function shared(): number {
  return 1;
}
""".lstrip(),
        encoding="utf-8",
    )
    caller.write_text(
        """
import { shared } from "./B";

export function run(): number {
  return shared();
}
""".lstrip(),
        encoding="utf-8",
    )

    graph_path = corpus / "graphify-out" / "graph.json"

    def load_graph():
        return json.loads(graph_path.read_text(encoding="utf-8"))

    def node_id(graph, label, source_file):
        return next(
            node["id"]
            for node in graph.get("nodes", [])
            if node.get("label") == label
            and node.get("source_file") == source_file
        )

    def has_call(graph):
        run_id = node_id(graph, "run()", "A.ts")
        shared_id = node_id(graph, "shared()", "B.ts")
        return any(
            edge.get("relation") == "calls"
            and edge.get("source") == run_id
            and edge.get("target") == shared_id
            for edge in graph.get("links", graph.get("edges", []))
        )

    # Full-corpus baseline resolves A.run() -> B.shared().
    assert _rebuild_code(
        corpus,
        no_cluster=True,
        acquire_lock=False,
    ) is True
    assert has_call(load_graph()), "full rebuild must create the cross-file call edge"

    # Change only the caller while retaining the same call. The unchanged target
    # must remain available to the cross-file resolver.
    caller.write_text(
        """
import { shared } from "./B";

export function run(): number {
  return shared() + 1;
}
""".lstrip(),
        encoding="utf-8",
    )
    assert _rebuild_code(
        corpus,
        changed_paths=[caller],
        no_cluster=True,
        acquire_lock=False,
    ) is True
    assert has_call(
        load_graph()
    ), "incremental rebuild dropped the call edge to an unchanged target"

    # Removing the call must remove the edge; do not merely preserve old outgoing
    # edges from changed files.
    caller.write_text(
        """
export function run(): number {
  return 1;
}
""".lstrip(),
        encoding="utf-8",
    )
    assert _rebuild_code(
        corpus,
        changed_paths=[caller],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    final_graph = load_graph()
    run_id = node_id(final_graph, "run()", "A.ts")
    assert not any(
        edge.get("relation") == "calls"
        and edge.get("source") == run_id
        for edge in final_graph.get("links", final_graph.get("edges", []))
    ), "removed call must not survive as a stale edge"


# --- #2406 incremental shared-direct-call resolution helpers -----------------

def _2406_graph(corpus):
    import json
    return json.loads(
        (corpus / "graphify-out" / "graph.json").read_text(encoding="utf-8")
    )


def _2406_nid(graph, label, source_file):
    return next(
        (
            node["id"]
            for node in graph.get("nodes", [])
            if node.get("label") == label and node.get("source_file") == source_file
        ),
        None,
    )


def _2406_calls(graph):
    """(source_id, target_id) of every `calls` edge."""
    return [
        (edge.get("source"), edge.get("target"))
        for edge in graph.get("links", graph.get("edges", []))
        if edge.get("relation") == "calls"
    ]


def _2406_seed(tmp_path, caller_src, target_src="export function shared(): number {\n  return 1;\n}\n"):
    """Build a two-file TS corpus and do the initial full rebuild."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir(parents=True)
    (corpus / "B.ts").write_text(target_src, encoding="utf-8")
    (corpus / "A.ts").write_text(caller_src, encoding="utf-8")
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    return corpus


_2406_CALLER = (
    'import { shared } from "./B";\n'
    "\n"
    "export function run(): number {\n"
    "  return shared();\n"
    "}\n"
)


def test_incremental_rebuild_drops_call_when_import_is_removed(tmp_path):
    """#2406: no import evidence => the persisted target must not be resolved."""
    from graphify.watch import _rebuild_code

    corpus = _2406_seed(tmp_path, _2406_CALLER)
    caller = corpus / "A.ts"
    # `shared` is now a local no-op call with no import backing it.
    caller.write_text(
        "declare function shared(): number;\n"
        "\n"
        "export function run(): number {\n"
        "  return shared();\n"
        "}\n",
        encoding="utf-8",
    )
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True

    graph = _2406_graph(corpus)
    shared_id = _2406_nid(graph, "shared()", "B.ts")
    run_id = _2406_nid(graph, "run()", "A.ts")
    assert (run_id, shared_id) not in _2406_calls(graph)


def test_incremental_rebuild_uses_fresh_nodes_when_target_also_changed(tmp_path):
    """#2406: a changed target's persisted symbols must never win over fresh ones."""
    from graphify.watch import _rebuild_code

    corpus = _2406_seed(tmp_path, _2406_CALLER)
    caller, target = corpus / "A.ts", corpus / "B.ts"
    target.write_text(
        "export function renamed(): number {\n  return 2;\n}\n", encoding="utf-8"
    )
    caller.write_text(
        'import { renamed } from "./B";\n'
        "\n"
        "export function run(): number {\n"
        "  return renamed();\n"
        "}\n",
        encoding="utf-8",
    )
    assert _rebuild_code(
        corpus,
        changed_paths=[caller, target],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    graph = _2406_graph(corpus)
    calls = _2406_calls(graph)
    run_id = _2406_nid(graph, "run()", "A.ts")
    assert (run_id, _2406_nid(graph, "renamed()", "B.ts")) in calls
    # The stale `shared()` node is gone entirely, so nothing can point at it.
    assert _2406_nid(graph, "shared()", "B.ts") is None


def test_incremental_rebuild_context_excludes_deleted_target(tmp_path):
    """#2406: a deleted file cannot remain a resolver target."""
    from graphify.watch import _rebuild_code

    corpus = _2406_seed(tmp_path, _2406_CALLER)
    caller, target = corpus / "A.ts", corpus / "B.ts"
    target.unlink()
    caller.write_text(
        "export function run(): number {\n  return shared();\n}\n", encoding="utf-8"
    )
    assert _rebuild_code(
        corpus,
        changed_paths=[caller, target],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    graph = _2406_graph(corpus)
    assert _2406_nid(graph, "shared()", "B.ts") is None
    assert _2406_calls(graph) == []


def test_incremental_rebuild_does_not_reparse_unchanged_targets(tmp_path, monkeypatch):
    """#2406 keeps the incremental contract: only changed files are extracted."""
    import graphify.extract as extract_mod
    from graphify.watch import _rebuild_code

    corpus = _2406_seed(tmp_path, _2406_CALLER)
    caller = corpus / "A.ts"

    seen: list[list[str]] = []
    real_extract = extract_mod.extract

    def spy(paths, *args, **kwargs):
        seen.append([Path(p).name for p in paths])
        return real_extract(paths, *args, **kwargs)

    monkeypatch.setattr(extract_mod, "extract", spy)
    caller.write_text(_2406_CALLER.replace("shared();", "shared() + 1;"), encoding="utf-8")
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True

    assert seen == [["A.ts"]]


def test_incremental_rebuild_matches_full_rebuild_and_does_not_duplicate(tmp_path):
    """#2406: full/incremental parity for edges sourced by the changed file."""
    from graphify.watch import _rebuild_code

    corpus = _2406_seed(tmp_path, _2406_CALLER)
    caller = corpus / "A.ts"
    edited = _2406_CALLER.replace("shared();", "shared() + 1;")
    caller.write_text(edited, encoding="utf-8")

    # Two incremental rebuilds in a row must be idempotent (no duplicate edges).
    for _ in range(2):
        assert _rebuild_code(
            corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
        ) is True
    incremental = _2406_calls(_2406_graph(corpus))
    assert len(incremental) == len(set(incremental))

    # Same final corpus, built from scratch.
    fresh = _2406_seed(tmp_path / "fresh", edited)
    assert sorted(_2406_calls(_2406_graph(fresh))) == sorted(incremental)


def test_incremental_rebuild_preserves_python_call_to_unchanged_target(tmp_path):
    """#2406 is language-agnostic: the shared DIRECT cross-file call pass carries it."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir()
    (corpus / "b.py").write_text("def shared():\n    return 1\n", encoding="utf-8")
    caller = corpus / "a.py"
    caller.write_text(
        "from b import shared\n\n\ndef run():\n    return shared()\n", encoding="utf-8"
    )
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    full = _2406_calls(_2406_graph(corpus))
    assert full, "full rebuild must resolve the cross-file python call"

    caller.write_text(
        "from b import shared\n\n\ndef run():\n    return shared() + 1\n",
        encoding="utf-8",
    )
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True
    assert sorted(_2406_calls(_2406_graph(corpus))) == sorted(full)


# --- #2437 / #2438: member + indirect calls into unchanged files -------------
# The #2406 resolution context now also carries the unchanged corpus's
# contains/method edges (member-call resolvers, #2437) and the persisted
# `_callable`/`_callable_class` markers (indirect_call guard, #2438), so both
# edge families survive an incremental rebuild exactly like shared direct calls.

_2437_TARGET = "export class Service {\n  ping(): number { return 1; }\n}\n"
_2437_CALLER = (
    'import { Service } from "./B";\n\n'
    "export function run(): number {\n"
    "  const service = new Service();\n"
    "  return service.ping()%s;\n}\n"
)


def _2437_seed(tmp_path, caller_suffix=""):
    """Build the member-call corpus (TS receiver-typed call) and full-rebuild it."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir(parents=True)
    (corpus / "B.ts").write_text(_2437_TARGET, encoding="utf-8")
    (corpus / "A.ts").write_text(_2437_CALLER % caller_suffix, encoding="utf-8")
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    return corpus


_2438_CALLER = (
    "from b import handler\n\n\ndef run(pool):\n%s    return pool.submit(handler)\n"
)


def _2438_seed(tmp_path, caller_prefix="", target_src="def handler():\n    return 1\n"):
    """Build the indirect-call corpus (py callback into b.py) and full-rebuild it."""
    from graphify.watch import _rebuild_code

    corpus = tmp_path / "corpus"
    corpus.mkdir(parents=True)
    (corpus / "b.py").write_text(target_src, encoding="utf-8")
    (corpus / "a.py").write_text(_2438_CALLER % caller_prefix, encoding="utf-8")
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    return corpus


def _2438_indirects(graph):
    """(source_id, target_id) of every `indirect_call` edge."""
    return [
        (edge.get("source"), edge.get("target"))
        for edge in graph.get("links", graph.get("edges", []))
        if edge.get("relation") == "indirect_call"
    ]


def test_incremental_rebuild_preserves_member_call_to_unchanged_target(tmp_path):
    """#2437: a changed caller keeps its `service.ping()` edge into an unchanged file."""
    from graphify.watch import _rebuild_code

    corpus = _2437_seed(tmp_path)
    full = _2406_calls(_2406_graph(corpus))
    assert full, "full rebuild must resolve the cross-file member call"

    caller = corpus / "A.ts"
    caller.write_text(_2437_CALLER % " + 1", encoding="utf-8")
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True
    assert sorted(_2406_calls(_2406_graph(corpus))) == sorted(full)


def test_incremental_rebuild_preserves_indirect_call_to_unchanged_target(tmp_path):
    """#2438: the persisted `_callable` marker keeps `pool.submit(handler)` resolving."""
    from graphify.watch import _rebuild_code

    corpus = _2438_seed(tmp_path)
    full = _2438_indirects(_2406_graph(corpus))
    assert full, "full rebuild must resolve the cross-file indirect call"

    caller = corpus / "a.py"
    caller.write_text(_2438_CALLER % "    x = 1\n", encoding="utf-8")
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True
    assert sorted(_2438_indirects(_2406_graph(corpus))) == sorted(full)


def test_incremental_rebuild_evicts_removed_member_call(tmp_path):
    """#2437: removing the member call from the caller must remove the edge —
    the fix regenerates edges from fresh raw_calls, it never preserves stale ones."""
    from graphify.watch import _rebuild_code

    corpus = _2437_seed(tmp_path)
    assert _2406_calls(_2406_graph(corpus)), "member-call baseline missing"

    caller = corpus / "A.ts"
    caller.write_text(
        "export function run(): number {\n  return 1;\n}\n", encoding="utf-8"
    )
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True
    assert _2406_calls(_2406_graph(corpus)) == []


def test_incremental_rebuild_evicts_member_call_when_target_deleted(tmp_path):
    """#2437: a deleted callee file must not resurrect through the context edges."""
    from graphify.watch import _rebuild_code

    corpus = _2437_seed(tmp_path)
    caller, target = corpus / "A.ts", corpus / "B.ts"
    target.unlink()
    assert _rebuild_code(
        corpus,
        changed_paths=[caller, target],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    graph = _2406_graph(corpus)
    assert _2406_calls(graph) == []
    assert not any(
        node.get("source_file") == "B.ts" for node in graph.get("nodes", [])
    ), "deleted target's nodes must be evicted, not resurrected as context"


def test_incremental_rebuild_evicts_indirect_call_when_target_deleted(tmp_path):
    """#2438: a deleted callback target must not resurrect through the context nodes."""
    from graphify.watch import _rebuild_code

    corpus = _2438_seed(tmp_path)
    caller, target = corpus / "a.py", corpus / "b.py"
    target.unlink()
    assert _rebuild_code(
        corpus,
        changed_paths=[caller, target],
        no_cluster=True,
        acquire_lock=False,
    ) is True

    graph = _2406_graph(corpus)
    assert _2438_indirects(graph) == []
    assert not any(
        node.get("source_file") == "b.py" for node in graph.get("nodes", [])
    ), "deleted target's nodes must be evicted, not resurrected as context"


def test_incremental_rebuild_callable_guard_excludes_unchanged_data_symbol(tmp_path):
    """#2438 keeps the #1566/#2137 guard: a same-named DATA symbol in an unchanged
    file (`handler = 1`) is not `_callable`, so `pool.submit(handler)` must emit no
    indirect_call on the full build or the incremental one."""
    from graphify.watch import _rebuild_code

    corpus = _2438_seed(tmp_path, target_src="handler = 1\n")
    assert _2438_indirects(_2406_graph(corpus)) == []

    caller = corpus / "a.py"
    caller.write_text(_2438_CALLER % "    x = 1\n", encoding="utf-8")
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True
    assert _2438_indirects(_2406_graph(corpus)) == []


def test_incremental_rebuild_legacy_graph_without_callable_markers(tmp_path):
    """#2438 degradation contract: a graph written before the `_callable` markers
    persisted must not crash the incremental rebuild — the guard fails closed (no
    indirect_call, pre-fix behavior) and the next full rebuild self-heals."""
    import json

    from graphify.watch import _rebuild_code

    corpus = _2438_seed(tmp_path)
    graph_path = corpus / "graphify-out" / "graph.json"
    assert _2438_indirects(_2406_graph(corpus)), "indirect-call baseline missing"

    # Simulate a pre-#2438 graph: strip the persisted callability markers.
    legacy = json.loads(graph_path.read_text(encoding="utf-8"))
    for node in legacy.get("nodes", []):
        node.pop("_callable", None)
        node.pop("_callable_class", None)
    graph_path.write_text(json.dumps(legacy), encoding="utf-8")

    caller = corpus / "a.py"
    caller.write_text(_2438_CALLER % "    x = 1\n", encoding="utf-8")
    assert _rebuild_code(
        corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
    ) is True
    assert _2438_indirects(_2406_graph(corpus)) == []

    # A full rebuild re-extracts the target, restores the markers, and the edge.
    assert _rebuild_code(corpus, no_cluster=True, acquire_lock=False) is True
    assert _2438_indirects(_2406_graph(corpus)), "full rebuild must self-heal"


def test_incremental_member_call_parity_and_idempotency(tmp_path):
    """#2437: repeated incremental rebuilds neither duplicate the member-call edge
    nor diverge from a from-scratch build of the same corpus."""
    from graphify.watch import _rebuild_code

    corpus = _2437_seed(tmp_path)
    caller = corpus / "A.ts"
    caller.write_text(_2437_CALLER % " + 1", encoding="utf-8")
    for _ in range(2):
        assert _rebuild_code(
            corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
        ) is True
    incremental = _2406_calls(_2406_graph(corpus))
    assert incremental, "member call lost across repeated incremental rebuilds"
    assert len(incremental) == len(set(incremental))

    fresh = _2437_seed(tmp_path / "fresh", caller_suffix=" + 1")
    assert sorted(_2406_calls(_2406_graph(fresh))) == sorted(incremental)


def test_incremental_indirect_call_parity_and_idempotency(tmp_path):
    """#2438: repeated incremental rebuilds neither duplicate the indirect_call edge
    nor diverge from a from-scratch build of the same corpus."""
    from graphify.watch import _rebuild_code

    corpus = _2438_seed(tmp_path)
    caller = corpus / "a.py"
    caller.write_text(_2438_CALLER % "    x = 1\n", encoding="utf-8")
    for _ in range(2):
        assert _rebuild_code(
            corpus, changed_paths=[caller], no_cluster=True, acquire_lock=False
        ) is True
    incremental = _2438_indirects(_2406_graph(corpus))
    assert incremental, "indirect call lost across repeated incremental rebuilds"
    assert len(incremental) == len(set(incremental))

    fresh = _2438_seed(tmp_path / "fresh", caller_prefix="    x = 1\n")
    assert sorted(_2438_indirects(_2406_graph(fresh))) == sorted(incremental)