diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl index 7d10a8f129..b5b5743dc3 100644 --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -1,3 +1,4 @@ +{"_type":"issue","id":"polylogue-jdesf","title":"verification: enforce hermetic test-path boundaries","description":"Complete the hermeticity half of the production-reachability oracle. Tests that certify production behavior must not read ambient user or live archive paths unless they declare and enter an explicit production-safe fixture boundary.","design":"Add a structured fixture-boundary declaration and verifier in devtools, integrate it with the production reachability seam checks, and cover ambient ~/.codex, ~/.claude, configured archive-root, and explicit temporary fixture paths. Keep tests deterministic and fail closed on undeclared escapes.","acceptance_criteria":"1. A production-reachability seam can declare its fixture boundary and the verifier rejects undeclared reads of ambient user/session/archive paths. 2. Explicit workspace_env or temporary fixture paths remain allowed and are checked against the declared boundary. 3. Mutation coverage proves removing the guard or widening the boundary makes the test fail. 4. Focused tests and devtools verify --quick pass. 5. No live archive or production mutation is used.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-10T07:49:21Z","created_by":"Sinity","updated_at":"2026-08-10T07:49:21Z","labels":["area:devtools","area:testing","horizon:frontier","lane:reindex"],"dependencies":[{"issue_id":"polylogue-jdesf","depends_on_id":"polylogue-4v2d3","type":"discovered-from","created_at":"2026-08-10T07:49:21Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-1mfxh","title":"raw-authority: persist paginated artifact census receipts","description":"Add a durable source-tier raw-authority artifact census and receipt route. It must use the canonical byte-duplicate authority planner, exclude parser-failed rows from the authoritative candidate universe, support resumable bounded pages, validate backup evidence before checkpointing, and persist apply receipts in source.db.","acceptance_criteria":"1. Census candidates are derived through the canonical duplicate-supersession planner and include only the declared accepted parser universe; parser-failed rows and alternate authority logic cannot enter the applied population. 2. Bounded apply supports an exclusive continuation cursor and durable receipt/checkpoint so successive pages cannot repeat or skip rows. 3. Backup manifest and source ownership are validated before checkpoint or mutation, and any invalid or changed evidence refuses without mutation. 4. Each apply page persists an immutable source-tier receipt bound to census, cursor, plan digest, before/after inventory, and command identity. 5. Real temporary-archive tests exercise two pages, stale/invalid backup refusal, parser-failure exclusion, receipt persistence, and red mutations; quick verification passes. 6. No live production apply is claimed by implementation closure.","status":"closed","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-10T00:51:50Z","created_by":"Sinity","updated_at":"2026-08-10T07:17:24Z","closed_at":"2026-08-10T07:17:24Z","close_reason":"Satisfied by merged PR #3911 (9f8a0a4e2). Paginated raw-authority census receipts use the canonical planner, parser-failure exclusion, resumable cursor/checkpoint, backup/ownership validation, immutable receipt binding, and red-mutation coverage. Verification: focused census suite and quick gate passed on exact head e2e80dfb; no live production apply claimed.","dependencies":[{"issue_id":"polylogue-1mfxh","depends_on_id":"polylogue-fbkr","type":"discovered-from","created_at":"2026-08-10T00:52:00Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-mupq0","title":"reindex: bind message-owner backfill to durable candidate authority","description":"Make message-owner scope backfill a shared, daemon-safe prerequisite for candidate creation and promotion. The route must reject unscoped assertions, bind receipts to current durable state and candidate ownership, and recover safely after a durable commit before receipt publication.","acceptance_criteria":"1. Direct rebuild and daemon bulk-rebuild paths invoke the same message-owner gate before candidate creation, after archive ownership acquisition, and before promotion. 2. The receipt binds exact durable assertion fingerprints, candidate generation/ownership, and source/index authority; missing, drifted, duplicated, or mismatched state refuses without candidate mutation. 3. A prepared marker supports restart-safe completion after commit-before-receipt interruption and rejects altered post-commit state. 4. Real SQLite route tests and mutation twins cover bypass, durable drift, candidate-owner mismatch, and recovery. 5. Implementation closure does not claim a live candidate or promotion; those remain under the reindex phase receipts.","status":"closed","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-10T00:43:21Z","created_by":"Sinity","updated_at":"2026-08-10T07:17:24Z","closed_at":"2026-08-10T07:17:24Z","close_reason":"Satisfied by merged PR #3909 (f33e859f9). Direct and daemon rebuild routes share the owner gate; durable assertion/source/index/candidate-owner bindings and prepared-marker recovery are implemented. Verification: 54 focused route tests passed on exact rebased head 68d54b736; merge-gate passed; implementation closure does not claim live candidate or promotion.","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"polylogue-d96ta","title":"testmon: run and publish a bounded fresh seed receipt","description":"Execute a fresh seed-testmon run after the concurrency cap, retain the complete selection denominator and resource receipt, and decide whether the result is a complete red baseline, release-eligible baseline, or typed incomplete/resource-timeout outcome.","acceptance_criteria":"1. The fresh seed runs on a clean selected code tree and records the complete expected node universe, selection digest, harness and dependency identity, and actual worker/resource evidence. 2. The run reaches a typed terminal outcome or records an explicit timeout/incomplete receipt; no partial selection can be promoted. 3. The resulting receipt is independently replayable and accepted only by the typed testmon promotion gate. 4. The exact command, resource envelope, result, and residual failure attribution are published for the release ledger.","status":"open","priority":0,"issue_type":"task","owner":"ezo.dev@gmail.com","created_at":"2026-08-10T00:40:19Z","created_by":"Sinity","updated_at":"2026-08-10T00:40:19Z","dependencies":[{"issue_id":"polylogue-d96ta","depends_on_id":"polylogue-817er","type":"discovered-from","created_at":"2026-08-10T00:40:29Z","created_by":"Sinity","metadata":"{}"}],"dependency_count":0,"dependent_count":0,"comment_count":0} diff --git a/devtools/production_reachability.py b/devtools/production_reachability.py new file mode 100644 index 0000000000..8dcbc4e751 --- /dev/null +++ b/devtools/production_reachability.py @@ -0,0 +1,388 @@ +"""Structured production-route reachability checks for proof tests. + +The check is deliberately based on Python's import and call structure. A test +may name the exact production symbol it exercises, but that symbol must also +be reachable from the declared production entrypoint. This prevents a test +that directly calls an orphaned helper from certifying a production route. +""" + +from __future__ import annotations + +import ast +import json +from collections import deque +from collections.abc import Iterable +from dataclasses import dataclass +from functools import lru_cache +from pathlib import Path + + +@dataclass(frozen=True, slots=True) +class ProductionSeamSpec: + """Machine-readable contract between one test and a production route.""" + + test_path: str + test_function: str + production_entrypoint: str + tested_symbols: tuple[str, ...] + required_symbols: tuple[str, ...] = () + production_namespace: str = "polylogue" + + def to_dict(self) -> dict[str, object]: + return { + "test_path": self.test_path, + "test_function": self.test_function, + "production_entrypoint": self.production_entrypoint, + "tested_symbols": list(self.tested_symbols), + "required_symbols": list(self.required_symbols), + "production_namespace": self.production_namespace, + } + + +@dataclass(frozen=True, slots=True) +class ProductionReachabilityViolation: + """One structured failure from a production seam check.""" + + code: str + symbol: str + + def to_dict(self) -> dict[str, str]: + return {"code": self.code, "symbol": self.symbol} + + +@dataclass(frozen=True, slots=True) +class ProductionReachabilityReport: + """Machine-readable result for one :class:`ProductionSeamSpec`.""" + + spec: ProductionSeamSpec + violations: tuple[ProductionReachabilityViolation, ...] + + @property + def ok(self) -> bool: + return not self.violations + + def to_dict(self) -> dict[str, object]: + return { + "spec": self.spec.to_dict(), + "ok": self.ok, + "violations": [violation.to_dict() for violation in self.violations], + } + + def to_json(self) -> str: + return json.dumps(self.to_dict(), sort_keys=True) + + +@dataclass(frozen=True, slots=True) +class _FunctionNode: + qualified_name: str + node: ast.FunctionDef | ast.AsyncFunctionDef + module: str + + +@dataclass(frozen=True, slots=True) +class _ParsedModule: + name: str + tree: ast.Module + path: Path + + +class _CallGraph: + def __init__(self, modules: Iterable[_ParsedModule]) -> None: + self.nodes: dict[str, _FunctionNode] = {} + self.edges: dict[str, frozenset[str]] = {} + self.add_modules(modules) + + def add_modules(self, modules: Iterable[_ParsedModule]) -> None: + parsed = tuple(modules) + for module in parsed: + self._index_functions(module) + for module in parsed: + self._index_edges(module) + + def _index_functions(self, module: _ParsedModule) -> None: + for statement in module.tree.body: + if isinstance(statement, (ast.FunctionDef, ast.AsyncFunctionDef)): + qualified_name = f"{module.name}.{statement.name}" + self.nodes[qualified_name] = _FunctionNode(qualified_name, statement, module.name) + elif isinstance(statement, ast.ClassDef): + for member in statement.body: + if isinstance(member, (ast.FunctionDef, ast.AsyncFunctionDef)): + qualified_name = f"{module.name}.{statement.name}.{member.name}" + self.nodes[qualified_name] = _FunctionNode(qualified_name, member, module.name) + + def _index_edges(self, module: _ParsedModule) -> None: + module_imports = _imports_from_nodes( + module.tree.body, module.name, is_package=module.path.name == "__init__.py" + ) + local_functions = { + name: f"{module.name}.{name}" + for name in ( + node.name for node in module.tree.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + ) + } + local_classes = { + node.name: f"{module.name}.{node.name}" for node in module.tree.body if isinstance(node, ast.ClassDef) + } + for function in tuple(node for node in self.nodes.values() if node.module == module.name): + bindings = { + **module_imports, + **local_functions, + **local_classes, + **_imports_from_nodes(function.node.body, module.name, is_package=module.path.name == "__init__.py"), + } + shadowed = _shadowed_names(function.node) + for name in shadowed: + bindings.pop(name, None) + qualified_parts = function.qualified_name.split(".") + if len(qualified_parts) >= 3: + bindings["self"] = ".".join(qualified_parts[:-1]) + targets: set[str] = set() + for call in _calls_in_function(function.node): + target = _resolve_call_target(call.func, bindings, self.nodes) + if target is not None: + targets.add(target) + self.edges[function.qualified_name] = frozenset(targets) + + def reachable_from(self, root: str) -> frozenset[str]: + seen: set[str] = set() + pending = deque([root]) + while pending: + current = pending.popleft() + if current in seen: + continue + seen.add(current) + pending.extend(self.edges.get(current, ())) + return frozenset(seen) + + +def _module_name(path: Path, source_root: Path) -> str: + relative = path.relative_to(source_root) + relative = relative.parent if relative.name == "__init__.py" else relative.with_suffix("") + return ".".join(relative.parts) + + +def _source_files(roots: Iterable[Path]) -> tuple[Path, ...]: + paths: set[Path] = set() + for root in roots: + if root.is_file() and root.suffix == ".py": + paths.add(root) + elif root.is_dir(): + paths.update(path for path in root.rglob("*.py") if path.is_file()) + return tuple(sorted(paths)) + + +def _parse_modules(source_root: Path, roots: Iterable[Path]) -> tuple[_ParsedModule, ...]: + modules: list[_ParsedModule] = [] + for path in _source_files(roots): + try: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + except SyntaxError as exc: + raise ValueError(f"cannot parse {path}: {exc}") from exc + modules.append(_ParsedModule(_module_name(path, source_root), tree, path)) + return tuple(modules) + + +def _resolve_relative_module(module: str, imported: str | None, level: int, *, is_package: bool = False) -> str: + if level == 0: + return imported or "" + parts = module.split(".") + base = parts[: len(parts) - level + 1] if is_package else parts[:-level] + return ".".join((*base, *(imported.split(".") if imported else ()))) + + +def _imports_from_nodes(nodes: Iterable[ast.AST], module: str, *, is_package: bool = False) -> dict[str, str]: + bindings: dict[str, str] = {} + for node in nodes: + if isinstance(node, ast.Import): + for alias in node.names: + bindings[alias.asname or alias.name.split(".")[0]] = ( + alias.name if alias.asname else alias.name.split(".")[0] + ) + elif isinstance(node, ast.ImportFrom): + imported_module = _resolve_relative_module(module, node.module, node.level, is_package=is_package) + for alias in node.names: + if alias.name == "*": + continue + local_name = alias.asname or alias.name + bindings[local_name] = f"{imported_module}.{alias.name}" + return bindings + + +def _attribute_parts(node: ast.AST) -> tuple[str, ...] | None: + if isinstance(node, ast.Name): + return (node.id,) + if isinstance(node, ast.Attribute): + parent = _attribute_parts(node.value) + return None if parent is None else (*parent, node.attr) + return None + + +def _resolve_call_target(function: ast.AST, bindings: dict[str, str], nodes: dict[str, _FunctionNode]) -> str | None: + if isinstance(function, ast.Attribute) and isinstance(function.value, ast.Call): + constructor = _resolve_call_target(function.value.func, bindings, nodes) + if constructor is not None: + candidate = f"{constructor}.{function.attr}" + if candidate in nodes: + return candidate + parts = _attribute_parts(function) + if parts is None or not parts: + return None + bound = bindings.get(parts[0]) + if bound is None: + return None + target = ".".join((bound, *parts[1:])) + if target in nodes: + return target + if bound in nodes or any(name.startswith(f"{bound}.") for name in nodes): + return bound + return None + + +def _test_function_name(spec: ProductionSeamSpec, source_root: Path) -> str: + test_path = Path(spec.test_path) + absolute_path = test_path if test_path.is_absolute() else source_root / test_path + return f"{_module_name(absolute_path, source_root)}.{spec.test_function}" + + +def _shadowed_names(function: ast.FunctionDef | ast.AsyncFunctionDef) -> set[str]: + names = {argument.arg for argument in (*function.args.posonlyargs, *function.args.args, *function.args.kwonlyargs)} + if function.args.vararg is not None: + names.add(function.args.vararg.arg) + if function.args.kwarg is not None: + names.add(function.args.kwarg.arg) + + class ShadowScanner(ast.NodeVisitor): + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + if node is function: + self.generic_visit(node) + else: + names.add(node.name) + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + if node is function: + self.generic_visit(node) + else: + names.add(node.name) + + def visit_Lambda(self, node: ast.Lambda) -> None: + return + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + names.add(node.name) + + def visit_Name(self, node: ast.Name) -> None: + if isinstance(node.ctx, ast.Store): + names.add(node.id) + + scanner = ShadowScanner() + scanner.visit(function) + return names + + +class _CallScanner(ast.NodeVisitor): + def __init__(self, root: ast.FunctionDef | ast.AsyncFunctionDef) -> None: + self.root = root + self.calls: list[ast.Call] = [] + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + if node is self.root: + # A function's decorators, annotations, defaults, and type + # parameters execute in the defining scope, not when the + # function body runs. The reachability contract describes the + # production route executed by the callable, so scan only its + # statements. Nested callable bodies are intentionally skipped. + for statement in node.body: + self.visit(statement) + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + if node is self.root: + for statement in node.body: + self.visit(statement) + + def visit_Lambda(self, node: ast.Lambda) -> None: + return + + def visit_ClassDef(self, node: ast.ClassDef) -> None: + return + + def visit_Call(self, node: ast.Call) -> None: + self.calls.append(node) + self.generic_visit(node) + + +def _calls_in_function(function: ast.FunctionDef | ast.AsyncFunctionDef) -> tuple[ast.Call, ...]: + scanner = _CallScanner(function) + scanner.visit(function) + return tuple(scanner.calls) + + +@lru_cache(maxsize=8) +def _production_modules(source_root: Path, signature: tuple[tuple[str, int, int], ...]) -> tuple[_ParsedModule, ...]: + del signature + production_root = source_root / "polylogue" + return _parse_modules(source_root, (production_root if production_root.is_dir() else source_root,)) + + +def _source_signature(root: Path) -> tuple[tuple[str, int, int], ...]: + return tuple((str(path), path.stat().st_mtime_ns, path.stat().st_size) for path in _source_files((root,))) + + +def _call_graph(source_root: Path, test_path: Path) -> _CallGraph: + production_root = source_root / "polylogue" + production_modules = _production_modules(source_root, _source_signature(production_root)) + production_graph = _CallGraph(production_modules) + graph = _CallGraph(()) + graph.nodes = dict(production_graph.nodes) + graph.edges = dict(production_graph.edges) + graph.add_modules(_parse_modules(source_root, (test_path,))) + return graph + + +def check_production_seam(spec: ProductionSeamSpec, *, source_root: Path) -> ProductionReachabilityReport: + """Check a test's direct call and its production entrypoint's call graph.""" + + test_path = Path(spec.test_path) + absolute_test_path = test_path if test_path.is_absolute() else source_root / test_path + graph = _call_graph(source_root.resolve(), absolute_test_path.resolve()) + violations: list[ProductionReachabilityViolation] = [] + entrypoint = spec.production_entrypoint + test_function = _test_function_name(spec, source_root) + if not entrypoint.startswith(f"{spec.production_namespace}."): + violations.append(ProductionReachabilityViolation("entrypoint_outside_production", entrypoint)) + if entrypoint not in graph.nodes: + violations.append(ProductionReachabilityViolation("missing_production_entrypoint", entrypoint)) + if test_function not in graph.nodes: + violations.append(ProductionReachabilityViolation("missing_test_function", test_function)) + + test_targets = graph.edges.get(test_function, frozenset()) + reachable = graph.reachable_from(entrypoint) + for symbol in spec.tested_symbols: + if symbol not in graph.nodes: + violations.append(ProductionReachabilityViolation("missing_tested_symbol", symbol)) + elif symbol not in test_targets: + violations.append(ProductionReachabilityViolation("test_symbol_not_called", symbol)) + elif symbol not in reachable: + violations.append(ProductionReachabilityViolation("tested_symbol_unreachable", symbol)) + for symbol in spec.required_symbols: + if symbol not in graph.nodes: + violations.append(ProductionReachabilityViolation("missing_required_symbol", symbol)) + elif symbol not in reachable: + violations.append(ProductionReachabilityViolation("required_symbol_unreachable", symbol)) + return ProductionReachabilityReport(spec, tuple(violations)) + + +def assert_production_seam(spec: ProductionSeamSpec, *, source_root: Path) -> None: + """Raise with the structured report when a seam contract is violated.""" + + report = check_production_seam(spec, source_root=source_root) + if not report.ok: + raise AssertionError(report.to_json()) + + +__all__ = [ + "ProductionReachabilityReport", + "ProductionReachabilityViolation", + "ProductionSeamSpec", + "assert_production_seam", + "check_production_seam", +] diff --git a/tests/fixtures/production_reachability/__init__.py b/tests/fixtures/production_reachability/__init__.py new file mode 100644 index 0000000000..23eed76f11 --- /dev/null +++ b/tests/fixtures/production_reachability/__init__.py @@ -0,0 +1 @@ +"""Fixture package for structured production-route reachability tests.""" diff --git a/tests/fixtures/production_reachability/fixture_test.py b/tests/fixtures/production_reachability/fixture_test.py new file mode 100644 index 0000000000..8bbea0fc4d --- /dev/null +++ b/tests/fixtures/production_reachability/fixture_test.py @@ -0,0 +1,46 @@ +"""Test-shaped fixture nodes for the reachability oracle tests.""" + +from __future__ import annotations + +from .nestedpkg import package_route +from .routes import ( + class_route, + dead_helper, + production_entrypoint, + route_accepts_helper, + route_with_nested, + route_with_signature_helper, + shadowed_route, +) + + +def test_wired_route() -> None: + assert production_entrypoint() == "live" + + +def test_dead_helper() -> None: + assert dead_helper() == "dead" + + +def test_nested_route() -> None: + assert route_with_nested() == "live" + + +def test_argument_route() -> None: + assert route_accepts_helper(dead_helper) == "live" + + +def test_signature_route() -> None: + assert route_with_signature_helper() == "live" + + +def test_shadowed_route() -> None: + assert shadowed_route() == "shadowed" + + +def test_class_route() -> None: + assert class_route() == "live" + + +def test_package_route() -> None: + assert package_route() == "child" diff --git a/tests/fixtures/production_reachability/nestedpkg/__init__.py b/tests/fixtures/production_reachability/nestedpkg/__init__.py new file mode 100644 index 0000000000..1624b63079 --- /dev/null +++ b/tests/fixtures/production_reachability/nestedpkg/__init__.py @@ -0,0 +1,5 @@ +from .child import child_route + + +def package_route() -> str: + return child_route() diff --git a/tests/fixtures/production_reachability/nestedpkg/child.py b/tests/fixtures/production_reachability/nestedpkg/child.py new file mode 100644 index 0000000000..aa5a139e9c --- /dev/null +++ b/tests/fixtures/production_reachability/nestedpkg/child.py @@ -0,0 +1,2 @@ +def child_route() -> str: + return "child" diff --git a/tests/fixtures/production_reachability/routes.py b/tests/fixtures/production_reachability/routes.py new file mode 100644 index 0000000000..7eac0fb1e5 --- /dev/null +++ b/tests/fixtures/production_reachability/routes.py @@ -0,0 +1,46 @@ +"""Small import/call graph fixture with one live and one unreachable helper.""" + +from __future__ import annotations + + +def production_entrypoint() -> str: + return live_helper() + + +def live_helper() -> str: + return "live" + + +def dead_helper() -> str: + return "dead" + + +def route_with_nested() -> str: + def nested() -> str: + return dead_helper() + + return live_helper() + + +def route_accepts_helper(helper: object) -> str: + return live_helper() + + +def route_with_signature_helper(default: object = dead_helper()) -> str: + return live_helper() + + +def shadowed_route() -> str: + def live_helper() -> str: + return "shadowed" + + return live_helper() + + +class Runner: + def run(self) -> str: + return live_helper() + + +def class_route() -> str: + return Runner().run() diff --git a/tests/unit/devtools/test_production_reachability.py b/tests/unit/devtools/test_production_reachability.py new file mode 100644 index 0000000000..13d1a8c664 --- /dev/null +++ b/tests/unit/devtools/test_production_reachability.py @@ -0,0 +1,140 @@ +"""Tests for the structured production-route reachability oracle.""" + +from __future__ import annotations + +from pathlib import Path + +from devtools.production_reachability import ProductionSeamSpec, check_production_seam + +_FIXTURE_ROOT = Path(__file__).parents[2] / "fixtures" / "production_reachability" + + +def test_wired_fixture_route_is_reachable() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_wired_route", + production_entrypoint="routes.production_entrypoint", + tested_symbols=("routes.production_entrypoint",), + required_symbols=("routes.live_helper",), + production_namespace="routes", + ), + source_root=_FIXTURE_ROOT, + ) + + assert report.ok, report.to_json() + + +def test_unreachable_tested_symbol_is_a_structured_failure() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_dead_helper", + production_entrypoint="routes.production_entrypoint", + tested_symbols=("routes.dead_helper",), + production_namespace="routes", + ), + source_root=_FIXTURE_ROOT, + ) + + assert not report.ok + assert [violation.to_dict() for violation in report.violations] == [ + {"code": "tested_symbol_unreachable", "symbol": "routes.dead_helper"} + ] + assert report.to_dict()["violations"] == [{"code": "tested_symbol_unreachable", "symbol": "routes.dead_helper"}] + + +def test_nested_callable_body_does_not_create_a_production_edge() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_nested_route", + production_entrypoint="routes.route_with_nested", + tested_symbols=("routes.route_with_nested",), + required_symbols=("routes.dead_helper",), + production_namespace="routes", + ), + source_root=_FIXTURE_ROOT, + ) + + assert [violation.code for violation in report.violations] == ["required_symbol_unreachable"] + + +def test_passed_callable_argument_does_not_create_a_test_edge() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_argument_route", + production_entrypoint="routes.route_accepts_helper", + tested_symbols=("routes.route_accepts_helper",), + required_symbols=("routes.dead_helper",), + production_namespace="routes", + ), + source_root=_FIXTURE_ROOT, + ) + + assert [violation.code for violation in report.violations] == ["required_symbol_unreachable"] + + +def test_signature_calls_do_not_create_a_production_edge() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_signature_route", + production_entrypoint="routes.route_with_signature_helper", + tested_symbols=("routes.route_with_signature_helper",), + required_symbols=("routes.dead_helper",), + production_namespace="routes", + ), + source_root=_FIXTURE_ROOT, + ) + + assert [violation.code for violation in report.violations] == ["required_symbol_unreachable"] + + +def test_shadowed_import_is_not_resolved_as_a_production_edge() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_shadowed_route", + production_entrypoint="routes.shadowed_route", + tested_symbols=("routes.shadowed_route",), + required_symbols=("routes.live_helper",), + production_namespace="routes", + ), + source_root=_FIXTURE_ROOT, + ) + + assert [violation.code for violation in report.violations] == ["required_symbol_unreachable"] + + +def test_class_constructor_and_method_are_reachable() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_class_route", + production_entrypoint="routes.class_route", + tested_symbols=("routes.class_route",), + required_symbols=("routes.Runner.run",), + production_namespace="routes", + ), + source_root=_FIXTURE_ROOT, + ) + + assert report.ok, report.to_json() + + +def test_package_initializer_relative_import_is_resolved() -> None: + report = check_production_seam( + ProductionSeamSpec( + test_path="fixture_test.py", + test_function="test_package_route", + production_entrypoint="nestedpkg.package_route", + tested_symbols=("nestedpkg.package_route",), + required_symbols=("nestedpkg.child.child_route",), + production_namespace="nestedpkg", + ), + source_root=_FIXTURE_ROOT, + ) + + assert report.ok, report.to_json() diff --git a/tests/unit/maintenance/test_rebuild_parse_apply_split.py b/tests/unit/maintenance/test_rebuild_parse_apply_split.py index 4364682e37..c34978f269 100644 --- a/tests/unit/maintenance/test_rebuild_parse_apply_split.py +++ b/tests/unit/maintenance/test_rebuild_parse_apply_split.py @@ -32,6 +32,7 @@ import pytest +from devtools.production_reachability import ProductionSeamSpec, check_production_seam from polylogue.config import Config from polylogue.maintenance.rebuild_index import RebuildIndexRequest, rebuild_index_from_source_sync from polylogue.sources import revision_backfill @@ -39,6 +40,47 @@ from polylogue.sources.revision_backfill import split_parse_and_apply_seconds from tests.infra.revision_backfill_benchmark import build_independent_raw_corpus +REINDEX_PRODUCTION_SEAMS = ( + ProductionSeamSpec( + test_path="tests/unit/maintenance/test_rebuild_parse_apply_split.py", + test_function="test_rebuild_records_parse_apply_split_summing_to_stage_total", + production_entrypoint="polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync", + tested_symbols=("polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync",), + required_symbols=( + "polylogue.sources.revision_backfill.backfill_historical_revision_evidence", + "polylogue.storage.repair.repair_session_insights", + ), + ), + ProductionSeamSpec( + test_path="tests/unit/maintenance/test_rebuild_parse_apply_split.py", + test_function="test_rebuild_index_from_source_sync_warms_prefetch_cache_when_caller_omits_one", + production_entrypoint="polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync", + tested_symbols=("polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync",), + required_symbols=( + "polylogue.sources.revision_backfill.backfill_historical_revision_evidence", + "polylogue.storage.repair.repair_session_insights", + ), + ), + ProductionSeamSpec( + test_path="tests/unit/maintenance/test_rebuild_parse_apply_split.py", + test_function="test_rebuild_index_from_source_sync_auto_engages_pipelined_decode", + production_entrypoint="polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync", + tested_symbols=("polylogue.maintenance.rebuild_index.rebuild_index_from_source_sync",), + required_symbols=( + "polylogue.sources.revision_backfill.backfill_historical_revision_evidence", + "polylogue.storage.repair.repair_session_insights", + ), + ), +) + + +def test_selected_reindex_proof_tests_are_production_reachable() -> None: + """The selected reindex proofs bind to replay and terminal convergence.""" + root = Path(__file__).resolve().parents[3] + for spec in REINDEX_PRODUCTION_SEAMS: + report = check_production_seam(spec, source_root=root) + assert report.ok, report.to_json() + def test_split_parse_and_apply_seconds_sums_to_total() -> None: """Pure rollup: parse is census+spill_load, apply is everything else."""