test(migration): characterize source stop and lease drain

docs/reference/configuration-backup.md

@@ -293,6 +293,39 @@ to resume a modern Agent from a backup.
 这是预览行为变更；执行时仍在原锁内重验，成功预览不保证以后可写。该旧工具和
 隔离 `authority-archive restore` 都不能作为备份后恢复现代 Agent 执行的捷径。
 
+### Source stop, admission closure and lease drain
+
+These existing operations have different responsibilities. `goal-lifecycle`
+stop suspends automatic advancement; it preserves an already admitted Agent's
+lease and permits its authorized Todo writeback. It is not a quiescence receipt.
+Withdrawing registered source peers rejects their current Todo writes and fresh
+lease acquisitions, but does not release stored lease tuples. Lease inspection
+can therefore report `active: false` with `executor_constraint.reason` equal to
+`owner_not_registered` while the retained `lease.status` is still `active`.
+That effective constraint must not be mistaken for a drained lease: membership
+can be restored during recovery.
+
+The exact original owner/key/version can still release a lease after withdrawal.
+After that release, restoring membership does not revive its old execution proof.
+A planned-transfer operation must compose these owners, settle in-flight work
+and external effects, preserve their original receipts, and separately establish
+destination admission. Neither stop nor withdrawal certifies that a process has
+stopped or that an external effect has completed. The CLI/provider regression
+in `test_planned_transfer_source_boundary.py` checks these distinctions on File
+and SQLite, including a cold runtime; it does not qualify a complete transfer.
+This is an integration contract, not an instruction for users to stitch a
+migration together manually.
+
+中文：停止 Goal 只暂停自动推进，保留已获准 Agent 的租约和合法 Todo 回写，
+不证明源端已静止。撤回已注册 peer 会拒绝其当前回写和新租约申请，但不释放
+保存的租约；检查中的 `active: false` 可能只是 `owner_not_registered` 的有效性
+限制，原 `lease.status` 仍为 `active`，恢复成员资格可能重新使它有效。
+原 owner/key/version 在撤回后仍可精确释放租约，释放后的旧执行证明不会因重新
+注册而复活。迁移编排须组合这些 owner，排空在途工作、确认外部 effect 的实际
+结果、保存原 receipt，并另行完成目标准入。stop 或撤回都不证明进程已停止、
+外部操作已完成。File/SQLite 及冷启动回归只验证这些现有边界，不代表完整迁移
+已交付；这也不是要求用户手工拼接迁移步骤。
+
 ## Full-state recovery after cold import
 
 Keep the reviewed pre-import backup as well as a new full-state backup after

tests/control_plane/test_planned_transfer_source_boundary.py

@@ -0,0 +1,133 @@
+"""Migration must compose admission closure and drain, not reinterpret Goal stop.
+
+These are real CLI/provider characterizations of the existing owners. They do
+not implement destination admission or qualify a complete Agent transfer.
+"""
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from loopx.control_plane.effect_runtime import restart_effect_runtime
+from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_stop_withdrawal_and_exact_release_have_distinct_authority(
+    tmp_path: Path, monkeypatch, provider: str,
+) -> None:
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    home = tmp_path / "home"
+    home.mkdir()
+    monkeypatch.setenv("HOME", str(home))
+    monkeypatch.setenv("USERPROFILE", str(home))
+    monkeypatch.setenv("LOOPX_USAGE_PING", "0")
+    project, worktree, runtime = (tmp_path / name for name in ("project", "agent", "runtime"))
+    project.mkdir()
+    registry = project / ".loopx/registry.json"
+    machine = runtime / "machine/configuration.json"
+    machine.parent.mkdir(parents=True)
+    machine.write_text(json.dumps({"schema_version": "loopx_machine_configuration_v0", "namespaces": {
+        "goal_storage": {"schema_version": "loopx_goal_storage_defaults_v1",
+                         "new_goal_provider": provider, "canonical_creation": True,
+                         "new_goal_handoff_mode": "hard_lease"}}}))
+
+    def git(*arguments: str) -> None:
+        subprocess.run(["git", "-C", str(project), *arguments], check=True, capture_output=True)
+
+    git("init", "-q")
+    git("remote", "add", "origin", "https://example.org/fixtures/agent-transfer.git")
+    (project / "input.txt").write_text("6\n7\n8\n")
+    git("add", "input.txt")
+    git("-c", "user.name=Fixture", "-c", "user.email=fixture@example.org", "commit", "-qm", "Input")
+    git("worktree", "add", "-qb", "agent", str(worktree))
+
+    def cli(*arguments: str, expected_code: int = 0) -> dict:
+        result = subprocess.run([sys.executable, "-I", "-m", "loopx.cli", "--format", "json",
+            "--registry", str(registry), "--runtime-root", str(runtime), *arguments],
+            cwd=worktree, capture_output=True, text=True, timeout=60)
+        assert result.returncode == expected_code, result.stdout + result.stderr
+        return json.loads(result.stdout)
+
+    def inspect() -> dict:
+        return cli("task-lease", "inspect", "--goal-id", "report", "--todo-id", todo)
+
+    def update(operation: str, note: str, *, expected_code: int = 0) -> dict:
+        return cli("todo", "update", "--goal-id", "report", "--todo-id", todo,
+            "--agent-id", "agent-a", "--task-lease-idempotency-key", lease["idempotency_key"],
+            "--task-lease-expected-version", str(lease["version"]), "--note", note,
+            "--update-operation-id", operation, "--execute", expected_code=expected_code)
+
+    try:
+        cli("bootstrap", "--project", str(project), "--goal-id", "report",
+            "--objective", "Produce report.txt from input.txt", "--no-global-sync")
+        cli("configure-goal", "--goal-id", "report", "--registered-agent", "agent-a",
+            "--write-scope", "report.txt", "--execute")
+        created = cli("todo", "add", "--goal-id", "report", "--role", "agent",
+            "--claimed-by", "agent-a", "--text", "Write the input sum to report.txt",
+            "--task-repository", "git:example.org/fixtures/agent-transfer",
+            "--required-write-scope", "report.txt", "--operation-id", "pending-report")
+        todo = created["todo_id"]
+        acquired = cli("task-lease", "acquire", "--goal-id", "report", "--todo-id", todo,
+            "--owner", "agent-a", "--idempotency-key", "source-work", "--write-scope", "report.txt",
+            "--write-worktree", str(worktree), "--ttl-seconds", "7200")
+        lease = acquired["lease"]
+        assert acquired["source_authority"] == f"{provider}_v0"
+        assert inspect()["active"] is True
+
+        # The accepted stop operation controls automatic advancement. It must
+        # not silently revoke an admitted executor's lease or discard its work.
+        before_preview = registry.read_bytes()
+        preview = cli("goal-lifecycle", "--goal-id", "report", "--operation", "stop")
+        assert registry.read_bytes() == before_preview
+        stopped = cli("goal-lifecycle", "--goal-id", "report", "--operation", "stop",
+            "--actor-kind", "owner", "--expected-state-fingerprint",
+            preview["observed_state_fingerprint"], "--execute")
+        assert stopped["after_state"] == "stopped" and stopped["readback"]["verified"] is True
+        assert inspect()["lease"] == lease
+        assert inspect()["active"] is True
+        assert update("finish-source-note", "Preserve the source continuation.")["ok"] is True
+
+        # Withdrawn membership blocks fresh writes and acquisitions. The native
+        # effective=false projection is not a release of the stored lease tuple.
+        cli("configure-goal", "--goal-id", "report", "--clear-registered-agents", "--execute")
+        withdrawn = inspect()
+        assert withdrawn["active"] is False and withdrawn["lease"] == lease
+        assert withdrawn["executor_constraint"]["reason"] == "owner_not_registered"
+        for cold in (False, True):
+            if cold:
+                restart_effect_runtime(wait_for_exit=True)
+            refusal = update(f"withdrawn-write-{cold}", "Must not replace the source note.", expected_code=1)
+            assert refusal["reason_code"] == "actor_not_registered"
+            refused = cli("task-lease", "acquire", "--goal-id", "report", "--todo-id", todo,
+                "--owner", "agent-a", "--idempotency-key", f"withdrawn-acquire-{cold}",
+                "--write-scope", "report.txt", "--write-worktree", str(worktree), expected_code=1)
+            assert "owner_not_registered" in refused["error"]
+            assert inspect()["provider_revision"] == withdrawn["provider_revision"]
+        retained = cli("todo", "list", "--goal-id", "report", "--todo-id", todo)["todo"]
+        assert retained["note"] == "Preserve the source continuation."
+        assert retained["claimed_by"] == "agent-a" and retained["status"] == "open"
+
+        # Membership restoration alone restores eligibility for the unreleased
+        # tuple. A transfer cannot use temporary withdrawal as its final fence.
+        cli("configure-goal", "--goal-id", "report", "--registered-agent", "agent-a", "--execute")
+        assert inspect()["active"] is True and inspect()["lease"] == lease
+        cli("configure-goal", "--goal-id", "report", "--clear-registered-agents", "--execute")
+
+        # Exact cleanup remains possible after withdrawal. Re-registering an
+        # actor for recovery must not turn the released original proof live.
+        cli("task-lease", "release", "--goal-id", "report", "--todo-id", todo, "--owner", "agent-a",
+            "--idempotency-key", lease["idempotency_key"], "--expected-version", str(lease["version"]))
+        assert inspect()["lease"]["status"] == "released"
+        cli("configure-goal", "--goal-id", "report", "--registered-agent", "agent-a", "--execute")
+        released = inspect()
+        assert released["active"] is False and released["lease"]["status"] == "released"
+        assert update("released-proof-write", "Must not revive released work.", expected_code=1)["ok"] is False
+        assert inspect()["provider_revision"] == released["provider_revision"]
+        assert cli("todo", "list", "--goal-id", "report", "--todo-id", todo)["todo"]["note"] == retained["note"]
+    finally:
+        restart_effect_runtime(wait_for_exit=True)