diff --git a/tests/test_delegation_authority_recheck.py b/tests/test_delegation_authority_recheck.py index 879e5a4992..2c49d24b0d 100644 --- a/tests/test_delegation_authority_recheck.py +++ b/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 ) diff --git a/tests/test_delegation_preflight.py b/tests/test_delegation_preflight.py index 3be137f435..c24286b183 100644 --- a/tests/test_delegation_preflight.py +++ b/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 ) diff --git a/tests/test_delegation_preview_reuse.py b/tests/test_delegation_preview_reuse.py index 6543ea67e8..d83ec65152 100644 --- a/tests/test_delegation_preview_reuse.py +++ b/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)] diff --git a/tests/test_local_delegation.py b/tests/test_local_delegation.py index 45c57a7948..957597b9c2 100644 --- a/tests/test_local_delegation.py +++ b/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