From 1ecfcec3db5bb3716f55e4b6128c4727317c7b19 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Fri, 2 Oct 2026 20:46:30 +0800 Subject: [PATCH 1/3] fix(delegation): renew original lease before final acceptance Signed-off-by: huangruiteng --- loopx/collaboration_mcp.py | 22 ++++ tests/test_delegation_lease_lifetime.py | 149 +++++++++++++++++++++++- 2 files changed, 170 insertions(+), 1 deletion(-) diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py index 87b98c6086..4ab5b658dd 100644 --- a/loopx/collaboration_mcp.py +++ b/loopx/collaboration_mcp.py @@ -1167,6 +1167,28 @@ def _complete_delegated_todo(self, row: dict, binding: dict) -> None: # key/epoch; it cannot reacquire an expired execution. Renewal has # changed its version, so the historical acquisition is not CAS. if "completion_lease_version" not in row: + # The Host supervisor has stopped. Renew the original execution + # before validation captures its provider revision; renewing + # during validation would invalidate that source witness. The + # canonical TS lease owner decides admission and replay. This + # adapter journals one intent, not a new lease or a longer TTL. + if "completion_lease_renewal_version" not in row: + proof = self._cli(binding, *self._delegation_claim_arguments(row, binding)) + if proof.get("ok") is not True: + raise ValueError("delegation current execution proof lost before completion") + row["completion_lease_renewal_version"] = proof["lease"]["version"] + _write(self.path(row["identity"]["operation_id"]), row) + renewed = self._cli( + binding, "task-lease", "renew", "--goal-id", self.goal_id, + "--todo-id", binding["todo_id"], "--owner", binding["agent_id"], + "--idempotency-key", lease["lease"]["idempotency_key"], + "--expected-version", str(row["completion_lease_renewal_version"]), + "--ttl-seconds", str(lease["lease"]["acquire_ttl_seconds"]), + ) + if renewed.get("ok") is not True: + raise ValueError("delegation original lease renewal rejected before completion") + # A renewal receipt can be historical after a lost reply. Read + # current authority before freezing the terminal intent below. proof = self._cli(binding, *self._delegation_claim_arguments(row, binding)) current = proof.get("lease", {}) if (proof.get("ok") is not True or current.get("owner") != binding["agent_id"] diff --git a/tests/test_delegation_lease_lifetime.py b/tests/test_delegation_lease_lifetime.py index 76b5dd8ee0..e26e5e3f5c 100644 --- a/tests/test_delegation_lease_lifetime.py +++ b/tests/test_delegation_lease_lifetime.py @@ -13,7 +13,7 @@ import pytest -from test_local_delegation import HOST, brief, service # noqa: F401 +from test_local_delegation import HOST, brief, demo, service # noqa: F401 from loopx.control_plane.collaboration.inbox import _read from loopx.control_plane.coordination.local_authority import read_canonical_todos_if_promoted from tests.control_plane.host_process_fixture import COUNTER_PROCESS_SOURCE @@ -54,6 +54,153 @@ def inspect(runner): "--todo-id", "todo_analyst-initial") +@pytest.fixture(params=["file", "sqlite"]) +def completion_service(tmp_path, request, monkeypatch): + """Pin a real slow final acceptance command before preparing authority.""" + validator = tmp_path / "completion-validator.py" + validator.write_text('''import sys, time, runpy +from pathlib import Path +root = Path(sys.argv[1]) +counter = root / 'validator-calls' +calls = int(counter.read_text()) + 1 if counter.exists() else 1 +counter.write_text(str(calls)) +if calls == 2: + time.sleep(8) + (root / 'completion-validation-ended').touch() +actual = root / 'project/validation/acceptance.py' +sys.path.insert(0, str(actual.parent)) +sys.argv[0] = str(actual) +runpy.run_path(str(actual), run_name='__main__') +''') + write = demo.write + + def configure(path, value): + if path.name == "bootstrap.json": + for criterion in value["document"]["criteria"]: + criterion["validation_timeout_seconds"] = 1 + if criterion["id"] == "analyst-initial": + criterion["validation_timeout_seconds"] = 15 + criterion["validation_argv"][1] = str(validator) + return write(path, value) + + monkeypatch.setattr(demo, "write", configure) + return service.__wrapped__(tmp_path, request, monkeypatch) + + +@pytest.mark.parametrize("lost_reply", [None, "renewal", "completion"]) +def test_completion_renews_before_validation_and_replays_each_intent(completion_service, monkeypatch, lost_reply): + root, runner = completion_service + original = prepare_lease(root, runner, monkeypatch) + complete, cli = runner._complete_delegated_todo, runner._cli + shortened = None + renewal_calls, completion_calls = [], [] + dropped = False + + def enter_completion(row, binding): + nonlocal shortened + if shortened is None: + current = inspect(runner)["lease"] + # Fix the phase boundary, independent of whether the Host happened + # to cross its earlier renewal timer. Use the real canonical API. + shortened = cli(binding, "task-lease", "renew", "--goal-id", runner.goal_id, + "--todo-id", binding["todo_id"], "--owner", binding["agent_id"], + "--idempotency-key", current["idempotency_key"], + "--expected-version", str(current["version"]), "--ttl-seconds", "6")["lease"] + return complete(row, binding) + + def observe_reply(binding, *args, **kwargs): + nonlocal dropped + result = cli(binding, *args, **kwargs) + phase = None + if args[:2] == ("task-lease", "renew"): + renewal_calls.append((args, result)) + phase = "renewal" + elif args[:2] == ("todo", "complete"): + completion_calls.append((args, result)) + phase = "completion" + if phase == lost_reply and phase is not None and not dropped: + dropped = True + raise ValueError("fixture dropped the committed " + phase + " reply") + return result + + monkeypatch.setattr(runner, "_complete_delegated_todo", enter_completion) + monkeypatch.setattr(runner, "_cli", observe_reply) + runner.execute("lease-lifetime") + if lost_reply: + uncertain = _read(runner.path("lease-lifetime")) + assert uncertain["status"] == "turn_returned", uncertain + assert "fixture dropped" in uncertain["error"] + assert uncertain["completion_lease_renewal_version"] == shortened["version"] + assert ("completion_lease_version" in uncertain) == (lost_reply == "completion") + runner.execute("lease-lifetime") + row = _read(runner.path("lease-lifetime")) + assert row["status"] == "accepted", row + assert row["turn_result"]["result_kind"] == "validated_progress" + assert (root / "completion-validation-ended").exists() + assert time.time() > datetime.fromisoformat(shortened["expires_at"].replace("Z", "+00:00")).timestamp() + assert (root / "analyst/initial/host-invocations").read_text() == "1" + final = inspect(runner)["lease"] + assert final["status"] == "released" + assert final["lease_epoch"] == original["lease_epoch"] + assert final["idempotency_key"] == original["idempotency_key"] + assert final["version"] == shortened["version"] + 1 + assert len(renewal_calls) == (2 if lost_reply == "renewal" else 1) + assert len(completion_calls) == (2 if lost_reply == "completion" else 1) + for calls in (renewal_calls, completion_calls): + assert all(args == calls[0][0] for args, _ in calls) + assert all(result["provider_revision"] == calls[0][1]["provider_revision"] for _, result in calls) + + +@pytest.mark.parametrize("authority_loss", ["expiry", "replacement"]) +def test_completion_renewal_receipt_cannot_revive_lost_execution(service, monkeypatch, authority_loss): + root, runner = service + prepare_lease(root, runner, monkeypatch) + cli = runner._cli + dropped = False + completions = [] + + def lose_renewal_reply(binding, *args, **kwargs): + nonlocal dropped + if args[:2] == ("todo", "complete"): + completions.append(args) + result = cli(binding, *args, **kwargs) + if args[:2] == ("task-lease", "renew") and not dropped: + dropped = True + raise ValueError("fixture lost renewal response before terminal intent") + return result + + monkeypatch.setattr(runner, "_cli", lose_renewal_reply) + runner.execute("lease-lifetime") + row = _read(runner.path("lease-lifetime")) + assert row["status"] == "turn_returned", row + assert "completion_lease_renewal_version" in row + assert "completion_lease_version" not in row + current = inspect(runner)["lease"] + binding = runner.binding("analysis") + args = ("--goal-id", runner.goal_id, "--todo-id", binding["todo_id"], + "--owner", binding["agent_id"], "--idempotency-key", current["idempotency_key"], + "--expected-version", str(current["version"])) + if authority_loss == "expiry": + expired = cli(binding, "task-lease", "renew", *args, "--ttl-seconds", "1")["lease"] + time.sleep(max(0, datetime.fromisoformat(expired["expires_at"].replace("Z", "+00:00")).timestamp() + - time.time()) + 0.1) + else: + assert cli(binding, "task-lease", "release", *args)["ok"] is True + replacement = cli(binding, "task-lease", "acquire", "--goal-id", runner.goal_id, + "--todo-id", binding["todo_id"], "--owner", binding["agent_id"], + "--idempotency-key", "replacement", "--expected-version", str(current["version"])) + assert replacement["lease"]["lease_epoch"] > current["lease_epoch"] + runner.execute("lease-lifetime") + rejected = _read(runner.path("lease-lifetime")) + assert rejected["status"] != "accepted", rejected + assert "completion_lease_version" not in rejected + assert not completions + assert (root / "analyst/initial/host-invocations").read_text() == "1" + snapshot = read_canonical_todos_if_promoted(runtime_root=runner.root, goal_id=runner.goal_id) + todo = next(item for item in snapshot["todos"] if item["todo_id"] == binding["todo_id"]) + assert todo["done"] is False + + def await_started(root, future, runner): deadline = time.monotonic() + 45 while time.monotonic() < deadline and not (root / "host-started").exists(): From 60a0523836e92957a1d180bc60947274bcb4e7f5 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Fri, 2 Oct 2026 20:46:51 +0800 Subject: [PATCH 2/3] docs(authority): record delegated completion lease boundary Signed-off-by: huangruiteng --- .../2026-09-28-retirement-cadence.md | 15 ++++++++++++--- .../2026-09-28-retirement-cadence.zh-CN.md | 12 ++++++++++++ 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md index f453715744..1619aeecbb 100644 --- a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md +++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.md @@ -475,8 +475,13 @@ Renewal uses the latest proved version, one unchanged-intent retry for a lost transport reply, and the last proved expiry even when renewal hangs. A rejected proof cancels the CLI; its TERM adapter unwinds nested managed Hosts before returning. Ordinary non-hard delegation keeps the existing subprocess route. -Completion reads the current claim, persists its terminal CAS intent before the -effect, and replays that exact completion after an ambiguous reply. Canonical +After the supervised CLI returns, completion renews the original execution +using its acquisition TTL before starting independent Todo acceptance. It +journals the renewal version before the effect, then proves current authority +and freezes the terminal CAS version. Lost renewal and completion replies each +replay their original intent; neither replay can revive an expired or replaced +execution. The validation phase makes no lease writes that would invalidate +its provider-revision witness. Canonical completion releases the execution lease; subsequent original-Turn accounting uses its terminal receipt rather than reacquiring an open-work lease. @@ -492,7 +497,11 @@ their own runtime instead of inheriting operator state. The acceptance slice uses disposable File/SQLite providers, real CLI/Turn execution and a synthetic model process: crossing the initial expiry, canonical release, a new execution epoch, lost completion/renewal replies and rejected or -hung renewal, including control-pipe loss with a TERM-resistant process. It +hung renewal, including control-pipe loss with a TERM-resistant process. Final +Todo acceptance also crosses a short remaining lease deadline, with lost-reply, +expiry and replacement controls. The Python delegation adapter only sequences +the existing TS-owned claim, renew and terminal contracts; no provider rule, +public setting or frontend permission changes. It does not qualify a paid model, remote job cancellation or Windows process-tree cleanup. Stop acknowledgements and interrupted-Turn no-progress settlement remain with the existing delegation-stop work (#5308); diff --git a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md index 92523f6dd7..69f6fa9afb 100644 --- a/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md +++ b/docs/architecture/rfcs/ledger/shared-goal-authority-state-provider-v0/2026-09-28-retirement-cadence.zh-CN.md @@ -321,3 +321,15 @@ Archive restore／audit 已使用 provider 通用的 1–64 个操作批量回 验收。File 每恢复一笔仍重写保留的文件；此前一次完整历史恢复超出调用方的 300 秒超时，随后才发布精确匹配的确认。批量回执优化没有闭合这项恢复成本。 #4224 的 soak 已启动；其最终证据和对当前候选的适用性仍待核对。 + +### 委派完成阶段的原租约衔接 + +#5436 的 TS Host 监督覆盖模型执行和 Turn 验证，最终 Todo 独立验收位于其后。 +因此完成前先按原获取 TTL 续租同一执行，分别持久化续租与完成的 CAS 版本，再 +进入验收。两种回复丢失都重放原请求；历史回执不能恢复已过期或被替换的执行。 +验收期间不写租约，避免自己改变验收所依赖的 provider revision。 + +隔离 File／SQLite 回归覆盖剩余短租约跨越最终验收，以及续租／完成回复丢失、 +过期和新 epoch。Python 仅衔接既有 TS claim、renew、terminal 权威合同，没有 +新增 provider 规则、公开配置或前端权限。Stop ACK 与中断后的无进展结算仍归 +#5308；持续 D2、默认准入和旧 writer 删除仍需各自证据。 From d60680076e7288758ec76c1ae2f0be546cee2e0c Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Sat, 3 Oct 2026 01:52:08 +0800 Subject: [PATCH 3/3] fix(turn): preserve empty-frontier replan after delegated completion Signed-off-by: huangruiteng --- loopx/control_plane/turn_driver/driver.py | 5 +++ tests/test_delegation_lease_lifetime.py | 42 +++++++++++++++---- tests/test_loopx_turn_driver.py | 49 ++++++++++++++++++++++- 3 files changed, 87 insertions(+), 9 deletions(-) diff --git a/loopx/control_plane/turn_driver/driver.py b/loopx/control_plane/turn_driver/driver.py index a68fb48726..9ffa6f1a4f 100644 --- a/loopx/control_plane/turn_driver/driver.py +++ b/loopx/control_plane/turn_driver/driver.py @@ -460,6 +460,11 @@ def build_loopx_turn_plan( else LoopXTurnRoute.CONTRACT_ERROR ) selected_todo = selected_turn_todo(envelope) + # Replanning an empty frontier is valid, but this Host executor settles + # against a Todo. Preserve the replan packet for the controller to resolve; + # do not turn the absence of a successor into a malformed Host session. + if route is LoopXTurnRoute.REPLAN_REQUIRED and not selected_todo: + route = LoopXTurnRoute.BLOCKED lineage = _turn_lineage(envelope, selected_todo=selected_todo) session, session_error = _session_plan( route=route, diff --git a/tests/test_delegation_lease_lifetime.py b/tests/test_delegation_lease_lifetime.py index e26e5e3f5c..e47deab6f9 100644 --- a/tests/test_delegation_lease_lifetime.py +++ b/tests/test_delegation_lease_lifetime.py @@ -42,6 +42,8 @@ def prepare_lease(root, runner, monkeypatch, *, ttl=20): binding = runner.binding("analysis") runner._acquire_delegation_lease(runner.path("lease-lifetime"), row, binding) lease = row["task_lease"]["lease"] + if ttl is None: + return lease renewed = runner._cli(binding, "task-lease", "renew", "--goal-id", runner.goal_id, "--todo-id", binding["todo_id"], "--owner", binding["agent_id"], "--idempotency-key", lease["idempotency_key"], "--expected-version", str(lease["version"]), @@ -59,13 +61,15 @@ def completion_service(tmp_path, request, monkeypatch): """Pin a real slow final acceptance command before preparing authority.""" validator = tmp_path / "completion-validator.py" validator.write_text('''import sys, time, runpy +from datetime import datetime from pathlib import Path root = Path(sys.argv[1]) counter = root / 'validator-calls' calls = int(counter.read_text()) + 1 if counter.exists() else 1 counter.write_text(str(calls)) if calls == 2: - time.sleep(8) + deadline = datetime.fromisoformat((root / 'completion-prior-expiry').read_text()) + time.sleep(max(0, deadline.timestamp() - time.time()) + 0.2) (root / 'completion-validation-ended').touch() actual = root / 'project/validation/acceptance.py' sys.path.insert(0, str(actual.parent)) @@ -79,7 +83,9 @@ def configure(path, value): for criterion in value["document"]["criteria"]: criterion["validation_timeout_seconds"] = 1 if criterion["id"] == "analyst-initial": - criterion["validation_timeout_seconds"] = 15 + # 21 + four one-second criteria stays within the public + # 25-second completion budget, including a 20s deadline. + criterion["validation_timeout_seconds"] = 21 criterion["validation_argv"][1] = str(validator) return write(path, value) @@ -90,7 +96,10 @@ def configure(path, value): @pytest.mark.parametrize("lost_reply", [None, "renewal", "completion"]) def test_completion_renews_before_validation_and_replays_each_intent(completion_service, monkeypatch, lost_reply): root, runner = completion_service - original = prepare_lease(root, runner, monkeypatch) + # This case shortens the lease at the completion boundary below. An + # unrelated short Host deadline can cancel execution before that boundary + # under load; the running-Host cases separately exercise that deadline. + original = prepare_lease(root, runner, monkeypatch, ttl=None) complete, cli = runner._complete_delegated_todo, runner._cli shortened = None renewal_calls, completion_calls = [], [] @@ -105,7 +114,11 @@ def enter_completion(row, binding): shortened = cli(binding, "task-lease", "renew", "--goal-id", runner.goal_id, "--todo-id", binding["todo_id"], "--owner", binding["agent_id"], "--idempotency-key", current["idempotency_key"], - "--expected-version", str(current["version"]), "--ttl-seconds", "6")["lease"] + "--expected-version", str(current["version"]), "--ttl-seconds", "20")["lease"] + # Cross the actual pre-renewal deadline, not an assumed amount of + # CLI startup time. Allow cold claim/renew commands to reach the + # boundary; the independent validator still outlives that lease. + (root / "completion-prior-expiry").write_text(shortened["expires_at"]) return complete(row, binding) def observe_reply(binding, *args, **kwargs): @@ -272,11 +285,24 @@ def test_real_revocation_or_new_execution_stops_nested_host_without_acceptance(s with ThreadPoolExecutor(max_workers=1) as pool: future = pool.submit(runner.execute, "lease-lifetime") await_started(root, future, runner) - current = inspect(runner)["lease"] binding = runner.binding("analysis") - runner._cli(binding, "task-lease", "release", "--goal-id", runner.goal_id, - "--todo-id", "todo_analyst-initial", "--owner", "analyst", - "--idempotency-key", original["idempotency_key"], "--expected-version", str(current["version"])) + # The live supervisor may renew between inspection and revocation. + # Retry only that CAS race, never a replacement execution or rejection. + for _ in range(5): + current = inspect(runner)["lease"] + assert current["lease_epoch"] == original["lease_epoch"] + assert current["idempotency_key"] == original["idempotency_key"] + try: + runner._cli(binding, "task-lease", "release", "--goal-id", runner.goal_id, + "--todo-id", "todo_analyst-initial", "--owner", "analyst", + "--idempotency-key", original["idempotency_key"], + "--expected-version", str(current["version"])) + break + except ValueError as exc: + if str(exc) != "canonical task lease release rejected: version_mismatch": + raise + else: + pytest.fail("could not revoke the original execution during concurrent renewal") if reclaim: acquired = runner._cli(binding, "task-lease", "acquire", "--goal-id", runner.goal_id, "--todo-id", "todo_analyst-initial", "--owner", "analyst", "--idempotency-key", "new-execution", diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index e06b21b2cc..531060444e 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -1111,6 +1111,45 @@ def test_turn_plan_rejects_contradictory_scheduler_owner() -> None: assert "cannot be owned by host_automation" in payload["error"] +@pytest.mark.parametrize( + "effective_action", + ["autonomous_replan", "autonomous_replan_required", "successor_replan_required"], +) +def test_turn_plan_keeps_todoless_replan_for_controller(effective_action: str) -> None: + envelope = _envelope(effective_action=effective_action) + envelope["action"]["selected_todo"] = None + payload = build_loopx_turn_plan( + envelope, host="generic-cli", execution_mode="isolated-headless", + ) + + assert payload["ok"] is True + assert payload["route"]["kind"] == "blocked" + assert payload["route"]["would_invoke_host"] is False + assert payload["session"]["action"] == "none" + assert payload["transaction"]["status"] == "not_applicable" + assert not any(payload["effects"].values()) + assert payload["turn_envelope"] == envelope + + +@pytest.mark.parametrize("invalid_signature", [False, True]) +def test_turn_plan_missing_todo_does_not_bypass_contract_errors( + invalid_signature: bool, +) -> None: + envelope = _envelope( + effective_action="autonomous_replan_required" if invalid_signature else "normal_run", + ) + envelope["action"]["selected_todo"] = None + if invalid_signature: + envelope["action_signature"]["matches"] = False + payload = build_loopx_turn_plan( + envelope, host="generic-cli", execution_mode="isolated-headless", + ) + + assert payload["ok"] is False + assert payload["route"]["kind"] == "contract_error" + assert payload["route"]["would_invoke_host"] is False + + def test_turn_plan_preserves_safe_bypass_when_user_action_is_visible() -> None: payload = build_loopx_turn_plan( _envelope(action_required=True), @@ -2154,8 +2193,13 @@ def run(turn_instance_id: str, context: str) -> dict[str, object]: assert state_path.read_text(encoding="utf-8") == before_state +@pytest.mark.parametrize( + "next_action", + ["Select the next Todo from a fresh decision.", "Assess remaining work."], +) def test_turn_run_once_cli_completes_selected_todo_after_validation( tmp_path: Path, + next_action: str, ) -> None: project, runtime, registry = _write_live_fixture(tmp_path) host_project = tmp_path / "isolated-host-workspace" @@ -2180,6 +2224,9 @@ def test_turn_run_once_cli_completes_selected_todo_after_validation( "summary": "One public fixture completed." }, sys.stdout) """ + host_script = host_script.replace( + "Select the next Todo from a fresh decision.", next_action, + ) validation_script = """ import json import pathlib @@ -2291,7 +2338,7 @@ def test_turn_run_once_cli_completes_selected_todo_after_validation( ] ) next_plan = json.loads(next_plan_output.getvalue()) - assert next_plan_exit_code == 0, next_plan + assert next_plan_exit_code == 0, json.dumps(next_plan, indent=2) # The completion-created obligation is deferred only for the causal # closeout write. It must be visible immediately at the next decision; # Turn itself stays blocked until that replan creates a runnable successor.