refactor: separate complete quota and Host request phases

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
 
 

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
 
 

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
 
 

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
 
 

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":

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",
+    )

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