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..45905824d --- /dev/null +++ b/features/steps/tdd_actors_loader_lock_filter_steps.py @@ -0,0 +1,206 @@ +"""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 pathlib import Path +from typing import Any + +from behave import given, then, when # type: ignore[attr-defined] + +from cleveragents.actor.loader import ActorLoader + + +def _make_actor_yaml(name: str, provider: str = "openai", model: str = "gpt-4") -> str: + return f"""name: {name} +type: llm +description: Test actor {name} +provider: {provider} +model: {model} +tools: [] +""" + + +# --------------------------------------------------------------------------- +# Fixture-like setup for the background scenario +# --------------------------------------------------------------------------- + + +def _setup_actor_dir() -> tuple[str, Path]: + """Create a temp directory seeded with actors from two namespaces.""" + tmp = tempfile.mkdtemp() + root = Path(tmp) / "actors" + root.mkdir(parents=True) + + for fname, content in { + "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"), + }.items(): + (root / fname).write_text(content) + + return tmp, root + + +# ------- Given steps ------------------------------------------------------- + + +@given("a temporary directory with multiple namespace groups of actor YAML files") +def step_given_actor_dir(context: Any) -> None: + """Set up temp directory with actors from two namespaces.""" + context._actor_tmp, context._actor_root = _setup_actor_dir() # type: ignore[attr-defined] + + +@given("the loader discovers all actors") +@when("the loader discovers all actors") +def step_when_discover_all(context: Any) -> None: + loader: ActorLoader = context.actor_loader # type: ignore[attr-defined] + loader.discover() + + +@given("I create an ActorLoader with initial actors from multiple namespaces") +@when("I create an ActorLoader with initial actors from multiple namespaces") +def step_given_loader_with_initial(context: Any) -> None: + """Set up a loader and seed it with actors.""" + loader = ActorLoader(search_roots=[context._actor_root]) # type: ignore[attr-defined] + loader.discover() + context.actor_loader = loader # type: ignore[attr-defined] + + +# ------- When steps -------------------------------------------------------- + + +@when("I create an ActorLoader from that directory") +def step_when_create_loader(context: Any) -> None: + root: Path = context._actor_root # type: ignore[attr-defined] + loader = ActorLoader(search_roots=[root]) + loader.discover() + context.actor_loader = loader # type: ignore[attr-defined] + + +@when( + "another thread concurrently adds and removes YAML files " + "(triggering ``discover()``)" +) +def step_when_concurrent_modifications(context: Any) -> None: + """Define a background thread that mutates actor files.""" + root: Path = context._actor_root # type: ignore[attr-defined] + + def _concurrent_worker() -> None: + """Repeatedly add and remove YAML files.""" + try: + for i in range(10): + tmp_fpath = root / f"conc_{i}.yaml" + tmp_fpath.write_text(_make_actor_yaml(f"conc/concurrent_{i}")) + loader = ActorLoader(search_roots=[root]) + loader.discover() + time.sleep(0.01) + if tmp_fpath.exists(): + tmp_fpath.unlink(missing_ok=True) + except Exception as exc: + errors = getattr(context, "_concurrent_errors", []) # type: ignore[attr-defined] + errors.append(exc) + context._concurrent_errors = errors # type: ignore[attr-defined] + + context._worker_thread = threading.Thread( # type: ignore[attr-defined] + target=_concurrent_worker, daemon=True + ) + + +@when("a worker repeatedly calls ``list_actors(namespace=...)`` in parallel") +def step_when_parallel_list(context: Any) -> None: + """Run parallel list_actors calls while a background thread mutates.""" + loader: ActorLoader = context.actor_loader # type: ignore[attr-defined] + + def _list_worker() -> None: + try: + for _ in range(20): + loader.list_actors(namespace="local") + loader.list_actors(namespace="remote") + time.sleep(0.01) + except Exception as exc: + errors = getattr(context, "_list_errors", []) # type: ignore[attr-defined] + errors.append(exc) + context._list_errors = errors # type: ignore[attr-defined] + + workers: list[threading.Thread] = [] + start_thread = getattr(context, "_worker_thread", None) # type: ignore[attr-defined] + + for _ in range(5): + t = threading.Thread(target=_list_worker, daemon=True) + workers.append(t) + t.start() + + if start_thread is not None: + start_thread.start() + + for t in workers: + t.join(timeout=10) + if start_thread is not None: + start_thread.join(timeout=10) + + collected_errors: list[Exception] = [ + *getattr(context, "_list_errors", []), # type: ignore[attr-defined] + *getattr(context, "_concurrent_errors", []), # type: ignore[attr-defined] + ] + context._final_concurrent_errors = collected_errors # type: ignore[attr-defined] + + +# ------- Then steps -------------------------------------------------------- + + +@then('filtering by namespace "local" should return only local actors') +def step_then_local_only(context: Any) -> None: + result = context.actor_loader.list_actors(namespace="local") # type: ignore[attr-defined] + assert len(result) == 2, ( + f"Expected 2 local actors, got {len(result)}: {[a.name for a in result]}" + ) + 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_then_remote_only(context: Any) -> None: + result = context.actor_loader.list_actors(namespace="remote") # type: ignore[attr-defined] + assert len(result) == 2, ( + f"Expected 2 remote actors, got {len(result)}: {[a.name for a in result]}" + ) + 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_then_all_actors(context: Any) -> None: + result = context.actor_loader.list_actors() # type: ignore[attr-defined] + assert len(result) == 4, f"Expected 4 total actors, got {len(result)}" + + +@then("no exceptions should be raised") +def step_then_no_errors(context: Any) -> None: + errors = getattr(context, "_final_concurrent_errors", []) # type: ignore[attr-defined] + assert len(errors) == 0, f"Concurrent errors occurred: {errors}" + + +@then("every returned result should be internally consistent") +def step_then_consistent(context: Any) -> None: + """Every thread-safe list_actors call returns a stable snapshot.""" + loader: ActorLoader = context.actor_loader # type: ignore[attr-defined] + result = loader.list_actors(namespace="local") + for actor in result: + assert hasattr(actor, "name"), f"Actor missing 'name' attribute: {actor}" + assert "/" in actor.name, ( + f"Namespaced actor name must contain '/': {actor.name}" + ) + + +@then("filtering by a non-existent namespace should return an empty list") +def step_then_empty_for_missing_ns(context: Any) -> None: + result = context.actor_loader.list_actors(namespace="nonexistent") # type: ignore[attr-defined] + assert result == [] diff --git a/features/tdd_actors_loader_lock_filter.feature b/features/tdd_actors_loader_lock_filter.feature new file mode 100644 index 000000000..6c7c63598 --- /dev/null +++ b/features/tdd_actors_loader_lock_filter.feature @@ -0,0 +1,36 @@ +@tdd_issue @tdd_issue_8588 +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 thread-safe 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/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..c2dc5d6c5 --- /dev/null +++ b/tests/actor/test_loader_list_actors_thread_safety.py @@ -0,0 +1,177 @@ +"""Thread-safety regression test for ActorLoader.list_actors (Issue #8660). + +Verifies that ``list_actors(namespace=...)`` applies the namespace filter +while holding ``threading.RLock``, preventing +``RuntimeError: dictionary changed size during iteration`` when other threads +call ``discover()`` or ``clear()`` concurrently. +""" + +from __future__ import annotations + +import tempfile +import threading +import time +from pathlib import Path + +from cleveragents.actor.loader import ActorLoader + + +def _make_actor_yaml(name: str, provider: str = "openai", model: str = "gpt-4") -> str: + return f"""name: {name} +type: llm +description: Test actor {name} +provider: {provider} +model: {model} +tools: [] +""" + + +class TestListActorsThreadSafety: + """Test thread-safety of ActorLoader.list_actors with namespace filter.""" + + def _setup_loader(self) -> tuple[ActorLoader, Path]: + """Create loader seeded with actors from multiple namespaces.""" + tmp = tempfile.mkdtemp() + root = Path(tmp) / "actors" + root.mkdir(parents=True) + + 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]) + loader.discover() + return loader, root + + def test_namespace_filter_thread_safe_under_concurrent_discover( + self, + ) -> None: + """list_actors(namespace=...) is safe while discover() mutates cache.""" + loader, root = self._setup_loader() + errors: list[Exception] = [] + + def _concurrent_discover() -> None: + """Worker that repeatedly discovers actors.""" + try: + for i in range(20): + extra_file = root / f"extra_{i}.yaml" + extra_file.write_text(_make_actor_yaml(f"conc/thread_{i}")) + loader.discover() + time.sleep(0.01) + if extra_file.exists(): + extra_file.unlink(missing_ok=True) + except Exception as exc: + errors.append(exc) + + def _list_worker() -> None: + """Worker that repeatedly calls list_actors with namespace filter.""" + try: + for _ in range(30): + local = loader.list_actors(namespace="local") + remote = loader.list_actors(namespace="remote") + # Each filtered result should only contain matching actors + for actor in local: + assert "/" in actor.name and actor.name.startswith("local/") + for actor in remote: + assert "/" in actor.name and actor.name.startswith("remote/") + time.sleep(0.01) + except Exception as exc: + errors.append(exc) + + # Run concurrent workers + threads = [ + threading.Thread(target=_concurrent_discover, daemon=True), + *[threading.Thread(target=_list_worker, daemon=True) for _ in range(5)], + ] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + assert not errors, f"Concurrent access raised {len(errors)} errors" + + def test_namespace_filter_thread_safe_under_concurrent_clear( + self, + ) -> None: + """list_actors(namespace=...) is safe while clear() empties cache.""" + loader, _ = self._setup_loader() + errors: list[Exception] = [] + + def _clear_worker() -> None: + try: + for _ in range(10): + loader.clear() + time.sleep(0.02) + loader.discover() + time.sleep(0.01) + except Exception as exc: + errors.append(exc) + + def _list_worker() -> None: + try: + for _ in range(30): + loader.list_actors() + loader.list_actors(namespace="local") + loader.list_actors(namespace="remote") + time.sleep(0.01) + except Exception as exc: + errors.append(exc) + + threads = [ + threading.Thread(target=_clear_worker, daemon=True), + *[threading.Thread(target=_list_worker, daemon=True) for _ in range(5)], + ] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + assert not errors, f"Concurrent clear raised {len(errors)} errors" + + def test_list_actors_returns_consistent_snapshot_no_namespace(self) -> None: + """list_actors() without namespace is atomic under concurrent mutation.""" + loader, root = self._setup_loader() + errors: list[Exception] = [] + + def _add_actor() -> None: + try: + for i in range(20): + (root / f"new_{i}.yaml").write_text( + _make_actor_yaml(f"conc/new_{i}") + ) + loader.discover() + time.sleep(0.01) + if (root / f"new_{i}.yaml").exists(): + (root / f"new_{i}.yaml").unlink(missing_ok=True) + except Exception as exc: + errors.append(exc) + + def _list_worker() -> None: + try: + for _ in range(30): + all_actors = loader.list_actors() + assert isinstance(all_actors, list) + time.sleep(0.01) + except Exception as exc: + errors.append(exc) + + threads = [ + threading.Thread(target=_add_actor, daemon=True), + *[threading.Thread(target=_list_worker, daemon=True) for _ in range(3)], + ] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + assert not errors, f"Concurrent mutation raised {len(errors)} errors" + + +if __name__ == "__main__": # pragma: no cover + import pytest + + pytest.main([__file__, "-v"])