diff --git a/loopx/capabilities/issue_fix/reviewer_notification.py b/loopx/capabilities/issue_fix/reviewer_notification.py index 56cb27d643..dd7aa62ef9 100644 --- a/loopx/capabilities/issue_fix/reviewer_notification.py +++ b/loopx/capabilities/issue_fix/reviewer_notification.py @@ -16,7 +16,11 @@ reviewer_notification_before_send_gate, ) from ...control_plane.content_digest import ENVELOPED_SHA256_PATTERN -from ...domain_packs.issue_fix import REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN +from ...domain_packs.issue_fix import ( + REVIEWER_NOTIFICATION_LOCAL_TIME_PATTERN, + REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION as + ISSUE_FIX_REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION, +) ISSUE_FIX_REVIEWER_NOTIFICATION_SINKS_INPUT_SCHEMA_VERSION = ( @@ -28,9 +32,6 @@ ISSUE_FIX_REVIEWER_NOTIFICATION_SINK_RESULT_SCHEMA_VERSION = ( "issue_fix_reviewer_notification_sink_result_v0" ) -ISSUE_FIX_REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION = ( - "issue_fix_reviewer_notification_queue_receipt_v1" -) ISSUE_FIX_REVIEWER_NOTIFICATION_LEGACY_QUEUE_RECEIPT_SCHEMA_VERSION = ( "issue_fix_reviewer_notification_queue_receipt_v0" ) diff --git a/tests/architecture/test_issue_fix_queue_receipt_version_owner.py b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py new file mode 100644 index 0000000000..4bc99c1658 --- /dev/null +++ b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py @@ -0,0 +1,346 @@ +"""One owner decides the reviewer-notification queue receipt schema version. + +``issue_fix_reviewer_notification_queue_receipt_v1`` names one wire contract, and +two modules stated it: ``loopx/domain_packs/issue_fix.py``, which the capability +surface already imports for the local-time shape, and +``loopx/capabilities/issue_fix/reviewer_notification.py``, which restated it under +an ``ISSUE_FIX_`` prefix. A version bump in one file would have left the other +accepting the previous contract, which is the failure the local-time owner test +was written to prevent. + +Layer 1 judges folded values rather than spellings, so a copy written across +lines, built with ``+``, declared with a type annotation, or interpolated from +same-file constants into an f-string is reported too. +Layer 2 pins the consuming module to the owner's object and checks that it keeps +no local binding. Layer 3 records that the current and legacy receipts stay +distinct at the boundary that reads a queued receipt. +""" + +from __future__ import annotations + +import ast +import pathlib + +import pytest + +from loopx.capabilities.issue_fix import reviewer_notification +from loopx.domain_packs import issue_fix as issue_fix_pack + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +OWNER_PATH = "loopx/domain_packs/issue_fix.py" +OWNER_NAME = "REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION" +CONSUMER_PATH = "loopx/capabilities/issue_fix/reviewer_notification.py" +CONSUMER_NAME = "ISSUE_FIX_REVIEWER_NOTIFICATION_QUEUE_RECEIPT_SCHEMA_VERSION" +QUEUE_RECEIPT_VALUE = "issue_fix_reviewer_notification_queue_receipt_v1" + +DECLARED_UNFOLDABLE_SITES: dict[str, int] = {} + + +def fold_string(node: ast.AST, constants: dict[str, str], depth: int = 0) -> str | None: + if isinstance(node, ast.Constant): + return node.value if isinstance(node.value, str) else None + if isinstance(node, ast.JoinedStr): + parts: list[str] = [] + for piece in node.values: + if isinstance(piece, ast.Constant): + if not isinstance(piece.value, str): + return None + parts.append(piece.value) + continue + if ( + isinstance(piece, ast.FormattedValue) + and piece.conversion in (-1, None) + and piece.format_spec is None + ): + folded = fold_string(piece.value, constants, depth + 1) + if folded is None: + return None + parts.append(folded) + continue + return None + return "".join(parts) + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + left = fold_string(node.left, constants, depth + 1) + right = fold_string(node.right, constants, depth + 1) + return None if left is None or right is None else left + right + if isinstance(node, ast.Name) and depth < 4: + return constants.get(node.id) + return None + + +def stated_bindings(tree: ast.Module) -> list[tuple[str, int, ast.expr]]: + """One extraction rule for module-level string bindings. + + A plain assignment and an annotated assignment with a value state the same + thing before anything runs, so both feed the fold and both are census + candidates; reading only ``ast.Assign`` leaves ``COPY: Final[str] = ...`` a + restatement this guard never sees. + """ + stated: list[tuple[str, int, ast.expr]] = [] + for node in tree.body: + if isinstance(node, ast.Assign): + if len(node.targets) != 1: + continue + target = node.targets[0] + elif isinstance(node, ast.AnnAssign): + if node.value is None: + continue + target = node.target + else: + continue + if isinstance(target, ast.Name): + stated.append((target.id, node.lineno, node.value)) + return stated + + +def module_constants(tree: ast.Module) -> dict[str, str]: + pending: list[tuple[str, ast.AST]] = [ + (bound, value) for bound, _line, value in stated_bindings(tree) + ] + constants: dict[str, str] = {} + for _ in range(4): + unresolved: list[tuple[str, ast.AST]] = [] + for name, value in pending: + folded = fold_string(value, {k: v for k, v in constants.items() if k != name}) + if folded is None: + unresolved.append((name, value)) + else: + constants[name] = folded + pending = unresolved + if not pending: + break + return constants + + +def version_rows(source: str, name: str) -> list[dict[str, object]]: + tree = ast.parse(source) + constants = module_constants(tree) + rows: list[dict[str, object]] = [] + for _bound, lineno, value in stated_bindings(tree): + if fold_string(value, constants) == QUEUE_RECEIPT_VALUE: + rows.append({"file": name, "line": lineno, "kind": "queue_receipt_version"}) + return rows + + +def collect(root: pathlib.Path) -> list[dict[str, object]]: + rows: list[dict[str, object]] = [] + for path in sorted((root / "loopx").rglob("*.py")): + if "__pycache__" in path.parts: + continue + try: + source = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + rows.extend(version_rows(source, path.relative_to(root).as_posix())) + return rows + + +def offenders(rows: list[dict[str, object]]) -> list[str]: + found = [ + f"{row['file']}:{row['line']} restates the queue receipt schema version" + for row in rows + if row["file"] != OWNER_PATH + ] + for file, allowed in DECLARED_UNFOLDABLE_SITES.items(): + count = sum(1 for row in rows if row["file"] == file) + if count != allowed: + found.append(f"{file} declares {allowed} site(s), found {count}") + return sorted(set(found)) + + +def test_only_the_domain_pack_states_the_queue_receipt_version() -> None: + rows = collect(REPO_ROOT) + reported = offenders(rows) + assert reported == [], reported + assert any(row["file"] == OWNER_PATH for row in rows), rows + + +def test_the_consumer_binds_the_owner_object_without_a_local_copy() -> None: + owner_value = getattr(issue_fix_pack, OWNER_NAME) + assert owner_value == QUEUE_RECEIPT_VALUE, owner_value + assert getattr(reviewer_notification, CONSUMER_NAME) is owner_value + source = (REPO_ROOT / CONSUMER_PATH).read_text(encoding="utf-8") + assert f"{CONSUMER_NAME} = (" not in source, "consumer re-declared the version" + + +def test_current_and_legacy_receipt_versions_stay_distinct_at_the_boundary() -> None: + owner_value = getattr(issue_fix_pack, OWNER_NAME) + legacy = getattr( + reviewer_notification, + "ISSUE_FIX_REVIEWER_NOTIFICATION_LEGACY_QUEUE_RECEIPT_SCHEMA_VERSION", + ) + assert owner_value != legacy, (owner_value, legacy) + assert owner_value.endswith("_v1") and legacy.endswith("_v0") + + +OFFENDING = ( + ( + "split literal addition", + 'COPY = "issue_fix_reviewer_notification_" + "queue_" + "receipt_v1"\n', + ), + ( + "adjacent literals across lines", + 'COPY = (\n "issue_fix_reviewer_notification_queue_" "receipt_v1"\n)\n', + ), + ( + "same-file constants concatenated", + 'HEAD = "issue_fix_reviewer_notification_queue_"\nTAIL = "receipt_v1"\n' + "COPY = HEAD + TAIL\n", + ), + ( + "constant pieces inside adjacent f-strings", + 'COPY = f"issue_fix_reviewer_notification_queue_rece" f"ipt_v1"\n', + ), + ( + "typed annotation with the full literal", + 'from typing import Final\n\nCOPY: Final[str] = "issue_fix_reviewer_notification_' + 'queue_receipt_v1"\n', + ), + ( + "typed annotation built with addition", + 'COPY: str = "issue_fix_reviewer_notification_" + "queue_" + "receipt_v1"\n', + ), + ( + "typed annotation over a typed same-file constant", + 'from typing import Final\nHEAD: Final[str] = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: Final[str] = HEAD + "receipt_v1"\n', + ), + ( + "interpolated same-file constant", + 'PREFIX: str = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: str = f"{PREFIX}receipt_v1"\n', + ), + ( + "typed annotation over an interpolated constant", + 'from typing import Final\nPREFIX: Final[str] = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: Final[str] = f"{PREFIX}receipt_v1"\n', + ), +) +NON_OFFENDING = ( + ( + "the legacy receipt is a different decision", + 'COPY = "issue_fix_reviewer_notification_queue_receipt_v0"\n', + ), + ( + "an unrelated contract name is unrelated", + 'COPY = "issue_fix_reviewer_notification_queue_projection_v1"\n', + ), + ( + "a typed legacy receipt is a different decision", + 'from typing import Final\n\nCOPY: Final[str] = "issue_fix_reviewer_notification_' + 'queue_receipt_v0"\n', + ), + ( + "an interpolated legacy receipt is a different decision", + 'PREFIX: str = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: str = f"{PREFIX}receipt_v0"\n', + ), + ( + "a repr conversion is not the stated literal", + 'PREFIX: str = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: str = f"{PREFIX!r}receipt_v1"\n', + ), + ( + "a format spec is not the stated literal", + 'PREFIX: str = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: str = f"{PREFIX:>60}receipt_v1"\n', + ), +) + + +@pytest.mark.parametrize( + ("label", "source"), OFFENDING, ids=[row[0] for row in OFFENDING] +) +def test_the_scan_reports_a_restated_queue_receipt_version(label: str, source: str) -> None: + rows = version_rows(source, "loopx/capabilities/probe.py") + assert rows, label + assert offenders(rows) != [], label + + +@pytest.mark.parametrize( + ("label", "source"), NON_OFFENDING, ids=[row[0] for row in NON_OFFENDING] +) +def test_the_scan_leaves_a_different_contract_alone(label: str, source: str) -> None: + assert version_rows(source, "loopx/capabilities/probe.py") == [], label + + +def test_a_split_copy_is_rejected_through_the_repository_scan( + tmp_path: pathlib.Path, +) -> None: + source = 'COPY = "issue_fix_reviewer_notification_" + "queue_" + "receipt_v1"\n' + probe_root = tmp_path / "loopx" / "capabilities" + probe_root.mkdir(parents=True) + (probe_root / "probe.py").write_text(source, encoding="utf-8") + rows = collect(tmp_path) + assert [row["file"] for row in rows] == ["loopx/capabilities/probe.py"] + assert offenders(rows)[0].startswith("loopx/capabilities/probe.py:") + + +def test_the_census_flags_a_typed_copy_beside_the_real_owner( + tmp_path: pathlib.Path, +) -> None: + owner_dir = tmp_path / "loopx" / "domain_packs" + owner_dir.mkdir(parents=True) + (owner_dir / "issue_fix.py").write_text( + (REPO_ROOT / OWNER_PATH).read_text(encoding="utf-8"), encoding="utf-8" + ) + copy_dir = tmp_path / "loopx" / "capabilities" + copy_dir.mkdir(parents=True) + (copy_dir / "copy.py").write_text( + f"from typing import Final\n\nCOPY: Final[str] = \"{QUEUE_RECEIPT_VALUE}\"\n", + encoding="utf-8", + ) + rows = collect(tmp_path) + assert sorted(str(row["file"]) for row in rows) == [ + "loopx/capabilities/copy.py", + OWNER_PATH, + ] + assert offenders(rows) != [], offenders(rows) + + +def test_the_census_leaves_a_typed_legacy_receipt_alone( + tmp_path: pathlib.Path, +) -> None: + copy_dir = tmp_path / "loopx" / "capabilities" + copy_dir.mkdir(parents=True) + (copy_dir / "legacy.py").write_text( + 'from typing import Final\n\nCOPY: Final[str] = ' + '"issue_fix_reviewer_notification_queue_receipt_v0"\n', + encoding="utf-8", + ) + assert collect(tmp_path) == [] + +def test_the_census_flags_an_interpolated_copy_beside_the_real_owner( + tmp_path: pathlib.Path, +) -> None: + owner_dir = tmp_path / "loopx" / "domain_packs" + owner_dir.mkdir(parents=True) + (owner_dir / "issue_fix.py").write_text( + (REPO_ROOT / OWNER_PATH).read_text(encoding="utf-8"), encoding="utf-8" + ) + copy_dir = tmp_path / "loopx" / "capabilities" + copy_dir.mkdir(parents=True) + (copy_dir / "copy.py").write_text( + 'PREFIX: str = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: str = f"{PREFIX}receipt_v1"\n', + encoding="utf-8", + ) + rows = collect(tmp_path) + assert sorted(str(row["file"]) for row in rows) == [ + "loopx/capabilities/copy.py", + OWNER_PATH, + ] + assert offenders(rows) != [], offenders(rows) + + +def test_the_census_leaves_an_interpolated_legacy_receipt_alone( + tmp_path: pathlib.Path, +) -> None: + copy_dir = tmp_path / "loopx" / "capabilities" + copy_dir.mkdir(parents=True) + (copy_dir / "legacy.py").write_text( + 'PREFIX: str = "issue_fix_reviewer_notification_queue_"\n' + 'COPY: str = f"{PREFIX}receipt_v0"\n', + encoding="utf-8", + ) + assert collect(tmp_path) == []