From c116c720f669885c61cc16710020d80aea7da7b9 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 00:55:28 -0400 Subject: [PATCH 1/8] refactor(quota): construct settled replay without action cleanup Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- .../quota/settlement_precedence.py | 163 +++--------- loopx/control_plane/quota/should_run.py | 18 +- .../control_plane/quota/should_run_packet.py | 233 ++++++++++++------ loopx/control_plane/todos/user_gate.py | 21 +- loopx/semantics/vocabulary_v0.json | 3 +- .../test_settled_replay_construction.py | 137 ++++++++++ 6 files changed, 352 insertions(+), 223 deletions(-) create mode 100644 tests/control_plane/test_settled_replay_construction.py diff --git a/loopx/control_plane/quota/settlement_precedence.py b/loopx/control_plane/quota/settlement_precedence.py index f1cf97b179..a50241ad9b 100644 --- a/loopx/control_plane/quota/settlement_precedence.py +++ b/loopx/control_plane/quota/settlement_precedence.py @@ -1,9 +1,8 @@ from __future__ import annotations from .effective_action import EffectiveAction -from typing import Any, Protocol +from typing import Any -from ..effect_program import ReceiptBoundReplayPhase HEARTBEAT_SETTLED_REPLAY_REASON = ( @@ -40,66 +39,6 @@ ) -class SettledReplayRoute(Protocol): - normal_delivery_allowed: bool - recovery_allowed: bool - self_repair_allowed: bool - capability_repair_allowed: bool - workspace_repair_allowed: bool - should_run: bool - effective_action: str - reason: str - replan_decision_allowed: bool - receipt_bound_replan_decision: bool - heartbeat_recommendation: dict[str, Any] - external_evidence_observation: dict[str, Any] | None - external_evidence_observation_recent: dict[str, Any] | None - selected_recommended_action: Any - agent_lane_next_action: dict[str, Any] | None - agent_scope_frontier: dict[str, Any] | None - agent_lane_frontier_hint: dict[str, Any] | None - state_action_projection_warning: dict[str, Any] | None - next_action_warning: dict[str, Any] | None - goal_route_hint: dict[str, Any] | None - - -def apply_settled_replay_route_precedence( - route: SettledReplayRoute, - *, - replay_phase: ReceiptBoundReplayPhase | None, -) -> None: - """Prevent a settled heartbeat identity from selecting work in the same turn.""" - - if replay_phase is not ReceiptBoundReplayPhase.SETTLED: - return - route.normal_delivery_allowed = False - route.recovery_allowed = False - route.self_repair_allowed = False - route.capability_repair_allowed = False - route.workspace_repair_allowed = False - route.should_run = False - route.effective_action = EffectiveAction.HEARTBEAT_SETTLED_SKIP.value - route.reason = HEARTBEAT_SETTLED_REPLAY_REASON - route.replan_decision_allowed = False - route.receipt_bound_replan_decision = False - route.heartbeat_recommendation = { - **route.heartbeat_recommendation, - "recommended_mode": route.effective_action, - "notify": "DONT_NOTIFY", - "reason": route.reason, - "spend_policy": "no quota spend for an already-settled heartbeat turn", - } - route.external_evidence_observation = None - route.external_evidence_observation_recent = None - route.selected_recommended_action = route.reason - route.agent_lane_next_action = None - route.agent_scope_frontier = None - route.agent_lane_frontier_hint = None - route.state_action_projection_warning = None - route.next_action_warning = None - route.goal_route_hint = None - - def clear_quota_action_projections( payload: dict[str, Any], *, @@ -109,72 +48,38 @@ def clear_quota_action_projections( payload.pop(key, None) -def apply_settled_replay_payload_precedence( - payload: dict[str, Any], - *, - replay_phase: ReceiptBoundReplayPhase | None, -) -> None: - """Keep late supporting projections from reopening a settled heartbeat turn.""" - - if replay_phase is not ReceiptBoundReplayPhase.SETTLED: - return +def settled_replay_fields() -> dict[str, Any]: + """Construct the authority fields of a verified, already-settled Turn.""" reason = HEARTBEAT_SETTLED_REPLAY_REASON - payload.update( - { - "decision": "skip", - "should_run": False, - "normal_delivery_allowed": False, - "recovery_delivery_allowed": False, - "self_repair_allowed": False, - "capability_repair_allowed": False, - "workspace_repair_allowed": False, - "effective_action": EffectiveAction.HEARTBEAT_SETTLED_SKIP.value, - "actionable_by_codex": False, - "reason": reason, - "requires_user_action": False, - "recommended_action": ( - "Finish this heartbeat without another action; use a fresh turn " - "identity for successor selection." - ), - "heartbeat_recommendation": { - "recommended_mode": "heartbeat_settled_skip", - "notify": "DONT_NOTIFY", - "reason": reason, - "spend_policy": "no quota spend for an already-settled heartbeat turn", - "agent_must_attempt": False, - }, - "execution_obligation": { - "must_attempt_work": False, - "kind": "heartbeat_settled_skip", - "delivery_allowed": False, - "notify_is_execution_gate": False, - "reason": reason, - "spend_policy": "no quota spend for an already-settled heartbeat turn", - }, - } - ) - frontier = payload.get("goal_frontier_projection") - if isinstance(frontier, dict): - payload["goal_frontier_projection"] = { - key: value - for key, value in frontier.items() - if key - not in { - "autonomous_replan_decision", - "replan_ack_feedback", - "replan_obligation", - "vision_continuation_audit", - "vision_wait_state", - } - } - clear_quota_action_projections( - payload, - additional_keys=( - "boundary_projection_gap", - "operator_question", - "selected_todo", - "state_projection_gap", - "task_orchestration_contract", - "todo_write_hint", + return { + "decision": "skip", + "should_run": False, + "normal_delivery_allowed": False, + "recovery_delivery_allowed": False, + "self_repair_allowed": False, + "capability_repair_allowed": False, + "workspace_repair_allowed": False, + "effective_action": EffectiveAction.HEARTBEAT_SETTLED_SKIP.value, + "actionable_by_codex": False, + "reason": reason, + "requires_user_action": False, + "recommended_action": ( + "Finish this heartbeat without another action; use a fresh turn " + "identity for successor selection." ), - ) + "heartbeat_recommendation": { + "recommended_mode": "heartbeat_settled_skip", + "notify": "DONT_NOTIFY", + "reason": reason, + "spend_policy": "no quota spend for an already-settled heartbeat turn", + "agent_must_attempt": False, + }, + "execution_obligation": { + "must_attempt_work": False, + "kind": "heartbeat_settled_skip", + "delivery_allowed": False, + "notify_is_execution_gate": False, + "reason": reason, + "spend_policy": "no quota spend for an already-settled heartbeat turn", + }, + } diff --git a/loopx/control_plane/quota/should_run.py b/loopx/control_plane/quota/should_run.py index b938f51d0e..e6424871a5 100644 --- a/loopx/control_plane/quota/should_run.py +++ b/loopx/control_plane/quota/should_run.py @@ -45,7 +45,6 @@ ReceiptBoundReplayPhase, ReceiptBoundTerminalPhase, ) -from .settlement_precedence import apply_settled_replay_route_precedence from .should_run_packet import ( _build_quota_should_run_payload, _execution_obligation, @@ -62,17 +61,6 @@ GOAL_STOPPED_MODE = "goal_stopped" -def _resolve_quota_route_with_settled_replay_precedence( - prepared: _QuotaDecisionPreparation, -) -> _QuotaDecisionRoute: - route = _resolve_quota_should_run_route(prepared) - apply_settled_replay_route_precedence( - route, - replay_phase=prepared.receipt_bound_replay_phase, - ) - return route - - def _apply_selected_todo_guards( prepared: _QuotaDecisionPreparation, route: _QuotaDecisionRoute, @@ -112,7 +100,7 @@ def _apply_selected_todo_guards( "Goal acceptance holds the current work; inspect its contract and " "ask the owner to configure or rebind the current Todo." ) - route = _resolve_quota_route_with_settled_replay_precedence(prepared) + route = _resolve_quota_should_run_route(prepared) workspace_guard = None if not prepared.inbox_priority_due: workspace_guard = build_agent_workspace_guard( @@ -143,7 +131,7 @@ def _apply_selected_todo_guards( prepared.reason = str( boundary_projection_repair.get("reason") or prepared.reason ) - return _resolve_quota_route_with_settled_replay_precedence(prepared) + return _resolve_quota_should_run_route(prepared) def build_quota_paused_should_run_payload( @@ -352,7 +340,7 @@ def build_quota_should_run( receipt_bound_replay_phase=receipt_bound_replay_phase, receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id, ) - route = _resolve_quota_route_with_settled_replay_precedence(prepared) + route = _resolve_quota_should_run_route(prepared) route = _apply_selected_todo_guards(prepared, route) return _build_quota_should_run_payload( prepared, diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 631a5c7619..ac7045c0b3 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -1,5 +1,6 @@ from __future__ import annotations from .effective_action import EffectiveAction +from .effect_program import ReceiptBoundReplayPhase from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path @@ -100,6 +101,7 @@ ) from ..todos.user_gate import ( apply_scoped_user_gate_fallback_projection as _apply_scoped_user_gate_fallback_projection, + scoped_user_gate_fallback_fields, ) from ..todos.user_gate import ( build_gate_prompt as _build_gate_prompt, @@ -133,7 +135,8 @@ work_lane_contract_is_receipt_bound_monitor_settled, ) from .settlement_precedence import ( - apply_settled_replay_payload_precedence, + settled_replay_fields, + HEARTBEAT_SETTLED_REPLAY_REASON, clear_quota_action_projections, ) from .should_run_prepare import _QuotaDecisionPreparation @@ -465,6 +468,7 @@ def _attach_quota_supporting_projections( state_action_projection_warning: dict[str, Any] | None, next_action_warning: dict[str, Any] | None, replan_obligation: dict[str, Any] | None, + notify_gate: bool = True, ) -> None: _attach_truthy_fields( payload, @@ -489,7 +493,8 @@ def _attach_quota_supporting_projections( gate_prompt := _build_gate_prompt(item, user_todo_summary=user_todo_summary) ): payload["gate_prompt"] = gate_prompt - payload["notify_user_on_gate"] = True + if notify_gate: + payload["notify_user_on_gate"] = True _attach_truthy_fields( payload, next_handoff_condition=item.get("next_handoff_condition"), agent_command=item.get("agent_command") if should_run else None, @@ -803,6 +808,37 @@ def _resolve_quota_should_run_route( goal_frontier_projection=prepared.goal_frontier_projection, task_orchestration_contract=prepared.task_orchestration_contract, ) + if prepared.receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED: + fields = settled_replay_fields() + return _QuotaDecisionRoute( + normal_delivery_allowed=fields["normal_delivery_allowed"], + recovery_allowed=fields["recovery_delivery_allowed"], + self_repair_allowed=fields["self_repair_allowed"], + capability_repair_allowed=fields["capability_repair_allowed"], + workspace_repair_allowed=fields["workspace_repair_allowed"], + should_run=fields["should_run"], effective_action=fields["effective_action"], + reason=fields["reason"], state=run_decision.state, quota=run_decision.quota, + replan_decision_allowed=False, receipt_bound_replan_decision=False, + heartbeat_recommendation=fields["heartbeat_recommendation"], + external_evidence_observation=None, external_evidence_observation_recent=None, + selected_recommended_action=HEARTBEAT_SETTLED_REPLAY_REASON, + agent_lane_next_action=None, agent_scope_frontier=None, + agent_lane_frontier_hint=None, state_action_projection_warning=None, + active_state_next_action_text=_protocol_action_text( + item.get("active_state_next_action") + or prepared.project_asset.get("active_state_next_action") + or prepared.project_asset.get("next_action"), limit=320, + ), + latest_run_recommended_action_text=_protocol_action_text( + item.get("latest_run_recommended_action") + or prepared.project_asset.get("latest_run_recommended_action"), limit=320, + ), + next_action_warning=None, goal_route_hint=None, + payload_work_lane_contract=_payload_work_lane_contract( + prepared.work_lane_contract, effective_action=fields["effective_action"], + recovery_allowed=False, agent_scope_frontier=None, + ), + ) normal_delivery_allowed = run_decision.normal_delivery_allowed recovery_allowed = run_decision.recovery_delivery_allowed self_repair_allowed = run_decision.self_repair_allowed @@ -1127,13 +1163,58 @@ def _resolve_quota_should_run_route( ) -def _build_quota_should_run_payload( +def _quota_payload_context(prepared: _QuotaDecisionPreparation, route: _QuotaDecisionRoute) -> dict[str, Any]: + return { + **_standing_decision_authority_payload_from_status_item( + prepared.item, + project_asset=prepared.project_asset, + agent_id=normalize_todo_claimed_by( + (prepared.agent_identity or {}).get("agent_id") + ), + ), + "ok": prepared.goal_health_ok + or route.self_repair_allowed + or route.capability_repair_allowed + or route.workspace_repair_allowed, + "status_health_ok": prepared.goal_health_ok, + "mode": "should-run", + "goal_id": prepared.safe_goal_id, + "quota": route.quota, + "state": route.state, + "blocked_action_scope": prepared.boundary_projection_repair.get("blocked_action_scope") + if prepared.boundary_projection_repair + else stall_repair_blocked_action_scope(prepared.stall_self_repair) + or route.quota.get("blocked_action_scope"), + "safe_bypass_allowed": bool(route.quota.get("safe_bypass_allowed")), + "safe_bypass_kind": route.quota.get("safe_bypass_kind"), + "safe_bypass_policy": route.quota.get("safe_bypass_policy"), + "waiting_on": prepared.item.get("waiting_on"), + "status": prepared.item.get("status"), + "lifecycle_phase": prepared.item.get("lifecycle_phase"), + "lifecycle_flags": prepared.item.get("lifecycle_flags"), + "source": prepared.item.get("source"), + "project_asset_source": prepared.item.get("project_asset_source"), + "active_state_next_action": route.active_state_next_action_text or None, + "latest_run_recommended_action": route.latest_run_recommended_action_text or None, + "execution_profile": _quota_execution_profile_summary( + prepared.project_asset.get("execution_profile") + ) + if prepared.project_asset + else None, + "long_task_cadence_hint": prepared.item.get("long_task_cadence_hint") + if isinstance(prepared.item.get("long_task_cadence_hint"), dict) + else None, + "handoff_readiness": prepared.item.get("handoff_readiness"), + "goal_boundary": prepared.goal_boundary, + "plan_summary": prepared.plan.get("summary") + } + + +def _build_active_quota_payload( prepared: _QuotaDecisionPreparation, route: _QuotaDecisionRoute, *, - turn_instance_id: str | None = None, include_agent_todo_detail: bool = False, - runtime_root: str | Path | None = None, ) -> dict[str, Any]: agent_scope_action = _agent_scope_frontier_action(route.effective_action) execution_obligation = _execution_obligation( @@ -1153,24 +1234,8 @@ def _build_quota_should_run_payload( execution_obligation.get("must_attempt_work") ) payload = { - **_standing_decision_authority_payload_from_status_item( - prepared.item, - project_asset=prepared.project_asset, - agent_id=normalize_todo_claimed_by( - (prepared.agent_identity or {}).get("agent_id") - ), - ), - "ok": ( - prepared.goal_health_ok - or route.self_repair_allowed - or route.capability_repair_allowed - or route.workspace_repair_allowed - ), - "status_health_ok": prepared.goal_health_ok, - "mode": "should-run", - "goal_id": prepared.safe_goal_id, - "decision": ( - AUTONOMOUS_REPLAN_REQUIRED_MODE + **_quota_payload_context(prepared, route), + "decision": AUTONOMOUS_REPLAN_REQUIRED_MODE if route.replan_decision_allowed else "run" if route.normal_delivery_allowed @@ -1190,8 +1255,7 @@ def _build_quota_should_run_payload( if agent_scope_action is not None else PEER_COORDINATION_BLOCKED_ACTION if route.effective_action == PEER_COORDINATION_BLOCKED_ACTION - else "skip" - ), + else "skip", "should_run": route.should_run, "normal_delivery_allowed": route.normal_delivery_allowed, "recovery_delivery_allowed": route.recovery_allowed, @@ -1206,53 +1270,15 @@ def _build_quota_should_run_payload( or route.capability_repair_allowed or route.workspace_repair_allowed ), - "reason": ( - str(prepared.stall_self_repair.get("reason")) + "reason": str(prepared.stall_self_repair.get("reason")) if route.self_repair_allowed and isinstance(prepared.stall_self_repair, dict) - else route.reason - ), - "quota": route.quota, - "state": route.state, - "blocked_action_scope": ( - prepared.boundary_projection_repair.get("blocked_action_scope") - if prepared.boundary_projection_repair - else stall_repair_blocked_action_scope(prepared.stall_self_repair) - or route.quota.get("blocked_action_scope") - ), - "safe_bypass_allowed": bool(route.quota.get("safe_bypass_allowed")), - "safe_bypass_kind": route.quota.get("safe_bypass_kind"), - "safe_bypass_policy": route.quota.get("safe_bypass_policy"), - "waiting_on": prepared.item.get("waiting_on"), - "status": prepared.item.get("status"), - "lifecycle_phase": prepared.item.get("lifecycle_phase"), - "lifecycle_flags": prepared.item.get("lifecycle_flags"), - "source": prepared.item.get("source"), - "project_asset_source": prepared.item.get("project_asset_source"), + else route.reason, "recommended_action": route.selected_recommended_action, - "active_state_next_action": route.active_state_next_action_text or None, - "latest_run_recommended_action": ( - route.latest_run_recommended_action_text or None - ), - "execution_profile": ( - _quota_execution_profile_summary( - prepared.project_asset.get("execution_profile") - ) - if prepared.project_asset - else None - ), - "long_task_cadence_hint": ( - prepared.item.get("long_task_cadence_hint") - if isinstance(prepared.item.get("long_task_cadence_hint"), dict) - else None - ), - "handoff_readiness": prepared.item.get("handoff_readiness"), "heartbeat_recommendation": route.heartbeat_recommendation, "execution_obligation": execution_obligation, - "goal_boundary": prepared.goal_boundary, "goal_frontier_projection": prepared.goal_frontier_projection, - "plan_summary": prepared.plan.get("summary"), - "todo_write_hint": build_todo_write_hint(prepared.safe_goal_id), + "todo_write_hint": build_todo_write_hint(prepared.safe_goal_id) } if payload["safe_bypass_policy"] is None: payload.pop("safe_bypass_policy") @@ -1441,10 +1467,6 @@ def _build_quota_should_run_payload( monitor_only=prepared.agent_monitor_only, inbox_priority_due=prepared.inbox_priority_due, ) - apply_settled_replay_payload_precedence( - payload, - replay_phase=prepared.receipt_bound_replay_phase, - ) if isinstance(payload.get("autonomous_replan_obligation"), dict): payload["replan_action_packet"] = build_replan_action_packet( payload["autonomous_replan_obligation"], @@ -1454,6 +1476,23 @@ def _build_quota_should_run_payload( ), bounded_research_frontier=bounded_research_frontier, ) + return payload + + +def _build_quota_should_run_payload( + prepared: _QuotaDecisionPreparation, + route: _QuotaDecisionRoute, + *, + turn_instance_id: str | None = None, + include_agent_todo_detail: bool = False, + runtime_root: str | Path | None = None, +) -> dict[str, Any]: + if prepared.receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED: + payload = _build_settled_quota_payload(prepared, route) + else: + payload = _build_active_quota_payload( + prepared, route, include_agent_todo_detail=include_agent_todo_detail, + ) payload["automation_liveness"] = build_automation_liveness(payload) payload["interaction_contract"] = build_interaction_contract( payload, @@ -1497,3 +1536,57 @@ def _build_quota_should_run_payload( ) payload["protocol_action_packet"] = build_protocol_action_packet(payload) return payload + + +def _build_settled_quota_payload( + prepared: _QuotaDecisionPreparation, route: _QuotaDecisionRoute, +) -> dict[str, Any]: + payload = { + **_quota_payload_context(prepared, route), + **settled_replay_fields(), + "goal_frontier_projection": { + key: value for key, value in prepared.goal_frontier_projection.items() + if key not in {"autonomous_replan_decision", "replan_ack_feedback", + "replan_obligation", "vision_continuation_audit", "vision_wait_state"} + }, + } + if payload["safe_bypass_policy"] is None: + payload.pop("safe_bypass_policy") + _attach_agent_identity_contracts(payload=payload, agent_identity=prepared.agent_identity) + _attach_truthy_fields( + payload, + action_selection_qualification=prepared.action_selection_qualification, + automation_prompt_upgrade=prepared.automation_prompt_upgrade, + work_lane_contract=route.payload_work_lane_contract, + monitor_debt_arbitration=prepared.monitor_debt_arbitration if prepared.monitor_debt_arbitration.get("active") else None, + external_evidence_observation_recent=route.external_evidence_observation_recent, + control_plane=compact_control_plane_policy(prepared.item.get("control_plane")), + missing_gates=prepared.item.get("missing_gates"), + agent_todo_summary=compact_quota_todo_summary_for_payload(prepared.agent_todo_summary) if prepared.agent_todo_summary else None, + user_todo_summary=compact_quota_todo_summary_for_payload(prepared.user_todo_summary) if prepared.user_todo_summary else None, + bounded_research_frontier=_dict_field(prepared.status_payload, "bounded_research_frontier"), + ) + if prepared.agent_scoped_user_todo_override: + payload[str(prepared.agent_scoped_user_todo_override["kind"])] = prepared.agent_scoped_user_todo_override + payload.update(stall_repair_payload(prepared.stall_self_repair)) + attention_queue = _dict_field(prepared.status_payload, "attention_queue") or {} + for field in ("autonomous_backlog_candidates", "autonomous_monitor_candidates"): + value = _compact_autonomous_candidate_context(attention_queue.get(field), goal_id=prepared.safe_goal_id) + if value: + payload[field] = value + _attach_quota_supporting_projections( + payload, status_payload=prepared.status_payload, item=prepared.item, + project_asset=prepared.project_asset, goal_id=prepared.safe_goal_id, + selected_recommended_action=route.selected_recommended_action, state=route.state, + user_todo_summary=prepared.user_todo_summary, should_run=False, + state_action_projection_warning=None, next_action_warning=None, + replan_obligation=None, notify_gate=False, + ) + if prepared.scoped_user_gate_fallback: + payload.update(scoped_user_gate_fallback_fields()) + # Work-mode diagnostics survive settlement, while execution authority comes + # only from the settled result, never the monitor precedence mutator. + if prepared.agent_monitor_only and not prepared.inbox_priority_due: + payload.update(agent_work_mode="monitor_only", blocked_action_scope="advancement_work", + safe_bypass_allowed=False, safe_bypass_kind=None, safe_bypass_policy=None) + return payload diff --git a/loopx/control_plane/todos/user_gate.py b/loopx/control_plane/todos/user_gate.py index 3a52355c6b..9262d667b2 100644 --- a/loopx/control_plane/todos/user_gate.py +++ b/loopx/control_plane/todos/user_gate.py @@ -146,6 +146,19 @@ def build_user_todo_notification( } +def scoped_user_gate_fallback_fields() -> dict[str, Any]: + """Project the existing scoped-fallback readback independently of execution.""" + return { + "safe_bypass_allowed": True, + "safe_bypass_kind": "scoped_user_gate_fallback", + "safe_bypass_policy": ( + "The user gate blocks only the matched agent action scope. Surface " + "that gate, then advance the selected non-gated fallback; spend only " + "after validated writeback." + ), + } + + def apply_scoped_user_gate_fallback_projection( payload: dict[str, Any], *, @@ -182,13 +195,7 @@ def apply_scoped_user_gate_fallback_projection( } ) projected["execution_obligation"] = execution_obligation - projected["safe_bypass_allowed"] = True - projected["safe_bypass_kind"] = "scoped_user_gate_fallback" - projected["safe_bypass_policy"] = ( - "The user gate blocks only the matched agent action scope. Surface " - "that gate, then advance the selected non-gated fallback; spend only " - "after validated writeback." - ) + projected.update(scoped_user_gate_fallback_fields()) projected["actionable_by_codex"] = True return projected diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json index b12d032080..d380802fd6 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -504,8 +504,7 @@ "loopx/control_plane/quota/live_decision.py::_apply_pending_capability_intent_precedence", "loopx/control_plane/quota/projection_repair.py::build_boundary_projection_repair_hint", "loopx/control_plane/quota/projection_repair.py::build_state_projection_gap_repair_hint", - "loopx/control_plane/quota/settlement_precedence.py::apply_settled_replay_payload_precedence", - "loopx/control_plane/quota/settlement_precedence.py::apply_settled_replay_route_precedence", + "loopx/control_plane/quota/settlement_precedence.py::settled_replay_fields", "loopx/control_plane/quota/should_run.py::build_quota_paused_should_run_payload", "loopx/control_plane/quota/should_run_packet.py::_apply_agent_monitor_only_precedence", "loopx/control_plane/quota/should_run_packet.py::_resolve_quota_should_run_route", diff --git a/tests/control_plane/test_settled_replay_construction.py b/tests/control_plane/test_settled_replay_construction.py new file mode 100644 index 0000000000..ba3abf5b17 --- /dev/null +++ b/tests/control_plane/test_settled_replay_construction.py @@ -0,0 +1,137 @@ +"""Settled Turns have observations, but never construct a successor action.""" +from __future__ import annotations + +import pytest + +from loopx.control_plane.effect_program import ReceiptBoundReplayPhase +from loopx.control_plane.quota import should_run_packet +from loopx.control_plane.quota.should_run import build_quota_should_run +from loopx.control_plane.testing.quota_fixtures import quota_status_payload + + +@pytest.mark.parametrize("quota_state", ["eligible", "operator_gate", "waiting_external", "exhausted"]) +@pytest.mark.parametrize("monitor", [False, True]) +def test_settled_turn_never_constructs_successor_or_replan( + monkeypatch: pytest.MonkeyPatch, quota_state: str, monitor: bool, +) -> None: + def unexpected(*args, **kwargs): + pytest.fail("settled replay entered an executable action construction path") + + monkeypatch.setattr(should_run_packet, "_resolve_agent_lane_delivery_route", unexpected) + monkeypatch.setattr(should_run_packet, "build_replan_action_packet", unexpected) + monkeypatch.setattr(should_run_packet, "_apply_agent_monitor_only_precedence", unexpected) + status = quota_status_payload( + goal_id="settled-fixture", status="active", quota_state=quota_state, + agent_todo_items=[{ + "todo_id": "todo_successor", "index": 1, "text": "[P1] Advance successor", + "role": "agent", "status": "open", "priority": "P1", + "task_class": "continuous_monitor" if monitor else "advancement_task", + }], + recommended_action="Advance successor", + ) + payload = build_quota_should_run( + status, goal_id="settled-fixture", available_capabilities=["shell"], + receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED, + ) + assert payload["effective_action"] == "heartbeat_settled_skip" + assert payload["decision"] == "skip" + for flag in ( + "should_run", "normal_delivery_allowed", "recovery_delivery_allowed", + "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed", + "actionable_by_codex", "requires_user_action", + ): + assert payload[flag] is False, flag + assert payload["execution_obligation"]["must_attempt_work"] is False + assert payload["heartbeat_recommendation"]["agent_must_attempt"] is False + for field in ("selected_todo", "replan_action_packet", "autonomous_replan_obligation", "action_portfolio"): + assert field not in payload + assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is False + assert payload["interaction_contract"]["cli_channel"]["spend_after_validation"] is False + assert payload["protocol_action_packet"]["summary"] + + +def test_pause_still_precedes_settled_replay() -> None: + status = quota_status_payload( + goal_id="settled-fixture", status="active", quota_state="paused", + recommended_action="Wait for owner", + ) + payload = build_quota_should_run( + status, goal_id="settled-fixture", + receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED, + ) + assert payload["should_run"] is False + assert payload["effective_action"] == "quota_skip" + assert payload["state"] == "paused" + + +def test_live_intent_can_follow_settled_quota_without_reopening_work( + tmp_path, monkeypatch: pytest.MonkeyPatch, +) -> None: + from types import SimpleNamespace + from loopx.control_plane.quota import live_decision + from loopx.control_plane.capability_hooks import ( + InteractionProjectionHookRegistration, + INTERACTION_PROJECTION_HOOK_RESULT_SCHEMA_VERSION, + ) + + # Receipt identity/refusal is covered through the real CLI settlement tests; + # this case isolates composition with the real typed hook decoder. + monkeypatch.setattr(live_decision, "read_heartbeat_settlement", lambda *args, **kwargs: SimpleNamespace( + replay_phase=ReceiptBoundReplayPhase.SETTLED, monitor_phase=None, + )) + command = "loopx periodic-report consume-pending --goal-id settled-fixture --agent-id fixture-agent --execute" + hook = InteractionProjectionHookRegistration( + hook_id="periodic_report.pending_intent", capability_id="periodic-report", + projection_slots=("pending_capability_intent",), + requested_read_scope=("post_writeback_intent_journal",), + producer=lambda: { + "schema_version": INTERACTION_PROJECTION_HOOK_RESULT_SCHEMA_VERSION, + "hook_id": "periodic_report.pending_intent", "capability_id": "periodic-report", + "phase": "interaction_projection", "status": "candidate", + "projection_slot": "pending_capability_intent", + "payload": { + "schema_version": "pending_capability_intent_projection_v0", + "capability_id": "periodic-report", "intent_kind": "periodic_report.trigger_evaluation", + "idempotency_key": "periodic-report:fixture", "intent_digest": "sha256:" + "a" * 64, + "goal_id": "settled-fixture", "agent_id": "fixture-agent", "state": "pending", + "action_kind": "consume_periodic_report_intent", + "action_summary": "Generate the report under its own receipt.", "command": command, + "generation_authorized": True, "external_delivery_authorized": True, + "agent_read_required": True, + }, + }, + ) + payload = live_decision.build_live_quota_should_run_decision( + quota_status_payload(goal_id="settled-fixture", status="active", recommended_action="Continue"), + goal_id="settled-fixture", agent_id=None, available_capabilities=["shell"], + include_scheduler_detail=False, codex_app_current_rrule=None, + registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", + interaction_projection_hooks=[hook], + ) + assert payload["effective_action"] == "governed_capability_intent" + assert payload["interaction_contract"]["cli_channel"]["next_cli_actions"] == [command] + assert payload.get("selected_todo") is None + assert payload["normal_delivery_allowed"] is False + + +def test_settled_fallback_readback_cannot_reopen_execution(monkeypatch: pytest.MonkeyPatch) -> None: + from loopx.control_plane.quota import should_run + original = should_run._prepare_quota_should_run_item + + def prepare(*args, **kwargs): + prepared = original(*args, **kwargs) + prepared.scoped_user_gate_fallback = {"reason": "scoped gate", "recommended_action": "Safe work"} + return prepared + + monkeypatch.setattr(should_run, "_prepare_quota_should_run_item", prepare) + payload = build_quota_should_run( + quota_status_payload(goal_id="settled-fixture", status="active", recommended_action="Continue"), + goal_id="settled-fixture", receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED, + ) + assert payload["safe_bypass_allowed"] is True + assert payload["safe_bypass_kind"] == "scoped_user_gate_fallback" + assert payload["should_run"] is False + assert payload["actionable_by_codex"] is False + assert payload["execution_obligation"]["must_attempt_work"] is False + assert payload["interaction_contract"]["cli_channel"]["spend_after_validation"] is False + assert "scoped_user_gate_fallback" not in payload From d29d43b308a9e1f91c077900bcc260bc5dff57e8 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 04:30:13 -0400 Subject: [PATCH 2/8] refactor(quota): isolate supporting packet projections Move the existing diagnostic projection helpers into a private quota module used by both active and settled packet construction. Keep settled replay construction and monitor precedence unchanged. Pass preparation fields directly into the run decision instead of aliasing and then overwriting them. The packet falls from 1592 to 1452 lines and the route from 92 to 86 AST statements without changing budgets, exceptions, or semantic producers. Validation: 251 focused tests, 4 real CLI settlement regressions, and 236 maintainability/semantic checks pass. All 40 base/head packets are equal and the five moved helper bodies retain identical ASTs. Add independent gate notification, command suppression, and goal-scoped reward diagnostics tests. Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- .../quota/_supporting_projections.py | 136 +++++++++++++++ .../control_plane/quota/should_run_packet.py | 162 ++---------------- .../test_quota_supporting_projections.py | 71 ++++++++ 3 files changed, 218 insertions(+), 151 deletions(-) create mode 100644 loopx/control_plane/quota/_supporting_projections.py create mode 100644 tests/control_plane/test_quota_supporting_projections.py diff --git a/loopx/control_plane/quota/_supporting_projections.py b/loopx/control_plane/quota/_supporting_projections.py new file mode 100644 index 0000000000..606ac75a49 --- /dev/null +++ b/loopx/control_plane/quota/_supporting_projections.py @@ -0,0 +1,136 @@ +"""Supporting diagnostics shared by active and settled quota packets.""" +from __future__ import annotations + +from typing import Any + +from ..agents.agent_scope import _action_scope_tokens_from_text +from ..runtime.decision_freshness import decision_freshness_warning as _decision_freshness_warning +from ..runtime.promotion_readiness import promotion_readiness_warning as _promotion_readiness_warning +from ..todos.user_gate import build_gate_prompt as _build_gate_prompt +from .recent_runs import goal_latest_runs as _goal_latest_runs + + +def _recent_reward_lessons(status_payload: dict[str, Any], *, goal_id: str) -> list[dict[str, Any]]: + lessons: list[dict[str, Any]] = [] + for run in _goal_latest_runs(status_payload, goal_id=goal_id): + reward = run.get("human_reward") if isinstance(run.get("human_reward"), dict) else {} + lesson = reward.get("lesson") if isinstance(reward.get("lesson"), dict) else {} + if not lesson: + continue + lessons.append( + { + "generated_at": run.get("generated_at"), + "decision": reward.get("decision"), + "reward": reward.get("reward"), + "kind": lesson.get("kind"), + "summary": lesson.get("summary"), + "avoid": lesson.get("avoid") if isinstance(lesson.get("avoid"), list) else [], + "prefer": lesson.get("prefer") if isinstance(lesson.get("prefer"), list) else [], + } + ) + return lessons + + +def _reward_lesson_projection_warning( + status_payload: dict[str, Any], + *, + goal_id: str, + recommended_action: str | None, +) -> dict[str, Any] | None: + action = str(recommended_action or "").strip() + if not action: + return None + action_lower = action.lower() + action_tokens = _action_scope_tokens_from_text(action) + matches: list[dict[str, Any]] = [] + for lesson in _recent_reward_lessons(status_payload, goal_id=goal_id): + for avoid in lesson.get("avoid") or []: + avoid_text = str(avoid or "").strip() + if not avoid_text: + continue + avoid_tokens = _action_scope_tokens_from_text(avoid_text) + exact_match = avoid_text.lower() in action_lower + if not exact_match and not avoid_tokens: + continue + token_overlap = sorted(action_tokens & avoid_tokens) + if not exact_match and len(token_overlap) < min(2, len(avoid_tokens)): + continue + matches.append( + { + "generated_at": lesson.get("generated_at"), + "decision": lesson.get("decision"), + "kind": lesson.get("kind"), + "summary": lesson.get("summary"), + "avoid": avoid_text, + "token_overlap": token_overlap[:5], + } + ) + if not matches: + return None + return { + "schema_version": "reward_lesson_projection_warning_v0", + "source": "run_history.human_reward.lesson", + "goal_id": goal_id, + "message": ( + "recommended_action overlaps a recent human_reward lesson avoid rule; " + "rebase the route or update the affected todo/next action before continuing" + ), + "recommended_action": action, + "match_count": len(matches), + "matches": matches[:3], + } + + +def _attach_truthy_fields(payload: dict[str, Any], **fields: Any) -> None: + payload.update({key: value for key, value in fields.items() if value}) + + +def _dict_field(payload: dict[str, Any], key: str) -> dict[str, Any] | None: + return payload.get(key) if isinstance(payload.get(key), dict) else None + + +def _attach_quota_supporting_projections( + payload: dict[str, Any], + *, + status_payload: dict[str, Any], + item: dict[str, Any], + project_asset: dict[str, Any], + goal_id: str, + selected_recommended_action: Any, + state: str, + user_todo_summary: dict[str, Any] | None, + should_run: bool, + state_action_projection_warning: dict[str, Any] | None, + next_action_warning: dict[str, Any] | None, + replan_obligation: dict[str, Any] | None, + notify_gate: bool = True, +) -> None: + _attach_truthy_fields( + payload, + stale_latest_run_warning=_dict_field(item, "stale_latest_run_warning"), + state_action_projection_warning=state_action_projection_warning, + next_action_projection_warning=next_action_warning, + backlog_hygiene_warning=_dict_field(item, "backlog_hygiene_warning"), + completed_todo_archive_warning=_dict_field(item, "completed_todo_archive_warning"), + autonomous_replan_obligation=replan_obligation, + dreaming_proposal=_dict_field(item, "dreaming_proposal"), + dreaming_lane_badge=_dict_field(item, "dreaming_lane_badge"), + interface_budget_cadence=_dict_field(project_asset, "interface_budget_cadence"), + decision_freshness_warning=_decision_freshness_warning(status_payload, goal_id=goal_id), + promotion_readiness_warning=_promotion_readiness_warning(status_payload), + reward_lesson_projection_warning=_reward_lesson_projection_warning( + status_payload, + goal_id=goal_id, + recommended_action=selected_recommended_action, + ), + ) + if state == "operator_gate" and ( + gate_prompt := _build_gate_prompt(item, user_todo_summary=user_todo_summary) + ): + payload["gate_prompt"] = gate_prompt + if notify_gate: + payload["notify_user_on_gate"] = True + _attach_truthy_fields( + payload, next_handoff_condition=item.get("next_handoff_condition"), + agent_command=item.get("agent_command") if should_run else None, + ) diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index ac7045c0b3..b10eb67dba 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -19,7 +19,6 @@ ) from ..agents.agent_scope import ( AgentScopeFrontierAction, - _action_scope_tokens_from_text, _agent_lane_frontier_hint, _agent_scope_deferred_resume_candidates, _agent_scope_frontier_action, @@ -47,9 +46,6 @@ AUTONOMOUS_CANDIDATE_CONTEXT_FIELDS, MONITOR_DUE_ITEM_LIMIT, ) -from ..quota.recent_runs import ( - goal_latest_runs as _goal_latest_runs, -) from ..quota.recent_runs import ( recent_external_monitor_observation_unchanged as _recent_external_monitor_observation_unchanged, ) @@ -73,12 +69,6 @@ from ..quota.task_orchestration import ( payload_work_lane_contract as _payload_work_lane_contract, ) -from ..runtime.decision_freshness import ( - decision_freshness_warning as _decision_freshness_warning, -) -from ..runtime.promotion_readiness import ( - promotion_readiness_warning as _promotion_readiness_warning, -) from ..scheduler.automation_liveness import build_automation_liveness from ..scheduler.execution_context import ( SchedulerExecutionContextResolution, @@ -103,9 +93,6 @@ apply_scoped_user_gate_fallback_projection as _apply_scoped_user_gate_fallback_projection, scoped_user_gate_fallback_fields, ) -from ..todos.user_gate import ( - build_gate_prompt as _build_gate_prompt, -) from ..todos.user_gate import ( build_user_todo_notification as _build_user_todo_notification, ) @@ -140,6 +127,11 @@ clear_quota_action_projections, ) from .should_run_prepare import _QuotaDecisionPreparation +from ._supporting_projections import ( + _attach_quota_supporting_projections, + _attach_truthy_fields, + _dict_field, +) def _interaction_runtime_root( @@ -242,77 +234,6 @@ def _execution_obligation( ) -def _recent_reward_lessons(status_payload: dict[str, Any], *, goal_id: str) -> list[dict[str, Any]]: - lessons: list[dict[str, Any]] = [] - for run in _goal_latest_runs(status_payload, goal_id=goal_id): - reward = run.get("human_reward") if isinstance(run.get("human_reward"), dict) else {} - lesson = reward.get("lesson") if isinstance(reward.get("lesson"), dict) else {} - if not lesson: - continue - lessons.append( - { - "generated_at": run.get("generated_at"), - "decision": reward.get("decision"), - "reward": reward.get("reward"), - "kind": lesson.get("kind"), - "summary": lesson.get("summary"), - "avoid": lesson.get("avoid") if isinstance(lesson.get("avoid"), list) else [], - "prefer": lesson.get("prefer") if isinstance(lesson.get("prefer"), list) else [], - } - ) - return lessons - - -def _reward_lesson_projection_warning( - status_payload: dict[str, Any], - *, - goal_id: str, - recommended_action: str | None, -) -> dict[str, Any] | None: - action = str(recommended_action or "").strip() - if not action: - return None - action_lower = action.lower() - action_tokens = _action_scope_tokens_from_text(action) - matches: list[dict[str, Any]] = [] - for lesson in _recent_reward_lessons(status_payload, goal_id=goal_id): - for avoid in lesson.get("avoid") or []: - avoid_text = str(avoid or "").strip() - if not avoid_text: - continue - avoid_tokens = _action_scope_tokens_from_text(avoid_text) - exact_match = avoid_text.lower() in action_lower - if not exact_match and not avoid_tokens: - continue - token_overlap = sorted(action_tokens & avoid_tokens) - if not exact_match and len(token_overlap) < min(2, len(avoid_tokens)): - continue - matches.append( - { - "generated_at": lesson.get("generated_at"), - "decision": lesson.get("decision"), - "kind": lesson.get("kind"), - "summary": lesson.get("summary"), - "avoid": avoid_text, - "token_overlap": token_overlap[:5], - } - ) - if not matches: - return None - return { - "schema_version": "reward_lesson_projection_warning_v0", - "source": "run_history.human_reward.lesson", - "goal_id": goal_id, - "message": ( - "recommended_action overlaps a recent human_reward lesson avoid rule; " - "rebase the route or update the affected todo/next action before continuing" - ), - "recommended_action": action, - "match_count": len(matches), - "matches": matches[:3], - } - - def _apply_agent_monitor_only_precedence( payload: dict[str, Any], *, @@ -446,61 +367,6 @@ def _apply_agent_monitor_only_precedence( clear_quota_action_projections(payload) -def _attach_truthy_fields(payload: dict[str, Any], **fields: Any) -> None: - payload.update({key: value for key, value in fields.items() if value}) - - -def _dict_field(payload: dict[str, Any], key: str) -> dict[str, Any] | None: - return payload.get(key) if isinstance(payload.get(key), dict) else None - - -def _attach_quota_supporting_projections( - payload: dict[str, Any], - *, - status_payload: dict[str, Any], - item: dict[str, Any], - project_asset: dict[str, Any], - goal_id: str, - selected_recommended_action: Any, - state: str, - user_todo_summary: dict[str, Any] | None, - should_run: bool, - state_action_projection_warning: dict[str, Any] | None, - next_action_warning: dict[str, Any] | None, - replan_obligation: dict[str, Any] | None, - notify_gate: bool = True, -) -> None: - _attach_truthy_fields( - payload, - stale_latest_run_warning=_dict_field(item, "stale_latest_run_warning"), - state_action_projection_warning=state_action_projection_warning, - next_action_projection_warning=next_action_warning, - backlog_hygiene_warning=_dict_field(item, "backlog_hygiene_warning"), - completed_todo_archive_warning=_dict_field(item, "completed_todo_archive_warning"), - autonomous_replan_obligation=replan_obligation, - dreaming_proposal=_dict_field(item, "dreaming_proposal"), - dreaming_lane_badge=_dict_field(item, "dreaming_lane_badge"), - interface_budget_cadence=_dict_field(project_asset, "interface_budget_cadence"), - decision_freshness_warning=_decision_freshness_warning(status_payload, goal_id=goal_id), - promotion_readiness_warning=_promotion_readiness_warning(status_payload), - reward_lesson_projection_warning=_reward_lesson_projection_warning( - status_payload, - goal_id=goal_id, - recommended_action=selected_recommended_action, - ), - ) - if state == "operator_gate" and ( - gate_prompt := _build_gate_prompt(item, user_todo_summary=user_todo_summary) - ): - payload["gate_prompt"] = gate_prompt - if notify_gate: - payload["notify_user_on_gate"] = True - _attach_truthy_fields( - payload, next_handoff_condition=item.get("next_handoff_condition"), - agent_command=item.get("agent_command") if should_run else None, - ) - - def _delivery_preemptions_for_route( prepared: _QuotaDecisionPreparation, *, @@ -780,20 +646,14 @@ def _resolve_quota_should_run_route( prepared: _QuotaDecisionPreparation, ) -> _QuotaDecisionRoute: item = prepared.item - normal_delivery_allowed = prepared.normal_delivery_allowed - recovery_allowed = prepared.recovery_allowed - self_repair_allowed = prepared.self_repair_allowed - state = prepared.state - quota = prepared.quota - reason = prepared.reason run_decision = resolve_quota_run_decision( - normal_delivery_allowed=normal_delivery_allowed, - recovery_delivery_allowed=recovery_allowed, - self_repair_allowed=self_repair_allowed, + normal_delivery_allowed=prepared.normal_delivery_allowed, + recovery_delivery_allowed=prepared.recovery_allowed, + self_repair_allowed=prepared.self_repair_allowed, stall_self_repair=prepared.stall_self_repair, - state=state, - quota=quota, - reason=reason, + state=prepared.state, + quota=prepared.quota, + reason=prepared.reason, capability_gate=prepared.capability_gate, capability_monitor_fallback=prepared.capability_monitor_fallback, workspace_guard=prepared.workspace_guard, diff --git a/tests/control_plane/test_quota_supporting_projections.py b/tests/control_plane/test_quota_supporting_projections.py new file mode 100644 index 0000000000..15a0d44305 --- /dev/null +++ b/tests/control_plane/test_quota_supporting_projections.py @@ -0,0 +1,71 @@ +"""Quota diagnostics preserve observations without granting execution authority.""" + +from copy import deepcopy + +import pytest + +from loopx.control_plane.quota.should_run_packet import ( + _attach_quota_supporting_projections, +) + + +@pytest.mark.parametrize("should_run,notify_gate", [(True, True), (False, False)]) +def test_gate_observations_do_not_reopen_settled_execution( + should_run: bool, notify_gate: bool, +) -> None: + payload = {"should_run": should_run} + item = { + "operator_question": "Approve the next step?", + "agent_command": "loopx status", + "stale_latest_run_warning": {"reason": "stale observation"}, + "backlog_hygiene_warning": "invalid diagnostic shape", + "next_handoff_condition": "Owner decision arrives", + } + before = deepcopy(item) + _attach_quota_supporting_projections( + payload, status_payload={}, item=item, project_asset={}, + goal_id="projection-fixture", selected_recommended_action="Wait for approval", + state="operator_gate", user_todo_summary=None, should_run=should_run, + state_action_projection_warning=None, next_action_warning=None, + replan_obligation=None, notify_gate=notify_gate, + ) + assert payload["should_run"] is should_run + assert payload["gate_prompt"] + assert payload.get("notify_user_on_gate", False) is notify_gate + assert ("agent_command" in payload) is should_run + assert payload["stale_latest_run_warning"] == {"reason": "stale observation"} + assert "backlog_hygiene_warning" not in payload + assert payload["next_handoff_condition"] == "Owner decision arrives" + assert item == before + + +@pytest.mark.parametrize("action,goal_id,matches", [ + ("Retry obsolete endpoint", "projection-fixture", True), + ("Validate fresh parser", "projection-fixture", False), + ("Retry obsolete endpoint", "other-goal", False), + (None, "projection-fixture", False), +]) +def test_reward_warning_is_scoped_to_selected_action_and_goal( + action: str | None, goal_id: str, matches: bool, +) -> None: + status = {"run_history": {"goals": [{ + "id": "projection-fixture", + "latest_runs": [{"human_reward": {"lesson": { + "summary": "Use the current endpoint", "avoid": ["obsolete endpoint"], + }}}], + }]}} + payload = {} + _attach_quota_supporting_projections( + payload, status_payload=status, item={}, project_asset={}, goal_id=goal_id, + selected_recommended_action=action, state="eligible", user_todo_summary=None, + should_run=False, state_action_projection_warning=None, + next_action_warning=None, replan_obligation=None, + ) + assert ("reward_lesson_projection_warning" in payload) is matches + if matches: + warning = payload["reward_lesson_projection_warning"] + assert warning["goal_id"] == goal_id + assert warning["match_count"] == 1 + assert warning["matches"][0]["avoid"] == "obsolete endpoint" + assert "agent_command" not in payload + assert "notify_user_on_gate" not in payload From b72a8d5c7c626233fa4ad3acbcc0e75f4f0285b1 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 08:06:33 -0400 Subject: [PATCH 3/8] fix(quota): keep settled turn closed before prior recovery Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- loopx/control_plane/quota/live_decision.py | 25 ++++++---- .../test_effect_turn_live_quota_decision.py | 50 ++++++++++++++++++- 2 files changed, 64 insertions(+), 11 deletions(-) diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index d364441f08..8c9e8487b2 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -13,6 +13,7 @@ InteractionProjectionHookRegistration, dispatch_interaction_projection_hooks, ) +from .effect_program import ReceiptBoundReplayPhase from .settlement import ( read_heartbeat_settlement, ) @@ -514,16 +515,20 @@ def build_live_quota_should_run_decision( interaction = payload.get("interaction_contract") if isinstance(interaction, dict): interaction.update(projections) - apply_unsettled_host_turn_recovery_if_required( - payload, - registry_path=registry_path, - runtime_root=runtime_root, - goal_id=goal_id, - agent_id=agent_id, - current_turn_instance_id=turn_instance_id, - available_capabilities=available_capabilities, - scheduler_execution_context=resolved_context, - ) + # The settled receipt owns the current Turn until it ends. An older + # unsettled Turn may be recovered only by a fresh Turn; allowing it to run + # here would overwrite heartbeat_settled_skip and reopen execution. + if receipt_bound_replay_phase is not ReceiptBoundReplayPhase.SETTLED: + apply_unsettled_host_turn_recovery_if_required( + payload, + registry_path=registry_path, + runtime_root=runtime_root, + goal_id=goal_id, + agent_id=agent_id, + current_turn_instance_id=turn_instance_id, + available_capabilities=available_capabilities, + scheduler_execution_context=resolved_context, + ) if hook_dispatch["failures"]: payload["capability_hook_dispatch"] = { key: value for key, value in hook_dispatch.items() if key != "projections" diff --git a/tests/control_plane/test_effect_turn_live_quota_decision.py b/tests/control_plane/test_effect_turn_live_quota_decision.py index d83c7a9853..1acb7e4338 100644 --- a/tests/control_plane/test_effect_turn_live_quota_decision.py +++ b/tests/control_plane/test_effect_turn_live_quota_decision.py @@ -6,6 +6,7 @@ import pytest from loopx.control_plane.effect_program import ( + ReceiptBoundReplayPhase, interpret_quota_should_run_packet, ) from loopx.control_plane.capability_hooks import ( @@ -174,7 +175,7 @@ def test_live_quota_decision_maps_to_effect_turn(tmp_path: Path) -> None: turn = interpret_quota_should_run_packet( packet, goal_id=GOAL_ID, - agent_id="codex-fixture", + agent_id=None, capabilities=["shell"], ) @@ -278,6 +279,53 @@ def test_managed_turn_projects_prior_unsettled_heartbeat_recovery( assert contract["cli_channel"]["spend_after_validation"] is False +def test_settled_turn_defers_prior_turn_recovery_to_a_fresh_turn( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + from types import SimpleNamespace + from loopx.control_plane.quota import live_decision + + monkeypatch.setattr( + live_decision, + "read_heartbeat_settlement", + lambda *_args, **_kwargs: SimpleNamespace( + replay_phase=ReceiptBoundReplayPhase.SETTLED, + monitor_phase=None, + ), + ) + + def forbidden_recovery(*_args, **_kwargs): + pytest.fail("settled current Turn entered prior-Turn recovery") + + monkeypatch.setattr( + live_decision, + "apply_unsettled_host_turn_recovery_if_required", + forbidden_recovery, + ) + payload = build_live_quota_should_run_decision( + _ordinary_status_payload(), + goal_id=GOAL_ID, + agent_id=None, + available_capabilities=["shell"], + include_scheduler_detail=False, + codex_app_current_rrule=None, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + route_source="loopx_turn_plan", + turn_instance_id="managed-current-turn", + scheduler_execution_context={ + "host_surface": "generic_cli", + "scheduler_owner": "agent_cli_loop", + "execution_mode": "interactive", + }, + ) + + assert payload["effective_action"] == "heartbeat_settled_skip" + assert payload["should_run"] is False + assert payload["execution_obligation"]["must_attempt_work"] is False + assert payload.get("unsettled_host_turn_recovery") is None + + def test_managed_turn_accepts_exact_material_monitor_poll_closeout( tmp_path: Path, ) -> None: From 91e6be96b57f818f9cc7f46511403929db47df90 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 08:18:12 -0400 Subject: [PATCH 4/8] fix(quota): resolve settled replay import merge Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- loopx/control_plane/quota/live_decision.py | 1 - 1 file changed, 1 deletion(-) diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index d9b1d5151a..4f8d4e1ae3 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -17,7 +17,6 @@ from .settlement import ( read_heartbeat_settlement, ) -from .effect_program import ReceiptBoundReplayPhase from ..work_items.interaction_contract import ( build_interaction_contract, build_protocol_action_packet, From 1fc7de952299ff080f045564cd4f6ce596279846 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 08:44:05 -0400 Subject: [PATCH 5/8] fix(quota): keep direct preflight calls compatible Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- loopx/cli_commands/quota.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index caa8594e1b..5774fa0052 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -209,8 +209,8 @@ def _apply_requested_quota_action_selection_preflight( requested_todo_id: str | None, receipt_bound_todo_id: str | None, receipt_bound_replan_obligation_id: str | None, - receipt_pending_action_todo_id: str | None, - receipt_identity_upgraded: bool, + receipt_pending_action_todo_id: str | None = None, + receipt_identity_upgraded: bool = False, ) -> bool: if not requested_todo_id: return False From 41e9de612a0ccb5b21162a924af7851e312350b2 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 09:40:28 -0400 Subject: [PATCH 6/8] test(semantics): preserve inventory smoke test facade Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- examples/semantic-vocabulary-drift-smoke.py | 1 + 1 file changed, 1 insertion(+) diff --git a/examples/semantic-vocabulary-drift-smoke.py b/examples/semantic-vocabulary-drift-smoke.py index 3e3f30efcc..61af9d0d9d 100755 --- a/examples/semantic-vocabulary-drift-smoke.py +++ b/examples/semantic-vocabulary-drift-smoke.py @@ -27,6 +27,7 @@ INVENTORY_SCHEMA_VERSION, SourceFile, build_inventory, + render_inventory, # noqa: F401 - re-exported to runpy-based contract tests collect_string_constants, load_sources, python_facts, From 969e94f8bbad4a33ee5307eb7516f097478a35ff Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 20:12:21 -0400 Subject: [PATCH 7/8] test(quota): preserve settlement Todo identity Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- tests/control_plane/test_quota_settlement_cli.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index b4eb4b9adc..239a191581 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -1549,6 +1549,8 @@ def test_standard_codex_app_settlement_is_receipted_and_idempotent( binding = ( "--agent-id", AGENT_ID, + "--todo-id", + TODO_ID, "--turn-instance-id", TURN_ID, ) From 2076d0ff80c55c1de2bf22964177eeedb485edbf Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 22:11:18 -0400 Subject: [PATCH 8/8] fix(quota): keep a settled turn's safe bypass closed A settled receipt proves the current Turn is finished, so the payload carried a no-work/no-spend obligation. But when a scoped user-gate fallback was prepared, the settled payload projected the fallback readback, which set safe_bypass_allowed=true with a policy that ends "spend only after validated writeback". The installed heartbeat task body reads should_run=false together with safe_bypass_allowed=true as permission to run one bounded safe-bypass step, write back and spend once, and the quota markdown surfaced both the settled no-spend policy and the fallback spend policy at the same time. The grant, not the readback, is the problem: it is an action authority, not a diagnostic. Construct the closed safe bypass in settled_replay_fields(), which is already the single construction point for settled authority, and stop re-granting it from the fallback readback. A fresh Turn still computes its own fallback, so the scoped-gate path keeps working; only the already-settled identity is denied. The monitor-only branch no longer needs to reset the same three fields. The regression asserted the old grant. It now pins the raw packet, the interaction contract, the heartbeat recommendation and the rendered guidance, so a settled Turn cannot carry a second, executable reading. Signed-off-by: song <22676124+songoow@users.noreply.github.com> Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- .../quota/settlement_precedence.py | 6 +++++ .../control_plane/quota/should_run_packet.py | 9 +++----- .../test_settled_replay_construction.py | 23 +++++++++++++++++-- 3 files changed, 30 insertions(+), 8 deletions(-) diff --git a/loopx/control_plane/quota/settlement_precedence.py b/loopx/control_plane/quota/settlement_precedence.py index 7e60fbb589..d3605e0449 100644 --- a/loopx/control_plane/quota/settlement_precedence.py +++ b/loopx/control_plane/quota/settlement_precedence.py @@ -59,6 +59,12 @@ def settled_replay_fields() -> dict[str, Any]: "self_repair_allowed": False, "capability_repair_allowed": False, "workspace_repair_allowed": False, + # A settled Turn grants no safe bypass: the heartbeat task body reads + # safe_bypass_allowed as permission to run a bounded step and spend, so + # a fresh Turn must recompute any fallback instead of inheriting one. + "safe_bypass_allowed": False, + "safe_bypass_kind": None, + "safe_bypass_policy": None, "effective_action": EffectiveAction.HEARTBEAT_SETTLED_SKIP.value, "actionable_by_codex": False, "reason": reason, diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index b10eb67dba..e7788b811d 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -91,7 +91,6 @@ ) from ..todos.user_gate import ( apply_scoped_user_gate_fallback_projection as _apply_scoped_user_gate_fallback_projection, - scoped_user_gate_fallback_fields, ) from ..todos.user_gate import ( build_user_todo_notification as _build_user_todo_notification, @@ -1442,11 +1441,9 @@ def _build_settled_quota_payload( state_action_projection_warning=None, next_action_warning=None, replan_obligation=None, notify_gate=False, ) - if prepared.scoped_user_gate_fallback: - payload.update(scoped_user_gate_fallback_fields()) # Work-mode diagnostics survive settlement, while execution authority comes - # only from the settled result, never the monitor precedence mutator. + # only from the settled result, never a fallback readback or the monitor + # precedence mutator. if prepared.agent_monitor_only and not prepared.inbox_priority_due: - payload.update(agent_work_mode="monitor_only", blocked_action_scope="advancement_work", - safe_bypass_allowed=False, safe_bypass_kind=None, safe_bypass_policy=None) + payload.update(agent_work_mode="monitor_only", blocked_action_scope="advancement_work") return payload diff --git a/tests/control_plane/test_settled_replay_construction.py b/tests/control_plane/test_settled_replay_construction.py index ba3abf5b17..b5b10230c8 100644 --- a/tests/control_plane/test_settled_replay_construction.py +++ b/tests/control_plane/test_settled_replay_construction.py @@ -7,6 +7,7 @@ from loopx.control_plane.quota import should_run_packet from loopx.control_plane.quota.should_run import build_quota_should_run from loopx.control_plane.testing.quota_fixtures import quota_status_payload +from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown @pytest.mark.parametrize("quota_state", ["eligible", "operator_gate", "waiting_external", "exhausted"]) @@ -128,10 +129,28 @@ def prepare(*args, **kwargs): quota_status_payload(goal_id="settled-fixture", status="active", recommended_action="Continue"), goal_id="settled-fixture", receipt_bound_replay_phase=ReceiptBoundReplayPhase.SETTLED, ) - assert payload["safe_bypass_allowed"] is True - assert payload["safe_bypass_kind"] == "scoped_user_gate_fallback" + # The heartbeat task body reads safe_bypass_allowed=true under + # should_run=false as permission to run one bounded step and spend once, so + # a settled Turn must not inherit the fallback grant from its readback. + assert payload["safe_bypass_allowed"] is False + assert payload["safe_bypass_kind"] is None + assert "safe_bypass_policy" not in payload assert payload["should_run"] is False assert payload["actionable_by_codex"] is False assert payload["execution_obligation"]["must_attempt_work"] is False + assert payload["interaction_contract"]["mode"] == "heartbeat_settled_skip" + assert payload["interaction_contract"]["agent_channel"]["must_attempt"] is False assert payload["interaction_contract"]["cli_channel"]["spend_after_validation"] is False assert "scoped_user_gate_fallback" not in payload + + recommendation = payload["heartbeat_recommendation"] + assert recommendation["recommended_mode"] == "heartbeat_settled_skip" + assert recommendation["agent_must_attempt"] is False + assert "no quota spend" in recommendation["spend_policy"] + + # The guidance the agent actually reads must not carry a second, executable + # reading of the same settled Turn. + guidance = render_quota_should_run_markdown(payload) + assert "safe_bypass" not in guidance + assert "spend only after validated writeback" not in guidance + assert "heartbeat_spend_policy: no quota spend" in guidance