From 03cbbdac41aa10d9c348b23b78941b6ee9d36f6c 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/2] 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 50ed43b9ac9688cb25a3679744159b40ebe4a265 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/2] 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"], + )