From eb27723de24bc34b1daeb227f079155a63c4ef51 Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:22:33 +0800 Subject: [PATCH 1/4] refactor(material-lifecycle): decide the apply outcome vocabulary in one owner `ranking.py` and `rebuild.py` each stated `{"applied", "no_change", "rejected", "rolled_back"}` at module level and each guarded its own receipt builder with its private copy, so one question -- is this token a legal apply outcome -- had two owners. A fifth outcome added on either side would have stayed illegal on the other, and no test could see the split because each copy only consulted its own file. The set now lives once in `_validation.py`, the module that already owns this package's shared validation vocabulary and the two validators both call sites invoke, and it is a `frozenset` so the owner cannot be mutated at runtime. Both consumers import and decide with that object; the accepted tokens, the refusal message and every companion-field rule are unchanged. Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- loopx/capabilities/material_lifecycle/_validation.py | 7 +++++++ loopx/capabilities/material_lifecycle/ranking.py | 6 ++---- loopx/capabilities/material_lifecycle/rebuild.py | 6 ++---- 3 files changed, 11 insertions(+), 8 deletions(-) diff --git a/loopx/capabilities/material_lifecycle/_validation.py b/loopx/capabilities/material_lifecycle/_validation.py index 38b3e4abfe..4fe3a5c724 100644 --- a/loopx/capabilities/material_lifecycle/_validation.py +++ b/loopx/capabilities/material_lifecycle/_validation.py @@ -41,6 +41,13 @@ } +# The outcome vocabulary of an owner-gated apply. Ranking records the result of +# one apply and rebuild replays a whole plan, so both ask the same question: is +# this token a legal apply outcome. It is stated once, here, because the two +# copies drifted apart only by accident, not by policy. +APPLY_STATUSES = frozenset({"applied", "no_change", "rejected", "rolled_back"}) + + def compact_text(value: Any, *, field: str, max_len: int = 320) -> str: text = " ".join(str(value or "").strip().split()) if not text: diff --git a/loopx/capabilities/material_lifecycle/ranking.py b/loopx/capabilities/material_lifecycle/ranking.py index 39f1a8efa5..d3372cd56c 100644 --- a/loopx/capabilities/material_lifecycle/ranking.py +++ b/loopx/capabilities/material_lifecycle/ranking.py @@ -9,6 +9,7 @@ from .ownership import MaterialProjectScope, material_owner_fields from ._validation import ( + APPLY_STATUSES, capability_contract, check_record_keys, compact_text, @@ -36,9 +37,6 @@ "reason_code", "to_rank", } -_APPLY_STATUSES = {"applied", "no_change", "rejected", "rolled_back"} - - def _ordered_material_refs(values: Sequence[str], *, field: str) -> list[str]: if isinstance(values, (str, bytes)) or not isinstance(values, Sequence): raise TypeError(f"{field} must be a sequence of compact tokens") @@ -283,7 +281,7 @@ def build_material_rerank_apply_receipt( """Record the result of an owner-gated apply, rejection, no-op, or rollback.""" normalized_status = compact_token(status, field="status") - if normalized_status not in _APPLY_STATUSES: + if normalized_status not in APPLY_STATUSES: raise ValueError(f"unsupported status: {normalized_status}") before = compact_token(before_revision, field="before_revision") after = compact_token(after_revision, field="after_revision") diff --git a/loopx/capabilities/material_lifecycle/rebuild.py b/loopx/capabilities/material_lifecycle/rebuild.py index 9bfcb0fcbd..86ef71d270 100644 --- a/loopx/capabilities/material_lifecycle/rebuild.py +++ b/loopx/capabilities/material_lifecycle/rebuild.py @@ -6,6 +6,7 @@ from typing import Any from ._validation import ( + APPLY_STATUSES, capability_contract, check_record_keys, compact_text, @@ -33,9 +34,6 @@ "target_rank", } _RANK_ANCHOR_FIELDS = {"material_ref", "rank"} -_APPLY_STATUSES = {"applied", "no_change", "rejected", "rolled_back"} - - def _ordered_token_list( values: Sequence[Any], *, @@ -404,7 +402,7 @@ def build_material_ranked_entry_rebuild_apply_receipt( """Record an owner-gated structural rebuild result.""" normalized_status = compact_token(status, field="status") - if normalized_status not in _APPLY_STATUSES: + if normalized_status not in APPLY_STATUSES: raise ValueError(f"unsupported status: {normalized_status}") before = compact_token(before_revision, field="before_revision") after = compact_token(after_revision, field="after_revision") From 4c3e87c7899b6c6034ca2e534c91cd89e919eaa1 Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Mon, 26 Oct 2026 21:22:33 +0800 Subject: [PATCH 2/4] test(architecture): guard the apply outcome vocabulary to one owner A census of the package reports every foldable set construction whose members are the apply outcomes, at module level *or* inline inside a function body -- module level alone let a consumer keep the owner import alive while deciding with its own literal, which is how the first version of this guard survived a mutation. Four probes pin the value-not-spelling rule (plain literal, reordered, `set(...)`, `frozenset(...)`), one pins that a widened copy is not a match, and the two consumers are checked for object identity, for naming the owner in the membership test, and for holding no private copy. Five behavior rows keep the convergence honest: each accepted status with the companion fields its own rule requires, an unknown status refused, and the legal-status table required to cover the owner set exactly. Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- ...t_material_lifecycle_apply_status_owner.py | 234 ++++++++++++++++++ 1 file changed, 234 insertions(+) create mode 100644 tests/architecture/test_material_lifecycle_apply_status_owner.py diff --git a/tests/architecture/test_material_lifecycle_apply_status_owner.py b/tests/architecture/test_material_lifecycle_apply_status_owner.py new file mode 100644 index 0000000000..621ef0f528 --- /dev/null +++ b/tests/architecture/test_material_lifecycle_apply_status_owner.py @@ -0,0 +1,234 @@ +"""Guard one owner for the material-lifecycle apply outcome vocabulary. + +`build_material_rerank_apply_receipt` records the result of one apply and +`build_material_ranked_entry_rebuild_apply_receipt` replays a whole plan. Both +ask the same question -- is this token a legal apply outcome -- and on the base +revision each stated the answer with its own module-level set, so a status added +on one side stayed illegal on the other while every scan that groups by name +still saw two matching definitions. + +Decisions are made on the folded value, not the spelling: a plain set literal, a +reordered literal, a `set(...)` or `frozenset(...)` wrapper and a comprehension +over the same members are all offenders, and a construction whose members cannot +be folded has to be declared with its file and count. +""" + +from __future__ import annotations + +import ast +import pathlib + +import pytest + +from loopx.capabilities.material_lifecycle import _validation +from loopx.capabilities.material_lifecycle._validation import APPLY_STATUSES +from loopx.capabilities.material_lifecycle.ranking import ( + build_material_rerank_apply_receipt, +) + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +PACKAGE_ROOT = REPO_ROOT / "loopx" / "capabilities" / "material_lifecycle" +OWNER_MODULE = "loopx/capabilities/material_lifecycle/_validation.py" +CONSUMERS = ( + "loopx/capabilities/material_lifecycle/ranking.py", + "loopx/capabilities/material_lifecycle/rebuild.py", +) + +VOCABULARY = {"applied", "no_change", "rejected", "rolled_back"} + + +def _folded_members(node: ast.AST) -> frozenset[str] | None: + """Return the member set of a literal set construction, or None.""" + + if isinstance(node, (ast.Set, ast.List, ast.Tuple)): + members = node.elts + elif isinstance(node, ast.Call) and getattr(node.func, "id", "") in {"set", "frozenset"}: + if len(node.args) != 1: + return None + inner = node.args[0] + if isinstance(inner, ast.SetComp): + return None if inner.generators else frozenset( + _folded_members(ast.Set(elts=inner.elt and [inner.elt])) or () + ) + members = inner.elts if isinstance(inner, (ast.Set, ast.List, ast.Tuple)) else None + if members is None: + return None + else: + return None + values = [ + element.value + for element in members + if isinstance(element, ast.Constant) and isinstance(element.value, str) + ] + return frozenset(values) if len(values) == len(members) else None + + +def set_literals(source: str) -> list[tuple[int, frozenset[str]]]: + """Every foldable set construction anywhere in the module. + + Module level alone is not enough: a consumer can keep the owner import alive + and still decide inline with its own literal inside a function body, which is + the same drift with one more step to find. + """ + + tree = ast.parse(source) + rows: list[tuple[int, frozenset[str]]] = [] + for node in ast.walk(tree): + candidates: list[ast.AST] = [] + if isinstance(node, ast.Compare): + candidates = list(node.comparators) + elif isinstance(node, (ast.Assign, ast.AnnAssign)): + targets = ( + [node.target] + if isinstance(node, ast.AnnAssign) + else node.targets + ) + if any(isinstance(target, ast.Name) for target in targets): + candidates = [node.value] + for candidate in candidates: + folded = _folded_members(candidate) + if folded is not None: + rows.append((getattr(node, "lineno", 0), folded)) + return rows + + +def vocabulary_sites( + root: pathlib.Path, *, rel_prefix: str = "" +) -> list[str]: + """Every module-level set in `root` that answers the apply-outcome question.""" + + sites: list[str] = [] + for path in sorted(root.rglob("*.py")): + if "__pycache__" in path.parts: + continue + try: + source = path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + prefix = rel_prefix or str(root.relative_to(REPO_ROOT)) + for line, members in set_literals(source): + if set(members) == VOCABULARY: + sites.append(f"{prefix}/{path.name}:{line}") + return sites + + +def test_only_the_owner_states_the_apply_vocabulary() -> None: + sites = vocabulary_sites(PACKAGE_ROOT) + + assert [site.split(":")[0] for site in sites] == [OWNER_MODULE], sites + assert len(sites) == 1, sites + + +@pytest.mark.parametrize( + "source", + [ + # a plain literal is the spelling the base revision used twice + pytest.param('APPLY = {"applied", "no_change", "rejected", "rolled_back"}\n', id="plain-literal"), + # reordering the members changes nothing about the decision + pytest.param('APPLY = {"rolled_back", "rejected", "no_change", "applied"}\n', id="reordered-literal"), + pytest.param('APPLY = set(["applied", "no_change", "rejected", "rolled_back"])\n', id="set-wrapper"), + pytest.param('APPLY = frozenset({"applied", "no_change", "rejected", "rolled_back"})\n', id="frozenset-wrapper"), + ], +) +def test_a_rewritten_copy_is_still_recognised(tmp_path: pathlib.Path, source: str) -> None: + (tmp_path / "copycat.py").write_text(source, encoding="utf-8") + + assert vocabulary_sites(tmp_path, rel_prefix="copy") == ["copy/copycat.py:1"] + + +def test_one_member_off_is_not_the_vocabulary(tmp_path: pathlib.Path) -> None: + """The census is exact, so a widened copy is a review event, not a match.""" + + (tmp_path / "widened.py").write_text( + 'APPLY = {"applied", "no_change", "rejected", "rolled_back", "undone"}\n', + encoding="utf-8", + ) + + assert vocabulary_sites(tmp_path, rel_prefix="copy") == [] + + +@pytest.mark.parametrize("relative", CONSUMERS) +def test_each_consumer_decides_with_the_owner_object(relative: str) -> None: + module_name = relative.rsplit("/", 1)[-1][:-3] + module = __import__( + f"loopx.capabilities.material_lifecycle.{module_name}", fromlist=["APPLY_STATUSES"] + ) + + assert module.APPLY_STATUSES is _validation.APPLY_STATUSES + source = pathlib.Path(REPO_ROOT / relative).read_text(encoding="utf-8") + assert source.count("not in APPLY_STATUSES") == 1, ( + f"{relative} must decide membership through the owner exactly once" + ) + assert "_APPLY_STATUSES" not in source, f"{relative} keeps a private copy" + + +LEGAL_CASES: dict[str, dict[str, object]] = { + # each shape is the minimal legal companion for its own status, which is why + # the vocabulary needs one owner: the accepted tokens and the rules attached + # to them must not be stated twice, once per consumer. + "applied": { + "applied_material_refs": ["material:one"], + "before_revision": "revision:before", + "after_revision": "revision:after", + "rollback_ref": "rollback:before", + "rejection_reason": None, + }, + "rolled_back": { + "applied_material_refs": [], + "before_revision": "revision:before", + "after_revision": "revision:after", + "rollback_ref": "rollback:before", + "rejection_reason": None, + }, + "no_change": { + "applied_material_refs": [], + "before_revision": "revision:same", + "after_revision": "revision:same", + "rollback_ref": None, + "rejection_reason": None, + }, + "rejected": { + "applied_material_refs": [], + "before_revision": "revision:same", + "after_revision": "revision:same", + "rollback_ref": None, + "rejection_reason": "owner declined the proposal", + }, +} + + +@pytest.mark.parametrize("status", sorted(LEGAL_CASES)) +def test_every_owner_status_is_a_legal_apply_outcome(status: str) -> None: + receipt = build_material_rerank_apply_receipt( + goal_id="goal:materials", + receipt_id=f"receipt:{status}", + proposal_ref="proposal:one", + observed_at="2026-01-01T00:00:00Z", + status=status, + owner_gate_ref="gate:exact-move", + validation_ref="validation:readback", + **LEGAL_CASES[status], + ) + + assert receipt["status"] == status + + +def test_the_legal_shapes_cover_the_whole_vocabulary() -> None: + assert sorted(LEGAL_CASES) == sorted(APPLY_STATUSES) + + +def test_a_status_outside_the_vocabulary_is_refused() -> None: + with pytest.raises(ValueError, match="unsupported status"): + build_material_rerank_apply_receipt( + goal_id="goal:materials", + receipt_id="receipt:refused", + proposal_ref="proposal:one", + observed_at="2026-01-01T00:00:00Z", + status="accepted", + before_revision="revision:before", + after_revision="revision:after", + owner_gate_ref="gate:exact-move", + validation_ref="validation:readback", + rollback_ref=None, + applied_material_refs=["material:one"], + ) From a771e2f7654d9337c36ddd07ca974b5362c7be58 Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:55:54 +0800 Subject: [PATCH 3/4] test(architecture): decide host separators and comprehension copies in the apply-status guard Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- ...t_material_lifecycle_apply_status_owner.py | 102 +++++++++++++++--- 1 file changed, 90 insertions(+), 12 deletions(-) diff --git a/tests/architecture/test_material_lifecycle_apply_status_owner.py b/tests/architecture/test_material_lifecycle_apply_status_owner.py index 621ef0f528..1f78a8ae41 100644 --- a/tests/architecture/test_material_lifecycle_apply_status_owner.py +++ b/tests/architecture/test_material_lifecycle_apply_status_owner.py @@ -8,15 +8,18 @@ still saw two matching definitions. Decisions are made on the folded value, not the spelling: a plain set literal, a -reordered literal, a `set(...)` or `frozenset(...)` wrapper and a comprehension -over the same members are all offenders, and a construction whose members cannot -be folded has to be declared with its file and count. +reordered literal, a `set(...)` or `frozenset(...)` wrapper, and a one-step +comprehension over the same members are all offenders. Constructions whose +members cannot be folded statically are not reported; widening the census to +them means teaching `_folded_members` to decide them and adding a case here, +rather than declaring a site and trusting it. """ from __future__ import annotations import ast import pathlib +import re import pytest @@ -37,22 +40,39 @@ VOCABULARY = {"applied", "no_change", "rejected", "rolled_back"} +def _folded_comprehension(node: ast.SetComp) -> frozenset[str] | None: + """Fold the one comprehension shape that cannot change the member set. + + ``{name for name in }`` with no filter selects exactly the + iterable's members, so it states the same vocabulary under another spelling. + Anything with a condition, a transform, or several generators is not decided + here, and claiming otherwise would put a comment over an unproved match. + """ + + if len(node.generators) != 1 or node.generators[0].ifs: + return None + generator = node.generators[0] + if not isinstance(generator.target, ast.Name) or not isinstance(node.elt, ast.Name): + return None + if node.elt.id != generator.target.id: + return None + return _folded_members(generator.iter) + + def _folded_members(node: ast.AST) -> frozenset[str] | None: """Return the member set of a literal set construction, or None.""" + if isinstance(node, ast.SetComp): + return _folded_comprehension(node) if isinstance(node, (ast.Set, ast.List, ast.Tuple)): members = node.elts elif isinstance(node, ast.Call) and getattr(node.func, "id", "") in {"set", "frozenset"}: - if len(node.args) != 1: + if len(node.args) != 1 or node.keywords: return None inner = node.args[0] - if isinstance(inner, ast.SetComp): - return None if inner.generators else frozenset( - _folded_members(ast.Set(elts=inner.elt and [inner.elt])) or () - ) - members = inner.elts if isinstance(inner, (ast.Set, ast.List, ast.Tuple)) else None - if members is None: - return None + if isinstance(inner, (ast.Set, ast.List, ast.Tuple, ast.SetComp)): + return _folded_members(inner) + return None else: return None values = [ @@ -92,6 +112,16 @@ def set_literals(source: str) -> list[tuple[int, frozenset[str]]]: return rows +def _posix_prefix(prefix: str) -> str: + """Join census sites with forward slashes whatever the host separator was. + + ``str(Path.relative_to(...))`` yields backslashes on Windows, which made the + owner comparison fail there even when exactly one correct owner existed. + """ + + return "/".join(part for part in re.split(r"[\\/]+", prefix) if part) + + def vocabulary_sites( root: pathlib.Path, *, rel_prefix: str = "" ) -> list[str]: @@ -108,7 +138,7 @@ def vocabulary_sites( prefix = rel_prefix or str(root.relative_to(REPO_ROOT)) for line, members in set_literals(source): if set(members) == VOCABULARY: - sites.append(f"{prefix}/{path.name}:{line}") + sites.append(f"{_posix_prefix(prefix)}/{path.name}:{line}") return sites @@ -128,6 +158,7 @@ def test_only_the_owner_states_the_apply_vocabulary() -> None: pytest.param('APPLY = {"rolled_back", "rejected", "no_change", "applied"}\n', id="reordered-literal"), pytest.param('APPLY = set(["applied", "no_change", "rejected", "rolled_back"])\n', id="set-wrapper"), pytest.param('APPLY = frozenset({"applied", "no_change", "rejected", "rolled_back"})\n', id="frozenset-wrapper"), + pytest.param('APPLY = {status for status in ("applied", "no_change", "rejected", "rolled_back")}\n', id="set-comprehension"), ], ) def test_a_rewritten_copy_is_still_recognised(tmp_path: pathlib.Path, source: str) -> None: @@ -232,3 +263,50 @@ def test_a_status_outside_the_vocabulary_is_refused() -> None: rollback_ref=None, applied_material_refs=["material:one"], ) + +def test_a_comprehension_over_other_members_is_not_the_vocabulary( + tmp_path: pathlib.Path, +) -> None: + """The comprehension fold widens reach, not matching: other members stay clean.""" + + (tmp_path / "other.py").write_text( + 'STATUS = {state for state in ("applied", "no_change", "rejected")}\n', + encoding="utf-8", + ) + + assert vocabulary_sites(tmp_path, rel_prefix="copy") == [] + + +@pytest.mark.parametrize( + ("prefix", "expected"), + [ + ("loopx\\capabilities\\material_lifecycle", "loopx/capabilities/material_lifecycle"), + ("loopx/capabilities/material_lifecycle", "loopx/capabilities/material_lifecycle"), + ("copy", "copy"), + ], +) +def test_sites_always_join_with_forward_slashes(prefix: str, expected: str) -> None: + """A Windows-shaped relative prefix must not turn the owner into a mismatch.""" + + assert _posix_prefix(prefix) == expected + + +def test_the_real_scan_reports_one_owner_regardless_of_separator( + tmp_path: pathlib.Path, +) -> None: + module = tmp_path / "loopx" / "capabilities" / "material_lifecycle" + module.mkdir(parents=True) + (module / "_validation.py").write_text( + 'APPLY_STATUSES = frozenset({"applied", "no_change", "rejected", "rolled_back"})\n', + encoding="utf-8", + ) + + sites = vocabulary_sites( + module, + rel_prefix=str(pathlib.PureWindowsPath("loopx", "capabilities", "material_lifecycle")), + ) + + assert sites == [ + "loopx/capabilities/material_lifecycle/_validation.py:1" + ], sites + From 1d3ee04084e2400290fb8207bfb826d8f3d8941c Mon Sep 17 00:00:00 2001 From: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:56:22 +0800 Subject: [PATCH 4/4] style(architecture): drop the trailing blank line in the apply-status guard Signed-off-by: Jiaqiiii523 <293262669+Jiaqiiii523@users.noreply.github.com> --- tests/architecture/test_material_lifecycle_apply_status_owner.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/architecture/test_material_lifecycle_apply_status_owner.py b/tests/architecture/test_material_lifecycle_apply_status_owner.py index 1f78a8ae41..f99b1f4365 100644 --- a/tests/architecture/test_material_lifecycle_apply_status_owner.py +++ b/tests/architecture/test_material_lifecycle_apply_status_owner.py @@ -309,4 +309,3 @@ def test_the_real_scan_reports_one_owner_regardless_of_separator( assert sites == [ "loopx/capabilities/material_lifecycle/_validation.py:1" ], sites -