test(architecture): close the two coverage gaps review found in the owner censuses

tests/architecture/test_chat_action_kind_owner.py

@@ -5,6 +5,11 @@
 ``loopx/chat_actions.py`` kept a byte-identical copy under a different name and
 gated dispatch on it, so a kind added to one side stayed invisible to the other
 and a proposal could be accepted by the service while the store rejected it.
+
+The census reads annotated module-level assignments as well as plain ones: a
+``LOCAL: set[str] = {...}`` restates the same decision as a copy, so a scanner
+that only visited ``ast.Assign`` would certify a fold while leaving a typed
+second owner in the tree.
 """
 
 from __future__ import annotations
@@ -55,21 +60,36 @@ def set_literal(node: ast.AST) -> frozenset[str] | None:
     return frozenset(members)
 
 
+def module_level_bindings(source: str) -> list[tuple[list[ast.expr], ast.expr | None]]:
+    """(targets, value) pairs for module-level assignments, typed or untyped."""
+    bindings: list[tuple[list[ast.expr], ast.expr | None]] = []
+    for statement in ast.parse(source).body:
+        if isinstance(statement, ast.Assign):
+            bindings.append((list(statement.targets), statement.value))
+        elif isinstance(statement, ast.AnnAssign):
+            bindings.append(([statement.target], statement.value))
+    return bindings
+
+
+def vocabulary_assignments(source: str) -> list[str]:
+    """Names in this source that restate the owner's vocabulary as their own set."""
+    held: list[str] = []
+    for targets, value in module_level_bindings(source):
+        if value is None or set_literal(value) != KIND_VALUE:
+            continue
+        held.extend(target.id for target in targets if isinstance(target, ast.Name))
+    return held
+
+
 def modules_holding_the_vocabulary() -> list[str]:
     offenders: list[str] = []
     for path in sorted(REPO_ROOT.glob("loopx/**/*.py")):
-        relative = path.relative_to(REPO_ROOT).as_posix()
         try:
-            tree = ast.parse(path.read_text(encoding="utf-8"))
+            held = vocabulary_assignments(path.read_text(encoding="utf-8"))
         except (SyntaxError, UnicodeDecodeError):  # pragma: no cover - guarded tree
             continue
-        for statement in tree.body:
-            if not isinstance(statement, ast.Assign):
-                continue
-            vocabulary = set_literal(statement.value)
-            if vocabulary == KIND_VALUE:
-                offenders.append(relative)
-                break
+        if held:
+            offenders.append(path.relative_to(REPO_ROOT).as_posix())
     return offenders
 
 
@@ -85,6 +105,21 @@ def test_no_module_outside_the_owner_holds_its_own_copy() -> None:
     assert modules_holding_the_vocabulary() == [OWNER_PATH]
 
 
+@pytest.mark.parametrize(
+    "declaration",
+    ["SHADOW = {MEMBERS}", "SHADOW: set[str] = {MEMBERS}", "SHADOW: frozenset[str] = frozenset({MEMBERS})"],
+)
+def test_the_census_reads_a_reintroduced_copy_whatever_it_is_called_or_annotated(
+    declaration: str,
+) -> None:
+    # Members are written in reverse so a reordered copy is proven to count too.
+    members = ", ".join(repr(value) for value in sorted(KIND_VALUE, reverse=True))
+
+    assert vocabulary_assignments(declaration.replace("{MEMBERS}", "{" + members + "}") + "\n") == [
+        "SHADOW"
+    ]
+
+
 @pytest.mark.parametrize("action_kind", sorted(KIND_VALUE))
 def test_every_declared_kind_is_dispatchable(action_kind: str) -> None:
     assert action_kind in chat_actions.SUPPORTED_ACTION_KINDS

tests/architecture/test_progress_review_pending_reason_owner.py

@@ -6,6 +6,14 @@
 pending reason was the exception: each side declared ``"pending_evaluation"``
 under its own name, so a rename on one side would leave the other silently
 comparing against a reason that no longer exists.
+
+A consumer must import the owner's name and then leave it alone. Checking only
+that the import exists, or that the value still compares equal, is not enough:
+rebinding the name below the import with a computed but equal string keeps the
+import, keeps the value and still installs a second owner, and an identity check
+cannot see it either because CPython folds and interns such a string. The guard
+therefore reads the import line and rejects any later module-level assignment to
+that name, whatever value shape it uses.
 """
 
 from __future__ import annotations
@@ -63,39 +71,79 @@ def modules_holding_the_reason() -> list[str]:
     return holders
 
 
-def owner_import_levels(source: str) -> dict[str, int]:
-    """Local names bound by an import from the owner module, with their levels."""
+def _first_line(lines: dict[str, int], name: str, lineno: int) -> dict[str, int]:
+    lines[name] = min(lineno, lines.get(name, lineno))
+    return lines
+
+
+def owner_import_lines(source: str) -> dict[str, int]:
+    """Local names bound by an import from the owner module, with their lines."""
     bindings: dict[str, int] = {}
     for node in ast.walk(ast.parse(source)):
         if isinstance(node, ast.ImportFrom) and (node.module or "").endswith(OWNER_MODULE):
             for alias in node.names:
-                bindings[alias.asname or alias.name] = node.level
+                bindings = _first_line(bindings, alias.asname or alias.name, node.lineno)
     return bindings
 
 
+def module_assignment_lines(source: str) -> dict[str, int]:
+    """Module-level names assigned a value, whatever shape that value has."""
+    assignments: dict[str, int] = {}
+    for statement in ast.parse(source).body:
+        if isinstance(statement, ast.Assign):
+            targets, value = list(statement.targets), statement.value
+        elif isinstance(statement, ast.AnnAssign):
+            targets, value = [statement.target], statement.value
+        else:
+            continue
+        if value is None:
+            continue
+        for target in targets:
+            if isinstance(target, ast.Name):
+                assignments = _first_line(assignments, target.id, statement.lineno)
+    return assignments
+
+
 def test_the_policy_owner_states_the_reason() -> None:
     assert progress_review_policy.PROGRESS_REVIEW_PENDING_REASON == REASON_VALUE
 
 
 @pytest.mark.parametrize(
     ("module", "local_name"),
     [
-        (receipt, "PROGRESS_REVIEW_PENDING_REASON"),
+        (receipt, OWNER_NAME),
         (external_progress_review, "EXTERNAL_PROGRESS_REVIEW_PENDING_REASON"),
     ],
     ids=["capability-receipt", "external-review"],
 )
-def test_each_consumer_imports_the_owner_name_and_carries_its_value(
+def test_each_consumer_imports_the_owner_name_and_never_rebinds_it(
     module: object, local_name: str
 ) -> None:
     source = Path(module.__file__).read_text(encoding="utf-8")  # type: ignore[attr-defined]
 
-    assert owner_import_levels(source).get(local_name), (
+    assert local_name in owner_import_lines(source), (
         f"{local_name} must be imported from {OWNER_MODULE}, not re-declared locally"
     )
+    assert local_name not in module_assignment_lines(source), (
+        f"{local_name} is assigned again in this module; an equal local value is a"
+        " second owner even while the owner import stays in place"
+    )
     assert getattr(module, local_name) == REASON_VALUE
 
 
+def test_the_census_flags_a_rebind_that_keeps_the_import_and_the_value() -> None:
+    # The counterexample review raised: the import survives, the value still
+    # compares equal, and a literal-keyed census sees nothing because the copy is
+    # built by concatenation.
+    source = (
+        f"from loopx.control_plane.work_items.{OWNER_MODULE} import {OWNER_NAME}\n"
+        f'{OWNER_NAME} = "pending_" + "evaluation"\n'
+    )
+
+    assert owner_import_lines(source) == {OWNER_NAME: 1}
+    assert module_assignment_lines(source) == {OWNER_NAME: 2}
+
+
 def test_no_module_outside_the_owner_holds_the_reason() -> None:
     assert modules_holding_the_reason() == [OWNER_PATH]