From ecb84c4ae29082b8c027fe12901d76aaa9399573 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 01:35:00 -0400 Subject: [PATCH 1/8] refactor(quota): construct unadmitted selection recovery directly Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- loopx/cli_commands/quota.py | 189 +++++++----------- .../control_plane/quota/should_run_packet.py | 4 +- .../work_items/action_selection_contract.py | 171 +++++++++++++--- .../work_items/interaction_contract.py | 132 +++++++++--- loopx/semantics/vocabulary_v0.json | 2 +- .../test_quota_action_selection_conflict.py | 8 +- .../test_unadmitted_selection_construction.py | 115 +++++++++++ 7 files changed, 431 insertions(+), 190 deletions(-) create mode 100644 tests/control_plane/test_unadmitted_selection_construction.py diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index cd0b74ecbc..cdf4237602 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -49,7 +49,10 @@ render_scheduler_execution_args, ) from ..control_plane.todos.contract import normalize_todo_id -from ..control_plane.work_items.action_selection_contract import apply_action_selection_recovery +from ..control_plane.work_items.action_selection_contract import ( + bind_action_selection_recovery_command, current_action_selection_admission, +) +from ..control_plane.work_items.interaction_contract import action_selection_recovery_fields from ..presentation.renderers.quota_event_markdown import ( render_quota_monitor_poll_markdown, render_quota_slot_preview_markdown, @@ -188,89 +191,29 @@ def _heartbeat_quota_action_selection_bindings( return existing, todo_id, replan_obligation_id -def _apply_requested_quota_action_selection_preflight( +def _requested_quota_action_selection_preflight( payload: dict[str, object], *, requested_todo_id: str | None, receipt_bound_todo_id: str | None, receipt_bound_replan_obligation_id: str | None, -) -> bool: +) -> dict[str, object] | None: if not requested_todo_id or ( receipt_bound_todo_id or receipt_bound_replan_obligation_id ): - return False - selected_todo = payload.get("selected_todo") - selected_todo_id = ( - normalize_todo_id(selected_todo.get("todo_id")) - if isinstance(selected_todo, Mapping) - else None + return None + interaction = payload.get("interaction_contract") + agent_channel = interaction.get("agent_channel") if isinstance(interaction, Mapping) else None + agent_channel = agent_channel if isinstance(agent_channel, Mapping) else {} + selected_todo_id, admitted = current_action_selection_admission( + payload, requested_todo_id=requested_todo_id, + agent_must_attempt=agent_channel.get("must_attempt") is True, + agent_delivery_refused=agent_channel.get("delivery_allowed") is False, ) + if admitted: + return None qualification_value = payload.get("action_selection_qualification") - qualification: Mapping[str, object] = ( - qualification_value if isinstance(qualification_value, Mapping) else {} - ) - if selected_todo_id is None and str(qualification.get("state") or "") == ( - "qualified" - ): - # An unsettled-host-turn recovery decision carries no top-level - # `selected_todo`: its qualification names the Todo that prior Turn - # has to settle, and binding the guard to that Todo is the documented - # closeout path rather than a conflict with the projection. - qualification_selected = qualification.get("selected_todo") - selected_todo_id = ( - normalize_todo_id(qualification_selected.get("todo_id")) - if isinstance(qualification_selected, Mapping) - else None - ) - selection_binding = ( - selected_todo.get("selection_binding") - if isinstance(selected_todo, Mapping) - else None - ) - execution_obligation_value = payload.get("execution_obligation") - execution_obligation: Mapping[str, object] = ( - execution_obligation_value - if isinstance(execution_obligation_value, Mapping) - else {} - ) - interaction_value = payload.get("interaction_contract") - interaction: Mapping[str, object] = ( - interaction_value if isinstance(interaction_value, Mapping) else {} - ) - agent_channel_value = interaction.get("agent_channel") - agent_channel: Mapping[str, object] = ( - agent_channel_value if isinstance(agent_channel_value, Mapping) else {} - ) - pending_selection_delivery_qualified = ( - selection_binding == "pending_action_selection" - and payload.get("normal_delivery_allowed") is True - ) - pending_selection_workspace_repair_qualified = ( - selection_binding == "pending_action_selection" - and payload.get("workspace_repair_allowed") is True - and payload.get("effective_action") == EffectiveAction.AGENT_WORKSPACE_REPAIR.value - and execution_obligation.get("kind") == "agent_workspace_repair" - and execution_obligation.get("must_attempt_work") is True - and agent_channel.get("must_attempt") is True - and agent_channel.get("delivery_allowed") is False - ) - exact_current_obligation_qualified = ( - selection_binding != "pending_action_selection" - and execution_obligation.get("must_attempt_work") is True - and agent_channel.get("must_attempt") is True - ) - if ( - selected_todo_id == requested_todo_id - and payload.get("ok") is True - and payload.get("should_run") is True - and ( - pending_selection_delivery_qualified - or pending_selection_workspace_repair_qualified - or exact_current_obligation_qualified - ) - ): - return False - + qualification = qualification_value if isinstance(qualification_value, Mapping) else {} if not isinstance(qualification_value, Mapping): raise QuotaActionSelectionConflictError( QuotaActionSelectionConflictKind.UNQUALIFIED, @@ -298,39 +241,44 @@ def _apply_requested_quota_action_selection_preflight( if deferred else "quota_action_selection_rejected" ) - payload.update( - { - "ok": False, - "decision": "skip", - "should_run": False, - "effective_action": EffectiveAction.QUOTA_SKIP.value, - "state": error_code, - "waiting_on": "codex", - "status": error_code, - "error_code": error_code, - "reason": ( - "explicit action selection was deferred by the current " - f"delivery frontier: {qualification_reason}" - if deferred - else "explicit action selection is not currently eligible: " - f"{qualification_reason}" - ), - "recommended_action": ( - "handle the current delivery preemption, then rerun quota " - "should-run with the same --turn-instance-id; omit --todo-id " - "first when a refreshed action portfolio is needed" - if deferred - else "the due monitor is visible as auxiliary context, not an " - "independently selectable action in the current advancement lane; " - "choose a current advancement Todo, or rerun after the monitor " - "becomes the hard lane" - if auxiliary_monitor - else "rerun quota should-run with the same --turn-instance-id " - "without --todo-id, then choose a currently eligible Todo" - ), - } - ) - return True + return { + "ok": False, + "spend_allowed_now": False, + "spend_after_validation": False, + "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, + "actionable_by_codex": False, + "effective_action": EffectiveAction.QUOTA_SKIP.value, + "state": error_code, + "waiting_on": "codex", + "status": error_code, + "error_code": error_code, + "reason": ( + "explicit action selection was deferred by the current " + f"delivery frontier: {qualification_reason}" + if deferred + else "explicit action selection is not currently eligible: " + f"{qualification_reason}" + ), + "recommended_action": ( + "handle the current delivery preemption, then rerun quota " + "should-run with the same --turn-instance-id; omit --todo-id " + "first when a refreshed action portfolio is needed" + if deferred + else "the due monitor is visible as auxiliary context, not an " + "independently selectable action in the current advancement lane; " + "choose a current advancement Todo, or rerun after the monitor " + "becomes the hard lane" + if auxiliary_monitor + else "rerun quota should-run with the same --turn-instance-id " + "without --todo-id, then choose a currently eligible Todo" + ), + } def _reconcile_requested_quota_action_selection( @@ -342,25 +290,22 @@ def _reconcile_requested_quota_action_selection( receipt_bound_todo_id: str | None, receipt_bound_replan_obligation_id: str | None, ) -> bool: - rejected = _apply_requested_quota_action_selection_preflight( + recovery = _requested_quota_action_selection_preflight( payload, requested_todo_id=_requested_quota_action_todo_id(args), receipt_bound_todo_id=receipt_bound_todo_id, receipt_bound_replan_obligation_id=receipt_bound_replan_obligation_id, ) - if rejected: - apply_action_selection_recovery( - payload, registry_path=str(registry_path), runtime_root=str(context.runtime_root), - goal_id=args.goal_id, agent_id=args.agent_id, - turn_instance_id=context.heartbeat_turn_id, - available_capabilities=args.available_capabilities, - scheduler_args=render_scheduler_execution_args( - scheduler_execution_context=context.scheduler_context), - ) - obligation = payload.get("execution_obligation") - if isinstance(obligation, dict): - obligation.update(must_attempt_work=False, delivery_allowed=False, - reason=payload["recommended_action"]) - return rejected + if recovery is None: + return False + payload.update(action_selection_recovery_fields(payload, recovery)) + bind_action_selection_recovery_command( + payload, registry_path=str(registry_path), runtime_root=str(context.runtime_root), + goal_id=args.goal_id, agent_id=args.agent_id, + turn_instance_id=context.heartbeat_turn_id, + available_capabilities=args.available_capabilities, + scheduler_args=render_scheduler_execution_args(scheduler_execution_context=context.scheduler_context), + ) + return True def _attach_uncommitted_action_selection_receipt( diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 631a5c7619..c532e4608a 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -117,6 +117,7 @@ from ..work_items.goal_route_hint import build_goal_route_hint from ..work_items.interaction_contract import ( build_interaction_contract, + unadmitted_action_selection, build_protocol_action_packet, finalize_user_gate_notification_cooldown, ) @@ -1445,7 +1446,8 @@ def _build_quota_should_run_payload( payload, replay_phase=prepared.receipt_bound_replay_phase, ) - if isinstance(payload.get("autonomous_replan_obligation"), dict): + if (isinstance(payload.get("autonomous_replan_obligation"), dict) + and not unadmitted_action_selection(payload)): payload["replan_action_packet"] = build_replan_action_packet( payload["autonomous_replan_obligation"], goal_id=prepared.safe_goal_id, diff --git a/loopx/control_plane/work_items/action_selection_contract.py b/loopx/control_plane/work_items/action_selection_contract.py index 18352d8adb..ee033291fb 100644 --- a/loopx/control_plane/work_items/action_selection_contract.py +++ b/loopx/control_plane/work_items/action_selection_contract.py @@ -5,6 +5,8 @@ from typing import Any from ..agents.capability_gate import runtime_capabilities_for_cli_projection +from ..todos.contract import normalize_todo_id +from ..quota.effective_action import EffectiveAction def render_cli_command_prefix(*, runtime_root: str | None = None) -> str: @@ -157,42 +159,145 @@ def delivery_spend_allowed( ) -def apply_action_selection_recovery( - payload: dict[str, Any], - *, - registry_path: str, - runtime_root: str, - goal_id: str, - agent_id: str, - turn_instance_id: str | None, - scheduler_args: str, - available_capabilities: Any = None, -) -> None: - """Render typed selection recovery before offering any settlement effects.""" - qualification = payload.get("action_selection_qualification") or {} +def current_action_selection_admission( + payload: Mapping[str, Any], *, requested_todo_id: str, + agent_must_attempt: bool, agent_delivery_refused: bool, +) -> tuple[str | None, bool]: + """Read current obligation admission, including repair and monitor positives.""" + selected_todo = payload.get("selected_todo") + selected_todo_id = ( + normalize_todo_id(selected_todo.get("todo_id")) + if isinstance(selected_todo, Mapping) + else None + ) + qualification_value = payload.get("action_selection_qualification") + qualification: Mapping[str, object] = ( + qualification_value if isinstance(qualification_value, Mapping) else {} + ) + if selected_todo_id is None and str(qualification.get("state") or "") == ( + "qualified" + ): + # An unsettled-host-turn recovery decision carries no top-level + # `selected_todo`: its qualification names the Todo that prior Turn + # has to settle, and binding the guard to that Todo is the documented + # closeout path rather than a conflict with the projection. + qualification_selected = qualification.get("selected_todo") + selected_todo_id = ( + normalize_todo_id(qualification_selected.get("todo_id")) + if isinstance(qualification_selected, Mapping) + else None + ) + selection_binding = ( + selected_todo.get("selection_binding") + if isinstance(selected_todo, Mapping) + else None + ) + execution_obligation_value = payload.get("execution_obligation") + execution_obligation: Mapping[str, object] = ( + execution_obligation_value + if isinstance(execution_obligation_value, Mapping) + else {} + ) + pending_selection_delivery_qualified = ( + selection_binding == "pending_action_selection" + and payload.get("normal_delivery_allowed") is True + ) + pending_selection_workspace_repair_qualified = ( + selection_binding == "pending_action_selection" + and payload.get("workspace_repair_allowed") is True + and payload.get("effective_action") == EffectiveAction.AGENT_WORKSPACE_REPAIR.value + and execution_obligation.get("kind") == "agent_workspace_repair" + and execution_obligation.get("must_attempt_work") is True + and agent_must_attempt + and agent_delivery_refused + ) + exact_current_obligation_qualified = ( + selection_binding != "pending_action_selection" + and execution_obligation.get("must_attempt_work") is True + and agent_must_attempt + ) + if ( + selected_todo_id == requested_todo_id + and payload.get("ok") is True + and payload.get("should_run") is True + and ( + pending_selection_delivery_qualified + or pending_selection_workspace_repair_qualified + or exact_current_obligation_qualified + ) + ): + return selected_todo_id, True + + return selected_todo_id, False + + +def action_selection_needs_recovery( + payload: Mapping[str, Any], *, agent_must_attempt: bool = False, + agent_delivery_refused: bool = False, +) -> bool: + """Read the typed qualifier's result without reimplementing admission.""" + qualification = payload.get("action_selection_qualification") + if not isinstance(qualification, Mapping) or qualification.get("state") not in {"deferred", "rejected"}: + return False + # A committed binding is reconciled by the receipt owner, not by pending + # selection recovery. Preserve the same exemption as CLI preflight. + selected = payload.get("selected_todo") + replan = payload.get("autonomous_replan_obligation") + if any(isinstance(value, Mapping) and value.get("selection_binding") == "heartbeat_receipt" + for value in (selected, replan)): + return False + _, admitted = current_action_selection_admission( + payload, requested_todo_id=str(qualification.get("requested_todo_id") or ""), + agent_must_attempt=agent_must_attempt, agent_delivery_refused=agent_delivery_refused, + ) + if admitted: + return False if qualification.get("recovery_action") != "reenter_guard_without_selection": raise RuntimeError("rejected action selection omitted its typed recovery action") - argv = ["loopx", "--registry", registry_path, "--runtime-root", runtime_root, - "--format", "json", "quota", "should-run", "--goal-id", goal_id, - "--agent-id", agent_id] + return True + + +def action_selection_recovery_command( + *, registry_path: str | None = None, runtime_root: str | None = None, + goal_id: str, agent_id: str | None, turn_instance_id: str | None, + scheduler_args: str, available_capabilities: Any = None, +) -> str: + argv = ["loopx"] + if registry_path: + argv.extend(["--registry", registry_path]) + if runtime_root: + argv.extend(["--runtime-root", runtime_root]) + argv.extend(["--format", "json", "quota", "should-run", "--goal-id", goal_id]) + if agent_id: + argv.extend(["--agent-id", agent_id]) if turn_instance_id: argv.extend(["--turn-instance-id", turn_instance_id]) for capability in runtime_capabilities_for_cli_projection(available_capabilities): argv.extend(["--available-capability", capability]) - command = shlex.join(argv) + scheduler_args - payload["spend_allowed_now"] = False - payload["spend_after_validation"] = False - # The current replan has not been admitted for this turn. Keeping its - # action packet would replace recovery in the compact TurnEnvelope. - payload.pop("replan_action_packet", None) - interaction = payload.get("interaction_contract") or {} - agent = interaction.get("agent_channel") or {} - agent.update(must_attempt=False, delivery_allowed=False, primary_action=command) - cli = interaction.get("cli_channel") or {} - for field in ("settlement_plan", "replan_settlement_contract", "selection_command", "selection_policy_ref"): - cli.pop(field, None) - cli.update(next_cli_actions=[command], selection_required=False, - spend_allowed_now=False, spend_after_validation=False, - spend_policy="rerun this turn's guard before delivery or settlement") - interaction.update(agent_channel=agent, cli_channel=cli) - payload["interaction_contract"] = interaction + return shlex.join(argv) + scheduler_args + + +def action_selection_recovery_cli_channel(command: str) -> dict[str, Any]: + return { + "next_cli_actions": [command], "selection_required": False, + "spend_allowed_now": False, "spend_after_validation": False, + "spend_policy": "rerun this turn's guard before delivery or settlement", + } + + +def bind_action_selection_recovery_command( + payload: dict[str, Any], *, registry_path: str, runtime_root: str, + goal_id: str, agent_id: str | None, turn_instance_id: str | None, + scheduler_args: str, available_capabilities: Any = None, +) -> None: + """Bind the existing recovery projection to the invoking CLI's exact argv.""" + if not action_selection_needs_recovery(payload): + raise RuntimeError("selection recovery binding requires a typed recovery result") + command = action_selection_recovery_command( + registry_path=registry_path, runtime_root=runtime_root, goal_id=goal_id, + agent_id=agent_id, turn_instance_id=turn_instance_id, + scheduler_args=scheduler_args, available_capabilities=available_capabilities, + ) + interaction = payload["interaction_contract"] + interaction["agent_channel"]["primary_action"] = command + interaction["cli_channel"]["next_cli_actions"] = [command] diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index a5ccbde6de..0d720211e8 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -662,6 +662,20 @@ def _terminal_cli_actions( return ["no quota spend without validated transition/blocker writeback"] +def _selection_recovery_command( + payload: dict[str, Any], *, available_capabilities: Any, + scheduler_execution_context: Mapping[str, Any] | SchedulerExecutionContextResolution | None, + turn_instance_id: str | None, runtime_root: str | None, +) -> str: + identity = payload.get("agent_identity") if isinstance(payload.get("agent_identity"), dict) else {} + return selection.action_selection_recovery_command( + goal_id=str(payload.get("goal_id") or ""), + agent_id=identity.get("agent_id"), runtime_root=runtime_root, + turn_instance_id=turn_instance_id, available_capabilities=available_capabilities, + scheduler_args=render_scheduler_execution_args(scheduler_execution_context=scheduler_execution_context), + ) + + def interaction_next_cli_actions( payload: dict[str, Any], *, @@ -676,6 +690,12 @@ def interaction_next_cli_actions( turn_instance_id: str | None = None, runtime_root: str | None = None, ) -> list[str]: + if unadmitted_action_selection(payload): + return [_selection_recovery_command( + payload, available_capabilities=available_capabilities, + scheduler_execution_context=scheduler_execution_context, + turn_instance_id=turn_instance_id, runtime_root=runtime_root, + )] goal_id = str(payload.get("goal_id") or "") command_prefix = selection.render_cli_command_prefix(runtime_root=runtime_root) agent_identity = payload.get("agent_identity") if isinstance(payload.get("agent_identity"), dict) else {} @@ -1277,6 +1297,12 @@ def _build_interaction_cli_channel( turn_instance_id: str | None = None, runtime_root: str | None = None, ) -> dict[str, Any]: + if unadmitted_action_selection(payload): + return selection.action_selection_recovery_cli_channel(_selection_recovery_command( + payload, available_capabilities=available_capabilities, + scheduler_execution_context=scheduler_execution_context, + turn_instance_id=turn_instance_id, runtime_root=runtime_root, + )) spend_after_selection = selection.delivery_spend_allowed(payload, spend_after_validation) settlement_plan, replan_settlement_contract = ( _turn_scoped_cli_settlement_context( @@ -1434,6 +1460,63 @@ def _interaction_fallback_policy_required(payload: dict[str, Any], *, mode: str) } or bool(payload.get("blocked_priority_fallback")) +def _interaction_execution_flags( + payload: dict[str, Any], *, mode: str, user_required: bool, +) -> tuple[bool, bool]: + execution_obligation = payload.get("execution_obligation") if isinstance(payload.get("execution_obligation"), dict) else {} + scoped_user_gate_fallback = mode == "scoped_user_gate_fallback" + bounded_delivery_with_user_notice = mode == "bounded_delivery_with_user_notice" + must_attempt = _interaction_must_attempt( + execution_obligation, + mode=mode, + user_required=user_required, + scoped_user_gate_fallback=scoped_user_gate_fallback, + bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, + ) + if mode == "automation_prompt_upgrade": + must_attempt = True + if _blocked_successor_wait_observation_required(payload): + must_attempt = True + delivery_allowed = _interaction_delivery_allowed( + payload, + execution_obligation, + mode=mode, + user_required=user_required, + scoped_user_gate_fallback=scoped_user_gate_fallback, + bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, + ) + return must_attempt, delivery_allowed + + +def action_selection_recovery_fields( + payload: Mapping[str, Any], recovery: dict[str, Any], +) -> dict[str, Any]: + """Complete the one preflight result in the owning projection module.""" + obligation = payload.get("execution_obligation") + if isinstance(obligation, dict): + recovery["execution_obligation"] = { + **obligation, "must_attempt_work": False, "delivery_allowed": False, + "reason": recovery["recommended_action"], + } + recommendation = payload.get("heartbeat_recommendation") + if isinstance(recommendation, dict): + recovery["heartbeat_recommendation"] = {**recommendation, "agent_must_attempt": False} + return recovery + + +def unadmitted_action_selection(payload: dict[str, Any]) -> bool: + """Use the same current-obligation facts as interaction and CLI preflight.""" + qualification = payload.get("action_selection_qualification") + if not isinstance(qualification, Mapping) or qualification.get("state") not in {"deferred", "rejected"}: + return False + mode = _interaction_mode(payload) + user_required = False if payload.get("agent_work_mode") == "monitor_only" else user_channel_action_required(payload) + must_attempt, delivery_allowed = _interaction_execution_flags(payload, mode=mode, user_required=user_required) + return selection.action_selection_needs_recovery( + payload, agent_must_attempt=must_attempt, agent_delivery_refused=delivery_allowed is False, + ) + + def build_interaction_contract( payload: dict[str, Any], *, @@ -1457,27 +1540,7 @@ def build_interaction_contract( mode = _interaction_mode(payload) monitor_only = payload.get("agent_work_mode") == "monitor_only" user_required = False if monitor_only else user_channel_action_required(payload) - scoped_user_gate_fallback = mode == "scoped_user_gate_fallback" - bounded_delivery_with_user_notice = mode == "bounded_delivery_with_user_notice" - must_attempt = _interaction_must_attempt( - execution_obligation, - mode=mode, - user_required=user_required, - scoped_user_gate_fallback=scoped_user_gate_fallback, - bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, - ) - if mode == "automation_prompt_upgrade": - must_attempt = True - if _blocked_successor_wait_observation_required(payload): - must_attempt = True - delivery_allowed = _interaction_delivery_allowed( - payload, - execution_obligation, - mode=mode, - user_required=user_required, - scoped_user_gate_fallback=scoped_user_gate_fallback, - bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, - ) + must_attempt, delivery_allowed = _interaction_execution_flags(payload, mode=mode, user_required=user_required) quiet_noop_allowed = _interaction_quiet_noop_allowed( mode=mode, user_required=user_required, @@ -1507,14 +1570,25 @@ def build_interaction_contract( "notify": "DONT_NOTIFY", "reason": payload.get("reason"), } - agent_channel = _build_interaction_agent_channel( - payload, - mode=mode, - must_attempt=must_attempt, - delivery_allowed=delivery_allowed, - quiet_noop_allowed=quiet_noop_allowed, - capability_reentry=capability_reentry, - ) + if unadmitted_action_selection(payload): + agent_channel = { + "must_attempt": False, "delivery_allowed": False, + "quiet_noop_allowed": quiet_noop_allowed, + "primary_action": _selection_recovery_command( + payload, available_capabilities=available_capabilities, + scheduler_execution_context=scheduler_execution_context, + turn_instance_id=turn_instance_id, runtime_root=runtime_root, + ), + } + else: + agent_channel = _build_interaction_agent_channel( + payload, + mode=mode, + must_attempt=must_attempt, + delivery_allowed=delivery_allowed, + quiet_noop_allowed=quiet_noop_allowed, + capability_reentry=capability_reentry, + ) contract: dict[str, Any] = { "schema_version": INTERACTION_CONTRACT_SCHEMA_VERSION, "mode": mode, diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json index f800f8e64b..a1726b9bff 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -496,7 +496,7 @@ "unsettled_host_turn_recovery": "A host Turn is unsettled and must be recovered before anything else: the selected Todo and action portfolio are dropped, should_run is set, and normal, recovery and self-repair delivery are all refused." }, "producers": [ - "loopx/cli_commands/quota.py::_apply_requested_quota_action_selection_preflight", + "loopx/cli_commands/quota.py::_requested_quota_action_selection_preflight", "loopx/control_plane/quota/decision_summary.py::_task_orchestration_effective_action", "loopx/control_plane/quota/decision_summary.py::quota_effective_action", "loopx/control_plane/quota/decision_summary.py::resolve_quota_run_decision", diff --git a/tests/control_plane/test_quota_action_selection_conflict.py b/tests/control_plane/test_quota_action_selection_conflict.py index 5bd0e8cd1a..c2fe422a43 100644 --- a/tests/control_plane/test_quota_action_selection_conflict.py +++ b/tests/control_plane/test_quota_action_selection_conflict.py @@ -7,7 +7,7 @@ import pytest -from loopx.cli_commands.quota import _apply_requested_quota_action_selection_preflight +from loopx.cli_commands.quota import _requested_quota_action_selection_preflight from loopx.cli_commands.quota_failure_report import quota_failure_payload from loopx.control_plane.quota.error_codes import ( QuotaActionSelectionConflictError, @@ -35,7 +35,7 @@ def _payload(**overrides: object) -> dict[str, object]: def _raise(payload: dict[str, object]) -> QuotaActionSelectionConflictError: with pytest.raises(QuotaActionSelectionConflictError) as raised: - _apply_requested_quota_action_selection_preflight( + _requested_quota_action_selection_preflight( payload, requested_todo_id=REQUESTED_TODO_ID, receipt_bound_todo_id=None, @@ -75,14 +75,14 @@ def test_missing_qualification_is_typed_rather_than_unexplained() -> None: def test_a_qualified_selection_for_the_requested_todo_is_not_a_conflict() -> None: - is_conflict = _apply_requested_quota_action_selection_preflight( + is_conflict = _requested_quota_action_selection_preflight( _payload(action_selection_qualification=_qualified_for(REQUESTED_TODO_ID)), requested_todo_id=REQUESTED_TODO_ID, receipt_bound_todo_id=None, receipt_bound_replan_obligation_id=None, ) - assert is_conflict is False + assert is_conflict is None def test_failure_payload_reports_the_conflict_instead_of_collection_failure() -> None: diff --git a/tests/control_plane/test_unadmitted_selection_construction.py b/tests/control_plane/test_unadmitted_selection_construction.py new file mode 100644 index 0000000000..d2aafc6df5 --- /dev/null +++ b/tests/control_plane/test_unadmitted_selection_construction.py @@ -0,0 +1,115 @@ +"""Typed rejection builds recovery before any settlement capability exists.""" +from __future__ import annotations + +from copy import deepcopy +import shlex + +import pytest + +from loopx.cli_commands.quota import _requested_quota_action_selection_preflight +from loopx.control_plane.work_items import interaction_contract +from loopx.control_plane.work_items.action_selection_contract import ( + action_selection_needs_recovery, + bind_action_selection_recovery_command, +) + + +def _source(state: str) -> dict: + return { + "goal_id": "selection-fixture", "agent_identity": {"agent_id": "fixture-agent"}, + "ok": True, "should_run": True, "effective_action": "autonomous_replan", + "recommended_action": "Handle the preempting replan", + "execution_obligation": {"must_attempt_work": True, "delivery_allowed": True}, + "action_selection_qualification": { + "schema_version": "action_selection_qualification_v0", "state": state, + "requested_todo_id": "todo_pending", "reason": "current_delivery_gate", + "recovery_action": "reenter_guard_without_selection", + }, + } + + +@pytest.mark.parametrize("state", ["deferred", "rejected"]) +def test_recovery_never_constructs_settlement_or_executable_primary_action(monkeypatch, state): + def unexpected(*args, **kwargs): + pytest.fail("unadmitted selection constructed an executable projection") + + monkeypatch.setattr(interaction_contract, "_turn_scoped_cli_settlement_context", unexpected) + monkeypatch.setattr(interaction_contract, "build_primary_action_projection", unexpected) + source = _source(state) + contract = interaction_contract.build_interaction_contract( + source, turn_instance_id="selection-turn", runtime_root="/runtime with spaces", + available_capabilities=["shell"], + ) + source["interaction_contract"] = contract + assert contract["agent_channel"]["must_attempt"] is False + assert contract["agent_channel"]["delivery_allowed"] is False + cli = contract["cli_channel"] + assert cli["spend_after_validation"] is False + assert cli["selection_required"] is False + for field in ("settlement_plan", "replan_settlement_contract", "selection_command", "selection_policy_ref"): + assert field not in cli + bind_action_selection_recovery_command( + source, registry_path="/registry with spaces.json", runtime_root="/runtime with spaces", + goal_id="selection-fixture", agent_id="fixture-agent", turn_instance_id="selection-turn", + scheduler_args=" --codex-app", available_capabilities=["shell"], + ) + [command] = cli["next_cli_actions"] + assert contract["agent_channel"]["primary_action"] == command + argv = shlex.split(command) + for flag, value in {"--registry": "/registry with spaces.json", "--runtime-root": "/runtime with spaces", + "--goal-id": "selection-fixture", "--agent-id": "fixture-agent", + "--turn-instance-id": "selection-turn", "--available-capability": "shell"}.items(): + assert argv[argv.index(flag) + 1] == value + assert "--codex-app" in argv + assert "--todo-id" not in argv and "--replan-obligation-id" not in argv + + +@pytest.mark.parametrize("state", ["deferred", "rejected"]) +def test_preflight_returns_one_result_without_mutating_source(state): + source = _source(state) + before = deepcopy(source) + result = _requested_quota_action_selection_preflight( + source, requested_todo_id="todo_pending", receipt_bound_todo_id=None, + receipt_bound_replan_obligation_id=None, + ) + assert source == before + assert result["error_code"] == f"quota_action_selection_{state}" + assert result["should_run"] is False + for flag in ("normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed", + "capability_repair_allowed", "workspace_repair_allowed", "actionable_by_codex"): + assert result[flag] is False + assert result["spend_after_validation"] is False + + +@pytest.mark.parametrize("bound_field", ["selected_todo", "autonomous_replan_obligation"]) +def test_committed_binding_is_owned_by_receipt_reconciliation(bound_field): + source = _source("deferred") + source[bound_field] = {"selection_binding": "heartbeat_receipt"} + assert action_selection_needs_recovery(source) is False + + +def test_qualified_selection_is_not_recovery(): + assert action_selection_needs_recovery(_source("qualified")) is False + + +@pytest.mark.parametrize("delivery_allowed", [False, True, None]) +def test_pending_workspace_repair_requires_explicit_delivery_refusal(delivery_allowed): + from loopx.control_plane.quota.error_codes import QuotaActionSelectionConflictError + + source = _source("qualified") + source.update( + effective_action="agent_workspace_repair", workspace_repair_allowed=True, + selected_todo={"todo_id": "todo_pending", "selection_binding": "pending_action_selection"}, + execution_obligation={"kind": "agent_workspace_repair", "must_attempt_work": True}, + ) + agent = {"must_attempt": True} + if delivery_allowed is not None: + agent["delivery_allowed"] = delivery_allowed + source["interaction_contract"] = {"agent_channel": agent} + kwargs = dict(requested_todo_id="todo_pending", receipt_bound_todo_id=None, + receipt_bound_replan_obligation_id=None) + if delivery_allowed is False: + assert _requested_quota_action_selection_preflight(source, **kwargs) is None + else: + with pytest.raises(QuotaActionSelectionConflictError): + _requested_quota_action_selection_preflight(source, **kwargs) From f50ec115ac1266c30015da376ee8d3d6696d01d4 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 04:33:28 -0400 Subject: [PATCH 2/8] refactor(work-items): keep execution flags with primary action projection Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- .../control_plane/quota/should_run_packet.py | 8 +- .../work_items/interaction_contract.py | 108 ++---------------- .../work_items/primary_action.py | 95 +++++++++++++++ 3 files changed, 108 insertions(+), 103 deletions(-) diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index c532e4608a..b0a3003d24 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -1426,12 +1426,8 @@ def _build_quota_should_run_payload( next_action_warning=route.next_action_warning, replan_obligation=prepared.replan_obligation, ) - bounded_research_frontier = ( - prepared.status_payload.get("bounded_research_frontier") - if isinstance( - prepared.status_payload.get("bounded_research_frontier"), dict - ) - else None + bounded_research_frontier = _dict_field( + prepared.status_payload, "bounded_research_frontier" ) _attach_truthy_fields( payload, diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 0d720211e8..34e4d0ffdc 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -44,6 +44,8 @@ from . import runtime_capability_reentry as capability_reentry_adapter from .primary_action import ( build_primary_action_projection, + interaction_execution_flags, + interaction_quiet_noop_allowed, protocol_action_label as _protocol_action_label, protocol_action_text, protocol_first_candidate_action as _protocol_first_candidate_action, @@ -1064,72 +1066,6 @@ def _blocked_priority_fallback_user_reason(payload: dict[str, Any]) -> str | Non return blocked_priority_fallback_owner_reason(fallback) -def _interaction_must_attempt( - execution_obligation: dict[str, Any], - *, - mode: str, - user_required: bool, - scoped_user_gate_fallback: bool, - bounded_delivery_with_user_notice: bool, -) -> bool: - if mode == "governed_capability_intent": - return bool(execution_obligation.get("must_attempt_work")) - if user_required and not ( - scoped_user_gate_fallback or bounded_delivery_with_user_notice - ): - return False - return bool(execution_obligation.get("must_attempt_work")) - - -def _interaction_delivery_allowed( - payload: dict[str, Any], - execution_obligation: dict[str, Any], - *, - mode: str, - user_required: bool, - scoped_user_gate_fallback: bool, - bounded_delivery_with_user_notice: bool, -) -> bool: - if mode == "governed_capability_intent": - return bool(execution_obligation.get("must_attempt_work")) - if mode == "mapped_noop_if_unchanged": - return False - if user_required and not ( - scoped_user_gate_fallback or bounded_delivery_with_user_notice - ): - return False - return bool( - execution_obligation.get( - "delivery_allowed", - payload.get("normal_delivery_allowed") - or payload.get("recovery_delivery_allowed") - or payload.get("self_repair_allowed") - or payload.get("should_run"), - ) - ) - - -def _interaction_quiet_noop_allowed( - *, - mode: str, - user_required: bool, - must_attempt: bool, -) -> bool: - if user_required or must_attempt: - return False - return _agent_scope_frontier_action(mode) is not None or mode in { - "monitor_quiet_skip", - "mapped_noop_if_unchanged", - "quota_throttled", - "blocked_wait", - "user_gate_cooldown_wait", - "terminal_no_followup", - "peer_coordination_blocked", - "agent_monitor_only", - "skip", - } - - def _interaction_spend_after_validation(mode: str) -> bool: return mode in { "bounded_delivery", @@ -1460,34 +1396,6 @@ def _interaction_fallback_policy_required(payload: dict[str, Any], *, mode: str) } or bool(payload.get("blocked_priority_fallback")) -def _interaction_execution_flags( - payload: dict[str, Any], *, mode: str, user_required: bool, -) -> tuple[bool, bool]: - execution_obligation = payload.get("execution_obligation") if isinstance(payload.get("execution_obligation"), dict) else {} - scoped_user_gate_fallback = mode == "scoped_user_gate_fallback" - bounded_delivery_with_user_notice = mode == "bounded_delivery_with_user_notice" - must_attempt = _interaction_must_attempt( - execution_obligation, - mode=mode, - user_required=user_required, - scoped_user_gate_fallback=scoped_user_gate_fallback, - bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, - ) - if mode == "automation_prompt_upgrade": - must_attempt = True - if _blocked_successor_wait_observation_required(payload): - must_attempt = True - delivery_allowed = _interaction_delivery_allowed( - payload, - execution_obligation, - mode=mode, - user_required=user_required, - scoped_user_gate_fallback=scoped_user_gate_fallback, - bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, - ) - return must_attempt, delivery_allowed - - def action_selection_recovery_fields( payload: Mapping[str, Any], recovery: dict[str, Any], ) -> dict[str, Any]: @@ -1511,7 +1419,10 @@ def unadmitted_action_selection(payload: dict[str, Any]) -> bool: return False mode = _interaction_mode(payload) user_required = False if payload.get("agent_work_mode") == "monitor_only" else user_channel_action_required(payload) - must_attempt, delivery_allowed = _interaction_execution_flags(payload, mode=mode, user_required=user_required) + must_attempt, delivery_allowed = interaction_execution_flags( + payload, mode=mode, user_required=user_required, + blocked_successor_wait_observation=_blocked_successor_wait_observation_required(payload), + ) return selection.action_selection_needs_recovery( payload, agent_must_attempt=must_attempt, agent_delivery_refused=delivery_allowed is False, ) @@ -1540,8 +1451,11 @@ def build_interaction_contract( mode = _interaction_mode(payload) monitor_only = payload.get("agent_work_mode") == "monitor_only" user_required = False if monitor_only else user_channel_action_required(payload) - must_attempt, delivery_allowed = _interaction_execution_flags(payload, mode=mode, user_required=user_required) - quiet_noop_allowed = _interaction_quiet_noop_allowed( + must_attempt, delivery_allowed = interaction_execution_flags( + payload, mode=mode, user_required=user_required, + blocked_successor_wait_observation=_blocked_successor_wait_observation_required(payload), + ) + quiet_noop_allowed = interaction_quiet_noop_allowed( mode=mode, user_required=user_required, must_attempt=must_attempt, diff --git a/loopx/control_plane/work_items/primary_action.py b/loopx/control_plane/work_items/primary_action.py index a4c53af5f7..1fc5a2080d 100644 --- a/loopx/control_plane/work_items/primary_action.py +++ b/loopx/control_plane/work_items/primary_action.py @@ -343,3 +343,98 @@ def build_primary_action_projection(payload: dict[str, Any], *, mode: str) -> di if resolution_trace: projection["resolution_trace"] = resolution_trace return projection + + +def _interaction_must_attempt( + execution_obligation: dict[str, Any], + *, + mode: str, + user_required: bool, + scoped_user_gate_fallback: bool, + bounded_delivery_with_user_notice: bool, +) -> bool: + if mode == "governed_capability_intent": + return bool(execution_obligation.get("must_attempt_work")) + if user_required and not ( + scoped_user_gate_fallback or bounded_delivery_with_user_notice + ): + return False + return bool(execution_obligation.get("must_attempt_work")) + + +def _interaction_delivery_allowed( + payload: dict[str, Any], + execution_obligation: dict[str, Any], + *, + mode: str, + user_required: bool, + scoped_user_gate_fallback: bool, + bounded_delivery_with_user_notice: bool, +) -> bool: + if mode == "governed_capability_intent": + return bool(execution_obligation.get("must_attempt_work")) + if mode == "mapped_noop_if_unchanged": + return False + if user_required and not ( + scoped_user_gate_fallback or bounded_delivery_with_user_notice + ): + return False + return bool( + execution_obligation.get( + "delivery_allowed", + payload.get("normal_delivery_allowed") + or payload.get("recovery_delivery_allowed") + or payload.get("self_repair_allowed") + or payload.get("should_run"), + ) + ) + + +def interaction_quiet_noop_allowed( + *, + mode: str, + user_required: bool, + must_attempt: bool, +) -> bool: + if user_required or must_attempt: + return False + return _agent_scope_frontier_action(mode) is not None or mode in { + "monitor_quiet_skip", + "mapped_noop_if_unchanged", + "quota_throttled", + "blocked_wait", + "user_gate_cooldown_wait", + "terminal_no_followup", + "peer_coordination_blocked", + "agent_monitor_only", + "skip", + } + + +def interaction_execution_flags( + payload: dict[str, Any], *, mode: str, user_required: bool, + blocked_successor_wait_observation: bool, +) -> tuple[bool, bool]: + execution_obligation = payload.get("execution_obligation") if isinstance(payload.get("execution_obligation"), dict) else {} + scoped_user_gate_fallback = mode == "scoped_user_gate_fallback" + bounded_delivery_with_user_notice = mode == "bounded_delivery_with_user_notice" + must_attempt = _interaction_must_attempt( + execution_obligation, + mode=mode, + user_required=user_required, + scoped_user_gate_fallback=scoped_user_gate_fallback, + bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, + ) + if mode == "automation_prompt_upgrade": + must_attempt = True + if blocked_successor_wait_observation: + must_attempt = True + delivery_allowed = _interaction_delivery_allowed( + payload, + execution_obligation, + mode=mode, + user_required=user_required, + scoped_user_gate_fallback=scoped_user_gate_fallback, + bounded_delivery_with_user_notice=bounded_delivery_with_user_notice, + ) + return must_attempt, delivery_allowed From 4fd8af3017ecfa83fa489f6d9bcbe4e38793db94 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 09:22:52 -0400 Subject: [PATCH 3/8] test(semantics): lock reduced inventory ratchets Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- examples/semantic-vocabulary-drift-smoke.py | 7 +++---- loopx/semantics/vocabulary_v0.json | 6 +++--- 2 files changed, 6 insertions(+), 7 deletions(-) diff --git a/examples/semantic-vocabulary-drift-smoke.py b/examples/semantic-vocabulary-drift-smoke.py index 03f16cc4ad..3e3f30efcc 100755 --- a/examples/semantic-vocabulary-drift-smoke.py +++ b/examples/semantic-vocabulary-drift-smoke.py @@ -27,7 +27,6 @@ INVENTORY_SCHEMA_VERSION, SourceFile, build_inventory, - render_inventory, collect_string_constants, load_sources, python_facts, @@ -232,8 +231,8 @@ TWIN_ROOT_ANCHOR = "loopx/control_plane" TWIN_BUDGET_ANCHOR = 43 BUDGET_ANCHOR = { - "same_runtime_forks": 18, - "same_runtime_fork_definitions": 41, + "same_runtime_forks": 17, + "same_runtime_fork_definitions": 39, "conflicting_values": 16, "conflicting_definitions": 55, "schema_version_same_runtime_forks": 7, @@ -241,7 +240,7 @@ "multi_value_forks": 2, "multi_value_forks_semantic": 1, "multi_value_fork_definitions": 6, - "same_runtime_forks_semantic": 11, + "same_runtime_forks_semantic": 10, "conflicting_values_semantic": 0, } # Budgets for the legacy should-run decision fields, anchored the same way so a diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json index 465e60277f..f58a31b9de 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -1124,15 +1124,15 @@ }, "inventory_ratchets": { "meaning": "Counts read from the generated inventory. A same-runtime fork is one constant name with one value defined in two or more modules of the same runtime; a conflicting value is one name with different values. Both the number of affected names and the number of definitions are budgets, so a third spelling of an already-conflicting name is still a regression.", - "same_runtime_forks": 18, - "same_runtime_fork_definitions": 41, + "same_runtime_forks": 17, + "same_runtime_fork_definitions": 39, "conflicting_values": 16, "conflicting_definitions": 55, "schema_version_same_runtime_forks": 7, "multi_value_twins": 13, "multi_value_forks": 2, "multi_value_fork_definitions": 6, - "same_runtime_forks_semantic": 11, + "same_runtime_forks_semantic": 10, "conflicting_values_semantic": 0, "multi_value_meaning": "Enums, named closed sets, Literal aliases, and TypeScript as-const arrays are vocabulary exactly as a NAME = \"value\" constant is, so they get the same collision rule. One name defined in two modules with identical values is a twin; with different values it is a fork. The semantic multi-value-fork budget excludes only names declared in scope_declarations.", "multi_value_forks_note": "The 4 counted forks include SOURCE_SURFACES, whose four definitions are four CLI commands each listing its own data sources; that is bounded-context reuse of one name, not drift. It stays in the budget until M0.5 adds a scope field (RFC Section 5) and must not be removed by renaming.", From 2d85dc6d51f6b533301c2d3643e9fefc04e10818 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 09:41:08 -0400 Subject: [PATCH 4/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 e3d101e6ed29a8612791518f115d7539115a8a7c Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 20:07:03 -0400 Subject: [PATCH 5/8] fix(quota): finalize selection recovery before projections Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- loopx/cli_commands/quota.py | 58 +---------- .../quota/heartbeat_recommendation.py | 17 ++++ .../control_plane/quota/should_run_packet.py | 27 ++++++ .../work_items/action_selection_contract.py | 95 +++++++++++++++++++ .../work_items/interaction_contract.py | 16 ---- loopx/semantics/vocabulary_v0.json | 2 +- .../test_quota_settlement_cli.py | 43 +++++++++ .../test_unadmitted_selection_construction.py | 2 + 8 files changed, 188 insertions(+), 72 deletions(-) diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 95feba583e..414749992b 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -55,11 +55,9 @@ from ..control_plane.todos.contract import normalize_todo_id from ..control_plane.work_items.action_selection_contract import ( bind_action_selection_recovery_command, + build_action_selection_recovery_fields, current_action_selection_admission, ) -from ..control_plane.work_items.interaction_contract import ( - action_selection_recovery_fields, -) from ..presentation.renderers.quota_event_markdown import ( render_quota_monitor_poll_markdown, render_quota_slot_preview_markdown, @@ -275,57 +273,7 @@ def _requested_quota_action_selection_preflight( selected_todo_id=selected_todo_id, qualification_state=qualification_state, ) - qualification_reason = str( - qualification.get("reason") or "candidate_not_currently_eligible" - ) - deferred = qualification_state == "deferred" - auxiliary_monitor = ( - qualification_reason - == "auxiliary_monitor_not_selectable_in_advancement_lane" - ) - error_code = ( - "quota_action_selection_deferred" - if deferred - else "quota_action_selection_rejected" - ) - return { - "ok": False, - "spend_allowed_now": False, - "spend_after_validation": False, - "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, - "actionable_by_codex": False, - "effective_action": EffectiveAction.QUOTA_SKIP.value, - "state": error_code, - "waiting_on": "codex", - "status": error_code, - "error_code": error_code, - "reason": ( - "explicit action selection was deferred by the current " - f"delivery frontier: {qualification_reason}" - if deferred - else "explicit action selection is not currently eligible: " - f"{qualification_reason}" - ), - "recommended_action": ( - "handle the current delivery preemption, then rerun quota " - "should-run with the same --turn-instance-id; omit --todo-id " - "first when a refreshed action portfolio is needed" - if deferred - else "the due monitor is visible as auxiliary context, not an " - "independently selectable action in the current advancement lane; " - "choose a current advancement Todo, or rerun after the monitor " - "becomes the hard lane" - if auxiliary_monitor - else "rerun quota should-run with the same --turn-instance-id " - "without --todo-id, then choose a currently eligible Todo" - ), - } + return build_action_selection_recovery_fields(payload) def _reconcile_requested_quota_action_selection( @@ -348,7 +296,7 @@ def _reconcile_requested_quota_action_selection( ) if recovery is None: return False - payload.update(action_selection_recovery_fields(payload, recovery)) + payload.update(recovery) bind_action_selection_recovery_command( payload, registry_path=str(registry_path), diff --git a/loopx/control_plane/quota/heartbeat_recommendation.py b/loopx/control_plane/quota/heartbeat_recommendation.py index 219194375d..50f6330423 100644 --- a/loopx/control_plane/quota/heartbeat_recommendation.py +++ b/loopx/control_plane/quota/heartbeat_recommendation.py @@ -264,6 +264,23 @@ def _recommendation(*parts: dict[str, Any]) -> dict[str, Any]: return payload +def build_action_selection_recovery_recommendation( + *, reason: str, +) -> dict[str, Any]: + """Project closed heartbeat guidance after a typed selection refusal.""" + + return _recommendation( + { + "source": "action_selection_recovery", + "recommended_mode": "quota_skip", + "notify": "DONT_NOTIFY", + "spend_policy": "no quota spend until an eligible Todo is selected", + "reason": reason, + "agent_must_attempt": False, + } + ) + + def _stall_self_repair_rule( facts: _HeartbeatRecommendationFacts, ) -> dict[str, Any] | None: diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index b0a3003d24..dc09328649 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -39,6 +39,7 @@ quota_execution_profile_summary as _quota_execution_profile_summary, ) from ..quota.heartbeat_recommendation import ( + build_action_selection_recovery_recommendation, build_heartbeat_recommendation, refine_heartbeat_recommendation, ) @@ -114,6 +115,9 @@ qualify_action_selection_from_inventory, ) from ..work_items.execution_obligation import build_execution_obligation +from ..work_items.action_selection_contract import ( + apply_action_selection_recovery_projection, +) from ..work_items.goal_route_hint import build_goal_route_hint from ..work_items.interaction_contract import ( build_interaction_contract, @@ -1442,6 +1446,29 @@ def _build_quota_should_run_payload( payload, replay_phase=prepared.receipt_bound_replay_phase, ) + if unadmitted_action_selection(payload): + for field in ( + "agent_lane_next_action", + "agent_scope_frontier", + "autonomous_replan_obligation", + "execution_profile", + "goal_route_hint", + "handoff_readiness", + "replan_action_packet", + "selected_todo", + "task_orchestration_contract", + "todo_id", + "todo_write_hint", + "work_lane_contract", + "workspace_guard", + ): + payload.pop(field, None) + apply_action_selection_recovery_projection(payload) + payload["heartbeat_recommendation"] = ( + build_action_selection_recovery_recommendation( + reason=str(payload.get("reason") or "") + ) + ) if (isinstance(payload.get("autonomous_replan_obligation"), dict) and not unadmitted_action_selection(payload)): payload["replan_action_packet"] = build_replan_action_packet( diff --git a/loopx/control_plane/work_items/action_selection_contract.py b/loopx/control_plane/work_items/action_selection_contract.py index ee033291fb..3044c2ba5a 100644 --- a/loopx/control_plane/work_items/action_selection_contract.py +++ b/loopx/control_plane/work_items/action_selection_contract.py @@ -257,6 +257,101 @@ def action_selection_needs_recovery( return True +def action_selection_recovery_fields( + recovery: dict[str, Any], +) -> dict[str, Any]: + """Complete the one preflight result in the owning projection module.""" + recovery["execution_obligation"] = { + "must_attempt_work": False, + "kind": EffectiveAction.QUOTA_SKIP.value, + "delivery_allowed": False, + "notify_is_execution_gate": False, + "reason": recovery["recommended_action"], + "spend_policy": "no quota spend until an eligible Todo is selected", + } + return recovery + + +def build_action_selection_recovery_fields( + payload: Mapping[str, Any], +) -> dict[str, Any]: + """Construct the closed root facts for one typed selection refusal.""" + + qualification = payload.get("action_selection_qualification") + if not isinstance(qualification, Mapping): + raise RuntimeError("selection recovery requires a typed qualification") + qualification_state = str(qualification.get("state") or "") + if qualification_state not in {"deferred", "rejected"}: + raise RuntimeError("selection recovery requires a deferred or rejected state") + qualification_reason = str( + qualification.get("reason") or "candidate_not_currently_eligible" + ) + deferred = qualification_state == "deferred" + auxiliary_monitor = ( + qualification_reason + == "auxiliary_monitor_not_selectable_in_advancement_lane" + ) + error_code = ( + "quota_action_selection_deferred" + if deferred + else "quota_action_selection_rejected" + ) + recovery = { + "ok": False, + "spend_allowed_now": False, + "spend_after_validation": False, + "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, + "actionable_by_codex": False, + "effective_action": EffectiveAction.QUOTA_SKIP.value, + "state": error_code, + "waiting_on": "codex", + "status": error_code, + "error_code": error_code, + "reason": ( + "explicit action selection was deferred by the current " + f"delivery frontier: {qualification_reason}" + if deferred + else "explicit action selection is not currently eligible: " + f"{qualification_reason}" + ), + "recommended_action": ( + "handle the current delivery preemption, then rerun quota " + "should-run with the same --turn-instance-id; omit --todo-id " + "first when a refreshed action portfolio is needed" + if deferred + else "the due monitor is visible as auxiliary context, not an " + "independently selectable action in the current advancement lane; " + "choose a current advancement Todo, or rerun after the monitor " + "becomes the hard lane" + if auxiliary_monitor + else "rerun quota should-run with the same --turn-instance-id " + "without --todo-id, then choose a currently eligible Todo" + ), + } + return action_selection_recovery_fields(recovery) + + +def apply_action_selection_recovery_projection(payload: dict[str, Any]) -> bool: + """Replace an unadmitted action with its closed pre-finalization facts.""" + + qualification = payload.get("action_selection_qualification") + if not isinstance(qualification, Mapping) or qualification.get("state") not in { + "deferred", + "rejected", + }: + return False + if qualification.get("recovery_action") != "reenter_guard_without_selection": + raise RuntimeError("rejected action selection omitted its typed recovery action") + payload.update(build_action_selection_recovery_fields(payload)) + return True + + def action_selection_recovery_command( *, registry_path: str | None = None, runtime_root: str | None = None, goal_id: str, agent_id: str | None, turn_instance_id: str | None, diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 6aba20f42f..8f91d84717 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -1487,22 +1487,6 @@ def _interaction_fallback_policy_required(payload: dict[str, Any], *, mode: str) } or bool(payload.get("blocked_priority_fallback")) -def action_selection_recovery_fields( - payload: Mapping[str, Any], recovery: dict[str, Any], -) -> dict[str, Any]: - """Complete the one preflight result in the owning projection module.""" - obligation = payload.get("execution_obligation") - if isinstance(obligation, dict): - recovery["execution_obligation"] = { - **obligation, "must_attempt_work": False, "delivery_allowed": False, - "reason": recovery["recommended_action"], - } - recommendation = payload.get("heartbeat_recommendation") - if isinstance(recommendation, dict): - recovery["heartbeat_recommendation"] = {**recommendation, "agent_must_attempt": False} - return recovery - - def unadmitted_action_selection(payload: dict[str, Any]) -> bool: """Use the same current-obligation facts as interaction and CLI preflight.""" qualification = payload.get("action_selection_qualification") diff --git a/loopx/semantics/vocabulary_v0.json b/loopx/semantics/vocabulary_v0.json index f58a31b9de..5e530f7268 100644 --- a/loopx/semantics/vocabulary_v0.json +++ b/loopx/semantics/vocabulary_v0.json @@ -496,7 +496,7 @@ "unsettled_host_turn_recovery": "A host Turn is unsettled and must be recovered before anything else: the selected Todo and action portfolio are dropped, should_run is set, and normal, recovery and self-repair delivery are all refused." }, "producers": [ - "loopx/cli_commands/quota.py::_requested_quota_action_selection_preflight", + "loopx/control_plane/work_items/action_selection_contract.py::build_action_selection_recovery_fields", "loopx/control_plane/quota/decision_summary.py::_task_orchestration_effective_action", "loopx/control_plane/quota/decision_summary.py::quota_effective_action", "loopx/control_plane/quota/decision_summary.py::resolve_quota_run_decision", diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index b3170f79ed..8b951e8307 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -35,6 +35,46 @@ SELECTED_REPLAN_TODO_ID = "todo_chain_000000000000" +def _assert_action_selection_recovery_projections(payload: dict[str, Any]) -> None: + from loopx.control_plane.quota.turn_envelope import build_turn_envelope + from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority + + interaction = payload["interaction_contract"] + assert interaction["mode"] == "skip" + assert interaction["agent_channel"]["must_attempt"] is False + assert interaction["agent_channel"]["delivery_allowed"] is False + assert interaction["cli_channel"]["spend_allowed_now"] is False + assert interaction["cli_channel"]["spend_after_validation"] is False + assert payload["execution_obligation"]["kind"] == "quota_skip" + assert payload["execution_obligation"]["must_attempt_work"] is False + assert payload["automation_liveness"]["automation_action"] == "keep_active" + assert payload["scheduler_hint"]["action"] == "backoff_until_state_change" + protocol_summary = payload["protocol_action_packet"]["summary"] + assert "agent_action_required=false" in protocol_summary + assert "agent_action_required=true" not in protocol_summary + assert "execute_bounded_work" not in protocol_summary + for field in ( + "autonomous_replan_obligation", + "replan_action_packet", + "selected_todo", + "work_lane_contract", + ): + assert field not in payload + + envelope = build_turn_envelope(payload) + assert envelope["contract_capsule"]["interaction_contract"]["mode"] == "skip" + assert envelope["contract_capsule"]["execution_obligation"][ + "must_attempt_work" + ] is False + assert envelope["writeback"]["spend_allowed_now"] is False + assert envelope["writeback"]["spend_after_validation"] is False + authority = extract_turn_authority({"turn_envelope": envelope}) + assert authority["primary_action"] == interaction["agent_channel"][ + "primary_action" + ] + assert authority["write_scope"] == [] + + def _write_fixture( root: Path, *, @@ -3027,6 +3067,7 @@ def test_agent_selection_rejects_unprojected_todo(tmp_path: Path) -> None: "requested_todo_id": "todo_not_projected", "reason": "candidate_not_currently_eligible", } + _assert_action_selection_recovery_projections(invalid) assert invalid["heartbeat_receipt"]["status"] == "replayed" assert invalid["rollout_event"]["appended"] is False assert _heartbeat_receipt_count(runtime, turn_instance_id) == 1 @@ -3107,6 +3148,7 @@ def test_first_call_rejected_selection_does_not_commit_a_false_receipt( assert rejected_rc == 1, rejected assert rejected["error_code"] == "quota_action_selection_rejected" + _assert_action_selection_recovery_projections(rejected) assert rejected["heartbeat_receipt"] == { "schema_version": "heartbeat_quota_receipt_v0", "turn_instance_id": turn_instance_id, @@ -3525,6 +3567,7 @@ def test_pending_action_selection_reports_autonomous_replan_preemption( "reason": "autonomous_replan", "delivery_preemptions": ["autonomous_replan", "delivery_not_allowed"], } + _assert_action_selection_recovery_projections(selected) assert selected["heartbeat_receipt"]["status"] == "selection_retained" assert selected["heartbeat_receipt"]["pending_action_selection"]["todo_id"] == ( ALTERNATIVE_TODO_ID diff --git a/tests/control_plane/test_unadmitted_selection_construction.py b/tests/control_plane/test_unadmitted_selection_construction.py index d2aafc6df5..cf12f234b2 100644 --- a/tests/control_plane/test_unadmitted_selection_construction.py +++ b/tests/control_plane/test_unadmitted_selection_construction.py @@ -75,6 +75,8 @@ def test_preflight_returns_one_result_without_mutating_source(state): assert source == before assert result["error_code"] == f"quota_action_selection_{state}" assert result["should_run"] is False + assert result["execution_obligation"]["kind"] == "quota_skip" + assert result["execution_obligation"]["must_attempt_work"] is False for flag in ("normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed", "actionable_by_codex"): assert result[flag] is False From be8b264f5077b70f29e54fe34877c9d14e5f2bcc Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 20:37:36 -0400 Subject: [PATCH 6/8] fix(quota): preserve settled replan replay precedence Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- .../control_plane/quota/should_run_packet.py | 63 ++++++++++++------- 1 file changed, 40 insertions(+), 23 deletions(-) diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index dc09328649..febdbc8a4b 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 @@ -448,6 +449,41 @@ def _apply_agent_monitor_only_precedence( clear_quota_action_projections(payload) +def _apply_unadmitted_action_selection_precedence( + payload: dict[str, Any], + *, + replay_phase: ReceiptBoundReplayPhase | None, +) -> None: + """Finalize one closed selection recovery before shared projections.""" + + if replay_phase is ReceiptBoundReplayPhase.SETTLED or not ( + unadmitted_action_selection(payload) + ): + return + for field in ( + "agent_lane_next_action", + "agent_scope_frontier", + "autonomous_replan_obligation", + "execution_profile", + "goal_route_hint", + "handoff_readiness", + "replan_action_packet", + "selected_todo", + "task_orchestration_contract", + "todo_id", + "todo_write_hint", + "work_lane_contract", + "workspace_guard", + ): + payload.pop(field, None) + apply_action_selection_recovery_projection(payload) + payload["heartbeat_recommendation"] = ( + build_action_selection_recovery_recommendation( + reason=str(payload.get("reason") or "") + ) + ) + + def _attach_truthy_fields(payload: dict[str, Any], **fields: Any) -> None: payload.update({key: value for key, value in fields.items() if value}) @@ -1446,29 +1482,10 @@ def _build_quota_should_run_payload( payload, replay_phase=prepared.receipt_bound_replay_phase, ) - if unadmitted_action_selection(payload): - for field in ( - "agent_lane_next_action", - "agent_scope_frontier", - "autonomous_replan_obligation", - "execution_profile", - "goal_route_hint", - "handoff_readiness", - "replan_action_packet", - "selected_todo", - "task_orchestration_contract", - "todo_id", - "todo_write_hint", - "work_lane_contract", - "workspace_guard", - ): - payload.pop(field, None) - apply_action_selection_recovery_projection(payload) - payload["heartbeat_recommendation"] = ( - build_action_selection_recovery_recommendation( - reason=str(payload.get("reason") or "") - ) - ) + _apply_unadmitted_action_selection_precedence( + payload, + replay_phase=prepared.receipt_bound_replay_phase, + ) if (isinstance(payload.get("autonomous_replan_obligation"), dict) and not unadmitted_action_selection(payload)): payload["replan_action_packet"] = build_replan_action_packet( From 75ac19d294b399334eae058f431c76b04c3b4af1 Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 21:12:53 -0400 Subject: [PATCH 7/8] test(quota): lock settled selection precedence Signed-off-by: song <22676124+songoow@users.noreply.github.com> --- tests/control_plane/test_quota_settlement_cli.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index 8b951e8307..c452af79fc 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -4411,7 +4411,10 @@ def test_todoless_blocked_replan_settles_read_only_external_evidence_without_wor assert replay_rc == 0, replay assert replay["effective_action"] == "heartbeat_settled_skip" + assert replay["interaction_contract"]["mode"] == "heartbeat_settled_skip" + assert replay["execution_obligation"]["kind"] == "heartbeat_settled_skip" assert replay["should_run"] is False + assert replay.get("error_code") is None assert replay.get("selected_todo") is None assert replay.get("unsettled_host_turn_recovery") is None From 7bdac93450686494b673626a7762c94792a14e6f Mon Sep 17 00:00:00 2001 From: song <22676124+songoow@users.noreply.github.com> Date: Sun, 20 Sep 2026 22:37:35 -0400 Subject: [PATCH 8/8] fix(quota): close the safe bypass an unadmitted selection inherits A refused selection already produced a closed quota_skip recovery, but a scoped user-gate fallback prepared earlier in the same packet survived it: safe_bypass_allowed stayed true, safe_bypass_kind/policy kept the fallback text, and the fallback projection short-circuited the interaction mode back to scoped_user_gate_fallback with agent_action_required=true. The shipped 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, so the host could still work and account on a Turn whose selection was refused. Close it in the recovery owner that already constructs the refusal: build_action_selection_recovery_fields() now denies the safe bypass, and the precedence pass drops scoped_user_gate_fallback with the other action projections it already clears. No new mode, schema or second cleanup owner. The fallback itself is untouched: without a refused selection the same fixture still runs it with safe_bypass_allowed=true, and a positive control pins that. The composed regression covers root, execution obligation, interaction, protocol summary, TurnEnvelope, turn authority and the rendered guidance, and both refusal states are pinned at the owner they share. 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> --- .../control_plane/quota/should_run_packet.py | 1 + .../work_items/action_selection_contract.py | 6 + .../test_unadmitted_selection_construction.py | 105 ++++++++++++++++++ 3 files changed, 112 insertions(+) diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index febdbc8a4b..fefcfc03d9 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -468,6 +468,7 @@ def _apply_unadmitted_action_selection_precedence( "goal_route_hint", "handoff_readiness", "replan_action_packet", + "scoped_user_gate_fallback", "selected_todo", "task_orchestration_contract", "todo_id", diff --git a/loopx/control_plane/work_items/action_selection_contract.py b/loopx/control_plane/work_items/action_selection_contract.py index 3044c2ba5a..e4622c48d8 100644 --- a/loopx/control_plane/work_items/action_selection_contract.py +++ b/loopx/control_plane/work_items/action_selection_contract.py @@ -308,6 +308,12 @@ def build_action_selection_recovery_fields( "capability_repair_allowed": False, "workspace_repair_allowed": False, "actionable_by_codex": False, + # An unadmitted selection grants no safe bypass either: the heartbeat + # task body reads safe_bypass_allowed as permission to run one bounded + # step and spend, so the refusal must close that authority too. + "safe_bypass_allowed": False, + "safe_bypass_kind": None, + "safe_bypass_policy": None, "effective_action": EffectiveAction.QUOTA_SKIP.value, "state": error_code, "waiting_on": "codex", diff --git a/tests/control_plane/test_unadmitted_selection_construction.py b/tests/control_plane/test_unadmitted_selection_construction.py index cf12f234b2..b3fc90d42c 100644 --- a/tests/control_plane/test_unadmitted_selection_construction.py +++ b/tests/control_plane/test_unadmitted_selection_construction.py @@ -115,3 +115,108 @@ def test_pending_workspace_repair_requires_explicit_delivery_refusal(delivery_al else: with pytest.raises(QuotaActionSelectionConflictError): _requested_quota_action_selection_preflight(source, **kwargs) + + +def _scoped_gate_status(): + """Reuse the real scoped user-gate fallback producer fixture.""" + import sys + + sys.path.insert(0, "tests/control_plane") + from test_user_gate_lane_progress import APP_CONTEXT, AGENT_ID, GOAL_ID, _status_payload + + return _status_payload(gate_action_kind="approve_product_first_screen"), { + "goal_id": GOAL_ID, "agent_id": AGENT_ID, + "scheduler_execution_context": APP_CONTEXT, + } + + +def test_scoped_fallback_cannot_execute_under_an_unadmitted_selection(): + """A runnable scoped fallback does not survive a refused selection. + + The fallback readback grants safe_bypass_allowed, and the shipped heartbeat + task body reads that under should_run=false as permission for one bounded + step plus a spend. The selection refusal has to close that authority too, + or the same packet says both "no work" and "do one step and spend". + """ + from loopx.control_plane.quota.turn_envelope import build_turn_envelope + from loopx.control_plane.turn_driver.host_candidate import extract_turn_authority + from loopx.presentation.renderers.quota_markdown import render_quota_should_run_markdown + from loopx.quota import build_quota_should_run + + status, kwargs = _scoped_gate_status() + payload = build_quota_should_run( + status, requested_action_todo_id="todo_not_projected", **kwargs, + ) + + assert payload["effective_action"] == "quota_skip" + assert payload["state"] == "quota_action_selection_rejected" + for flag in ( + "ok", "should_run", "actionable_by_codex", "safe_bypass_allowed", + "spend_allowed_now", "spend_after_validation", "normal_delivery_allowed", + "recovery_delivery_allowed", "self_repair_allowed", + "capability_repair_allowed", "workspace_repair_allowed", + ): + assert payload[flag] is False, flag + assert payload["safe_bypass_kind"] is None + assert payload["safe_bypass_policy"] is None + assert "scoped_user_gate_fallback" not in payload + + obligation = payload["execution_obligation"] + assert obligation["kind"] == "quota_skip" + assert obligation["must_attempt_work"] is False + assert obligation["delivery_allowed"] is False + assert "no quota spend" in obligation["spend_policy"] + + interaction = payload["interaction_contract"] + assert interaction["mode"] != "scoped_user_gate_fallback" + assert interaction["agent_channel"]["must_attempt"] is False + assert interaction["agent_channel"]["delivery_allowed"] is False + assert interaction["cli_channel"]["spend_allowed_now"] is False + assert interaction["cli_channel"]["spend_after_validation"] is False + + summary = payload["protocol_action_packet"]["summary"] + assert "agent_action_required=false" in summary + assert "agent_action_required=true" not in summary + + envelope = build_turn_envelope(payload) + assert envelope["writeback"]["spend_allowed_now"] is False + assert envelope["writeback"]["spend_after_validation"] is False + capsule = envelope["contract_capsule"] + assert capsule["interaction_contract"]["mode"] != "scoped_user_gate_fallback" + assert capsule["execution_obligation"]["must_attempt_work"] is False + assert extract_turn_authority({"turn_envelope": envelope})["write_scope"] == [] + + guidance = render_quota_should_run_markdown(payload) + assert "safe_bypass" not in guidance + assert "spend only after validated writeback" not in guidance + + +def test_scoped_fallback_still_runs_without_a_refused_selection(): + """The refusal closes the grant; it does not retire the fallback path.""" + from loopx.quota import build_quota_should_run + + status, kwargs = _scoped_gate_status() + payload = build_quota_should_run(status, **kwargs) + + assert payload["should_run"] is True + assert payload["safe_bypass_allowed"] is True + assert payload["safe_bypass_kind"] == "scoped_user_gate_fallback" + assert payload["interaction_contract"]["mode"] == "scoped_user_gate_fallback" + + +@pytest.mark.parametrize("state", ["deferred", "rejected"]) +def test_recovery_fields_close_the_safe_bypass_grant(state): + """Both refusal states share one owner, so both close the same authority.""" + from loopx.control_plane.work_items.action_selection_contract import ( + build_action_selection_recovery_fields, + ) + + source = _source(state) + source.update( + safe_bypass_allowed=True, safe_bypass_kind="scoped_user_gate_fallback", + safe_bypass_policy="advance the fallback; spend only after validated writeback", + ) + recovery = build_action_selection_recovery_fields(source) + assert recovery["safe_bypass_allowed"] is False + assert recovery["safe_bypass_kind"] is None + assert recovery["safe_bypass_policy"] is None