tests/test_cache.py
Rama v8 · réplica de lectura del repositorio original.
"""Tests for graphify/cache.py."""
import json
import os
import pytest
from pathlib import Path
from graphify.cache import file_hash, cache_dir, load_cached, save_cached, cached_files, clear_cache, _body_content
@pytest.fixture
def tmp_file(tmp_path):
f = tmp_path / "sample.txt"
f.write_text("hello world")
return f
@pytest.fixture
def cache_root(tmp_path):
return tmp_path
def test_file_hash_consistent(tmp_file):
"""Same file gives same hash on repeated calls."""
h1 = file_hash(tmp_file)
h2 = file_hash(tmp_file)
assert h1 == h2
assert isinstance(h1, str)
assert len(h1) == 64 # SHA256 hex digest length
def test_file_hash_changes(tmp_path):
"""Different file contents give different hashes."""
f1 = tmp_path / "a.txt"
f2 = tmp_path / "b.txt"
f1.write_text("content one")
f2.write_text("content two")
assert file_hash(f1) != file_hash(f2)
def test_cache_roundtrip(tmp_file, cache_root):
"""Save then load returns the same result dict."""
result = {"nodes": [{"id": "n1", "label": "Node1"}], "edges": []}
save_cached(tmp_file, result, root=cache_root)
loaded = load_cached(tmp_file, root=cache_root)
assert loaded == result
def test_cache_miss_on_change(tmp_file, cache_root):
"""After file content changes, load_cached returns None."""
result = {"nodes": [], "edges": [{"source": "a", "target": "b"}]}
save_cached(tmp_file, result, root=cache_root)
# Modify the file
tmp_file.write_text("completely different content")
assert load_cached(tmp_file, root=cache_root) is None
def test_cached_files(tmp_path, cache_root):
"""cached_files returns the set of cached hashes."""
f1 = tmp_path / "file1.py"
f2 = tmp_path / "file2.py"
f1.write_text("alpha")
f2.write_text("beta")
save_cached(f1, {"nodes": [], "edges": []}, root=cache_root)
save_cached(f2, {"nodes": [], "edges": []}, root=cache_root)
hashes = cached_files(cache_root)
assert file_hash(f1, cache_root) in hashes
assert file_hash(f2, cache_root) in hashes
def test_clear_cache(tmp_file, cache_root):
"""clear_cache removes all .json files from graphify-out/cache/ (all subdirs)."""
save_cached(tmp_file, {"nodes": [], "edges": []}, root=cache_root)
# Since v0.5.3 entries go into cache/ast/, not the flat cache/ dir
cache_base = cache_root / "graphify-out" / "cache"
assert len(list(cache_base.rglob("*.json"))) > 0
clear_cache(cache_root)
assert len(list(cache_base.rglob("*.json"))) == 0
def test_md_frontmatter_only_change_same_hash(tmp_path):
"""Changing only frontmatter fields in a .md file does not change the hash."""
f = tmp_path / "doc.md"
f.write_text("---\nreviewed: 2026-01-01\n---\n\n# Title\n\nBody text.")
h1 = file_hash(f)
f.write_text("---\nreviewed: 2026-04-09\n---\n\n# Title\n\nBody text.")
h2 = file_hash(f)
assert h1 == h2
def test_md_body_change_different_hash(tmp_path):
"""Changing the body of a .md file produces a different hash."""
f = tmp_path / "doc.md"
f.write_text("---\nreviewed: 2026-01-01\n---\n\n# Title\n\nOriginal body.")
h1 = file_hash(f)
f.write_text("---\nreviewed: 2026-01-01\n---\n\n# Title\n\nChanged body.")
h2 = file_hash(f)
assert h1 != h2
def test_md_no_frontmatter_hashed_normally(tmp_path):
"""A .md file with no frontmatter is hashed by its full content."""
f = tmp_path / "doc.md"
f.write_text("# Just a heading\n\nNo frontmatter here.")
h1 = file_hash(f)
f.write_text("# Just a heading\n\nDifferent content.")
h2 = file_hash(f)
assert h1 != h2
def test_non_md_file_hashed_fully(tmp_path):
"""Non-.md files are still hashed by their full content."""
f = tmp_path / "script.py"
f.write_text("# comment\nx = 1")
h1 = file_hash(f)
f.write_text("# changed comment\nx = 1")
h2 = file_hash(f)
assert h1 != h2
def test_body_content_strips_frontmatter():
"""_body_content correctly strips YAML frontmatter."""
content = b"---\ntitle: Test\n---\n\nActual body."
assert _body_content(content) == b"\n\nActual body."
def test_body_content_no_frontmatter():
"""_body_content returns content unchanged when no frontmatter present."""
content = b"No frontmatter here."
assert _body_content(content) == content
# --- #1259: frontmatter delimiters must be whole `---` lines -----------------
def test_body_content_hr_start_is_not_frontmatter():
"""A document opening with a ``----`` thematic break has no frontmatter;
a later ``---`` hr must not be mistaken for a close delimiter."""
content = b"----\nIntro paragraph that must be hashed.\n\n---\nbody"
assert _body_content(content) == content
def test_body_content_dash_title_start_is_not_frontmatter():
"""``--- title`` on the first line is prose, not an open delimiter."""
content = b"--- title\nIntro that must be hashed.\n\n---\nbody"
assert _body_content(content) == content
def test_body_content_dash_text_line_is_not_close_delimiter():
"""``--- text`` and ``----`` lines inside opened frontmatter are not the
close; without a proper close the content passes through unchanged."""
content = b"---\ntitle: Test\nbody starts here\n--- not a delimiter\n----\nreal content"
assert _body_content(content) == content
def test_body_content_later_proper_close_skips_dash_text_lines():
"""A ``--- text`` line is skipped; the next whole ``---`` line closes."""
content = b"---\ntitle: Test\nnote: --- inline\n---\nreal body"
assert _body_content(content) == b"\nreal body"
def test_body_content_well_formed_output_byte_identical():
"""For well-formed frontmatter the stripped body must stay byte-identical
to the historical substring implementation, so existing semantic-cache
hashes do not churn (re-extraction is billed LLM work)."""
cases = [
# (input, output of the historical text.find("\n---")+4 algorithm)
(b"---\ntitle: Test\n---\n\nActual body.", b"\n\nActual body."),
(b"---\nreviewed: 2026-01-01\n---\n\n# Title\n\nBody text.", b"\n\n# Title\n\nBody text."),
# close delimiter with trailing whitespace keeps it in the body
(b"---\ntitle: Test\n--- \nbody", b" \nbody"),
# CRLF line endings
(b"---\r\ntitle: Test\r\n---\r\nbody", b"\r\nbody"),
# empty frontmatter block
(b"---\n---\nbody", b"\nbody"),
# close as the very last line, no trailing newline
(b"---\ntitle: Test\n---", b""),
]
for content, expected in cases:
assert _body_content(content) == expected, content
def test_md_edit_above_hr_changes_hash(tmp_path):
"""Editing content above a mid-document ``----`` break must change the
hash -- previously that region was silently excluded from hashing."""
f = tmp_path / "doc.md"
f.write_text("----\nIntro paragraph.\n\n---\nbody")
h1 = file_hash(f)
f.write_text("----\nEdited intro paragraph.\n\n---\nbody")
h2 = file_hash(f)
assert h1 != h2
# --- #777: portable cache source_file fields --------------------------------
# ``save_cached`` relativizes ``source_file`` entries inside the cache file
# so a committed ``graphify-out/cache/`` is portable across machines and
# CI runners. ``load_cached`` re-absolutizes them so consumers (extract,
# merge into graph.json) see the same shape that fresh extraction emits.
def test_save_cached_relativizes_source_file(tmp_path):
"""The on-disk cache JSON contains forward-slash relative source_file
entries — no absolute prefix from the saving machine leaks in."""
import json
from graphify.cache import save_cached, file_hash, cache_dir
(tmp_path / "src").mkdir()
src = tmp_path / "src" / "foo.py"
src.write_text("def x(): pass\n")
abs_src = str(src.resolve())
result = {
"nodes": [{"id": "n1", "label": "foo", "source_file": abs_src}],
"edges": [{"source": "n1", "target": "n1", "source_file": abs_src}],
}
save_cached(src, result, root=tmp_path, kind="ast")
h = file_hash(src, tmp_path)
entry = cache_dir(tmp_path, "ast") / f"{h}.json"
on_disk = json.loads(entry.read_text(encoding="utf-8"))
node_sources = {n["source_file"] for n in on_disk["nodes"]}
edge_sources = {e["source_file"] for e in on_disk["edges"]}
assert node_sources == {"src/foo.py"}, (
f"cache nodes must store relative source_file; got {node_sources}"
)
assert edge_sources == {"src/foo.py"}
def test_save_cached_survives_a_winerror_17_replace(tmp_path, monkeypatch):
"""#3508: on some Windows/filesystem combinations, `os.replace` raises
WinError 17 ("cannot move to a different disk drive") even for a temp file
and destination in the same directory on the same drive -- reported for
exactly this AST cache write path. WinError 17 is a plain OSError, not a
PermissionError, so it must still trigger the copy-then-delete fallback."""
import os
from graphify.cache import save_cached, file_hash, cache_dir
src = tmp_path / "foo.py"
src.write_text("def x(): pass\n")
real_replace = os.replace
def flaky_replace(a, b):
exc = OSError("cannot move to a different disk drive")
exc.winerror = 17
raise exc
monkeypatch.setattr(os, "replace", flaky_replace)
try:
save_cached(src, {"nodes": [], "edges": []}, root=tmp_path, kind="ast")
finally:
monkeypatch.setattr(os, "replace", real_replace)
h = file_hash(src, tmp_path)
entry = cache_dir(tmp_path, "ast") / f"{h}.json"
assert entry.exists()
assert not any(p.name.endswith(".tmp") for p in entry.parent.iterdir())
def test_load_cached_absolutizes_source_file(tmp_path):
"""``load_cached`` returns the same absolute-path shape that a fresh
extraction produces, so consumers don't need to special-case cache
hits vs. fresh extraction."""
from graphify.cache import save_cached, load_cached
(tmp_path / "src").mkdir()
src = tmp_path / "src" / "foo.py"
src.write_text("def x(): pass\n")
abs_src = str(src.resolve())
save_cached(src, {
"nodes": [{"id": "n1", "source_file": abs_src}],
"edges": [{"source": "n1", "target": "n1", "source_file": abs_src}],
}, root=tmp_path, kind="ast")
loaded = load_cached(src, root=tmp_path, kind="ast")
assert loaded is not None
assert loaded["nodes"][0]["source_file"] == abs_src
assert loaded["edges"][0]["source_file"] == abs_src
def test_load_cached_passes_through_legacy_absolute_source_file(tmp_path):
"""Cache entries written by an older graphify (with absolute source_file
inside) must still load correctly: the absolutize step is a no-op for
already-absolute values."""
import json
from graphify.cache import load_cached, file_hash, cache_dir
(tmp_path / "src").mkdir()
src = tmp_path / "src" / "foo.py"
src.write_text("pass\n")
abs_src = str(src.resolve())
# Hand-write a legacy-format cache entry (absolute source_file).
h = file_hash(src, tmp_path)
entry = cache_dir(tmp_path, "ast") / f"{h}.json"
entry.write_text(json.dumps({
"nodes": [{"id": "n1", "source_file": abs_src}],
"edges": [],
}))
loaded = load_cached(src, root=tmp_path, kind="ast")
assert loaded is not None
assert loaded["nodes"][0]["source_file"] == abs_src
def test_cache_portable_across_roots(tmp_path):
"""End-to-end portability: a cache entry written at one root can be
consumed at a different absolute root because the file is content-hashed
AND its embedded source_file is stored relative."""
import json
import shutil
from graphify.cache import save_cached, load_cached, file_hash, cache_dir
repo_a = tmp_path / "repo_a"
repo_a.mkdir()
(repo_a / "src").mkdir()
src_a = repo_a / "src" / "foo.py"
src_a.write_text("def x(): pass\n")
save_cached(src_a, {
"nodes": [{"id": "n1", "source_file": str(src_a.resolve())}],
"edges": [],
}, root=repo_a, kind="ast")
# Copy corpus + cache to a second location with a different absolute prefix.
repo_b = tmp_path / "repo_b"
shutil.copytree(repo_a, repo_b)
src_b = repo_b / "src" / "foo.py"
loaded = load_cached(src_b, root=repo_b, kind="ast")
assert loaded is not None, (
"cache must port across absolute prefixes (content hash + relative source_file)"
)
# Source path re-anchored to the new root, not the old one.
assert loaded["nodes"][0]["source_file"] == str(src_b.resolve())
assert not str(repo_a) in loaded["nodes"][0]["source_file"]
# --- AST cache id portability (#2257) ---------------------------------------
# Sibling of the source_file portability above. Extractors mint node ids from
# the path STRING they were handed, so a cached entry embeds the absolute scan
# root in every id (`<root-slug>_pkg_mod_base`). extract()'s id-remap keys those
# rewrites off the CURRENT path, so on a warm hit under a different root the
# stored ids match no key and the original root's slug survives into graph.json.
def _reset_stat_index():
"""The stat-index location/anchor are chosen once per process via module
globals (#1747/#2199). Reset them so each test sees a fresh-process
decision — same pattern as tests/test_stat_index_portability.py."""
from graphify import cache as _cache
_cache._stat_index_root = None
_cache._stat_index_anchor = None
_cache._stat_index = {}
_cache._stat_index_dirty = False
def _portability_corpus(base: Path) -> Path:
"""A corpus covering every id/path carrier a cache entry can hold.
Deliberately NOT JavaScript/TypeScript: those suffixes are in
``_JS_CACHE_BYPASS_SUFFIXES`` and are never cached, so a JS fixture would
make the warm-hit assertions below pass vacuously.
- python -> cross-file `imports_from` with an already-canonical target
- C -> `edges[].target_file` (absolute) on the `#include` edge
- bash -> `bash_sources[].source_file` plus the `__entry` id suffix
- markdown-> `references` edge with a `target_file` stamp
"""
c = base / "corpus"
(c / "pkg").mkdir(parents=True)
(c / "lib").mkdir(parents=True)
(c / "pkg" / "mod.py").write_text("class Base:\n def hello(self):\n return 1\n")
(c / "app.py").write_text(
"from pkg.mod import Base\n\n\ndef run():\n return Base().hello()\n"
)
(c / "lib" / "util.h").write_text("int util_add(int a, int b);\n")
(c / "main.c").write_text('#include "lib/util.h"\nint main(void) { return util_add(1,2); }\n')
(c / "lib" / "common.sh").write_text("greet() { echo hi; }\n")
(c / "run.sh").write_text("#!/bin/bash\nsource ./lib/common.sh\ngreet\n")
(c / "doc.md").write_text("# Doc\n\nSee [util](lib/util.h).\n")
return c
def _graph_ids(result: dict) -> tuple[list[str], list[tuple]]:
"""Node ids + edge endpoint pairs — the granularity #2257 is about.
Deliberately not whole-dict equality: transient hints such as
``target_file`` are minted from the resolved path while other fields keep
the given spelling, so they can differ harmlessly between a warm and a cold
run under a symlinked root (consumers ``.resolve()`` them anyway).
"""
return (
sorted(str(n.get("id")) for n in result["nodes"]),
sorted((str(e.get("source")), str(e.get("target"))) for e in result["edges"]),
)
def test_warm_cache_from_another_root_does_not_leak_that_root(tmp_path, monkeypatch):
"""#2257: extract corpus under root A (populating the cache), copy the tree
AND graphify-out to root B, extract under B on the warm cache.
No node id or edge endpoint may carry root A's slug, and the ids must match
a cold B extraction exactly.
"""
import shutil
import graphify.extract as ex
a_slug = "aaa_root_marker"
b_slug = "bbb_root_marker"
corpus_a = _portability_corpus(tmp_path / a_slug)
paths_a = sorted(p for p in corpus_a.rglob("*") if p.is_file())
_reset_stat_index()
result_a = ex.extract(paths_a, cache_root=corpus_a, root=corpus_a, parallel=False)
from graphify import cache as _cache
_cache._flush_stat_index()
assert result_a["nodes"], "run A should have extracted something"
# The entries on disk must be portable BY CONSTRUCTION: neither the scan
# root's slug (ids are casefolded, paths are not — compare case-insensitively)
# nor any absolute path from root A may be embedded in them.
entries = sorted((corpus_a / "graphify-out" / "cache" / "ast").rglob("*.json"))
assert entries, "run A should have written AST cache entries"
for entry in entries:
blob = entry.read_text(encoding="utf-8")
assert a_slug not in blob.lower(), (
f"{entry.name} embeds the scan root's slug, so replaying it under a "
f"different root replays root A's ids (#2257)"
)
assert str(corpus_a) not in blob, f"{entry.name} embeds an absolute scan path"
# Move the corpus; graphify-out/ (cache + stat index) rides along. copy2
# preserves mtime_ns so the stat-index fastpath stays warm.
corpus_b = tmp_path / b_slug / "corpus"
corpus_b.parent.mkdir()
shutil.copytree(corpus_a, corpus_b, copy_function=shutil.copy2)
paths_b = sorted(p for p in corpus_b.rglob("*") if p.is_file()
and "graphify-out" not in p.parts)
# Warmth probe: _safe_extract_with_xaml_root runs only on a cache MISS. If
# run B silently re-extracts, cold ids come out clean and every assertion
# below passes while proving nothing. The probe requires parallel=False —
# the process pool extracts in a subprocess where this patch is invisible,
# so switching this call to parallel=True would make `misses` vacuously [].
misses = []
real_extract = ex._safe_extract_with_xaml_root
def _counting(extractor, path, root):
misses.append(str(path))
return real_extract(extractor, path, root)
monkeypatch.setattr(ex, "_safe_extract_with_xaml_root", _counting)
_reset_stat_index()
warm_b = ex.extract(paths_b, cache_root=corpus_b, root=corpus_b, parallel=False)
assert misses == [], f"run B must be served entirely from the cache, re-extracted: {misses}"
warm_ids, warm_edges = _graph_ids(warm_b)
leaked = [i for i in warm_ids if a_slug in i] + [
p for p in warm_edges if any(a_slug in x for x in p)
]
assert not leaked, f"root A's slug survived a warm cache hit into run B (#2257): {leaked}"
assert not [i for i in warm_ids if "$" in i], "the storage placeholder escaped into the graph"
# ...and the replay is not merely clean but IDENTICAL to a cold B run.
monkeypatch.undo()
shutil.rmtree(corpus_b / "graphify-out")
_reset_stat_index()
cold_b = ex.extract(paths_b, cache_root=corpus_b, root=corpus_b, parallel=False)
cold_ids, cold_edges = _graph_ids(cold_b)
# Guards the save-side transform against mutating the caller's dict: a cold
# run's ids must still be the canonical root-relative spec form, since
# extract()'s id-remap is keyed on the ABSOLUTE form the extractor minted.
assert {"app", "app_run", "pkg_mod", "pkg_mod_base", "pkg_mod_base_hello"} <= set(cold_ids), (
f"cold run no longer produces canonical ids: {cold_ids}"
)
assert (warm_ids, warm_edges) == (cold_ids, cold_edges), (
"a warm cross-root cache hit must reproduce the cold extraction exactly"
)
def test_cached_ids_round_trip_under_the_same_root(tmp_path):
"""The stored placeholder form must restore to the exact absolute-derived id
the extractor minted, or a same-root warm hit would break extract()'s
id-remap (which is keyed on that absolute form)."""
root = tmp_path / "repo"
(root / "src").mkdir(parents=True)
f = root / "src" / "foo.py"
f.write_text("def x(): pass\n")
from graphify.extract import _make_id
minted = _make_id(str(f))
result = {
"nodes": [{"id": minted, "source_file": str(f)}],
"edges": [{"source": minted, "target": minted + "_x", "source_file": str(f)}],
"raw_calls": [{"caller_nid": minted + "_x", "source_file": str(f)}],
}
save_cached(f, result, root=root, kind="ast")
assert result["nodes"][0]["id"] == minted, "the caller's dict must not be mutated"
loaded = load_cached(f, root=root, kind="ast")
assert loaded["nodes"][0]["id"] == minted
assert loaded["edges"][0]["source"] == minted
assert loaded["edges"][0]["target"] == minted + "_x"
assert loaded["raw_calls"][0]["caller_nid"] == minted + "_x"
def test_relative_root_does_not_reanchor_an_already_canonical_id(tmp_path, monkeypatch):
"""A relative ``root`` (what save_semantic_cache forwards) must not be used
as an id anchor: with an absolute path the restore form is the RESOLVED
slug, so stripping the relative spelling would rewrite an already-canonical
id into an absolute-derived one — the very leak this guards against."""
monkeypatch.chdir(tmp_path)
(tmp_path / "src" / "utils").mkdir(parents=True)
f = (tmp_path / "src" / "utils" / "foo.py").resolve()
f.write_text("x = 1\n")
canonical = {"nodes": [{"id": "src_utils_foo", "source_file": str(f)}], "edges": []}
save_cached(f, canonical, root=Path("src"), kind="semantic")
loaded = load_cached(f, root=Path("src"), kind="semantic")
assert loaded["nodes"][0]["id"] == "src_utils_foo"
def test_warm_hit_with_relative_inputs_from_above_the_root(tmp_path, monkeypatch):
"""#2630: relative inputs handed to extract() from a CWD above the root.
Extractors stamp ``source_file`` with the path STRING they were handed, so
a relative input yields a CWD-relative stamp — but the stored format is
root-relative and ``load_cached`` re-anchors it as such. When CWD is not
the inferred root the two disagree and a warm hit resurrects a path naming
no file (``<root>/src/pages/index.astro``). Every source_file-GATED remap
in extract() then misses, so the warm hit keeps the raw-path symbol ids a
cold run canonicalizes: the astro frontmatter variable came back as
``src_pages_index_posts``, no longer under its ``pages_index`` file node's
stem — the prefix ``build.py`` reconciles symbols to files by.
Astro is the fixture because ``.astro`` is cached while every other
JS-family suffix is in ``_JS_CACHE_BYPASS_SUFFIXES``; the defect itself is
language-agnostic (the gate is shared).
"""
import graphify.extract as ex
project = tmp_path / "project"
(project / "src" / "pages").mkdir(parents=True)
(project / "src" / "lib").mkdir(parents=True)
(project / "src" / "lib" / "content.ts").write_text(
"export function getPosts() { return []; }\n"
)
(project / "src" / "pages" / "index.astro").write_text(
"---\n"
"import { getPosts } from '../lib/content';\n"
"const posts = getPosts();\n"
"---\n"
"<h1>{posts.length}</h1>\n"
)
# CWD is the project; the root extract() infers is the common parent `src/`.
monkeypatch.chdir(project)
rel_paths = [Path("src/lib/content.ts"), Path("src/pages/index.astro")]
_reset_stat_index()
cold = ex.extract(rel_paths, parallel=False)
# Warmth probe (see test_warm_cache_from_another_root...): a silent
# re-extraction would make the assertions below pass vacuously. Only the
# .astro file is cached, so it is the one that must not be re-extracted.
misses: list[str] = []
real_extract = ex._safe_extract_with_xaml_root
def _counting(extractor, path, root):
misses.append(str(path))
return real_extract(extractor, path, root)
monkeypatch.setattr(ex, "_safe_extract_with_xaml_root", _counting)
_reset_stat_index()
warm = ex.extract(rel_paths, parallel=False)
assert not [m for m in misses if m.endswith(".astro")], (
f"the .astro file must be served from the cache, re-extracted: {misses}"
)
assert _graph_ids(cold) == _graph_ids(warm), (
"a warm cache hit must reproduce the cold extraction's ids exactly"
)
for label, graph in (("cold", cold), ("warm", warm)):
ids = {str(n["id"]) for n in graph["nodes"]}
assert {"pages_index", "pages_index_posts"} <= ids, (label, sorted(ids))
# The stem the frontmatter variable must NOT keep: `src_`-prefixed is
# the pre-remap form derived from the raw input path.
assert not [i for i in ids if i.startswith("src_pages_index")], (
label, sorted(ids)
)
# --- AST cache versioning ----------------------------------------------------
# AST cache entries are the output of graphify's own extractor code, so they
# are only valid for the graphify version that wrote them. Keying purely on
# file content meant extractor fixes shipped in a new release kept serving
# stale pre-fix results. The AST cache is therefore namespaced by package
# version; the semantic cache is NOT (invalidating it would re-bill LLM
# extraction for unchanged files).
def test_ast_cache_invalidated_on_version_bump(tmp_path, monkeypatch):
"""An AST entry written by version X must not be served after upgrading
to version Y — the file is unchanged but the extractor is not."""
import graphify.cache as cache_mod
f = tmp_path / "mod.py"
f.write_text("def f(): pass\n")
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "0.8.0", raising=False)
save_cached(f, {"nodes": [{"id": "n1"}], "edges": []}, root=tmp_path, kind="ast")
assert load_cached(f, root=tmp_path, kind="ast") is not None
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "0.8.1", raising=False)
assert load_cached(f, root=tmp_path, kind="ast") is None, (
"AST cache entry from a previous graphify version must not be served"
)
def test_ast_cache_schema_rejects_same_version_legacy_collision(
tmp_path, monkeypatch
):
"""A key-schema change must not replay a poisoned same-version AST entry."""
import json
import graphify.cache as cache_mod
target = tmp_path / "real.py"
target.write_text("value = 1\n")
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "0.9.46", raising=False)
old_dir = tmp_path / cache_mod._GRAPHIFY_OUT / "cache" / "ast" / "v0.9.46"
old_dir.mkdir(parents=True)
old_hash = file_hash(target, tmp_path)
(old_dir / f"{old_hash}.json").write_text(json.dumps({
"nodes": [{"id": "alias", "source_file": "alias.py"}],
"edges": [],
}))
monkeypatch.setattr(cache_mod, "_cleaned_ast_dirs", set(), raising=False)
assert load_cached(target, root=tmp_path, kind="ast") is None
assert not old_dir.exists()
def test_python_receiver_shadow_schema_retires_same_version_raw_calls(
tmp_path, monkeypatch,
):
"""Schema-4 raw calls lack the lexical fact required by external resolution."""
import graphify.cache as cache_mod
from graphify.extract import extract
target = tmp_path / "caller.py"
target.write_text("import requests\n\ndef fetch():\n return requests.get('/data')\n")
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "same-version")
monkeypatch.setattr(cache_mod, "_AST_CACHE_SCHEMA", 4)
monkeypatch.setattr(cache_mod, "_cleaned_ast_dirs", set())
save_cached(target, {"nodes": [], "edges": [], "raw_calls": [
{"caller_nid": "old", "receiver": "requests", "callee": "get"}
]}, root=tmp_path, kind="ast")
old_dir = cache_dir(tmp_path, "ast")
monkeypatch.setattr(cache_mod, "_AST_CACHE_SCHEMA", 5)
assert load_cached(target, root=tmp_path, kind="ast") is None
assert not old_dir.exists()
cold = extract([target], cache_root=tmp_path, root=tmp_path)
fresh = load_cached(target, root=tmp_path, kind="ast")
assert fresh is not None
assert any(rc.get("_python_receiver_shadowed") is False
for rc in fresh.get("raw_calls", []))
warm = extract([target], cache_root=tmp_path, root=tmp_path)
assert len([e for e in cold["edges"] if e["relation"] == "calls"]) == 1
assert [e for e in cold["edges"] if e["relation"] == "calls"] == [
e for e in warm["edges"] if e["relation"] == "calls"
]
def test_ast_cache_version_bump_cleans_stale_entries(tmp_path, monkeypatch):
"""Upgrading removes AST entries left behind by previous versions so the
cache directory does not grow one full copy per release."""
import graphify.cache as cache_mod
f = tmp_path / "mod.py"
f.write_text("def f(): pass\n")
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "0.8.0", raising=False)
save_cached(f, {"nodes": [{"id": "n1"}], "edges": []}, root=tmp_path, kind="ast")
old_dir = cache_dir(tmp_path, "ast")
assert any(old_dir.glob("*.json"))
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "0.8.1", raising=False)
monkeypatch.setattr(cache_mod, "_cleaned_ast_dirs", set(), raising=False)
cache_dir(tmp_path, "ast")
assert not old_dir.exists(), (
"stale AST version directory must be removed on upgrade"
)
def test_legacy_unversioned_ast_entries_not_served(tmp_path):
"""Entries written by pre-versioning graphify (flat cache/ or unversioned
cache/ast/) are by definition from an older extractor and must not be
served — that staleness is exactly what version namespacing fixes."""
import json
from graphify.cache import file_hash, _GRAPHIFY_OUT
f = tmp_path / "mod.py"
f.write_text("def f(): pass\n")
h = file_hash(f, tmp_path)
payload = json.dumps({"nodes": [{"id": "stale"}], "edges": []})
# Unversioned cache/ast/{hash}.json (pre-versioning layout)
unversioned = tmp_path / _GRAPHIFY_OUT / "cache" / "ast"
unversioned.mkdir(parents=True)
(unversioned / f"{h}.json").write_text(payload)
# Legacy flat cache/{hash}.json (pre-0.5.3 layout)
(unversioned.parent / f"{h}.json").write_text(payload)
assert load_cached(f, root=tmp_path, kind="ast") is None
def test_semantic_cache_survives_version_bump(tmp_path, monkeypatch):
"""The semantic cache is deliberately not versioned: entries are produced
by the LLM from file contents, and re-extraction costs real money."""
import graphify.cache as cache_mod
f = tmp_path / "doc.md"
f.write_text("# Title\n\nBody.\n")
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "0.8.0", raising=False)
save_cached(f, {"nodes": [{"id": "n1"}], "edges": []}, root=tmp_path, kind="semantic")
semantic_dir = cache_dir(tmp_path, "semantic")
monkeypatch.setattr(cache_mod, "_EXTRACTOR_VERSION", "0.8.1", raising=False)
monkeypatch.setattr(cache_mod, "_cleaned_ast_dirs", set(), raising=False)
cache_dir(tmp_path, "ast") # triggers stale-AST cleanup
assert load_cached(f, root=tmp_path, kind="semantic") is not None
assert any(semantic_dir.glob("*.json")), (
"semantic entries must survive both the version bump and AST cleanup"
)
def test_save_cached_in_root_symlink_keeps_symlink_name(tmp_path):
"""``source_file`` for an in-root symlink must be stored under the
symlink's own name, not the resolved target. Lower-impact than the
manifest case (cache lookup is content-hashed, not key-matched), but
keeps the on-disk shape consistent with what callers passed in."""
import json
from graphify.cache import save_cached, file_hash, cache_dir
(tmp_path / "sub").mkdir()
target = tmp_path / "sub" / "target.py"
target.write_text("pass\n")
alias = tmp_path / "alias.py"
try:
alias.symlink_to(target)
except (OSError, NotImplementedError):
import pytest
pytest.skip("filesystem does not support symlinks")
abs_alias = str(alias) # caller's view — the symlink path, unresolved
save_cached(alias, {
"nodes": [{"id": "n1", "source_file": abs_alias}],
"edges": [],
}, root=tmp_path, kind="ast")
h = file_hash(alias, tmp_path)
entry = cache_dir(tmp_path, "ast") / f"{h}.json"
on_disk = json.loads(entry.read_text(encoding="utf-8"))
assert on_disk["nodes"][0]["source_file"] == "alias.py", (
f"cache must store symlink name, not resolved target; got "
f"{on_disk['nodes'][0]['source_file']!r}"
)
def test_file_hash_distinguishes_walked_symlink_paths_portably(
requires_symlinks, tmp_path
):
"""Aliases of one target need separate portable extraction-cache keys."""
from graphify import cache as cache_mod
_reset_stat_index()
hashes_by_root = []
for dirname in ("repo_a", "repo_b"):
root = tmp_path / dirname
(root / "sub").mkdir(parents=True)
target = root / "real.py"
target.write_text("def value():\n return 1\n")
aliases = (root / "alias.py", root / "sub" / "link.py")
aliases[0].symlink_to(target)
aliases[1].symlink_to(target)
hashes = tuple(file_hash(path, root) for path in (target, *aliases))
assert len(set(hashes)) == 3
assert len(cache_mod._stat_index[str(target.resolve())]["hashes"]) == 3
hashes_by_root.append(hashes)
assert hashes_by_root[0] == hashes_by_root[1]
def test_file_hash_keeps_resolved_fallback_for_external_symlink(
requires_symlinks, tmp_path
):
"""An out-of-root target retains the existing resolved-path identity."""
_reset_stat_index()
root = tmp_path / "repo"
root.mkdir()
target = tmp_path / "external.py"
target.write_text("external = True\n")
alias = root / "external.py"
alias.symlink_to(target)
assert file_hash(alias, root) == file_hash(target, root)
def test_warm_cache_keeps_target_and_symlink_sources_distinct(
requires_symlinks, tmp_path, monkeypatch
):
"""#2832: a warm cache must not move target nodes onto its symlink."""
from collections import Counter
import graphify.extract as extract_mod
_reset_stat_index()
physical_root = tmp_path / "repo"
(physical_root / "sub").mkdir(parents=True)
target = physical_root / "real.py"
target.write_text("def value():\n return 1\n")
alias = physical_root / "sub" / "link.py"
alias.symlink_to(target)
root = tmp_path / "scan"
root.symlink_to(physical_root, target_is_directory=True)
paths = extract_mod.collect_files(root)
assert [path.relative_to(root).as_posix() for path in paths] == [
"real.py",
"sub/link.py",
]
cold = extract_mod.extract(paths, cache_root=root, root=root, parallel=False)
misses = []
real_extract = extract_mod._safe_extract_with_xaml_root
def counting_extract(extractor, path, extract_root):
misses.append(path)
return real_extract(extractor, path, extract_root)
monkeypatch.setattr(extract_mod, "_safe_extract_with_xaml_root", counting_extract)
warm = extract_mod.extract(paths, cache_root=root, root=root, parallel=False)
assert misses == []
cold_counts = Counter(n.get("source_file") for n in cold["nodes"])
warm_counts = Counter(n.get("source_file") for n in warm["nodes"])
assert warm_counts == cold_counts
assert len(cold_counts) == 2
assert {Path(source).name for source in cold_counts} == {"real.py", "link.py"}
@pytest.mark.skipif(os.name != "nt", reason="directory junctions are Windows-only")
def test_file_hash_distinguishes_walked_junction_paths(tmp_path):
r"""#2832 with junctions, which need no symlink privilege. Under a junctioned
scan root the walked-parent match compared a plain spelling with the
``\\?\`` form of the root (#4925), never matched, and the in-root alias fell
back to its target's salt and shared its digest."""
import subprocess
_reset_stat_index()
root = tmp_path / "repo"
(root / "sub").mkdir(parents=True)
target = root / "sub" / "g.py"
target.write_text("def g():\n return 1\n")
for link, dest in ((root / "alias", root / "sub"), (tmp_path / "scan", root)):
subprocess.run(["cmd", "/c", "mklink", "/J", str(link), str(dest)],
check=True, capture_output=True)
walked_alias = tmp_path / "scan" / "alias" / "g.py"
assert file_hash(walked_alias, root) != file_hash(target, root)
def test_semantic_cache_self_heals_legacy_symlink_collision(
requires_symlinks, tmp_path
):
"""A poisoned legacy entry misses once, then walked groups round-trip."""
import json
from graphify.cache import check_semantic_cache, save_semantic_cache
_reset_stat_index()
physical_root = tmp_path / "repo"
physical_root.mkdir()
(physical_root / "real.md").write_text("# Shared\n")
(physical_root / "alias.md").symlink_to(physical_root / "real.md")
root = tmp_path / "scan"
root.symlink_to(physical_root, target_is_directory=True)
target = root / "real.md"
alias = root / "alias.md"
legacy_hash = file_hash(target, root)
legacy_entry = cache_dir(root, "semantic") / f"{legacy_hash}.json"
legacy_entry.write_text(json.dumps({
"nodes": [{"id": "alias-old", "source_file": "alias.md"}],
"edges": [],
}))
nodes, _, _, uncached = check_semantic_cache(
[str(target), str(alias)], root=root
)
assert nodes == []
assert uncached == [str(target), str(alias)]
saved = save_semantic_cache(
[
{"id": "real", "source_file": str(target)},
{"id": "alias", "source_file": str(alias)},
],
[],
root=root,
)
stored_sources = []
for path in (target, alias):
entry = cache_dir(root, "semantic") / f"{file_hash(path, root)}.json"
stored_sources.append(json.loads(entry.read_text())["nodes"][0]["source_file"])
nodes, _, _, uncached = check_semantic_cache(
[str(target), str(alias)], root=root
)
assert saved == 2
assert stored_sources == ["real.md", "alias.md"]
assert [node["id"] for node in nodes] == ["real", "alias"]
assert uncached == []
def test_semantic_symlink_policy_uses_walked_identity(
requires_symlinks, tmp_path
):
"""Alias authorization and partial state must not leak to its target."""
from graphify.cache import load_cached, save_semantic_cache
_reset_stat_index()
target = tmp_path / "real.md"
target.write_text("# Shared\n")
alias = tmp_path / "alias.md"
alias.symlink_to(target)
with pytest.warns(RuntimeWarning, match="out-of-scope source_file 'alias.md'"):
saved = save_semantic_cache(
[
{"id": "real", "source_file": "real.md"},
{"id": "alias", "source_file": "alias.md"},
],
[],
root=tmp_path,
allowed_source_files=[target],
partial_source_files=[alias],
)
target_entry = load_cached(
target, root=tmp_path, kind="semantic", allow_partial=True
)
assert saved == 1
assert target_entry is not None
assert target_entry.get("partial") is not True
assert load_cached(alias, root=tmp_path, kind="semantic") is None
def test_semantic_symlink_root_accepts_resolved_policy_paths_without_alias_leak(
requires_symlinks, tmp_path
):
"""Resolved root spellings apply only to the matching walked identity."""
from graphify.cache import load_cached, save_semantic_cache
_reset_stat_index()
physical_root = tmp_path / "repo"
physical_root.mkdir()
target = physical_root / "real.md"
target.write_text("# Shared\n")
alias = physical_root / "alias.md"
alias.symlink_to(target)
root = tmp_path / "scan"
root.symlink_to(physical_root, target_is_directory=True)
walked_target = root / "real.md"
walked_alias = root / "alias.md"
with pytest.warns(RuntimeWarning, match="out-of-scope source_file"):
saved = save_semantic_cache(
[
{"id": "real", "source_file": str(walked_target)},
{"id": "alias", "source_file": str(walked_alias)},
],
[],
root=root,
allowed_source_files=[walked_target.resolve()],
partial_source_files=[walked_target.resolve()],
)
target_entry = load_cached(
walked_target, root=root, kind="semantic", allow_partial=True
)
assert saved == 1
assert target_entry is not None
assert [node["id"] for node in target_entry["nodes"]] == ["real"]
assert target_entry["partial"] is True
assert load_cached(walked_target, root=root, kind="semantic") is None
assert load_cached(walked_alias, root=root, kind="semantic") is None
def test_semantic_symlink_root_keeps_external_policy_path_absolute(
requires_symlinks, tmp_path
):
"""An allowed absolute source outside a symlinked root stays external."""
from graphify.cache import load_cached, save_semantic_cache
_reset_stat_index()
physical_root = tmp_path / "repo"
physical_root.mkdir()
root = tmp_path / "scan"
root.symlink_to(physical_root, target_is_directory=True)
external = tmp_path / "external.md"
external.write_text("# External\n")
saved = save_semantic_cache(
[{"id": "external", "source_file": str(external)}],
[],
root=root,
allowed_source_files=[external],
partial_source_files=[external],
)
entry = load_cached(external, root=root, kind="semantic", allow_partial=True)
assert saved == 1
assert entry is not None
assert Path(entry["nodes"][0]["source_file"]) == external
assert entry["partial"] is True
assert load_cached(external, root=root, kind="semantic") is None
def test_semantic_prune_removes_orphan_entries(tmp_path):
"""Changing a file's content leaves the old content-hash entry orphaned;
pruning against the new live hash removes the stale entry and keeps the
current one."""
from graphify.cache import prune_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# A\n\nContent A.\n")
h_a = file_hash(f, tmp_path)
save_cached(f, {"nodes": [{"id": "a"}], "edges": []}, root=tmp_path, kind="semantic")
f.write_text("# B\n\nContent B.\n")
h_b = file_hash(f, tmp_path)
save_cached(f, {"nodes": [{"id": "b"}], "edges": []}, root=tmp_path, kind="semantic")
semantic_dir = cache_dir(tmp_path, "semantic")
assert (semantic_dir / f"{h_a}.json").exists()
assert (semantic_dir / f"{h_b}.json").exists()
pruned = prune_semantic_cache(tmp_path, {h_b})
assert pruned == 1
assert not (semantic_dir / f"{h_a}.json").exists()
assert (semantic_dir / f"{h_b}.json").exists()
def test_semantic_prune_keeps_live_unchanged_entries(tmp_path):
"""Pruning against the FULL live set must keep every live entry — guards
the trap of pruning against an incremental changed-subset, which would
delete all unchanged docs' valid entries."""
from graphify.cache import prune_semantic_cache
live_hashes = set()
for i in range(5):
f = tmp_path / f"doc{i}.md"
f.write_text(f"# Doc {i}\n\nBody {i}.\n")
save_cached(f, {"nodes": [{"id": str(i)}], "edges": []}, root=tmp_path, kind="semantic")
live_hashes.add(file_hash(f, tmp_path))
semantic_dir = cache_dir(tmp_path, "semantic")
assert len(list(semantic_dir.glob("*.json"))) == 5
pruned = prune_semantic_cache(tmp_path, live_hashes)
assert pruned == 0
assert len(list(semantic_dir.glob("*.json"))) == 5
def test_semantic_prune_handles_deleted_file(tmp_path):
"""An entry for a file that no longer exists (dropped from the live set) is
pruned."""
from graphify.cache import prune_semantic_cache
f = tmp_path / "gone.md"
f.write_text("# Gone\n\nWill be deleted.\n")
h = file_hash(f, tmp_path)
save_cached(f, {"nodes": [{"id": "g"}], "edges": []}, root=tmp_path, kind="semantic")
semantic_dir = cache_dir(tmp_path, "semantic")
assert (semantic_dir / f"{h}.json").exists()
f.unlink()
# Live set is empty: the file is gone, so its entry must be pruned.
pruned = prune_semantic_cache(tmp_path, set())
assert pruned == 1
assert not (semantic_dir / f"{h}.json").exists()
def test_semantic_prune_ignores_ast_and_tmp(tmp_path):
"""Prune touches only cache/semantic/*.json: AST entries and atomic-write
*.tmp temporaries are left untouched."""
from graphify.cache import prune_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n\nBody.\n")
# AST entry (different subtree) must survive.
save_cached(f, {"nodes": [{"id": "ast"}], "edges": []}, root=tmp_path, kind="ast")
ast_dir = cache_dir(tmp_path, "ast")
assert len(list(ast_dir.glob("*.json"))) == 1
# A semantic orphan .json (to be pruned) plus a .tmp temporary (to survive).
semantic_dir = cache_dir(tmp_path, "semantic")
(semantic_dir / "deadbeef.json").write_text('{"nodes": [], "edges": []}')
tmp_entry = semantic_dir / "deadbeef.tmp"
tmp_entry.write_text("partial")
pruned = prune_semantic_cache(tmp_path, set())
assert pruned == 1
assert not (semantic_dir / "deadbeef.json").exists()
assert tmp_entry.exists(), "*.tmp temporaries must not be swept"
assert len(list(ast_dir.glob("*.json"))) == 1, "AST entries must not be touched"
def test_save_semantic_cache_overwrites_by_default(tmp_path):
"""Default save_semantic_cache replaces a file's cached entry (the final,
authoritative write in the extract pipeline)."""
from graphify.cache import save_semantic_cache
f = tmp_path / "doc.md"; f.write_text("# Doc\n")
save_semantic_cache([{"id": "a", "source_file": "doc.md"}], [], root=tmp_path)
save_semantic_cache([{"id": "b", "source_file": "doc.md"}], [], root=tmp_path)
cached = load_cached(f, root=tmp_path, kind="semantic")
ids = {n["id"] for n in cached["nodes"]}
assert ids == {"b"}, "default must overwrite, not accumulate"
def test_save_semantic_cache_rejects_out_of_scope_source_file(tmp_path):
"""#1757: an undispatched file must keep its complete cache entry when a
semantic result misattributes a node to it."""
from graphify.cache import save_semantic_cache
intended = tmp_path / "intended.md"
intended.write_text("# Intended\n")
protected = tmp_path / "protected.md"
protected.write_text("# Protected\n")
save_semantic_cache(
[{"id": "original", "source_file": "protected.md"}],
[],
root=tmp_path,
)
nodes = [
{"id": "expected", "source_file": str(intended.resolve())},
{"id": "stray", "source_file": "protected.md"},
]
edges = [
{"source": "stray", "target": "expected", "source_file": "protected.md"},
]
hyperedges = [
{"id": "stray_hyperedge", "nodes": ["stray"], "source_file": "protected.md"},
]
with pytest.warns(RuntimeWarning, match="out-of-scope source_file 'protected.md'"):
saved = save_semantic_cache(
nodes,
edges,
hyperedges,
root=tmp_path,
allowed_source_files=["intended.md"],
)
assert saved == 1
intended_cache = load_cached(intended, root=tmp_path, kind="semantic")
assert {node["id"] for node in intended_cache["nodes"]} == {"expected"}
protected_cache = load_cached(protected, root=tmp_path, kind="semantic")
assert {node["id"] for node in protected_cache["nodes"]} == {"original"}
assert protected_cache["edges"] == []
assert protected_cache["hyperedges"] == []
# --- #1894: mode-namespaced semantic cache -----------------------------------
# `extract --mode deep` produces richer results than standard extraction, so
# deep entries live in their own namespace (cache/semantic-deep/). mode=None
# must stay byte-identical to the historical behavior: older installed skill
# flows call check/save without the parameter and must be unaffected.
def test_semantic_cache_deep_mode_roundtrip_under_deep_namespace(tmp_path):
"""mode='deep' saves under cache/semantic-deep/ and reads back from it."""
from graphify.cache import check_semantic_cache, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n\nBody.\n")
saved = save_semantic_cache(
[{"id": "deep_n", "source_file": "doc.md"}], [], root=tmp_path, mode="deep"
)
assert saved == 1
deep_dir = tmp_path / "graphify-out" / "cache" / "semantic-deep"
h = file_hash(f, tmp_path)
assert (deep_dir / f"{h}.json").exists(), (
"deep entry must land under cache/semantic-deep/"
)
# And NOT in the plain namespace.
plain_dir = tmp_path / "graphify-out" / "cache" / "semantic"
assert not (plain_dir / f"{h}.json").exists()
nodes, edges, hyper, uncached = check_semantic_cache(
[str(f)], root=tmp_path, mode="deep"
)
assert [n["id"] for n in nodes] == ["deep_n"]
assert uncached == []
def test_semantic_cache_deep_invisible_to_plain_reads_and_vice_versa(tmp_path):
"""Deep entries must not satisfy mode=None reads (and plain entries must
not satisfy deep reads) — the namespaces are fully isolated."""
from graphify.cache import check_semantic_cache, save_semantic_cache
deep_doc = tmp_path / "deep.md"
deep_doc.write_text("# Deep\n")
plain_doc = tmp_path / "plain.md"
plain_doc.write_text("# Plain\n")
save_semantic_cache([{"id": "d", "source_file": "deep.md"}], [],
root=tmp_path, mode="deep")
save_semantic_cache([{"id": "p", "source_file": "plain.md"}], [],
root=tmp_path) # mode omitted: historical call shape
# Plain read: deep entry is a miss, plain entry is a hit.
nodes, _, _, uncached = check_semantic_cache(
[str(deep_doc), str(plain_doc)], root=tmp_path
)
assert [n["id"] for n in nodes] == ["p"]
assert uncached == [str(deep_doc)]
# Deep read: mirror image.
nodes, _, _, uncached = check_semantic_cache(
[str(deep_doc), str(plain_doc)], root=tmp_path, mode="deep"
)
assert [n["id"] for n in nodes] == ["d"]
assert uncached == [str(plain_doc)]
def test_semantic_cache_mode_none_layout_unchanged(tmp_path):
"""Omitting mode writes exactly the historical cache/semantic/ layout —
forward-compat for older installed callers that never pass mode."""
from graphify.cache import check_semantic_cache, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "n", "source_file": "doc.md"}], [], root=tmp_path)
h = file_hash(f, tmp_path)
assert (tmp_path / "graphify-out" / "cache" / "semantic" / f"{h}.json").exists()
assert not (tmp_path / "graphify-out" / "cache" / "semantic-deep").exists(), (
"mode=None must never create the deep namespace"
)
nodes, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path)
assert [n["id"] for n in nodes] == ["n"] and uncached == []
def test_clear_cache_removes_deep_namespace(tmp_path):
"""clear_cache sweeps cache/semantic-deep/ alongside semantic/ and ast/."""
from graphify.cache import save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "p", "source_file": "doc.md"}], [], root=tmp_path)
save_semantic_cache([{"id": "d", "source_file": "doc.md"}], [],
root=tmp_path, mode="deep")
base = tmp_path / "graphify-out" / "cache"
assert list((base / "semantic").glob("*.json"))
assert list((base / "semantic-deep").glob("*.json"))
clear_cache(tmp_path)
assert not list(base.rglob("*.json")), (
"clear_cache must remove entries in BOTH semantic namespaces"
)
def test_cached_files_includes_deep_namespace(tmp_path):
"""cached_files reports deep-namespace entries too."""
from graphify.cache import save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "d", "source_file": "doc.md"}], [],
root=tmp_path, mode="deep")
assert file_hash(f, tmp_path) in cached_files(tmp_path)
def test_semantic_prune_sweeps_both_namespaces_against_same_live_set(tmp_path):
"""#1894 follow-up to #1527: prune must sweep cache/semantic/ AND
cache/semantic-deep/ against the SAME live-hash set (liveness is
content-based, mode-independent). Orphans go in both namespaces; live
entries survive in both."""
from graphify.cache import prune_semantic_cache, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# A\n\nContent A.\n")
h_old = file_hash(f, tmp_path)
save_semantic_cache([{"id": "pa", "source_file": "doc.md"}], [], root=tmp_path)
save_semantic_cache([{"id": "da", "source_file": "doc.md"}], [],
root=tmp_path, mode="deep")
f.write_text("# B\n\nContent B.\n")
h_live = file_hash(f, tmp_path)
save_semantic_cache([{"id": "pb", "source_file": "doc.md"}], [], root=tmp_path)
save_semantic_cache([{"id": "db", "source_file": "doc.md"}], [],
root=tmp_path, mode="deep")
plain_dir = tmp_path / "graphify-out" / "cache" / "semantic"
deep_dir = tmp_path / "graphify-out" / "cache" / "semantic-deep"
for d in (plain_dir, deep_dir):
assert (d / f"{h_old}.json").exists()
assert (d / f"{h_live}.json").exists()
pruned = prune_semantic_cache(tmp_path, {h_live})
assert pruned == 2, "one orphan in EACH namespace must be pruned"
for d in (plain_dir, deep_dir):
assert not (d / f"{h_old}.json").exists(), f"orphan survived in {d.name}"
assert (d / f"{h_live}.json").exists(), f"live entry pruned from {d.name}"
def test_save_semantic_cache_merge_existing_unions(tmp_path):
"""#1715: merge_existing=True unions with the prior entry so a file split
across chunks (checkpointed per chunk) keeps every slice."""
from graphify.cache import save_semantic_cache
f = tmp_path / "big.md"; f.write_text("# Big\n")
# chunk 1 slice
save_semantic_cache([{"id": "a", "source_file": "big.md"}],
[{"source": "a", "target": "x", "source_file": "big.md"}],
root=tmp_path, merge_existing=True)
# chunk 2 slice for the same file
save_semantic_cache([{"id": "b", "source_file": "big.md"}], [],
root=tmp_path, merge_existing=True)
cached = load_cached(f, root=tmp_path, kind="semantic")
ids = {n["id"] for n in cached["nodes"]}
assert ids == {"a", "b"}, "merge_existing must union both chunk slices"
assert len(cached["edges"]) == 1
def test_save_semantic_cache_drops_edges_to_out_of_scope_nodes(tmp_path):
"""#1916: an edge in an ALLOWED file's group referencing a node grouped
under an out-of-scope REAL file used to be written verbatim, so on replay
(check_semantic_cache) it dangled forever — the #1895 merged-result filter
runs after this checkpoint write and is bypassed entirely on replay. The
written entry must carry no reference to the skipped id, while a
duplicate-attribution node (also defined in a written group) must not be
over-pruned."""
from graphify.cache import check_semantic_cache, save_semantic_cache
allowed = tmp_path / "allowed.md"
allowed.write_text("# Allowed\n")
outside = tmp_path / "outside.md"
outside.write_text("# Outside\n")
nodes = [
{"id": "kept", "source_file": "allowed.md"},
{"id": "stray", "source_file": "outside.md"},
# duplicate attribution: same id defined in a written AND a skipped group
{"id": "dup", "source_file": "allowed.md"},
{"id": "dup", "source_file": "outside.md"},
]
edges = [
{"source": "kept", "target": "stray", "source_file": "allowed.md"},
{"source": "stray", "target": "kept", "source_file": "allowed.md"},
{"source": "kept", "target": "dup", "source_file": "allowed.md"},
]
with pytest.warns(RuntimeWarning, match="out-of-scope source_file"):
saved = save_semantic_cache(
nodes, edges, root=tmp_path, allowed_source_files=["allowed.md"]
)
assert saved == 1
cached_nodes, cached_edges, _, uncached = check_semantic_cache(
[str(allowed)], root=tmp_path
)
assert uncached == []
assert {n["id"] for n in cached_nodes} == {"kept", "dup"}
pairs = [(e["source"], e["target"]) for e in cached_edges]
assert pairs == [("kept", "dup")], "edges touching the skipped id must be dropped"
def test_save_semantic_cache_drops_edges_to_ghost_file_nodes(tmp_path):
"""#1916 (ghost variant): a node group whose source_file does not exist is
silently skipped by the write loop; edges in a written group referencing
its node ids must not survive into the cache."""
from graphify.cache import check_semantic_cache, save_semantic_cache
real = tmp_path / "real.md"
real.write_text("# Real\n")
nodes = [
{"id": "kept", "source_file": "real.md"},
{"id": "phantom", "source_file": "ghost.md"}, # no such file on disk
]
edges = [
{"source": "kept", "target": "phantom", "source_file": "real.md"},
{"source": "kept", "target": "kept", "relation": "self", "source_file": "real.md"},
]
saved = save_semantic_cache(
nodes, edges, root=tmp_path, allowed_source_files=["real.md"]
)
assert saved == 1
cached_nodes, cached_edges, _, uncached = check_semantic_cache(
[str(real)], root=tmp_path
)
assert uncached == []
assert {n["id"] for n in cached_nodes} == {"kept"}
pairs = [(e["source"], e["target"]) for e in cached_edges]
assert pairs == [("kept", "kept")]
def test_save_semantic_cache_drops_hyperedges_touching_skipped_nodes(tmp_path):
"""#1916: a hyperedge whose member list intersects the skipped ids is
dropped whole (mirroring the #1895 semantics), while hyperedges over
surviving nodes are kept."""
from graphify.cache import check_semantic_cache, save_semantic_cache
allowed = tmp_path / "allowed.md"
allowed.write_text("# Allowed\n")
outside = tmp_path / "outside.md"
outside.write_text("# Outside\n")
nodes = [
{"id": "kept", "source_file": "allowed.md"},
{"id": "kept2", "source_file": "allowed.md"},
{"id": "stray", "source_file": "outside.md"},
]
hyperedges = [
{"id": "he_bad", "nodes": ["kept", "stray"], "source_file": "allowed.md"},
{"id": "he_ok", "nodes": ["kept", "kept2"], "source_file": "allowed.md"},
]
with pytest.warns(RuntimeWarning, match="out-of-scope source_file"):
save_semantic_cache(
nodes, [], hyperedges, root=tmp_path, allowed_source_files=["allowed.md"]
)
_, _, cached_hyperedges, uncached = check_semantic_cache(
[str(allowed)], root=tmp_path
)
assert uncached == []
assert {h["id"] for h in cached_hyperedges} == {"he_ok"}
def test_save_semantic_cache_unscoped_preserves_dangling_refs_verbatim(tmp_path):
"""#1916 guard-rail: unscoped callers (allowed_source_files=None) must stay
byte-identical — no pruning happens even when an edge or hyperedge
references a node grouped under a ghost file."""
from graphify.cache import save_semantic_cache
doc = tmp_path / "doc.md"
doc.write_text("# Doc\n")
nodes = [
{"id": "a", "source_file": "doc.md"},
{"id": "ghost_n", "source_file": "ghost.md"}, # skipped group (no file)
]
edges = [{"source": "a", "target": "ghost_n", "source_file": "doc.md"}]
hyperedges = [{"id": "he", "nodes": ["a", "ghost_n"], "source_file": "doc.md"}]
saved = save_semantic_cache(nodes, edges, hyperedges, root=tmp_path)
assert saved == 1
import json
raw = json.loads(
(cache_dir(tmp_path, "semantic") / f"{file_hash(doc, tmp_path)}.json").read_text()
)
assert raw["edges"] == edges
assert raw["hyperedges"] == hyperedges
def test_save_semantic_cache_merge_existing_prunes_only_incoming(tmp_path):
"""#1916 + #1715: with merge_existing=True (the llm.py checkpoint path),
only the INCOMING slice is pruned before the union — the prior cached
entry's valid edges must survive untouched."""
from graphify.cache import save_semantic_cache
big = tmp_path / "big.md"
big.write_text("# Big\n")
other = tmp_path / "other.md"
other.write_text("# Other\n")
# checkpoint 1: a clean slice
save_semantic_cache(
[{"id": "a", "source_file": "big.md"}],
[{"source": "a", "target": "a", "relation": "self", "source_file": "big.md"}],
root=tmp_path,
merge_existing=True,
allowed_source_files=["big.md"],
)
# checkpoint 2: incoming slice with a dangling edge to an out-of-scope node
nodes2 = [
{"id": "b", "source_file": "big.md"},
{"id": "stray", "source_file": "other.md"},
]
edges2 = [
{"source": "b", "target": "stray", "source_file": "big.md"},
{"source": "a", "target": "b", "source_file": "big.md"},
]
with pytest.warns(RuntimeWarning, match="out-of-scope source_file"):
save_semantic_cache(
nodes2, edges2, root=tmp_path, merge_existing=True,
allowed_source_files=["big.md"],
)
cached = load_cached(big, root=tmp_path, kind="semantic")
assert {n["id"] for n in cached["nodes"]} == {"a", "b"}
pairs = [(e["source"], e["target"]) for e in cached["edges"]]
assert ("a", "a") in pairs, "prior entry's valid edge must survive the union"
assert ("a", "b") in pairs, "incoming valid edge must be kept"
assert not any("stray" in p for p in pairs)
# --- extraction-prompt fingerprinting (#1939) -------------------------------
def test_prompt_fingerprint_stable_and_prompt_sensitive(tmp_path):
"""The fingerprint is stable for identical prompts and differs when the
prompt text changes — the whole invalidation signal rests on this."""
from graphify.cache import prompt_fingerprint
assert prompt_fingerprint("extract a graph") == prompt_fingerprint("extract a graph")
assert prompt_fingerprint("extract a graph") != prompt_fingerprint("extract a graph v2")
# A Path is read and hashed as its contents, so the skill path (which loads
# references/extraction-spec.md) and the Python path agree on the same text.
spec = tmp_path / "extraction-spec.md"
spec.write_text("extract a graph", encoding="utf-8")
assert prompt_fingerprint(spec) == prompt_fingerprint("extract a graph")
def test_prompt_fingerprint_ignores_line_endings(tmp_path):
"""A CRLF checkout of the same spec must not look like a prompt change —
otherwise every Windows run re-bills the whole corpus."""
from graphify.cache import prompt_fingerprint
assert prompt_fingerprint("a\r\nb\r\n") == prompt_fingerprint("a\nb\n")
assert prompt_fingerprint("a \nb\n") == prompt_fingerprint("a\nb\n")
def test_semantic_cache_prompt_change_invalidates(tmp_path):
"""The reported bug (#1939): after the extraction prompt changes, an
unchanged file must MISS instead of replaying the older vintage."""
from graphify.cache import check_semantic_cache, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n\nBody.\n")
save_semantic_cache([{"id": "old_vintage", "source_file": "doc.md"}], [],
root=tmp_path, prompt="PROMPT V1")
# Same prompt: hit.
nodes, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path, prompt="PROMPT V1")
assert [n["id"] for n in nodes] == ["old_vintage"]
assert uncached == []
# Prompt changed (an upgrade shipped a new extraction-spec): must re-extract.
nodes, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path, prompt="PROMPT V2")
assert nodes == []
assert uncached == [str(f)], "a new prompt must not replay the old prompt's entry"
# V2's results land in their own namespace and do not clobber V1's, so
# rolling back to V1 still hits rather than re-billing.
save_semantic_cache([{"id": "new_vintage", "source_file": "doc.md"}], [],
root=tmp_path, prompt="PROMPT V2")
nodes, _, _, _ = check_semantic_cache([str(f)], root=tmp_path, prompt="PROMPT V2")
assert [n["id"] for n in nodes] == ["new_vintage"]
nodes, _, _, _ = check_semantic_cache([str(f)], root=tmp_path, prompt="PROMPT V1")
assert [n["id"] for n in nodes] == ["old_vintage"]
def test_semantic_cache_prompt_namespaced_layout(tmp_path):
"""Fingerprinted entries live under cache/semantic/p{fp}/, never flat."""
from graphify.cache import prompt_fingerprint, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "n", "source_file": "doc.md"}], [],
root=tmp_path, prompt="PROMPT V1")
sem = tmp_path / "graphify-out" / "cache" / "semantic"
h = file_hash(f, tmp_path)
assert (sem / f"p{prompt_fingerprint('PROMPT V1')}" / f"{h}.json").exists()
assert not (sem / f"{h}.json").exists(), (
"a known-vintage entry must never be written into the flat unknown-vintage layout"
)
def test_semantic_cache_prompt_and_mode_compose(tmp_path):
"""The prompt fingerprint nests inside the deep namespace (#1894), so the
two dimensions are independent."""
from graphify.cache import check_semantic_cache, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "d", "source_file": "doc.md"}], [],
root=tmp_path, mode="deep", prompt="PROMPT V1")
deep = tmp_path / "graphify-out" / "cache" / "semantic-deep"
assert list(deep.glob("p*/*.json")), "deep + prompt must nest under semantic-deep/p{fp}/"
# Right mode, wrong prompt -> miss. Right prompt, wrong mode -> miss.
_, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path, mode="deep",
prompt="PROMPT V2")
assert uncached == [str(f)]
_, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path, prompt="PROMPT V1")
assert uncached == [str(f)]
# Both right -> hit.
nodes, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path, mode="deep",
prompt="PROMPT V1")
assert [n["id"] for n in nodes] == ["d"] and uncached == []
def test_semantic_cache_legacy_entries_served_with_warning(tmp_path):
"""Entries written before fingerprinting have unknowable vintage. They are
still served — dropping them would re-bill a whole corpus on upgrade — but
the user is told how many, which is the signal #1939 says is missing today."""
from graphify.cache import check_semantic_cache, save_semantic_cache
a = tmp_path / "a.md"
a.write_text("# A\n")
b = tmp_path / "b.md"
b.write_text("# B\n")
# Pre-fingerprint writes: the historical flat layout.
save_semantic_cache([{"id": "a_old", "source_file": "a.md"},
{"id": "b_old", "source_file": "b.md"}], [], root=tmp_path)
with pytest.warns(RuntimeWarning, match="2 semantic cache entries predate"):
nodes, _, _, uncached = check_semantic_cache(
[str(a), str(b)], root=tmp_path, prompt="PROMPT V1"
)
assert {n["id"] for n in nodes} == {"a_old", "b_old"}
assert uncached == []
def test_semantic_cache_fingerprinted_entry_beats_legacy(tmp_path):
"""Once a file is re-extracted under the current prompt, its fingerprinted
entry wins and the stale flat one is no longer consulted (no warning)."""
import warnings as _warnings
from graphify.cache import check_semantic_cache, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "unknown_vintage", "source_file": "doc.md"}], [],
root=tmp_path) # legacy flat
save_semantic_cache([{"id": "current", "source_file": "doc.md"}], [],
root=tmp_path, prompt="PROMPT V1")
with _warnings.catch_warnings():
_warnings.simplefilter("error") # any legacy warning would raise here
nodes, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path,
prompt="PROMPT V1")
assert [n["id"] for n in nodes] == ["current"]
assert uncached == []
def test_semantic_cache_merge_existing_never_fuses_legacy_vintage(tmp_path):
"""merge_existing must not union a pre-fingerprint entry into a write it is
about to stamp as current-vintage — that would mix two prompts inside one
entry and then attest the result to a prompt that produced half of it."""
from graphify.cache import load_cached, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "unknown_vintage", "source_file": "doc.md"}], [],
root=tmp_path) # legacy flat
save_semantic_cache([{"id": "current", "source_file": "doc.md"}], [],
root=tmp_path, merge_existing=True, prompt="PROMPT V1")
entry = load_cached(f, root=tmp_path, kind="semantic", prompt="PROMPT V1")
assert [n["id"] for n in entry["nodes"]] == ["current"]
# Within one prompt, merge_existing still unions across checkpoints.
save_semantic_cache([{"id": "second_chunk", "source_file": "doc.md"}], [],
root=tmp_path, merge_existing=True, prompt="PROMPT V1")
entry = load_cached(f, root=tmp_path, kind="semantic", prompt="PROMPT V1")
assert {n["id"] for n in entry["nodes"]} == {"current", "second_chunk"}
def test_semantic_prune_and_clear_reach_fingerprint_subdirs(tmp_path):
"""A glob that stopped at the top level would leave every fingerprinted
entry unprunable, re-growing the unbounded-orphan problem of #1527."""
from graphify.cache import (
cached_files, clear_cache, prune_semantic_cache, save_semantic_cache,
)
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "n", "source_file": "doc.md"}], [],
root=tmp_path, prompt="PROMPT V1")
h = file_hash(f, tmp_path)
assert h in cached_files(tmp_path), "cached_files must see fingerprinted entries"
# Live: kept.
assert prune_semantic_cache(tmp_path, {h}) == 0
# Orphaned (content changed / file deleted): pruned.
assert prune_semantic_cache(tmp_path, set()) == 1
save_semantic_cache([{"id": "n", "source_file": "doc.md"}], [],
root=tmp_path, prompt="PROMPT V1")
clear_cache(tmp_path)
assert not list((tmp_path / "graphify-out" / "cache" / "semantic").glob("**/*.json"))
def test_semantic_cache_unreadable_prompt_file_warns_and_falls_back(tmp_path):
"""A skill snippet substitutes SPEC_PATH by hand. If it lands on a path that
isn't there, the fallback to the unattributed layout must be loud: silently
reverting to unversioned keying is exactly the #1939 behavior being fixed."""
from graphify.cache import check_semantic_cache, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "n", "source_file": "doc.md"}], [], root=tmp_path)
with pytest.warns(RuntimeWarning, match="could not read extraction prompt"):
nodes, _, _, uncached = check_semantic_cache(
[str(f)], root=tmp_path, prompt_file=str(tmp_path / "nope.md")
)
# Fell back rather than aborting the run.
assert [n["id"] for n in nodes] == ["n"] and uncached == []
def test_prompt_file_reflects_edited_spec(tmp_path):
"""The prompt-file fingerprint is memoized per (path, size, mtime); an edited
spec must still register as a new prompt rather than reusing a stale memo."""
from graphify.cache import check_semantic_cache, save_semantic_cache
spec = tmp_path / "extraction-spec.md"
spec.write_text("prompt one", encoding="utf-8")
f = tmp_path / "doc.md"
f.write_text("# Doc\n")
save_semantic_cache([{"id": "v1", "source_file": "doc.md"}], [],
root=tmp_path, prompt_file=str(spec))
nodes, _, _, _ = check_semantic_cache([str(f)], root=tmp_path, prompt_file=str(spec))
assert [n["id"] for n in nodes] == ["v1"]
# An upgrade rewrites the spec: the same file path is now a different prompt.
import os as _os
spec.write_text("prompt two — rewritten by an upgrade", encoding="utf-8")
_os.utime(spec, ns=(0, 0)) # force a distinct stat signature
_, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path, prompt_file=str(spec))
assert uncached == [str(f)], "an edited spec must invalidate, not reuse the memo"
# --- stat-fastpath racily-clean guard ---------------------------------------
# (size, mtime_ns) alone cannot prove a file is unchanged: NTFS advances mtime
# on a ~15.6 ms tick, so a same-length rewrite inside one tick leaves the
# signature identical and the memo used to return the PREVIOUS content's
# digest. These two tests pin both halves of the fix — the hole is closed, and
# the fastpath still actually fires for files whose mtime tick has closed.
def test_file_hash_detects_same_size_rewrite_within_one_mtime_tick(tmp_path):
"""A same-length edit must change the digest even when the filesystem
reports an identical (size, mtime_ns) for both writes.
The collision is forced with utime rather than raced for: on a filesystem
with fine-grained timestamps the two writes would land in different ticks
and the memo would never be consulted, making the test vacuous. Pinning
both writes to one mtime models the coarse-granularity filesystem (NTFS,
FAT, NFS) on every host.
"""
import os as _os
_reset_stat_index()
f = tmp_path / "mod.py"
f.write_text("x = 1 # aaa\n", encoding="utf-8")
st = f.stat()
h1 = file_hash(f, tmp_path)
f.write_text("x = 2 # bbb\n", encoding="utf-8") # same length, new content
_os.utime(f, ns=(st.st_atime_ns, st.st_mtime_ns)) # ...inside the same tick
assert f.stat().st_size == st.st_size and f.stat().st_mtime_ns == st.st_mtime_ns, (
"test setup failed to reproduce an identical stat signature"
)
h2 = file_hash(f, tmp_path)
assert h1 != h2, "same-size rewrite returned the previous content's digest"
def test_file_hash_fastpath_still_serves_a_settled_file(tmp_path, monkeypatch):
"""The guard must not disable the cache: once a file's mtime tick has
closed, the digest is served from the index without re-reading."""
_reset_stat_index()
f = tmp_path / "mod.py"
f.write_text("x = 1\n", encoding="utf-8")
# Backdate well past the granularity window so the entry is provably clean.
import os as _os
old_ns = f.stat().st_mtime_ns - 60 * 1_000_000_000
_os.utime(f, ns=(old_ns, old_ns))
first = file_hash(f, tmp_path)
reads = []
real_read_bytes = Path.read_bytes
monkeypatch.setattr(Path, "read_bytes",
lambda self: (reads.append(self), real_read_bytes(self))[1])
second = file_hash(f, tmp_path)
assert second == first
assert reads == [], "settled file was re-read; the stat fastpath is dead"
def test_corrupt_semantic_entry_warns_and_is_a_miss(tmp_path):
"""A corrupt (invalid-JSON) cache entry must not be silently swallowed
(#2405). Left unreported it fails to parse on every future run, re-billing
the semantic extraction forever with no diagnostic. check_semantic_cache
treats it as a miss (uncached) AND emits one aggregate warning naming the
count, mirroring the pre-fingerprint legacy-hit warning."""
from graphify.cache import (
check_semantic_cache,
save_semantic_cache,
cache_dir,
)
f = tmp_path / "doc.md"
f.write_text("# Doc\n\nBody.\n")
save_semantic_cache([{"id": "n", "source_file": "doc.md"}], [], root=tmp_path)
# Corrupt the on-disk entry (e.g. an old producer wrote unescaped
# backslashes, or a partial write left truncated JSON).
h = file_hash(f, tmp_path)
entry = cache_dir(tmp_path, "semantic") / f"{h}.json"
assert entry.exists()
entry.write_text('{"nodes": [ this is not valid json')
with pytest.warns(RuntimeWarning, match="corrupt"):
nodes, _, _, uncached = check_semantic_cache([str(f)], root=tmp_path)
# The corrupt entry is a miss, so the file is re-dispatched for extraction.
assert nodes == []
assert uncached == [str(f)]
# --- #2927: zero-node semantic cache rejection and healing -------------------
def test_edge_only_semantic_result_not_cached(tmp_path):
"""#2927: an edge-only semantic result (0 nodes, 0 hyperedges) represents an
omission by the model and must NOT be written to cache, so subsequent runs
can re-dispatch and retry the file (#933/#1666)."""
from graphify.cache import check_semantic_cache, load_cached, save_semantic_cache
f = tmp_path / "doc.md"
f.write_text("# Architecture\nSome prose.\n", encoding="utf-8")
edges = [{"source": "auth_a", "target": "auth_b", "source_file": "doc.md"}]
saved = save_semantic_cache([], edges, root=tmp_path, prompt="PROMPT V1")
assert saved == 0, "edge-only result must not be saved to cache"
# load_cached must return None (miss)
assert load_cached(f, root=tmp_path, kind="semantic", prompt="PROMPT V1") is None
# check_semantic_cache must treat it as uncached
nodes, edges_out, hyper_out, uncached = check_semantic_cache([str(f)], root=tmp_path, prompt="PROMPT V1")
assert nodes == [] and edges_out == [] and hyper_out == []
assert uncached == [str(f)]
def test_node_only_and_node_edge_semantic_results_cached(tmp_path):
"""Normal extractions (nodes-only and nodes+edges) continue to cache normally."""
from graphify.cache import load_cached, save_semantic_cache
f1 = tmp_path / "doc1.md"
f1.write_text("# Doc 1\n", encoding="utf-8")
f2 = tmp_path / "doc2.md"
f2.write_text("# Doc 2\n", encoding="utf-8")
# Node-only
saved1 = save_semantic_cache([{"id": "n1", "source_file": "doc1.md"}], [], root=tmp_path, prompt="P")
assert saved1 == 1
loaded1 = load_cached(f1, root=tmp_path, kind="semantic", prompt="P")
assert loaded1 is not None and len(loaded1["nodes"]) == 1
# Node + edge
saved2 = save_semantic_cache(
[{"id": "n2", "source_file": "doc2.md"}],
[{"source": "n2", "target": "n2", "source_file": "doc2.md"}],
root=tmp_path,
prompt="P",
)
assert saved2 == 1
loaded2 = load_cached(f2, root=tmp_path, kind="semantic", prompt="P")
assert loaded2 is not None and len(loaded2["nodes"]) == 1 and len(loaded2["edges"]) == 1
def test_hyperedge_only_semantic_result_cached(tmp_path):
"""#1920: hyperedge-only documents are valid semantic output and must be cached."""
from graphify.cache import check_semantic_cache, load_cached, save_semantic_cache
f = tmp_path / "hyper.md"
f.write_text("# Pipeline Concept\n", encoding="utf-8")
hyperedges = [
{"id": "h1", "label": "Pipeline", "nodes": ["a", "b", "c"], "source_file": "hyper.md"}
]
saved = save_semantic_cache([], [], hyperedges, root=tmp_path, prompt="PROMPT V1")
assert saved == 1, "hyperedge-only result must be saved to cache (#1920)"
loaded = load_cached(f, root=tmp_path, kind="semantic", prompt="PROMPT V1")
assert loaded is not None
assert len(loaded["hyperedges"]) == 1
_, _, cached_hyper, uncached = check_semantic_cache([str(f)], root=tmp_path, prompt="PROMPT V1")
assert uncached == []
assert len(cached_hyper) == 1
def test_poisoned_edge_only_cache_entry_treated_as_miss(tmp_path):
"""#2927 healing: a legacy on-disk cache entry containing edges but no nodes
or hyperedges must be rejected by load_cached as a cache MISS."""
import json
from graphify.cache import cache_dir, file_hash, load_cached, prompt_fingerprint
f = tmp_path / "poisoned.md"
f.write_text("# Poisoned\n", encoding="utf-8")
# Manually seed a legacy poisoned cache file (nodes: [], edges: [...])
prompt = "PROMPT V1"
fp = prompt_fingerprint(prompt)
cdir = cache_dir(tmp_path, "semantic", fp)
cdir.mkdir(parents=True, exist_ok=True)
h = file_hash(f, tmp_path)
(cdir / f"{h}.json").write_text(
json.dumps({
"nodes": [],
"edges": [{"source": "x", "target": "y", "source_file": "poisoned.md"}],
"hyperedges": [],
}),
encoding="utf-8",
)
# load_cached must reject the poisoned entry
assert load_cached(f, root=tmp_path, kind="semantic", prompt=prompt) is None
def test_existing_hyperedge_only_cache_entry_remains_hit(tmp_path):
"""#1920 / #2927: an existing on-disk cache entry with hyperedges but no nodes
remains a valid cache hit."""
import json
from graphify.cache import cache_dir, file_hash, load_cached, prompt_fingerprint
f = tmp_path / "valid_hyper.md"
f.write_text("# Hyper\n", encoding="utf-8")
prompt = "PROMPT V1"
fp = prompt_fingerprint(prompt)
cdir = cache_dir(tmp_path, "semantic", fp)
cdir.mkdir(parents=True, exist_ok=True)
h = file_hash(f, tmp_path)
(cdir / f"{h}.json").write_text(
json.dumps({
"nodes": [],
"edges": [],
"hyperedges": [{"id": "h1", "label": "Group", "nodes": ["a", "b", "c"], "source_file": "valid_hyper.md"}],
}),
encoding="utf-8",
)
loaded = load_cached(f, root=tmp_path, kind="semantic", prompt=prompt)
assert loaded is not None
assert len(loaded["hyperedges"]) == 1
# --- #2926: graph-side scope filter -------------------------------------------
# The #1757 guard protects the cache write, but the unfiltered fresh result
# also feeds build_merge(), whose replace-set logic swaps a non-dispatched
# file's entire prior contribution for a stray fragment. scope_semantic_result
# applies the same allowlist to the result dict before it reaches the merge.
def test_scope_semantic_result_drops_out_of_scope_groups(tmp_path):
"""Stray items attributed to a non-dispatched file are removed; allowed
and source-less items pass through."""
from graphify.cache import scope_semantic_result
result = {
"nodes": [
{"id": "kept", "source_file": "intended.md"},
{"id": "stray", "source_file": "protected.md"},
{"id": "phantom", "source_file": "src/foo.ts"}, # nonexistent path
{"id": "no_source"}, # no source_file: passes through
],
"edges": [
{"source": "kept", "target": "other", "source_file": "intended.md"},
{"source": "stray", "target": "kept", "source_file": "protected.md"},
],
"hyperedges": [
{"id": "h_kept", "nodes": ["kept"], "source_file": "intended.md"},
{"id": "h_stray", "nodes": ["stray"], "source_file": "protected.md"},
],
}
dropped_files, dropped_items = scope_semantic_result(
result, root=tmp_path, allowed_source_files=["intended.md"],
)
assert [n["id"] for n in result["nodes"]] == ["kept", "no_source"]
assert [e["source"] for e in result["edges"]] == ["kept"]
assert [h["id"] for h in result["hyperedges"]] == ["h_kept"]
assert dropped_files == {"protected.md", "src/foo.ts"}
assert dropped_items == 4 # 2 stray nodes + 1 stray edge + 1 stray hyperedge
def test_scope_semantic_result_matches_absolute_and_relative_forms(tmp_path):
"""An absolute in-root source_file and its relative form are the same
identity — both must match the allowlist entry (#2197 normalization)."""
from graphify.cache import scope_semantic_result
result = {
"nodes": [
{"id": "abs", "source_file": str(tmp_path / "doc.md")},
{"id": "rel", "source_file": "doc.md"},
],
"edges": [],
"hyperedges": [],
}
dropped_files, dropped_items = scope_semantic_result(
result, root=tmp_path, allowed_source_files=["doc.md"],
)
assert [n["id"] for n in result["nodes"]] == ["abs", "rel"]
assert dropped_files == set()
assert dropped_items == 0
def test_scope_semantic_result_prunes_edges_referencing_dropped_ids(tmp_path):
"""#1916 mirror: an edge attributed to an ALLOWED file that references a
node id only defined by a DROPPED group would materialize that id as a
phantom node at build time; it must be dropped too. A duplicate-attribution
id (defined by both a kept and a dropped group) keeps its edges."""
from graphify.cache import scope_semantic_result
result = {
"nodes": [
{"id": "kept", "source_file": "a.md"},
{"id": "shared", "source_file": "a.md"}, # also defined by b.md
{"id": "shared", "source_file": "b.md"}, # duplicate attribution
{"id": "gone", "source_file": "c.md"},
],
"edges": [
{"source": "kept", "target": "shared", "source_file": "a.md"},
{"source": "kept", "target": "gone", "source_file": "a.md"},
],
"hyperedges": [
{"id": "h1", "nodes": ["kept", "gone"], "source_file": "a.md"},
{"id": "h2", "nodes": ["kept", "shared"], "source_file": "a.md"},
],
}
scope_semantic_result(result, root=tmp_path,
allowed_source_files=["a.md"])
# The b.md COPY of "shared" is dropped as out-of-scope, but because a kept
# node also defines that id, references to it must NOT be pruned.
assert [n["id"] for n in result["nodes"]] == ["kept", "shared"]
assert [e["target"] for e in result["edges"]] == ["shared"]
assert [h["id"] for h in result["hyperedges"]] == ["h2"]
def test_scope_semantic_result_unscoped_is_a_no_op():
"""allowed_source_files=None must leave the result untouched (same contract
as save_semantic_cache's unscoped callers)."""
from graphify.cache import scope_semantic_result
result = {
"nodes": [{"id": "n", "source_file": "anywhere.md"}],
"edges": [],
"hyperedges": [],
}
dropped_files, dropped_items = scope_semantic_result(result, root=Path("."))
assert (dropped_files, dropped_items) == (set(), 0)
assert [n["id"] for n in result["nodes"]] == ["n"]
# -- #5083: mark a degraded entry partial so the next run self-heals ----------
def test_mark_semantic_cache_partial_forces_next_run_miss(tmp_path):
"""A degraded entry stays on disk for inspection but stops being served."""
from graphify.cache import load_cached, mark_semantic_cache_partial, save_semantic_cache
_reset_stat_index()
doc = tmp_path / "doc.md"
doc.write_text("# Doc\n")
save_semantic_cache([{"id": "doc", "source_file": "doc.md"}], [], root=tmp_path)
assert load_cached(doc, root=tmp_path, kind="semantic") is not None
marked = mark_semantic_cache_partial(["doc.md"], root=tmp_path)
assert len(marked) == 1
assert marked[0].exists()
assert load_cached(doc, root=tmp_path, kind="semantic") is None
assert load_cached(doc, root=tmp_path, kind="semantic", allow_partial=True)[
"partial"
] is True
# A complete re-extraction overwrites the same content-hash key cleanly.
save_semantic_cache([{"id": "doc", "source_file": "doc.md"}], [], root=tmp_path)
assert load_cached(doc, root=tmp_path, kind="semantic") is not None
def test_mark_semantic_cache_partial_marks_the_flat_entry_it_would_serve(tmp_path):
"""A pre-fingerprint flat entry must be marked, not shadowed by a new one."""
from graphify.cache import (
cache_dir,
file_hash,
load_cached,
mark_semantic_cache_partial,
)
_reset_stat_index()
doc = tmp_path / "doc.md"
doc.write_text("# Doc\n")
flat = cache_dir(tmp_path, "semantic") / f"{file_hash(doc, tmp_path)}.json"
flat.write_text(
json.dumps({"nodes": [{"id": "doc", "source_file": "doc.md"}]}),
encoding="utf-8",
)
prompt = "a prompt"
# The fingerprinted namespace misses, so the read path falls back to flat.
assert load_cached(doc, root=tmp_path, kind="semantic", prompt=prompt) is not None
marked = mark_semantic_cache_partial(["doc.md"], root=tmp_path, prompt=prompt)
assert marked == [flat]
assert load_cached(doc, root=tmp_path, kind="semantic", prompt=prompt) is None
def test_mark_semantic_cache_partial_skips_missing_and_already_partial(tmp_path):
from graphify.cache import load_cached, mark_semantic_cache_partial, save_semantic_cache
_reset_stat_index()
doc = tmp_path / "doc.md"
doc.write_text("# Doc\n")
save_semantic_cache(
[{"id": "doc", "source_file": "doc.md"}],
[],
root=tmp_path,
partial_source_files=[doc],
)
# Already partial -> nothing new to mark; a missing source is skipped.
assert mark_semantic_cache_partial(["doc.md"], root=tmp_path) == []
assert mark_semantic_cache_partial(["absent.md"], root=tmp_path) == []
assert load_cached(doc, root=tmp_path, kind="semantic") is None