diff --git a/tests/conftest.py b/tests/conftest.py
index da0bfb0b83..0cded7d70e 100644
--- a/tests/conftest.py
+++ b/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
diff --git a/tests/control_plane/test_canonical_handoff_invariants.py b/tests/control_plane/test_canonical_handoff_invariants.py
index 0f1d9602f1..1159645ece 100644
--- a/tests/control_plane/test_canonical_handoff_invariants.py
+++ b/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)
diff --git a/tests/control_plane/test_canonical_handoff_mode.py b/tests/control_plane/test_canonical_handoff_mode.py
index 1050a967d8..ddb0ea243e 100644
--- a/tests/control_plane/test_canonical_handoff_mode.py
+++ b/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
diff --git a/tests/control_plane/test_canonical_todo_writer_isolation.py b/tests/control_plane/test_canonical_todo_writer_isolation.py
index ea12517f8a..4e5cdcc270 100644
--- a/tests/control_plane/test_canonical_todo_writer_isolation.py
+++ b/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,
diff --git a/tests/control_plane/test_chat_cold_source_import.py b/tests/control_plane/test_chat_cold_source_import.py
index e5bedaa830..af755f0892 100644
--- a/tests/control_plane/test_chat_cold_source_import.py
+++ b/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"))
diff --git a/tests/control_plane/test_cold_source_import_cli.py b/tests/control_plane/test_cold_source_import_cli.py
index da84677406..99f96bf957 100644
--- a/tests/control_plane/test_cold_source_import_cli.py
+++ b/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",
diff --git a/tests/control_plane/test_quota_authority_settlement_journey.py b/tests/control_plane/test_quota_authority_settlement_journey.py
index 10aec6500e..9e720c0161 100644
--- a/tests/control_plane/test_quota_authority_settlement_journey.py
+++ b/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"
diff --git a/tests/control_plane/test_research_execution_authority.py b/tests/control_plane/test_research_execution_authority.py
index 409afe327b..5c4f61af80 100644
--- a/tests/control_plane/test_research_execution_authority.py
+++ b/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,
diff --git a/tests/control_plane/test_soft_claim_migration_continuation.py b/tests/control_plane/test_soft_claim_migration_continuation.py
index a50dcbd7e0..48c4ed7417 100644
--- a/tests/control_plane/test_soft_claim_migration_continuation.py
+++ b/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
diff --git a/tests/control_plane/test_task_lease_cli_native.py b/tests/control_plane/test_task_lease_cli_native.py
index abe849eb25..d9be522cb4 100644
--- a/tests/control_plane/test_task_lease_cli_native.py
+++ b/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)
diff --git a/tests/control_plane/test_todo_consumer_host_retirement.py b/tests/control_plane/test_todo_consumer_host_retirement.py
index 43b792f062..a80c99a903 100644
--- a/tests/control_plane/test_todo_consumer_host_retirement.py
+++ b/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):
diff --git a/tests/managed_effect_runtime_fixture.py b/tests/managed_effect_runtime_fixture.py
new file mode 100644
index 0000000000..bdd9b5aeb9
--- /dev/null
+++ b/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}"
diff --git a/tests/test_managed_effect_runtime_fixture.py b/tests/test_managed_effect_runtime_fixture.py
new file mode 100644
index 0000000000..7ef8c10023
--- /dev/null
+++ b/tests/test_managed_effect_runtime_fixture.py
@@ -0,0 +1,140 @@
+"""Disposable pytest namespaces must not retain managed runtime processes."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import textwrap
+
+import pytest
+
+from managed_effect_runtime_fixture import stop_disposable_effect_runtimes
+from loopx.control_plane import effect_runtime
+
+
+def _has_exited(pid):
+    if not effect_runtime._pid_is_alive(pid):
+        return True
+    if sys.platform == "linux":
+        # A detached child's zombie has exited but can await the container
+        # init's reap. kill(pid, 0) alone incorrectly calls it a live runtime.
+        try:
+            state = (Path("/proc") / str(pid) / "status").read_text(encoding="utf-8")
+        except FileNotFoundError:
+            return True
+        return "\nState:\tZ" in "\n" + state
+    return False
+
+
+@pytest.mark.parametrize("failing", [False, True])
+@pytest.mark.parametrize("retention", ["all", "none"])
+def test_pytest_retires_source_and_copied_package_runtimes(tmp_path, failing, retention):
+    repository = Path(__file__).resolve().parents[1]
+    records = tmp_path / "runtimes.jsonl"
+    probe = tmp_path / "test_disposable_namespace.py"
+    child_script = """
+import json, sys
+from pathlib import Path
+sys.path.insert(0, sys.argv[1])
+from loopx.control_plane import effect_runtime
+effect_runtime.effect_runtime_result("runtime.ping", {})
+locator = effect_runtime._runtime_info_path(effect_runtime._runtime_fingerprint())
+print(json.dumps({"path": str(locator), "info": json.loads(locator.read_text(encoding="utf-8"))}))
+"""
+    probe.write_text(textwrap.dedent(f'''
+        import json, os, shutil, subprocess, sys
+        from pathlib import Path
+        import pytest
+
+        @pytest.mark.parametrize("copied", [False, True])
+        def test_owned_runtime(tmp_path, copied):
+            source = Path({str(repository)!r})
+            if copied:
+                receiver = tmp_path / "receiver"
+                shutil.copytree(source / "loopx", receiver / "loopx",
+                    ignore=shutil.ignore_patterns("__pycache__", "*.pyc"))
+                source = receiver
+            script = {child_script!r}
+            environment = {{**os.environ, "TMPDIR": str(tmp_path), "TEMP": str(tmp_path),
+                "TMP": str(tmp_path), "PYTHONPATH": str(source)}}
+            child = subprocess.run([sys.executable, "-c", script, str(source)],
+                cwd=tmp_path, env=environment, capture_output=True, text=True, timeout=60)
+            assert child.returncode == 0, child.stdout + child.stderr
+            with Path({str(records)!r}).open("a", encoding="utf-8") as stream:
+                stream.write(child.stdout.strip() + "\\n")
+            assert not {failing!r}, "fixture failure retained"
+    '''), encoding="utf-8")
+    environment = {**os.environ, "PYTHONPATH": os.pathsep.join([
+        str(repository / "tests"), str(repository),
+    ])}
+    observed = []
+    try:
+        result = subprocess.run([
+            sys.executable, "-m", "pytest", "-q", "-p", "conftest",
+            "-o", f"tmp_path_retention_policy={retention}", str(probe),
+        ], cwd=tmp_path, env=environment, capture_output=True, text=True, timeout=120)
+        assert result.returncode == (1 if failing else 0), result.stdout + result.stderr
+        if failing:
+            assert "2 failed" in result.stdout and "fixture failure retained" in result.stdout
+        observed = [json.loads(line) for line in records.read_text(encoding="utf-8").splitlines()]
+        assert len(observed) == 2
+        assert all(_has_exited(row["info"]["pid"]) for row in observed), (
+            "pytest retained a source or copied-package runtime after its test namespace ended"
+        )
+    finally:
+        # Keep the failing baseline bounded too; these exact locators came from
+        # the two child fixtures above, never from a developer's runtime.
+        if records.exists() and not observed:
+            observed = [json.loads(line) for line in records.read_text(encoding="utf-8").splitlines()]
+        for row in observed:
+            effect_runtime._stop_effect_runtime(
+                Path(row["path"]), row["info"], timeout=5, wait_for_process_exit=True,
+            )
+
+
+def test_cleanup_preserves_foreign_namespace_and_rejects_wrong_pid(tmp_path, monkeypatch):
+    foreign = tmp_path / "foreign"
+    owned = tmp_path / "owned"
+    foreign.mkdir()
+    owned.mkdir()
+    with monkeypatch.context() as isolated:
+        isolated.setattr(effect_runtime.tempfile, "tempdir", str(foreign))
+        effect_runtime.effect_runtime_result("runtime.ping", {})
+        locator = effect_runtime._runtime_info_path(effect_runtime._runtime_fingerprint())
+        info = json.loads(locator.read_text(encoding="utf-8"))
+        try:
+            # A directory link must not turn an unrelated namespace into ours.
+            (owned / locator.parent.name).symlink_to(locator.parent, target_is_directory=True)
+            stop_disposable_effect_runtimes(owned)
+            assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == info["pid"]
+            (owned / locator.parent.name).unlink()
+            forged = owned / locator.parent.name / locator.name
+            forged.parent.mkdir()
+            forged.write_text(json.dumps({**info, "pid": os.getpid()}), encoding="utf-8")
+            stop_disposable_effect_runtimes(owned)
+            assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == info["pid"]
+            forged.write_text("not JSON", encoding="utf-8")
+            stop_disposable_effect_runtimes(owned)
+            forged.write_bytes(b"\xff")
+            stop_disposable_effect_runtimes(owned)
+            assert effect_runtime.effect_runtime_result("runtime.ping", {})["pid"] == info["pid"]
+        finally:
+            assert effect_runtime.restart_effect_runtime(wait_for_exit=True)["stopped"]
+
+
+def test_cleanup_does_not_hide_an_unfinished_shutdown(tmp_path, monkeypatch):
+    fingerprint = "a" * 64
+    locator = tmp_path / "loopx-effect-runtime-fixture" / f"runtime-{fingerprint[:16]}.json"
+    locator.parent.mkdir()
+    locator.write_text("{}", encoding="utf-8")
+    monkeypatch.setattr(effect_runtime, "_read_info", lambda *_a, **_k: {
+        "pid": 123, "fingerprint": fingerprint,
+    })
+    monkeypatch.setattr(effect_runtime, "_request_with_info", lambda *_a, **_k: {
+        "result": {"ready": True, "pid": 123, "fingerprint": fingerprint},
+    })
+    monkeypatch.setattr(effect_runtime, "_stop_effect_runtime", lambda *_a, **_k: {"stopped": False})
+    with pytest.raises(AssertionError, match="Disposable runtime did not exit"):
+        stop_disposable_effect_runtimes(tmp_path)
diff --git a/tests/test_projection_envelope.py b/tests/test_projection_envelope.py
index 62a309f987..6d474a3769 100644
--- a/tests/test_projection_envelope.py
+++ b/tests/test_projection_envelope.py
@@ -584,12 +584,20 @@ def test_receipt_case_collection_does_not_depend_on_the_collecting_process_clock
         "test_global_risks_discloses_unreadable_host_poll_receipts_and_recovers"
     )
     collections = []
-    for clock in ("2026-01-01T00:00:00Z", "2026-01-01T00:00:01Z"):
+    for index, clock in enumerate(("2026-01-01T00:00:00Z", "2026-01-01T00:00:01Z")):
+        # A collector must not observe routes created by its still-running
+        # parent test as writes to the collector's own default namespace.
+        home = tmp_path / f"collector-home-{index}"
+        temporary = tmp_path / f"collector-tmp-{index}"
+        home.mkdir()
+        temporary.mkdir()
         result = subprocess.run(
             [sys.executable, "-m", "pytest", target, "--collect-only", "-q",
              "-o", "addopts=", "-p", "receipt_collection_clock"],
             cwd=Path(__file__).resolve().parents[1],
             env={**os.environ, "LOOPX_RECEIPT_COLLECTION_CLOCK": clock,
+                 "HOME": str(home), "USERPROFILE": str(home),
+                 "TMPDIR": str(temporary), "TEMP": str(temporary), "TMP": str(temporary),
                  "PYTHONPATH": os.pathsep.join((str(tmp_path), os.environ.get("PYTHONPATH", "")))},
             capture_output=True, text=True, timeout=60,
         )
diff --git a/tests/test_zcode_host_diagnostics.py b/tests/test_zcode_host_diagnostics.py
index 71b92f2f80..1ec43cb9fe 100644
--- a/tests/test_zcode_host_diagnostics.py
+++ b/tests/test_zcode_host_diagnostics.py
@@ -338,13 +338,14 @@ def test_probe_deadline_survives_descendant_inheriting_stdout(monkeypatch, capsy
 
 @pytest.mark.parametrize("mode,expected,child_startup_delay", [
     ("timeout_default", "timeout", 0), ("timeout", "timeout", 0),
-    ("output_limit", "output_limit", 0), ("exit_pipe", "timeout", 0),
+    ("output_limit", "output_limit", 0), ("output_limit", "output_limit", .6),
+    ("exit_pipe", "timeout", 0),
     ("exit_closed", "observed", 0), ("exit_closed", "observed", .6),
 ])
 def test_probe_reclaims_persistent_descendants_before_return(tmp_path, monkeypatch, mode, expected, child_startup_delay):
-    # Successful closure uses the shipped deadline so interpreter and child
-    # startup are not constrained by a deadline for deliberately blocked probes.
-    if mode not in {"timeout_default", "exit_closed"}:
+    # Output and successful closure use the shipped deadline so two interpreter
+    # startups are not constrained by the synthetic timeout-case deadline.
+    if mode in {"timeout", "exit_pipe"}:
         monkeypatch.setattr(diagnostics, "PROBE_TIMEOUT_SECONDS", .5)
     # Publishing the ready PID must imply that the first pulse already exists.
     child_script = f"import time; time.sleep({child_startup_delay!r})\nimport pathlib, sys, os; p=pathlib.Path(sys.argv[1]); p.with_suffix('.pulse').write_text('0'); p.write_text(str(os.getpid())); n=0\nwhile True:\n n+=1; p.with_suffix('.pulse').write_text(str(n)); time.sleep(.01)"