fix(lease): align cutoff recovery across CLI formats

docs/reference/canonical-lease-renew.md

@@ -4,8 +4,15 @@ An already promoted File or SQLite Goal runs `task-lease acquire`, `renew`,
 `transfer` and `release` through TS-owned provider transactions. Fresh execution
 and expired/ineffective-holder takeover now use the complete canonical head;
 previously standalone acquire still entered the fenced legacy file writer.
-Unpromoted Goals retain their legacy wire and transaction. A provider selector
-does not promote a Goal or grant execution authority.
+Unpromoted Goals retain historical lease inspection, exact release and original
+receipt recovery. Fresh acquire, renew and transfer require reviewed import and
+return `canonical_authority_required` with backup-and-migrate recovery, including
+`transfer --transfer-claim`. A provider selector does not promote a Goal or grant
+execution authority.
+
+> **2.0 cutoff:** the capture-isolation section below records the pre-cutoff
+> migration stage, not an available Markdown write route. Retained receipts,
+> lease cleanup and pending outbox recovery do not authorize new execution.
 
 For a new execution on an open, eligible Todo with no prior lease:
 
@@ -505,7 +512,10 @@ not replace [D2 capacity and continuity qualification](sqlite-authority-store.md
 
 已 promoted 的 File/SQLite Goal，其 acquire、renew、transfer、release 现在使用
 同一 provider opening/source fence 与 TS 规则；新领取和失效持有者接管不再进入
-旧文件 writer。旧 wire 与未 promoted 路径保留，选择 provider 不构成 promotion。
+旧文件 writer。未迁移 Goal 只保留历史租约检查、精确释放与原回执恢复；新领取、
+续约和转交（包括 `--transfer-claim`）统一返回 `canonical_authority_required`
+及备份后迁移提示。选择 provider 不构成 promotion，也不授予执行权。下方 capture
+隔离章节保留退役前阶段的历史，不代表 2.0 仍支持 Markdown 普通写入。
 
 新执行按上面的 acquire 命令领取；没有旧 lease 时 expected-version 为 0，否则
 用 inspect 的当前版本及新的 execution key。领取/接管同时增加 version 和 epoch，

loopx/cli_commands/task_lease.py

@@ -39,7 +39,10 @@ def render_task_lease_markdown(payload: dict[str, object]) -> str:
     if payload.get("error_code"):
         lines.append(f"- error_code: `{payload.get('error_code')}`")
     recovery = payload.get("recovery")
-    if isinstance(recovery, dict):
+    if isinstance(recovery, dict) and payload.get("error_code") == "canonical_authority_required":
+        lines.append(f"- recovery: `{recovery['action']}`")
+        lines.append(f"- backup_required: `{recovery['backup_required']}`")
+    elif isinstance(recovery, dict):
         import shlex
         lines.append(f"- repair: `{shlex.join(['loopx', *recovery['cli_args']])}`")
         lines.append(f"- supply: `{', '.join(recovery['requires_flags']) or 'none'}`")

loopx/control_plane/work_items/task_lease_acquire_adapter.py

@@ -477,10 +477,9 @@ def execute_native_task_lease_lifecycle(
                 runtime_root=runtime_root, goal_id=goal_id
             )
         if transfer_claim and not canonical_lifecycle:
-            raise TaskLeaseError(
-                "atomic claim transfer requires canonical authority; inspect provider readiness first",
-                code="claim_transfer_requires_canonical_authority",
-            )
+            from ..coordination.local_authority import require_canonical_authority
+
+            require_canonical_authority(goal_id=goal_id, operation="task_lease_transfer")
         if registry_path is not None and needs_authority:
             authority = _canonical_lease_authority_facts(registry_path, goal_id)
         request: dict[str, Any] = {

tests/control_plane/test_canonical_claim_transfer.py

@@ -3,6 +3,7 @@
 import os
 import subprocess
 import sys
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 import pytest
@@ -97,9 +98,25 @@ def cli(command, *args, expected_exit=0):
                        cwd=cli_repo, capture_output=True, text=True, timeout=30, check=True)
 
 
-def test_claim_transfer_never_emulates_a_legacy_two_write_handover(tmp_path):
+@pytest.mark.parametrize("handoff_mode", ["legacy", "soft_claim", "hard_lease"])
+@pytest.mark.parametrize("output_format", ["json", "markdown", None])
+def test_claim_transfer_never_emulates_a_legacy_two_write_handover(tmp_path, handoff_mode, output_format):
     registry = tmp_path / "registry.json"
     runtime = tmp_path / "runtime"
+    state = tmp_path / "state.md"
+    state.write_text(f"---\nhandoff_mode: {handoff_mode}\n---\n"
+                     "# Retained source\n\n## Agent Todo\n- [ ] Historical work <!-- todo_id=todo_example -->\n")
+    lease_path = runtime / "goals/example/task-leases/todo_example.json"
+    lease_path.parent.mkdir(parents=True)
+    now = datetime.now(UTC).replace(microsecond=0)
+    lease_path.write_text(json.dumps({
+        "schema_version": "task_lease_v0", "goal_id": "example", "todo_id": "todo_example",
+        "owner": "agent-a", "idempotency_key": "source-a", "version": 1, "lease_epoch": 1,
+        "status": "active", "write_scopes": ["src/**"],
+        "acquire_ttl_seconds": 600, "acquired_at": now.isoformat(), "updated_at": now.isoformat(),
+        "expires_at": (now + timedelta(minutes=10)).isoformat(),
+    }))
+    original_state, original_lease = state.read_bytes(), lease_path.read_bytes()
     registry.write_text(json.dumps({
         "common_runtime_root": str(runtime),
         "goals": [{
@@ -119,4 +136,23 @@ def test_claim_transfer_never_emulates_a_legacy_two_write_handover(tmp_path):
     assert caught.value.code == "canonical_authority_required"
     assert caught.value.payload["operation"] == "task_lease_transfer"
     assert caught.value.payload["write_admitted"] is False
-    assert not (runtime / "goals/example/task-leases/todo_example.json").exists()
+    child = subprocess.run([
+        sys.executable, "-m", "loopx.cli", "--registry", str(registry),
+        *(["--format", output_format] if output_format is not None else []),
+        "task-lease", "transfer", "--goal-id", "example", "--todo-id", "todo_example",
+        "--owner", "agent-a", "--idempotency-key", "source-a", "--expected-version", "1",
+        "--new-owner", "agent-b", "--new-idempotency-key", "receiver-b", "--transfer-claim",
+    ], cwd=REPO, capture_output=True, text=True, timeout=60, check=False)
+    assert child.returncode == 1, child.stdout + child.stderr
+    assert "Traceback" not in child.stderr
+    if output_format == "json":
+        rejected = json.loads(child.stdout)
+        assert rejected["error_code"] == "canonical_authority_required"
+        assert rejected["write_admitted"] is False
+        assert rejected["recovery"] == {"action": "migrate_goal_authority", "backup_required": True}
+    else:
+        assert "canonical_authority_required" in child.stdout
+        assert "migrate_goal_authority" in child.stdout
+        assert "backup_required: `True`" in child.stdout
+    assert state.read_bytes() == original_state
+    assert lease_path.read_bytes() == original_lease

tests/control_plane/test_todo_update_lease_recovery.py

@@ -4,6 +4,7 @@
 import json
 import subprocess
 import sys
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 import pytest
@@ -15,8 +16,8 @@
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-@pytest.mark.parametrize("mode", ["legacy", "hard_lease"])
-def test_released_lease_recovery_through_public_cli(tmp_path: Path, monkeypatch, provider: str, mode: str) -> None:
+def test_hard_lease_recovery_through_public_cli(tmp_path: Path, monkeypatch, provider: str) -> None:
+    mode = "hard_lease"
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     runtime = tmp_path / "runtime"
     state = tmp_path / "ACTIVE_GOAL_STATE.md"
@@ -94,6 +95,64 @@ def snapshot() -> dict:
     assert "Recovered edit" in state.read_text(encoding="utf-8")
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_legacy_history_remains_readable_and_releasable_without_execution(tmp_path, monkeypatch, provider):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    runtime, state, registry = tmp_path / "runtime", tmp_path / "state.md", tmp_path / "registry.json"
+    state.write_text("# Retained legacy display\n")
+    registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{
+        "id": "legacy-history", "repo": str(tmp_path), "state_file": state.name,
+        "coordination": {"registered_agents": ["agent-a"]},
+    }]}))
+    now = datetime.now(UTC).replace(microsecond=0)
+    lease = {"schema_version": "task_lease_v0", "goal_id": "legacy-history", "todo_id": "todo_history",
+             "owner": "agent-a", "idempotency_key": "historical-execution", "version": 3, "lease_epoch": 2,
+             "status": "active", "write_scopes": [], "acquire_ttl_seconds": 600,
+             "acquired_at": now.isoformat(), "updated_at": now.isoformat(),
+             "expires_at": (now + timedelta(minutes=10)).isoformat()}
+    projection = build_todo_runtime_shadow_projection(goal_id="legacy-history", handoff_mode="legacy", todos=[{
+        "schema_version": "todo_item_v0", "source_section": "Agent Todo", "index": 1,
+        "todo_id": "todo_history", "text": "Retained task", "role": "agent", "status": "open",
+        "done": False, "archive_state": "active", "claimed_by": "agent-a", "task_class": "advancement_task",
+    }])
+    projection["leases"] = [lease]
+    initialize_canonical_authority(runtime, "legacy-history", projection, state_path=state, provider=provider)
+    registry_bytes, display_bytes = registry.read_bytes(), state.read_bytes()
+
+    def cli(*args, ok=True):
+        child = subprocess.run([sys.executable, "-m", "loopx.cli", "--format", "json",
+            "--registry", str(registry), "--runtime-root", str(runtime), *args,
+            "--goal-id", "legacy-history", "--todo-id", "todo_history"],
+            capture_output=True, text=True, timeout=45)
+        assert (child.returncode == 0) is ok, child.stdout + child.stderr
+        return json.loads(child.stdout)
+
+    def snapshot():
+        return read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="legacy-history", include_leases=True)
+
+    before = snapshot()
+    inspected = cli("task-lease", "inspect")
+    assert inspected["active"] is False and inspected["lease"] == lease
+    rejected = cli("task-lease", "acquire", "--owner", "agent-a", "--idempotency-key", "new-execution",
+                   "--expected-version", "3", ok=False)
+    assert rejected["error_code"] == "handoff_mode_retired"
+    rejected = cli("todo", "update", "--agent-id", "agent-a", "--note", "Forbidden execution",
+                   "--task-lease-idempotency-key", "historical-execution", "--task-lease-expected-version", "3", ok=False)
+    assert rejected["error_code"] == "handoff_mode_retired"
+    assert snapshot() == before
+    release = ("task-lease", "release", "--owner", "agent-a", "--idempotency-key", "historical-execution",
+               "--expected-version", "3")
+    released = cli(*release)
+    assert released["lease"]["status"] == "released"
+    assert released["lease"]["version"] == 3 and released["lease"]["lease_epoch"] == 2
+    after = snapshot()
+    assert after["todos"] == before["todos"]
+    assert cli(*release)["original_receipt"] == released["original_receipt"]
+    assert snapshot() == after
+    assert cli("task-lease", "inspect")["active"] is False
+    assert registry.read_bytes() == registry_bytes and state.read_bytes() == display_bytes
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("claimed,lease_state", [(False, None), (True, None), (True, "expired"), (True, "released"), (True, "active")])
 def test_blocked_edit_recovers_by_separate_lifecycle_before_execution(