diff --git a/loopx/control_plane/quota/should_run.py b/loopx/control_plane/quota/should_run.py new file mode 100644 index 000000000..2f1cf1f72 --- /dev/null +++ b/loopx/control_plane/quota/should_run.py @@ -0,0 +1,235 @@ +"""Bounded `quota should-run` decision and packet builder.""" + +from __future__ import annotations + +from collections.abc import Callable, Mapping +from typing import Any + +from ...quota import ( + _build_quota_plan_for_goal, + _build_quota_should_run_payload, + _execution_obligation, + _prepare_quota_should_run_item, + _resolve_quota_should_run_route, + _scheduler_hint, +) +from ..agents.agent_scope import _attach_agent_identity_contracts +from ..agents.identity import build_quota_agent_identity +from ..quota.decision_summary import quota_plan_items as _quota_plan_items +from ..quota.goal_boundary import registry_goal_by_id as _registry_goal_by_id +from ..quota.states import quota_item_is_paused as _quota_item_is_paused +from ..scheduler.automation_liveness import build_automation_liveness +from ..scheduler.execution_context import ( + SchedulerExecutionContextResolution, + resolve_scheduler_execution_context, +) +from ..todos.write_hint import build_todo_write_hint +from ..work_items.interaction_contract import ( + build_interaction_contract, + build_protocol_action_packet, +) + + +QUOTA_PAUSED_MODE = "quota_paused" + + +def build_quota_paused_should_run_payload( + status_payload: dict[str, Any], + *, + safe_goal_id: str, + requested_agent_id: str | None, + item: dict[str, Any], + plan: dict[str, Any], + goal_health_ok: bool, + include_scheduler_detail: bool, + codex_app_current_rrule: Any, + resolved_scheduler_context: SchedulerExecutionContextResolution, +) -> dict[str, Any]: + """Project one canonical paused contract with no contradicting lane authority. + + The whole Goal is hard-paused, so every automatic authority field resolves to + the same terminal decision: `should_run=false`, all delivery/repair + permissions false, `DONT_NOTIFY`, no quota spend, and a scheduler cadence that + is never `run_now`. No capability_gate, workspace_guard, replan, monitor, or + inbox candidate is constructed here. + """ + + quota = item.get("quota") if isinstance(item.get("quota"), dict) else {} + quota = {**quota, "state": "paused"} + reason = str( + quota.get("reason") + or "compute quota is 0; the whole Goal is hard-paused and automatic agent turns stop" + ) + agent_identity = build_quota_agent_identity(item, agent_id=requested_agent_id) + heartbeat_recommendation = { + "source": "quota.should-run", + "recommended_mode": QUOTA_PAUSED_MODE, + "notify": "DONT_NOTIFY", + "reason": reason, + "spend_policy": "do not append quota spend while the Goal is paused", + } + execution_obligation = _execution_obligation( + should_run=False, + effective_action="quota_skip", + heartbeat_recommendation=heartbeat_recommendation, + ) + payload: dict[str, Any] = { + "ok": goal_health_ok, + "status_health_ok": goal_health_ok, + "mode": "should-run", + "goal_id": safe_goal_id, + "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": "quota_skip", + "actionable_by_codex": False, + "reason": reason, + "quota": quota, + "state": "paused", + "safe_bypass_allowed": False, + "waiting_on": item.get("waiting_on"), + "status": item.get("status"), + "lifecycle_phase": item.get("lifecycle_phase"), + "lifecycle_flags": item.get("lifecycle_flags"), + "source": item.get("source"), + "recommended_action": reason, + "requires_user_action": False, + "heartbeat_recommendation": heartbeat_recommendation, + "execution_obligation": execution_obligation, + "plan_summary": plan.get("summary"), + "todo_write_hint": build_todo_write_hint(safe_goal_id), + } + payload = _attach_agent_identity_contracts( + payload=payload, + agent_identity=agent_identity, + ) + payload["automation_liveness"] = build_automation_liveness(payload) + payload["interaction_contract"] = build_interaction_contract( + payload, + available_capabilities=None, + scheduler_execution_context=resolved_scheduler_context, + ) + payload["scheduler_hint"] = _scheduler_hint( + payload, + include_detail=include_scheduler_detail, + available_capabilities=None, + codex_app_current_rrule=codex_app_current_rrule, + scheduler_execution_context=resolved_scheduler_context, + ) + payload["protocol_action_packet"] = build_protocol_action_packet(payload) + return payload + + +def build_quota_should_run( + status_payload: dict[str, Any], + *, + goal_id: str, + agent_id: str | None = None, + available_capabilities: Any = None, + include_scheduler_detail: bool = False, + codex_app_current_rrule: Any = None, + scheduler_execution_context: ( + Mapping[str, Any] | SchedulerExecutionContextResolution | None + ) = None, + operator_inbox_urgency_projector: Callable[..., dict[str, Any]] | None = None, +) -> dict[str, Any]: + safe_goal_id = str(goal_id or "").strip() + resolved_scheduler_context = resolve_scheduler_execution_context( + scheduler_execution_context + ) + registry_goal = _registry_goal_by_id(status_payload).get(safe_goal_id) or {} + plan, goal_health_ok = _build_quota_plan_for_goal( + status_payload, + goal_id=safe_goal_id, + ) + item = next( + ( + candidate + for candidate in _quota_plan_items(plan) + if candidate.get("goal_id") == safe_goal_id + ), + None, + ) + health_items = ( + plan.get("health_items") + if isinstance(plan.get("health_items"), list) + else [] + ) + health_item = next( + ( + candidate + for candidate in health_items + if isinstance(candidate, dict) and candidate.get("goal_id") == safe_goal_id + ), + None, + ) + if item: + if _quota_item_is_paused(item): + return build_quota_paused_should_run_payload( + status_payload, + safe_goal_id=safe_goal_id, + requested_agent_id=agent_id, + item=item, + plan=plan, + goal_health_ok=goal_health_ok, + include_scheduler_detail=include_scheduler_detail, + codex_app_current_rrule=codex_app_current_rrule, + resolved_scheduler_context=resolved_scheduler_context, + ) + prepared = _prepare_quota_should_run_item( + status_payload, + safe_goal_id=safe_goal_id, + requested_agent_id=agent_id, + available_capabilities=available_capabilities, + include_scheduler_detail=include_scheduler_detail, + codex_app_current_rrule=codex_app_current_rrule, + resolved_scheduler_context=resolved_scheduler_context, + operator_inbox_urgency_projector=operator_inbox_urgency_projector, + registry_goal=registry_goal, + plan=plan, + goal_health_ok=goal_health_ok, + item=item, + health_items=health_items, + ) + return _build_quota_should_run_payload( + prepared, + _resolve_quota_should_run_route(prepared), + ) + if health_item: + return { + "ok": False, + "mode": "should-run", + "goal_id": safe_goal_id, + "decision": "skip", + "should_run": False, + "reason": str( + health_item.get("recommended_action") + or "health item blocks automatic compute" + ), + "state": "blocked_health", + "waiting_on": health_item.get("waiting_on"), + "status": health_item.get("status"), + "source": health_item.get("source"), + "recommended_action": health_item.get("recommended_action"), + "plan_summary": plan.get("summary"), + } + return { + "ok": False, + "mode": "should-run", + "goal_id": safe_goal_id, + "decision": "skip", + "should_run": False, + "reason": "goal is not present in the registered quota plan", + "state": "unknown", + "waiting_on": None, + "status": "goal_not_found", + "source": "quota", + "recommended_action": ( + "run `loopx registry` and connect or sync the goal before spending compute" + ), + "plan_summary": plan.get("summary"), + } diff --git a/loopx/quota.py b/loopx/quota.py index ae1d28dfa..d0a2c02a2 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -66,7 +66,6 @@ from .control_plane.quota.decision_summary import ( goal_status_health_ok as _goal_status_health_ok, quota_decision_agent_id, - quota_plan_items as _quota_plan_items, refine_quota_recommended_action, resolve_quota_run_decision, ) @@ -115,7 +114,7 @@ from .control_plane.quota.spend_sources import ( DEFAULT_SLOT_SPEND_SOURCE, ) -from .control_plane.quota.states import QUOTA_STATE_ORDER, quota_item_is_paused as _quota_item_is_paused +from .control_plane.quota.states import QUOTA_STATE_ORDER from .control_plane.quota.policy_constants import ( AUTONOMOUS_CANDIDATE_CONTEXT_FIELDS, DEFAULT_COMPUTE_QUOTA, @@ -142,7 +141,6 @@ from .control_plane.scheduler.scheduler_hint import build_scheduler_hint from .control_plane.scheduler.execution_context import ( SchedulerExecutionContextResolution, - resolve_scheduler_execution_context, ) from .control_plane.scheduler.external_evidence_observation import build_external_evidence_observation_obligation from .control_plane.scheduler.automation_liveness import build_automation_liveness @@ -2411,100 +2409,6 @@ def _build_quota_should_run_payload( return payload -QUOTA_PAUSED_MODE = "quota_paused" - - -def _build_quota_paused_should_run_payload( - status_payload: dict[str, Any], - *, - safe_goal_id: str, - requested_agent_id: str | None, - item: dict[str, Any], - plan: dict[str, Any], - goal_health_ok: bool, - include_scheduler_detail: bool, - codex_app_current_rrule: Any, - resolved_scheduler_context: SchedulerExecutionContextResolution, -) -> dict[str, Any]: - """Project one canonical paused contract with no contradicting lane authority. - - The whole Goal is hard-paused, so every automatic authority field resolves to - the same terminal decision: `should_run=false`, all delivery/repair - permissions false, `DONT_NOTIFY`, no quota spend, and a scheduler cadence that - is never `run_now`. No capability_gate, workspace_guard, replan, monitor, or - inbox candidate is constructed here. - """ - - quota = item.get("quota") if isinstance(item.get("quota"), dict) else {} - quota = {**quota, "state": "paused"} - reason = str( - quota.get("reason") - or "compute quota is 0; the whole Goal is hard-paused and automatic agent turns stop" - ) - agent_identity = build_quota_agent_identity(item, agent_id=requested_agent_id) - heartbeat_recommendation = { - "source": "quota.should-run", - "recommended_mode": QUOTA_PAUSED_MODE, - "notify": "DONT_NOTIFY", - "reason": reason, - "spend_policy": "do not append quota spend while the Goal is paused", - } - execution_obligation = _execution_obligation( - should_run=False, - effective_action="quota_skip", - heartbeat_recommendation=heartbeat_recommendation, - ) - payload: dict[str, Any] = { - "ok": goal_health_ok, - "status_health_ok": goal_health_ok, - "mode": "should-run", - "goal_id": safe_goal_id, - "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": "quota_skip", - "actionable_by_codex": False, - "reason": reason, - "quota": quota, - "state": "paused", - "safe_bypass_allowed": False, - "waiting_on": item.get("waiting_on"), - "status": item.get("status"), - "lifecycle_phase": item.get("lifecycle_phase"), - "lifecycle_flags": item.get("lifecycle_flags"), - "source": item.get("source"), - "recommended_action": reason, - "requires_user_action": False, - "heartbeat_recommendation": heartbeat_recommendation, - "execution_obligation": execution_obligation, - "plan_summary": plan.get("summary"), - "todo_write_hint": build_todo_write_hint(safe_goal_id), - } - payload = _attach_agent_identity_contracts( - payload=payload, - agent_identity=agent_identity, - ) - payload["automation_liveness"] = build_automation_liveness(payload) - payload["interaction_contract"] = build_interaction_contract( - payload, - available_capabilities=None, - scheduler_execution_context=resolved_scheduler_context, - ) - payload["scheduler_hint"] = _scheduler_hint( - payload, - include_detail=include_scheduler_detail, - available_capabilities=None, - codex_app_current_rrule=codex_app_current_rrule, - scheduler_execution_context=resolved_scheduler_context, - ) - payload["protocol_action_packet"] = build_protocol_action_packet(payload) - return payload - - def build_quota_should_run( status_payload: dict[str, Any], *, @@ -2518,102 +2422,20 @@ def build_quota_should_run( ) = None, operator_inbox_urgency_projector: Callable[..., dict[str, Any]] | None = None, ) -> dict[str, Any]: - safe_goal_id = str(goal_id or "").strip() - resolved_scheduler_context = resolve_scheduler_execution_context( - scheduler_execution_context + from .control_plane.quota.should_run import ( + build_quota_should_run as _build_quota_should_run, ) - registry_goal = _registry_goal_by_id(status_payload).get(safe_goal_id) or {} - plan, goal_health_ok = _build_quota_plan_for_goal( + + return _build_quota_should_run( status_payload, - goal_id=safe_goal_id, - ) - item = next( - ( - candidate - for candidate in _quota_plan_items(plan) - if candidate.get("goal_id") == safe_goal_id - ), - None, - ) - health_items = ( - plan.get("health_items") - if isinstance(plan.get("health_items"), list) - else [] + goal_id=goal_id, + agent_id=agent_id, + available_capabilities=available_capabilities, + include_scheduler_detail=include_scheduler_detail, + codex_app_current_rrule=codex_app_current_rrule, + scheduler_execution_context=scheduler_execution_context, + operator_inbox_urgency_projector=operator_inbox_urgency_projector, ) - health_item = next( - ( - candidate - for candidate in health_items - if isinstance(candidate, dict) and candidate.get("goal_id") == safe_goal_id - ), - None, - ) - if item: - if _quota_item_is_paused(item): - return _build_quota_paused_should_run_payload( - status_payload, - safe_goal_id=safe_goal_id, - requested_agent_id=agent_id, - item=item, - plan=plan, - goal_health_ok=goal_health_ok, - include_scheduler_detail=include_scheduler_detail, - codex_app_current_rrule=codex_app_current_rrule, - resolved_scheduler_context=resolved_scheduler_context, - ) - prepared = _prepare_quota_should_run_item( - status_payload, - safe_goal_id=safe_goal_id, - requested_agent_id=agent_id, - available_capabilities=available_capabilities, - include_scheduler_detail=include_scheduler_detail, - codex_app_current_rrule=codex_app_current_rrule, - resolved_scheduler_context=resolved_scheduler_context, - operator_inbox_urgency_projector=operator_inbox_urgency_projector, - registry_goal=registry_goal, - plan=plan, - goal_health_ok=goal_health_ok, - item=item, - health_items=health_items, - ) - return _build_quota_should_run_payload( - prepared, - _resolve_quota_should_run_route(prepared), - ) - if health_item: - return { - "ok": False, - "mode": "should-run", - "goal_id": safe_goal_id, - "decision": "skip", - "should_run": False, - "reason": str( - health_item.get("recommended_action") - or "health item blocks automatic compute" - ), - "state": "blocked_health", - "waiting_on": health_item.get("waiting_on"), - "status": health_item.get("status"), - "source": health_item.get("source"), - "recommended_action": health_item.get("recommended_action"), - "plan_summary": plan.get("summary"), - } - return { - "ok": False, - "mode": "should-run", - "goal_id": safe_goal_id, - "decision": "skip", - "should_run": False, - "reason": "goal is not present in the registered quota plan", - "state": "unknown", - "waiting_on": None, - "status": "goal_not_found", - "source": "quota", - "recommended_action": ( - "run `loopx registry` and connect or sync the goal before spending compute" - ), - "plan_summary": plan.get("summary"), - } def build_quota_slot_preview( diff --git a/tests/control_plane/test_quota_should_run_parity.py b/tests/control_plane/test_quota_should_run_parity.py index c3176fb72..2a711845a 100644 --- a/tests/control_plane/test_quota_should_run_parity.py +++ b/tests/control_plane/test_quota_should_run_parity.py @@ -1,13 +1,43 @@ from __future__ import annotations +from loopx.control_plane.quota.should_run import ( + build_quota_should_run as bounded_build_quota_should_run, +) from loopx.control_plane.testing.quota_fixtures import quota_status_payload from loopx.control_plane.testing.quota_should_run_parity import ( build_quota_should_run_parity, ) +from loopx.quota import build_quota_should_run as facade_build_quota_should_run GOAL_ID = "quota-parity-fixture" +def test_facade_builds_through_bounded_should_run_module() -> None: + todo_text = "[P1] Advance the bounded slice." + payload = quota_status_payload( + goal_id=GOAL_ID, + status="active", + agent_todo_items=[ + { + "index": 1, + "text": todo_text, + "role": "agent", + "status": "open", + "priority": "P1", + "task_class": "advancement_task", + } + ], + recommended_action=todo_text, + next_action=todo_text, + ) + + facade_packet = facade_build_quota_should_run(payload, goal_id=GOAL_ID) + bounded_packet = bounded_build_quota_should_run(payload, goal_id=GOAL_ID) + + assert facade_packet == bounded_packet + assert bounded_packet["decision"] == "run" + + def test_advancement_run_parity_surface() -> None: todo_text = "[P1] Advance the bounded slice." payload = quota_status_payload(