"""Deterministic orphan sweep for the `transcription` package. `.github/skills/python-code-reviewer/skill.md` requires every review to report orphaned code as removed, retained-with-justification, or uncertain-follow-up. This guard makes that sweep reproducible: it locks the current set of unreferenced public definitions, so a newly stranded function fails the build instead of accumulating silently, and deleting a known orphan requires deleting its entry here. The sweep is intentionally conservative. It considers module-level public definitions and public methods on public classes, and honours the dynamic-wiring exceptions the skill calls out: framework route registration, string-based entrypoint references, and use from `tests/` or `tools/`. """ from __future__ import annotations import ast from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[1] SOURCE_DIR = PROJECT_ROOT / "src" / "transcription" # Every tree that may legitimately consume package API. REFERENCE_ROOTS = (SOURCE_DIR, PROJECT_ROOT / "tests", PROJECT_ROOT / "tools") # Decorators that hand a callable to a framework registry, making the definition # reachable without any in-repo reference to its name. REGISTRATION_DECORATOR_PREFIXES = ("router.", "app.", "ui.page") # Confirmed orphans, retained by decision rather than by reference. Each entry needs a # rationale. Removing the code means removing the entry; adding an entry means an # explicit decision to keep unreferenced code. KNOWN_ORPHANS: dict[str, str] = { "DocumentService.is_document_type_referenced": ( "Read helper currently bypassed by callers in favor of direct delete guards. " "Retained for now to preserve service API stability during cleanup." ), "DocumentService.is_tag_referenced": ( "Read helper currently bypassed by callers in favor of direct delete guards. " "Retained for now to preserve service API stability during cleanup." ), "DocumentService.read_document": ( "Public CRUD read method currently unused by runtime routes, but retained as " "part of the service API shape for downstream call sites." ), "DocumentService.read_document_type": ( "Public CRUD read method currently unused by runtime routes, but retained as " "part of the service API shape for downstream call sites." ), "DocumentService.read_tag": ( "Public CRUD read method currently unused by runtime routes, but retained as " "part of the service API shape for downstream call sites." ), "JSONBCompat.load_dialect_impl": ( "SQLAlchemy type hook is invoked by ORM internals via subclass protocol rather " "than direct in-repo references; keep this implementation method." ), "JobService.delete_job": ( "Legacy delete method retained as a compatibility shim while guarded deletion " "flows migrate fully to delete_job_with_guardrails." ), "JobService.update_job": ( "Legacy update method retained for compatibility while route and workflow " "callers continue converging on narrower state-transition APIs." ), "PeopleService.is_person_role_referenced": ( "Reference-check helper currently not called by route workflows, but kept with " "the service surface while role-management cleanup remains in progress." ), "PeopleService.read_person": ( "Public CRUD read method currently unused by runtime routes, but retained as " "part of the service API shape for downstream call sites." ), "PeopleService.read_person_role": ( "Public CRUD read method currently unused by runtime routes, but retained as " "part of the service API shape for downstream call sites." ), "Settings.normalize_provider_models": ( "Pydantic model-level validator is executed by framework hooks using decorator " "registration and therefore has no direct symbolic call site." ), "Settings.validate_provider_models_input": ( "Pydantic field validator is executed by framework hooks using decorator " "registration and therefore has no direct symbolic call site." ), "SourceService.read_job_source_for_job": ( "Public read helper currently unused by runtime flows, but retained as part of " "the SourceService API pending endpoint consolidation." ), } def _source_files() -> list[Path]: return sorted(SOURCE_DIR.rglob("*.py")) def _is_registered_with_framework(node: ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef) -> bool: return any(ast.unparse(decorator).startswith(REGISTRATION_DECORATOR_PREFIXES) for decorator in node.decorator_list) def _public_definitions() -> dict[str, str]: """Public definitions mapped to `path:line` (module-level + class methods).""" definitions: dict[str, str] = {} for path in _source_files(): tree = ast.parse(path.read_text(encoding="utf-8")) for node in tree.body: if not isinstance(node, ast.FunctionDef | ast.AsyncFunctionDef | ast.ClassDef): continue if node.name.startswith("_") or _is_registered_with_framework(node): continue definitions[node.name] = f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}" if isinstance(node, ast.ClassDef): for method in node.body: if not isinstance(method, ast.FunctionDef | ast.AsyncFunctionDef): continue if method.name.startswith("_"): continue definitions[f"{node.name}.{method.name}"] = ( f"{path.relative_to(PROJECT_ROOT).as_posix()}:{method.lineno}" ) return definitions def _referenced_names() -> tuple[set[str], str]: """Names referenced anywhere, plus every string literal joined for dotted lookups.""" names: set[str] = set() literals: list[str] = [] for root in REFERENCE_ROOTS: for path in sorted(root.rglob("*.py")): # This module names every known orphan in `KNOWN_ORPHANS`; counting those # strings as references would make the allowlist self-satisfying. if path == Path(__file__).resolve(): continue tree = ast.parse(path.read_text(encoding="utf-8")) for node in ast.walk(tree): if isinstance(node, ast.Name): names.add(node.id) elif isinstance(node, ast.Attribute): names.add(node.attr) elif isinstance(node, ast.ImportFrom): for alias in node.names: names.add(alias.name) names.add(alias.asname or alias.name) elif isinstance(node, ast.Constant) and isinstance(node.value, str): literals.append(node.value) return names, "\n".join(literals) def _orphans() -> dict[str, str]: definitions = _public_definitions() names, literal_blob = _referenced_names() return { name: location for name, location in definitions.items() # A definition is referenced if its name is used directly, or appears inside a # string such as "transcription.__main__:create_cli_app". if ( name not in names and name not in literal_blob and name.split(".")[-1] not in names and name.split(".")[-1] not in literal_blob ) } def test_public_definitions_are_discovered(): """Guard the guard: the sweep is meaningless if nothing is scanned.""" definitions = _public_definitions() assert len(definitions) >= 420 assert "create_app" in definitions def test_framework_registered_routes_are_exempt(): """Route handlers are reachable via decorator registration, not by name.""" definitions = _public_definitions() assert "healthz_route" not in definitions assert "read_document_source_media" not in definitions def test_no_unexpected_orphaned_definitions(): """Check 10: no public definition becomes unreferenced without a recorded decision.""" unexpected = {name: location for name, location in _orphans().items() if name not in KNOWN_ORPHANS} assert unexpected == {} def test_known_orphans_are_still_orphaned(): """Keep the allowlist honest: an entry that regained callers must be removed.""" current = set(_orphans()) stale = sorted(name for name in KNOWN_ORPHANS if name not in current) assert stale == [], "these definitions are referenced again; drop them from KNOWN_ORPHANS" def test_known_orphans_document_a_rationale(): """An allowlist without reasons is just suppressed output.""" missing = sorted(name for name, reason in KNOWN_ORPHANS.items() if len(reason.strip()) < 40) assert missing == []