diff --git a/examples/loopx-turn-dsh-e2e-smoke.py b/examples/loopx-turn-dsh-e2e-smoke.py index 18706ff7d2..40cd7d4f3f 100755 --- a/examples/loopx-turn-dsh-e2e-smoke.py +++ b/examples/loopx-turn-dsh-e2e-smoke.py @@ -126,6 +126,10 @@ def _write_fixture(root: Path) -> tuple[Path, Path, Path, Path]: + "\n", encoding="utf-8", ) + from loopx.control_plane.testing.canary_harness import import_fixture_authority + + for record in json.loads(registry.read_text())["goals"]: + import_fixture_authority(registry, record["id"], provider="sqlite") return project, runtime, workspace, registry diff --git a/examples/loopx-turn-managed-executor-binding-smoke.py b/examples/loopx-turn-managed-executor-binding-smoke.py index 14fb5a5324..5adbfba8eb 100644 --- a/examples/loopx-turn-managed-executor-binding-smoke.py +++ b/examples/loopx-turn-managed-executor-binding-smoke.py @@ -126,6 +126,10 @@ def _write_fixture(root: Path) -> tuple[Path, Path, Path, Path]: + "\n", encoding="utf-8", ) + from loopx.control_plane.testing.canary_harness import import_fixture_authority + + for record in json.loads(registry.read_text())["goals"]: + import_fixture_authority(registry, record["id"], provider="sqlite") return project, runtime, workspace, registry diff --git a/examples/loopx-turn-managed-step-self-heal-smoke.py b/examples/loopx-turn-managed-step-self-heal-smoke.py index 4b9c0bda2f..b062379874 100644 --- a/examples/loopx-turn-managed-step-self-heal-smoke.py +++ b/examples/loopx-turn-managed-step-self-heal-smoke.py @@ -123,6 +123,10 @@ def _write_fixture(root: Path) -> tuple[Path, Path, Path, Path]: + "\n", encoding="utf-8", ) + from loopx.control_plane.testing.canary_harness import import_fixture_authority + + for record in json.loads(registry.read_text())["goals"]: + import_fixture_authority(registry, record["id"], provider="sqlite") return project, runtime, workspace, registry diff --git a/examples/loopx-turn-traex-e2e-smoke.py b/examples/loopx-turn-traex-e2e-smoke.py index 8ecb5d7efb..05ec59246f 100644 --- a/examples/loopx-turn-traex-e2e-smoke.py +++ b/examples/loopx-turn-traex-e2e-smoke.py @@ -115,6 +115,10 @@ def _write_fixture(root: Path) -> tuple[Path, Path, Path, Path]: + "\n", encoding="utf-8", ) + from loopx.control_plane.testing.canary_harness import import_fixture_authority + + for record in json.loads(registry.read_text())["goals"]: + import_fixture_authority(registry, record["id"], provider="sqlite") return project, runtime, workspace, registry diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 8723780835..d63e3d3a34 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -27,7 +27,6 @@ ) from ..control_plane.effect_runtime import EffectRuntimeRejected from ..control_plane.coordination.local_authority import CanonicalAuthorityRequired -from ..control_plane.capability_hooks import InteractionProjectionHookRegistration from ..control_plane.goals.first_party_host_admission import ( capture_first_party_host_goal_ref, ) @@ -37,7 +36,6 @@ compact_quota_plan_cli_payload, compact_quota_should_run_cli_payload, ) -from ..control_plane.quota.effective_action import EffectiveAction from ..control_plane.quota.effect_program import SettlementIdentity from ..control_plane.quota.error_codes import ( CloseoutQueryUnavailableError, @@ -108,6 +106,7 @@ ) from .quota_host_poll import attach_host_poll_receipt from .quota_monitor_poll import ( + record_automatic_heartbeat_stall as _record_automatic_heartbeat_stall, record_quota_monitor_poll_for_cli, recover_quota_monitor_poll_for_cli, ) @@ -200,106 +199,6 @@ def _quota_renderer( }.get(command, render_quota_markdown) -def _record_automatic_heartbeat_stall( - payload: dict[str, object], - status_payload: dict[str, object], - args: argparse.Namespace, - *, - registry_path: Path, - runtime_root_arg: str | None, - context: QuotaCommandContext, - interaction_projection_hooks: tuple[InteractionProjectionHookRegistration, ...], - turn_start_hook_dispatch: Mapping[str, object] | None, - action_selection: RequestedQuotaActionSelection, - cache_metadata: object, - goal_ref: Mapping[str, object] | None, -) -> tuple[dict[str, object], dict[str, object], object, str]: - """Commit and reproject the automatic no-spend heartbeat observation.""" - - turn_id = context.heartbeat_turn_id - if turn_id is None: - return payload, status_payload, cache_metadata, "not_applicable" - existing_stall = find_quota_monitor_poll_turn( - context.runtime_root, - goal_id=args.goal_id, - agent_id=args.agent_id, - turn_instance_id=turn_id, - goal_ref=goal_ref, - ) - if ( - payload.get("effective_action") - != EffectiveAction.MONITOR_QUIET_SKIP.value - and existing_stall is None - ): - return payload, status_payload, cache_metadata, "not_applicable" - poll = record_quota_monitor_poll( - status_payload, - goal_id=args.goal_id, - registry_path=registry_path, - execute=True, - source="heartbeat", - agent_id=args.agent_id, - available_capabilities=args.available_capabilities, - turn_instance_id=turn_id, - goal_ref=goal_ref, - scheduler_execution_context=context.scheduler_context, - operator_inbox_urgency_projector=context.operator_inbox_urgency_projector, - bounded_research_frontier_projector=project_live_explore_composition_frontier, - ) - if not poll.get("ok"): - raise RuntimeError( - "heartbeat stall observation writeback failed: " - f"{poll.get('reason') or 'missing follow-up quota decision'}" - ) - reloaded = collect_status( - registry_path=registry_path, - runtime_root_override=runtime_root_arg, - limit=context.status_limit, - goal_id=context.status_goal_id, - available_capabilities=args.available_capabilities, - ) - rebuilt = build_live_quota_should_run_decision( - reloaded, - goal_id=args.goal_id, - agent_id=args.agent_id, - available_capabilities=args.available_capabilities, - include_scheduler_detail="scheduler" in context.detail_sections, - include_agent_todo_detail=( - "agent-todos" in context.detail_sections - and not bool(getattr(args, "turn_envelope", False)) - ), - codex_app_current_rrule=args.app_automation_current_rrule, - registry_path=registry_path, - runtime_root=context.runtime_root, - host_observation_resolver=resolve_codex_app_automation_rrule, - scheduler_execution_context=context.scheduler_context, - operator_inbox_urgency_projector=context.operator_inbox_urgency_projector, - bounded_research_frontier_projector=project_live_explore_composition_frontier, - receipt_bound_todo_id=action_selection.receipt_bound_todo_id, - receipt_bound_replan_obligation_id=( - action_selection.receipt_bound_replan_obligation_id - ), - retained_action_selection_todo_id=( - action_selection.retained_todo_id_for_decision - ), - turn_instance_id=turn_id, - interaction_projection_hooks=interaction_projection_hooks, - turn_start_hook_dispatch=turn_start_hook_dispatch, - goal_ref=goal_ref, - ) - rebuilt["heartbeat_stall_writeback"] = { - "turn_instance_id": turn_id, - "status": "replayed" if poll.get("replayed") else "appended", - "generated_at": poll.get("generated_at"), - } - return ( - rebuilt, - reloaded, - None, - "replayed" if poll.get("replayed") else "appended", - ) - - def _dispatch_quota_turn_start_hooks( args: argparse.Namespace, *, @@ -810,6 +709,11 @@ def handle_quota_command( action_selection=action_selection, cache_metadata=cache_metadata, goal_ref=goal_ref, + find_monitor_poll_turn=find_quota_monitor_poll_turn, + monitor_poll_recorder=record_quota_monitor_poll, + status_collector=collect_status, + decision_builder=build_live_quota_should_run_decision, + host_observation_resolver=resolve_codex_app_automation_rrule, ) heartbeat_receipt_ready = True elif args.quota_command == "monitor-poll": diff --git a/loopx/cli_commands/quota_monitor_poll.py b/loopx/cli_commands/quota_monitor_poll.py index bbef4b2e77..d3efa7890f 100644 --- a/loopx/cli_commands/quota_monitor_poll.py +++ b/loopx/cli_commands/quota_monitor_poll.py @@ -5,6 +5,10 @@ from pathlib import Path from ..control_plane.coordination.local_authority import CanonicalAuthorityRequired +from ..control_plane.capability_hooks import InteractionProjectionHookRegistration +from ..control_plane.quota.effective_action import EffectiveAction +from .quota_context import QuotaCommandContext +from .quota_action_selection import RequestedQuotaActionSelection from ..capabilities.explore.composition_frontier import ( project_live_explore_composition_frontier, ) @@ -158,3 +162,108 @@ def recover_quota_monitor_poll_for_cli( return quota_failure_payload( args, registry_path=registry_path, runtime_root_arg=runtime_root_arg, error=error, ) + + +def record_automatic_heartbeat_stall( + payload: dict[str, object], + status_payload: dict[str, object], + args: argparse.Namespace, + *, + registry_path: Path, + runtime_root_arg: str | None, + context: QuotaCommandContext, + interaction_projection_hooks: tuple[InteractionProjectionHookRegistration, ...], + turn_start_hook_dispatch: Mapping[str, object] | None, + action_selection: RequestedQuotaActionSelection, + cache_metadata: object, + goal_ref: Mapping[str, object] | None, + find_monitor_poll_turn: Callable[..., Mapping[str, object] | None], + monitor_poll_recorder: Callable[..., dict[str, object]], + status_collector: Callable[..., dict[str, object]], + decision_builder: Callable[..., dict[str, object]], + host_observation_resolver: Callable[..., object], +) -> tuple[dict[str, object], dict[str, object], object, str]: + """Commit and reproject the automatic no-spend heartbeat observation.""" + + turn_id = context.heartbeat_turn_id + if turn_id is None: + return payload, status_payload, cache_metadata, "not_applicable" + existing_stall = find_monitor_poll_turn( + context.runtime_root, + goal_id=args.goal_id, + agent_id=args.agent_id, + turn_instance_id=turn_id, + goal_ref=goal_ref, + ) + if ( + payload.get("effective_action") + != EffectiveAction.MONITOR_QUIET_SKIP.value + and existing_stall is None + ): + return payload, status_payload, cache_metadata, "not_applicable" + poll = monitor_poll_recorder( + status_payload, + goal_id=args.goal_id, + registry_path=registry_path, + execute=True, + source="heartbeat", + agent_id=args.agent_id, + available_capabilities=args.available_capabilities, + turn_instance_id=turn_id, + goal_ref=goal_ref, + scheduler_execution_context=context.scheduler_context, + operator_inbox_urgency_projector=context.operator_inbox_urgency_projector, + bounded_research_frontier_projector=project_live_explore_composition_frontier, + ) + if not poll.get("ok"): + raise RuntimeError( + "heartbeat stall observation writeback failed: " + f"{poll.get('reason') or 'missing follow-up quota decision'}" + ) + reloaded = status_collector( + registry_path=registry_path, + runtime_root_override=runtime_root_arg, + limit=context.status_limit, + goal_id=context.status_goal_id, + available_capabilities=args.available_capabilities, + ) + rebuilt = decision_builder( + reloaded, + goal_id=args.goal_id, + agent_id=args.agent_id, + available_capabilities=args.available_capabilities, + include_scheduler_detail="scheduler" in context.detail_sections, + include_agent_todo_detail=( + "agent-todos" in context.detail_sections + and not bool(getattr(args, "turn_envelope", False)) + ), + codex_app_current_rrule=args.app_automation_current_rrule, + registry_path=registry_path, + runtime_root=context.runtime_root, + host_observation_resolver=host_observation_resolver, + scheduler_execution_context=context.scheduler_context, + operator_inbox_urgency_projector=context.operator_inbox_urgency_projector, + bounded_research_frontier_projector=project_live_explore_composition_frontier, + receipt_bound_todo_id=action_selection.receipt_bound_todo_id, + receipt_bound_replan_obligation_id=( + action_selection.receipt_bound_replan_obligation_id + ), + retained_action_selection_todo_id=( + action_selection.retained_todo_id_for_decision + ), + turn_instance_id=turn_id, + interaction_projection_hooks=interaction_projection_hooks, + turn_start_hook_dispatch=turn_start_hook_dispatch, + goal_ref=goal_ref, + ) + rebuilt["heartbeat_stall_writeback"] = { + "turn_instance_id": turn_id, + "status": "replayed" if poll.get("replayed") else "appended", + "generated_at": poll.get("generated_at"), + } + return ( + rebuilt, + reloaded, + None, + "replayed" if poll.get("replayed") else "appended", + ) diff --git a/loopx/cli_commands/turn_host_request.py b/loopx/cli_commands/turn_host_request.py new file mode 100644 index 0000000000..7cc0f4a41d --- /dev/null +++ b/loopx/cli_commands/turn_host_request.py @@ -0,0 +1,76 @@ +"""Parse the existing CLI Host request before any Turn execution. + +This is CLI argument transport and host-adapter binding, not a decision owner. +The Turn driver retains admission, effects and settlement. +""" +from __future__ import annotations + +import argparse +import json +from collections.abc import Callable, Mapping +from pathlib import Path +from typing import Any, cast + + +def prepare_turn_host_request( + args: argparse.Namespace, payload: Mapping[str, Any], *, + native_host_runner: Callable[..., dict[str, Any]] | None, + native_session_binding_resolver: Callable[[Mapping[str, Any]], dict[str, Any]] | None, + native_task_validator: Callable[[Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any]] | None, + validator_builder: Callable[..., Callable[..., Mapping[str, Any]]], +) -> tuple[Path, list[str] | None, Callable[..., Mapping[str, Any]] | None, dict[str, Any]]: + project = Path(args.project).expanduser().resolve() + planned_host = ( + cast(dict[str, Any], payload.get("host")) if isinstance(payload.get("host"), dict) else {} + ) + if planned_host.get("kind") != args.host: + raise ValueError("--host must match the journaled LoopX Turn plan") + if args.host == "codex-app-server" and ( + native_host_runner is None or native_session_binding_resolver is None + ): + raise ValueError("app-server execution requires its original in-process Chat connection") + if args.host != "codex-app-server" and ( + native_host_runner is not None or native_session_binding_resolver is not None + ): + raise ValueError("a native Chat connection cannot replace a selected CLI host") + if args.host == "generic-cli": + if not args.host_command_json: + raise ValueError( + "generic-cli requires --host-adapter-command-json " + "(alias --host-command-json)" + ) + raw_argv = json.loads(args.host_command_json) + if not isinstance(raw_argv, list) or not all( + isinstance(item, str) for item in raw_argv + ): + raise ValueError( + "--host-adapter-command-json must be a JSON string array" + ) + else: + if args.host_command_json: + raise ValueError(f"{args.host} does not accept --host-command-json") + raw_argv = None + if native_task_validator is not None and args.host != "codex-app-server": + raise ValueError("a native task validator cannot replace a selected CLI validator") + task_validator = native_task_validator + if args.validation_command_json: + raw_validation_argv = json.loads(args.validation_command_json) + if not isinstance(raw_validation_argv, list) or not all( + isinstance(item, str) for item in raw_validation_argv + ): + raise ValueError( + "--validation-command-json must be a JSON string array" + ) + if task_validator is None: + task_validator = validator_builder( + raw_validation_argv, + project=project, + timeout_seconds=args.validation_timeout_seconds, + failure_recovery_kind=args.validation_failure_kind, + ) + envelope = ( + cast(dict[str, Any], payload.get("turn_envelope")) + if isinstance(payload.get("turn_envelope"), dict) + else {} + ) + return project, raw_argv, task_validator, envelope diff --git a/loopx/cli_commands/turn_run_once.py b/loopx/cli_commands/turn_run_once.py index 995c837cdd..7f0dd781fd 100644 --- a/loopx/cli_commands/turn_run_once.py +++ b/loopx/cli_commands/turn_run_once.py @@ -1,7 +1,6 @@ from __future__ import annotations import argparse -import json from collections.abc import Callable, Mapping from pathlib import Path from typing import Any @@ -18,9 +17,7 @@ from ..control_plane.quota.heartbeat_receipt import ( ensure_turn_heartbeat_settlement_receipt, ) -from ..control_plane.quota.live_decision import ( - build_live_quota_should_run_decision, -) +from ..control_plane.quota.live_decision import build_live_quota_should_run_decision from ..control_plane.quota.settlement import ( SettlementIdentity, SettlementStepKind, @@ -50,6 +47,7 @@ from ..state_refresh import refresh_state_run from ..control_plane.todos.path_resolution import resolve_todo_state_path from .turn_cadence import managed_cadence_start +from .turn_host_request import prepare_turn_host_request from .turn_decision import collect_turn_status_payload, current_turn_host_launch_decision from .turn_dsh_host import build_dsh_host_runner, build_dsh_operation_host_runner from .turn_rendering import build_turn_error_payload @@ -82,59 +80,11 @@ def execute_turn_run_once( """Adapt one prepared CLI Turn plan to the bounded Turn driver.""" execution_started = False try: - project = Path(args.project).expanduser().resolve() - planned_host = ( - payload.get("host") if isinstance(payload.get("host"), dict) else {} - ) - if planned_host.get("kind") != args.host: - raise ValueError("--host must match the journaled LoopX Turn plan") - if args.host == "codex-app-server" and ( - native_host_runner is None or native_session_binding_resolver is None - ): - raise ValueError("app-server execution requires its original in-process Chat connection") - if args.host != "codex-app-server" and ( - native_host_runner is not None or native_session_binding_resolver is not None - ): - raise ValueError("a native Chat connection cannot replace a selected CLI host") - if args.host == "generic-cli": - if not args.host_command_json: - raise ValueError( - "generic-cli requires --host-adapter-command-json " - "(alias --host-command-json)" - ) - raw_argv = json.loads(args.host_command_json) - if not isinstance(raw_argv, list) or not all( - isinstance(item, str) for item in raw_argv - ): - raise ValueError( - "--host-adapter-command-json must be a JSON string array" - ) - else: - if args.host_command_json: - raise ValueError(f"{args.host} does not accept --host-command-json") - raw_argv = None - if native_task_validator is not None and args.host != "codex-app-server": - raise ValueError("a native task validator cannot replace a selected CLI validator") - task_validator = native_task_validator - if args.validation_command_json: - raw_validation_argv = json.loads(args.validation_command_json) - if not isinstance(raw_validation_argv, list) or not all( - isinstance(item, str) for item in raw_validation_argv - ): - raise ValueError( - "--validation-command-json must be a JSON string array" - ) - if task_validator is None: - task_validator = build_loopx_turn_command_validator( - raw_validation_argv, - project=project, - timeout_seconds=args.validation_timeout_seconds, - failure_recovery_kind=args.validation_failure_kind, - ) - envelope = ( - payload.get("turn_envelope") - if isinstance(payload.get("turn_envelope"), dict) - else {} + project, raw_argv, task_validator, envelope = prepare_turn_host_request( + args, payload, native_host_runner=native_host_runner, + native_session_binding_resolver=native_session_binding_resolver, + native_task_validator=native_task_validator, + validator_builder=build_loopx_turn_command_validator, ) selected_todo = selected_turn_todo(envelope) expected_todo_text_sha256 = captured_todo_text_sha256(