diff --git a/graph/epistemic.py b/graph/epistemic.py index 4ea2712b..061814bb 100644 --- a/graph/epistemic.py +++ b/graph/epistemic.py @@ -337,8 +337,14 @@ async def count_edges(self, user_id: str | None = None) -> dict[str, int]: Returns `{"total", "live", "dead"}`. `epi_edges` carries no `layer` column, so an edge belongs to the layer - of its SOURCE node — the same convention `memory_graph_edges` already - uses. An edge whose source node is gone is counted by none of the three + of its SOURCE node — that is the convention counted here. It is NOT the + convention `memory_graph_edges` lists by: that reader requires both + endpoints in the requested layer, because it returns the target node's + content and a source-only filter would pull a foreign layer's text into + the answer. The two therefore diverge on a cross-layer edge: counted + here, not listed there. Today no such edge exists; a node re-layering + would create them, and this counter would then exceed what the tool + shows. An edge whose source node is gone is counted by none of the three numbers rather than guessed into one. "Dead" is `weight <= 0`, not `weight < NREM_FLOOR`: measured on all diff --git a/mcp_server/tools/graph.py b/mcp_server/tools/graph.py index 4bf937fc..c7075315 100644 --- a/mcp_server/tools/graph.py +++ b/mcp_server/tools/graph.py @@ -170,6 +170,9 @@ async def memory_graph_edges( sibling node readers. An empty user_id keeps the layer-only scope for internal callers that genuinely want the whole layer; the default is `"default"` so an omitted argument cannot read another user's edges. + The `layer` is scoped on both endpoints too: an edge belongs to its + endpoints' layer, so filtering only the source would pull a foreign + layer's node content through the join. """ limit = _clamp_limit(limit, default=20) app = _get_ctx(ctx) @@ -179,7 +182,8 @@ async def memory_graph_edges( graph = _get_graph(app, layer) dispatch = get_write_dispatch(graph._cm) # F54: the edge is scoped on BOTH endpoints — a join on one side alone - # pulls a foreign node's content into the answer. + # pulls a foreign node's content into the answer. F105: the layer is scoped + # the same way, on the target as well as the source. user_clause = "" user_params: tuple[Any, ...] = () if user_id: @@ -199,9 +203,9 @@ async def memory_graph_edges( FROM epi_edges e JOIN epi_nodes s ON e.source_id = s.node_id JOIN epi_nodes t ON e.target_id = t.node_id - WHERE {where} AND s.layer = ?{user_clause} + WHERE {where} AND s.layer = ? AND t.layer = ?{user_clause} ORDER BY e.weight DESC LIMIT ?""", - (*params, graph.layer, *user_params, limit), + (*params, graph.layer, graph.layer, *user_params, limit), ) else: rows = await dispatch.query( @@ -211,9 +215,9 @@ async def memory_graph_edges( FROM epi_edges e JOIN epi_nodes s ON e.source_id = s.node_id JOIN epi_nodes t ON e.target_id = t.node_id - WHERE s.layer = ?{user_clause} + WHERE s.layer = ? AND t.layer = ?{user_clause} ORDER BY e.weight DESC LIMIT ?""", - (graph.layer, *user_params, limit), + (graph.layer, graph.layer, *user_params, limit), ) edges = [ { diff --git a/scripts/migrate_graph_layers.py b/scripts/migrate_graph_layers.py index 52ee46a6..57131058 100755 --- a/scripts/migrate_graph_layers.py +++ b/scripts/migrate_graph_layers.py @@ -25,10 +25,14 @@ MARKER_VERSION = 1 +EXCLUDE_VERSION = 1 MARKER_SOURCES = {"none", "cli", "file", "cli+file"} +EXCLUDE_SOURCES = {"none", "cli", "file", "cli+file"} REQUIRED_NODE_COLUMNS = {"node_id", "layer", "user_id", "content", "node_type", "tags"} REQUIRED_EDGE_COLUMNS = {"source_id", "target_id", "relation", "tags"} -AGENT_LIKE_NODE_TYPES = {"personality_trait", "action", "outcome", "error_outcome", "wiki_page"} +# wiki_page is deliberately absent: its `content` is a filesystem path, not prose, +# so the node type alone carries no agent-layer text evidence. +AGENT_LIKE_NODE_TYPES = {"personality_trait", "action", "outcome", "error_outcome"} AGENT_LIKE_TAGS = { "learned_from", "decided_because", @@ -154,47 +158,85 @@ def _json_list(value: Any) -> list[str]: return [item for item in decoded if isinstance(item, str)] +def _normalize_patterns(patterns: Sequence[str]) -> list[str]: + return sorted({pattern for pattern in patterns if pattern}) + + def _normalize_markers(markers: Sequence[str]) -> list[str]: - return sorted({marker for marker in markers if marker}) + return _normalize_patterns(markers) -def _marker_policy(markers: Sequence[str], source: str | None) -> dict[str, Any]: - normalized = _normalize_markers(markers) +def _normalize_excludes(excludes: Sequence[str]) -> list[str]: + return _normalize_patterns(excludes) + + +def _pattern_policy( + patterns: Sequence[str], + source: str | None, + *, + version: int, + sources: set[str], + label: str, +) -> dict[str, Any]: + normalized = _normalize_patterns(patterns) resolved = source if source is not None else ("none" if not normalized else "cli") - if resolved not in MARKER_SOURCES: - raise SchemaError(f"unknown marker source: {resolved}") + if resolved not in sources: + raise SchemaError(f"unknown {label} source: {resolved}") digest = hashlib.sha256("\n".join(normalized).encode("utf-8")).hexdigest() - return {"version": MARKER_VERSION, "source": resolved, "count": len(normalized), "sha256": digest} + return {"version": version, "source": resolved, "count": len(normalized), "sha256": digest} + + +def _marker_policy(markers: Sequence[str], source: str | None) -> dict[str, Any]: + return _pattern_policy(markers, source, version=MARKER_VERSION, sources=MARKER_SOURCES, label="marker") + +def _exclude_policy(excludes: Sequence[str], source: str | None) -> dict[str, Any]: + return _pattern_policy(excludes, source, version=EXCLUDE_VERSION, sources=EXCLUDE_SOURCES, label="exclude") -def _read_markers_file(path: Path) -> list[str]: - markers: list[str] = [] + +def _read_patterns_file(path: Path) -> list[str]: + patterns: list[str] = [] for line in path.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#"): continue - markers.append(stripped) - return markers + patterns.append(stripped) + return patterns -def _resolve_markers(markers_file: Path | None, markers: Sequence[str] | None) -> tuple[list[str], str]: - from_file = _read_markers_file(markers_file) if markers_file is not None else [] - from_cli = list(markers) if markers else [] - if markers_file is not None and from_cli: +def _resolve_patterns_file( + patterns_file: Path | None, + patterns: Sequence[str] | None, +) -> tuple[list[str], str]: + from_file = _read_patterns_file(patterns_file) if patterns_file is not None else [] + from_cli = list(patterns) if patterns else [] + if patterns_file is not None and from_cli: source = "cli+file" - elif markers_file is not None: + elif patterns_file is not None: source = "file" elif from_cli: source = "cli" else: source = "none" - return _normalize_markers([*from_file, *from_cli]), source + return _normalize_patterns([*from_file, *from_cli]), source + + +def _resolve_markers(markers_file: Path | None, markers: Sequence[str] | None) -> tuple[list[str], str]: + return _resolve_patterns_file(markers_file, markers) + + +def _resolve_excludes(excludes_file: Path | None, excludes: Sequence[str] | None) -> tuple[list[str], str]: + return _resolve_patterns_file(excludes_file, excludes) def _marker_reasons(content: str, markers: Sequence[str]) -> list[str]: return [f"marker:{marker}" for marker in markers if marker in content] +def _exclude_reasons(content: str, excludes: Sequence[str]) -> list[str]: + return [f"excluded:{pattern}" for pattern in excludes if pattern in content] + + def _node_evidence(node: sqlite3.Row) -> tuple[list[str], bool]: reasons: list[str] = [] node_type = str(node["node_type"]) @@ -273,8 +315,11 @@ def _candidate( conn: sqlite3.Connection, edge_pk: list[str], markers: Sequence[str], + excludes: Sequence[str], ) -> Candidate | None: - marker_reasons = _marker_reasons(str(node["content"]), markers) + content = str(node["content"]) + marker_reasons = _marker_reasons(content, markers) + exclude_reasons = _exclude_reasons(content, excludes) node_reasons, has_node_evidence = _node_evidence(node) edge_ids, edge_impact, edge_reasons = _edge_inventory(conn, int(node["node_id"]), str(node["user_id"]), edge_pk) reasons = marker_reasons + node_reasons + edge_reasons @@ -288,6 +333,7 @@ def _candidate( if not reasons: return None confidence = "high" if categories >= 2 else "medium" if categories == 1 else "low" + decision = "exclude" if exclude_reasons else "review" duplicate_node_ids = _duplicate_node_ids( conn, int(node["node_id"]), @@ -303,8 +349,8 @@ def _candidate( "current_layer": "user", "target_layer": "agent", "confidence": confidence, - "reasons": reasons, - "decision": "review", + "reasons": reasons + exclude_reasons, + "decision": decision, "edge_ids": edge_ids, "edge_impact": edge_impact, "duplicate_node_ids": duplicate_node_ids, @@ -340,6 +386,8 @@ def inventory( *, markers: Sequence[str] | None = None, marker_source: str | None = None, + excludes: Sequence[str] | None = None, + exclude_source: str | None = None, ) -> dict[str, Any]: """Build a deterministic, read-only inventory for one memory base. @@ -351,8 +399,11 @@ def inventory( candidate itself is excluded. ``target_collision_node_ids`` identifies existing agent rows with the same user, node type, and content. Edge evidence includes the opposite user and only same-user evidence affects - confidence. All reads and the fingerprint use one temporary SQLite online - backup snapshot, which is removed before return. + confidence. A candidate matching an exclude pattern is still emitted but + carries ``decision="exclude"`` and an ``excluded:`` reason; the + exclude words never appear in ``exclude_policy``. All reads and the + fingerprint use one temporary SQLite online backup snapshot, which is + removed before return. """ db_path = _database_path(data_dir).resolve() if not db_path.is_file(): @@ -360,6 +411,8 @@ def inventory( normalized_markers = _normalize_markers(list(markers) if markers else []) policy = _marker_policy(normalized_markers, marker_source) + normalized_excludes = _normalize_excludes(list(excludes) if excludes else []) + exclude_policy = _exclude_policy(normalized_excludes, exclude_source) label = _database_label(data_dir, database) encoded_db_path = quote(str(db_path), safe="/") with closing(sqlite3.connect(f"file:{encoded_db_path}?mode=ro", uri=True)) as source_conn: @@ -374,13 +427,22 @@ def inventory( rows = conn.execute( "SELECT node_id, layer, user_id, content, node_type, tags FROM epi_nodes WHERE layer = 'user' ORDER BY node_id" ).fetchall() - candidates = [candidate for row in rows if (candidate := _candidate(row, conn, edge_pk, normalized_markers)) is not None] + candidates = [ + candidate for row in rows if (candidate := _candidate(row, conn, edge_pk, normalized_markers, normalized_excludes)) is not None + ] layer_rows = conn.execute("SELECT layer, COUNT(*) AS row_count FROM epi_nodes GROUP BY layer ORDER BY layer").fetchall() user_rows = conn.execute("SELECT user_id, COUNT(*) AS row_count FROM epi_nodes GROUP BY user_id ORDER BY user_id").fetchall() fingerprint = _fingerprint(snapshot_path, source_metadata) + excluded_count = sum(candidate["decision"] == "exclude" for candidate in candidates) + # The confidence buckets describe every emitted candidate, excluded ones + # included — they are a property of the evidence, not of the decision. + # `reviewable` is the number a reviewer actually acts on, so a consumer never + # has to know to subtract `excluded` itself. summary = { "candidates": len(candidates), + "excluded": excluded_count, + "reviewable": len(candidates) - excluded_count, "high": sum(candidate["confidence"] == "high" for candidate in candidates), "medium": sum(candidate["confidence"] == "medium" for candidate in candidates), "low": sum(candidate["confidence"] == "low" for candidate in candidates), @@ -393,6 +455,7 @@ def inventory( "source_layer": "user", "target_layer": "agent", "marker_policy": policy, + "exclude_policy": exclude_policy, "edge_id_field": edge_pk, "db_fingerprint": fingerprint, "candidates": candidates, @@ -407,10 +470,19 @@ def write_inventory( *, markers: Sequence[str] | None = None, marker_source: str | None = None, + excludes: Sequence[str] | None = None, + exclude_source: str | None = None, ) -> None: """Write a manifest only after the complete read-only inventory succeeds.""" _, output_path = _resolve_output_paths(data_dir, output) - manifest = inventory(data_dir, database, markers=markers, marker_source=marker_source) + manifest = inventory( + data_dir, + database, + markers=markers, + marker_source=marker_source, + excludes=excludes, + exclude_source=exclude_source, + ) temp_path: Path | None = None try: with tempfile.NamedTemporaryFile( @@ -441,6 +513,8 @@ def _parser() -> argparse.ArgumentParser: inventory_parser.add_argument("--database", type=str) inventory_parser.add_argument("--markers-file", type=Path) inventory_parser.add_argument("--marker", action="append", dest="markers", metavar="TEXT") + inventory_parser.add_argument("--exclude-file", type=Path) + inventory_parser.add_argument("--exclude", action="append", dest="excludes", metavar="TEXT") return parser @@ -450,7 +524,16 @@ def main(argv: list[str] | None = None) -> int: raise AssertionError(f"unsupported command: {args.command}") try: markers, marker_source = _resolve_markers(args.markers_file, args.markers) - write_inventory(args.data_dir, args.output, args.database, markers=markers, marker_source=marker_source) + excludes, exclude_source = _resolve_excludes(args.exclude_file, args.excludes) + write_inventory( + args.data_dir, + args.output, + args.database, + markers=markers, + marker_source=marker_source, + excludes=excludes, + exclude_source=exclude_source, + ) except (FileNotFoundError, OSError, SchemaError, sqlite3.Error) as exc: print(str(exc), file=sys.stderr) return 2 diff --git a/tests/test_mcp/test_graph_edges_layer_scope.py b/tests/test_mcp/test_graph_edges_layer_scope.py new file mode 100644 index 00000000..8f057531 --- /dev/null +++ b/tests/test_mcp/test_graph_edges_layer_scope.py @@ -0,0 +1,147 @@ +"""F105 — ``memory_graph_edges`` scopes the layer on BOTH endpoints. + +The user half of this asymmetry was closed by F54 (``AND s.user_id = ? AND +t.user_id = ?``); the layer half was missed: only the source node was filtered, +so an edge leaving the requested layer returned the foreign layer's target +content. Each test names the mutation it kills. +""" + +import time +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from shared.connection import connection_manager +from shared.migrations import MigrationManager + + +def _ctx(app: object) -> object: + ctx = MagicMock() + ctx.request_context = MagicMock() + ctx.request_context.lifespan_context = app + return ctx + + +@pytest.fixture +async def layer_db(tmp_path, monkeypatch): + """A migrated temp database wired into the global manager, no live bases.""" + monkeypatch.setattr(connection_manager, "base_dir", tmp_path) + connection_manager._conns.clear() + await MigrationManager(cm=connection_manager).migrate() + yield connection_manager + connection_manager._conns.clear() + + +def _app(cm: object) -> SimpleNamespace: + from core import MemoryManager + from graph.epistemic import EpistemicGraph + + return SimpleNamespace(mm=MemoryManager(cm=cm), user_graph=EpistemicGraph(layer="user", cm=cm)) + + +async def _node(cm: object, layer: str, user_id: str, content: str) -> int: + conn = await cm.get("memory.db") + cur = await conn.execute( + "INSERT INTO epi_nodes (layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, 'fact', '[]', 0.5, ?)", + (layer, user_id, content, time.time()), + ) + await conn.commit() + return int(cur.lastrowid) + + +async def _edge(cm: object, src: int, tgt: int, relation: str) -> None: + conn = await cm.get("memory.db") + await conn.execute( + "INSERT INTO epi_edges (source_id, target_id, relation, weight, created_at) VALUES (?, ?, ?, 0.8, ?)", + (src, tgt, relation, time.time()), + ) + await conn.commit() + + +async def _seed_cross_layer(cm: object) -> dict[str, int]: + """user A/B in layer ``user``, agent C in layer ``agent`` (all user ``default``).""" + a = await _node(cm, "user", "default", "user-a") + b = await _node(cm, "user", "default", "user-b") + c = await _node(cm, "agent", "default", "agent-c") + await _edge(cm, a, b, "same_layer") + await _edge(cm, a, c, "user_to_agent") + await _edge(cm, c, a, "agent_to_user") + return {"a": a, "b": b, "c": c} + + +def _relations(out: dict[str, object]) -> set[str]: + return {str(edge["relation"]) for edge in out["edges"]} # type: ignore[index] + + +def _contents(out: dict[str, object]) -> set[str]: + seen: set[str] = set() + for edge in out["edges"]: # type: ignore[index] + seen.add(str(edge["source_content"])) + seen.add(str(edge["target_content"])) + return seen + + +async def test_same_layer_edge_survives(layer_db) -> None: + """Kills over-filtering: the legitimate A→B edge must not disappear.""" + await _seed_cross_layer(layer_db) + + from mcp_server.tools.graph import memory_graph_edges + + out = await memory_graph_edges(layer="user", user_id="default", ctx=_ctx(_app(layer_db))) + + assert "same_layer" in _relations(out), f"same-layer edge lost: {out}" + + +async def test_out_does_not_cross_the_layer(layer_db) -> None: + """Kills the bug: ``s.layer`` only, so A→C leaks agent content on the target.""" + await _seed_cross_layer(layer_db) + + from mcp_server.tools.graph import memory_graph_edges + + out = await memory_graph_edges(layer="user", user_id="default", direction="out", ctx=_ctx(_app(layer_db))) + + assert "user_to_agent" not in _relations(out), f"cross-layer edge returned on out: {out}" + assert "agent-c" not in _contents(out), f"agent content leaked on out: {out}" + + +async def test_in_does_not_cross_the_layer(layer_db) -> None: + """Kills the bug for ``direction="in"``: the requested node is the target, so + a user-layer source reaching an agent-layer target must not return it.""" + ids = await _seed_cross_layer(layer_db) + + from mcp_server.tools.graph import memory_graph_edges + + out = await memory_graph_edges(layer="user", user_id="default", node_id=ids["c"], direction="in", ctx=_ctx(_app(layer_db))) + + assert "user_to_agent" not in _relations(out), f"cross-layer backlink returned on in: {out}" + assert "agent-c" not in _contents(out), f"agent content leaked on in: {out}" + + +async def test_both_does_not_cross_the_layer(layer_db) -> None: + """Kills the bug for ``direction="both"``: either endpoint must stay in-layer.""" + ids = await _seed_cross_layer(layer_db) + + from mcp_server.tools.graph import memory_graph_edges + + out = await memory_graph_edges(layer="user", user_id="default", node_id=ids["a"], direction="both", ctx=_ctx(_app(layer_db))) + + assert "agent-c" not in _contents(out), f"agent content leaked on both: {out}" + assert _relations(out) == {"same_layer"}, f"both must keep only the same-layer edge: {out}" + + +async def test_explicit_user_id_still_scopes_both_endpoints(layer_db) -> None: + """Guard (stays green): F54's user predicate is untouched by the layer fix.""" + a = await _node(layer_db, "user", "default", "default-a") + b = await _node(layer_db, "user", "default", "default-b") + bob = await _node(layer_db, "user", "bob", "bob-x") + await _edge(layer_db, a, b, "default_rel") + await _edge(layer_db, a, bob, "default_to_bob") + await _edge(layer_db, bob, a, "bob_to_default") + + from mcp_server.tools.graph import memory_graph_edges + + out = await memory_graph_edges(layer="user", user_id="default", direction="both", ctx=_ctx(_app(layer_db))) + + assert _relations(out) == {"default_rel"}, f"foreign user's edges leaked: {out}" + assert "bob-x" not in _contents(out), f"foreign user's content leaked: {out}" diff --git a/tests/test_scripts/test_migrate_graph_layers.py b/tests/test_scripts/test_migrate_graph_layers.py index f46a1590..267fc843 100644 --- a/tests/test_scripts/test_migrate_graph_layers.py +++ b/tests/test_scripts/test_migrate_graph_layers.py @@ -81,6 +81,28 @@ def fixture_base(tmp_path: Path) -> Generator[Path, None, None]: yield base +PROFILE_LINE = "name: X. calls me Y. i call them Z." +EXCLUDE_PATTERN = "calls me" + + +@pytest.fixture +def exclusion_base(tmp_path: Path) -> Generator[Path, None, None]: + base = tmp_path / "exclusion" + base.mkdir() + db = base / "memory.db" + with sqlite3.connect(db) as conn: + conn.executescript(NODE_DDL + EDGE_DDL) + conn.executemany( + "INSERT INTO epi_nodes (node_id, layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + [ + (1, "user", "default", f"{PROFILE_LINE} alpha_marker", "fact", "[]", 0.8, 1.0), + (2, "user", "default", f"{PROFILE_LINE} alpha_marker", "fact", "[]", 0.8, 2.0), + (3, "user", "default", "alpha_marker only", "fact", "[]", 0.8, 3.0), + ], + ) + yield base + + def _db_hash(db: Path) -> str: return hashlib.sha256(db.read_bytes()).hexdigest() @@ -133,6 +155,8 @@ def test_inventory_manifest_has_fingerprint_and_candidate_fields(fixture_base: P assert manifest["db_fingerprint"]["source"]["filename"] == "memory.db" assert set(manifest["summary"]) == { "candidates", + "excluded", + "reviewable", "high", "medium", "low", @@ -656,3 +680,136 @@ def test_output_cannot_be_a_database_sidecar(fixture_base: Path, suffix: str) -> assert sidecar.read_bytes() == original, "the sidecar was overwritten" assert _db_hash(db_path) == before + + +def test_excluded_candidate_is_emitted_and_marked_exclude(exclusion_base: Path) -> None: + """Kills: dropping excluded rows instead of emitting them, or forcing decision=review.""" + manifest = migration.inventory(exclusion_base, markers=["alpha_marker"], excludes=[EXCLUDE_PATTERN]) + excluded = next(item for item in manifest["candidates"] if item["node_id"] == 1) + review = next(item for item in manifest["candidates"] if item["node_id"] == 3) + baseline = next(item for item in migration.inventory(exclusion_base, markers=["alpha_marker"])["candidates"] if item["node_id"] == 1) + + assert excluded["decision"] == "exclude" + assert f"excluded:{EXCLUDE_PATTERN}" in excluded["reasons"] + assert review["decision"] == "review" + assert excluded["confidence"] == baseline["confidence"] == "medium" + + +def test_exclude_policy_is_metadata_only(exclusion_base: Path) -> None: + """Kills: leaking exclude words into the policy, or deviating from the marker policy shape. + + The pattern here matches a real node on purpose. An earlier version used a + pattern that matched nothing, so "the word is not in the manifest" was true + no matter what the code did. The honest contract is narrower: the *policy* + carries only metadata, while a matching candidate's `reasons` legitimately + echo the pattern as `excluded:` — that is the evidence a reviewer + reads, so this test asserts that echo is present rather than absent. + """ + pattern = EXCLUDE_PATTERN + manifest = migration.inventory(exclusion_base, markers=["alpha_marker"], excludes=[pattern]) + policy = manifest["exclude_policy"] + + assert set(policy) == {"version", "source", "count", "sha256"} + assert policy["source"] == "cli" + assert policy["count"] == 1 + assert policy["sha256"] == hashlib.sha256(pattern.encode("utf-8")).hexdigest() + assert pattern not in json.dumps(policy, ensure_ascii=False) + + matched = [c for c in manifest["candidates"] if f"excluded:{pattern}" in c["reasons"]] + assert matched, "the fixture must produce at least one excluded candidate" + assert all(c["decision"] == "exclude" for c in matched) + assert manifest["summary"]["excluded"] == len(matched) + assert manifest["summary"]["reviewable"] == len(manifest["candidates"]) - len(matched) + + +def test_exclude_file_and_option_combine_dedupe_and_sort(exclusion_base: Path) -> None: + """Kills: skipping file/CLI merge, dedupe, sort, or blank/comment filtering.""" + output = exclusion_base / "manifest.json" + exclude_file = exclusion_base / "excludes.txt" + exclude_file.write_text("# exclusions\n\n calls me \nother_pattern\ncalls me\n", encoding="utf-8") + + result = _run_inventory_cli( + exclusion_base, + output, + "--exclude-file", + str(exclude_file), + "--exclude", + "zzz_pattern", + "--exclude", + EXCLUDE_PATTERN, + ) + + assert result.returncode == 0, result.stderr + policy = json.loads(output.read_text(encoding="utf-8"))["exclude_policy"] + assert policy["source"] == "cli+file" + assert policy["count"] == 3 + assert policy["sha256"] == _expected_marker_sha(["calls me", "other_pattern", "zzz_pattern"]) + + +def test_wiki_page_path_content_is_not_a_candidate(tmp_path: Path) -> None: + """Kills: treating wiki_page as agent prose when its content is a filesystem path.""" + base = tmp_path / "wiki" + base.mkdir() + db = base / "memory.db" + with sqlite3.connect(db) as conn: + conn.executescript(NODE_DDL + EDGE_DDL) + conn.execute( + "INSERT INTO epi_nodes (node_id, layer, user_id, content, node_type, tags, confidence, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (1, "user", "default", "/home/user/wiki/user/diary/MOC_diary.md", "wiki_page", "[]", 0.5, 1.0), + ) + + manifest = _inventory(base) + + assert manifest["candidates"] == [] + + +def test_summary_excluded_counts_excluded_candidates(exclusion_base: Path) -> None: + """Kills: reporting candidate totals without a separate excluded count.""" + manifest = migration.inventory(exclusion_base, markers=["alpha_marker"], excludes=[EXCLUDE_PATTERN]) + + assert manifest["summary"]["excluded"] == 2 + assert manifest["summary"]["candidates"] == 3 + assert sum(item["decision"] == "exclude" for item in manifest["candidates"]) == 2 + + +def test_exclusion_keeps_inventory_read_only_deterministic_and_apply_rejected(exclusion_base: Path) -> None: + """Kills: exclusion bypassing the snapshot/read-only guarantees or enabling apply.""" + db = exclusion_base / "memory.db" + before_hash = _db_hash(db) + + first = migration.inventory(exclusion_base, markers=["alpha_marker"], excludes=[EXCLUDE_PATTERN]) + second = migration.inventory(exclusion_base, markers=["alpha_marker"], excludes=[EXCLUDE_PATTERN]) + + assert first == second + assert json.dumps(first, ensure_ascii=False, sort_keys=True) == json.dumps(second, ensure_ascii=False, sort_keys=True) + assert _db_hash(db) == before_hash + + output = exclusion_base / "manifest.json" + result = _run_inventory_cli( + exclusion_base, + output, + "--marker", + "alpha_marker", + "--exclude", + EXCLUDE_PATTERN, + ) + assert result.returncode == 0, result.stderr + assert _db_hash(db) == before_hash + + apply_result = subprocess.run( + [ + sys.executable, + str(REPO_ROOT / "scripts/migrate_graph_layers.py"), + "apply", + "--data-dir", + str(exclusion_base), + "--output", + str(output), + ], + cwd=REPO_ROOT, + capture_output=True, + text=True, + check=False, + ) + assert apply_result.returncode != 0 + assert _db_hash(db) == before_hash