diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index caa8594e1..5774fa005 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 diff --git a/loopx/control_plane/quota/_supporting_projections.py b/loopx/control_plane/quota/_supporting_projections.py new file mode 100644 index 000000000..606ac75a4 --- /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/live_decision.py b/loopx/control_plane/quota/live_decision.py index 7da31ca94..ea27270b1 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -13,10 +13,10 @@ InteractionProjectionHookRegistration, dispatch_interaction_projection_hooks, ) +from .effect_program import ReceiptBoundReplayPhase from .settlement import ( read_heartbeat_settlement, ) -from .effect_program import ReceiptBoundReplayPhase from ..work_items.interaction_contract import ( build_interaction_contract, build_protocol_action_packet, diff --git a/loopx/control_plane/quota/settlement_precedence.py b/loopx/control_plane/quota/settlement_precedence.py index a9894339e..d3605e044 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,44 @@ 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, + # 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, + "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 b938f51d0..e6424871a 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 631a5c761..e7788b811 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 @@ -18,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, @@ -46,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, ) @@ -72,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, @@ -101,9 +92,6 @@ from ..todos.user_gate import ( apply_scoped_user_gate_fallback_projection as _apply_scoped_user_gate_fallback_projection, ) -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, ) @@ -133,10 +121,16 @@ 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 +from ._supporting_projections import ( + _attach_quota_supporting_projections, + _attach_truthy_fields, + _dict_field, +) def _interaction_runtime_root( @@ -239,77 +233,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], *, @@ -443,59 +366,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, -) -> 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 - 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, *, @@ -775,20 +645,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, @@ -803,6 +667,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 +1022,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 +1093,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 +1114,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 +1129,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 +1326,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 +1335,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 +1395,55 @@ 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, + ) + # Work-mode diagnostics survive settlement, while execution authority comes + # 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") + return payload diff --git a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts index 43373553b..9a968c078 100644 --- a/loopx/control_plane/quota/unsettled_host_turn_recovery.ts +++ b/loopx/control_plane/quota/unsettled_host_turn_recovery.ts @@ -222,34 +222,40 @@ export async function preflightPriorHostTurnCloseout( turns_validated: turnsValidated, }; } - const selected = candidates[0]!; - const readback = await readQuotaSettlement( - settlementReadbackRequest(request, selected), - ); - if (readback.found === true && !bundleFailed(readback, "settlement")) { - // The Turn has a validated settlement closeout, so it needs no recovery. + let newestSettledTurn: string | null = null; + for (const selected of candidates) { + const readback = await readQuotaSettlement( + settlementReadbackRequest(request, selected), + ); + if (readback.found === true && !bundleFailed(readback, "settlement")) { + // A newer settled Turn cannot hide an older missing closeout. Keep + // scanning in persisted newest-first order until the first unsettled + // candidate is found. + newestSettledTurn ??= selected.prior_turn_instance_id; + continue; + } + const missingReceipts: string[] = []; + if (readback.found !== true || bundleFailed(readback, "writeback")) { + missingReceipts.push(WRITEBACK_RECEIPT); + } + if (readback.found !== true || bundleFailed(readback, "spend")) { + missingReceipts.push(SPEND_RECEIPT); + } return { schema_version: PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA, - status: "none", - reason: "prior_turn_settlement_validated", + status: "candidate", turns_validated: turnsValidated, - prior_turn_instance_id: selected.prior_turn_instance_id, - accepted_closeout: "validated_writeback_and_quota_spend", + candidate: selected, + missing_receipts: missingReceipts, }; } - const missingReceipts: string[] = []; - if (readback.found !== true || bundleFailed(readback, "writeback")) { - missingReceipts.push(WRITEBACK_RECEIPT); - } - if (readback.found !== true || bundleFailed(readback, "spend")) { - missingReceipts.push(SPEND_RECEIPT); - } return { schema_version: PRIOR_HOST_TURN_CLOSEOUT_PREFLIGHT_RESULT_SCHEMA, - status: "candidate", + status: "none", + reason: "prior_turn_settlement_validated", turns_validated: turnsValidated, - candidate: selected, - missing_receipts: missingReceipts, + prior_turn_instance_id: newestSettledTurn, + accepted_closeout: "validated_writeback_and_quota_spend", }; } diff --git a/loopx/control_plane/todos/user_gate.py b/loopx/control_plane/todos/user_gate.py index 3a52355c6..9262d667b 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 f0c4b66e2..054688550 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -505,8 +505,7 @@ "loopx/control_plane/work_items/action_portfolio.ts::reconcileRetainedActionSelection", "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_effect_turn_live_quota_decision.py b/tests/control_plane/test_effect_turn_live_quota_decision.py index 77463bad9..d352ff8d6 100644 --- a/tests/control_plane/test_effect_turn_live_quota_decision.py +++ b/tests/control_plane/test_effect_turn_live_quota_decision.py @@ -177,7 +177,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"], ) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index b3170f79e..239a19158 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -20,6 +20,7 @@ AUTONOMOUS_REPLAN_PERIODIC_RUN_THRESHOLD, ) from loopx.heartbeat_prompt import build_heartbeat_prompt +from loopx.rollout_event_log import build_rollout_event REPO_ROOT = Path(__file__).resolve().parents[2] GOAL_ID = "settlement-cli-fixture" @@ -1564,6 +1565,8 @@ def test_standard_codex_app_settlement_is_receipted_and_idempotent( GOAL_ID, "--agent-id", AGENT_ID, + "--todo-id", + TODO_ID, "--turn-instance-id", TURN_ID, "--scan-path", @@ -5514,6 +5517,147 @@ def test_same_turn_terminal_receipt_replay_preempts_autonomous_replan( assert fresh["replan_action_packet"]["obligation_id"] +def test_settled_turn_defers_prior_unsettled_history_to_fresh_turn( + tmp_path: Path, +) -> None: + project, runtime, registry_path = _write_fixture(tmp_path) + _configure_selectable_alternative(project) + guard_args = ( + "quota", + "should-run", + "--codex-app", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--todo-id", + TODO_ID, + "--turn-instance-id", + TURN_ID, + "--scan-path", + str(project), + ) + binding = ( + "--agent-id", + AGENT_ID, + "--todo-id", + TODO_ID, + "--turn-instance-id", + TURN_ID, + ) + + first_rc, first = _run_cli(registry_path, runtime, *guard_args) + assert first_rc == 0, first + assert first["selected_todo"]["todo_id"] == TODO_ID + + refresh_rc, refresh = _run_cli( + registry_path, + runtime, + "refresh-state", + "--goal-id", + GOAL_ID, + "--classification", + "settled_turn_recovery_order_validated", + "--delivery-batch-scale", + "single_surface", + "--delivery-outcome", + "outcome_progress", + *binding, + "--no-global-sync", + "--suppress-external-sinks", + ) + assert refresh_rc == 0, refresh + + spend_rc, spend = _run_cli( + registry_path, + runtime, + "quota", + "spend-slot", + "--goal-id", + GOAL_ID, + "--slots", + "1", + "--source", + "heartbeat", + "--execute", + *binding, + "--scan-path", + str(project), + ) + assert spend_rc == 0, spend + + complete_rc, complete = _run_cli( + registry_path, + runtime, + "todo", + "complete", + "--goal-id", + GOAL_ID, + *binding, + "--claimed-by", + AGENT_ID, + "--evidence", + "settled Turn recovery order validated", + "--no-follow-up", + ) + assert complete_rc == 0, complete + + prior_turn_id = "turn-unsettled-prior" + prior = build_rollout_event( + goal_id=GOAL_ID, + event_kind="quota_should_run", + agent_id=AGENT_ID, + todo_id=ALTERNATIVE_TODO_ID, + run_id=prior_turn_id, + status="normal_run", + summary="prior host Turn requires closeout", + recorded_at="2025-12-31T23:59:00Z", + details={ + "todo_id": ALTERNATIVE_TODO_ID, + "settlement_effect_id": ( + f"{GOAL_ID}:{AGENT_ID}:{ALTERNATIVE_TODO_ID}:{prior_turn_id}" + ), + "closeout_required": True, + }, + ) + log_path = runtime / "goals" / GOAL_ID / "rollout-event-log.jsonl" + committed_history = log_path.read_text(encoding="utf-8") + log_path.write_text( + json.dumps(prior) + "\n" + committed_history, + encoding="utf-8", + ) + + replay_rc, replay = _run_cli(registry_path, runtime, *guard_args) + assert replay_rc == 0, replay + assert replay["decision"] == "skip" + assert replay["effective_action"] == "heartbeat_settled_skip" + assert replay["should_run"] is False + assert replay.get("unsettled_host_turn_recovery") is None + + fresh_args = ( + "quota", + "should-run", + "--codex-app", + "--goal-id", + GOAL_ID, + "--agent-id", + AGENT_ID, + "--turn-instance-id", + "turn-settlement-cli-2", + "--scan-path", + str(project), + ) + fresh_rc, fresh = _run_cli(registry_path, runtime, *fresh_args) + assert fresh_rc == 0, fresh + assert fresh["effective_action"] == "unsettled_host_turn_recovery" + assert fresh["unsettled_host_turn_recovery"]["prior_turn_instance_id"] == ( + prior_turn_id + ) + assert fresh["unsettled_host_turn_recovery"]["binding_id"] == ( + ALTERNATIVE_TODO_ID + ) + + def test_legacy_read_only_workspace_mismatch_fails_then_corrects_from_todo_contract( tmp_path: Path, ) -> None: 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 000000000..15a0d4430 --- /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 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 000000000..b5b10230c --- /dev/null +++ b/tests/control_plane/test_settled_replay_construction.py @@ -0,0 +1,156 @@ +"""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 +from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown + + +@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, + ) + # 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