refactor(lease): read source todos without the mutation facade

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>

Author: loopx-agent

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -1515,6 +1515,15 @@ Python writer retirement remain separate exits.
   change, and provider outage without a Markdown fallback. Unpromoted writers,
   lease capture recovery and the legacy migration window remain separate
   retirement obligations; this closes the canonical setter's import dependency.
+- Source task-lease facts now use `todos/list_readback.py` directly instead of
+  the historic mutation facade. TypeScript retains acquire/lifecycle decisions;
+  Markdown IO, source capture and public compatibility exports remain live.
+  The CLI distribution counterfactual removes the facade and disables editable
+  install finders, which could otherwise borrow deleted modules from a checkout.
+  Source and installed-wheel coverage includes capture off/on, all three handoff
+  modes, replay, ownership/version rejection, transfer, release and retained
+  history. This removes a reader-to-writer dependency; full writer deletion still
+  needs its remaining callers and migration window qualified.
 
 **Validation and stop rules for every card**
 

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -1041,6 +1041,13 @@ acceptance 和 vision。展示限制不能变成结算限制。在 legacy、File
   File/SQLite CLI 验证覆盖预览、写入、后续模式变化后的原操作回放及 provider
   故障时禁止 Markdown 回退。本批闭合 canonical setter 的 import 依赖；未晋升
   writer、lease capture 恢复及旧源迁移窗口仍保留独立退役验收。
+- 旧源 task-lease facts 直接复用 `todos/list_readback.py`，移除对历史 mutation
+  facade 的依赖。领取及生命周期决策仍由 TypeScript 拥有；Markdown IO、source
+  capture 与公开兼容 export 仍有实际 caller。CLI 分发反例移除 facade 并禁用
+  editable install finder，防止从 checkout 偷读已删除模块。源码与安装 wheel
+  覆盖 capture 关闭／开启、三种 handoff 模式、回放、owner/version 拒绝、转交、
+  释放及保留历史。本批闭合 reader 到 writer facade 的依赖；完整 writer 删除
+  仍须验收剩余 caller 与迁移窗口。
 
 **每张执行卡的验证与停止规则**
 

loopx/control_plane/work_items/task_lease_acquire_adapter.py

@@ -165,7 +165,7 @@ def _task_lease_authority_projection(
     state_file: Path | None,
     project_todos: bool = True,
 ) -> tuple[Any, list[Any], list[dict[str, Any]], dict[str, Any] | None]:
-    from ...todos import list_goal_todos
+    from ..todos.list_readback import list_goal_todos
     from ..goals.active_state_metadata import parse_state_frontmatter
 
     if goal is None:

tests/control_plane/test_canonical_todo_writer_isolation.py

@@ -9,6 +9,7 @@
 import shutil
 import subprocess
 import sys
+import sysconfig
 
 import pytest
 
@@ -46,7 +47,7 @@ def isolated_todo_distribution(tmp_path, monkeypatch, request):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     package_root = tmp_path / "package"
     package = package_root / "loopx"
-    command = [sys.executable, "-m", "loopx.cli"]
+    command = [sys.executable, "-S", "-m", "loopx.cli"]
     if request.param == "source":
         shutil.copytree(Path(__file__).resolve().parents[2] / "loopx", package,
                         ignore=shutil.ignore_patterns("__pycache__"))
@@ -59,10 +60,13 @@ def isolated_todo_distribution(tmp_path, monkeypatch, request):
         assert installed.returncode == 0, installed.stdout + installed.stderr
         entry = package_root / "bin/loopx"
         assert entry.is_file()
-        command = [sys.executable, str(entry)]
-    monkeypatch.setenv("PYTHONPATH", str(package_root))
+        command = [sys.executable, "-S", str(entry)]
+    # Keep dependencies available without processing editable-install .pth files.
+    # Their import finder can silently borrow a deleted module from the checkout.
+    dependencies = dict.fromkeys(sysconfig.get_paths()[key] for key in ("purelib", "platlib"))
+    monkeypatch.setenv("PYTHONPATH", os.pathsep.join([str(package_root), *dependencies]))
     provenance = json.loads(subprocess.check_output(
-        [sys.executable, "-c", "import json,loopx; from importlib import metadata; "
+        [sys.executable, "-S", "-c", "import json,loopx; from importlib import metadata; "
          "print(json.dumps({'module':loopx.__file__, "
          "'distribution':str(metadata.distribution('loopx').locate_file(''))}))"],
         cwd=tmp_path, text=True,
@@ -122,7 +126,7 @@ def test_pending_source_outbox_survives_producer_retirement(
 
     def history():
         result = subprocess.run(
-            [sys.executable, "-c",
+            [sys.executable, "-S", "-c",
              "import json,sys;from pathlib import Path;"
              "from loopx.control_plane.coordination.local_authority_shadow_adapter "
              "import read_local_authority_shadow;"
@@ -253,7 +257,7 @@ def test_handoff_mode_without_source_todo_writers(tmp_path, provider, without_so
 
     def cli(*args, succeeds=True, override=False):
         result = subprocess.run(
-            [sys.executable, "-m", "loopx.cli", "--format", "json", "--registry", str(registry),
+            [*without_source_todo_writers, "--format", "json", "--registry", str(registry),
              *(["--runtime-root", str(runtime)] if override else []), *args, "--goal-id", "goal-a"],
             cwd=caller, capture_output=True, text=True, timeout=45,
         )

tests/control_plane/test_source_lease_reader_isolation.py

@@ -0,0 +1,178 @@
+"""Supported source leases retain readback without the historic mutation facade."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+
+import pytest
+
+from shadow_e2e_fixture import workspace
+from test_canonical_todo_writer_isolation import (
+    isolated_todo_distribution,  # noqa: F401 -- distribution fixture
+    retirement_wheel,  # noqa: F401 -- distribution fixtures
+)
+
+
+@pytest.mark.parametrize("facade", ["present", "absent"])
+@pytest.mark.parametrize("capture", [False, True], ids=["capture-off", "capture-on"])
+@pytest.mark.parametrize("mode", ["legacy", "hard_lease", "soft_claim"])
+def test_source_lease_lifecycle_uses_retained_reader(
+    tmp_path,
+    isolated_todo_distribution,  # noqa: F811
+    facade,
+    capture,
+    mode,  # noqa: F811
+):
+    package, command = isolated_todo_distribution
+    w = workspace(tmp_path / "source", bootstrap=False)
+    w.command, w.cwd, w.package = tuple(command), tmp_path, package
+    w.state.write_text(
+        w.state.read_text().replace("handoff_mode: hard_lease", f"handoff_mode: {mode}")
+        + "- [ ] [P1] Preserve source work — 完整内容.\n"
+        "  <!-- loopx:todo todo_id=todo_source status=open task_class=advancement_task -->\n"
+    )
+    config = json.loads(w.registry.read_text())
+    config["goals"][0]["coordination"]["runtime_shadow"]["enabled"] = capture
+    w.registry.write_text(json.dumps(config))
+    if capture:
+        assert (
+            w.cli("coordination-shadow", "bootstrap", "--execute")["bootstrap"][
+                "status"
+            ]
+            == "applied"
+        )
+    if facade == "absent":
+        (package / "todos.py").unlink()
+        subprocess.run(
+            [
+                sys.executable,
+                "-S",
+                "-c",
+                "from importlib.util import find_spec;assert find_spec('loopx.todos') is None",
+            ],
+            cwd=tmp_path,
+            check=True,
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+    original = w.state.read_bytes(), w.registry.read_bytes()
+
+    def lease(action, *args, success=True):
+        return w.cli(
+            "task-lease", action, "--todo-id", "todo_source", *args, success=success
+        )
+
+    first = lease(
+        "acquire",
+        "--owner",
+        "agent-a",
+        "--idempotency-key",
+        "source-a",
+        "--ttl-seconds",
+        "600",
+        success=mode != "soft_claim",
+    )
+    if mode == "soft_claim":
+        assert not first["ok"] and first["error_code"] == "handoff_mode_forbids_lease"
+        assert lease("inspect")["lease"] is None
+        assert (w.state.read_bytes(), w.registry.read_bytes()) == original
+        return
+
+    assert first["ok"] and first["acquired"] and first["lease"]["owner"] == "agent-a"
+    assert first["lease"]["version"] == first["lease"]["lease_epoch"] == 1
+    assert lease("inspect")["active"] is True
+    retry = lease(
+        "acquire",
+        "--owner",
+        "agent-a",
+        "--idempotency-key",
+        "source-a",
+        "--ttl-seconds",
+        "600",
+    )
+    assert retry["idempotent"] and retry["lease"] == first["lease"]
+    denied = lease(
+        "acquire", "--owner", "agent-b", "--idempotency-key", "other", success=False
+    )
+    assert not denied["ok"] and denied["error_code"] == "todo_lease_conflict"
+    renewed = lease(
+        "renew",
+        "--owner",
+        "agent-a",
+        "--idempotency-key",
+        "source-a",
+        "--expected-version",
+        "1",
+    )
+    assert renewed["lease"]["version"] == 2 and renewed["lease"]["lease_epoch"] == 1
+    # The exact original renewal is receipt recovery, not a stale new write.
+    replay = lease(
+        "renew",
+        "--owner",
+        "agent-a",
+        "--idempotency-key",
+        "source-a",
+        "--expected-version",
+        "1",
+    )
+    assert replay["idempotent"] and replay["lease"] == renewed["lease"]
+    assert lease("inspect")["lease"] == renewed["lease"]
+    transferred = lease(
+        "transfer",
+        "--owner",
+        "agent-a",
+        "--idempotency-key",
+        "source-a",
+        "--expected-version",
+        "2",
+        "--new-owner",
+        "agent-b",
+        "--new-idempotency-key",
+        "source-b",
+    )
+    assert (
+        transferred["lease"]["version"] == 3
+        and transferred["lease"]["lease_epoch"] == 2
+    )
+    assert transferred["lease"]["owner"] == "agent-b"
+    stale = lease(
+        "renew",
+        "--owner",
+        "agent-b",
+        "--idempotency-key",
+        "source-b",
+        "--expected-version",
+        "2",
+        success=False,
+    )
+    assert not stale["ok"] and stale["error_code"] == "version_mismatch"
+    assert lease("inspect")["lease"] == transferred["lease"]
+    released = lease(
+        "release",
+        "--owner",
+        "agent-b",
+        "--idempotency-key",
+        "source-b",
+        "--expected-version",
+        "3",
+    )
+    assert released["lease"]["status"] == "released"
+    retained = lease("inspect")
+    assert not retained["active"] and retained["lease"] == released["lease"]
+    assert (w.state.read_bytes(), w.registry.read_bytes()) == original
+    if capture:
+        # Read the retained journal after the lifecycle; no capture remains pending.
+        history = json.loads(subprocess.check_output(
+            [sys.executable, "-S", "-c",
+             "import json,sys;from pathlib import Path;"
+             "from loopx.control_plane.coordination.local_authority_shadow_history "
+             "import outbox_summary;"
+             "print(json.dumps(outbox_summary(Path(sys.argv[1]),sys.argv[2])))",
+             str(w.runtime), w.goal],
+            cwd=tmp_path, text=True, timeout=30,
+        ))
+        assert history["leases"]["prepared_only"] == 0
+        assert history["leases"]["committed_pending"] == 0