test(delegation): isolate and retire fixture-owned Effect runtimes

tests/test_delegation_authority_recheck.py

@@ -11,9 +11,11 @@
 
 
 @pytest.fixture
-def sqlite_service(tmp_path, monkeypatch):
+def sqlite_service(tmp_path, request, monkeypatch):
     return delegation_service.__wrapped__(
-        tmp_path, SimpleNamespace(param="sqlite"), monkeypatch
+        tmp_path,
+        SimpleNamespace(param="sqlite", addfinalizer=request.addfinalizer),
+        monkeypatch
     )
 
 

tests/test_delegation_preflight.py

@@ -123,7 +123,10 @@ def test_structured_previews_isolate_concurrent_registry_and_workspace_facts(
 ):
     """Same public Goal name, different authorities: no cwd or decision leakage."""
     first_root, first = service
-    provider_request = SimpleNamespace(param=request.node.callspec.params["service"])
+    provider_request = SimpleNamespace(
+        param=request.node.callspec.params["service"],
+        addfinalizer=request.addfinalizer,
+    )
     second_root, second = delegation_service.__wrapped__(
         tmp_path / "second", provider_request, monkeypatch
     )

tests/test_delegation_preview_reuse.py

@@ -282,7 +282,12 @@ def read():
 def test_concurrent_services_preserve_registry_runtime_and_workspace_partition(service, tmp_path, request, monkeypatch):
     root, first = service
     second_root, second = delegation_service.__wrapped__(
-        tmp_path / "second", SimpleNamespace(param=request.node.callspec.params["delegation_service"]), monkeypatch
+        tmp_path / "second",
+        SimpleNamespace(
+            param=request.node.callspec.params["delegation_service"],
+            addfinalizer=request.addfinalizer,
+        ),
+        monkeypatch
     )
     second = reusable_service(second)
     expected = [runner.inspect("analysis") for runner in (first, second)]

tests/test_local_delegation.py

@@ -1,15 +1,18 @@
 """Production delegation/Turn/TS completion with an explicit fixture model host."""
 import json
+import os
 import asyncio
 from pathlib import Path
 import subprocess
 import sys
+import tempfile
 import time
 from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeout
 from contextlib import contextmanager
 from threading import Event, get_ident
 
 import pytest
+from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
 
@@ -47,8 +50,23 @@
 
 @pytest.fixture(params=["file", "sqlite"])
 def service(tmp_path, request, monkeypatch):
-    for name in ("TMPDIR", "TEMP", "TMP"):
-        monkeypatch.setenv(name, str(tmp_path))
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    # Each case owns a private server. Do not accumulate five-minute idle
+    # runtimes across the delegation suite, including failed setup/test cases.
+    runtime_env = os.environ.copy()
+    run_cleanup = subprocess.run  # Tests may replace the shared module before teardown.
+    def retire_runtime():
+        run_cleanup([
+            sys.executable, "-c",
+            "from pathlib import Path; import sys; "
+            "from loopx.control_plane.effect_runtime import _runtime_dir, restart_effect_runtime; "
+            "assert _runtime_dir().parent == Path(sys.argv[1]); "
+            "result = restart_effect_runtime(); "
+            "assert result['status'] in {'stopped', 'not_running'}, result",
+            str(tmp_path),
+        ], cwd=Path(__file__).resolve().parents[1], env=runtime_env,
+            capture_output=True, text=True, timeout=30, check=True)
+    request.addfinalizer(retire_runtime)
     root = tmp_path / "team"
     demo.prepare(root, provider=request.param)
     fixture(root)
@@ -71,6 +89,74 @@ def brief():
             "return_requirement": "Return the independently checked artifact"}
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_delegation_fixture_isolates_cached_and_child_runtime_routes(tmp_path, monkeypatch, provider):
+    """A warmed parent must use the same private Effect server as its CLI."""
+    from types import SimpleNamespace
+
+    from loopx.control_plane.effect_runtime import (
+        _runtime_dir, _serving_runtime_identity, restart_effect_runtime, effect_runtime_result,
+    )
+
+    cached, isolated = tmp_path / "cached", tmp_path / "isolated"
+    cached.mkdir()
+    isolated.mkdir()
+    monkeypatch.setattr(tempfile, "tempdir", str(cached))
+    finalizers = []
+    service.__wrapped__(isolated, SimpleNamespace(param=provider, addfinalizer=finalizers.append), monkeypatch)
+
+    assert _runtime_dir().parent == isolated
+    child = subprocess.check_output([
+        sys.executable, "-c",
+        "from loopx.control_plane.effect_runtime import _runtime_dir; print(_runtime_dir())",
+    ], text=True).strip()
+    assert Path(child) == _runtime_dir()
+    effect_runtime_result("runtime.ping", {})
+    assert _serving_runtime_identity() is not None
+    try:
+        # Teardown targets the captured route, even if another test changed
+        # the parent cache/environment. A neighboring runtime must survive.
+        with monkeypatch.context() as neighbor:
+            isolate_sqlite_runtime(cached, neighbor)
+            neighbor_pid = effect_runtime_result("runtime.ping", {})["pid"]
+            try:
+                with monkeypatch.context() as mocked_calls:
+                    mocked_calls.setattr(subprocess, "run", lambda *args, **kwargs:
+                                         pytest.fail("teardown must retain its original runner"))
+                    for finalize in reversed(finalizers):
+                        finalize()
+                assert effect_runtime_result("runtime.ping", {})["pid"] == neighbor_pid
+            finally:
+                restart_effect_runtime()
+        assert _serving_runtime_identity() is None, "fixture must retire its private runtime before the next case"
+    finally:
+        restart_effect_runtime()
+
+
+def test_delegation_fixture_retires_runtime_after_setup_failure(tmp_path, monkeypatch):
+    from types import SimpleNamespace
+    from loopx.control_plane.effect_runtime import (
+        _serving_runtime_identity, restart_effect_runtime, effect_runtime_result,
+    )
+
+    finalizers = []
+    def fail_after_runtime_start(*args, **kwargs):
+        effect_runtime_result("runtime.ping", {})
+        raise RuntimeError("fixture setup interrupted")
+
+    monkeypatch.setattr(demo, "prepare", fail_after_runtime_start)
+    try:
+        with pytest.raises(RuntimeError, match="fixture setup interrupted"):
+            service.__wrapped__(tmp_path, SimpleNamespace(param="file", addfinalizer=finalizers.append), monkeypatch)
+        assert _serving_runtime_identity() is not None
+        assert len(finalizers) == 1
+        for finalize in reversed(finalizers):
+            finalize()
+        assert _serving_runtime_identity() is None
+    finally:
+        restart_effect_runtime()
+
+
 @pytest.mark.parametrize("operation", ["--help", "x y", "x\ny", "x;echo", "x/../y"])
 def test_worker_rejects_unbounded_operation_arguments(tmp_path, monkeypatch, operation):
     from loopx import collaboration_mcp as delegation