refactor(status): single-source shared projection carriers

examples/semantic-vocabulary-drift-smoke.py

@@ -232,16 +232,16 @@
 TWIN_ROOT_ANCHOR = "loopx/control_plane"
 TWIN_BUDGET_ANCHOR = 43
 BUDGET_ANCHOR = {
-    "same_runtime_forks": 12,
-    "same_runtime_fork_definitions": 28,
+    "same_runtime_forks": 11,
+    "same_runtime_fork_definitions": 25,
     "conflicting_values": 16,
     "conflicting_definitions": 55,
     "schema_version_same_runtime_forks": 2,
     "multi_value_twins": 8,
     "multi_value_forks": 2,
     "multi_value_forks_semantic": 1,
     "multi_value_fork_definitions": 6,
-    "same_runtime_forks_semantic": 10,
+    "same_runtime_forks_semantic": 9,
     "conflicting_values_semantic": 0,
 }
 # Budgets for the legacy should-run decision fields, anchored the same way so a

loopx/control_plane/status/attention_projection.py

@@ -21,9 +21,9 @@
     autonomous_backlog_candidates as _autonomous_backlog_candidates,
     autonomous_monitor_candidates as _autonomous_monitor_candidates,
 )
+from .monitor_display_projection import MONITOR_SIGNAL_WAITING_ON
 
 
-MONITOR_SIGNAL_WAITING_ON = "monitor_signal"
 MAX_AUTONOMOUS_BACKLOG_CANDIDATES = MAX_AUTONOMOUS_TODO_CANDIDATES
 
 

loopx/control_plane/status/goal_attention_projection.py

@@ -24,7 +24,6 @@
 )
 from ..work_items.attention_item import attention_item as _attention_item
 from ..work_items.attention_routing import goal_attention as _goal_attention
-from ..work_items.project_asset import build_project_asset
 from .adapter_status_vocabulary import (
     CONNECTED_ADAPTER_STATUSES,
     CONNECTED_DELIVERY_ADAPTER_STATUSES,
@@ -38,10 +37,13 @@
     operator_gate_attention_fields,
     readiness_attention_fields,
 )
+from .monitor_display_projection import (
+    MONITOR_SIGNAL_WAITING_ON,
+    build_project_asset,
+)
 from .run_projection import latest_run
 
 
-MONITOR_SIGNAL_WAITING_ON = "monitor_signal"
 PLANNED_CONTROLLER_OPT_IN_RECOMMENDED_ACTION = (
     "先在 LoopX 完成 operator 判断；同意后项目 Agent 只执行 read-only map dry-run"
 )

loopx/control_plane/status/monitor_display_projection.py

@@ -16,6 +16,9 @@
     open_todo_items,
     todo_item_is_actionable_open,
 )
+from ..work_items.project_asset import (
+    build_project_asset as _build_project_asset,
+)
 
 
 MONITOR_DISPLAY_SCHEMA_VERSION = "monitor_quiet_display_v0"
@@ -29,6 +32,29 @@
 MONITOR_SIGNAL_WAITING_ON = "monitor_signal"
 
 
+def build_project_asset(
+    *,
+    status: str,
+    waiting_on: str,
+    recommended_action: str,
+    operator_question: str | None,
+    agent_command: str | None,
+    missing_gates: list[str] | None,
+    next_handoff_condition: str | None,
+) -> dict[str, Any]:
+    return _build_project_asset(
+        status=status,
+        waiting_on=waiting_on,
+        recommended_action=recommended_action,
+        operator_question=operator_question,
+        agent_command=agent_command,
+        missing_gates=missing_gates,
+        next_handoff_condition=next_handoff_condition,
+        monitor_signal_waiting_on=MONITOR_SIGNAL_WAITING_ON,
+        monitor_display_stop_condition=MONITOR_DISPLAY_STOP_CONDITION,
+    )
+
+
 def todo_summary_open_count(summary: dict[str, Any] | None) -> int:
     return _todo_summary_open_count(
         summary,

loopx/control_plane/status/registry_health_projection.py

@@ -17,12 +17,12 @@
     same_path,
 )
 from ..runtime.time import parse_timestamp
-from ..work_items.project_asset import build_project_asset
 from ..work_items.attention_item import attention_item
 from ..work_items.attention_queue import (
     merge_global_registry_findings as _merge_global_registry_findings,
 )
 from .dreaming_projection import compact_dreaming_lane_badge
+from .monitor_display_projection import build_project_asset
 
 
 SOURCE_REGISTRY_SHADOW_FINDINGS = {

loopx/control_plane/work_items/project_asset.py

@@ -13,10 +13,6 @@
 from .autonomous_replan_obligation import run_history_agent_id
 
 
-DEFAULT_MONITOR_SIGNAL_WAITING_ON = "monitor_signal"
-DEFAULT_MONITOR_DISPLAY_STOP_CONDITION = (
-    "stop until a material monitor transition, regression, or concrete blocker appears"
-)
 TODO_PROJECTION_VIEW_SCHEMA_VERSION = "todo_projection_view_v0"
 TODO_PROJECTION_DETAIL_POINTER_SCHEMA_VERSION = "todo_projection_detail_pointer_v0"
 PROJECT_ASSET_TODO_PROJECTION_GAP_SCHEMA_VERSION = "project_asset_todo_projection_gap_v0"
@@ -34,7 +30,7 @@ def project_asset_public_safe_compact_text(value: Any, *, limit: int = 220) -> s
 def project_asset_owner(
     waiting_on: str,
     *,
-    monitor_signal_waiting_on: str = DEFAULT_MONITOR_SIGNAL_WAITING_ON,
+    monitor_signal_waiting_on: str,
 ) -> str:
     if waiting_on == "codex":
         return "codex"
@@ -55,7 +51,7 @@ def project_asset_gate(
     operator_question: str | None,
     missing_gates: list[str] | None,
     status: str,
-    monitor_signal_waiting_on: str = DEFAULT_MONITOR_SIGNAL_WAITING_ON,
+    monitor_signal_waiting_on: str,
 ) -> str:
     if operator_question:
         return "operator_question"
@@ -75,8 +71,8 @@ def project_asset_stop_condition(
     waiting_on: str,
     next_handoff_condition: str | None,
     agent_command: str | None,
-    monitor_signal_waiting_on: str = DEFAULT_MONITOR_SIGNAL_WAITING_ON,
-    monitor_display_stop_condition: str = DEFAULT_MONITOR_DISPLAY_STOP_CONDITION,
+    monitor_signal_waiting_on: str,
+    monitor_display_stop_condition: str,
 ) -> str:
     if next_handoff_condition:
         return next_handoff_condition
@@ -101,7 +97,7 @@ def project_asset_support_mode(
     status: str,
     recommended_action: str,
     agent_command: str | None,
-    monitor_signal_waiting_on: str = DEFAULT_MONITOR_SIGNAL_WAITING_ON,
+    monitor_signal_waiting_on: str,
 ) -> str:
     surface = " ".join(
         str(value or "")
@@ -411,14 +407,20 @@ def build_project_asset(
     agent_command: str | None,
     missing_gates: list[str] | None,
     next_handoff_condition: str | None,
+    monitor_signal_waiting_on: str,
+    monitor_display_stop_condition: str,
 ) -> dict[str, Any]:
     asset = {
-        "owner": project_asset_owner(waiting_on),
+        "owner": project_asset_owner(
+            waiting_on,
+            monitor_signal_waiting_on=monitor_signal_waiting_on,
+        ),
         "gate": project_asset_gate(
             waiting_on=waiting_on,
             operator_question=operator_question,
             missing_gates=missing_gates,
             status=status,
+            monitor_signal_waiting_on=monitor_signal_waiting_on,
         ),
         "support_mode": project_asset_support_mode(
             waiting_on=waiting_on,
@@ -427,12 +429,15 @@ def build_project_asset(
             status=status,
             recommended_action=recommended_action,
             agent_command=agent_command,
+            monitor_signal_waiting_on=monitor_signal_waiting_on,
         ),
         "next_action": recommended_action,
         "stop_condition": project_asset_stop_condition(
             waiting_on=waiting_on,
             next_handoff_condition=next_handoff_condition,
             agent_command=agent_command,
+            monitor_signal_waiting_on=monitor_signal_waiting_on,
+            monitor_display_stop_condition=monitor_display_stop_condition,
         ),
     }
     next_safe_command = project_asset_next_safe_command(agent_command)

loopx/diagnose.py

@@ -8,6 +8,7 @@
     GENERIC_CLI_OUTER_CONTROLLER_SCHEDULER_CONTEXT,
     render_scheduler_execution_args,
 )
+from .control_plane.status.contract_projection import STATUS_CONTRACT_SIGNAL_LIMIT
 from .control_plane.goals.orphaned_goal_state import (
     ORPHANED_GOAL_STATE_CONNECTION,
     ORPHANED_GOAL_STATE_REASON,
@@ -25,7 +26,6 @@
 
 DIAGNOSIS_SCHEMA_VERSION = "loopx_agent_diagnosis_packet_v0"
 PACKET_KIND = "agent_reasoning_evidence_packet"
-STATUS_CONTRACT_SIGNAL_LIMIT = 3
 ORPHANED_GOAL_STATE_RECOMMENDED_ACTION = (
     "inspect the listed orphaned state and preview its backup; do not bootstrap "
     "until an operator explicitly resolves it"

loopx/semantics/vocabulary_v0.json

@@ -1126,15 +1126,15 @@
   },
   "inventory_ratchets": {
     "meaning": "Counts read from the generated inventory. A same-runtime fork is one constant name with one value defined in two or more modules of the same runtime; a conflicting value is one name with different values. Both the number of affected names and the number of definitions are budgets, so a third spelling of an already-conflicting name is still a regression.",
-    "same_runtime_forks": 12,
-    "same_runtime_fork_definitions": 28,
+    "same_runtime_forks": 11,
+    "same_runtime_fork_definitions": 25,
     "conflicting_values": 16,
     "conflicting_definitions": 55,
     "schema_version_same_runtime_forks": 2,
     "multi_value_twins": 8,
     "multi_value_forks": 2,
     "multi_value_fork_definitions": 6,
-    "same_runtime_forks_semantic": 10,
+    "same_runtime_forks_semantic": 9,
     "conflicting_values_semantic": 0,
     "multi_value_meaning": "Enums, named closed sets, Literal aliases, and TypeScript as-const arrays are vocabulary exactly as a NAME = \"value\" constant is, so they get the same collision rule. One name defined in two modules with identical values is a twin; with different values it is a fork. The semantic multi-value-fork budget excludes only names declared in scope_declarations.",
     "multi_value_forks_note": "The 4 counted forks include SOURCE_SURFACES, whose four definitions are four CLI commands each listing its own data sources; that is bounded-context reuse of one name, not drift. It stays in the budget until M0.5 adds a scope field (RFC Section 5) and must not be removed by renaming.",

loopx/state_projection.py

@@ -4,6 +4,7 @@
 from typing import Any
 
 from .control_plane.goals.active_state_metadata import todo_role_for_heading
+from .control_plane.status.active_state_projection import SECTION_HEADING_PATTERN
 from .control_plane.todos.contract import (
     TODO_TASK_PATTERN,
     build_todo_id,
@@ -43,7 +44,6 @@
 ACTIVE_STATE_PROJECTION_DIAGNOSTICS_SCHEMA_VERSION = "active_state_projection_diagnostics_v0"
 TODO_ITEM_SCHEMA_VERSION = "todo_item_v0"
 
-SECTION_HEADING_PATTERN = re.compile(r"^##+\s+(.+?)\s*$")
 BULLET_PATTERN = re.compile(r"^\s*(?:[-*]|\d+[.)])\s+(.+?)\s*$")
 PRIORITY_PATTERN = re.compile(r"^\[(P[0-4])\]\s+(.+)$", re.IGNORECASE)
 NEXT_ACTION_EXECUTABLE_PATTERN = re.compile(

loopx/status.py

@@ -62,6 +62,7 @@
     MONITOR_DISPLAY_SCHEMA_VERSION as MONITOR_DISPLAY_SCHEMA_VERSION,
     MONITOR_DISPLAY_STOP_CONDITION as MONITOR_DISPLAY_STOP_CONDITION,
     MONITOR_SIGNAL_WAITING_ON,
+    build_project_asset,
 )
 from .control_plane.status.registry_health_projection import (
     SOURCE_REGISTRY_SHADOW_FINDINGS,
@@ -106,7 +107,6 @@
     TODO_PROJECTION_DETAIL_POINTER_SCHEMA_VERSION as TODO_PROJECTION_DETAIL_POINTER_SCHEMA_VERSION,
     TODO_PROJECTION_VIEW_SCHEMA_VERSION as TODO_PROJECTION_VIEW_SCHEMA_VERSION,
     attach_active_state_project_asset_fields as _attach_active_state_project_asset_fields,
-    build_project_asset,
     enrich_project_asset as _enrich_project_asset_read_model,
     project_asset_handoff_check_projection,
     project_asset_latest_validation,

tests/control_plane/test_status_facade_vocabulary_owners.py

@@ -1,20 +1,17 @@
-"""Refs #4447: the status facade stops restating what its own projections own.
+"""Refs #4447: status consumers stop restating what their projections own.
 
 `loopx/status.py` is the public facade over `loopx.control_plane.status`, and it had
 grown its own copy of fifteen module-level carriers that a projection already defined
 with an identical value: the three status-contract schema numbers, the monitor display
 schema version, the goal-attention override set and legacy evidence prefixes, and the
 active-state, autonomous-replan, dead-monitor, backlog-hygiene and agent-lane carriers.
 
-Each of those values feeds an injected parameter of a read model inside the owning
-projection, so the projection is the owner and the facade's job is only to keep
-exporting the established name. Before this change the two copies were free to
-diverge: an edit on one side changed the read model and left `loopx.status` reporting
-the other literal, and the generated inventory counted every such pair as a duplicate
-it had to carry in a budget.
+The same follow-up applies to status consumers outside that facade. Each value feeds
+the same read-model contract as its owning projection, so consumers import the owner
+instead of maintaining an equal-looking declaration.
 
 This test pins three things: the facade declares none of these names itself, every
-facade binding is the owner's object rather than an equal-looking re-typed literal,
+importing binding is the owner's object rather than an equal-looking re-typed literal,
 and the merged values are what the projections shipped with.
 """
 
@@ -25,9 +22,12 @@
 import inspect
 from pathlib import Path
 
-from loopx import status
+import pytest
+
+from loopx import diagnose, state_projection, status
 from loopx.control_plane.status import (
     active_state_projection,
+    attention_projection,
     autonomous_replan_projection,
     contract_projection,
     goal_attention_projection,
@@ -41,8 +41,7 @@
 FACADE = "loopx/status.py"
 
 # name -> (module holding the one remaining definition, other modules importing it).
-# These names were shared by the facade and exactly this owner, so after the merge the
-# owner is the only declarer left in the package (twelve of the fifteen).
+# Each name has one owner and any public or internal consumers import that object.
 FULLY_MERGED: dict[str, tuple[object, tuple[object, ...]]] = {
     "AUTONOMOUS_REPLAN_PERIODIC_RUN_THRESHOLD": (
         autonomous_replan_projection,
@@ -63,8 +62,20 @@
         (status,),
     ),
     "MONITOR_DISPLAY_SCHEMA_VERSION": (monitor_display_projection, (status,)),
+    "MONITOR_SIGNAL_WAITING_ON": (
+        monitor_display_projection,
+        (attention_projection, goal_attention_projection, status),
+    ),
     "REGISTRY_WAITING_ON_OVERRIDES": (goal_attention_projection, (status,)),
+    "SECTION_HEADING_PATTERN": (
+        active_state_projection,
+        (state_projection, status),
+    ),
     "STATUS_CONTRACT_SCHEMA_VERSION": (contract_projection, (status,)),
+    "STATUS_CONTRACT_SIGNAL_LIMIT": (
+        contract_projection,
+        (diagnose, status),
+    ),
 }
 
 # The facade copy is gone here too, but these names are still restated by modules
@@ -77,21 +88,11 @@
         "loopx/history.py",
         "loopx/state_refresh.py",
     ],
-    "SECTION_HEADING_PATTERN": [
-        "loopx/control_plane/status/active_state_projection.py",
-        "loopx/state_projection.py",
-    ],
-    "STATUS_CONTRACT_SIGNAL_LIMIT": [
-        "loopx/control_plane/status/contract_projection.py",
-        "loopx/diagnose.py",
-    ],
 }
 
 OWNERSHIPS: dict[str, tuple[object, tuple[object, ...]]] = {
     **FULLY_MERGED,
     "AGENT_LANE_PROGRESS_SCOPE": (run_projection, (status,)),
-    "SECTION_HEADING_PATTERN": (active_state_projection, (status,)),
-    "STATUS_CONTRACT_SIGNAL_LIMIT": (contract_projection, (status,)),
 }
 
 # The values those carriers already had in the projections before the merge.
@@ -108,6 +109,7 @@
     ),
     "MINIMUM_DASHBOARD_STATUS_CONTRACT_SCHEMA_VERSION": 2,
     "MONITOR_DISPLAY_SCHEMA_VERSION": "monitor_quiet_display_v0",
+    "MONITOR_SIGNAL_WAITING_ON": "monitor_signal",
     "REGISTRY_WAITING_ON_OVERRIDES": {
         "user_or_controller",
         "controller",
@@ -204,6 +206,36 @@ def test_merging_the_copies_changed_no_value() -> None:
         assert getattr(OWNERSHIPS[name][0], name).pattern == expected, name
 
 
+def test_monitor_project_asset_uses_the_projection_owner(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monitor_signal = "changed_monitor_signal"
+    stop_condition = "stop for the changed monitor signal"
+    monkeypatch.setattr(
+        monitor_display_projection,
+        "MONITOR_SIGNAL_WAITING_ON",
+        monitor_signal,
+    )
+    monkeypatch.setattr(
+        monitor_display_projection,
+        "MONITOR_DISPLAY_STOP_CONDITION",
+        stop_condition,
+    )
+
+    item = goal_attention_projection.attention_item(
+        goal_id="goal-a",
+        status="waiting",
+        waiting_on=monitor_signal,
+        severity="info",
+        recommended_action="wait for monitor evidence",
+        source="monitor",
+        agent_command="loopx monitor poll",
+    )
+
+    assert item["project_asset"]["support_mode"] == "read_only_observer"
+    assert item["project_asset"]["stop_condition"] == stop_condition
+
+
 def test_the_lifecycle_priority_pair_is_left_in_place() -> None:
     """Deliberate exclusion, recorded so a later reader sees it was a decision.