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