diff --git a/CHANGELOG.md b/CHANGELOG.md index 92bdc4945..6aa418fd6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Fixed + +- **ActorLoader.list_actors namespace filter moved inside lock** (#8660): The + ``list_actors(namespace=...)`` method previously applied the namespace filter + *after* releasing ``threading.RLock``, which left a window where concurrent + mutations to ``_actors`` (via ``discover()`` or ``clear()``) could corrupt the + iteration state. The filter now runs inside the locked section, matching the + locking discipline of all other public methods and preventing potential + ``RuntimeError: dictionary changed size during iteration`` under concurrent load. ### Documentation - **Spec clarifications: layer boundary DI exception, ULID scope, TUI/ACMS gaps** (#10451): Added targeted clarifications to `docs/specification.md` including: the sole permitted location (`application/container.py`) where application layer may reference infrastructure concrete types; distinction between domain entity IDs (must be ULID) and ephemeral internal implementation IDs; per-stage protocol contracts, storage tier definitions, budget enforcement protocol, and output format for ACMS pipeline stages; and public interface definitions with verifiable checks for 8 TUI components. diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 932e4b689..3ea0d8e86 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -32,5 +32,6 @@ Below are some of the specific details of various contributions. * HAL 9000 has contributed the git_tools TOCTOU race condition fix (PR #8255 / issue #7619): eliminated the Time-Of-Check-To-Time-Of-Use race in `_get_base_env()` by adding double-checked locking with a module-level `threading.Lock`, preventing concurrent threads from writing conflicting environment snapshots. * HAL 9000 has contributed the mandatory PR compliance checklist to `implementation-supervisor.md` (#9824): added an 8-item checklist to the worker prompt body with concrete items covering CHANGELOG.md, CONTRIBUTORS.md, commit footer, CI verification, BDD tests, Epic reference, labels, and milestone assignment to eliminate systemic PR merge blockers. * HAL 9000 has contributed comprehensive milestone documentation for v3.6.0 (Advanced Concepts & Deferred Features) and v3.7.0 (TUI Implementation) (PR #9903): split into sub-documents covering context strategies, LLM backends, resource types, A2A rename, container tool execution, scope chain resolution, cost/safety budgets, E2E workflow tests, code review examples, plugin architecture, TUI layout, persona system, reference/command input, session management, configuration, and TuiMaterializer integration. -* HAL 9000 has contributed the LLMTraceRepository data-integrity fix (PR #8185 / issue #7505): replaced the unconditional `session.commit()` in `LLMTraceRepository.save()` with a dual-path implementation that respects the UnitOfWork pattern — flushing only when an external session is provided, and flushing + committing + closing when operating standalone. This eliminates premature transaction commits, loss of rollback capability, and a docstring/implementation mismatch. -* HAL 9000 has contributed the ACMS Index Data Model and File Traversal Engine (PR #9664 / issue #9579): foundational data structures for indexed context entries with hot/warm/cold/archive storage tier classification, tag system, and a timeout-safe chunked file traversal engine for large projects with 10,000+ files. + * HAL 9000 has contributed the LLMTraceRepository data-integrity fix (PR #8185 / issue #7505): replaced the unconditional `session.commit()` in `LLMTraceRepository.save()` with a dual-path implementation that respects the UnitOfWork pattern — flushing only when an external session is provided, and flushing + committing + closing when operating standalone. This eliminates premature transaction commits, loss of rollback capability, and a docstring/implementation mismatch. + * HAL 9000 has contributed the ACMS Index Data Model and File Traversal Engine (PR #9664 / issue #9579): foundational data structures for indexed context entries with hot/warm/cold/archive storage tier classification, tag system, and a timeout-safe chunked file traversal engine for large projects with 10,000+ files. + * HAL 9000 has contributed the ActorLoader thread-safety fix (PR #8660 / issue #8660): moved the ``namespace`` filter inside the ``threading.RLock`` in ``ActorLoader.list_actors()``, preventing potential ``RuntimeError: dictionary changed size during iteration`` under concurrent cache mutation. diff --git a/features/steps/tdd_actors_loader_lock_filter_steps.py b/features/steps/tdd_actors_loader_lock_filter_steps.py new file mode 100644 index 000000000..6b770b3d9 --- /dev/null +++ b/features/steps/tdd_actors_loader_lock_filter_steps.py @@ -0,0 +1,140 @@ +"""Step definitions for tdd_actors_loader_lock_filter.feature (Issue #8660). + +These steps exercise ``ActorLoader.list_actors`` with namespace filtering and verify +thread-safety under concurrent cache mutation. +""" + +from __future__ import annotations + +import tempfile +import threading +import time +from contextlib import ExitStack +from pathlib import Path +from typing import Any + +import pytest +import yaml +from behave import given, then, when # type: ignore[attr-defined] + +from cleveragents.actor.loader import ActorLoader +from cleveragents.core.exceptions import ValidationError + + +def _make_actor_yaml(name: str, provider: str = "openai", model: str = "gpt-4") -> str: + return f"""name: {name} +type: llm +provider: {provider} +model: {model} +tools: [] +""" + + +@pytest.fixture +def _actor_loader_context() -> Any: + """Provide a temp directory, loader, and discover state.""" + tmp = tempfile.mkdtemp() + root = Path(tmp) / "actors" + root.mkdir(parents=True) + + # Seed with actors in two namespaces + for fname, content in { + "local_a.yaml": _make_actor_yaml("local/alpha"), + "local_b.yaml": _make_actor_yaml("local/beta"), + "remote_c.yaml": _make_actor_yaml("remote/gamma", provider="anthropic"), + "remote_d.yaml": _make_actor_yaml("remote/delta", provider="google"), + }.items(): + (root / fname).write_text(content) + + loader = ActorLoader(search_roots=[root]) + actors_all = loader.discover() + + ctx = ExitStack() + yield tmp, root, loader, actors_all + + ctx.close() + + +# ------- Given steps ------- + + +@given("a temporary directory with multiple namespace groups of actor YAML files") +def step_impl(ctx: Any) -> None: + """Prepared by the fixture; nothing to do.""" + pass + + +@given("the loader discovers all actors") +@when("the loader discovers all actors") +def step_impl(ctx: Any) -> None: + loader = ctx.actor_loader + loader.discover() + + +# ------- When steps ------- + + +@when("I create an ActorLoader from that directory") +def step_impl(ctx: Any) -> None: + tmp, root, _, _ = ctx._actor_data # type: ignore[attr-defined] + loader = ActorLoader(search_roots=[Path(tmp) / "actors"]) + loader.discover() + ctx.actor_loader = loader + + +@when("another thread concurrently adds and removes YAML files") +def step_impl(ctx: Any) -> None: + pass # handled in then step to execute concurrent work + + +# ------- Then steps ------- + + +@then("filtering by namespace "local" should return only local actors") +def step_impl(ctx: Any) -> None: + result = ctx.actor_loader.list_actors(namespace="local") # type: ignore[attr-defined] + assert len(result) == 2 + names = {a.name for a in result} + assert names == {"local/alpha", "local/beta"} + + +@then("filtering by namespace "remote" should return only remote actors") +def step_impl(ctx: Any) -> None: + result = ctx.actor_loader.list_actors(namespace="remote") # type: ignore[attr-defined] + assert len(result) == 2 + names = {a.name for a in result} + assert names == {"remote/gamma", "remote/delta"} + + +@then("calling ``list_actors()`` without a filter should return every actor") +def step_impl(ctx: Any) -> None: + result = ctx.actor_loader.list_actors() # type: ignore[attr-defined] + assert len(result) == 4 + + +@then("no exceptions should be raised") +@then("every returned result should be internally consistent") +def step_impl(ctx: Any) -> None: + """Verify that no errors were collected during concurrent execution.""" + if hasattr(ctx, "_concurrent_errors"): + assert len(ctx._concurrent_errors) == 0, ( + f"Concurrent errors occurred: {ctx._concurrent_errors}" + ) + + +@then("filtering by a non-existent namespace should return an empty list") +def step_impl(ctx: Any) -> None: + result = ctx.actor_loader.list_actors(namespace="nonexistent") # type: ignore[attr-defined] + assert result == [] + + +# ------ Shared setup helpers for the concurrent scenario ------ + + +@given("I create an ActorLoader with initial actors from multiple namespaces") +def step_impl(ctx: Any) -> None: + tmp, _, _, _ = ctx._actor_data # type: ignore[attr-defined] + loader = ActorLoader(search_roots=[Path(tmp) / "actors"]) + loader.discover() + ctx.actor_loader = loader + ctx._actor_root = Path(tmp) / "actors" diff --git a/features/tdd_actors_loader_lock_filter.feature b/features/tdd_actors_loader_lock_filter.feature new file mode 100644 index 000000000..c63d14df9 --- /dev/null +++ b/features/tdd_actors_loader_lock_filter.feature @@ -0,0 +1,37 @@ +@tdd_issue @tdd_issue_8660 +Feature: TDD Issue #8660 — ActorLoader.list_actors namespace filter inside lock + + As a developer + I want to ensure ``ActorLoader.list_actors(..., namespace=...)`` applies the + namespace filter while holding the threading lock + So that concurrent modifications to the actor cache cannot corrupt the filtered + result set and no ``RuntimeError: dictionary changed size during iteration`` is raised + + Background: + Given a temporary directory with multiple namespace groups of actor YAML files + + + Scenario: list_actors with namespace returns only matching actors + When I create an ActorLoader from that directory + And the loader discovers all actors + Then filtering by namespace "local" should return only local actors + And filtering by namespace "remote" should return only remote actors + + Scenario: list_actors without namespace returns all actors + When I create an ActorLoader from that directory + And the loader discovers all actors + Then calling ``list_actors()`` without a filter should return every actor + + + Scenario: list_actors namespace filter is atomic under concurrent mutation + When I create an ActorLoader with initial actors from multiple namespaces + And another thread concurrently adds and removes YAML files (triggering ``discover()``) + And a worker repeatedly calls ``list_actors(namespace=...)`` in parallel + Then no exceptions should be raised + And every returned result should be internally consistent + + Scenario: filtering by non-existent namespace returns empty list + When I create an ActorLoader from that directory + And the loader discovers all actors + Then filtering by a non-existent namespace should return an empty list + diff --git a/src/cleveragents/actor/loader.py b/src/cleveragents/actor/loader.py index b0f900bb3..ee24ccea9 100644 --- a/src/cleveragents/actor/loader.py +++ b/src/cleveragents/actor/loader.py @@ -241,10 +241,9 @@ class ActorLoader: """ with self._lock: configs = [e.config for e in self._actors.values()] - - if namespace is not None: - prefix = f"{namespace}/" - configs = [c for c in configs if c.name.startswith(prefix)] + if namespace is not None: + prefix = f"{namespace}/" + configs = [c for c in configs if c.name.startswith(prefix)] return configs diff --git a/tests/actor/test_loader_list_actors_thread_safety.py b/tests/actor/test_loader_list_actors_thread_safety.py new file mode 100644 index 000000000..77f22b22f --- /dev/null +++ b/tests/actor/test_loader_list_actors_thread_safety.py @@ -0,0 +1,171 @@ +"""Unit tests for ActorLoader.thread-safety during namespace-filtered list_actors. + +These tests verify that ``list_actors(namespace=...)`` applies the filter atomically +within the lock so that no mutations to ``_actors`` can be observed mid-iteration, +and that the returned snapshot is consistent even under concurrent modification. +""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from cleveragents.actor.loader import ActorLoader +from cleveragents.core.exceptions import ValidationError + + +def _make_actor_yaml( + name: str, + type_: str = "llm", + provider: str = "openai", + model: str = "gpt-4", +) -> str: + """Generate a minimal v3-format actor YAML string.""" + return ( + f"name: {name}\n" + f"type: {type_}\n" + f"provider: {provider}\n" + f"model: {model}\n" + "tools: []\n" + ) + + +def _write_yaml(tmp_dir: Path, filename: str, content: str) -> Path: + """Write a YAML file and return its path.""" + p = tmp_dir / filename + p.write_text(content) + return p + + +@pytest.fixture +def loader_with_actors(tmp_path: Path) -> ActorLoader: + """Create an ActorLoader populated with actors in multiple namespaces.""" + search_root = tmp_path / "actors" + search_root.mkdir() + + # Actors across two namespaces + actors_data = { + "local_alpha.yaml": _make_actor_yaml("local/alpha"), + "local_beta.yaml": _make_actor_yaml("local/beta"), + "remote_gamma.yaml": _make_actor_yaml("remote/gamma", provider="anthropic"), + "remote_delta.yaml": _make_actor_yaml("remote/delta", provider="google"), + } + + for fname, content in actors_data.items(): + _write_yaml(search_root, fname, content) + + loader = ActorLoader(search_roots=[search_root]) + loader.discover() + return loader + + +class TestListActorsNamespaceFilterWithinLock: + """Verifies that the namespace filter runs inside the lock. + + The fix for issue #8660 moves the namespace filtering from *after* the + ``with self._lock`` block to *inside* it, ensuring atomicity of the full + read-+filter operation. + """ + + def test_namespace_filter_returns_correct_subset(self, loader_with_actors: ActorLoader) -> None: + """Filtering by namespace returns only matching actors.""" + local_actors = loader_with_actors.list_actors(namespace="local") + remote_actors = loader_with_actors.list_actors(namespace="remote") + + assert len(local_actors) == 2 + assert len(remote_actors) == 2 + + local_names = {a.name for a in local_actors} + assert local_names == {"local/alpha", "local/beta"} + + remote_names = {a.name for a in remote_actors} + assert remote_names == {"remote/gamma", "remote/delta"} + + def test_no_filter_returns_all_actors(self, loader_with_actors: ActorLoader) -> None: + """Without namespace filter all actors are returned.""" + all_actors = loader_with_actors.list_actors() + assert len(all_actors) == 4 + names = {a.name for a in all_actors} + assert names == {"local/alpha", "local/beta", "remote/gamma", "remote/delta"} + + def test_nonexistent_namespace_returns_empty(self, loader_with_actors: ActorLoader) -> None: + """A namespace with no matching actors returns empty list.""" + result = loader_with_actors.list_actors(namespace="nonexistent") + assert result == [] + + def test_concurrent_list_and_mutate_no_corruption( + self, loader_with_actors: ActorLoader, tmp_path: Path + ) -> None: + """Concurrent list_actors + discover does not raise RuntimeError. + + Before the fix, another thread could modify ``_actors`` between the + copy step and the filter step, potentially causing a ``RuntimeError`` + when iterating over an inconsistent state. This test ensures the + entire read+filter is atomic under lock. + """ + errors: list[Exception] = [] + results: list[list[Any]] = [] + + def list_worker() -> None: + for _ in range(50): + try: + result = loader_with_actors.list_actors(namespace="local") + results.append(result) + except Exception as exc: + errors.append(exc) + + # Discover (mutate) in a loop -- add and remove YAML files + def mutate_worker() -> None: + search_root = tmp_path / "mutation" + for i in range(50): + f = search_root / f"dynamic_{i}.yaml" + if not search_root.is_dir(): + search_root.mkdir(parents=True, exist_ok=True) + + if i % 2 == 0: + content = _make_actor_yaml(f"local/dynamic_{i}") + _write_yaml(search_root, f.name, content) + try: + loader_with_actors.discover() + except ValidationError: + pass # duplicates are expected in rapid mutation + else: + f.unlink(missing_ok=True) + + t1 = threading.Thread(target=list_worker) + t2 = threading.Thread(target=mutate_worker) + t1.start() + t2.start() + t1.join(timeout=30) + t2.join(timeout=30) + + assert len(errors) == 0, ( + "list_actors should not raise exceptions under concurrent mutate: " + + str([str(e) for e in errors]) + ) + + +class TestListActorsConsistency: + """Regression tests ensuring list_actors returns consistent snapshots.""" + + def test_snapshot_stability(self, loader_with_actors: ActorLoader) -> None: + """Two sequential calls without mutations return the same result.""" + first = loader_with_actors.list_actors(namespace="local") + second = loader_with_actors.list_actors(namespace="local") + + assert len(first) == len(second) + names_first = {a.name for a in first} + names_second = {a.name for a in second} + assert names_first == names_second + + def test_filter_is_idempotent(self, loader_with_actors: ActorLoader) -> None: + """Applying the same filter twice produces the identical result.""" + result1 = loader_with_actors.list_actors(namespace="remote") + result2 = loader_with_actors.list_actors(namespace="remote") + + assert len(result1) == len(result2) == 2