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 == []

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

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"])