Fix automatic task lease preparation before execution

docs/status-data-contract.md

@@ -1376,8 +1376,11 @@ capability-ready Todo. That request is a pending selection, not a committed
 receipt identity. Quota first re-runs current lane arbitration and eligibility;
 a newly due hard-priority monitor, blocking user gate, or other preemption
 defers the request and keeps the receipt identity-less. A qualified request does
-not need to have appeared in the bounded suggestions. Only the upgraded response
-restores delivery and its settlement plan. If there is only one admitted action,
+not need to have appeared in the bounded suggestions. A qualified selection with
+a missing hard lease binds the same Todo/Turn for preparation only: delivery stays
+denied and the only action is the ordinary claim. The same-Turn guard must observe
+`execution_preparation.state=ready` before restoring delivery and its settlement
+plan. If there is only one admitted action,
 no portfolio selection phase is added.
 
 When selected work has meaningful typed lineage, waiting, sibling, or
@@ -1404,6 +1407,37 @@ completeness counters remain authoritative for omissions. Consumers that need
 the complete open source follow `detail_refs.full_todo_list` rather than
 assuming either bounded projection is exhaustive.
 
+After a new Turn has selected one Todo, `quota should-run` also checks that
+Todo's canonical hard lease. `execution_preparation_v0` is `ready` only for a
+matching active lease. A missing, released, or expired lease produces
+`claim_required` with one ordinary managed claim and
+`host_execution_allowed=true`; a capable Host may execute it before launching
+the model, then must rerun the guard and observe the same Todo as `ready`.
+The read-only guard never acquires authority itself, and the claim still
+atomically rechecks current ownership, write scopes, physical workspace and
+lease generation. Foreign ownership, conflicting leases, or unreadable
+canonical authority fail closed. Explicit action selection must bind one Todo
+before preparation. Receipt-bound recovery rechecks that same Todo and Turn, so
+the post-claim guard can prove `ready` without replacing their identities. Hosts
+without this integration expose only the claim command for
+the Agent to execute, avoiding unrelated context reads or delivery work first.
+
+For enabled native Codex App/SSH context delivery, the existing trusted
+`SessionStart` / `PostToolUse` lifecycle hook is a capable Host executor. The
+private session journal retains only the exact projected claim identity beside
+the immutable guard capture, including the absolute authority route that
+produced it. The hook uses that retained route even when its process runs from
+another project; it never rediscovers Goal authority from the hook cwd. The
+hook accepts only
+`host_execution_allowed=true`, executes the ordinary managed claim, reruns the
+guard for the same Todo and receipt-bound Turn, and injects a compact `ready`
+receipt before the next model step. A successful receipt suppresses repeated
+preparation for that capture; session resume rechecks it because the lease may
+have expired while the model was absent. Malformed, foreign, moved-workspace,
+timed-out, or failed preparation preserves the original explicit claim/recheck
+fallback and never erases context-recovery output. Context delivery remains an
+opt-in transport; it does not make a read-only guard itself acquire authority.
+
 Correctly typed future work is handled earlier. A higher-priority
 `continuous_monitor` with a valid future `next_due_at` is not executable; quota
 selects the next ready advancement todo and records the future monitor under

loopx/cli_commands/quota_capture.py

@@ -8,6 +8,7 @@
 import shlex
 import socket
 import stat
+import subprocess
 import tempfile
 from collections.abc import Callable
 from pathlib import Path
@@ -1110,23 +1111,65 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
         from ..control_plane.effect_runtime import MAX_REQUEST_BYTES
         from ..control_plane.turn_driver.codex_context import (
             CONTEXT_LIFECYCLE_HOOKS, codex_context_directory, codex_context_hook,
+            native_context_event_eligible, native_execution_preparation_request,
+            record_native_execution_preparation,
+        )
+        from ..control_plane.turn_driver.codex_execution_preparation import (
+            codex_execution_preparation_context,
+            execute_codex_execution_preparation,
         )
 
-        # Native transport only: no registry discovery, guard or quota effects.
+        # Context transport remains read-only.  An exact typed Host preparation
+        # request may additionally execute its ordinary claim and fresh guard.
         try:
             if not activation["enabled"]:
                 print("{}")
                 return 0
             raw = sys.stdin.buffer.read(MAX_REQUEST_BYTES + 1)
             event = json.loads(raw) if len(raw) <= MAX_REQUEST_BYTES else None
-            if (isinstance(event, dict) and isinstance(event.get("session_id"), str) and event["session_id"]
+            if (isinstance(event, dict) and native_context_event_eligible(event)
+                    and isinstance(event.get("session_id"), str) and event["session_id"]
                     and event.get("hook_event_name") in CONTEXT_LIFECYCLE_HOOKS):
                 directory = codex_context_directory(directory=args.codex_context_dir,
                     root=getattr(args, "codex_context_root", None), session=event.get("session_id"), create=True)
                 payload = codex_context_hook(directory, event)
+                try:
+                    request = native_execution_preparation_request(directory, event)
+                    if request is not None:
+                        workspace = Path(event.get("cwd") or "")
+                        if workspace.is_absolute() and workspace.resolve() == Path.cwd().resolve():
+                            receipt = execute_codex_execution_preparation(
+                                request,
+                                cwd=workspace.resolve(),
+                                scenario=activation["scenario"],
+                            )
+                            record_native_execution_preparation(
+                                directory,
+                                event,
+                                source_sha256=request["source_sha256"],
+                                receipt=receipt,
+                            )
+                            existing = payload.get("hookSpecificOutput")
+                            existing_context = (
+                                str(existing.get("additionalContext") or "")
+                                if isinstance(existing, dict)
+                                else ""
+                            )
+                            payload["hookSpecificOutput"] = {
+                                "hookEventName": event["hook_event_name"],
+                                "additionalContext": " ".join(filter(None, [
+                                    existing_context,
+                                    codex_execution_preparation_context(receipt),
+                                ])),
+                            }
+                except (OSError, RuntimeError, subprocess.SubprocessError, ValueError, TypeError, RecursionError):
+                    # Preserve the original explicit action and any context
+                    # recovery output when Host-side preparation cannot prove
+                    # its postcondition.
+                    pass
             else:
                 payload = {}
-        except (OSError, ValueError, TypeError, RecursionError):
+        except (OSError, RuntimeError, subprocess.SubprocessError, ValueError, TypeError, RecursionError):
             payload = {}
         print(json.dumps(payload, ensure_ascii=False, separators=(",", ":")))
         return 0

loopx/control_plane/effect_runtime_handlers.ts

@@ -779,6 +779,7 @@ export function createEffectRuntimeHandlers(
     ["turn.settlement.reduce", lazyHandler(() => import("./turn_driver/settlement.ts"), ({reduceTurnSettlementTransaction}) => reduceTurnSettlementTransaction)],
     ["turn.host_todo_completion.evaluate", lazyHandler(() => import("./turn_driver/host_todo_completion.ts"), ({evaluateHostTodoCompletion}) => evaluateHostTodoCompletion)],
     ["work_item.interaction_execution.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectInteractionExecution}) => projectInteractionExecution)],
+    ["work_item.execution_preparation.project", lazyHandler(() => import("./work_items/execution_preparation.ts"), ({projectExecutionPreparation}) => projectExecutionPreparation)],
     ["work_item.interaction_reads.project", lazyHandler(() => import("./quota/turn_envelope.ts"), ({projectInteractionReadsAndGuidance}) => projectInteractionReadsAndGuidance)],
     ["work_item.blocked_wait_observation.project", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({projectBlockedSuccessorWaitObservation}) => projectBlockedSuccessorWaitObservation)],
     ["work_item.context.plan", lazyHandler(() => import("./work_items/interaction_contract.ts"), ({planInteractionWorkContext}) => planInteractionWorkContext)],

loopx/control_plane/quota/execution_preparation.py

@@ -0,0 +1,189 @@
+from __future__ import annotations
+
+import shlex
+from collections.abc import Mapping
+from pathlib import Path
+from typing import Any
+
+from ..coordination.local_authority import read_canonical_todos_if_promoted
+from ..effect_runtime import effect_runtime_result
+from ..todos.contract import normalize_todo_claimed_by, normalize_todo_id
+from ..work_items.action_selection_contract import (
+    action_portfolio_requires_explicit_selection,
+)
+from ..work_items.task_lease import lease_is_active
+
+
+def _unavailable_facts(*, goal_id: str, todo_id: str, agent_id: str) -> dict[str, Any]:
+    return {
+        "schema_version": "execution_preparation_facts_v0",
+        "goal_id": goal_id,
+        "agent_id": agent_id,
+        "handoff_mode": "hard_lease",
+        "authority_available": False,
+        "selected_todo": {"todo_id": todo_id, "status": "open"},
+        "leases": [],
+    }
+
+
+def project_selected_todo_execution_preparation(
+    *,
+    runtime_root: str | Path | None,
+    goal_id: str,
+    selected_todo: Mapping[str, Any],
+    agent_id: str | None,
+) -> dict[str, Any] | None:
+    """Read the selected Todo's canonical lease and project one preparation step.
+
+    No lease is acquired here.  ``claim.cli_args`` is an ordinary managed
+    claim that a Host may execute before launching the model; the claim remains
+    the mutation authority and rechecks current workspace and conflicts.
+    """
+
+    todo_id = normalize_todo_id(selected_todo.get("todo_id"))
+    safe_agent_id = normalize_todo_claimed_by(agent_id)
+    if not todo_id or not safe_agent_id or runtime_root is None:
+        return None
+    try:
+        snapshot = read_canonical_todos_if_promoted(
+            runtime_root=Path(runtime_root), goal_id=goal_id, include_leases=True,
+        )
+        authoritative = [
+            row for row in snapshot["todos"]
+            if normalize_todo_id(row.get("todo_id")) == todo_id
+        ]
+        if len(authoritative) != 1:
+            facts = _unavailable_facts(
+                goal_id=goal_id, todo_id=todo_id, agent_id=safe_agent_id,
+            )
+        else:
+            todo = authoritative[0]
+            leases = []
+            for lease in snapshot["leases"]:
+                if normalize_todo_id(lease.get("todo_id")) != todo_id:
+                    continue
+                status = str(lease.get("status") or "unknown").strip().lower()
+                leases.append({
+                    "status": status if status in {"active", "released"} else "unknown",
+                    "active": lease_is_active(dict(lease)),
+                    **({"owner": lease["owner"]} if normalize_todo_claimed_by(lease.get("owner")) else {}),
+                })
+            facts = {
+                "schema_version": "execution_preparation_facts_v0",
+                "goal_id": goal_id,
+                "agent_id": safe_agent_id,
+                "handoff_mode": str(snapshot.get("handoff_mode") or "hard_lease"),
+                "authority_available": True,
+                "selected_todo": {
+                    "todo_id": todo_id,
+                    "status": todo.get("status"),
+                    "claimed_by": normalize_todo_claimed_by(todo.get("claimed_by")),
+                },
+                "leases": leases,
+            }
+    except (OSError, RuntimeError, TypeError, ValueError):
+        facts = _unavailable_facts(
+            goal_id=goal_id, todo_id=todo_id, agent_id=safe_agent_id,
+        )
+
+    result = effect_runtime_result("work_item.execution_preparation.project", facts)
+    if not isinstance(result, dict) or result.get("schema_version") != "execution_preparation_v0":
+        raise TypeError("typed execution preparation returned an invalid projection")
+    claim = result.get("claim")
+    if isinstance(claim, dict) and isinstance(claim.get("cli_args"), list):
+        claim["command_line"] = shlex.join(["loopx", *map(str, claim["cli_args"])])
+    return result
+
+
+def apply_selected_todo_execution_preparation(
+    payload: dict[str, Any],
+    *,
+    runtime_root: str | Path | None,
+    goal_id: str,
+    agent_id: str | None,
+    obligation: dict[str, Any] | None = None,
+    recommendation: dict[str, Any] | None = None,
+    receipt_bound: bool = False,
+) -> None:
+    selected = payload.get("selected_todo")
+    if (
+        not isinstance(selected, dict)
+        or payload.get("should_run") is not True
+        or action_portfolio_requires_explicit_selection(payload)
+    ):
+        return
+    preparation = project_selected_todo_execution_preparation(
+        runtime_root=runtime_root,
+        goal_id=goal_id,
+        selected_todo=selected,
+        agent_id=agent_id,
+    )
+    # Existing receipt-bound recovery predates this projection and may be
+    # reconstructed without canonical lease authority in synthetic/legacy
+    # environments. Preserve that recovery path, while a Host that just ran a
+    # projected claim still requires the explicit `ready` postcondition before
+    # launching work. Available authority continues to surface conflicts and
+    # stale leases for the same bound Todo.
+    if receipt_bound and (
+        preparation is None
+        or preparation.get("state") == "authority_unavailable"
+    ):
+        return
+    if preparation is None or preparation.get("state") == "not_required":
+        return
+    payload["execution_preparation"] = preparation
+    state = preparation.get("state")
+    if state == "ready":
+        return
+    payload["normal_delivery_allowed"] = False
+    if obligation is not None:
+        obligation["delivery_allowed"] = False
+    reason = str(preparation.get("reason") or "Selected Todo execution is not prepared.")
+    payload["reason"] = reason
+    if recommendation is not None:
+        recommendation["reason"] = reason
+    if state == "claim_required":
+        selected["claim_required_before_work"] = True
+        payload["actionable_by_codex"] = True
+        if obligation is not None:
+            obligation["must_attempt_work"] = True
+            obligation["reason"] = reason
+        command = (preparation.get("claim") or {}).get("command_line")
+        if isinstance(command, str):
+            payload["recommended_action"] = command
+    else:
+        payload["should_run"] = False
+        payload["actionable_by_codex"] = False
+        if obligation is not None:
+            obligation["must_attempt_work"] = False
+
+
+def attach_interaction_execution_preparation(
+    contract: dict[str, Any], payload: dict[str, Any],
+) -> None:
+    preparation = payload.get("execution_preparation")
+    if not isinstance(preparation, dict):
+        return
+    contract["agent_channel"]["execution_preparation"] = preparation
+    contract["cli_channel"]["execution_preparation"] = preparation
+    state = preparation.get("state")
+    if state == "ready":
+        return
+    reason = str(preparation.get("reason") or "Selected Todo execution is not prepared.")
+    contract["agent_channel"]["delivery_allowed"] = False
+    contract["agent_channel"]["primary_action"] = reason
+    if state != "claim_required":
+        return
+    claim = preparation.get("claim")
+    command = claim.get("command_line") if isinstance(claim, dict) else None
+    if isinstance(command, str) and command:
+        contract["agent_channel"]["must_attempt"] = True
+        contract["agent_channel"]["quiet_noop_allowed"] = False
+        contract["agent_channel"]["primary_action"] = (
+            f"Run the atomic execution preparation action: {command}"
+        )
+        contract["cli_channel"]["next_cli_actions"] = [command]
+        for key in (
+            "settlement_plan", "quota_spend_source", "replan_settlement_contract",
+        ):
+            contract["cli_channel"].pop(key, None)

loopx/control_plane/quota/progressive_context.ts

@@ -301,7 +301,7 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
     include(...source, "content", "harness", "guidance");
     include(...source, "content", "harness", "orchestration_gate");
   });
-  for (const field of ["next_cli_actions", "spend_allowed_now", "spend_after_validation", "spend_policy",
+  for (const field of ["execution_preparation", "next_cli_actions", "spend_allowed_now", "spend_after_validation", "spend_policy",
     "quota_spend_source", "delivery_workspace_causality", "vision_continuation_audit"])
     include("interaction_contract", "cli_channel", field);
   const planPath = ["interaction_contract", "cli_channel", "settlement_plan"];

loopx/control_plane/quota/should_run_packet.py

@@ -65,6 +65,9 @@
 from ..quota.selected_todo_projection import (
     selected_todo_projection as _selected_todo_projection,
 )
+from ..quota.execution_preparation import (
+    apply_selected_todo_execution_preparation,
+)
 from ..quota.stall_repair import (
     stall_repair_blocked_action_scope,
     stall_repair_payload,
@@ -1697,6 +1700,23 @@ def _build_quota_should_run_payload(
         payload["delivery_workspace_isolation"] = observe_goal_delivery_workspace_isolation(
             prepared.item, selected_todo,
         )
+    apply_selected_todo_execution_preparation(
+        payload,
+        runtime_root=_interaction_runtime_root(runtime_root, prepared.status_payload),
+        goal_id=prepared.safe_goal_id,
+        agent_id=quota_decision_agent_id(payload) or prepared.requested_agent_id,
+        obligation=(
+            payload["execution_obligation"]
+            if isinstance(payload.get("execution_obligation"), dict)
+            else None
+        ),
+        recommendation=(
+            payload["heartbeat_recommendation"]
+            if isinstance(payload.get("heartbeat_recommendation"), dict)
+            else None
+        ),
+        receipt_bound=prepared.receipt_bound_todo_id is not None,
+    )
     _attach_automation_cadence_readback(
         payload,
         goal_id=prepared.safe_goal_id,