From bfabddd2c8836c72a3f58685c7cabf321a414a6f Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 3 Nov 2026 19:29:54 +0800 Subject: [PATCH 01/10] fix(control-plane): prepare task leases before execution Signed-off-by: huangruiteng --- docs/status-data-contract.md | 14 ++ .../control_plane/effect_runtime_handlers.ts | 1 + .../quota/execution_preparation.py | 148 +++++++++++++ .../control_plane/quota/should_run_packet.py | 20 ++ .../work_items/execution_preparation.ts | 123 +++++++++++ .../work_items/interaction_contract.py | 35 +++ loopx/kunluncode_goal_mode/runtime.py | 30 +++ .../test_quota_execution_preparation.py | 199 ++++++++++++++++++ .../execution_preparation.test.ts | 87 ++++++++ tests/test_kunluncode_goal_mode.py | 48 +++++ 10 files changed, 705 insertions(+) create mode 100644 loopx/control_plane/quota/execution_preparation.py create mode 100644 loopx/control_plane/work_items/execution_preparation.ts create mode 100644 tests/control_plane/test_quota_execution_preparation.py create mode 100644 tests/control_plane_ts/execution_preparation.test.ts diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index ec6478918f..85d5632f1f 100644 --- a/docs/status-data-contract.md +++ b/docs/status-data-contract.md @@ -1404,6 +1404,20 @@ 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 and receipt-bound +recovery keep their existing identities and are never replaced by automatic +preparation. Hosts without this integration expose only the claim command for +the Agent to execute, avoiding unrelated context reads or delivery work first. + 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 diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index b98ba473d4..54c36e6726 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -777,6 +777,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)], diff --git a/loopx/control_plane/quota/execution_preparation.py b/loopx/control_plane/quota/execution_preparation.py new file mode 100644 index 0000000000..7827c8c513 --- /dev/null +++ b/loopx/control_plane/quota/execution_preparation.py @@ -0,0 +1,148 @@ +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 receipt_bound + 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, + ) + 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 diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 7598873e65..3d2ac0f931 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -64,6 +64,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, @@ -1696,6 +1699,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, diff --git a/loopx/control_plane/work_items/execution_preparation.ts b/loopx/control_plane/work_items/execution_preparation.ts new file mode 100644 index 0000000000..25fd0996fc --- /dev/null +++ b/loopx/control_plane/work_items/execution_preparation.ts @@ -0,0 +1,123 @@ +import type {JsonObject} from "../effect_program.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; +import { + optionalNonEmptyString, + requireBoolean, + requireJsonObject, + requireNonEmptyString, + requireStringLiteral, +} from "../runtime_decode.ts"; + +export const EXECUTION_PREPARATION_FACTS_SCHEMA_VERSION = + "execution_preparation_facts_v0"; +export const EXECUTION_PREPARATION_SCHEMA_VERSION = + "execution_preparation_v0"; + +const HANDOFF_MODES = ["soft_claim", "hard_lease"] as const; +const LEASE_STATUSES = ["active", "released", "unknown"] as const; + +function leaseFact(value: unknown, label: string): JsonObject { + const lease = requireJsonObject(value, label); + const status = requireStringLiteral(lease.status, LEASE_STATUSES, `${label}.status`); + const active = requireBoolean(lease.active, `${label}.active`); + const owner = optionalNonEmptyString(lease.owner, `${label}.owner`); + if (active && status !== "active") { + throw new EffectRuntimeRequestError(`${label}.active requires status=active`); + } + return {status, active, ...(owner === null ? {} : {owner})}; +} + +function claimAction(goalId: string, todoId: string, agentId: string): JsonObject { + return { + command: "loopx todo claim", + cli_args: [ + "todo", "claim", "--goal-id", goalId, "--todo-id", todoId, + "--agent-id", agentId, "--claimed-by", agentId, + ], + host_execution_allowed: true, + postcondition: "rerun quota should-run and require execution_preparation.state=ready", + safety: "ordinary managed claim atomically rechecks current ownership, write scopes, workspace isolation, and lease generation", + }; +} + +/** Decide whether the selected Todo can execute under its current hard lease. + * + * This is a read-model decision. It never acquires a lease. A capable Host + * may execute the returned ordinary claim before starting the model; every + * authority and workspace check remains owned by that atomic claim. + */ +export function projectExecutionPreparation(value: unknown): JsonObject { + const request = requireJsonObject(value, "execution preparation facts"); + if (request.schema_version !== EXECUTION_PREPARATION_FACTS_SCHEMA_VERSION) { + throw new EffectRuntimeRequestError( + `execution_preparation_facts.schema_version must be ${EXECUTION_PREPARATION_FACTS_SCHEMA_VERSION}`, + ); + } + const goalId = requireNonEmptyString(request.goal_id, "goal_id"); + const agentId = requireNonEmptyString(request.agent_id, "agent_id"); + const handoffMode = requireStringLiteral(request.handoff_mode, HANDOFF_MODES, "handoff_mode"); + const authorityAvailable = requireBoolean(request.authority_available, "authority_available"); + const selected = requireJsonObject(request.selected_todo, "selected_todo"); + const todoId = requireNonEmptyString(selected.todo_id, "selected_todo.todo_id"); + const claimedBy = optionalNonEmptyString(selected.claimed_by, "selected_todo.claimed_by"); + const status = optionalNonEmptyString(selected.status, "selected_todo.status"); + const rawLeases = request.leases; + if (!Array.isArray(rawLeases)) { + throw new EffectRuntimeRequestError("leases must be an array"); + } + const leases = rawLeases.map((lease, index) => leaseFact(lease, `leases[${index}]`)); + const base = { + schema_version: EXECUTION_PREPARATION_SCHEMA_VERSION, + todo_id: todoId, + agent_id: agentId, + handoff_mode: handoffMode, + }; + + if (!authorityAvailable) { + return {...base, state: "authority_unavailable", delivery_allowed: false, + reason_code: "canonical_authority_unavailable", + reason: "Canonical Todo and lease authority is unavailable; do not start delivery work."}; + } + if (status !== "open") { + return {...base, state: "blocked_by_conflict", delivery_allowed: false, + reason_code: "selected_todo_not_open", + reason: "The selected Todo is no longer open; select current work before execution."}; + } + if (claimedBy !== null && claimedBy !== agentId) { + return {...base, state: "blocked_by_conflict", delivery_allowed: false, + reason_code: "todo_claimed_by_other_agent", owner: claimedBy, + reason: "The selected Todo is claimed by another agent; do not take over its execution."}; + } + if (handoffMode === "soft_claim") { + return {...base, state: "not_required", delivery_allowed: true, + reason_code: "soft_claim_has_no_execution_lease"}; + } + + const active = leases.filter(lease => lease.active === true); + if (active.length > 1) { + return {...base, state: "blocked_by_conflict", delivery_allowed: false, + reason_code: "multiple_active_task_leases", + reason: "Multiple active leases were observed for one Todo; repair canonical authority before execution."}; + } + if (active.length === 1) { + const owner = optionalNonEmptyString(active[0].owner, "active lease owner"); + if (owner !== agentId) { + return {...base, state: "blocked_by_conflict", delivery_allowed: false, + reason_code: "active_lease_owned_by_other_agent", ...(owner ? {owner} : {}), + reason: "The selected Todo has an active lease owned by another agent; do not take over its execution."}; + } + return {...base, state: "ready", delivery_allowed: true, + reason_code: "matching_active_task_lease"}; + } + + const prior = leases[0]; + const reasonCode = prior?.status === "released" + ? "task_lease_released" + : prior?.status === "active" + ? "task_lease_expired" + : "task_lease_missing"; + return {...base, state: "claim_required", delivery_allowed: false, + reason_code: reasonCode, + reason: "Prepare the selected Todo with one ordinary managed claim before delivery work.", + claim: claimAction(goalId, todoId, agentId)}; +} diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 0efce2fed4..bf63a0bb60 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -1500,6 +1500,9 @@ def _attach_interaction_required_reads( def _attach_interaction_post_writeback_actions( contract: dict[str, Any], payload: dict[str, Any] ) -> None: + preparation = payload.get("execution_preparation") + if isinstance(preparation, Mapping) and preparation.get("state") != "ready": + return goal_boundary = ( payload.get("goal_boundary") if isinstance(payload.get("goal_boundary"), Mapping) @@ -1575,6 +1578,37 @@ def _attach_interaction_vision_wait_state( } +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) + + def _interaction_fallback_policy_required(payload: Mapping[str, object], *, mode: str) -> bool: return mode in { "user_gate", @@ -1713,6 +1747,7 @@ def _build_interaction_contract_projection( registry_path=registry_path, ), } + _attach_interaction_execution_preparation(contract, payload) response_plan = _build_interaction_response_plan( user_channel=user_channel, agent_channel=agent_channel, diff --git a/loopx/kunluncode_goal_mode/runtime.py b/loopx/kunluncode_goal_mode/runtime.py index 7254d1812b..ce9b38dfd2 100644 --- a/loopx/kunluncode_goal_mode/runtime.py +++ b/loopx/kunluncode_goal_mode/runtime.py @@ -649,6 +649,36 @@ def run_native_goal( "LoopX allowed a KunlunCode turn without selecting an agent todo" ) todo_id = str(todo.get("todo_id") or "") + preparation = ( + decision.get("execution_preparation") + if isinstance(decision.get("execution_preparation"), Mapping) + else {} + ) + if preparation.get("state") == "claim_required": + # The Host performs the projected atomic claim before any model + # work. A fresh guard must observe the resulting lease; the + # preview itself never grants execution authority. + goal_admission.require_current() + control.claim(todo_id) + decision = control.should_run() + if decision.get("should_run") is not True: + raise KunlunNativeGoalRuntimeError( + "LoopX did not admit execution after automatic task preparation" + ) + todo = decision.get("selected_todo") + preparation = ( + decision.get("execution_preparation") + if isinstance(decision.get("execution_preparation"), Mapping) + else {} + ) + if ( + not isinstance(todo, Mapping) + or str(todo.get("todo_id") or "") != todo_id + or preparation.get("state") != "ready" + ): + raise KunlunNativeGoalRuntimeError( + "LoopX automatic task preparation did not preserve the selected Todo and current lease" + ) claimed_by = str(todo.get("claimed_by") or "") agent_id = str(context.get("agent_id") or "") if claimed_by and claimed_by != agent_id: diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py new file mode 100644 index 0000000000..a1c2904fe2 --- /dev/null +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +from datetime import timedelta +import json + +import pytest + +from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection +from loopx.control_plane.runtime.time import now_utc, utc_isoformat +from loopx.control_plane.testing.canary_harness import write_fixture_registry, run_json_cli_result +from loopx.control_plane.quota.execution_preparation import ( + apply_selected_todo_execution_preparation, +) + + +GOAL = "goal-execution-preparation" +TODO = "todo_execution_preparation" +AGENT = "agent-a" + + +def _todo() -> dict[str, object]: + return { + "schema_version": "todo_item_v0", + "todo_id": TODO, + "role": "agent", + "source_section": "Agent Todo", + "archive_state": "active", + "text": "Deliver one validated synthetic change", + "status": "open", + "done": False, + "task_class": "advancement_task", + "priority": "P1", + "claimed_by": AGENT, + "index": 1, + } + + +def _lease(*, owner: str = AGENT, status: str = "active", expired: bool = False) -> dict[str, object]: + at = now_utc() + return { + "schema_version": "task_lease_v0", + "goal_id": GOAL, + "todo_id": TODO, + "owner": owner, + "idempotency_key": f"{owner}-execution", + "write_scopes": [], + "acquire_ttl_seconds": 600, + "version": 2, + "lease_epoch": 1, + "status": status, + "acquired_at": utc_isoformat(at - timedelta(seconds=120)), + "updated_at": utc_isoformat(at - timedelta(seconds=120)), + "expires_at": utc_isoformat(at - timedelta(seconds=60) if expired else at + timedelta(seconds=600)), + **({"released_at": utc_isoformat(at)} if status == "released" else {}), + } + + +def _run(tmp_path, monkeypatch, provider: str, leases: list[dict[str, object]]): + isolate_sqlite_runtime(tmp_path, monkeypatch) + runtime = tmp_path / "runtime" + registry = tmp_path / "registry.json" + state = tmp_path / "state.md" + state.write_text( + "---\nstatus: active\n---\n# Goal\n\n## Objective\n" + "Deliver one validated synthetic change.\n\n## Next Action\n" + "Run the selected Todo.\n", + encoding="utf-8", + ) + write_fixture_registry( + project=tmp_path, + runtime_root=runtime, + registry_path=registry, + goal_id=GOAL, + domain="execution-preparation", + adapter_kind="generic_project_goal_v0", + state_file=str(state), + registered_agents=[AGENT, "agent-b"], + quota_allowed_slots=None, + extra_goal_fields={"coordination": { + "registered_agents": [AGENT, "agent-b"], + "handoff_mode": "hard_lease", + }}, + ) + projection = build_todo_runtime_shadow_projection( + goal_id=GOAL, + todos=[_todo()], + leases=leases, + handoff_mode="hard_lease", + ) + initialize_canonical_authority( + runtime, GOAL, projection, state_path=state, provider=provider, + ) + code, packet = run_json_cli_result( + "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + registry_path=registry, runtime_root=runtime, + ) + assert code == 0, json.dumps(packet, ensure_ascii=False, indent=2) + return packet + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize( + "leases,reason_code", + [ + ([], "task_lease_missing"), + ([_lease(status="released")], "task_lease_released"), + ([_lease(expired=True)], "task_lease_expired"), + ], +) +def test_should_run_requires_one_atomic_claim_before_delivery( + tmp_path, monkeypatch, provider, leases, reason_code, +): + packet = _run(tmp_path, monkeypatch, provider, leases) + preparation = packet["execution_preparation"] + assert preparation["state"] == "claim_required" + assert preparation["reason_code"] == reason_code + assert preparation["claim"]["host_execution_allowed"] is True + assert preparation["claim"]["command_line"].startswith("loopx todo claim ") + assert packet["should_run"] is True + assert packet["normal_delivery_allowed"] is False + assert packet["selected_todo"]["claim_required_before_work"] is True + contract = packet["interaction_contract"] + assert contract["agent_channel"]["must_attempt"] is True + assert contract["agent_channel"]["delivery_allowed"] is False + assert contract["cli_channel"]["next_cli_actions"][0] == preparation["claim"]["command_line"] + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_matching_active_lease_preserves_delivery(tmp_path, monkeypatch, provider): + packet = _run(tmp_path, monkeypatch, provider, [_lease()]) + assert packet["execution_preparation"]["state"] == "ready" + assert packet["normal_delivery_allowed"] is True + assert packet["interaction_contract"]["agent_channel"]["delivery_allowed"] is True + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_projected_claim_prepares_the_same_todo_for_a_fresh_guard( + tmp_path, monkeypatch, provider, +): + before = _run(tmp_path, monkeypatch, provider, []) + claim_args = before["execution_preparation"]["claim"]["cli_args"] + + claim_code, claim = run_json_cli_result( + *claim_args, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + ) + assert claim_code == 0, json.dumps(claim, ensure_ascii=False, indent=2) + + guard_code, after = run_json_cli_result( + "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + ) + assert guard_code == 0, json.dumps(after, ensure_ascii=False, indent=2) + assert after["selected_todo"]["todo_id"] == TODO + assert after["execution_preparation"]["state"] == "ready" + assert after["normal_delivery_allowed"] is True + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_foreign_active_lease_blocks_without_claim_action(tmp_path, monkeypatch, provider): + packet = _run(tmp_path, monkeypatch, provider, [_lease(owner="agent-b")]) + preparation = packet["execution_preparation"] + assert preparation["state"] == "blocked_by_conflict" + assert preparation["reason_code"] == "active_lease_owned_by_other_agent" + assert "claim" not in preparation + assert packet["should_run"] is False + assert packet["normal_delivery_allowed"] is False + + +@pytest.mark.parametrize("boundary", ["selection", "receipt"]) +def test_preparation_does_not_replace_selection_or_receipt_identity( + monkeypatch, boundary, +): + from loopx.control_plane.quota import execution_preparation as module + + monkeypatch.setattr( + module, + "project_selected_todo_execution_preparation", + lambda **_kwargs: pytest.fail("preparation must wait for the current binding"), + ) + payload = { + "should_run": True, + "selected_todo": {"todo_id": TODO}, + } + if boundary == "selection": + payload["action_portfolio"] = { + "selection_policy": {"requires_explicit_turn_binding": True}, + } + apply_selected_todo_execution_preparation( + payload, + runtime_root="/synthetic/runtime", + goal_id=GOAL, + agent_id=AGENT, + receipt_bound=boundary == "receipt", + ) + assert "execution_preparation" not in payload diff --git a/tests/control_plane_ts/execution_preparation.test.ts b/tests/control_plane_ts/execution_preparation.test.ts new file mode 100644 index 0000000000..9e1c3c7491 --- /dev/null +++ b/tests/control_plane_ts/execution_preparation.test.ts @@ -0,0 +1,87 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import {EffectRuntimeRequestError} from "../../loopx/control_plane/effect_runtime_errors.ts"; +import {projectExecutionPreparation} from "../../loopx/control_plane/work_items/execution_preparation.ts"; + +function facts(patch: Record = {}) { + return { + schema_version: "execution_preparation_facts_v0", + goal_id: "goal-example", + agent_id: "agent-a", + handoff_mode: "hard_lease", + authority_available: true, + selected_todo: {todo_id: "todo_example", status: "open", claimed_by: "agent-a"}, + leases: [], + ...patch, + }; +} + +test("matching active lease is ready", () => { + assert.deepEqual(projectExecutionPreparation(facts({leases: [ + {status: "active", active: true, owner: "agent-a"}, + ]})), { + schema_version: "execution_preparation_v0", + todo_id: "todo_example", + agent_id: "agent-a", + handoff_mode: "hard_lease", + state: "ready", + delivery_allowed: true, + reason_code: "matching_active_task_lease", + }); +}); + +for (const [label, leases, reasonCode] of [ + ["missing", [], "task_lease_missing"], + ["released", [{status: "released", active: false, owner: "agent-a"}], "task_lease_released"], + ["expired", [{status: "active", active: false, owner: "agent-a"}], "task_lease_expired"], +] as const) { + test(`${label} lease returns one atomic managed claim`, () => { + const result = projectExecutionPreparation(facts({leases})); + assert.equal(result.state, "claim_required"); + assert.equal(result.delivery_allowed, false); + assert.equal(result.reason_code, reasonCode); + assert.deepEqual((result.claim as Record).cli_args, [ + "todo", "claim", "--goal-id", "goal-example", "--todo-id", "todo_example", + "--agent-id", "agent-a", "--claimed-by", "agent-a", + ]); + assert.equal((result.claim as Record).host_execution_allowed, true); + }); +} + +test("foreign ownership remains blocked", () => { + const lease = projectExecutionPreparation(facts({leases: [ + {status: "active", active: true, owner: "agent-b"}, + ]})); + assert.equal(lease.state, "blocked_by_conflict"); + assert.equal(lease.reason_code, "active_lease_owned_by_other_agent"); + assert.equal(lease.delivery_allowed, false); + assert.equal(Object.hasOwn(lease, "claim"), false); + + const todo = projectExecutionPreparation(facts({selected_todo: { + todo_id: "todo_example", status: "open", claimed_by: "agent-b", + }})); + assert.equal(todo.state, "blocked_by_conflict"); + assert.equal(todo.reason_code, "todo_claimed_by_other_agent"); +}); + +test("soft claims preserve delivery and unavailable authority fails closed", () => { + const soft = projectExecutionPreparation(facts({handoff_mode: "soft_claim"})); + assert.equal(soft.state, "not_required"); + assert.equal(soft.delivery_allowed, true); + + const unavailable = projectExecutionPreparation(facts({authority_available: false})); + assert.equal(unavailable.state, "authority_unavailable"); + assert.equal(unavailable.delivery_allowed, false); +}); + +test("malformed facts are rejected", () => { + assert.throws( + () => projectExecutionPreparation(facts({handoff_mode: "future"})), + EffectRuntimeRequestError, + ); + assert.throws( + () => projectExecutionPreparation(facts({leases: [{status: "released", active: true}]})), + /active requires status=active/, + ); +}); diff --git a/tests/test_kunluncode_goal_mode.py b/tests/test_kunluncode_goal_mode.py index 69815dceaa..bd9fe3f927 100644 --- a/tests/test_kunluncode_goal_mode.py +++ b/tests/test_kunluncode_goal_mode.py @@ -1409,6 +1409,54 @@ def factory(command: list[str], **kwargs) -> _FakeAppServer: assert "Implement and verify the native Goal Pro adapter" not in journal +def test_native_goal_prepares_released_execution_before_model_launch( + tmp_path: Path, +) -> None: + todo = _native_todo(claimed=True) + + class PreparationControl(_FakeControlPlane): + def should_run(self) -> dict[str, object]: + self.calls.append("should_run") + ready = any(call.startswith("claim:") for call in self.calls) + return { + "ok": True, + "should_run": True, + "selected_todo": self.todo, + "normal_delivery_allowed": ready, + "execution_preparation": { + "schema_version": "execution_preparation_v0", + "state": "ready" if ready else "claim_required", + "delivery_allowed": ready, + }, + } + + control = PreparationControl(todo) + clients: list[_FakeAppServer] = [] + + def factory(command: list[str], **kwargs) -> _FakeAppServer: + client = _FakeAppServer(command, **kwargs) + clients.append(client) + return client + + result = run_native_goal( + tmp_path, + _native_context(tmp_path), + mode="goal-pro", + permission_mode="auto", + max_duration_secs=60, + controller_timeout_secs=120, + kunluncode_bin="/usr/bin/kunluncode", + control_plane=control, + client_factory=factory, + ) + + assert result["status"] == "committed" + assert control.calls[:3] == [ + "should_run", "claim:todo-native-1", "should_run", + ] + assert clients and clients[0].calls[0] == "initialize" + + def test_source_native_goal_rejects_alias_only_runtime_state( tmp_path: Path, ) -> None: From fd50d6d9f0755113a4296166217a654ffb011ce5 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 3 Nov 2026 19:45:50 +0800 Subject: [PATCH 02/10] fix(control-plane): preserve lease readiness on turn reentry Signed-off-by: huangruiteng --- docs/status-data-contract.md | 7 +- .../quota/execution_preparation.py | 12 ++- .../test_quota_execution_preparation.py | 81 ++++++++++++++++--- 3 files changed, 86 insertions(+), 14 deletions(-) diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index 85d5632f1f..07eab032e6 100644 --- a/docs/status-data-contract.md +++ b/docs/status-data-contract.md @@ -1413,9 +1413,10 @@ 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 and receipt-bound -recovery keep their existing identities and are never replaced by automatic -preparation. Hosts without this integration expose only the claim command for +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. Correctly typed future work is handled earlier. A higher-priority diff --git a/loopx/control_plane/quota/execution_preparation.py b/loopx/control_plane/quota/execution_preparation.py index 7827c8c513..e379d02b9d 100644 --- a/loopx/control_plane/quota/execution_preparation.py +++ b/loopx/control_plane/quota/execution_preparation.py @@ -109,7 +109,6 @@ def apply_selected_todo_execution_preparation( if ( not isinstance(selected, dict) or payload.get("should_run") is not True - or receipt_bound or action_portfolio_requires_explicit_selection(payload) ): return @@ -119,6 +118,17 @@ def apply_selected_todo_execution_preparation( 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 diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index a1c2904fe2..5b1c4d91ef 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -56,7 +56,13 @@ def _lease(*, owner: str = AGENT, status: str = "active", expired: bool = False) } -def _run(tmp_path, monkeypatch, provider: str, leases: list[dict[str, object]]): +def _run( + tmp_path, + monkeypatch, + provider: str, + leases: list[dict[str, object]], + *guard_args: str, +): isolate_sqlite_runtime(tmp_path, monkeypatch) runtime = tmp_path / "runtime" registry = tmp_path / "registry.json" @@ -93,6 +99,7 @@ def _run(tmp_path, monkeypatch, provider: str, leases: list[dict[str, object]]): ) code, packet = run_json_cli_result( "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + *guard_args, registry_path=registry, runtime_root=runtime, ) assert code == 0, json.dumps(packet, ensure_ascii=False, indent=2) @@ -159,6 +166,65 @@ def test_projected_claim_prepares_the_same_todo_for_a_fresh_guard( assert after["normal_delivery_allowed"] is True +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_projected_claim_prepares_the_same_receipt_bound_turn( + tmp_path, monkeypatch, provider, +): + unbound = _run( + tmp_path, + monkeypatch, + provider, + [], + "--runtime-profile", + "codex_app_ssh_goal", + "--context-delivery", + "off", + "--begin-turn", + ) + assert "heartbeat_receipt" in unbound, json.dumps( + unbound, ensure_ascii=False, indent=2, + ) + turn = unbound["heartbeat_receipt"]["turn_instance_id"] + guard_code, before = run_json_cli_result( + "quota", "should-run", + "--runtime-profile", "codex_app_ssh_goal", + "--context-delivery", "off", + "--goal-id", GOAL, + "--agent-id", AGENT, + "--turn-instance-id", turn, + "--todo-id", TODO, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + ) + assert guard_code == 0, json.dumps(before, ensure_ascii=False, indent=2) + assert before["execution_preparation"]["state"] == "claim_required" + assert before["heartbeat_receipt"]["turn_instance_id"] == turn + + claim_code, claim = run_json_cli_result( + *before["execution_preparation"]["claim"]["cli_args"], + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + ) + assert claim_code == 0, json.dumps(claim, ensure_ascii=False, indent=2) + + guard_code, after = run_json_cli_result( + "quota", "should-run", + "--runtime-profile", "codex_app_ssh_goal", + "--context-delivery", "off", + "--goal-id", GOAL, + "--agent-id", AGENT, + "--turn-instance-id", turn, + "--todo-id", TODO, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + ) + assert guard_code == 0, json.dumps(after, ensure_ascii=False, indent=2) + assert after["heartbeat_receipt"]["turn_instance_id"] == turn + assert after["selected_todo"]["todo_id"] == TODO + assert after["execution_preparation"]["state"] == "ready" + assert after["normal_delivery_allowed"] is True + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_foreign_active_lease_blocks_without_claim_action(tmp_path, monkeypatch, provider): packet = _run(tmp_path, monkeypatch, provider, [_lease(owner="agent-b")]) @@ -170,10 +236,7 @@ def test_foreign_active_lease_blocks_without_claim_action(tmp_path, monkeypatch, assert packet["normal_delivery_allowed"] is False -@pytest.mark.parametrize("boundary", ["selection", "receipt"]) -def test_preparation_does_not_replace_selection_or_receipt_identity( - monkeypatch, boundary, -): +def test_preparation_does_not_replace_explicit_action_selection(monkeypatch): from loopx.control_plane.quota import execution_preparation as module monkeypatch.setattr( @@ -185,15 +248,13 @@ def test_preparation_does_not_replace_selection_or_receipt_identity( "should_run": True, "selected_todo": {"todo_id": TODO}, } - if boundary == "selection": - payload["action_portfolio"] = { - "selection_policy": {"requires_explicit_turn_binding": True}, - } + payload["action_portfolio"] = { + "selection_policy": {"requires_explicit_turn_binding": True}, + } apply_selected_todo_execution_preparation( payload, runtime_root="/synthetic/runtime", goal_id=GOAL, agent_id=AGENT, - receipt_bound=boundary == "receipt", ) assert "execution_preparation" not in payload From 752a858e0eb1465a2c5bbe106d3c3baa17e17ad8 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 3 Nov 2026 20:13:27 +0800 Subject: [PATCH 03/10] fix(codex): auto-prepare recovered task leases Signed-off-by: huangruiteng --- docs/status-data-contract.md | 16 ++ loopx/cli_commands/quota_capture.py | 46 +++- .../turn_driver/codex_context.py | 76 +++++- .../codex_execution_preparation.py | 243 ++++++++++++++++++ .../test_codex_context_lifecycle.py | 58 +++++ .../test_codex_execution_preparation.py | 150 +++++++++++ .../test_quota_execution_preparation.py | 106 ++++++++ 7 files changed, 691 insertions(+), 4 deletions(-) create mode 100644 loopx/control_plane/turn_driver/codex_execution_preparation.py create mode 100644 tests/control_plane/test_codex_execution_preparation.py diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index 07eab032e6..935ee56c36 100644 --- a/docs/status-data-contract.md +++ b/docs/status-data-contract.md @@ -1419,6 +1419,22 @@ 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 diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 37a31d584f..53ad215a3e 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/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,9 +1111,16 @@ 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_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("{}") @@ -1124,9 +1132,43 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca 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 diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index c6a0f848e2..829d6c818f 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -7,7 +7,7 @@ from __future__ import annotations from contextlib import contextmanager -from collections.abc import Iterator +from collections.abc import Iterator, Mapping from typing import Any import json import os @@ -26,6 +26,7 @@ NATIVE_CONTEXT_LARGE_OUTPUT_BYTES, NATIVE_CONTEXT_HISTORY_TOKENS, NATIVE_CONTEXT_OUTPUT_ENV, native_context_page_budget, ) +from .codex_execution_preparation import compact_execution_preparation_request # Native adapter activation reuses scheduler-owned host declarations. These # settings are presentation opt-ins, never host authentication or authority. @@ -370,6 +371,75 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str return {} +def native_execution_preparation_request( + directory: str | None, event: dict[str, Any], +) -> dict[str, Any] | None: + """Read one compact Host action from this exact native session journal.""" + + if directory is None or event.get("hook_event_name") not in {"SessionStart", "PostToolUse"}: + return None + session, cwd = event.get("session_id"), event.get("cwd") + if not isinstance(session, str) or not session or not isinstance(cwd, str) or not cwd: + return None + try: + with _locked(directory) as root: + state = _state(root) + if state.get("session_id") != session or state.get("cwd") != cwd: + return None + recovery = state.get("recovery") + if not isinstance(recovery, dict): + return None + request = recovery.get("execution_preparation") + source_sha256 = recovery.get("sha256") + if not isinstance(request, dict) or not isinstance(source_sha256, str): + return None + previous = state.get("host_execution_preparation") + if ( + event.get("hook_event_name") != "SessionStart" + and isinstance(previous, dict) + and previous.get("source_sha256") == source_sha256 + and previous.get("state") == "ready" + ): + return None + return {**request, "source_sha256": source_sha256} + except (OSError, ValueError, TypeError, RecursionError): + return None + + +def record_native_execution_preparation( + directory: str | None, + event: dict[str, Any], + *, + source_sha256: str, + receipt: Mapping[str, Any], +) -> None: + """Record a ready receipt only while its source/session is still current.""" + + if directory is None or receipt.get("event") != "host_execution_prepared": + return + with _locked(directory) as root: + state = _state(root) + recovery = state.get("recovery") + if ( + state.get("session_id") != event.get("session_id") + or state.get("cwd") != event.get("cwd") + or not isinstance(recovery, dict) + or recovery.get("sha256") != source_sha256 + ): + raise ValueError("execution-preparation source moved before receipt publication") + state["host_execution_preparation"] = { + "source_sha256": source_sha256, + "state": "ready", + "todo_id": receipt.get("todo_id"), + **( + {"turn_instance_id": receipt["turn_instance_id"]} + if receipt.get("turn_instance_id") + else {} + ), + } + _save(root, state) + + def _context_pages(value: object) -> list[dict[str, Any]]: """Decode native shell/MCP JSON records; never classify arbitrary prose. @@ -537,9 +607,11 @@ def project_codex_context(payload: dict[str, Any], *, directory: str | None, cap def publish_recovery(reference: dict[str, Any]) -> None: reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"], page_bytes=page_bytes, max_page_output_bytes=output_bytes) + preparation = compact_execution_preparation_request(payload) state["recovery"] = {**{key: reference[key] for key in ("path", "sha256", "command")}, "goal_id": goal_id, "agent_id": agent_id, - "total_bytes": Path(reference["path"]).stat().st_size} + "total_bytes": Path(reference["path"]).stat().st_size, + **({"execution_preparation": preparation} if preparation else {})} state["pending"] = {"path": reference["path"], "sha256": reference["sha256"], "total_bytes": state["recovery"]["total_bytes"], "ranges": {}} _save(root, state) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py new file mode 100644 index 0000000000..f4c2d8c7c1 --- /dev/null +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -0,0 +1,243 @@ +"""Codex lifecycle-hook executor for the typed execution-preparation action. + +The TypeScript control plane remains the decision owner. This adapter accepts +only its exact ordinary claim, executes it through the public CLI, and proves +the same Todo (and Turn when present) is ready before returning a Host receipt. +""" +from __future__ import annotations + +from collections.abc import Callable, Mapping, Sequence +import json +from pathlib import Path +import subprocess +from typing import Any + +from ...command_invocation import current_loopx_cli_argv + + +CommandRunner = Callable[..., subprocess.CompletedProcess[str]] + + +def compact_execution_preparation_request( + payload: Mapping[str, Any], +) -> dict[str, Any] | None: + """Retain only the exact Host-executable action beside a private capture.""" + + preparation = payload.get("execution_preparation") + selected = payload.get("selected_todo") + if not isinstance(preparation, Mapping) or not isinstance(selected, Mapping): + return None + claim = preparation.get("claim") + goal_id = str(payload.get("goal_id") or "").strip() + agent_id = str(preparation.get("agent_id") or "").strip() + todo_id = str(preparation.get("todo_id") or "").strip() + selected_todo_id = str(selected.get("todo_id") or "").strip() + registry_path = Path(str(payload.get("registry") or "")).expanduser() + runtime_root = Path(str(payload.get("runtime_root") or "")).expanduser() + if ( + preparation.get("schema_version") != "execution_preparation_v0" + or preparation.get("state") != "claim_required" + or not isinstance(claim, Mapping) + or claim.get("host_execution_allowed") is not True + or not goal_id + or not agent_id + or not todo_id + or selected_todo_id != todo_id + or not registry_path.is_absolute() + or registry_path.resolve() != registry_path + or not runtime_root.is_absolute() + or runtime_root.resolve() != runtime_root + ): + return None + claim_args = claim.get("cli_args") + expected_claim = [ + "todo", "claim", "--goal-id", goal_id, "--todo-id", todo_id, + "--agent-id", agent_id, "--claimed-by", agent_id, + ] + if claim_args != expected_claim: + return None + receipt = payload.get("heartbeat_receipt") + turn_instance_id = ( + str(receipt.get("turn_instance_id") or "").strip() + if isinstance(receipt, Mapping) + else "" + ) + return { + "goal_id": goal_id, + "agent_id": agent_id, + "todo_id": todo_id, + "claim_args": expected_claim, + # The lifecycle hook can run from any project. Retain the exact route + # that produced this digest-bound action instead of rediscovering Goal + # authority from the hook process cwd. + "authority_route": { + "registry_path": str(registry_path), + "runtime_root": str(runtime_root), + }, + **({"turn_instance_id": turn_instance_id} if turn_instance_id else {}), + } + + +def _default_runner( + command: Sequence[str], *, cwd: Path, timeout: float, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + list(command), cwd=str(cwd), capture_output=True, text=True, + encoding="utf-8", errors="replace", timeout=timeout, + ) + + +def _run_json( + arguments: Sequence[str], + *, + registry_path: Path, + runtime_root: str | Path | None, + cwd: Path, + timeout: float, + runner: CommandRunner, +) -> dict[str, Any]: + command = [ + *current_loopx_cli_argv(), + "--registry", str(registry_path), + *(["--runtime-root", str(runtime_root)] if runtime_root is not None else []), + "--format", "json", + *arguments, + ] + completed = runner(command, cwd=cwd, timeout=timeout) + try: + payload = json.loads(completed.stdout or "{}") + except json.JSONDecodeError as exc: + raise RuntimeError("LoopX Host preparation returned non-JSON output") from exc + if ( + completed.returncode != 0 + or not isinstance(payload, dict) + or payload.get("ok") is False + ): + reason = payload.get("reason") if isinstance(payload, dict) else None + raise RuntimeError(str(reason or completed.stderr or "LoopX Host preparation failed")) + return payload + + +def execute_codex_execution_preparation( + request: Mapping[str, Any], + *, + cwd: Path, + scenario: str, + timeout: float = 8, + runner: CommandRunner | None = None, +) -> dict[str, Any]: + """Execute one projected claim and prove same-identity readiness. + + The hook never selects work. Unknown/malformed requests fail before a + subprocess. The ordinary claim remains the authority and rechecks all + ownership, lease-generation, workspace and write-scope constraints. + """ + + goal_id = str(request.get("goal_id") or "").strip() + agent_id = str(request.get("agent_id") or "").strip() + todo_id = str(request.get("todo_id") or "").strip() + turn_instance_id = str(request.get("turn_instance_id") or "").strip() + authority_route = request.get("authority_route") + registry_path = Path( + str(authority_route.get("registry_path") or "") + if isinstance(authority_route, Mapping) + else "" + ).expanduser() + runtime_root = Path( + str(authority_route.get("runtime_root") or "") + if isinstance(authority_route, Mapping) + else "" + ).expanduser() + expected_claim = [ + "todo", "claim", "--goal-id", goal_id, "--todo-id", todo_id, + "--agent-id", agent_id, "--claimed-by", agent_id, + ] + if ( + not goal_id + or not agent_id + or not todo_id + or request.get("claim_args") != expected_claim + or not registry_path.is_absolute() + or registry_path.resolve() != registry_path + or not runtime_root.is_absolute() + or runtime_root.resolve() != runtime_root + or scenario not in {"codex_app_heartbeat", "codex_app_ssh_goal"} + ): + raise ValueError("invalid Codex execution-preparation request") + workspace = Path(cwd) + if ( + not workspace.is_absolute() + or not workspace.is_dir() + or workspace.resolve() != workspace + ): + raise ValueError("Codex execution preparation requires a physical workspace") + invoke = runner or _default_runner + _run_json( + expected_claim, + registry_path=registry_path, + runtime_root=runtime_root, + cwd=workspace, + timeout=timeout / 2, + runner=invoke, + ) + guard = [ + "quota", "should-run", + "--goal-id", goal_id, + "--agent-id", agent_id, + *( ["--codex-app"] if scenario == "codex_app_heartbeat" else + ["--runtime-profile", "codex_app_ssh_goal"] ), + "--context-delivery", "off", + "--todo-id", todo_id, + *(["--turn-instance-id", turn_instance_id] if turn_instance_id else []), + ] + after = _run_json( + guard, + registry_path=registry_path, + runtime_root=runtime_root, + cwd=workspace, + timeout=timeout / 2, + runner=invoke, + ) + preparation = after.get("execution_preparation") + selected = after.get("selected_todo") + receipt = after.get("heartbeat_receipt") + same_turn = not turn_instance_id or ( + isinstance(receipt, Mapping) + and receipt.get("turn_instance_id") == turn_instance_id + ) + ready = ( + after.get("should_run") is True + and after.get("normal_delivery_allowed") is True + and isinstance(selected, Mapping) + and selected.get("todo_id") == todo_id + and isinstance(preparation, Mapping) + and preparation.get("state") == "ready" + and preparation.get("delivery_allowed") is True + and same_turn + ) + if not ready: + raise RuntimeError( + "LoopX Host claim completed without a same-Todo ready postcondition" + ) + return { + "event": "host_execution_prepared", + "goal_id": goal_id, + "agent_id": agent_id, + "todo_id": todo_id, + **({"turn_instance_id": turn_instance_id} if turn_instance_id else {}), + "execution_preparation": dict(preparation), + "normal_delivery_allowed": True, + } + + +def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str: + """Render the small Host receipt that supersedes an earlier stale guard.""" + + return ( + "LoopX Host automatically restored the exact selected Todo lease and " + "rechecked the same Todo/Turn before this model step. Treat the earlier " + "claim_required state as superseded by this ready receipt; do not repeat " + "its claim or guard solely for that lease state. Current gates and later " + "settlement checks still apply. Receipt: " + + json.dumps(dict(receipt), ensure_ascii=False, separators=(",", ":")) + ) diff --git a/tests/control_plane/test_codex_context_lifecycle.py b/tests/control_plane/test_codex_context_lifecycle.py index d7c855c7f6..30386f5414 100644 --- a/tests/control_plane/test_codex_context_lifecycle.py +++ b/tests/control_plane/test_codex_context_lifecycle.py @@ -35,6 +35,64 @@ def test_shared_root_keeps_interleaved_native_sessions_independent(tmp_path, mon assert (directory / "session.json").read_bytes() == raw +def test_native_journal_projects_one_host_execution_request(tmp_path, monkeypatch): + event = setup(tmp_path, monkeypatch) + payload = { + "ok": True, + "goal_id": "fixture", + "registry": str(tmp_path / "registry.json"), + "runtime_root": str(tmp_path / "runtime"), + "should_run": True, + "selected_todo": {"todo_id": "todo_fixture", "claimed_by": "agent"}, + "execution_preparation": { + "schema_version": "execution_preparation_v0", + "state": "claim_required", + "todo_id": "todo_fixture", + "agent_id": "agent", + "claim": { + "host_execution_allowed": True, + "cli_args": [ + "todo", "claim", "--goal-id", "fixture", + "--todo-id", "todo_fixture", "--agent-id", "agent", + "--claimed-by", "agent", + ], + }, + }, + "heartbeat_receipt": {"turn_instance_id": "turn-fixture"}, + } + _, wire = project(tmp_path, payload=payload) + request = context.native_execution_preparation_request( + str(tmp_path), {**event, "hook_event_name": "PostToolUse"}, + ) + assert request == { + "goal_id": "fixture", + "agent_id": "agent", + "todo_id": "todo_fixture", + "claim_args": payload["execution_preparation"]["claim"]["cli_args"], + "authority_route": { + "registry_path": str(tmp_path / "registry.json"), + "runtime_root": str(tmp_path / "runtime"), + }, + "turn_instance_id": "turn-fixture", + "source_sha256": wire["recovery"]["sha256"], + } + receipt = { + "event": "host_execution_prepared", + "todo_id": "todo_fixture", + "turn_instance_id": "turn-fixture", + } + context.record_native_execution_preparation( + str(tmp_path), event, + source_sha256=request["source_sha256"], receipt=receipt, + ) + assert context.native_execution_preparation_request( + str(tmp_path), {**event, "hook_event_name": "PostToolUse"}, + ) is None + assert context.native_execution_preparation_request( + str(tmp_path), {**event, "hook_event_name": "SessionStart"}, + )["todo_id"] == "todo_fixture" + + def test_shared_root_read_does_not_create_unknown_session_or_follow_alias(tmp_path, monkeypatch): tmp_path.chmod(0o700) monkeypatch.setenv("CODEX_THREAD_ID", "unknown") diff --git a/tests/control_plane/test_codex_execution_preparation.py b/tests/control_plane/test_codex_execution_preparation.py new file mode 100644 index 0000000000..c346845c70 --- /dev/null +++ b/tests/control_plane/test_codex_execution_preparation.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import json +from pathlib import Path +import subprocess + +import pytest + +from loopx.control_plane.turn_driver.codex_execution_preparation import ( + codex_execution_preparation_context, + compact_execution_preparation_request, + execute_codex_execution_preparation, +) + + +GOAL = "goal-example" +TODO = "todo_example" +AGENT = "agent-a" + + +def _packet(*, turn: str | None = "turn-example") -> dict[str, object]: + return { + "ok": True, + "goal_id": GOAL, + "registry": "/tmp/loopx-project-a/.loopx/registry.json", + "runtime_root": "/tmp/loopx-runtime-a", + "should_run": True, + "selected_todo": {"todo_id": TODO, "claimed_by": AGENT}, + "execution_preparation": { + "schema_version": "execution_preparation_v0", + "state": "claim_required", + "todo_id": TODO, + "agent_id": AGENT, + "claim": { + "host_execution_allowed": True, + "cli_args": [ + "todo", "claim", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", AGENT, "--claimed-by", AGENT, + ], + }, + }, + **( + {"heartbeat_receipt": {"turn_instance_id": turn}} + if turn is not None + else {} + ), + } + + +def _completed(payload: dict[str, object], *, code: int = 0) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess( + args=[], returncode=code, stdout=json.dumps(payload), stderr="", + ) + + +def test_compact_request_accepts_only_exact_host_action() -> None: + request = compact_execution_preparation_request(_packet()) + assert request == { + "goal_id": GOAL, + "agent_id": AGENT, + "todo_id": TODO, + "turn_instance_id": "turn-example", + "claim_args": [ + "todo", "claim", "--goal-id", GOAL, "--todo-id", TODO, + "--agent-id", AGENT, "--claimed-by", AGENT, + ], + "authority_route": { + "registry_path": "/tmp/loopx-project-a/.loopx/registry.json", + "runtime_root": "/tmp/loopx-runtime-a", + }, + } + + malformed = _packet() + malformed["execution_preparation"]["claim"]["cli_args"][-1] = "agent-b" + assert compact_execution_preparation_request(malformed) is None + + +def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path) -> None: + request = compact_execution_preparation_request(_packet()) + assert request is not None + calls: list[list[str]] = [] + + def runner(command, *, cwd, timeout): + calls.append(list(command)) + if "should-run" not in command: + return _completed({"ok": True, "claimed": True}) + return _completed({ + "ok": True, + "goal_id": GOAL, + "should_run": True, + "normal_delivery_allowed": True, + "selected_todo": {"todo_id": TODO}, + "execution_preparation": { + "schema_version": "execution_preparation_v0", + "state": "ready", + "todo_id": TODO, + "agent_id": AGENT, + "delivery_allowed": True, + }, + "heartbeat_receipt": {"turn_instance_id": "turn-example"}, + }) + + receipt = execute_codex_execution_preparation( + request, + cwd=tmp_path.resolve(), + scenario="codex_app_ssh_goal", + runner=runner, + ) + + assert receipt["event"] == "host_execution_prepared" + assert receipt["normal_delivery_allowed"] is True + assert len(calls) == 2 + for call in calls: + assert call[call.index("--registry") + 1] == ( + "/tmp/loopx-project-a/.loopx/registry.json" + ) + assert call[call.index("--runtime-root") + 1] == "/tmp/loopx-runtime-a" + assert calls[0][-len(request["claim_args"]):] == request["claim_args"] + assert calls[1][-4:] == ["--todo-id", TODO, "--turn-instance-id", "turn-example"] + context = codex_execution_preparation_context(receipt) + assert "do not repeat" in context and '"state":"ready"' in context + + +def test_executor_rejects_changed_todo_or_turn(tmp_path: Path) -> None: + request = compact_execution_preparation_request(_packet()) + assert request is not None + + def runner(command, *, cwd, timeout): + if "should-run" not in command: + return _completed({"ok": True}) + return _completed({ + "ok": True, + "should_run": True, + "normal_delivery_allowed": True, + "selected_todo": {"todo_id": "todo-other"}, + "execution_preparation": { + "schema_version": "execution_preparation_v0", + "state": "ready", + "delivery_allowed": True, + }, + "heartbeat_receipt": {"turn_instance_id": "turn-other"}, + }) + + with pytest.raises(RuntimeError, match="same-Todo ready"): + execute_codex_execution_preparation( + request, + cwd=tmp_path.resolve(), + scenario="codex_app_ssh_goal", + runner=runner, + ) diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index 5b1c4d91ef..25b712c966 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -2,6 +2,10 @@ from datetime import timedelta import json +import os +from pathlib import Path +import subprocess +import sys import pytest @@ -258,3 +262,105 @@ def test_preparation_does_not_replace_explicit_action_selection(monkeypatch): agent_id=AGENT, ) assert "execution_preparation" not in payload + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_codex_lifecycle_hook_uses_captured_authority_from_another_project( + tmp_path, monkeypatch, provider, +): + from loopx.control_plane.turn_driver.codex_context import ( + codex_context_hook, + project_codex_context, + ) + + before = _run( + tmp_path, + monkeypatch, + provider, + [], + "--runtime-profile", "codex_app_ssh_goal", + "--context-delivery", "off", + "--begin-turn", + ) + turn = before["heartbeat_receipt"]["turn_instance_id"] + context_root = tmp_path / "context" + capture_root = tmp_path / "capture" + context_root.mkdir(mode=0o700) + capture_root.mkdir(mode=0o700) + foreign_workspace = tmp_path / "project-b" + foreign_workspace.mkdir() + monkeypatch.setenv("CODEX_THREAD_ID", "codex-hook-session") + monkeypatch.chdir(foreign_workspace) + event = { + "session_id": "codex-hook-session", + "cwd": str(foreign_workspace.resolve()), + "hook_event_name": "SessionStart", + "source": "startup", + } + assert codex_context_hook(str(context_root), event) == {} + project_codex_context( + before, + directory=str(context_root), + capture_directory=capture_root, + goal_id=GOAL, + agent_id=AGENT, + surface="quota_packet", + capture_id="claim-required", + ) + + command = [ + sys.executable, "-m", "loopx.entrypoint", + # The lifecycle process may be rooted in another project. These + # deliberately wrong invocation routes must not replace the route + # retained with the captured preparation action. + "--registry", str(foreign_workspace / ".loopx" / "registry.json"), + "--runtime-root", str(foreign_workspace / ".loopx" / "runtime"), + "--format", "json", + "quota", "context-hook", + "--context-delivery", "on", + "--runtime-profile", "codex_app_ssh_goal", + "--codex-context-dir", str(context_root), + ] + post_tool = {**event, "hook_event_name": "PostToolUse", "tool_response": "{}"} + environment = dict(os.environ) + environment["PYTHONPATH"] = str(Path(__file__).resolve().parents[2]) + prepared = subprocess.run( + command, + input=json.dumps(post_tool), + cwd=foreign_workspace, + env=environment, + capture_output=True, + text=True, + timeout=15, + ) + assert prepared.returncode == 0, prepared.stderr + hook = json.loads(prepared.stdout) + additional = hook["hookSpecificOutput"]["additionalContext"] + assert "automatically restored" in additional + assert '"state":"ready"' in additional + assert turn in additional + + guard_code, after = run_json_cli_result( + "quota", "should-run", + "--runtime-profile", "codex_app_ssh_goal", + "--context-delivery", "off", + "--goal-id", GOAL, + "--agent-id", AGENT, + "--turn-instance-id", turn, + "--todo-id", TODO, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + ) + assert guard_code == 0 + assert after["execution_preparation"]["state"] == "ready" + + repeated = subprocess.run( + command, + input=json.dumps(post_tool), + cwd=tmp_path, + env=environment, + capture_output=True, + text=True, + timeout=15, + ) + assert repeated.returncode == 0 and json.loads(repeated.stdout) == {} From e2ca1098cf0ef3eea8132fb235b441216d9ccc3f Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 3 Nov 2026 20:51:54 +0800 Subject: [PATCH 04/10] fix(control-plane): satisfy host preparation lint Signed-off-by: huangruiteng --- .../control_plane/turn_driver/codex_execution_preparation.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index f4c2d8c7c1..a29d19100f 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -6,10 +6,10 @@ """ from __future__ import annotations -from collections.abc import Callable, Mapping, Sequence import json -from pathlib import Path import subprocess +from collections.abc import Callable, Mapping, Sequence +from pathlib import Path from typing import Any from ...command_invocation import current_loopx_cli_argv @@ -84,6 +84,7 @@ def _default_runner( return subprocess.run( list(command), cwd=str(cwd), capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout, + check=False, ) From 459a63e31cd3f3302ad5ba5ebb2a404c90759a94 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Tue, 3 Nov 2026 20:56:04 +0800 Subject: [PATCH 05/10] test(control-plane): cover expired lease recovery Signed-off-by: huangruiteng --- tests/control_plane/test_quota_execution_preparation.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index 25b712c966..7b31c11373 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -265,8 +265,9 @@ def test_preparation_does_not_replace_explicit_action_selection(monkeypatch): @pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("lease_state", ["missing", "expired"]) def test_codex_lifecycle_hook_uses_captured_authority_from_another_project( - tmp_path, monkeypatch, provider, + tmp_path, monkeypatch, provider, lease_state, ): from loopx.control_plane.turn_driver.codex_context import ( codex_context_hook, @@ -277,7 +278,7 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project( tmp_path, monkeypatch, provider, - [], + [] if lease_state == "missing" else [_lease(expired=True)], "--runtime-profile", "codex_app_ssh_goal", "--context-delivery", "off", "--begin-turn", From 3bb33ea13ce7b50d6d1112aedb7992101c483834 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 4 Nov 2026 14:06:49 +0800 Subject: [PATCH 06/10] fix(codex): isolate execution preparation from subagents Apply the native lifecycle parent-event qualification before directory IO, claim execution, manifest reads and receipt publication. Use physical tmp_path routes in executor tests so macOS exercises the same production path checks. Signed-off-by: huangruiteng --- loopx/cli_commands/quota_capture.py | 5 ++-- .../turn_driver/codex_context.py | 14 ++++++++-- .../test_codex_context_lifecycle.py | 8 ++++++ .../test_codex_execution_preparation.py | 27 ++++++++++-------- .../test_quota_execution_preparation.py | 28 +++++++++++++++++++ 5 files changed, 65 insertions(+), 17 deletions(-) diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 53ad215a3e..458c3418f4 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -1111,7 +1111,7 @@ 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_execution_preparation_request, + native_context_event_eligible, native_execution_preparation_request, record_native_execution_preparation, ) from ..control_plane.turn_driver.codex_execution_preparation import ( @@ -1127,7 +1127,8 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca 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) diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index 829d6c818f..370132ce0d 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -39,6 +39,12 @@ class NativeContextSessionUnavailable(ValueError): """No native identity for a session journal; explicit file reads still work.""" +def native_context_event_eligible(event: Mapping[str, Any]) -> bool: + """Reject subagent hooks before parent-session context or lease effects.""" + agent_id = event.get("agent_id") + return not (isinstance(agent_id, str) and agent_id) + + def context_delivery_for_arguments(args: Any, *, codex_machine_preference: bool = False) -> dict[str, Any]: """One declaration adapter for explicit CLI and inherited machine opt-ins.""" fields = {key: getattr(args, key, None) for key in ("host_surface", "scheduler_owner", "execution_mode")} @@ -321,7 +327,7 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str name, session = event.get("hook_event_name"), event.get("session_id") if not isinstance(session, str) or not session: return {} - if isinstance(event.get("agent_id"), str) and event["agent_id"]: + if not native_context_event_eligible(event): return {} # no parent recovery reference may escape to a subagent pages = _context_pages(event.get("tool_response")) if name == "PostToolUse" else [] if name == "PostToolUse" and not pages: @@ -376,7 +382,8 @@ def native_execution_preparation_request( ) -> dict[str, Any] | None: """Read one compact Host action from this exact native session journal.""" - if directory is None or event.get("hook_event_name") not in {"SessionStart", "PostToolUse"}: + if (directory is None or not native_context_event_eligible(event) + or event.get("hook_event_name") not in {"SessionStart", "PostToolUse"}): return None session, cwd = event.get("session_id"), event.get("cwd") if not isinstance(session, str) or not session or not isinstance(cwd, str) or not cwd: @@ -415,7 +422,8 @@ def record_native_execution_preparation( ) -> None: """Record a ready receipt only while its source/session is still current.""" - if directory is None or receipt.get("event") != "host_execution_prepared": + if (directory is None or not native_context_event_eligible(event) + or receipt.get("event") != "host_execution_prepared"): return with _locked(directory) as root: state = _state(root) diff --git a/tests/control_plane/test_codex_context_lifecycle.py b/tests/control_plane/test_codex_context_lifecycle.py index 30386f5414..30ccd76d69 100644 --- a/tests/control_plane/test_codex_context_lifecycle.py +++ b/tests/control_plane/test_codex_context_lifecycle.py @@ -76,11 +76,19 @@ def test_native_journal_projects_one_host_execution_request(tmp_path, monkeypatc "turn_instance_id": "turn-fixture", "source_sha256": wire["recovery"]["sha256"], } + child_event = {**event, "hook_event_name": "PostToolUse", "agent_id": "child"} + assert context.native_execution_preparation_request(str(tmp_path), child_event) is None receipt = { "event": "host_execution_prepared", "todo_id": "todo_fixture", "turn_instance_id": "turn-fixture", } + state_before_child = (tmp_path / "session.json").read_bytes() + context.record_native_execution_preparation( + str(tmp_path), child_event, + source_sha256=request["source_sha256"], receipt=receipt, + ) + assert (tmp_path / "session.json").read_bytes() == state_before_child context.record_native_execution_preparation( str(tmp_path), event, source_sha256=request["source_sha256"], receipt=receipt, diff --git a/tests/control_plane/test_codex_execution_preparation.py b/tests/control_plane/test_codex_execution_preparation.py index c346845c70..dd8cbd0904 100644 --- a/tests/control_plane/test_codex_execution_preparation.py +++ b/tests/control_plane/test_codex_execution_preparation.py @@ -18,12 +18,14 @@ AGENT = "agent-a" -def _packet(*, turn: str | None = "turn-example") -> dict[str, object]: +def _packet(tmp_path: Path, *, turn: str | None = "turn-example") -> dict[str, object]: + registry = (tmp_path / "loopx-project-a" / ".loopx" / "registry.json").resolve() + runtime_root = (tmp_path / "loopx-runtime-a").resolve() return { "ok": True, "goal_id": GOAL, - "registry": "/tmp/loopx-project-a/.loopx/registry.json", - "runtime_root": "/tmp/loopx-runtime-a", + "registry": str(registry), + "runtime_root": str(runtime_root), "should_run": True, "selected_todo": {"todo_id": TODO, "claimed_by": AGENT}, "execution_preparation": { @@ -53,8 +55,9 @@ def _completed(payload: dict[str, object], *, code: int = 0) -> subprocess.Compl ) -def test_compact_request_accepts_only_exact_host_action() -> None: - request = compact_execution_preparation_request(_packet()) +def test_compact_request_accepts_only_exact_host_action(tmp_path: Path) -> None: + packet = _packet(tmp_path) + request = compact_execution_preparation_request(packet) assert request == { "goal_id": GOAL, "agent_id": AGENT, @@ -65,18 +68,18 @@ def test_compact_request_accepts_only_exact_host_action() -> None: "--agent-id", AGENT, "--claimed-by", AGENT, ], "authority_route": { - "registry_path": "/tmp/loopx-project-a/.loopx/registry.json", - "runtime_root": "/tmp/loopx-runtime-a", + "registry_path": str((tmp_path / "loopx-project-a" / ".loopx" / "registry.json").resolve()), + "runtime_root": str((tmp_path / "loopx-runtime-a").resolve()), }, } - malformed = _packet() + malformed = _packet(tmp_path) malformed["execution_preparation"]["claim"]["cli_args"][-1] = "agent-b" assert compact_execution_preparation_request(malformed) is None def test_executor_claims_then_proves_same_turn_ready(tmp_path: Path) -> None: - request = compact_execution_preparation_request(_packet()) + request = compact_execution_preparation_request(_packet(tmp_path)) assert request is not None calls: list[list[str]] = [] @@ -112,9 +115,9 @@ def runner(command, *, cwd, timeout): assert len(calls) == 2 for call in calls: assert call[call.index("--registry") + 1] == ( - "/tmp/loopx-project-a/.loopx/registry.json" + str((tmp_path / "loopx-project-a" / ".loopx" / "registry.json").resolve()) ) - assert call[call.index("--runtime-root") + 1] == "/tmp/loopx-runtime-a" + assert call[call.index("--runtime-root") + 1] == str((tmp_path / "loopx-runtime-a").resolve()) assert calls[0][-len(request["claim_args"]):] == request["claim_args"] assert calls[1][-4:] == ["--todo-id", TODO, "--turn-instance-id", "turn-example"] context = codex_execution_preparation_context(receipt) @@ -122,7 +125,7 @@ def runner(command, *, cwd, timeout): def test_executor_rejects_changed_todo_or_turn(tmp_path: Path) -> None: - request = compact_execution_preparation_request(_packet()) + request = compact_execution_preparation_request(_packet(tmp_path)) assert request is not None def runner(command, *, cwd, timeout): diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index 7b31c11373..526b260c40 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -325,6 +325,34 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project( post_tool = {**event, "hook_event_name": "PostToolUse", "tool_response": "{}"} environment = dict(os.environ) environment["PYTHONPATH"] = str(Path(__file__).resolve().parents[2]) + + # A subagent may share its parent's native session and cwd, but it must not + # execute or receive the parent's claim action. + child = subprocess.run( + command, + input=json.dumps({**post_tool, "agent_id": "child-agent"}), + cwd=foreign_workspace, + env=environment, + capture_output=True, + text=True, + timeout=15, + ) + assert child.returncode == 0, child.stderr + assert json.loads(child.stdout) == {} + child_guard_code, child_guard = run_json_cli_result( + "quota", "should-run", + "--runtime-profile", "codex_app_ssh_goal", + "--context-delivery", "off", + "--goal-id", GOAL, + "--agent-id", AGENT, + "--turn-instance-id", turn, + "--todo-id", TODO, + registry_path=tmp_path / "registry.json", + runtime_root=tmp_path / "runtime", + ) + assert child_guard_code == 0 + assert child_guard["execution_preparation"]["state"] == "claim_required" + prepared = subprocess.run( command, input=json.dumps(post_tool), From c10b100c049e8298622e5bc96e6756631dd3d8b5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 4 Nov 2026 22:12:28 +0800 Subject: [PATCH 07/10] fix(quota): bind qualified selections before lease preparation Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/status-data-contract.md | 7 ++-- .../work_items/action_selection_admission.ts | 13 +++++++- .../work_items/action_selection_contract.py | 1 + .../test_quota_execution_preparation.py | 23 +++++++++++-- .../action_selection_admission.test.ts | 33 +++++++++++++++++++ 5 files changed, 72 insertions(+), 5 deletions(-) diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index 935ee56c36..deb62052bf 100644 --- a/docs/status-data-contract.md +++ b/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 diff --git a/loopx/control_plane/work_items/action_selection_admission.ts b/loopx/control_plane/work_items/action_selection_admission.ts index 22e4a0b34a..2557b3f747 100644 --- a/loopx/control_plane/work_items/action_selection_admission.ts +++ b/loopx/control_plane/work_items/action_selection_admission.ts @@ -5,6 +5,7 @@ import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; import {jsonObject, requireBoolean, requireJsonObject} from "../runtime_decode.ts"; import {EffectiveAction} from "../quota/effective_action.generated.ts"; import {isCanonicalTodoId} from "../todos/identity.ts"; +import {EXECUTION_PREPARATION_SCHEMA_VERSION} from "./execution_preparation.ts"; interface AgentAdmissionFacts {must_attempt: boolean; delivery_allowed: boolean | null} interface SelectionAdmission {selected_todo_id: string | null; admitted: boolean} @@ -41,10 +42,20 @@ export function currentActionSelectionAdmission( const workspaceRepair = pending && packet.workspace_repair_allowed === true && packet.effective_action === EffectiveAction.AGENT_WORKSPACE_REPAIR && agent.must_attempt && agent.delivery_allowed === false; + const preparation = jsonObject(packet.execution_preparation); + // Binding qualified work may precede its ordinary claim; it does not grant delivery. + const preparationOnly = pending && qualification.state === "qualified" + && selectionTodoId(qualification.requested_todo_id) === requestedTodoId + && preparation?.schema_version === EXECUTION_PREPARATION_SCHEMA_VERSION + && preparation.state === "claim_required" + && selectionTodoId(preparation.todo_id) === selectedId + && preparation.delivery_allowed === false && packet.normal_delivery_allowed === false + && agent.must_attempt && agent.delivery_allowed === false; const currentWork = !pending && agent.must_attempt; return {selected_todo_id: selectedId, admitted: selectedId !== null && selectedId === requestedTodoId - && packet.ok === true && packet.should_run === true && (delivery || scopedDelivery || workspaceRepair || currentWork)}; + && packet.ok === true && packet.should_run === true + && (delivery || scopedDelivery || workspaceRepair || preparationOnly || currentWork)}; } export function actionSelectionNeedsRecovery(packet: JsonObject, agent: AgentAdmissionFacts): boolean { diff --git a/loopx/control_plane/work_items/action_selection_contract.py b/loopx/control_plane/work_items/action_selection_contract.py index aed37f1eb5..dbb17630ba 100644 --- a/loopx/control_plane/work_items/action_selection_contract.py +++ b/loopx/control_plane/work_items/action_selection_contract.py @@ -168,6 +168,7 @@ def action_selection_context(payload: Mapping[str, Any]) -> dict[str, Any]: ("selected_todo", ("todo_id", "selection_binding")), ("autonomous_replan_obligation", ("selection_binding",)), ("action_selection_qualification", ("state", "requested_todo_id", "recovery_action")), + ("execution_preparation", ("schema_version", "state", "todo_id", "delivery_allowed")), ): raw = payload.get(field) if isinstance(raw, Mapping): diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index 526b260c40..99246a9b45 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -10,6 +10,7 @@ import pytest from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime +from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.runtime.time import now_utc, utc_isoformat from loopx.control_plane.testing.canary_harness import write_fixture_registry, run_json_cli_result @@ -66,6 +67,7 @@ def _run( provider: str, leases: list[dict[str, object]], *guard_args: str, + todos: list[dict[str, object]] | None = None, ): isolate_sqlite_runtime(tmp_path, monkeypatch) runtime = tmp_path / "runtime" @@ -94,7 +96,7 @@ def _run( ) projection = build_todo_runtime_shadow_projection( goal_id=GOAL, - todos=[_todo()], + todos=[_todo()] if todos is None else todos, leases=leases, handoff_mode="hard_lease", ) @@ -171,8 +173,9 @@ def test_projected_claim_prepares_the_same_todo_for_a_fresh_guard( @pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("multiple_candidates", [False, True]) def test_projected_claim_prepares_the_same_receipt_bound_turn( - tmp_path, monkeypatch, provider, + tmp_path, monkeypatch, provider, multiple_candidates, ): unbound = _run( tmp_path, @@ -184,11 +187,15 @@ def test_projected_claim_prepares_the_same_receipt_bound_turn( "--context-delivery", "off", "--begin-turn", + todos=[_todo(), {**_todo(), "todo_id": "todo_other_candidate", "index": 2}] + if multiple_candidates else None, ) assert "heartbeat_receipt" in unbound, json.dumps( unbound, ensure_ascii=False, indent=2, ) turn = unbound["heartbeat_receipt"]["turn_instance_id"] + if multiple_candidates: + assert unbound["interaction_contract"]["cli_channel"]["selection_required"] is True guard_code, before = run_json_cli_result( "quota", "should-run", "--runtime-profile", "codex_app_ssh_goal", @@ -203,6 +210,18 @@ def test_projected_claim_prepares_the_same_receipt_bound_turn( assert guard_code == 0, json.dumps(before, ensure_ascii=False, indent=2) assert before["execution_preparation"]["state"] == "claim_required" assert before["heartbeat_receipt"]["turn_instance_id"] == turn + assert before["selected_todo"]["todo_id"] == TODO + assert before["selected_todo"]["selection_binding"] == "heartbeat_receipt" + assert before["normal_delivery_allowed"] is False + contract = before["interaction_contract"] + assert contract["agent_channel"]["delivery_allowed"] is False + assert contract["cli_channel"]["next_cli_actions"] == [ + before["execution_preparation"]["claim"]["command_line"], + ] + assert "settlement_plan" not in contract["cli_channel"] + assert not read_canonical_todos_if_promoted( + runtime_root=tmp_path / "runtime", goal_id=GOAL, include_leases=True, + )["leases"] claim_code, claim = run_json_cli_result( *before["execution_preparation"]["claim"]["cli_args"], diff --git a/tests/control_plane_ts/action_selection_admission.test.ts b/tests/control_plane_ts/action_selection_admission.test.ts index ce7cb07c7d..d3802df582 100644 --- a/tests/control_plane_ts/action_selection_admission.test.ts +++ b/tests/control_plane_ts/action_selection_admission.test.ts @@ -72,6 +72,39 @@ test("pending delivery cannot borrow root permission from a denied or missing ch } }); +test("qualified claim preparation binds only its exact Todo while delivery stays denied", () => { + const preparation = {schema_version: "execution_preparation_v0", state: "claim_required", + todo_id: "todo_target", delivery_allowed: false}; + const packet: JsonObject = {...source("qualified"), normal_delivery_allowed: false, + selected_todo: {todo_id: "todo_target", selection_binding: "pending_action_selection"}, + execution_preparation: preparation}; + const preparing = {must_attempt: true, delivery_allowed: false}; + assert.equal(currentActionSelectionAdmission(packet, "todo_target", preparing).admitted, true); + assert.equal(packet.normal_delivery_allowed, false); + assert.equal(preparation.delivery_allowed, false); + for (const patch of [ + {execution_preparation: null}, + ...["schema_version", "state", "todo_id", "delivery_allowed"].map(field => ({ + execution_preparation: {...preparation, [field]: field === "delivery_allowed" ? true : "other"}, + })), + ...["ready", "blocked_by_conflict", "authority_unavailable"].map(state => ({ + execution_preparation: {...preparation, state}, + })), + {action_selection_qualification: {state: "deferred", requested_todo_id: "todo_target"}}, + {action_selection_qualification: {state: "rejected", requested_todo_id: "todo_target"}}, + {action_selection_qualification: {state: "qualified", requested_todo_id: "todo_other"}}, + {normal_delivery_allowed: true}, {ok: false}, {should_run: false}, + ]) { + assert.equal(currentActionSelectionAdmission({...packet, ...patch}, "todo_target", preparing).admitted, + false, JSON.stringify(patch)); + } + for (const facts of [{must_attempt: false, delivery_allowed: false}, agent, + {must_attempt: true, delivery_allowed: null}]) { + assert.equal(currentActionSelectionAdmission(packet, "todo_target", facts).admitted, false); + } + assert.equal(currentActionSelectionAdmission(packet, "todo_other", preparing).admitted, false); +}); + test("compatibility annotations neither grant nor veto canonical current-work admission", () => { for (const legacy of [undefined, {}, {must_attempt_work: false}, {must_attempt_work: true}]) { const packet = {...source(), selected_todo: {todo_id: "todo_target"}}; From 73d10d1c5d7163755397f8925c3fd0a3e7983f6e Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Wed, 4 Nov 2026 22:26:48 +0800 Subject: [PATCH 08/10] refactor(quota): keep execution preparation presentation with its owner Signed-off-by: huangruiteng --- .../quota/execution_preparation.py | 31 +++++++++++++++++ .../work_items/interaction_contract.py | 34 ++----------------- 2 files changed, 33 insertions(+), 32 deletions(-) diff --git a/loopx/control_plane/quota/execution_preparation.py b/loopx/control_plane/quota/execution_preparation.py index e379d02b9d..792c508937 100644 --- a/loopx/control_plane/quota/execution_preparation.py +++ b/loopx/control_plane/quota/execution_preparation.py @@ -156,3 +156,34 @@ def apply_selected_todo_execution_preparation( 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) diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index bf63a0bb60..dad4f6819b 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -3,6 +3,7 @@ from ..goals.acceptance_readback import configured_checks from ..quota.blocked_transition_notice import blocked_priority_fallback_owner_reason from ..quota.effective_action import EffectiveAction +from ..quota.execution_preparation import attach_interaction_execution_preparation from ..effect_runtime import effect_runtime_result import json import shlex @@ -1578,37 +1579,6 @@ def _attach_interaction_vision_wait_state( } -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) - - def _interaction_fallback_policy_required(payload: Mapping[str, object], *, mode: str) -> bool: return mode in { "user_gate", @@ -1747,7 +1717,7 @@ def _build_interaction_contract_projection( registry_path=registry_path, ), } - _attach_interaction_execution_preparation(contract, payload) + attach_interaction_execution_preparation(contract, payload) response_plan = _build_interaction_response_plan( user_channel=user_channel, agent_channel=agent_channel, From f433a88d080f8144612c12f0aaafed3479bd9271 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 00:11:42 +0800 Subject: [PATCH 09/10] test(leases): qualify restored work on the ordinary preparation path Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../test_full_loopx_checkpoint_restore.py | 31 ++++++++++++++++++- .../test_quota_execution_preparation.py | 5 ++- 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/tests/control_plane/test_full_loopx_checkpoint_restore.py b/tests/control_plane/test_full_loopx_checkpoint_restore.py index 1a8e66dd57..6c8b8d6d9f 100644 --- a/tests/control_plane/test_full_loopx_checkpoint_restore.py +++ b/tests/control_plane/test_full_loopx_checkpoint_restore.py @@ -23,7 +23,12 @@ @pytest.fixture -def retained_point(tmp_path, monkeypatch, capsys, request): +def release_source_lease(): + return False + + +@pytest.fixture +def retained_point(tmp_path, monkeypatch, capsys, request, release_source_lease): isolate_sqlite_runtime(tmp_path, monkeypatch) workspace, runtime, registry = tmp_path / "workspace", tmp_path / "runtime", tmp_path / "control/registry.json" workspace.mkdir() @@ -102,6 +107,9 @@ def cli(*args): settled = read_heartbeat_settlement(runtime, goal_id="research", agent_id="researcher", todo_id=todo, turn_instance_id="source-work", registry_path=registry) assert settled.progress["state"] == "settled" + if release_source_lease: + cli("task-lease", "release", "--goal-id", "research", "--todo-id", todo, "--owner", "researcher", + "--idempotency-key", lease["idempotency_key"], "--expected-version", str(lease["version"])) metadata = {"worker_file_snapshot": {"components": [{"id": "workspace", "source_path": str(workspace)}]}, "session": {"session_id": source_session, "session_profile_digest": wake.digest}} if real_host: @@ -180,6 +188,27 @@ def fork(*args, **kwargs): assert len(c["calls"]) == (1 if c["real_host"] else 0) +@pytest.mark.parametrize("release_source_lease", [True]) +@pytest.mark.parametrize("retained_point", [("file", False), ("sqlite", False)], indirect=True) +def test_restored_released_lease_requires_claim_for_same_work(retained_point): + c = retained_point + _empty_receiver(c) + warm.restore_worker(c["declaration"], c["workspace"]) + cli = c["cli"] + binding = ("--goal-id", "research", "--agent-id", "researcher", "--todo-id", c["todo"], + "--turn-instance-id", "resumed-work") + args = ("quota", "should-run", *binding, "--runtime-profile", "generic_cli", "--context-delivery", "off") + pending = cli(*args) + assert pending["execution_preparation"]["reason_code"] == "task_lease_released" + assert pending["normal_delivery_allowed"] is False + assert cli(*pending["execution_preparation"]["claim"]["cli_args"])["ok"] + ready = cli(*args) + assert ready["execution_preparation"]["state"] == "ready" and ready["normal_delivery_allowed"] + assert ready["selected_todo"]["todo_id"] == c["todo"] + assert ready["heartbeat_receipt"]["turn_instance_id"] == "resumed-work" + assert len(cli("todo", "list", "--goal-id", "research")["todos"]) == 2 + + def test_full_restore_cannot_replace_a_binding_changed_during_native_fork(retained_point, monkeypatch): c = retained_point _empty_receiver(c) diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index 99246a9b45..bc5733821a 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -105,6 +105,7 @@ def _run( ) code, packet = run_json_cli_result( "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + "--context-delivery", "off", *guard_args, registry_path=registry, runtime_root=runtime, ) @@ -163,6 +164,7 @@ def test_projected_claim_prepares_the_same_todo_for_a_fresh_guard( guard_code, after = run_json_cli_result( "quota", "should-run", "--goal-id", GOAL, "--agent-id", AGENT, + "--context-delivery", "off", registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime", ) @@ -184,8 +186,6 @@ def test_projected_claim_prepares_the_same_receipt_bound_turn( [], "--runtime-profile", "codex_app_ssh_goal", - "--context-delivery", - "off", "--begin-turn", todos=[_todo(), {**_todo(), "todo_id": "todo_other_candidate", "index": 2}] if multiple_candidates else None, @@ -299,7 +299,6 @@ def test_codex_lifecycle_hook_uses_captured_authority_from_another_project( provider, [] if lease_state == "missing" else [_lease(expired=True)], "--runtime-profile", "codex_app_ssh_goal", - "--context-delivery", "off", "--begin-turn", ) turn = before["heartbeat_receipt"]["turn_instance_id"] From 5d174e710228a8198cfd804a9c4ebe560954161c Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 00:26:39 +0800 Subject: [PATCH 10/10] fix(leases): surface execution preparation in the first context read Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/progressive_context.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index b0ca41608a..eacd124481 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/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"];