From b4237bb86edad5d145aaff7c40a3aa4732e8fbc5 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 01:10:06 +0800 Subject: [PATCH] fix(tasks): manage execution leases through ordinary task operations Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/handoff-mode.md | 51 +++++ loopx/cli_commands/todo.py | 18 ++ .../cli_commands/todo_argument_validation.py | 4 +- loopx/cli_commands/todo_registration.py | 2 +- .../coordination/claim_execution.ts | 106 ++++++++++ .../coordination/local_authority.py | 5 +- .../coordination/local_authority_runtime.ts | 26 ++- .../control_plane/coordination/todo_claim.ts | 58 ++++-- .../coordination/todo_terminal_lifecycle.ts | 13 +- .../control_plane/coordination/todo_update.ts | 22 ++- .../coordination/todo_update_intent.ts | 4 + loopx/control_plane/todos/markdown.py | 4 + loopx/control_plane/todos/mutation_api.py | 7 +- .../todos/provider_terminal_lifecycle.py | 6 +- loopx/control_plane/todos/provider_update.py | 8 +- .../test_sqlite_authority_cli.py | 10 +- .../test_todo_claim_argument_recovery.py | 36 +++- .../authority_store_conformance.ts | 2 + .../local_authority_runtime.test.ts | 18 ++ .../managed_claim_execution_conformance.ts | 182 ++++++++++++++++++ 20 files changed, 545 insertions(+), 37 deletions(-) create mode 100644 loopx/control_plane/coordination/claim_execution.ts create mode 100644 tests/control_plane_ts/managed_claim_execution_conformance.ts diff --git a/docs/reference/handoff-mode.md b/docs/reference/handoff-mode.md index b58e75a0e7..79dd2e8a21 100644 --- a/docs/reference/handoff-mode.md +++ b/docs/reference/handoff-mode.md @@ -26,6 +26,42 @@ lease or change ownership policy to make the completion succeed. Other preview failures retain their own diagnostics. This explains an unavailable operation; it does not qualify App completion under `hard_lease`. +## Ordinary Agent CLI work + +A CLI self-claim (`--claimed-by` equal to the explicit `--agent-id`) now prepares +execution in the same canonical transaction. In `hard_lease`, it acquires the +required scopes or reuses the caller's still-active grant. It renews only in the +last third of the grant's original TTL. `soft_claim` does not create a lease. +Assignment by another caller and explicit task-lease commands keep their existing +semantics; this does not change any Goal's policy. + +```bash +loopx --format json todo claim --goal-id example-goal --todo-id todo_example \ + --agent-id worker-a --claimed-by worker-a --claim-operation-id work-a +loopx --format json todo update --goal-id example-goal --todo-id todo_example \ + --agent-id worker-a --claim-operation-id work-a --update-operation-id progress-a \ + --note 'Validated progress' +loopx --format json todo complete --goal-id example-goal --todo-id todo_example \ + --agent-id worker-a --claim-operation-id work-a --evidence evidence:validated-output +``` + +Claim readback supplies ready progress/completion commands in JSON and Markdown +with the original registry/runtime route. Normal self-claim output carries the +claim reference instead of raw lease counters or a duplicated receipt; explicit +lease commands retain their diagnostic output. Pre-upgrade plain-claim retries +recover only their original assignment and never become managed execution. +Keep any required original Turn identity and settlement. +Progress can renew the still-active execution in the **same** write transaction; +completion uses the existing terminal owner and release. Neither needs a separate +inspection or a copied lease key/version. The reference resolves only the execution +accepted by that claim: a later generation, foreign holder, expired grant, changed +policy or unavailable receipt cannot be borrowed or silently reacquired. An +explicit proof cannot be combined with the reference. Receipt recovery happens +before maintenance, so retrying an accepted operation does not renew or repeat it. +A read-only quota guard never claims or renews. Long-running delegated Hosts retain +their existing supervision; this CLI path does not supervise an idle external +process. The App's manual completion boundary described above is unchanged. + ## Choose the policy independently of storage Use `soft_claim` for an explicitly single executor without parallel takeover. @@ -279,3 +315,18 @@ claim+lease 均不可执行，必须先审核迁移；合并式 claim+lease 仍 冲突不会得到不可执行的 acquire 建议；修改已租用工作的要求或状态须走对应 lifecycle。 提示不包含 execution key、不授予权限，也不修改 mode 或 provider。并发造成 version 变化时重新读取，不能绕过 CAS；dry-run 和拒绝路径不产生 Todo 或租约写入。 + +### Agent 普通 CLI 操作 + +CLI 由明确的 `--agent-id` 认领给自己时，会在同一 canonical 事务中准备执行； +`hard_lease` 自动取得任务要求的 scope，或复用自己尚有效的执行。仅在原租期最后 +三分之一内续租，`soft_claim` 不创建租约，其他调用方分配任务和显式租约操作保持原义。 +认领结果在 JSON 和 Markdown 中返回保留 registry/runtime 路由的进展与完成命令， +常规输出不再展开租约计数和重复回执，显式租约操作保留诊断读回。升级前旧认领只 +恢复原来的分配，不被升级成新执行。新认领的进展与完成命令均携带原 +`--claim-operation-id`，无需单独 inspect 或复制 key/version。进展写回按需在同一事务 +续租，完成复用原终态 owner 及释放操作；必要的原 Turn 身份与结算仍须保留。 +该引用只承接当初认领接受的执行，过期、换代、外部占用、策略变化或回执不可用时拒绝， +不自动借用或重新取得证明；显式证明不能与引用混用。原操作回执先于维护恢复，重试不 +重复写入或续租。只读 quota guard 不做认领或续租；长期委派 Host 仍走已有监督， +本 CLI 路径不监督空闲外部进程，也不改变 App 手工完成的现有边界。 diff --git a/loopx/cli_commands/todo.py b/loopx/cli_commands/todo.py index 999d436409..9c5a8765b8 100644 --- a/loopx/cli_commands/todo.py +++ b/loopx/cli_commands/todo.py @@ -494,12 +494,28 @@ def handle_todo_command( claimed_by=args.claimed_by, agent_id=args.agent_id, claim_only=True, + manage_execution=(args.agent_id is not None and args.agent_id == args.claimed_by + and args.task_lease_idempotency_key is None and args.task_lease_expected_version is None), claim_operation_id=args.claim_operation_id, task_lease_idempotency_key=args.task_lease_idempotency_key, task_lease_expected_version=args.task_lease_expected_version, **_todo_path_args(args), dry_run=bool(args.dry_run), ) + if payload.get("ok") and payload.get("execution_managed") and not args.dry_run: + route = ["loopx", "--registry", str(registry_path), "--runtime-root", + str(resolve_runtime_root(load_registry(registry_path), runtime_root_arg)), "--format", "json"] + selected = ["--goal-id", args.goal_id, "--todo-id", args.todo_id, + "--agent-id", args.agent_id, "--claim-operation-id", payload["claim_operation_id"]] + payload["next_cli_actions"] = [shlex.join([*route, "todo", action, *selected, *extra]) for action, extra in ( + ("update", ["--update-operation-id", "", "--note", ""]), + ("complete", ["--evidence", ""]), + )] + payload["instruction"] = "Continue the task. Use this original claim reference for progress or completion; retain normal Turn settlement when applicable." + # Normal callers need the accepted task reference, not lease + # counters or a second inline copy of the native receipt. + for internal in ("lease", "lease_changed", "lease_idempotent", "original_receipt"): + payload.pop(internal, None) elif args.todo_command == "update": validate_todo_update_options(args) payload = update_goal_todo( @@ -507,6 +523,7 @@ def handle_todo_command( runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, todo_id=args.todo_id, + claim_operation_id=args.claim_operation_id, text=args.text, priority=args.priority, clear_priority=args.clear_priority, @@ -667,6 +684,7 @@ def handle_todo_command( runtime_root_arg=runtime_root_arg, goal_id=args.goal_id, todo_id=args.todo_id, + claim_operation_id=args.claim_operation_id, role=args.role, decision_outcome=args.decision_outcome, evidence=args.evidence, diff --git a/loopx/cli_commands/todo_argument_validation.py b/loopx/cli_commands/todo_argument_validation.py index aacf8820a5..969994b3b7 100644 --- a/loopx/cli_commands/todo_argument_validation.py +++ b/loopx/cli_commands/todo_argument_validation.py @@ -612,8 +612,8 @@ def validate_shared_todo_options(args: argparse.Namespace) -> None: raise ValueError("--update-operation-id is supported only by todo update") if getattr(args, "update_expected_provider_revision", None) is not None and args.todo_command != "update": raise ValueError("--update-expected-provider-revision is supported only by todo update") - if getattr(args, "claim_operation_id", None) is not None and args.todo_command != "claim": - raise ValueError("--claim-operation-id is supported only by todo claim") + if getattr(args, "claim_operation_id", None) is not None and args.todo_command not in {"claim", "update", "complete"}: + raise ValueError("--claim-operation-id is supported only by todo claim/update/complete") if ( getattr(args, "completion_identity_key", None) and args.todo_command != "complete" diff --git a/loopx/cli_commands/todo_registration.py b/loopx/cli_commands/todo_registration.py index d53ed67c9d..79b4a87e8a 100644 --- a/loopx/cli_commands/todo_registration.py +++ b/loopx/cli_commands/todo_registration.py @@ -75,7 +75,7 @@ def register_todo_command( todo_parser.add_argument( "--claim-operation-id", help=( - "For todo claim on promoted canonical authority only, reuse this public-safe " + "For claim/update/complete on canonical authority, reuse this public-safe " "operation id across retries. Changed intent with the same id is rejected; " "receipt replay proves historical acceptance, not current lease ownership. " "Omit to retain a fresh operation id per invocation." diff --git a/loopx/control_plane/coordination/claim_execution.ts b/loopx/control_plane/coordination/claim_execution.ts new file mode 100644 index 0000000000..9e74eb517f --- /dev/null +++ b/loopx/control_plane/coordination/claim_execution.ts @@ -0,0 +1,106 @@ +/** Ordinary task operations carry the accepted claim, not hand-copied lease + * counters. The canonical lifecycle/fence owners still decide every effect. */ +import type {JsonObject} from "../effect_program.ts"; +import type {AuthorityStore} from "./authority_store.ts"; +import {AuthorityStoreProtocolError, canonicalAuthorityObject, canonicalAuthoritySha256, + requireAuthorityStoreId} from "./authority_store_codec.ts"; +import {indexCoordinationProjection} from "./coordination_projection.ts"; +import {canonicalTaskLease, canonicalTaskLeaseAcquireFacts} from "./task_lease_state.ts"; +import {coordinationTodoWriteScopes} from "./todo_write_scopes.ts"; +import {leaseEpoch, leaseVersion, leaseIsActive, normalizeTtl, type LeaseRecord} from "../work_items/task_lease_acquire.ts"; +import {sameLeaseWorkspace} from "../work_items/task_lease_workspace.ts"; +import {leaseWriteRepository} from "../work_items/task_lease_repository.ts"; +import {evaluateTaskLeaseAcquireDecision, materializeTaskLeaseAcquire} from "../work_items/task_lease_acquire_decision.ts"; +import {decideTaskLeaseLifecycle, materializeTaskLeaseLifecycle} from "../work_items/task_lease_lifecycle_decision.ts"; + +export function claimExecutionReference(id: unknown, key: unknown, version: unknown): string | undefined { + if (id == null) return undefined; + const value = requireAuthorityStoreId(id, "claim_operation_id"); + if (key != null || version != null) throw new AuthorityStoreProtocolError( + "Use the original claim operation or explicit task-lease proof, not both"); + return value; +} + +type ExecutionInput = {goal_id: string; todo_id: string; actor_agent_id: string | null; + registered_agents: readonly string[]; now: Date}; + +/** Resolve only this accepted execution. No acquisition, adoption of a newer + * generation, or write retry is hidden behind a progress/completion request. */ +export async function claimExecutionProof(store: AuthorityStore, head: JsonObject, + input: ExecutionInput, operationId: string, allowReleasedCloseout = false): Promise<{lease_idempotency_key: string; lease_expected_version: number} | null> { + const receipt = await store.readReceipt(operationId); + if (receipt.status !== "found" || receipt.receipts.length !== 1) { + throw new AuthorityStoreProtocolError("Original task claim unavailable; recover that claim before writing"); + } + const original = receipt.receipts[0]; + const result = canonicalAuthorityObject(original.result, "original task claim"); + if (original.schema_version !== "loopx_coordination_todo_claim_receipt_v0" || + original.operation_id !== operationId || original.goal_id !== input.goal_id || + result.todo_id !== input.todo_id || result.claimed_by !== input.actor_agent_id || + result.execution_managed !== true) throw new AuthorityStoreProtocolError( + "The claim reference does not name this actor's managed task execution"); + // A soft claim has no execution lease. Its normal writer still checks policy. + if (result.handoff_mode !== "hard_lease") { + if (result.handoff_mode === "soft_claim" && head.handoff_mode === "soft_claim") return null; + throw new AuthorityStoreProtocolError("Task ownership policy changed; review a new task claim"); + } + if (head.handoff_mode !== "hard_lease") throw new AuthorityStoreProtocolError("Task ownership policy changed; review the current task before continuing"); + const accepted = canonicalTaskLease(canonicalAuthorityObject(result.lease, "accepted execution"), input.goal_id, input.todo_id); + const index = indexCoordinationProjection(head, input.goal_id); + const raw = index.leases.get(input.todo_id); + const current = raw && canonicalTaskLease(raw, input.goal_id, input.todo_id); + const retainedCloseout = allowReleasedCloseout && current?.status === "released" && + index.todos.get(input.todo_id)?.status === "done"; + if (accepted.owner !== input.actor_agent_id || !current || !(leaseIsActive(current, input.now) || retainedCloseout) || current.owner !== input.actor_agent_id || current.idempotency_key !== accepted.idempotency_key || + leaseEpoch(current) !== leaseEpoch(accepted) || leaseVersion(current) < leaseVersion(accepted) || + !sameLeaseWorkspace(current.write_workspace, accepted.write_workspace) || + leaseWriteRepository(current.write_repository) !== leaseWriteRepository(accepted.write_repository)) { + throw new AuthorityStoreProtocolError("Original task execution changed; reconcile task ownership before a new claim"); + } + return {lease_idempotency_key: String(current.idempotency_key), lease_expected_version: leaseVersion(current)}; +} + +/** Maintain only a still-active grant, using the normal lifecycle decision. + * Normal activity renews in the final third of its original TTL; an earlier + * read/write is not a renewal, and an expired grant cannot be resurrected. */ +export function maintainClaimExecution(head: JsonObject, input: ExecutionInput, lease: LeaseRecord): LeaseRecord | null { + // Historical grants without an original TTL cannot justify a new duration. + if (typeof lease.acquire_ttl_seconds !== "number") return null; + const ttl = normalizeTtl(lease.acquire_ttl_seconds); + if (!leaseIsActive(lease, input.now) || new Date(String(lease.expires_at)).valueOf() - input.now.valueOf() > ttl * 1000 / 3) return null; + const facts = canonicalTaskLeaseAcquireFacts(indexCoordinationProjection(head, input.goal_id), + input.goal_id, input.todo_id, input.registered_agents, input.now); + const command = {operation: "renew" as const, owner: String(lease.owner), + idempotency_key: String(lease.idempotency_key), expected_version: leaseVersion(lease), + ttl_seconds: ttl, new_owner: null, new_idempotency_key: null}; + const decision = decideTaskLeaseLifecycle({handoff_mode: String(head.handoff_mode), + registered_agents: input.registered_agents, todo: facts.todo, lease: facts.lease, command}); + if (decision.outcome !== "apply") throw new AuthorityStoreProtocolError(`Task execution maintenance rejected: ${decision.code}`); + return materializeTaskLeaseLifecycle(lease, command, decision, input.now); +} + +/** Atomic claim preparation, reusing acquire/lifecycle semantics and the + * caller's provider CAS. A foreign retained execution needs explicit recovery. */ +export function prepareClaimExecution(head: JsonObject, + input: ExecutionInput & {operation_id: string; claimed_by: string}): {lease: LeaseRecord; changed: boolean} { + const index = indexCoordinationProjection(head, input.goal_id); + const facts = canonicalTaskLeaseAcquireFacts(index, input.goal_id, input.todo_id, input.registered_agents, input.now); + const current = facts.current; + if (current && current.owner !== input.claimed_by && current.status !== "released") { + throw new AuthorityStoreProtocolError("Another task execution is retained; its owner must release or hand it over"); + } + if (current && leaseIsActive(current, input.now)) { + if (current.owner !== input.claimed_by) throw new AuthorityStoreProtocolError("Task is occupied by another actor"); + const renewed = maintainClaimExecution(head, input, current); + return {lease: renewed ?? current, changed: renewed !== null}; + } + const command = {owner: input.claimed_by, + idempotency_key: `claim-${canonicalAuthoritySha256({goal_id: input.goal_id, todo_id: input.todo_id, + owner: input.claimed_by, operation_id: input.operation_id})}`, + expected_version: current ? leaseVersion(current) : 0, ttl_seconds: normalizeTtl(null), + write_scopes: coordinationTodoWriteScopes(index.todos.get(input.todo_id)!)}; + const decision = evaluateTaskLeaseAcquireDecision({handoff_mode: String(head.handoff_mode), + registered_agents: [...input.registered_agents], ...facts, command}); + if (decision.outcome !== "apply") throw new AuthorityStoreProtocolError(`Task claim execution rejected: ${decision.code}`); + return {lease: materializeTaskLeaseAcquire(input, command, decision, input.now), changed: true}; +} diff --git a/loopx/control_plane/coordination/local_authority.py b/loopx/control_plane/coordination/local_authority.py index e5b315e237..883489bc13 100644 --- a/loopx/control_plane/coordination/local_authority.py +++ b/loopx/control_plane/coordination/local_authority.py @@ -154,6 +154,7 @@ def claim_canonical_todo_if_promoted( operation_id: str | None = None, task_lease_idempotency_key: str | None = None, task_lease_expected_version: int | None = None, + manage_execution: bool = False, project: Path | None = None, state_file: Path | None = None, ) -> dict[str, Any]: @@ -163,7 +164,8 @@ def claim_canonical_todo_if_promoted( result = effect_runtime_result( LOCAL_COORDINATION_TODO_CLAIM_METHOD, { - "schema_version": LOCAL_COORDINATION_TODO_CLAIM_WITNESSED_REQUEST_SCHEMA, + "schema_version": ("loopx_local_coordination_todo_claim_request_v2" if manage_execution + else LOCAL_COORDINATION_TODO_CLAIM_WITNESSED_REQUEST_SCHEMA), "runtime_root": str(runtime_root.expanduser().resolve(strict=False)), "goal_id": goal_id, "todo_id": todo_id, @@ -177,6 +179,7 @@ def claim_canonical_todo_if_promoted( if operation_id is not None else f"todo-claim:{goal_id}:{todo_id}:{uuid4().hex}" ), + **({"manage_execution": True} if manage_execution else {}), "lease_request": ( { "idempotency_key": task_lease_idempotency_key, diff --git a/loopx/control_plane/coordination/local_authority_runtime.ts b/loopx/control_plane/coordination/local_authority_runtime.ts index 37d07db622..8d400d3c9a 100644 --- a/loopx/control_plane/coordination/local_authority_runtime.ts +++ b/loopx/control_plane/coordination/local_authority_runtime.ts @@ -104,6 +104,9 @@ import { export const LOCAL_COORDINATION_TODO_CLAIM_REQUEST_SCHEMA = "loopx_local_coordination_todo_claim_request_v0"; export const LOCAL_COORDINATION_TODO_CLAIM_WITNESSED_REQUEST_SCHEMA = "loopx_local_coordination_todo_claim_request_v1"; +export const LOCAL_COORDINATION_TODO_MANAGED_CLAIM_REQUEST_SCHEMA = "loopx_local_coordination_todo_claim_request_v2"; +export const LOCAL_COORDINATION_TODO_EXECUTION_UPDATE_REQUEST_SCHEMA = "loopx_local_coordination_todo_update_request_v6"; +export const LOCAL_COORDINATION_TODO_EXECUTION_TERMINAL_REQUEST_SCHEMA = "loopx_local_coordination_todo_terminal_lifecycle_request_v4"; export const LOCAL_COORDINATION_TODO_CREATE_REQUEST_SCHEMA = "loopx_local_coordination_todo_create_request_v0"; export const LOCAL_COORDINATION_TODO_CREATE_WITNESSED_REQUEST_SCHEMA = "loopx_local_coordination_todo_create_request_v1"; @@ -1026,11 +1029,14 @@ export async function claimLocalCoordinationTodo( try { const input = requireJsonObject(value, "local coordination Todo claim request"); if (input.schema_version !== LOCAL_COORDINATION_TODO_CLAIM_REQUEST_SCHEMA && - input.schema_version !== LOCAL_COORDINATION_TODO_CLAIM_WITNESSED_REQUEST_SCHEMA) { + input.schema_version !== LOCAL_COORDINATION_TODO_CLAIM_WITNESSED_REQUEST_SCHEMA && + input.schema_version !== LOCAL_COORDINATION_TODO_MANAGED_CLAIM_REQUEST_SCHEMA) { throw new Error("local coordination Todo claim request schema mismatch"); } + if (Object.hasOwn(input, "manage_execution") && input.schema_version !== LOCAL_COORDINATION_TODO_MANAGED_CLAIM_REQUEST_SCHEMA) + throw new TypeError("Managed execution requires claim request v2"); const authoritySourcesCurrent = registryAuthoritySourceCheck(input, - input.schema_version === LOCAL_COORDINATION_TODO_CLAIM_WITNESSED_REQUEST_SCHEMA); + input.schema_version !== LOCAL_COORDINATION_TODO_CLAIM_REQUEST_SCHEMA); if (typeof input.dry_run !== "boolean") { throw new Error("dry_run must be a JSON boolean"); } @@ -1076,6 +1082,7 @@ export async function claimLocalCoordinationTodo( registered_agents: registeredAgents, operation_id: requireAuthorityStoreId(input.operation_id, "operation id"), lease_request: leaseRequest, + ...(input.manage_execution === undefined ? {} : {manage_execution: input.manage_execution as boolean}), dry_run: input.dry_run === true, now: claimObservedAt(input.observed_at), }, authoritySourcesCurrent); @@ -1186,9 +1193,12 @@ export async function updateLocalCoordinationTodo( input.schema_version !== COORDINATION_TODO_REVIEWED_UPDATE_REQUEST_SCHEMA && input.schema_version !== COORDINATION_TODO_COMPLETION_UPDATE_REQUEST_SCHEMA && input.schema_version !== COORDINATION_TODO_OBSERVATION_UPDATE_REQUEST_SCHEMA && - input.schema_version !== COORDINATION_TODO_VALIDATION_REVISION_REQUEST_SCHEMA) { + input.schema_version !== COORDINATION_TODO_VALIDATION_REVISION_REQUEST_SCHEMA && + input.schema_version !== LOCAL_COORDINATION_TODO_EXECUTION_UPDATE_REQUEST_SCHEMA) { throw new TypeError("local coordination Todo update request schema mismatch"); } + if (Object.hasOwn(input, "claim_operation_id") && input.schema_version !== LOCAL_COORDINATION_TODO_EXECUTION_UPDATE_REQUEST_SCHEMA) + throw new TypeError("Claim execution reference requires update request v6"); const planningIntent = input.planning_intent == null ? undefined : requireJsonObject(input.planning_intent, "Todo planning intent"); if (planningIntent && Object.keys(planningIntent).length && @@ -1212,7 +1222,8 @@ export async function updateLocalCoordinationTodo( if (validationRevisionUpdate && input.completion_validation_revision == null) { throw new TypeError("Completion validation revision update requires its revision payload"); } - const reviewed = observationUpdate || completionUpdate || validationRevisionUpdate || + const reviewed = input.schema_version === LOCAL_COORDINATION_TODO_EXECUTION_UPDATE_REQUEST_SCHEMA || + observationUpdate || completionUpdate || validationRevisionUpdate || input.schema_version === COORDINATION_TODO_REVIEWED_UPDATE_REQUEST_SCHEMA; if (!reviewed && ["lifecycle_grants", "authority_reason", "registry_source", "expected_provider_revision", "expected_registry_sha256"].some(field => Object.hasOwn(input, field))) { @@ -1250,6 +1261,7 @@ export async function updateLocalCoordinationTodo( lease_idempotency_key: input.lease_idempotency_key == null ? null : requireAuthorityStoreId(input.lease_idempotency_key, "lease_idempotency_key"), lease_expected_version: optionalNonNegativeSafeInteger(input.lease_expected_version, "lease_expected_version"), + ...(input.claim_operation_id == null ? {} : {claim_operation_id: requireAuthorityStoreId(input.claim_operation_id, "claim_operation_id")}), patch: requireJsonObject(input.patch, "Todo update patch"), planning_intent: planningIntent, ...(observationUpdate ? { @@ -1287,9 +1299,12 @@ export async function terminalLifecycleLocalCoordinationTodo( decision_read_from_provider: true, legacy_fallback_used: false}; try { const input = requireJsonObject(value, "local coordination Todo terminal request"); - if (input.schema_version !== LOCAL_COORDINATION_TODO_TERMINAL_LIFECYCLE_REQUEST_SCHEMA) { + if (input.schema_version !== LOCAL_COORDINATION_TODO_TERMINAL_LIFECYCLE_REQUEST_SCHEMA && + input.schema_version !== LOCAL_COORDINATION_TODO_EXECUTION_TERMINAL_REQUEST_SCHEMA) { throw new TypeError("local coordination Todo terminal request schema mismatch; regenerate with the current runtime"); } + if (Object.hasOwn(input, "claim_operation_id") && input.schema_version !== LOCAL_COORDINATION_TODO_EXECUTION_TERMINAL_REQUEST_SCHEMA) + throw new TypeError("Claim execution reference requires terminal request v4"); const operationIntent = decodeTerminalOperationIntent(input); const reviewBasis = input.review_basis == null ? undefined : requireJsonObject(input.review_basis, "terminal review basis"); const authoritySourcesCurrent = registryAuthoritySourceCheck(input, true, reviewBasis?.registry_sha256); @@ -1353,6 +1368,7 @@ export async function terminalLifecycleLocalCoordinationTodo( input.lease_idempotency_key === null || input.lease_idempotency_key === undefined ? null : requireAuthorityStoreId(input.lease_idempotency_key, "lease idempotency key"), lease_expected_version: leaseExpectedVersion, + ...(input.claim_operation_id == null ? {} : {claim_operation_id: requireAuthorityStoreId(input.claim_operation_id, "claim_operation_id")}), allow_user_gate_auto_acquire: input.allow_user_gate_auto_acquire as boolean, requested_no_followup: input.requested_no_followup as boolean, requested_completion_identity_source: diff --git a/loopx/control_plane/coordination/todo_claim.ts b/loopx/control_plane/coordination/todo_claim.ts index ae688c8acc..8de1db416d 100644 --- a/loopx/control_plane/coordination/todo_claim.ts +++ b/loopx/control_plane/coordination/todo_claim.ts @@ -1,3 +1,4 @@ +import {prepareClaimExecution} from "./claim_execution.ts"; import {retiredHandoffModeRejection, HANDOFF_MODE_RETIRED_REASON} from "./handoff_mode_policy.ts"; import {AUTHORITY_SOURCE_CHANGED, uncheckedAuthoritySource, type AuthoritySourceCheck} from "./authority_source.ts"; import {currentLeaseAcquisitionProof} from "./lease_acquisition_proof.ts"; @@ -61,6 +62,7 @@ export interface CoordinationTodoClaimInput { /** Optional causal binding materialized atomically with an actual runnable claim. */ readonly replan_obligation_id?: string; readonly lease_request?: CoordinationTodoClaimLeaseRequest | null; + readonly manage_execution?: boolean; readonly dry_run: boolean; readonly now: Date; } @@ -366,6 +368,12 @@ async function executeClaimAttempt( registered_agents: normalizeRegisteredTodoAgents(rawInput.registered_agents), lease_request: normalizeLeaseRequest(rawInput.lease_request), }; + if (input.manage_execution !== undefined && typeof input.manage_execution !== "boolean") { + throw new AuthorityStoreProtocolError("manage_execution must be a boolean"); + } + if (input.manage_execution && (input.lease_request != null || input.actor_agent_id !== input.claimed_by)) { + throw new AuthorityStoreProtocolError("Managed execution requires the claiming actor and no explicit lease request"); + } if (typeof input.dry_run !== "boolean") { throw new AuthorityStoreProtocolError("dry_run must be a boolean"); } @@ -387,7 +395,7 @@ async function executeClaimAttempt( // Intent identity excludes observation time and current authorization facts. // A receipt proves historical acceptance, never a renewed/current lease. - const requestSha = canonicalAuthoritySha256({ + const requestIdentity = { goal_id: input.goal_id, todo_id: input.todo_id, claimed_by: input.claimed_by, @@ -398,8 +406,10 @@ async function executeClaimAttempt( ...(input.transfer_grant === undefined ? {} : {transfer_grant: input.transfer_grant}), dry_run: input.dry_run, ...(leaseRequest === null ? {} : {lease_request: leaseRequest}), + ...(input.manage_execution ? {manage_execution: true} : {}), ...(input.replan_obligation_id === undefined ? {} : {replan_obligation_id: input.replan_obligation_id}), - }); + }; + const requestSha = canonicalAuthoritySha256(requestIdentity); const receipt = new CoordinationCommandReceipt({result_schema: COORDINATION_TODO_CLAIM_RESULT_SCHEMA, identity: {schema_version: COORDINATION_TODO_CLAIM_RECEIPT_SCHEMA, operation_id: input.operation_id, goal_id: input.goal_id, request_sha256: requestSha}, @@ -414,13 +424,16 @@ async function executeClaimAttempt( result.mutation_authority === null || typeof result.mutation_authority !== "object") { throw new AuthorityStoreProtocolError("original claim result is invalid"); } + if (input.manage_execution && (result.execution_managed !== true || result.claim_operation_id !== input.operation_id)) { + throw new AuthorityStoreProtocolError("managed claim receipt lost its execution reference"); + } if (input.replan_obligation_id !== undefined && result.replan_obligation_id !== input.replan_obligation_id) { throw new AuthorityStoreProtocolError("claim replan binding differs from its original receipt"); } - if (leaseRequest !== null) { + if (leaseRequest !== null || result.execution_managed === true && result.handoff_mode === "hard_lease") { const lease = canonicalAuthorityObject(result.lease, "original claim lease"); if (lease.todo_id !== input.todo_id || lease.owner !== input.claimed_by || - lease.idempotency_key !== leaseRequest.idempotency_key) { + leaseRequest !== null && lease.idempotency_key !== leaseRequest.idempotency_key) { throw new AuthorityStoreProtocolError("original claim lease identity is invalid"); } } @@ -430,10 +443,11 @@ async function executeClaimAttempt( const currentProof = async (result: CoordinationTodoClaimResult): Promise => { if (!await authoritySourcesCurrent()) return failure(AUTHORITY_SOURCE_CHANGED.code, AUTHORITY_SOURCE_CHANGED.reason, {original_receipt: result.original_receipt}, "decision_rejection"); - if (leaseRequest !== null) { + if (leaseRequest !== null || result.execution_managed === true && result.handoff_mode === "hard_lease") { try { + const key = leaseRequest?.idempotency_key ?? String(canonicalAuthorityObject(result.lease, "managed claim lease").idempotency_key); result = await currentLeaseAcquisitionProof(store, {...input, owner: input.claimed_by, - idempotency_key: leaseRequest.idempotency_key, required_handoff_mode: "hard_lease"}, result, + idempotency_key: key, required_handoff_mode: "hard_lease"}, result, (code, reason, detail = {}) => failure(code, reason, detail, "decision_rejection")); } catch (error) { return failure("invalid_coordination_task_lease", error instanceof Error ? error.message : "invalid canonical task lease", @@ -454,11 +468,27 @@ async function executeClaimAttempt( {original_receipt: result.original_receipt}, "decision_rejection"); return result; }; + const recoverProof = async (result: CoordinationTodoClaimResult): Promise => { + if (input.manage_execution && result.reason_code === "coordination_operation_identity_mismatch") { + const original = await store.readReceipt(input.operation_id); + const legacyIntent = {...requestIdentity}; + delete legacyIntent.manage_execution; + if (original.status === "found" && original.receipts.length === 1 && + original.receipts[0]?.schema_version === COORDINATION_TODO_CLAIM_RECEIPT_SCHEMA && + original.receipts[0]?.goal_id === input.goal_id && + original.receipts[0]?.request_sha256 === canonicalAuthoritySha256(legacyIntent)) { + // Preserve pre-upgrade assignment recovery. It remains historical, not + // a new managed execution; fresh work needs a new ordinary claim. + return executeClaimAttempt(store, {...input, manage_execution: false}, authoritySourcesCurrent); + } + } + return currentProof(result); + }; const existing = await receipt.read(store); - if (existing !== null) return currentProof(existing); + if (existing !== null) return recoverProof(existing); if (!await authoritySourcesCurrent()) return failure(AUTHORITY_SOURCE_CHANGED.code, AUTHORITY_SOURCE_CHANGED.reason, {}, "decision_rejection"); const observation = await receipt.observe(store); - if (observation.kind === "receipt") return currentProof(observation.result); + if (observation.kind === "receipt") return recoverProof(observation.result); const head = observation.authority; if (head.status !== "loaded") { return { @@ -545,7 +575,12 @@ async function executeClaimAttempt( let leaseIdempotent = false; try { const currentLease = projection.leases.get(input.todo_id); - if (handoffMode === "hard_lease" && leaseRequest !== null) { + if (handoffMode === "hard_lease" && input.manage_execution === true) { + const prepared = prepareClaimExecution(head.head, input); + lease = prepared.lease; + leaseChanged = prepared.changed; + leaseIdempotent = !leaseChanged; + } else if (handoffMode === "hard_lease" && leaseRequest !== null) { // Validate required scopes before planning; a caller cannot omit a required conflict. const writeScopes = coordinationTodoWriteScopes(todo); const facts = canonicalTaskLeaseAcquireFacts(projection, input.goal_id, input.todo_id, input.registered_agents, input.now); @@ -625,7 +660,7 @@ async function executeClaimAttempt( return failure(String(acceptance.reason_code), `${String(acceptance.reason)} Inspect Goal acceptance and ask the owner to configure or rebind this Todo.`, {goal_acceptance_guard: acceptance}, "decision_rejection"); } - if (leaseRequest !== null) { + if (leaseRequest !== null || input.manage_execution) { const dependency = todoExecutionDependencyRejection(projection.todos, input.todo_id, input.now); if (dependency !== null) return failure(dependency.code, dependency.reason, {resume_condition: dependency.condition}, "decision_rejection"); @@ -647,7 +682,8 @@ async function executeClaimAttempt( handoff_mode: handoffMode, ...(changed ? {updated_at: updatedAt} : {}), mutation_authority: mutationAuthority, - ...(leaseRequest === null ? {} : { + ...(input.manage_execution ? {execution_managed: true, claim_operation_id: input.operation_id} : {}), + ...(leaseRequest === null && !(input.manage_execution && handoffMode === "hard_lease") ? {} : { todo_changed: todoChanged, lease_changed: leaseChanged, lease_idempotent: leaseIdempotent, diff --git a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts index 06c1771705..764d222a4a 100644 --- a/loopx/control_plane/coordination/todo_terminal_lifecycle.ts +++ b/loopx/control_plane/coordination/todo_terminal_lifecycle.ts @@ -1,3 +1,4 @@ +import {claimExecutionReference, claimExecutionProof} from "./claim_execution.ts"; import {HANDOFF_MODE_RETIRED_REASON} from "./handoff_mode_policy.ts"; import {planUserCompletion, requireCompletionDecisionOutcome} from "../todos/user_completion.ts"; import {AUTHORITY_SOURCE_CHANGED, uncheckedAuthoritySource, type AuthoritySourceCheck} from "./authority_source.ts"; @@ -113,6 +114,7 @@ interface CoordinationTodoTerminalLifecycleBaseInput { readonly lifecycle_grants: readonly JsonObject[]; readonly authority_reason: string | null; readonly decision_outcome: typeof DECISION_OUTCOMES[number] | null; + readonly claim_operation_id?: string; readonly lease_idempotency_key: string | null; readonly lease_expected_version: number | null; readonly allow_user_gate_auto_acquire: boolean; @@ -418,6 +420,7 @@ function normalizeTerminalInput( authority_reason: optionalString(raw.authority_reason, "authority_reason"), decision_outcome: raw.decision_outcome === null ? null : requireLiteral(raw.decision_outcome, DECISION_OUTCOMES, "decision_outcome"), + ...(raw.claim_operation_id == null ? {} : {claim_operation_id: claimExecutionReference(raw.claim_operation_id, raw.lease_idempotency_key, raw.lease_expected_version)}), lease_idempotency_key: optionalString(raw.lease_idempotency_key, "lease_idempotency_key"), lease_expected_version: optionalSafeInteger( raw.lease_expected_version, @@ -518,6 +521,7 @@ function terminalRequestSha(input: CoordinationTodoTerminalLifecycleInput): stri expected_provider_revision: input.user_update.expected_provider_revision ?? null, expected_registry_sha256: input.user_update.expected_registry_sha256 ?? null, }}), + ...(input.claim_operation_id === undefined ? {} : {claim_operation_id: input.claim_operation_id}), actor_agent_id: input.actor_agent_id, authority_reason: input.authority_reason, decision_outcome: input.decision_outcome, @@ -1250,6 +1254,10 @@ export async function executeCoordinationTodoTerminalLifecycle( } head = observation.authority; } + if (input.claim_operation_id !== undefined) { + try { input = {...input, ...await claimExecutionProof(store, head.head, input, input.claim_operation_id, input.requested_no_followup), claim_operation_id: undefined}; } + catch (error) { return terminalFailure("claim_execution_unavailable", error instanceof Error ? error.message : "Original task claim unavailable"); } + } let update: NormalizedCoordinationTodoUpdateInput | undefined; if (input.user_update !== undefined) { try { @@ -1553,7 +1561,10 @@ export async function executeCoordinationTodoTerminalLifecycle( error instanceof Error ? error.message : "Invalid original completion identity"); } if (fence.outcome === "continue" && fence.reason === "same_turn_terminal_upgrade") { - const originalInput = {...input, requested_no_followup: false, + const originalInput = {...input, + ...(normalized.claim_operation_id === undefined ? {} : {claim_operation_id: normalized.claim_operation_id, + lease_idempotency_key: normalized.lease_idempotency_key, lease_expected_version: normalized.lease_expected_version}), + requested_no_followup: false, operation_id: completionTurnOperationId(input, false)}; const original = await terminalReceipt(originalInput, terminalRequestSha(originalInput)).read(store); if (original === null) return terminalFailure("terminal_completion_receipt_required", diff --git a/loopx/control_plane/coordination/todo_update.ts b/loopx/control_plane/coordination/todo_update.ts index 89810da0c0..891376e718 100644 --- a/loopx/control_plane/coordination/todo_update.ts +++ b/loopx/control_plane/coordination/todo_update.ts @@ -1,3 +1,5 @@ +import {claimExecutionProof, maintainClaimExecution} from "./claim_execution.ts"; +import {canonicalTaskLease} from "./task_lease_state.ts"; import {AUTHORITY_SOURCE_CHANGED, uncheckedAuthoritySource, type AuthoritySourceCheck} from "./authority_source.ts"; import type { JsonObject } from "../effect_program.ts"; import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; @@ -118,6 +120,7 @@ function updateRequestSha(input: CoordinationTodoUpdateInput): string { ...(input.completion_validation_revision === undefined ? {} : {completion_validation_revision: input.completion_validation_revision}), ...(Object.keys(input.planning_intent ?? {}).length ? {planning_intent: input.planning_intent} : {}), + ...(input.claim_operation_id === undefined ? {} : {claim_operation_id: input.claim_operation_id}), // Preserve receipt identity for pre-proof requests already persisted in v0. ...(input.lease_idempotency_key != null || input.lease_expected_version != null ? { lease_idempotency_key: input.lease_idempotency_key, @@ -162,6 +165,7 @@ export async function executeCoordinationTodoUpdate( command: "complete", actor_agent_id: input.actor_agent_id, registered_agents: input.registered_agents, lifecycle_grants: input.lifecycle_grants ?? [], authority_reason: input.authority_reason ?? null, decision_outcome: null, operation_identity: {kind: "explicit", operation_id: input.operation_id}, + ...(input.claim_operation_id === undefined ? {} : {claim_operation_id: input.claim_operation_id}), lease_idempotency_key: input.lease_idempotency_key ?? null, lease_expected_version: input.lease_expected_version ?? null, allow_user_gate_auto_acquire: input.lease_idempotency_key == null && input.lease_expected_version == null, requested_no_followup: intent.no_followup === true, requested_completion_turn_key: null, requested_completion_identity_source: null, @@ -214,6 +218,11 @@ export async function executeCoordinationTodoUpdate( const target = loadUpdateTarget(head.head, input); if (isFailure(target)) return target; + const claimReference = input.claim_operation_id; + if (claimReference !== undefined) { + try { input = {...input, ...await claimExecutionProof(store, head.head, input, claimReference), claim_operation_id: undefined}; } + catch (error) { return rejection("claim_execution_unavailable", error instanceof Error ? error.message : "Original task claim unavailable"); } + } const rejected = todoUpdateAdmissionRejection(head.head, target.todo, target.leases, input); if (rejected !== null) return {...rejection(rejected.code, rejected.reason), ...(rejected.handoff_mode === undefined ? {} : {handoff_mode: rejected.handoff_mode}), @@ -299,10 +308,21 @@ export async function executeCoordinationTodoUpdate( const workRequirementsTransition = workRequirements?.transition == null ? {} : { work_requirements_transition: workRequirements.transition, }; + let maintainedLease: JsonObject | null = null; + if (claimReference !== undefined && target.todo.status === "open" && next.status === "open" && + input.lease_idempotency_key != null && + ![...cycle.mutations, ...(deferredCycle?.mutations ?? []), ...(blockedCycle?.mutations ?? []), + ...(workRequirements?.mutations ?? [])].some(m => m.kind === "lease_upsert")) { + try { maintainedLease = maintainClaimExecution(head.head, input, canonicalTaskLease(target.leases.get(input.todo_id)!, input.goal_id, input.todo_id)); } + catch (error) { return rejection("claim_execution_unavailable", error instanceof Error ? error.message : "Task execution maintenance rejected"); } + } + const todoChanged = changed; + changed ||= maintainedLease !== null; const commit: AuthorityStoreCommit = changed ? prepareCoordinationProjectionCommit({ goal_id: input.goal_id, operation_id: input.operation_id, expected_provider_revision: head.provider_revision, projection: head.head, - mutations: [{kind: "todo_upsert", todo: next, clear_fields: clearFields}, + mutations: [...(todoChanged ? [{kind: "todo_upsert" as const, todo: next, clear_fields: clearFields}] : []), + ...(maintainedLease ? [{kind: "lease_upsert" as const, lease: maintainedLease}] : []), ...cycle.mutations, ...(deferredCycle?.mutations ?? []), ...(blockedCycle?.mutations ?? []), ...(workRequirements?.mutations ?? [])], }) : {operation_id: input.operation_id, diff --git a/loopx/control_plane/coordination/todo_update_intent.ts b/loopx/control_plane/coordination/todo_update_intent.ts index a822c07fab..5000acc0ba 100644 --- a/loopx/control_plane/coordination/todo_update_intent.ts +++ b/loopx/control_plane/coordination/todo_update_intent.ts @@ -1,3 +1,4 @@ +import {claimExecutionReference} from "./claim_execution.ts"; /** Canonical edit decoding and materialization shared by ordinary updates and * User completion. Admission, validation effects and commits stay with callers. */ import {decodeTaskLeaseProof} from "./task_lease_proof.ts"; @@ -36,6 +37,7 @@ export interface CoordinationTodoUpdateInput { readonly clear_fields: readonly string[]; readonly dry_run: boolean; readonly now: Date; + readonly claim_operation_id?: string; readonly lease_idempotency_key?: string | null; readonly lease_expected_version?: number | null; readonly planning_intent?: JsonObject; @@ -88,6 +90,8 @@ export function normalizeTodoUpdateEnvelope( if (raw.completion !== undefined && planningIntent.status !== "done") { throw new AuthorityStoreProtocolError("Completion payload requires status=done"); } + claimExecutionReference(raw.claim_operation_id, raw.lease_idempotency_key, raw.lease_expected_version); + if (raw.claim_operation_id != null && raw.actor_agent_id === null) throw new AuthorityStoreProtocolError("A task claim reference requires its actor"); const key = raw.lease_idempotency_key ?? null; const version = raw.lease_expected_version ?? null; if (key !== null && (typeof key !== "string" || !key.trim() || key !== key.trim())) { diff --git a/loopx/control_plane/todos/markdown.py b/loopx/control_plane/todos/markdown.py index 8b0ed115a7..5702c9cd4d 100644 --- a/loopx/control_plane/todos/markdown.py +++ b/loopx/control_plane/todos/markdown.py @@ -309,6 +309,10 @@ def render_todo_markdown(payload: dict[str, Any]) -> str: if payload.get("instruction"): lines.extend(["", str(payload["instruction"])]) lines.extend(_render_settlement_plan(payload["settlement_plan"])) + if not payload.get("error") and payload.get("execution_managed"): + lines.extend(["", f"- claim_operation_id: `{payload['claim_operation_id']}`"]) + for command in payload.get("next_cli_actions", []): + lines.extend(["", "```sh", command, "```"]) correctness = payload.get("local_state_write_correctness") if isinstance(correctness, dict): intent = correctness.get("write_intent") if isinstance(correctness.get("write_intent"), dict) else {} diff --git a/loopx/control_plane/todos/mutation_api.py b/loopx/control_plane/todos/mutation_api.py index 5a56c05ff7..af75eccbd2 100644 --- a/loopx/control_plane/todos/mutation_api.py +++ b/loopx/control_plane/todos/mutation_api.py @@ -303,6 +303,7 @@ def update_goal_todo( clear_claim: bool = False, claim_only: bool = False, claim_operation_id: str | None = None, + manage_execution: bool = False, update_operation_id: str | None = None, update_expected_provider_revision: str | None = None, update_expected_registry_sha256: str | None = None, @@ -337,9 +338,6 @@ def update_goal_todo( raise ValueError( "todo update accepts either resume_when or clear_resume_when, not both" ) - if claim_operation_id is not None: - if not claim_only: - raise ValueError("claim_operation_id is supported only by todo claim") if task_lease_expected_version is not None and task_lease_idempotency_key is None: raise ValueError( "--task-lease-expected-version requires --task-lease-idempotency-key" @@ -379,6 +377,7 @@ def update_goal_todo( actor_agent_id=agent_id, dry_run=dry_run, operation_id=claim_operation_id, + manage_execution=manage_execution, task_lease_idempotency_key=task_lease_idempotency_key, task_lease_expected_version=task_lease_expected_version, project=project, @@ -423,6 +422,7 @@ def update_goal_todo( actor_agent_id=agent_id, role=role, text=text, note=note, dry_run=dry_run, project=project, state_file=state_file, operation_id=update_operation_id, + claim_operation_id=claim_operation_id, authority_reason=authority_reason, expected_provider_revision=update_expected_provider_revision, expected_registry_sha256=update_expected_registry_sha256, @@ -455,6 +455,7 @@ def complete_goal_todo( completion_validation_workspace_path: Path | None = None, task_lease_idempotency_key: str | None = None, task_lease_expected_version: int | None = None, + claim_operation_id: str | None = None, note: str | None = None, no_followup: bool = False, successor_todo_ids: list[str] | None = None, diff --git a/loopx/control_plane/todos/provider_terminal_lifecycle.py b/loopx/control_plane/todos/provider_terminal_lifecycle.py index eff337a6fb..1b8c7ff7a9 100644 --- a/loopx/control_plane/todos/provider_terminal_lifecycle.py +++ b/loopx/control_plane/todos/provider_terminal_lifecycle.py @@ -134,6 +134,7 @@ def route_terminal_call(command: str, call: Mapping[str, Any]) -> dict[str, Any] completion_validation_workspace_path=( call.get("completion_validation_workspace_path") if complete else None ), + claim_operation_id=call.get("claim_operation_id"), task_lease_idempotency_key=call.get("task_lease_idempotency_key"), task_lease_expected_version=_non_negative_integer( call.get("task_lease_expected_version"), @@ -259,6 +260,7 @@ def terminal_canonical_todo_if_promoted( completion_identity_source: str | None, completion_delivery_workspace: Mapping[str, Any] | None, completion_validation_workspace_path: Path | None, + claim_operation_id: str | None, task_lease_idempotency_key: str | None, task_lease_expected_version: int | None, no_followup: bool, @@ -362,7 +364,8 @@ def terminal_canonical_todo_if_promoted( and target.get("task_class") == "continuous_monitor" ) request = { - "schema_version": _TERMINAL_REQUEST_SCHEMA, + "schema_version": ("loopx_local_coordination_todo_terminal_lifecycle_request_v4" + if claim_operation_id is not None else _TERMINAL_REQUEST_SCHEMA), **({"review_basis": dict(review_basis)} if review_basis is not None else {}), **({"expected_todo_text_sha256": expected_todo_text_sha256} if expected_todo_text_sha256 is not None or require_captured_todo_text else {}), @@ -389,6 +392,7 @@ def terminal_canonical_todo_if_promoted( completion_turn_key=completion_turn_key, )} ), + **({"claim_operation_id": claim_operation_id} if claim_operation_id is not None else {}), "lease_idempotency_key": task_lease_idempotency_key, "lease_expected_version": task_lease_expected_version, "allow_user_gate_auto_acquire": command == "complete", diff --git a/loopx/control_plane/todos/provider_update.py b/loopx/control_plane/todos/provider_update.py index 04e121e04a..2fe7943858 100644 --- a/loopx/control_plane/todos/provider_update.py +++ b/loopx/control_plane/todos/provider_update.py @@ -234,6 +234,7 @@ def update_canonical_todo_if_promoted( actor_agent_id: str | None, role: str | None, text: str | None, note: str | None, dry_run: bool, project: Path | None = None, state_file: Path | None = None, + claim_operation_id: str | None = None, operation_id: str | None = None, task_lease_idempotency_key: str | None = None, task_lease_expected_version: int | None = None, planning_intent: dict[str, Any] | None = None, @@ -257,6 +258,9 @@ def update_canonical_todo_if_promoted( normalized_note = compact_todo_text(note) if normalized_note: patch["note"] = normalized_note + if claim_operation_id is not None and (completion_validation_revision is not None or monitor_observation is not None + or str((planning_intent or {}).get("status", "")).lower() == "done"): + raise ValueError("Claim execution progress does not change validators or observe Monitors; use todo complete for completion") completion = None validation_revision = None if completion_validation_revision is not None: @@ -282,7 +286,8 @@ def update_canonical_todo_if_promoted( persist_if_resolved=not dry_run) completion = {"validation_declaration": declaration} request = { - "schema_version": ("loopx_local_coordination_todo_update_request_v5" if validation_revision is not None + "schema_version": ("loopx_local_coordination_todo_update_request_v6" if claim_operation_id is not None + else "loopx_local_coordination_todo_update_request_v5" if validation_revision is not None else "loopx_local_coordination_todo_update_request_v4" if monitor_observation is not None else "loopx_local_coordination_todo_update_request_v3" if completion is not None else "loopx_local_coordination_todo_update_request_v2"), @@ -296,6 +301,7 @@ def update_canonical_todo_if_promoted( "operation_id": (operation_id if operation_id is not None else (monitor_observation.monitor_effect_id if monitor_observation else None) or f"todo-update:{uuid4().hex}"), + **({"claim_operation_id": claim_operation_id} if claim_operation_id is not None else {}), "lease_idempotency_key": task_lease_idempotency_key, "lease_expected_version": task_lease_expected_version, "patch": patch, "clear_fields": [], "dry_run": dry_run, diff --git a/tests/control_plane/test_sqlite_authority_cli.py b/tests/control_plane/test_sqlite_authority_cli.py index 79af170859..c7288d7c37 100644 --- a/tests/control_plane/test_sqlite_authority_cli.py +++ b/tests/control_plane/test_sqlite_authority_cli.py @@ -348,12 +348,9 @@ def listed(*archived_ids): todo = command(*add)["todo_id"] claim = ["todo", "claim", "--goal-id", goal, "--todo-id", todo, "--agent-id", agent, "--claimed-by", agent, "--claim-operation-id", "ordinary-claim"] - assert command(*claim, expected_code=1)["error_code"] == "handoff_mode_requires_lease" - lease = command("task-lease", "acquire", "--goal-id", goal, "--todo-id", todo, - "--owner", agent, "--idempotency-key", "ordinary-lease") - proof = ["--task-lease-idempotency-key", "ordinary-lease", - "--task-lease-expected-version", str(lease["lease"]["version"])] - command(*claim, *proof) + accepted = command(*claim) + assert accepted["execution_managed"] is True + proof = ["--claim-operation-id", accepted["claim_operation_id"]] guard = ["quota", "should-run", "--goal-id", goal, "--agent-id", agent, "--todo-id", todo, "--codex-app", "--turn-instance-id", turn, "--available-capability", "filesystem_read", ] @@ -373,6 +370,7 @@ def listed(*archived_ids): "--agent-id", agent, "--turn-instance-id", turn, "--evidence", "synthetic:ordinary-lifecycle-validated", *proof] command(*complete) + command(*complete, "--no-follow-up") # Same original managed claim; no copied released proof. # Replanning is a normal caller, too: replace another task through its owner, # then retain both the archived predecessor and the new successor on exit. obsolete = command("todo", "add", "--goal-id", goal, "--role", "agent", "--claimed-by", agent, diff --git a/tests/control_plane/test_todo_claim_argument_recovery.py b/tests/control_plane/test_todo_claim_argument_recovery.py index 85f87be862..08d0f8125c 100644 --- a/tests/control_plane/test_todo_claim_argument_recovery.py +++ b/tests/control_plane/test_todo_claim_argument_recovery.py @@ -2,6 +2,7 @@ from __future__ import annotations import json +import shlex import subprocess import sys from pathlib import Path @@ -158,15 +159,14 @@ def test_lease_input_requirement_preserves_zero_cas(claim_cli): assert applied["lease"]["version"] == 1 -def test_grammar_packet_does_not_guess_authority_mode(claim_cli): +def test_grammar_packet_does_not_guess_execution_actor(claim_cli): _, run, snapshot, prefix, claim, _, _, _ = claim_cli before = snapshot() - flags = ["--claimed-by", "peer-a"] + flags = ["--claimed-by", "peer-b"] rejected = run([*prefix, *claim, *flags], expected_exit=1) assert rejected.get("error_code") != "todo_claim_invalid_arguments" assert snapshot() == before - assert rejected["error_code"] == "handoff_mode_requires_lease" - assert rejected["recovery"]["requires_flags"] == ["--task-lease-idempotency-key"] + assert rejected["error_code"] == "claim_actor_mismatch" def test_markdown_exposes_the_same_required_input(): @@ -182,3 +182,31 @@ def test_markdown_exposes_the_same_required_input(): rendered = render_todo_markdown(payload) assert "requires_flags: `--claimed-by`" in rendered assert json.dumps(payload["recovery"]["cli_args"]) in rendered + + +def test_ordinary_claim_progress_completion_need_no_lease_flags(claim_cli): + _, run, snapshot, prefix, claim, _, _, _ = claim_cli + arguments = [*prefix, *claim, "--claimed-by", "peer-a", "--claim-operation-id", "managed-claim"] + accepted = run(arguments) + assert accepted["ok"] is True + assert accepted["execution_managed"] is True + assert "lease" not in accepted and "original_receipt" not in accepted + from loopx.control_plane.todos.markdown import render_todo_markdown + rendered = render_todo_markdown(accepted) + assert all(action in rendered for action in accepted["next_cli_actions"]) + assert accepted["claim_operation_id"] == "managed-claim" + assert len(accepted["next_cli_actions"]) == 2 + assert all("--claim-operation-id managed-claim" in action for action in accepted["next_cli_actions"]) + assert all("task-lease" not in action for action in accepted["next_cli_actions"]) + common = ["--goal-id", GOAL, "--todo-id", TODO, "--agent-id", "peer-a", "--claim-operation-id", "managed-claim"] + progress_args = shlex.split(accepted["next_cli_actions"][0])[1:] + progress_args = ["managed-progress" if part == "" else "Validation passed" if part == "" else part for part in progress_args] + progress = run(progress_args) + assert progress["ok"] is True + complete_args = ["evidence:validated-output" if part == "" else part for part in shlex.split(accepted["next_cli_actions"][1])[1:]] + completed = run(complete_args) + assert completed["ok"] is True + before = snapshot() + replay = run([*prefix, "todo", "complete", *common, "--evidence", "evidence:validated-output"]) + assert replay["ok"] is True + assert snapshot() == before diff --git a/tests/control_plane_ts/authority_store_conformance.ts b/tests/control_plane_ts/authority_store_conformance.ts index 3138e7b1eb..b5a4392686 100644 --- a/tests/control_plane_ts/authority_store_conformance.ts +++ b/tests/control_plane_ts/authority_store_conformance.ts @@ -1,3 +1,4 @@ +import {registerManagedClaimExecutionConformance} from "./managed_claim_execution_conformance.ts"; import {registerClaimContentionConformance} from "./claim_contention_conformance.ts"; import {registerMonitorGateScopeConformance} from "./monitor_gate_scope_conformance.ts"; import {projectCoordinationSource, SOURCE_PROJECTION_REQUEST_SCHEMA} from "../../loopx/control_plane/coordination/source_projection.ts"; @@ -393,6 +394,7 @@ export function registerAuthorityStoreConformance( } }); registerCommandObservationConformance(providerName, factory); + registerManagedClaimExecutionConformance(providerName, factory); registerClaimAcquisitionProofConformance(providerName, factory); test(`${providerName} imported legacy policy requires upgrade before new execution`, async t => { const {store} = await factory(t); diff --git a/tests/control_plane_ts/local_authority_runtime.test.ts b/tests/control_plane_ts/local_authority_runtime.test.ts index ca393c328c..4370c0ada3 100644 --- a/tests/control_plane_ts/local_authority_runtime.test.ts +++ b/tests/control_plane_ts/local_authority_runtime.test.ts @@ -1836,3 +1836,21 @@ test("an engagement fence does not authorize retired legacy task-lease writers", backup_required: true, }); }); + +test("managed execution fields cannot be ignored by older public transports", async () => { + const runtime = await import("../../loopx/control_plane/coordination/local_authority_runtime.ts"); + const noStore = {createStore: () => {throw new Error("must reject before opening a provider");}}; + for (const v of [0, 1]) { + const result = await runtime.claimLocalCoordinationTodo({schema_version: `loopx_local_coordination_todo_claim_request_v${v}`, + manage_execution: true}, noStore); + assert.match(String(result.reason), /requires claim request v2/); + } + for (const v of [0, 1, 2, 3, 4, 5]) { + const result = await runtime.updateLocalCoordinationTodo({schema_version: `loopx_local_coordination_todo_update_request_v${v}`, + claim_operation_id: "original-claim"}, noStore); + assert.match(String(result.reason), /requires update request v6/); + } + const terminal = await runtime.terminalLifecycleLocalCoordinationTodo({ + schema_version: "loopx_local_coordination_todo_terminal_lifecycle_request_v3", claim_operation_id: "original-claim"}, noStore); + assert.match(String(terminal.reason), /requires terminal request v4/); +}); diff --git a/tests/control_plane_ts/managed_claim_execution_conformance.ts b/tests/control_plane_ts/managed_claim_execution_conformance.ts new file mode 100644 index 0000000000..e871c1b619 --- /dev/null +++ b/tests/control_plane_ts/managed_claim_execution_conformance.ts @@ -0,0 +1,182 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import type {AuthorityStore} from "../../loopx/control_plane/coordination/authority_store.ts"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; +import type {AuthorityStoreConformanceFactory} from "./authority_store_conformance.ts"; +import {executeCoordinationTodoClaim as claim} from "../../loopx/control_plane/coordination/todo_claim.ts"; +import {executeCoordinationTodoTerminalLifecycle as complete} from "../../loopx/control_plane/coordination/todo_terminal_lifecycle.ts"; +import {executeCoordinationTodoUpdate as update} from "../../loopx/control_plane/coordination/todo_update.ts"; +import {executeCanonicalTaskLeaseLifecycle as maintain} from "../../loopx/control_plane/coordination/task_lease_lifecycle.ts"; +import {prepareCoordinationProjectionCommit} from "../../loopx/control_plane/coordination/coordination_projection.ts"; +import {authorityProjectionFixture} from "./authority_projection_fixture.ts"; + +async function loaded(store: AuthorityStore) { + const value = await store.loadAuthority(); + if (value.status !== "loaded") throw new Error("synthetic authority unavailable"); + return value; +} +export function registerManagedClaimExecutionConformance(provider: string, factory: AuthorityStoreConformanceFactory) { + async function setup(t: test.TestContext, mode = "hard_lease") { + const {store} = await factory(t); + const projection = authorityProjectionFixture("goal-a", [{todo_id: "todo_managed", role: "agent", + status: "open", done: false, archive_state: "active", text: "Deliver validated output", + task_class: "advancement_task", required_write_scopes: ["src/**"]}], [], "native", {handoff_mode: mode}); + assert.equal((await store.commitAuthority({operation_id: "seed", expected_provider_revision: null, + next_projection: projection, events: [], receipts: []})).status, "applied"); + const now = new Date("2026-09-13T10:00:00Z"); + const request = {goal_id: "goal-a", todo_id: "todo_managed", claimed_by: "agent-a", actor_agent_id: "agent-a", + expected_role: "agent", registered_agents: ["agent-a", "agent-b"], operation_id: "claim-a", + manage_execution: true, dry_run: false, now}; + const progress = {goal_id: request.goal_id, todo_id: request.todo_id, actor_agent_id: "agent-a", + expected_role: "agent", registered_agents: request.registered_agents, operation_id: "progress-a", + patch: {note: "Validated progress"}, clear_fields: [], dry_run: false, now, + claim_operation_id: request.operation_id}; + return {store, request, progress}; + } + test(`${provider} managed execution: ordinary claim/progress carry one execution, renew only near expiry, recover without writes`, async t => { + const {store, request, progress} = await setup(t); + const before = await loaded(store); + const preview = await claim(store, {...request, dry_run: true}); + assert.equal(preview.status, "planned", JSON.stringify(preview)); + assert.deepEqual(await loaded(store), before); + const accepted = await claim(store, request); + assert.equal(accepted.status, "applied", JSON.stringify(accepted)); + const original = accepted.lease as JsonObject; + assert.deepEqual(original.write_scopes, ["src/**"]); + assert.equal(original.expires_at, "2026-09-13T10:45:00Z"); + assert.equal(accepted.claim_operation_id, "claim-a"); + assert.equal((await update(store, progress)).status, "applied"); + let current = await loaded(store); + assert.deepEqual((current.head.leases as JsonObject[])[0], original, "early activity never renews"); + const renewed = await update(store, {...progress, operation_id: "progress-near-expiry", + patch: {note: "New validated evidence"}, now: new Date("2026-09-13T10:31:00Z")}); + assert.equal(renewed.status, "applied", JSON.stringify(renewed)); + current = await loaded(store); + const lease = (current.head.leases as JsonObject[])[0]!; + assert.equal(lease.version, 2); assert.equal(lease.lease_epoch, original.lease_epoch); + assert.equal(lease.idempotency_key, original.idempotency_key); + assert.equal(lease.expires_at, "2026-09-13T11:16:00Z"); + assert.equal((current.head.todos as JsonObject[])[0]!.note, "New validated evidence"); + assert.equal((await update(store, {...progress, operation_id: "progress-near-expiry", + patch: {note: "New validated evidence"}, now: new Date("2026-09-13T10:55:00Z")})).status, "replayed"); + assert.deepEqual(await loaded(store), current, "lost response recovery cannot renew again"); + assert.equal((await claim(store, {...request, now: new Date("2026-09-13T10:55:00Z")})).status, "replayed"); + assert.deepEqual(await loaded(store), current, "claim replay is read-only"); + }); + test(`${provider} managed execution: missing/foreign/expired/replaced execution references cannot write or reacquire`, async t => { + const {store, request, progress} = await setup(t); + const accepted = await claim(store, request); + const lease = accepted.lease as JsonObject; + const before = await loaded(store); + for (const change of [{claim_operation_id: "missing"}, {actor_agent_id: "agent-b"}, + {now: new Date("2026-09-13T10:45:00Z")}, + {lease_idempotency_key: lease.idempotency_key as string, lease_expected_version: 1}]) { + assert.equal((await update(store, {...progress, ...change})).status, "failed"); + assert.deepEqual(await loaded(store), before); + } + const released = await maintain(store, {operation: "release", goal_id: request.goal_id, todo_id: request.todo_id, + owner: "agent-a", idempotency_key: String(lease.idempotency_key), expected_version: 1, + ttl_seconds: null, registered_agents: request.registered_agents, now: request.now}); + assert.equal(released.status, "applied", JSON.stringify(released)); + const replacement = await claim(store, {...request, operation_id: "new-execution"}); + assert.equal(replacement.status, "applied", JSON.stringify(replacement)); + const replaced = await loaded(store); + assert.equal((await update(store, progress)).reason_code, "claim_execution_unavailable"); + assert.deepEqual(await loaded(store), replaced); + assert.equal((await claim(store, request)).status, "failed", "old claim never revives retired execution"); + }); + test(`${provider} managed execution: claim conflict, exclusion, policy loss and provider CAS preserve the original task`, async t => { + const {store, request, progress} = await setup(t); + assert.equal((await claim(store, request)).status, "applied"); + const before = await loaded(store); + assert.equal((await claim(store, {...request, actor_agent_id: "agent-b", claimed_by: "agent-b", + operation_id: "foreign"})).status, "failed"); + assert.deepEqual(await loaded(store), before); + const racing = new Proxy(store, {get(target, key) { + if (key === "commitAuthority") return async () => ({status: "conflict", conflict_kind: "provider_revision_mismatch"}); + const value = Reflect.get(target, key, target); + return typeof value === "function" ? value.bind(target) : value; + }}); + assert.equal((await update(racing, {...progress, now: new Date("2026-09-13T10:31:00Z")})).status, "conflict"); + assert.deepEqual(await loaded(store), before, "progress and renewal are one CAS"); + const todo = (before.head.todos as JsonObject[])[0]!; + assert.equal((await store.commitAuthority(prepareCoordinationProjectionCommit({goal_id: request.goal_id, + operation_id: "exclude", expected_provider_revision: before.provider_revision, projection: before.head, + mutations: [{kind: "todo_upsert", todo: {...todo, excluded_agents: ["agent-a"]}}]}))).status, "applied"); + const excluded = await loaded(store); + assert.equal((await update(store, progress)).status, "failed"); + assert.deepEqual(await loaded(store), excluded); + }); + test(`${provider} managed execution: unavailable receipt and changed policy reject without maintenance`, async t => { + const {store, request, progress} = await setup(t); + assert.equal((await claim(store, request)).status, "applied"); + const before = await loaded(store); + const unavailable = new Proxy(store, {get(target, key) { + if (key === "readReceipt") return async (id: string) => id === request.operation_id + ? {status: "unavailable", reason_code: "offline", reason: "synthetic transport failure"} + : target.readReceipt(id); + const value = Reflect.get(target, key, target); + return typeof value === "function" ? value.bind(target) : value; + }}); + assert.equal((await update(unavailable, progress)).status, "failed"); + assert.deepEqual(await loaded(store), before); + assert.equal((await update(store, progress, async () => false)).status, "failed"); + assert.deepEqual(await loaded(store), before); + assert.equal((await update(store, progress)).status, "applied", "normal recovery retains the same original operation"); + const recovered = await loaded(store); + assert.equal((await store.commitAuthority({operation_id: "policy-change", expected_provider_revision: recovered.provider_revision, + next_projection: {...recovered.head, handoff_mode: "soft_claim"}, events: [], receipts: []})).status, "applied"); + const changed = await loaded(store); + assert.equal((await update(store, {...progress, operation_id: "post-policy-change"})).reason_code, "claim_execution_unavailable"); + assert.deepEqual(await loaded(store), changed); + }); + test(`${provider} managed execution: pre-upgrade plain claim recovery never changes its intent`, async t => { + const {store, request} = await setup(t); + assert.equal((await claim(store, request)).status, "applied"); + const legacy = {...request, operation_id: "pre-upgrade-claim", manage_execution: false}; + assert.equal((await claim(store, legacy)).status, "no_change"); + const before = await loaded(store); + const recovered = await claim(store, {...legacy, manage_execution: true}); + assert.equal(recovered.status, "replayed", JSON.stringify(recovered)); + assert.equal(recovered.execution_managed, undefined, "old assignment is not upgraded into execution proof"); + assert.deepEqual(await loaded(store), before); + assert.equal((await claim(store, {...legacy, manage_execution: true, expected_role: null})).reason_code, + "coordination_operation_identity_mismatch", "other changed intent remains rejected"); + assert.deepEqual(await loaded(store), before); + }); + test(`${provider} managed execution: completion releases once and the original Turn closes without new execution`, async t => { + const {store, request} = await setup(t); + assert.equal((await claim(store, request)).status, "applied"); + const terminal = {goal_id: request.goal_id, todo_id: request.todo_id, actor_agent_id: "agent-a", + expected_role: "agent" as const, registered_agents: request.registered_agents, command: "complete" as const, + operation_identity: {kind: "completion_turn" as const}, claim_operation_id: request.operation_id, + lifecycle_grants: [], authority_reason: null, decision_outcome: null, + lease_idempotency_key: null, lease_expected_version: null, allow_user_gate_auto_acquire: false, + requested_no_followup: false, requested_completion_turn_key: "original-turn", + requested_completion_identity_source: "turn_settlement" as const, linked_successor_todo_ids: [], + successor_intents: [], note: null, evidence: "evidence:validated-output", reason: null, + clear_claim: false, validation_declaration: null, validation_receipt: null, + completion_policy_request: null, dry_run: false, now: request.now}; + const done = await complete(store, terminal); + assert.equal(done.status, "applied", JSON.stringify(done)); + const completed = await loaded(store); + assert.equal((completed.head.leases as JsonObject[])[0]!.status, "released"); + const final = await complete(store, {...terminal, requested_no_followup: true}); + assert.equal(final.status, "applied", JSON.stringify(final)); + const closed = await loaded(store); + assert.equal((closed.head.todos as JsonObject[])[0]!.no_followup, true); + assert.equal((closed.head.leases as JsonObject[])[0]!.version, 1); + assert.equal((await complete(store, {...terminal, requested_no_followup: true, + now: new Date("2030-01-01T00:00:00Z")})).status, "replayed"); + assert.deepEqual(await loaded(store), closed); + assert.equal((await complete(store, {...terminal, + operation_identity: {kind: "explicit", operation_id: "unrelated-new-completion"}})).status, "failed"); + assert.deepEqual(await loaded(store), closed, "released proof cannot authorize new work"); + }); + test(`${provider} managed execution: soft claims do not create or maintain leases`, async t => { + const {store, request, progress} = await setup(t, "soft_claim"); + assert.equal((await claim(store, request)).status, "applied"); + assert.equal((await update(store, progress)).status, "applied"); + assert.deepEqual((await loaded(store)).head.leases, []); + }); +}