diff --git a/.github/skills/python-code-reviewer/skill.md b/.github/skills/python-code-reviewer/skill.md index ecfae3f..f0aac16 100644 --- a/.github/skills/python-code-reviewer/skill.md +++ b/.github/skills/python-code-reviewer/skill.md @@ -53,14 +53,14 @@ Where **Enforced by** reads *unenforced*, recommending a deterministic test is i | :-- | :--- | :--- | | 1 | **Service boundary rule:** no service-to-service imports | `tests/test_service_boundaries.py` | | 2 | **UI boundary rule:** pages/components do not perform persistence access | `tests/test_ui_boundaries.py` | -| 3 | **Status vocabulary conformance:** `JobStatus`/`JobSourceStatus` usage matches current enums in `src/transcription/db/models.py`; no stringly-typed status literals | *unenforced* — only incidental coverage via `tests/services/test_job_service.py` | +| 3 | **Status vocabulary conformance:** `JobStatus`/`JobSourceStatus`/`JobPurpose` usage matches current enums in `src/transcription/db/models.py`; no stringly-typed status literals | `tests/test_model_contract_guards.py` | | 4 | **Evidence ownership conformance:** append-only attempt history is preserved and projection writes are not mistaken for history mutation (`src/transcription/services/sources.py`, `src/transcription/services/evidence.py`) | `tests/test_v42_evidence.py::test_attempts_are_append_only_and_exported_with_integrity` | | 5 | **Canonical authority:** findings must resolve against `docs/*` first | `tests/test_meta_contract_guards.py::test_canonical_authority_references_are_present` | -| 6 | **Schema contract fidelity:** when model/persistence behavior changes, `docs/schema.md` remains field-accurate with `src/transcription/db/models.py` | *partial* — `tests/test_meta_contract_guards.py` verifies presence and references only, **not** field accuracy | +| 6 | **Schema contract fidelity:** when model/persistence behavior changes, `docs/schema.md` remains field-accurate with `src/transcription/db/models.py` | `tests/test_model_contract_guards.py` (field names, ordering, enum members, table coverage), `tests/test_meta_contract_guards.py` (presence and references) | | 7 | **Media boundary conformance:** print/export media is record-validated and UI media URL generation uses controlled resolver paths | `tests/test_media_path_safety.py`, `tests/ui/test_media_urls.py` | -| 8 | **Eager-loading conformance:** service/UI read paths satisfy `lazy="raise"` expectations | *unenforced* — no guard test; only incidental use in `tests/ui/test_sources_page.py` | +| 8 | **Eager-loading conformance:** service/UI read paths satisfy `lazy="raise"` expectations | `tests/test_model_contract_guards.py` (declaration-side; documented `noload` exceptions must match `docs/schema.md`) | | 9 | **Cross-cutting error conformance:** service/API/UI translation and retry behavior align with `.github/instructions/error-handling.instructions.md` | `tests/test_errors.py`, `tests/api/test_error_responses.py`, `tests/ui/test_error_presenter.py` | -| 10 | **Orphaned/dead-code conformance:** include a deterministic orphan sweep and report confirmed orphans removed/retained with rationale | *unenforced* | +| 10 | **Orphaned/dead-code conformance:** include a deterministic orphan sweep and report confirmed orphans removed/retained with rationale | `tests/test_orphan_sweep.py` (`KNOWN_ORPHANS` records each retained orphan and its rationale) | ## Core Review Areas diff --git a/src/transcription/ui/pages/sources_page.py b/src/transcription/ui/pages/sources_page.py index 6b1619f..8b58893 100644 --- a/src/transcription/ui/pages/sources_page.py +++ b/src/transcription/ui/pages/sources_page.py @@ -652,7 +652,7 @@ def _render_machine_candidates( successful = [ attempt for attempt in attempts - if attempt.status.value == "transcribed" and attempt.raw_transcription + if attempt.status == JobSourceStatus.TRANSCRIBED and attempt.raw_transcription ] candidates = [ attempt for attempt in successful if attempt.id != source.preferred_execution_attempt_id diff --git a/tests/test_model_contract_guards.py b/tests/test_model_contract_guards.py new file mode 100644 index 0000000..4a61ac1 --- /dev/null +++ b/tests/test_model_contract_guards.py @@ -0,0 +1,278 @@ +"""Deterministic guards for the model and persistence contracts. + +These cover three checks that `.github/skills/python-code-reviewer/skill.md` requires +on every review but that previously had no automated enforcement: + +* **Status vocabulary conformance** - status comparisons and assignments must use the + `JobStatus` / `JobSourceStatus` / `JobPurpose` enums rather than string literals. +* **Relationship loading contract** - relationships declare ``lazy="raise"`` so read + paths must eager-load explicitly, and the documented exceptions in `docs/schema.md` + must match the code exactly. +* **Schema contract fidelity** - the "Field-Accurate Table Contracts" tables in + `docs/schema.md` must list exactly the fields each SQLModel table declares. + +See `docs/requirements.md` (REQ-4-102), `docs/schema.md` ("Relationship Loading +Contract"), and `.github/instructions/services.instructions.md`. +""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path + +from sqlmodel import SQLModel + +from transcription.db import models as models_module +from transcription.db.models import JobPurpose +from transcription.db.models import JobSourceStatus +from transcription.db.models import JobStatus + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +SOURCE_DIR = PROJECT_ROOT / "src" / "transcription" +MODELS_PATH = SOURCE_DIR / "db" / "models.py" +SCHEMA_DOC = PROJECT_ROOT / "docs" / "schema.md" + +STATUS_ENUMS = (JobStatus, JobSourceStatus, JobPurpose) +STATUS_VALUES = frozenset(member.value for enum in STATUS_ENUMS for member in enum) + +# Attribute names that carry a status enum. A string literal compared against or +# assigned to one of these is a stringly-typed status, even if it happens to match. +STATUS_ATTRIBUTES = frozenset({"status", "purpose"}) + +# `JobSource.execution_attempts` loads attempt evidence on demand rather than raising, +# because evidence is fetched deliberately by the services that own it. Documented in +# `docs/schema.md` under "Relationship Loading Contract". +DOCUMENTED_LOADING_EXCEPTIONS = {"execution_attempts": "noload"} + + +def _python_files() -> list[Path]: + return sorted(SOURCE_DIR.rglob("*.py")) + + +def _relative(path: Path) -> str: + return path.relative_to(PROJECT_ROOT).as_posix() + + +def _is_status_target(node: ast.expr) -> bool: + """True for `x.status`, `x.purpose`, and their `.value` unwrappings.""" + if isinstance(node, ast.Attribute): + if node.attr in STATUS_ATTRIBUTES: + return True + if node.attr == "value": + return _is_status_target(node.value) + return False + + +def _string_constant(node: ast.expr) -> str | None: + return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) else None + + +def _status_literal_violations(tree: ast.Module) -> list[tuple[int, str]]: + found: list[tuple[int, str]] = [] + for node in ast.walk(tree): + if isinstance(node, ast.Compare): + operands = [node.left, *node.comparators] + if not any(_is_status_target(operand) for operand in operands): + continue + for operand in operands: + literal = _string_constant(operand) + if literal is not None: + found.append((node.lineno, literal)) + elif isinstance(node, ast.Call): + for keyword in node.keywords: + if keyword.arg not in STATUS_ATTRIBUTES: + continue + literal = _string_constant(keyword.value) + if literal is not None: + found.append((node.lineno, literal)) + return found + + +def test_status_enums_expose_expected_vocabulary(): + """Guard the guard: the scan below is meaningless if the enums are empty.""" + assert {member.value for member in JobStatus} == { + "queued", + "processing", + "transcribed", + "partial_success", + "failed", + } + assert {member.value for member in JobSourceStatus} == { + "pending", + "transcribed", + "failed", + "cancelled", + } + assert {member.value for member in JobPurpose} == {"transcription", "retranscription"} + + +def test_no_stringly_typed_status_comparisons_or_assignments(): + """Status handling must go through the enums, never raw strings. + + `attempt.status.value == "transcribed"` silently survives an enum rename and + compares a projection of the value rather than the value itself. + """ + violations: dict[str, list[tuple[int, str]]] = {} + for path in _python_files(): + tree = ast.parse(path.read_text(encoding="utf-8")) + found = _status_literal_violations(tree) + if found: + violations[_relative(path)] = found + assert violations == {} + + +def test_status_string_literals_outside_models_are_accounted_for(): + """Any bare status-valued literal in the package must be a known non-status use. + + This is deliberately narrower than the comparison scan: it catches literals that + merely *look* like statuses, so a genuine new one cannot slip in unnoticed. + """ + allowed = { + # `JobStatus` / `JobSourceStatus` / `JobPurpose` member definitions. + "src/transcription/db/models.py", + # `WorkerHealthState` is a separate Literal vocabulary that reuses "failed". + "src/transcription/worker.py", + # Distribution name lookups for `transcription`, not the JobPurpose member. + "src/transcription/config.py", + "src/transcription/providers/evidence.py", + # UI placeholder copy where a value is absent, not a status render. + "src/transcription/ui/pages/jobs_page.py", + } + unexpected: dict[str, list[tuple[int, str]]] = {} + for path in _python_files(): + relative = _relative(path) + if relative in allowed: + continue + tree = ast.parse(path.read_text(encoding="utf-8")) + found = [ + (node.lineno, node.value) + for node in ast.walk(tree) + if isinstance(node, ast.Constant) and isinstance(node.value, str) and node.value in STATUS_VALUES + ] + if found: + unexpected[relative] = found + assert unexpected == {} + + +def _relationship_loading_strategies() -> dict[str, dict[str, str | None]]: + """Map each model attribute defined via `Relationship(...)` to its lazy strategy.""" + tree = ast.parse(MODELS_PATH.read_text(encoding="utf-8")) + strategies: dict[str, dict[str, str | None]] = {} + for class_node in tree.body: + if not isinstance(class_node, ast.ClassDef): + continue + for statement in class_node.body: + if not isinstance(statement, ast.AnnAssign) or statement.value is None: + continue + call = statement.value + if not isinstance(call, ast.Call) or getattr(call.func, "id", None) != "Relationship": + continue + attribute = statement.target.id if isinstance(statement.target, ast.Name) else "" + lazy: str | None = None + for keyword in call.keywords: + if keyword.arg != "sa_relationship_kwargs" or not isinstance(keyword.value, ast.Dict): + continue + for key, value in zip(keyword.value.keys, keyword.value.values, strict=True): + if isinstance(key, ast.Constant) and key.value == "lazy" and isinstance(value, ast.Constant): + lazy = value.value + strategies.setdefault(class_node.name, {})[attribute] = lazy + return strategies + + +def test_relationships_are_discovered(): + """Guard the guard: the loading rules below are meaningless if nothing is scanned.""" + strategies = _relationship_loading_strategies() + assert {"Document", "Job", "JobSource", "Source"} <= set(strategies) + assert sum(len(attributes) for attributes in strategies.values()) >= 25 + + +def test_relationships_declare_lazy_raise_except_documented_cases(): + """REQ-4-102: relationships raise on implicit load so read shape stays explicit.""" + violations: dict[str, str | None] = {} + for model_name, attributes in _relationship_loading_strategies().items(): + for attribute, lazy in attributes.items(): + expected = DOCUMENTED_LOADING_EXCEPTIONS.get(attribute, "raise") + if lazy != expected: + violations[f"{model_name}.{attribute}"] = lazy + assert violations == {} + + +def test_documented_loading_exceptions_match_schema_doc(): + """The exception list is only trustworthy while `docs/schema.md` agrees with it.""" + schema_text = SCHEMA_DOC.read_text(encoding="utf-8") + contract = schema_text.split("## Relationship Loading Contract", 1)[1] + for attribute, strategy in DOCUMENTED_LOADING_EXCEPTIONS.items(): + assert attribute in contract, f"{attribute} is exempted in code but not documented" + assert f'`lazy="{strategy}"`' in contract + + +def _documented_table_fields() -> dict[str, list[str]]: + schema_text = SCHEMA_DOC.read_text(encoding="utf-8") + section = schema_text.split("## Field-Accurate Table Contracts", 1)[1] + documented: dict[str, list[str]] = {} + for block in re.split(r"\n### ", section)[1:]: + heading = block.splitlines()[0].strip().strip("`") + documented[heading] = re.findall(r"^\| `([^`]+)` \|", block, re.MULTILINE) + return documented + + +def _table_models() -> dict[str, type[SQLModel]]: + return { + name: attribute + for name, attribute in vars(models_module).items() + if isinstance(attribute, type) + and issubclass(attribute, SQLModel) + and attribute is not SQLModel + and getattr(attribute, "__table__", None) is not None + } + + +def test_schema_doc_documents_every_table_model(): + """Every persisted table needs a field contract, and vice versa.""" + documented = set(_documented_table_fields()) + actual = set(_table_models()) + assert actual, "no table models discovered" + assert documented - actual == set(), "schema.md documents tables that no longer exist" + assert actual - documented == set(), "schema.md is missing tables that exist in models.py" + + +def test_schema_doc_field_contracts_match_models(): + """Check 6: `docs/schema.md` stays field-accurate with `db/models.py`.""" + documented = _documented_table_fields() + drift: dict[str, dict[str, list[str]]] = {} + for name, model in _table_models().items(): + expected = set(model.model_fields) + listed = set(documented.get(name, [])) + if expected != listed: + drift[name] = { + "undocumented_fields": sorted(expected - listed), + "stale_doc_entries": sorted(listed - expected), + } + assert drift == {} + + +def test_schema_doc_lists_fields_in_declaration_order(): + """Ordering drift is how a doc silently stops being reviewable against the model.""" + documented = _documented_table_fields() + misordered = { + name: {"documented": documented[name], "declared": list(model.model_fields)} + for name, model in _table_models().items() + if documented.get(name, []) != list(model.model_fields) + } + assert misordered == {} + + +def test_schema_doc_enumerations_match_status_enums(): + """The "Authoritative Enumerations" section must list the real members.""" + schema_text = SCHEMA_DOC.read_text(encoding="utf-8") + section = schema_text.split("## Authoritative Enumerations", 1)[1].split("\n## ", 1)[0] + drift: dict[str, dict[str, list[str]]] = {} + for enum in STATUS_ENUMS: + block = re.split(rf"\n### {enum.__name__}\n", section) + assert len(block) == 2, f"{enum.__name__} has no section in docs/schema.md" + listed = re.findall(r"^- `([^`]+)`", block[1].split("\n### ", 1)[0], re.MULTILINE) + expected = [member.value for member in enum] + if listed != expected: + drift[enum.__name__] = {"documented": listed, "declared": expected} + assert drift == {} diff --git a/tests/test_orphan_sweep.py b/tests/test_orphan_sweep.py new file mode 100644 index 0000000..d551d29 --- /dev/null +++ b/tests/test_orphan_sweep.py @@ -0,0 +1,146 @@ +"""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 only considers module-level public +definitions, and it 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] = { + "BenchmarkManifest": ( + "Benchmark manifest model in benchmarking.py with no current caller. " + "Uncertain - follow-up: confirm whether the benchmarking entrypoint is still " + "intended before removing." + ), + "dispose_all_engines": ( + "Engine lifecycle helper in db/engine.py. Uncertain - follow-up: operational " + "teardown utility with no runtime or test caller." + ), + "refresh_engine": ( + "Engine lifecycle helper in db/engine.py. Uncertain - follow-up: paired with " + "dispose_all_engines and equally unreferenced." + ), + "summarize_error": ( + "Error-presentation helper in ui/components/error_presenter.py that no page or " + "component calls. Uncertain - follow-up: superseded by the presenter's other " + "entrypoints." + ), +} + + +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 module-level definitions, mapped to `path:line`.""" + 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}" + 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 + } + + +def test_public_definitions_are_discovered(): + """Guard the guard: the sweep is meaningless if nothing is scanned.""" + definitions = _public_definitions() + assert len(definitions) >= 200 + 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 == []