From b6b09fb5e1e1611f0c9a58e6d240812997b9e6e5 Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Fri, 6 Nov 2026 20:42:30 +0800 Subject: [PATCH 1/4] refactor(issue-fix): read the queue receipt version from its domain pack Two modules stated issue_fix_reviewer_notification_queue_receipt_v1, and the capability surface already imported the local-time shape from the same domain pack. The guard follows the existing owner-test shape: census the folded value, pin the consumer to the owner object, and keep current and legacy receipts distinct. Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- .../issue_fix/reviewer_notification.py | 9 +- ...t_issue_fix_queue_receipt_version_owner.py | 144 ++++++++++++++++++ 2 files changed, 149 insertions(+), 4 deletions(-) create mode 100644 tests/architecture/test_issue_fix_queue_receipt_version_owner.py 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..564f4fe4af --- /dev/null +++ b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py @@ -0,0 +1,144 @@ +"""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 ``+``, or interpolated 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 + +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 = [ + value.value + for value in node.values + if isinstance(value, ast.Constant) and isinstance(value.value, str) + ] + return "".join(parts) if len(parts) == len(node.values) else None + 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 module_constants(tree: ast.Module) -> dict[str, str]: + pending: list[tuple[str, ast.AST]] = [] + constants: dict[str, str] = {} + for node in tree.body: + if isinstance(node, ast.Assign) and len(node.targets) == 1: + target = node.targets[0] + if isinstance(target, ast.Name): + pending.append((target.id, node.value)) + 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]]: + if "queue_receipt" not in source: + return [] + tree = ast.parse(source) + constants = module_constants(tree) + rows: list[dict[str, object]] = [] + for node in tree.body: + if not isinstance(node, ast.Assign) or len(node.targets) != 1: + continue + target = node.targets[0] + if not isinstance(target, ast.Name): + continue + if fold_string(node.value, constants) == QUEUE_RECEIPT_VALUE: + rows.append( + {"file": name, "line": node.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") From f107a3d100046557360efcf4671349e2d79efc3e Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Sat, 7 Nov 2026 09:45:02 +0800 Subject: [PATCH 2/4] test(issue-fix): fold the queue receipt guard instead of prefiltering source text Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- ...t_issue_fix_queue_receipt_version_owner.py | 63 ++++++++++++++++++- 1 file changed, 61 insertions(+), 2 deletions(-) diff --git a/tests/architecture/test_issue_fix_queue_receipt_version_owner.py b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py index 564f4fe4af..d046e51d0c 100644 --- a/tests/architecture/test_issue_fix_queue_receipt_version_owner.py +++ b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py @@ -20,6 +20,8 @@ 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 @@ -75,8 +77,6 @@ def module_constants(tree: ast.Module) -> dict[str, str]: def version_rows(source: str, name: str) -> list[dict[str, object]]: - if "queue_receipt" not in source: - return [] tree = ast.parse(source) constants = module_constants(tree) rows: list[dict[str, object]] = [] @@ -142,3 +142,62 @@ def test_current_and_legacy_receipt_versions_stay_distinct_at_the_boundary() -> ) 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', + ), +) +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', + ), +) + + +@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:") From c7280b3579c2d6f1d1290ec1126b303f0bbc7ce6 Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Sat, 7 Nov 2026 11:48:44 +0800 Subject: [PATCH 3/4] test(issue-fix): fold annotated module bindings in the queue receipt census Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- ...t_issue_fix_queue_receipt_version_owner.py | 104 +++++++++++++++--- 1 file changed, 87 insertions(+), 17 deletions(-) diff --git a/tests/architecture/test_issue_fix_queue_receipt_version_owner.py b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py index d046e51d0c..53def44f7b 100644 --- a/tests/architecture/test_issue_fix_queue_receipt_version_owner.py +++ b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py @@ -9,7 +9,8 @@ was written to prevent. Layer 1 judges folded values rather than spellings, so a copy written across -lines, built with ``+``, or interpolated into an f-string is reported too. +lines, built with ``+``, interpolated into an f-string, or declared with a type +annotation 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. @@ -54,14 +55,36 @@ def fold_string(node: ast.AST, constants: dict[str, str], depth: int = 0) -> str return None -def module_constants(tree: ast.Module) -> dict[str, str]: - pending: list[tuple[str, ast.AST]] = [] - constants: dict[str, str] = {} +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) and len(node.targets) == 1: + if isinstance(node, ast.Assign): + if len(node.targets) != 1: + continue target = node.targets[0] - if isinstance(target, ast.Name): - pending.append((target.id, node.value)) + 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: @@ -80,16 +103,9 @@ 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 node in tree.body: - if not isinstance(node, ast.Assign) or len(node.targets) != 1: - continue - target = node.targets[0] - if not isinstance(target, ast.Name): - continue - if fold_string(node.value, constants) == QUEUE_RECEIPT_VALUE: - rows.append( - {"file": name, "line": node.lineno, "kind": "queue_receipt_version"} - ) + 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 @@ -162,6 +178,20 @@ def test_current_and_legacy_receipt_versions_stay_distinct_at_the_boundary() -> "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', + ), ) NON_OFFENDING = ( ( @@ -172,6 +202,11 @@ def test_current_and_legacy_receipt_versions_stay_distinct_at_the_boundary() -> "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', + ), ) @@ -201,3 +236,38 @@ def test_a_split_copy_is_rejected_through_the_repository_scan( 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) == [] From 2be88a04981fa78aa7c875dfb658fe87e686185e Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Sat, 7 Nov 2026 13:56:54 +0800 Subject: [PATCH 4/4] test(issue-fix): fold statically evaluable interpolations in the queue receipt guard Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- ...t_issue_fix_queue_receipt_version_owner.py | 89 +++++++++++++++++-- 1 file changed, 81 insertions(+), 8 deletions(-) diff --git a/tests/architecture/test_issue_fix_queue_receipt_version_owner.py b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py index 53def44f7b..4bc99c1658 100644 --- a/tests/architecture/test_issue_fix_queue_receipt_version_owner.py +++ b/tests/architecture/test_issue_fix_queue_receipt_version_owner.py @@ -9,8 +9,8 @@ was written to prevent. Layer 1 judges folded values rather than spellings, so a copy written across -lines, built with ``+``, interpolated into an f-string, or declared with a type -annotation is reported too. +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. @@ -40,12 +40,25 @@ def fold_string(node: ast.AST, constants: dict[str, str], depth: int = 0) -> str if isinstance(node, ast.Constant): return node.value if isinstance(node.value, str) else None if isinstance(node, ast.JoinedStr): - parts = [ - value.value - for value in node.values - if isinstance(value, ast.Constant) and isinstance(value.value, str) - ] - return "".join(parts) if len(parts) == len(node.values) else None + 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) @@ -192,6 +205,16 @@ def test_current_and_legacy_receipt_versions_stay_distinct_at_the_boundary() -> '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 = ( ( @@ -207,6 +230,21 @@ def test_current_and_legacy_receipt_versions_stay_distinct_at_the_boundary() -> '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', + ), ) @@ -271,3 +309,38 @@ def test_the_census_leaves_a_typed_legacy_receipt_alone( 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) == []