From 87f9810703487cb11d92f9049b6e80d08c9e548c Mon Sep 17 00:00:00 2001 From: "lusendong.6789" Date: Sat, 31 Oct 2026 21:45:51 +0800 Subject: [PATCH] perf(quota): skip presentation-only status projections on decision reads Quota commands and Turn decisions read collect_status only to derive the should-run decision, yet every read also built the runtime-wide promotion_gate (a scan of every Goal's rollout log for the latest promotion-readiness event) and a per-item goal_channel_projection. Neither is read by quota, scheduling, Todo selection or settlement; only the status and dashboard presentations consume them. Add include_presentation_projections (default True) to collect_status and include_goal_channel_projection to build_attention_queue. Decision callers pass False, and the status projection cache keys that reduced payload separately so loopx status never serves it. The default status output and its existing cache keys are unchanged. Signed-off-by: lusendong.6789 Co-authored-by: TRAE CLI --- docs/status-data-contract.md | 3 + loopx/cli_commands/quota.py | 1 + loopx/cli_commands/quota_context.py | 3 + loopx/cli_commands/quota_monitor_poll.py | 1 + loopx/cli_commands/turn_decision.py | 1 + .../runtime/status_projection_cache.py | 11 +++ loopx/control_plane/status/collection.py | 23 +++-- .../work_items/attention_queue.py | 14 +-- loopx/status.py | 4 + ...test_goal_channel_hard_lease_visibility.py | 50 +++++++++++ .../test_quota_observation_scope.py | 2 + ...s_collection_material_capability_wiring.py | 88 +++++++++++++++++++ 12 files changed, 190 insertions(+), 11 deletions(-) diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index 107234f182..1765599761 100644 --- a/docs/status-data-contract.md +++ b/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 diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 2e3d752737..d56252bf79 100644 --- a/loopx/cli_commands/quota.py +++ b/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 { diff --git a/loopx/cli_commands/quota_context.py b/loopx/cli_commands/quota_context.py index e5f7cd2c30..bb4213b48a 100644 --- a/loopx/cli_commands/quota_context.py +++ b/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"]) diff --git a/loopx/cli_commands/quota_monitor_poll.py b/loopx/cli_commands/quota_monitor_poll.py index d3efa7890f..a303f19eae 100644 --- a/loopx/cli_commands/quota_monitor_poll.py +++ b/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, diff --git a/loopx/cli_commands/turn_decision.py b/loopx/cli_commands/turn_decision.py index 80f19f851e..0bbdaf7ac3 100644 --- a/loopx/cli_commands/turn_decision.py +++ b/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, ) diff --git a/loopx/control_plane/runtime/status_projection_cache.py b/loopx/control_plane/runtime/status_projection_cache.py index a622871080..d79f525d1a 100644 --- a/loopx/control_plane/runtime/status_projection_cache.py +++ b/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) diff --git a/loopx/control_plane/status/collection.py b/loopx/control_plane/status/collection.py index ffe2a9ea5e..1cfb9c8d5f 100644 --- a/loopx/control_plane/status/collection.py +++ b/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", diff --git a/loopx/control_plane/work_items/attention_queue.py b/loopx/control_plane/work_items/attention_queue.py index 9bfbfdf99b..320a2eb59b 100644 --- a/loopx/control_plane/work_items/attention_queue.py +++ b/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( diff --git a/loopx/status.py b/loopx/status.py index 5cabaab6aa..ed06f42c9e 100644 --- a/loopx/status.py +++ b/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(), ) diff --git a/tests/control_plane/test_goal_channel_hard_lease_visibility.py b/tests/control_plane/test_goal_channel_hard_lease_visibility.py index 9f3bd0681e..a1bb8c66ea 100644 --- a/tests/control_plane/test_goal_channel_hard_lease_visibility.py +++ b/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" + } diff --git a/tests/control_plane/test_quota_observation_scope.py b/tests/control_plane/test_quota_observation_scope.py index 7b2a35865f..664d594136 100644 --- a/tests/control_plane/test_quota_observation_scope.py +++ b/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)], diff --git a/tests/control_plane/test_status_collection_material_capability_wiring.py b/tests/control_plane/test_status_collection_material_capability_wiring.py index ab492bc2fe..9ff2c473ff 100644 --- a/tests/control_plane/test_status_collection_material_capability_wiring.py +++ b/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: