test: align CI consumers and isolate disposable runtimes

tests/conftest.py

@@ -16,6 +16,11 @@
 
 import pytest  # noqa: E402
 
+from managed_effect_runtime_fixture import (  # noqa: E402
+    capture_disposable_effect_runtimes,
+    stop_disposable_effect_runtimes,
+)
+
 from loopx import paths  # noqa: E402
 from loopx.canary.runner import SMOKE_SUITE_CHOICES  # noqa: E402
 from loopx.semantics.production import NPM_DEV_DEPENDENCIES_MISSING  # noqa: E402
@@ -37,6 +42,20 @@
 _REAL_DEFAULT_ROUTES = (paths.DEFAULT_RUNTIME_ROOT, paths.LEGACY_RUNTIME_ROOT)
 
 
+@pytest.hookimpl(wrapper=True, tryfirst=True)
+def pytest_runtest_teardown(item):
+    temporary = item.funcargs.get("tmp_path")
+    captured = capture_disposable_effect_runtimes(temporary) if isinstance(temporary, Path) else []
+    try:
+        return (yield)
+    finally:
+        # Run after fixture finalizers and monkeypatch restoration, including
+        # failed tests. Ending a fixture must not leave its copied-source Node
+        # processes alive for the runtime's normal five-minute reuse window.
+        if isinstance(temporary, Path):
+            stop_disposable_effect_runtimes(temporary, captured)
+
+
 @pytest.fixture(scope="session", autouse=True)
 def _isolated_worker_effect_runtime(tmp_path_factory):
     # HOME does not namespace the Effect process on POSIX: its locator uses

tests/control_plane/test_canonical_handoff_invariants.py

@@ -540,17 +540,27 @@ def test_metadata_cannot_rewrite_hard_lease_execution_grants(
     _unchanged(registry, state, before, display)
 
 
-def test_metadata_edit_requires_current_execution_proof(tmp_path, canonical_workspace):
+def test_administrative_copy_preserves_explicit_execution_fence(tmp_path, canonical_workspace):
     registry, state = canonical_workspace(tmp_path, handoff_mode="hard_lease")
     todo = _add_todo(registry, claimed_by=AGENT_A)
+    original = _snapshot(registry)
+    updated = update_goal_todo(
+        registry_path=registry, goal_id=GOAL_ID, todo_id=todo["todo_id"],
+        agent_id=AGENT_A, note="Reviewed context",
+    )
+    assert updated["ok"]
+    assert _snapshot(registry)["leases"] == original["leases"]
     before, display = _snapshot(registry), state.read_bytes()
+    # Copy edits grant no execution. Supplying proof retains the strict fence.
     with pytest.raises(LocalCoordinationAuthorityRejection):
         update_goal_todo(
             registry_path=registry,
             goal_id=GOAL_ID,
             todo_id=todo["todo_id"],
             agent_id=AGENT_A,
-            note="Validated progress",
+            note="Unqualified execution",
+            task_lease_idempotency_key="not-acquired",
+            task_lease_expected_version=1,
         )
     _unchanged(registry, state, before, display)
     _acquire(registry, tmp_path, todo["todo_id"], owner=AGENT_A)

tests/control_plane/test_canonical_handoff_mode.py

@@ -186,9 +186,15 @@ def work(*args, expected_exit=0):
             *args, "--goal-id", "mode-goal", "--todo-id", todo["todo_id"]], capture_output=True, text=True, timeout=90)
         assert child.returncode == expected_exit, child.stdout + child.stderr
         return json.loads(child.stdout)
-    # Policy migration preserves assignment, but does not authorize an execution.
-    rejected = work("todo", "update", "--agent-id", "agent-a", "--text", "Continued work", expected_exit=1)
+    # Ordinary copy is administrative after migration; lease lineage stays intact.
+    assert work("todo", "update", "--agent-id", "agent-a", "--text", "Continued work")["ok"]
+    copied = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="mode-goal", include_leases=True)
+    assert copied["leases"] == after["leases"]
+    rejected = work("todo", "update", "--agent-id", "agent-a", "--note", "Unqualified execution",
+                    "--task-lease-idempotency-key", "not-acquired", "--task-lease-expected-version", "1",
+                    expected_exit=1)
     assert rejected["error_code"] == "handoff_mode_requires_lease"
+    assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id="mode-goal", include_leases=True) == copied
     acquired = work("task-lease", "acquire", "--owner", "agent-a", "--idempotency-key", "continued-execution",
                     "--expected-version", "0", "--ttl-seconds", "600", "--write-scope", "src/**")
     assert acquired["acquired"] is True

tests/control_plane/test_canonical_todo_writer_isolation.py

@@ -310,7 +310,9 @@ def guard(turn):
         return run("quota", "should-run", *identity, "--codex-app", "--turn-instance-id", turn)
 
     update = ["todo", "update", *identity, "--note", "Canonical delivery retained"]
-    rejected = run(*update, succeeds=False)
+    # Copy is administrative, while explicitly supplied execution stays fenced.
+    rejected = run(*update, "--task-lease-idempotency-key", "not-acquired",
+                   "--task-lease-expected-version", "1", succeeds=False)
     assert rejected["error_code"] == "handoff_mode_requires_lease"
     acquired = run("task-lease", "acquire", "--goal-id", settlement.GOAL_ID,
                    "--todo-id", settlement.TODO_ID, "--owner", settlement.AGENT_ID,

tests/control_plane/test_chat_cold_source_import.py

@@ -335,16 +335,24 @@ def test_registered_runtime_alias_retarget_rejects_without_import(cold_api, prov
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_app_enforces_lease_and_missing_provider_without_legacy_fallback(cold_api, provider):
+def test_app_copy_preserves_execution_and_provider_fences_without_legacy_fallback(cold_api, provider):
     call, read, state, _, runtime, _, _, _ = cold_api
     plan = prepare(call, provider)
     assert call("apply", {**carrier(plan), "writers_stopped": True})[1]["status"] == "applied"
     original_state, original_head = state.read_bytes(), read()
+    code, preview = call("/api/actions/preview", {
+        "action_kind": "todo.update", "summary": "Review an administrative edit",
+        "normalized_parameters": {"goal_id": "cold", "todo_id": "todo_current",
+            "operation": "edit", "text": "Reviewed imported work", "agent_id": "agent-a"},
+        "context": {}, "idempotency_key": "administrative-copy"})
+    assert code == 201 and preview["proposal"]["status"] == "preview_ready", preview
+    assert read() == original_head and state.read_bytes() == original_state
+    # Import and copy preview do not authorize completion of the imported work.
     code, refused = call("/api/actions/preview", {
-        "action_kind": "todo.update", "summary": "An edit without execution proof",
+        "action_kind": "todo.update", "summary": "Completion without execution proof",
         "normalized_parameters": {"goal_id": "cold", "todo_id": "todo_current",
-            "operation": "edit", "text": "Forbidden edit", "agent_id": "agent-a"},
-        "context": {}, "idempotency_key": "missing-lease"})
+            "operation": "complete", "agent_id": "agent-a", "no_followup": True},
+        "context": {}, "idempotency_key": "missing-execution"})
     assert code == 400 and refused["error_code"] == "handoff_mode_requires_lease", refused
     assert read() == original_head and state.read_bytes() == original_state
     stores = list(runtime.rglob("authority-store-*.json" if provider == "file" else "authority-*.sqlite"))

tests/control_plane/test_cold_source_import_cli.py

@@ -368,8 +368,9 @@ def listing():
         "--agent-id", "agent-a", "--note", "Work retained after migration",
         "--update-operation-id", "later-update",
         "--update-expected-provider-revision", before["authority_read"]["provider_revision"]]
-    # Importing source history grants no lease to mutate the imported work.
-    denied = run(*update, succeeds=False)
+    # Administrative copy is proof-free; import does not qualify explicit proof.
+    denied = run(*update, "--task-lease-idempotency-key", "not-acquired",
+        "--task-lease-expected-version", "1", succeeds=False)
     assert denied["error_code"] == "handoff_mode_requires_lease"
     assert listing()["todos"] == before["todos"]
     acquired = run("task-lease", "acquire", "--goal-id", "cold", "--todo-id", "todo_current",

tests/control_plane/test_quota_authority_settlement_journey.py

@@ -294,7 +294,9 @@ def run(*args):
     assert guard["heartbeat_receipt"]["settlement_identity"]["todo_id"] == cli.TODO_ID
     update = ("todo", "update", "--todo-id", cli.TODO_ID, "--agent-id", cli.AGENT_ID,
               "--text", "Continue after policy migration")
-    code, rejected = run(*update)
+    # A migrated owner may edit copy, but migration supplies no execution proof.
+    code, rejected = run(*update, "--task-lease-idempotency-key", "not-acquired",
+                         "--task-lease-expected-version", "1")
     assert code == 1 and rejected["error_code"] == "handoff_mode_requires_lease", rejected
 
     lease_key = "migrated-delivery"

tests/control_plane/test_research_execution_authority.py

@@ -242,16 +242,21 @@ def capture_native(method, params, **kwargs):
             from loopx.todos import update_goal_todo
             from loopx.capabilities.explore.composition_frontier import project_live_explore_composition_frontier
             from loopx.control_plane.work_items.semantic_replan_writeback import qualify_replan_writeback
-            from loopx.control_plane.work_items.task_lease import release_task_lease
-            with pytest.raises(LocalCoordinationAuthorityUnavailable, match="Release the active execution lease"):
+            before_pause = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal, include_leases=True)
+            with pytest.raises(LocalCoordinationAuthorityUnavailable):
                 update_goal_todo(registry_path=registry, goal_id=goal, todo_id="todo_joint", agent_id=agent,
-                    status="blocked", resume_when=f"todo_done:{blocker['todo_id']}", reason="Await the bounded dependency.")
-            released = release_task_lease(registry_path=registry, runtime_root=runtime, goal_id=goal,
-                todo_id="todo_joint", owner=agent, idempotency_key="native-research-proof",
-                expected_version=acquired["lease"]["version"])
-            assert released["ok"]
-            update_goal_todo(registry_path=registry, goal_id=goal, todo_id="todo_joint", agent_id=agent,
+                    status="blocked", resume_when=f"todo_done:{blocker['todo_id']}", reason="Await the bounded dependency.",
+                    task_lease_idempotency_key="native-research-proof",
+                    task_lease_expected_version=acquired["lease"]["version"] + 1)
+            assert read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal, include_leases=True) == before_pause
+            # The owner's proof-free pause retires its lease in the same CAS.
+            paused = update_goal_todo(registry_path=registry, goal_id=goal, todo_id="todo_joint", agent_id=agent,
                 status="blocked", resume_when=f"todo_done:{blocker['todo_id']}", reason="Await the bounded dependency.")
+            assert paused["blocked_lifecycle_transition"]["lease_retirement"] == "released"
+            retained = read_canonical_todos_if_promoted(runtime_root=runtime, goal_id=goal, include_leases=True)
+            old_execution = next(lease for lease in retained["leases"] if lease["todo_id"] == "todo_joint")
+            assert old_execution["status"] == "released"
+            assert old_execution["version"] == acquired["lease"]["version"]
             source = json.loads(registry.read_text())["goals"][0]
             def live_deferred():
                 return project_live_explore_composition_frontier(runtime_root=runtime, goal_id=goal,

tests/control_plane/test_soft_claim_migration_continuation.py

@@ -107,11 +107,16 @@ def lost_response(*args, **kwargs):
     assert {k: snapshot()["todos"][0][k] for k in preserved} == preserved
     assert cli(*migration)["status"] == "replayed" and snapshot() == current
     assert registry.read_bytes() == registry_bytes
-    # Returning to hard lease restores its execution requirement; soft copy
-    # admission did not consume, replace or silently upgrade the retained grant.
+    # Ordinary owner copy stays administrative after policy migration; explicit
+    # execution proof still cannot consume or upgrade the released grant.
     plan = registry.parent / "hard-again.json"
     preview = cli("handoff-mode", "plan-migration", "--mode", "hard_lease", "--plan", plan)
     cli("handoff-mode", "migrate", "--plan", plan, "--plan-sha256", preview["plan_sha256"], "--execute")
     hard = snapshot()
-    assert cli(*edit, ok=False)["error_code"] == "handoff_mode_requires_lease"
-    assert snapshot() == hard
+    assert cli(*edit)["status"] == "applied"
+    copied = snapshot()
+    assert copied["leases"] == hard["leases"]
+    assert copied["todos"][0]["note"] == "CLI observation"
+    assert cli(*edit, "--task-lease-idempotency-key", "previous-execution",
+        "--task-lease-expected-version", str(lease["version"]), ok=False)["error_code"] == "handoff_mode_requires_lease"
+    assert snapshot() == copied

tests/control_plane/test_task_lease_cli_native.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from copy import deepcopy
 from pathlib import Path
 
 from loopx.cli import build_parser
@@ -13,7 +14,7 @@ def _run_acquire(monkeypatch, result: dict) -> dict:
     monkeypatch.setattr(
         task_lease_cli,
         "execute_native_task_lease_acquire",
-        lambda **kwargs: result,
+        lambda **kwargs: deepcopy(result),
     )
     args = build_parser().parse_args(
         [
@@ -77,6 +78,19 @@ def raise_error(**_kwargs):
     return captured[0]
 
 
+def _assert_refusal(payload: dict, expected: dict) -> None:
+    inspection = payload["inspection"]
+    assert set(inspection) == {"cli_args", "reason", "execution_authority_granted"}
+    assert inspection["cli_args"] == [
+        f"--registry={Path('/tmp/registry.json')}", "--format", "json",
+        "task-lease", "inspect", "--goal-id=cli-native-goal",
+        "--todo-id=todo_cli_native",
+    ]
+    assert isinstance(inspection["reason"], str) and inspection["reason"].strip()
+    assert inspection["execution_authority_granted"] is False
+    assert {key: value for key, value in payload.items() if key != "inspection"} == expected
+
+
 def test_cli_envelope_keys_win_over_a_typed_exception_payload(monkeypatch) -> None:
     """A typed payload that carries its own schema_version cannot clobber the envelope's."""
 
@@ -88,14 +102,18 @@ def test_cli_envelope_keys_win_over_a_typed_exception_payload(monkeypatch) -> No
         "canonical Todo authority is unavailable",
         code="local_authority_todo_list_unavailable",
         payload={
+            "ok": True,
             "schema_version": "loopx_local_coordination_todo_list_result_v0",
+            "action": "release",
+            "error": "payload must not replace the original refusal",
+            "error_code": "payload_error",
             "status": "missing",
             "source_authority": "file_v0",
             "decision_read_from_provider": True,
             "legacy_fallback_used": False,
         },
     )
-    assert _run_acquire_raising(monkeypatch, error) == {
+    _assert_refusal(_run_acquire_raising(monkeypatch, error), {
         "ok": False,
         "schema_version": "task_lease_v0",
         "action": "acquire",
@@ -105,7 +123,7 @@ def test_cli_envelope_keys_win_over_a_typed_exception_payload(monkeypatch) -> No
         "source_authority": "file_v0",
         "decision_read_from_provider": True,
         "legacy_fallback_used": False,
-    }
+    })
 
 
 def test_cli_acquire_passes_through_native_success(monkeypatch) -> None:
@@ -177,4 +195,7 @@ def test_cli_acquire_passes_through_native_typed_failure(monkeypatch) -> None:
             },
         },
     }
-    assert _run_acquire(monkeypatch, result) == result
+    expected = deepcopy(result)
+    actual = _run_acquire(monkeypatch, result)
+    assert result == expected
+    _assert_refusal(actual, expected)

tests/control_plane/test_todo_consumer_host_retirement.py

@@ -80,8 +80,8 @@ def test_app_import_actions_and_restart_without_legacy_consumers(cold_api, provi
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_app_rejects_missing_lease_and_provider_without_legacy_consumers(cold_api, provider):
-    cold.test_app_enforces_lease_and_missing_provider_without_legacy_fallback(cold_api, provider)
+def test_app_copy_preserves_execution_and_provider_fences_without_legacy_consumers(cold_api, provider):
+    cold.test_app_copy_preserves_execution_and_provider_fences_without_legacy_fallback(cold_api, provider)
 
 
 def test_app_rejects_changed_source_and_backup_without_legacy_consumers(cold_api):

tests/managed_effect_runtime_fixture.py

@@ -0,0 +1,58 @@
+"""Retire runtimes in a finished test's disposable namespace, via their owner."""
+from __future__ import annotations
+
+from pathlib import Path
+import tempfile
+from uuid import uuid4
+
+from loopx.control_plane import effect_runtime
+
+
+def capture_disposable_effect_runtimes(root: Path) -> list[tuple[Path, str]]:
+    if root.is_symlink():
+        return []
+    owned_root = root.resolve()
+    captured = []
+    for locator in root.glob("**/loopx-effect-runtime-*/runtime-*.json"):
+        # Never follow a fixture's borrowed namespace or locator outside the
+        # test directory. Copied packages can legitimately own several revisions.
+        if locator.is_symlink() or not locator.resolve().is_relative_to(owned_root):
+            continue
+        try:
+            captured.append((locator, locator.read_text(encoding="utf-8")))
+        except (OSError, UnicodeError):
+            continue
+    return captured
+
+
+def stop_disposable_effect_runtimes(
+    root: Path, captured: list[tuple[Path, str]] | None = None,
+) -> None:
+    records = dict.fromkeys([*(captured or []), *capture_disposable_effect_runtimes(root)])
+    if not records:
+        return
+    # pytest may remove passed tmp_path directories during fixture teardown.
+    # Validate the captured bytes with the existing reader after monkeypatches
+    # are restored; the authenticated ping still verifies the exact live owner.
+    with tempfile.TemporaryDirectory(prefix="loopx-test-runtime-cleanup-") as scratch:
+        for locator, raw in records:
+            validation_file = Path(scratch) / locator.name
+            validation_file.write_text(raw, encoding="utf-8")
+            info = effect_runtime._read_info(validation_file, fingerprint=None)
+            if info is None or locator.name != f"runtime-{info['fingerprint'][:16]}.json":
+                continue
+            try:
+                response = effect_runtime._request_with_info(
+                    info, request_id=uuid4().hex, method="runtime.ping", params={}, timeout=1,
+                )
+            except (OSError, RuntimeError):
+                # Corrupt, stale or rejected metadata cannot authorize shutdown.
+                continue
+            ping = response.get("result")
+            if (not isinstance(ping, dict) or ping.get("ready") is not True or ping.get("pid") != info["pid"]
+                    or ping.get("fingerprint") != info["fingerprint"]):
+                continue
+            result = effect_runtime._stop_effect_runtime(
+                locator, info, timeout=5, wait_for_process_exit=True,
+            )
+            assert result["stopped"], f"Disposable runtime did not exit: {locator}"