refactor(progress-review): read the pending reason from the policy that owns it

loopx/capabilities/progress_review/receipt.py

@@ -18,6 +18,7 @@
 from typing import Any
 from ...control_plane.content_digest import BARE_SHA256_PATTERN
 from ...control_plane.work_items.progress_review_policy import (
+    PROGRESS_REVIEW_PENDING_REASON,
     PROGRESS_REVIEW_SIGNALS as PROGRESS_REVIEW_SIGNAL_KEYS,
 )
 
@@ -44,7 +45,6 @@
     PROGRESS_REVIEW_LEGACY_SIGNAL_RULE_VERSION,
     PROGRESS_REVIEW_SIGNAL_RULE_VERSION,
 )
-PROGRESS_REVIEW_PENDING_REASON = "pending_evaluation"
 MAX_RECEIPT_BYTES = 65536
 MAX_LOADED_RECEIPTS = 256
 _TEXT_LIMIT = 200

loopx/control_plane/work_items/external_progress_review.py

@@ -20,13 +20,13 @@
     progress_observation_from_run,
 )
 from .progress_review_policy import (
+    PROGRESS_REVIEW_PENDING_REASON as EXTERNAL_PROGRESS_REVIEW_PENDING_REASON,
     PROGRESS_REVIEW_SIGNALS as EXTERNAL_PROGRESS_REVIEW_SIGNALS,
 )
 
 EXTERNAL_PROGRESS_REVIEW_TRIGGER_KIND = "external_progress_review_drift"
 EXTERNAL_PROGRESS_REVIEW_TRIGGER_SCHEMA_VERSION = "external_progress_review_trigger_v0"
 EXTERNAL_PROGRESS_REVIEW_FRONTIER_PREFIX = "progress_review:"
-EXTERNAL_PROGRESS_REVIEW_PENDING_REASON = "pending_evaluation"
 # Why a captured transition carries no verdict. None of these is drift and none
 # is progress: they neither form a streak nor dissolve one that already formed.
 EXTERNAL_PROGRESS_REVIEW_UNEVALUATED_REASONS: tuple[str, ...] = (

loopx/control_plane/work_items/progress_review_policy.py

@@ -22,6 +22,7 @@
 PROGRESS_REVIEW_DEFAULT_DRIFT_THRESHOLD = 2
 PROGRESS_REVIEW_MIN_DRIFT_THRESHOLD = 2
 PROGRESS_REVIEW_MAX_DRIFT_THRESHOLD = 20
+PROGRESS_REVIEW_PENDING_REASON = "pending_evaluation"
 
 
 def normalize_progress_review_mode(value: Any) -> str:
@@ -138,6 +139,7 @@ def progress_review_goal_policy_summary(goal: Mapping[str, Any]) -> dict[str, An
     "PROGRESS_REVIEW_MAX_DRIFT_THRESHOLD",
     "PROGRESS_REVIEW_MIN_DRIFT_THRESHOLD",
     "PROGRESS_REVIEW_MODES",
+    "PROGRESS_REVIEW_PENDING_REASON",
     "PROGRESS_REVIEW_POLICY_SCHEMA_VERSION",
     "PROGRESS_REVIEW_SIGNALS",
     "normalize_progress_review_contract_revision",

tests/architecture/test_progress_review_pending_reason_owner.py

@@ -0,0 +1,108 @@
+"""One owner decides the reason a scoped progress-review receipt carries no verdict.
+
+``loopx/control_plane/work_items/progress_review_policy.py`` owns the
+progress-review vocabulary that both the capability receipt and the control-plane
+consumer read, and both already imported ``PROGRESS_REVIEW_SIGNALS`` from it. The
+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.
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+import pytest
+
+from loopx.capabilities.progress_review import receipt
+from loopx.control_plane.work_items import (
+    external_progress_review,
+    progress_review_policy,
+)
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+OWNER_PATH = "loopx/control_plane/work_items/progress_review_policy.py"
+OWNER_NAME = "PROGRESS_REVIEW_PENDING_REASON"
+OWNER_MODULE = "progress_review_policy"
+REASON_VALUE = "pending_evaluation"
+
+
+def reason_assignments(source: str) -> list[str]:
+    """Names bound to the pending-reason literal anywhere in a module."""
+    tree = ast.parse(source)
+    held: list[str] = []
+    for node in ast.walk(tree):
+        if isinstance(node, (ast.Assign, ast.AnnAssign)):
+            targets = (
+                node.targets
+                if isinstance(node, ast.Assign)
+                else [node.target]
+                if node.target is not None
+                else []
+            )
+            for target in targets:
+                if (
+                    isinstance(target, ast.Name)
+                    and isinstance(node.value, ast.Constant)
+                    and node.value.value == REASON_VALUE
+                ):
+                    held.append(target.id)
+    return held
+
+
+def modules_holding_the_reason() -> list[str]:
+    holders: list[str] = []
+    for path in sorted(REPO_ROOT.glob("loopx/**/*.py")):
+        try:
+            held = reason_assignments(path.read_text(encoding="utf-8"))
+        except (SyntaxError, UnicodeDecodeError):  # pragma: no cover - guarded tree
+            continue
+        if held:
+            holders.append(path.relative_to(REPO_ROOT).as_posix())
+    return holders
+
+
+def owner_import_levels(source: str) -> dict[str, int]:
+    """Local names bound by an import from the owner module, with their levels."""
+    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
+    return bindings
+
+
+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"),
+        (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(
+    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), (
+        f"{local_name} must be imported from {OWNER_MODULE}, not re-declared locally"
+    )
+    assert getattr(module, local_name) == REASON_VALUE
+
+
+def test_no_module_outside_the_owner_holds_the_reason() -> None:
+    assert modules_holding_the_reason() == [OWNER_PATH]
+
+
+def test_the_reason_stays_distinct_from_the_unevaluated_vocabulary() -> None:
+    # The receipt reason is input; the classifier projects it onto its own
+    # `"pending"` outcome reason, so the two vocabularies must not merge.
+    assert REASON_VALUE not in (
+        external_progress_review.EXTERNAL_PROGRESS_REVIEW_UNEVALUATED_REASONS
+    )