diff --git a/loopx/cli_runtime.py b/loopx/cli_runtime.py index 2373deee19..9cd5ba29cf 100644 --- a/loopx/cli_runtime.py +++ b/loopx/cli_runtime.py @@ -221,14 +221,12 @@ def _build_selected_parser(command: str) -> LoopXArgumentParser: return parser -def dispatch_common_command( +def _dispatch_common_command( args: argparse.Namespace, *, registry_path: Path, allow_missing_registry: bool, ) -> int | None: - """Dispatch one selected command through the shared canonical wiring.""" - if args.command == "change-window": from .capabilities.repository_change_window.cli import handle_repository_change_window_command @@ -325,6 +323,22 @@ def dispatch_common_command( return None +def dispatch_common_command( + args: argparse.Namespace, + *, + registry_path: Path, + allow_missing_registry: bool, +) -> int | None: + from .control_plane.effect_runtime import effect_runtime_request_scope + + with effect_runtime_request_scope(): + return _dispatch_common_command( + args, + registry_path=registry_path, + allow_missing_registry=allow_missing_registry, + ) + + def _dispatch_selected(args: argparse.Namespace, raw_argv: list[str]) -> int: args.format = resolve_global_output_format(args) guard_result = enforce_native_controller_guard(args) diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py index 92524ed552..a2cc3d1cb2 100644 --- a/loopx/control_plane/effect_runtime.py +++ b/loopx/control_plane/effect_runtime.py @@ -11,9 +11,13 @@ import tempfile import time import uuid -from collections.abc import Mapping +from collections.abc import Iterator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar +from dataclasses import dataclass from functools import lru_cache from pathlib import Path +from threading import Lock from typing import IO, Any from ..file_lock import process_is_alive @@ -38,6 +42,46 @@ _RuntimeSourceSnapshot = tuple[tuple[str, int, int, int], ...] +@dataclass(frozen=True) +class _RuntimeRevision: + fingerprint: str + + +class _RequestRuntimeRevision: + def __init__(self) -> None: + self._lock = Lock() + self._revision: _RuntimeRevision | None = None + self._scope_count = 1 + self._closed = False + + def try_join(self) -> bool: + with self._lock: + if self._closed: + return False + self._scope_count += 1 + return True + + def resolve(self) -> _RuntimeRevision: + with self._lock: + if not self._closed: + if self._revision is None: + self._revision = _RuntimeRevision(_runtime_fingerprint()) + return self._revision + return _RuntimeRevision(_runtime_fingerprint()) + + def leave(self) -> None: + with self._lock: + self._scope_count -= 1 + if self._scope_count == 0: + self._closed = True + self._revision = None + + +_REQUEST_RUNTIME_REVISION: ContextVar[_RequestRuntimeRevision | None] = ( + ContextVar("loopx_effect_runtime_request_revision", default=None) +) + + class EffectRuntimeRemoteError(RuntimeError): """A typed exception returned by the managed TypeScript runtime.""" @@ -235,6 +279,33 @@ def _runtime_fingerprint() -> str: ) from exc +@contextmanager +def effect_runtime_request_scope() -> Iterator[None]: + """Pin one managed runtime revision for a logical request.""" + + current = _REQUEST_RUNTIME_REVISION.get() + if current is not None and current.try_join(): + try: + yield + finally: + current.leave() + return + state = _RequestRuntimeRevision() + token = _REQUEST_RUNTIME_REVISION.set(state) + try: + yield + finally: + state.leave() + _REQUEST_RUNTIME_REVISION.reset(token) + + +def _runtime_fingerprint_for_request() -> str: + state = _REQUEST_RUNTIME_REVISION.get() + if state is None: + return _runtime_fingerprint() + return state.resolve().fingerprint + + def _runtime_dir() -> Path: owner = str(getattr(os, "getuid", lambda: Path.home())()) suffix = hashlib.sha256(owner.encode("utf-8")).hexdigest()[:12] @@ -697,7 +768,7 @@ def effect_runtime_request( ) -> dict[str, Any]: """Call the managed TS runtime, retrying only idempotent typed effects.""" - fingerprint = _runtime_fingerprint() + fingerprint = _runtime_fingerprint_for_request() info_path = _runtime_info_path(fingerprint) request_id = str(uuid.uuid4()) last_error: OSError | RuntimeError | None = None diff --git a/loopx/status.py b/loopx/status.py index 73213c6e6d..9dec20bf9f 100644 --- a/loopx/status.py +++ b/loopx/status.py @@ -6,6 +6,7 @@ from typing import Any from .control_plane import compact_control_plane_policy +from .control_plane.effect_runtime import effect_runtime_request_scope from .control_plane.status.collection import ( StatusCollectionContext, collect_status as _collect_status_read_model, @@ -1284,20 +1285,21 @@ def collect_status( activation_state_filter: str | None = None, agent_lane_id: str | None = None, ) -> dict[str, Any]: - return _collect_status_read_model( - registry_path=registry_path, - runtime_root_override=runtime_root_override, - scan_roots=scan_roots, - limit=limit, - include_task_graph=include_task_graph, - goal_id=goal_id, - available_capabilities=available_capabilities, - include_public_boundary_scan=include_public_boundary_scan, - recent_run_limit=recent_run_limit, - include_goal_subagent_configuration=( - include_goal_subagent_configuration - ), - activation_state_filter=activation_state_filter, - agent_lane_id=agent_lane_id, - context=build_status_collection_context(), - ) + with effect_runtime_request_scope(): + return _collect_status_read_model( + registry_path=registry_path, + runtime_root_override=runtime_root_override, + scan_roots=scan_roots, + limit=limit, + include_task_graph=include_task_graph, + goal_id=goal_id, + available_capabilities=available_capabilities, + include_public_boundary_scan=include_public_boundary_scan, + recent_run_limit=recent_run_limit, + include_goal_subagent_configuration=( + include_goal_subagent_configuration + ), + activation_state_filter=activation_state_filter, + agent_lane_id=agent_lane_id, + context=build_status_collection_context(), + ) diff --git a/tests/control_plane/test_effect_runtime_request_scope.py b/tests/control_plane/test_effect_runtime_request_scope.py new file mode 100644 index 0000000000..dffbd7e468 --- /dev/null +++ b/tests/control_plane/test_effect_runtime_request_scope.py @@ -0,0 +1,294 @@ +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from contextvars import copy_context +from itertools import count +from pathlib import Path +from threading import Barrier, Event, Lock +from time import sleep + +import pytest + +import loopx.cli_runtime as cli_runtime_module +import loopx.status as status_module +from loopx.control_plane import effect_runtime + + +def _install_fingerprint_counter( + monkeypatch: pytest.MonkeyPatch, +) -> list[str]: + fingerprints: list[str] = [] + counter_lock = Lock() + + def fingerprint() -> str: + with counter_lock: + value = f"revision-{len(fingerprints) + 1}" + fingerprints.append(value) + return value + + monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", fingerprint) + return fingerprints + + +def test_scope_reuses_one_revision_and_nested_scopes_join_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fingerprints = _install_fingerprint_counter(monkeypatch) + + assert effect_runtime._runtime_fingerprint_for_request() == "revision-1" + assert effect_runtime._runtime_fingerprint_for_request() == "revision-2" + with effect_runtime.effect_runtime_request_scope(): + assert effect_runtime._runtime_fingerprint_for_request() == "revision-3" + with effect_runtime.effect_runtime_request_scope(): + assert effect_runtime._runtime_fingerprint_for_request() == "revision-3" + assert effect_runtime._runtime_fingerprint_for_request() == "revision-3" + assert effect_runtime._runtime_fingerprint_for_request() == "revision-4" + + assert fingerprints == [ + "revision-1", + "revision-2", + "revision-3", + "revision-4", + ] + + +def test_scope_retries_resolution_after_a_fingerprint_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + attempts = count(1) + + def fingerprint() -> str: + attempt = next(attempts) + if attempt == 1: + raise FileNotFoundError("source changed during the scan") + return "stable-revision" + + monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", fingerprint) + + with effect_runtime.effect_runtime_request_scope(): + with pytest.raises(FileNotFoundError): + effect_runtime._runtime_fingerprint_for_request() + assert ( + effect_runtime._runtime_fingerprint_for_request() + == "stable-revision" + ) + + assert next(attempts) == 3 + + +def test_copied_context_cannot_reuse_a_closed_request_revision( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fingerprints = _install_fingerprint_counter(monkeypatch) + + with effect_runtime.effect_runtime_request_scope(): + assert effect_runtime._runtime_fingerprint_for_request() == "revision-1" + inherited = copy_context() + + assert inherited.run( + effect_runtime._runtime_fingerprint_for_request + ) == "revision-2" + assert inherited.run( + effect_runtime._runtime_fingerprint_for_request + ) == "revision-3" + assert fingerprints == ["revision-1", "revision-2", "revision-3"] + + +def test_source_changes_become_visible_on_the_next_request( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + source = tmp_path / "runtime.ts" + source.write_text("export const revision = 1;\n", encoding="utf-8") + monkeypatch.setattr( + effect_runtime, + "_control_plane_root", + lambda: tmp_path, + ) + + with effect_runtime.effect_runtime_request_scope(): + first = effect_runtime._runtime_fingerprint_for_request() + source.write_text("export const revision = 2;\n", encoding="utf-8") + assert effect_runtime._runtime_fingerprint_for_request() == first + + with effect_runtime.effect_runtime_request_scope(): + assert effect_runtime._runtime_fingerprint_for_request() != first + + +def test_concurrent_requests_resolve_independent_revisions( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fingerprints = _install_fingerprint_counter(monkeypatch) + ready = Barrier(2) + + def resolve_request() -> tuple[str, str]: + with effect_runtime.effect_runtime_request_scope(): + ready.wait() + first = effect_runtime._runtime_fingerprint_for_request() + second = effect_runtime._runtime_fingerprint_for_request() + return first, second + + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map(lambda _index: resolve_request(), range(2))) + + assert all(first == second for first, second in results) + assert len({first for first, _second in results}) == 2 + assert fingerprints == ["revision-1", "revision-2"] + + +def test_one_request_serializes_concurrent_first_resolution( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = 0 + calls_lock = Lock() + + def fingerprint() -> str: + nonlocal calls + with calls_lock: + calls += 1 + sleep(0.01) + return "shared-revision" + + monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", fingerprint) + state = effect_runtime._RequestRuntimeRevision() + + with ThreadPoolExecutor(max_workers=4) as executor: + revisions = list(executor.map(lambda _index: state.resolve(), range(8))) + + assert {revision.fingerprint for revision in revisions} == { + "shared-revision" + } + assert calls == 1 + + +def test_copied_contexts_share_one_concurrent_first_resolution( + monkeypatch: pytest.MonkeyPatch, +) -> None: + calls = 0 + calls_lock = Lock() + ready = Barrier(2) + + def fingerprint() -> str: + nonlocal calls + with calls_lock: + calls += 1 + sleep(0.01) + return "shared-revision" + + monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", fingerprint) + + with effect_runtime.effect_runtime_request_scope(): + inherited = [copy_context(), copy_context()] + + def resolve(context_index: int) -> str: + ready.wait() + return inherited[context_index].run( + effect_runtime._runtime_fingerprint_for_request + ) + + with ThreadPoolExecutor(max_workers=2) as executor: + revisions = list(executor.map(resolve, range(2))) + + assert revisions == ["shared-revision", "shared-revision"] + assert calls == 1 + + +def test_joined_copied_context_keeps_revision_after_parent_exits( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fingerprints = _install_fingerprint_counter(monkeypatch) + joined = Event() + parent_exited = Event() + + def resolve_after_parent_exit() -> tuple[str, str]: + with effect_runtime.effect_runtime_request_scope(): + joined.set() + assert parent_exited.wait(timeout=2) + return ( + effect_runtime._runtime_fingerprint_for_request(), + effect_runtime._runtime_fingerprint_for_request(), + ) + + with ThreadPoolExecutor(max_workers=1) as executor: + with effect_runtime.effect_runtime_request_scope(): + inherited = copy_context() + future = executor.submit(inherited.run, resolve_after_parent_exit) + assert joined.wait(timeout=2) + parent_exited.set() + first, second = future.result(timeout=2) + + assert first == second == "revision-1" + assert fingerprints == ["revision-1"] + + +def test_common_command_dispatch_defines_one_effect_runtime_request( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fingerprints = _install_fingerprint_counter(monkeypatch) + observed: list[str] = [] + + def run_command( + _args: object, + *, + registry_path: Path, + allow_missing_registry: bool, + ) -> int: + assert registry_path == Path("registry.json") + assert allow_missing_registry is False + observed.append(effect_runtime._runtime_fingerprint_for_request()) + observed.append(effect_runtime._runtime_fingerprint_for_request()) + return 17 + + monkeypatch.setattr( + cli_runtime_module, + "_dispatch_common_command", + run_command, + ) + + assert cli_runtime_module.dispatch_common_command( + object(), + registry_path=Path("registry.json"), + allow_missing_registry=False, + ) == 17 + assert observed == ["revision-1", "revision-1"] + assert fingerprints == ["revision-1"] + + +def test_long_running_full_cli_does_not_pin_a_revision( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fingerprints = _install_fingerprint_counter(monkeypatch) + + def run_command(_argv: list[str]) -> int: + effect_runtime._runtime_fingerprint_for_request() + effect_runtime._runtime_fingerprint_for_request() + return 0 + + monkeypatch.setattr(cli_runtime_module, "_run_full_cli", run_command) + + assert cli_runtime_module.main(["serve-status"]) == 0 + assert fingerprints == ["revision-1", "revision-2"] + + +def test_collect_status_defines_one_programmatic_request( + monkeypatch: pytest.MonkeyPatch, +) -> None: + fingerprints = _install_fingerprint_counter(monkeypatch) + + def collect(**_kwargs: object) -> dict[str, object]: + return { + "first": effect_runtime._runtime_fingerprint_for_request(), + "second": effect_runtime._runtime_fingerprint_for_request(), + } + + monkeypatch.setattr(status_module, "_collect_status_read_model", collect) + + payload = status_module.collect_status( + registry_path=Path("registry.json"), + runtime_root_override=None, + scan_roots=[], + limit=1, + ) + + assert payload == {"first": "revision-1", "second": "revision-1"} + assert fingerprints == ["revision-1"] diff --git a/tests/control_plane/test_status_rollout_event_snapshot.py b/tests/control_plane/test_status_rollout_event_snapshot.py index 0c0f2f981f..c22c19fff4 100644 --- a/tests/control_plane/test_status_rollout_event_snapshot.py +++ b/tests/control_plane/test_status_rollout_event_snapshot.py @@ -8,6 +8,7 @@ import loopx.rollout_event_log as rollout_event_log_module import loopx.status as status_module +from loopx.control_plane import effect_runtime from loopx.control_plane.todos.todo_index import build_todo_index from loopx.rollout_event_log import ( ROLLOUT_EVENT_SCHEMA_VERSION, @@ -343,3 +344,37 @@ def test_todo_index_rejects_a_supplied_lookup_with_a_different_tail( public_safe_compact_text=status_module.public_safe_compact_text, events_for_goal=snapshot.events_for_goal, ) + + +def test_collect_status_scans_effect_runtime_sources_once_per_request( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + registry_path, runtime_root, _goal = _write_fixture(tmp_path) + original_scan = effect_runtime._scan_runtime_source_files + scanned_roots: list[Path] = [] + + def counted_scan(root: Path) -> tuple[str, ...]: + scanned_roots.append(root) + return original_scan(root) + + monkeypatch.setattr( + effect_runtime, + "_scan_runtime_source_files", + counted_scan, + ) + + for _request in range(2): + status_module.collect_status( + registry_path=registry_path, + runtime_root_override=str(runtime_root), + scan_roots=[tmp_path], + limit=20, + goal_id=GOAL_ID, + include_public_boundary_scan=False, + ) + + assert scanned_roots == [ + Path(effect_runtime.__file__).resolve().parent, + Path(effect_runtime.__file__).resolve().parent, + ]