perf(quota): skip presentation-only status projections on decision reads

docs/status-data-contract.md

@@ -51,6 +51,9 @@ todos, quota, soft claims, compact run events, source warnings, and an explicit
 truth contract. It is not a write API. Dashboards may render it as a channel
 card or timeline, but project truth still comes from the registry, active
 state, quota guard, and append-only run history.
+Decision reads (`loopx quota ...` and Turn decisions) omit this row field and
+the top-level `promotion_gate`; only display consumers read them, and their
+status projection cache entries are keyed separately from `loopx status`.
 
 Rows may also include an optional `task_graph_projection` object with
 `schema_version=task_graph_projection_v0`. The default `status` hot path omits

loopx/cli_commands/quota.py

@@ -681,6 +681,7 @@ def handle_quota_command(
                     limit=status_limit,
                     goal_id=status_goal_id,
                     available_capabilities=args.available_capabilities,
+                    include_presentation_projections=False,
                 ),
             )
         elif args.quota_command in {

loopx/cli_commands/quota_context.py

@@ -314,6 +314,7 @@ def prepare_quota_command_context(
             max_age_seconds=projection_cache_ttl_seconds,
             available_capabilities=args.available_capabilities,
             agent_lane_id=args.agent_id,
+            include_presentation_projections=False,
         )
         if (
             status_payload is not None
@@ -344,6 +345,7 @@ def prepare_quota_command_context(
             goal_id=status_goal_id,
             available_capabilities=args.available_capabilities,
             agent_lane_id=args.agent_id,
+            include_presentation_projections=False,
         )
         if bool(getattr(args, "write_projection_cache", False)):
             cache_metadata = write_status_projection_cache(
@@ -356,6 +358,7 @@ def prepare_quota_command_context(
                 max_age_seconds=projection_cache_ttl_seconds,
                 available_capabilities=args.available_capabilities,
                 agent_lane_id=args.agent_id,
+                include_presentation_projections=False,
             )
     elif isinstance(status_payload.get("projection_cache"), dict):
         cache_metadata = dict(status_payload["projection_cache"])

loopx/cli_commands/quota_monitor_poll.py

@@ -226,6 +226,7 @@ def record_automatic_heartbeat_stall(
         limit=context.status_limit,
         goal_id=context.status_goal_id,
         available_capabilities=args.available_capabilities,
+        include_presentation_projections=False,
     )
     rebuilt = decision_builder(
         reloaded,

loopx/cli_commands/turn_decision.py

@@ -126,6 +126,7 @@ def collect_turn_status_payload(
         goal_id=args.goal_id,
         available_capabilities=args.available_capabilities,
         agent_lane_id=args.agent_id,
+        include_presentation_projections=False,
     )
 
 

loopx/control_plane/runtime/status_projection_cache.py

@@ -59,6 +59,7 @@ def status_projection_cache_key(
     goal_id: str | None,
     available_capabilities: Any = None,
     agent_lane_id: str | None = None,
+    include_presentation_projections: bool = True,
 ) -> str:
     request = {
         "schema_version": STATUS_PROJECTION_CACHE_SCHEMA_VERSION,
@@ -73,6 +74,10 @@ def status_projection_cache_key(
             available_capabilities
         ),
     }
+    if not include_presentation_projections:
+        # Decision-only payloads omit presentation projections; keep them out of
+        # the default status key so a display read never serves a reduced view.
+        request["presentation_projections"] = False
     encoded = json.dumps(
         request,
         ensure_ascii=True,
@@ -136,6 +141,7 @@ def status_projection_cache_metadata(
     max_age_seconds: int,
     available_capabilities: Any = None,
     agent_lane_id: str | None = None,
+    include_presentation_projections: bool = True,
 ) -> dict[str, Any]:
     key = status_projection_cache_key(
         registry_path=registry_path,
@@ -145,6 +151,7 @@ def status_projection_cache_metadata(
         goal_id=goal_id,
         available_capabilities=available_capabilities,
         agent_lane_id=agent_lane_id,
+        include_presentation_projections=include_presentation_projections,
     )
     normalized_capabilities = _normalized_available_capabilities(
         available_capabilities
@@ -172,6 +179,7 @@ def load_status_projection_cache(
     max_age_seconds: int,
     available_capabilities: Any = None,
     agent_lane_id: str | None = None,
+    include_presentation_projections: bool = True,
 ) -> tuple[dict[str, Any] | None, dict[str, Any]]:
     metadata = status_projection_cache_metadata(
         registry_path=registry_path,
@@ -182,6 +190,7 @@ def load_status_projection_cache(
         max_age_seconds=max_age_seconds,
         available_capabilities=available_capabilities,
         agent_lane_id=agent_lane_id,
+        include_presentation_projections=include_presentation_projections,
     )
     path = Path(str(metadata["path"]))
     metadata["hit"] = False
@@ -246,6 +255,7 @@ def write_status_projection_cache(
     max_age_seconds: int,
     available_capabilities: Any = None,
     agent_lane_id: str | None = None,
+    include_presentation_projections: bool = True,
 ) -> dict[str, Any]:
     metadata = status_projection_cache_metadata(
         registry_path=registry_path,
@@ -256,6 +266,7 @@ def write_status_projection_cache(
         max_age_seconds=max_age_seconds,
         available_capabilities=available_capabilities,
         agent_lane_id=agent_lane_id,
+        include_presentation_projections=include_presentation_projections,
     )
     path = Path(str(metadata["path"]))
     path.parent.mkdir(parents=True, exist_ok=True)

loopx/control_plane/status/collection.py

@@ -88,7 +88,14 @@ def collect_status(
     activation_state_filter: GoalActivationState | str | None = None,
     agent_lane_id: str | None = None,
     allow_missing_registry: bool = False,
+    include_presentation_projections: bool = True,
 ) -> dict[str, Any]:
+    """Collect the status read model.
+
+    ``include_presentation_projections=False`` omits the runtime-wide
+    ``promotion_gate`` and per-item ``goal_channel_projection``, which only the
+    status/dashboard presentations read; decision callers such as quota use it.
+    """
     display_limit = max(0, limit)
     control_plane_limit = max(
         display_limit,
@@ -179,6 +186,7 @@ def canonical_runtime_for_goal(current_goal_id: str) -> Path:
         events_for_goal=rollout_events.events_for_goal,
         current_registry=registry,
         registry_path=registry_path,
+        include_goal_channel_projection=include_presentation_projections,
     )
     # No later projection consumes canonical rows; release retained archive
     # data before assembling the rest of the display.
@@ -197,10 +205,14 @@ def canonical_runtime_for_goal(current_goal_id: str) -> Path:
         ),
         events_for_goal=rollout_events.events_for_goal,
     )
-    promotion_gate = context.build_promotion_gate(
-        registry_path=registry_path,
-        runtime_root_override=str(runtime_root),
-        registry=registry,
+    promotion_gate = (
+        context.build_promotion_gate(
+            registry_path=registry_path,
+            runtime_root_override=str(runtime_root),
+            registry=registry,
+        )
+        if include_presentation_projections
+        else None
     )
     zcode_eligibility: dict[str, list[str]] = {}
     source_goal_observer: Callable[[str, dict[str, Any] | None], None] | None = None
@@ -264,8 +276,9 @@ def observe_source_goal(current_goal_id: str, source_goal: dict[str, Any] | None
         "global_registry": global_registry,
         "attention_queue": queue,
         **runtime_summaries,
-        "promotion_gate": promotion_gate,
     }
+    if include_presentation_projections:
+        payload["promotion_gate"] = promotion_gate
     if activation_filter is not None:
         payload["goal_projection"] = {
             "schema_version": "loopx_goal_projection_scope_v0",

loopx/control_plane/work_items/attention_queue.py

@@ -189,6 +189,7 @@ def build_attention_queue(
     include_task_graph: bool = False,
     goal_id_filter: str | None = None,
     include_stopped_goal_context: bool = False,
+    include_goal_channel_projection: bool = True,
 ) -> dict[str, Any]:
     """Build the executable attention queue or a stopped-Goal archive view.
 
@@ -380,12 +381,13 @@ def build_attention_queue(
                 )
                 if task_graph_projection:
                     item["task_graph_projection"] = task_graph_projection
-            context.attach_goal_channel_projection(
-                item,
-                goal=goal,
-                goal_latest_runs=goal_latest_runs,
-                runtime_root=runtime_root,
-            )
+            if include_goal_channel_projection:
+                context.attach_goal_channel_projection(
+                    item,
+                    goal=goal,
+                    goal_latest_runs=goal_latest_runs,
+                    runtime_root=runtime_root,
+                )
             history_items.append(item)
 
     merge_global_registry_findings(

loopx/status.py

@@ -1099,6 +1099,7 @@ def build_attention_queue(
     todo_snapshot: _CanonicalTodoSnapshot | None = None,
     current_registry: dict[str, Any] | None = None,
     registry_path: Path | None = None,
+    include_goal_channel_projection: bool = True,
 ) -> dict[str, Any]:
     def request_active_state_todo_fields(
         goal: dict[str, Any],
@@ -1166,6 +1167,7 @@ def request_active_state_todo_fields(
         include_task_graph=include_task_graph,
         goal_id_filter=goal_id_filter,
         include_stopped_goal_context=include_stopped_goal_context,
+        include_goal_channel_projection=include_goal_channel_projection,
     )
     if runtime_root is not None:
         for item in queue.get("items") or []:
@@ -1363,6 +1365,7 @@ def collect_status(
     activation_state_filter: str | None = None,
     agent_lane_id: str | None = None,
     allow_missing_registry: bool = False,
+    include_presentation_projections: bool = True,
 ) -> dict[str, Any]:
     with effect_runtime_request_scope():
         return _collect_status_read_model(
@@ -1380,5 +1383,6 @@ def collect_status(
             include_zcode_goal_eligibility=include_zcode_goal_eligibility,
             activation_state_filter=activation_state_filter,
             agent_lane_id=agent_lane_id,
+            include_presentation_projections=include_presentation_projections,
             context=build_status_collection_context(),
         )

tests/control_plane/test_goal_channel_hard_lease_visibility.py

@@ -410,3 +410,53 @@ def test_status_attention_queue_surfaces_hard_lease_end_to_end(
             "claimed_by": "agent-b",
         }
     ]
+
+
+def test_decision_status_read_omits_only_presentation_projections(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The real collector's decision read keeps every attention-item field."""
+
+    from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
+    from loopx.control_plane.todos.todo_summary import canonical_todo_read_record, structured_todo_item
+    from loopx.status import collect_status
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    registry, repo, runtime_root = _write_status_workspace(tmp_path)
+    lease = _write_active_lease(runtime_root, owner="agent-a")
+    initialize_canonical_authority(
+        runtime_root, GOAL_ID,
+        build_todo_runtime_shadow_projection(
+            goal_id=GOAL_ID, handoff_mode="hard_lease", leases=[lease],
+            todos=[canonical_todo_read_record(structured_todo_item({
+                "todo_id": TODO_ID, "role": "agent", "status": "open",
+                "text": "Deliver one bounded control-plane change.",
+                "task_class": "advancement_task", "claimed_by": "agent-b",
+            }, role="agent", source_section="Agent Todo"))],
+        ),
+        state_path=repo / "ACTIVE_GOAL_STATE.md", provider="sqlite",
+    )
+
+    def read(**kwargs: Any) -> dict[str, Any]:
+        return collect_status(
+            registry_path=registry, runtime_root_override=None, limit=5,
+            goal_id=GOAL_ID, **kwargs,
+        )
+
+    full = read()
+    decision = read(include_presentation_projections=False)
+
+    assert "promotion_gate" in full
+    assert "promotion_gate" not in decision
+    full_items = full["attention_queue"]["items"]
+    decision_items = decision["attention_queue"]["items"]
+    assert [item.get("goal_id") for item in decision_items] == [
+        item.get("goal_id") for item in full_items
+    ]
+    for full_item, decision_item in zip(full_items, decision_items):
+        assert "goal_channel_projection" not in decision_item
+        assert decision_item == {
+            key: value
+            for key, value in full_item.items()
+            if key != "goal_channel_projection"
+        }

tests/control_plane/test_quota_observation_scope.py

@@ -113,6 +113,8 @@ def test_scope_reaches_collector_and_cache_identity(tmp_path, mode, goal_id):
 
     def collect(**kwargs):
         calls.append(kwargs["goal_id"])
+        # Quota decisions never read the status presentation projections.
+        assert kwargs["include_presentation_projections"] is False
         now = now_utc_iso()
         envelope = seal_projection_envelope(projection="status", observed_at=now,
             sources=[source_fact("registry", last_read_at=now)],

tests/control_plane/test_status_collection_material_capability_wiring.py

@@ -286,6 +286,94 @@ def test_stopped_scope_filters_goal_channel_rows_and_requests_archive_context(
     ]
 
 
+def _collect_recording_presentation(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    **collect_kwargs: Any,
+) -> tuple[dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
+    monkeypatch.setattr(
+        status_collection,
+        "collect_runtime_projection_route_diagnostics",
+        lambda **_kwargs: {"available": False},
+    )
+    queue_calls: list[dict[str, Any]] = []
+    gate_calls: list[dict[str, Any]] = []
+    context = _context(tmp_path)
+    base_builder = context.build_attention_queue
+    context = status_collection.StatusCollectionContext(
+        **{
+            **context.__dict__,
+            "build_attention_queue": lambda **kwargs: (
+                queue_calls.append(kwargs),
+                base_builder(**kwargs),
+            )[1],
+            "build_promotion_gate": lambda **kwargs: (
+                gate_calls.append(kwargs),
+                {"gate_state": "fixture"},
+            )[1],
+        }
+    )
+    payload = status_collection.collect_status(
+        registry_path=tmp_path / "registry.json",
+        runtime_root_override=None,
+        limit=5,
+        context=context,
+        goal_id=GOAL_ID,
+        **collect_kwargs,
+    )
+    return payload, queue_calls, gate_calls
+
+
+def test_status_collection_builds_presentation_projections_by_default(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    payload, queue_calls, gate_calls = _collect_recording_presentation(
+        tmp_path, monkeypatch
+    )
+
+    assert queue_calls[0]["include_goal_channel_projection"] is True
+    assert len(gate_calls) == 1
+    assert payload["promotion_gate"] == {"gate_state": "fixture"}
+
+
+def test_decision_status_read_skips_presentation_projections(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    payload, queue_calls, gate_calls = _collect_recording_presentation(
+        tmp_path, monkeypatch, include_presentation_projections=False
+    )
+
+    assert queue_calls[0]["include_goal_channel_projection"] is False
+    assert gate_calls == []
+    assert "promotion_gate" not in payload
+    # Decision inputs stay in the reduced payload.
+    assert payload["attention_queue"] == {"items": [], "item_count": 0}
+    assert payload["agent_management_projection"]["agents"]
+
+
+def test_status_projection_cache_separates_presentation_scope(
+    tmp_path: Path,
+) -> None:
+    kwargs = {
+        "registry_path": tmp_path / "registry.json",
+        "runtime_root": tmp_path / "runtime",
+        "limit": 5,
+        "include_task_graph": False,
+        "goal_id": GOAL_ID,
+    }
+
+    default_key = status_projection_cache_key(**kwargs)
+
+    assert default_key == status_projection_cache_key(
+        **kwargs, include_presentation_projections=True
+    )
+    assert default_key != status_projection_cache_key(
+        **kwargs, include_presentation_projections=False
+    )
+
+
 def test_status_projection_cache_separates_capability_envelopes(
     tmp_path: Path,
 ) -> None: