"""Test tool metadata JSON snapshot for dashboard consumption."""

import gc
import json
import sys
from dataclasses import replace
from pathlib import Path
from types import ModuleType, SimpleNamespace
from typing import Never
from unittest.mock import AsyncMock

import agno.tools.crawl4ai as agno_crawl4ai
import pytest
from agno.tools import Toolkit

import mindroom.tool_system.metadata as metadata_module

# Import tools to trigger tool registration
import mindroom.tools  # noqa: F401
import mindroom.tools.custom_api as custom_api_module
from mindroom.config.main import Config, ConfigRuntimeValidationError, load_config
from mindroom.constants import resolve_runtime_paths
from mindroom.redaction import REDACTED
from mindroom.server_fetch_url import ServerFetchUrlError
from mindroom.tool_system.bootstrap import ensure_tool_registry_loaded
from mindroom.tool_system.metadata import (
    _AUTHORED_OVERRIDE_INHERIT,
    ConfigField,
    ToolAuthoredOverrideValidator,
    ToolCategory,
    ToolConfigOverrideError,
    ToolManagedInitArg,
    _execute_validation_plugin_module,
    _validate_authored_overrides,
    deserialize_tool_validation_snapshot,
    export_tools_metadata,
    get_tool_by_name,
    resolved_tool_validation_snapshot_for_runtime,
    serialize_tool_validation_snapshot,
)
from mindroom.tool_system.registration import register_tool_with_metadata
from mindroom.tool_system.registry_state import (
    BUILTIN_TOOL_METADATA,
    BUILTIN_TOOL_REGISTRY,
    PLUGIN_MODULE_PREFIX,
    TOOL_METADATA,
    TOOL_REGISTRY,
    capture_tool_registry_snapshot,
    reconcile_dynamic_tool_state,
    restore_tool_registry_snapshot,
)
from mindroom.tool_system.worker_routing import (
    ToolExecutionIdentity,
    resolve_worker_target,
)
from mindroom.tools.crawl4ai import crawl4ai_tools
from mindroom.tools.custom_api import custom_api_tools

_BASE_TOOL_REGISTRY = TOOL_REGISTRY.copy()
_BASE_TOOL_METADATA = TOOL_METADATA.copy()


def _restore_builtin_tool_metadata_state() -> None:
    """Reset tool registries to the built-in metadata snapshot."""
    TOOL_REGISTRY.clear()
    TOOL_REGISTRY.update(_BASE_TOOL_REGISTRY)
    TOOL_METADATA.clear()
    TOOL_METADATA.update(_BASE_TOOL_METADATA)


def _clear_module_origin_caches() -> None:
    """Clear module-origin caches between tests."""
    metadata_module._resolved_module_file.cache_clear()
    metadata_module._module_file_within_root.cache_clear()


def test_reconcile_dynamic_tool_state_replaces_only_owned_entries() -> None:
    """Dynamic registry reconciliation should preserve unrelated tools and remove stale owned entries."""
    factory = TOOL_REGISTRY["shell"]
    replacement_factory = TOOL_REGISTRY["python"]
    unrelated_metadata = replace(TOOL_METADATA["shell"], name="unrelated")
    stale_metadata = replace(TOOL_METADATA["shell"], name="stale_owned")
    desired_metadata = replace(TOOL_METADATA["python"], name="owned")
    metadata_only = replace(TOOL_METADATA["shell"], name="metadata_only")
    current_registry = {
        "unrelated": factory,
        "stale_owned": factory,
        "owned": factory,
        "metadata_only": factory,
    }
    current_metadata = {
        "unrelated": unrelated_metadata,
        "stale_owned": stale_metadata,
        "owned": stale_metadata,
        "metadata_only": stale_metadata,
    }

    reconciled_tool_names = reconcile_dynamic_tool_state(
        current_registry,
        current_metadata,
        {"owned": replacement_factory},
        {"owned": desired_metadata, "metadata_only": metadata_only},
        owned_tool_names={"stale_owned", "owned", "metadata_only"},
        collision_error=ValueError,
    )

    assert reconciled_tool_names == {"owned", "metadata_only"}
    assert current_registry == {"unrelated": factory, "owned": replacement_factory}
    assert current_metadata == {
        "unrelated": unrelated_metadata,
        "owned": desired_metadata,
        "metadata_only": metadata_only,
    }


def test_reconcile_dynamic_tool_state_rejects_non_owned_collisions() -> None:
    """Desired dynamic entries must not overwrite tools outside the owned namespace."""
    factory = TOOL_REGISTRY["shell"]
    metadata = replace(TOOL_METADATA["shell"], name="shared")

    with pytest.raises(ValueError, match="shared collision"):
        reconcile_dynamic_tool_state(
            {"shared": factory},
            {"shared": metadata},
            {"shared": factory},
            {"shared": metadata},
            owned_tool_names=set(),
            collision_error=lambda tool_name: ValueError(f"{tool_name} collision"),
        )


def test_export_tools_metadata_json() -> None:
    """Verify committed tool metadata JSON matches the current registry export."""
    output_path = Path(__file__).parent.parent / "src/mindroom/tools_metadata.json"
    _restore_builtin_tool_metadata_state()

    committed_content = output_path.read_text(encoding="utf-8")
    tools = export_tools_metadata()
    expected_content = json.dumps({"tools": tools}, indent=2, sort_keys=True) + "\n"

    assert committed_content == expected_content, (
        "tools_metadata.json is out of date, regenerate it with "
        './.venv/bin/python -c "import json; import mindroom.tools; '
        "from pathlib import Path; "
        "from mindroom.tool_system.metadata import export_tools_metadata; "
        "Path('src/mindroom/tools_metadata.json').write_text("
        "json.dumps({'tools': export_tools_metadata()}, indent=2, sort_keys=True) + '\\n', "
        "encoding='utf-8')\""
    )

    with output_path.open(encoding="utf-8") as f:
        data = json.load(f)
        assert "tools" in data
        assert len(data["tools"]) > 0

        # Verify structure of first tool
        first_tool = data["tools"][0]
        required_fields = ["name", "display_name", "description", "category", "status", "setup_type"]
        for field in required_fields:
            assert field in first_tool, f"Missing required field: {field}"
        assert "managed_init_args" not in first_tool


def test_export_tools_metadata_json_resets_leaked_registry_entries() -> None:
    """Export should ignore temporary registry contamination from earlier tests."""
    tool_name = "test_leaked_tool"

    class LeakedTool(Toolkit):
        def __init__(self) -> None:
            super().__init__(name="leaked", tools=[])

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Leaked Tool",
        description="Temporary leaked tool metadata",
        category=ToolCategory.DEVELOPMENT,
    )
    def leaked_tool_factory() -> type[Toolkit]:
        return LeakedTool

    try:
        assert tool_name in TOOL_METADATA

        _restore_builtin_tool_metadata_state()

        exported_names = {tool["name"] for tool in export_tools_metadata()}
        assert tool_name not in exported_names
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)
        _restore_builtin_tool_metadata_state()


def test_homeassistant_private_url_metadata_defaults_to_false() -> None:
    """The private Home Assistant URL opt-in should expose an explicit secure default."""
    _restore_builtin_tool_metadata_state()
    homeassistant = TOOL_METADATA["homeassistant"]
    fields = {field.name: field for field in homeassistant.config_fields or []}

    assert fields["HOMEASSISTANT_ALLOW_PRIVATE_URL"].default is False


@pytest.mark.parametrize(
    ("base_url", "endpoint", "reason"),
    [
        (None, "http://127.0.0.1:8000/admin", "private_address"),
        ("http://169.254.169.254", "/latest/meta-data/", "metadata_address"),
    ],
)
def test_custom_api_tool_rejects_unsafe_url_before_request(
    base_url: str | None,
    endpoint: str,
    reason: str,
) -> None:
    """Custom API calls should validate final URLs before making requests."""
    tool = custom_api_tools()(base_url=base_url)

    with pytest.raises(ServerFetchUrlError) as exc_info:
        tool.make_request(endpoint)

    assert exc_info.value.reason == reason


def test_custom_api_tool_filters_sensitive_response_headers(monkeypatch: pytest.MonkeyPatch) -> None:
    """Custom API output should keep safe response headers without exposing credentials."""

    class FakeResponse:
        status_code = 200
        text = "{}"
        is_success = True

        def __init__(self) -> None:
            self.headers = {
                "content-type": "application/json",
                "x-request-id": "req-123",
                "set-cookie": "session=secret",
                "authorization": "Bearer secret",
                "proxy-authorization": "Basic secret",
                "cookie": "session=secret",
                "www-authenticate": "Bearer challenge",
                "authentication-info": "nextnonce=secret",
                "x-api-key": "secret",
                "x-auth-token": "secret",
                "x-api-token": "secret",
                "api-token": "secret",
                "x-token": "secret",
                "token": "secret",
                "x-amz-security-token": "secret",
                "x_api_token": "secret",
                "x-ratelimit-remaining-tokens": "99",
                "x-total-tokens": "100",
            }

        def json(self) -> dict[str, str]:
            return {"ok": "true"}

    class FakeClient:
        def __init__(self, *_args: object, **_kwargs: object) -> None:
            pass

        def __enter__(self) -> object:
            return self

        def __exit__(self, *_args: object) -> None:
            pass

        def request(self, **_kwargs: object) -> FakeResponse:
            return FakeResponse()

    monkeypatch.setattr(custom_api_module, "validate_server_fetch_url", lambda url: url)
    monkeypatch.setattr(custom_api_module.httpx, "Client", FakeClient)

    tool = custom_api_tools()()

    payload = json.loads(tool.make_request("https://example.com/data"))

    assert payload["headers"] == {
        "content-type": "application/json",
        "x-request-id": "req-123",
        "set-cookie": REDACTED,
        "authorization": REDACTED,
        "proxy-authorization": REDACTED,
        "cookie": REDACTED,
        "www-authenticate": REDACTED,
        "authentication-info": REDACTED,
        "x-api-key": REDACTED,
        "x-auth-token": REDACTED,
        "x-api-token": REDACTED,
        "api-token": REDACTED,
        "x-token": REDACTED,
        "token": REDACTED,
        "x-amz-security-token": REDACTED,
        "x_api_token": REDACTED,
        "x-ratelimit-remaining-tokens": "99",
        "x-total-tokens": "100",
    }


def test_crawl4ai_tool_rejects_private_url_before_crawl(monkeypatch: pytest.MonkeyPatch) -> None:
    """Crawl4AI should reject unsafe URLs before starting browser-backed crawling."""

    async def forbidden_crawl(*_args: object, **_kwargs: object) -> str:
        msg = "unsafe crawl4ai URL should be rejected before crawling starts"
        raise AssertionError(msg)

    tool = crawl4ai_tools()()
    monkeypatch.setattr(tool, "_async_crawl", forbidden_crawl)

    with pytest.raises(ServerFetchUrlError) as exc_info:
        tool.crawl("http://127.0.0.1:8000/private")

    assert exc_info.value.reason == "private_address"


@pytest.mark.asyncio
async def test_crawl4ai_tool_installs_server_fetch_route_guard(monkeypatch: pytest.MonkeyPatch) -> None:
    """Crawl4AI should install a Playwright route guard before crawling."""
    installed_hook = None

    class FakeCrawlerStrategy:
        def set_hook(self, name: str, hook: object) -> None:
            nonlocal installed_hook
            assert name == "on_page_context_created"
            installed_hook = hook

    class FakeAsyncWebCrawler:
        def __init__(self, *, config: object) -> None:
            del config
            self.crawler_strategy = FakeCrawlerStrategy()

        async def __aenter__(self) -> object:
            return self

        async def __aexit__(self, *_args: object) -> None:
            return None

        async def arun(self, *, url: str, config: object) -> object:
            del config
            assert url == "https://example.com"
            assert installed_hook is not None
            page = SimpleNamespace(route=AsyncMock())
            await installed_hook(page)
            route_handler = page.route.await_args.args[1]
            unsafe_route = SimpleNamespace(
                request=SimpleNamespace(url="http://127.0.0.1/admin"),
                abort=AsyncMock(),
                continue_=AsyncMock(),
            )
            await route_handler(unsafe_route)
            unsafe_route.abort.assert_awaited_once_with("blockedbyclient")
            unsafe_route.continue_.assert_not_called()
            return SimpleNamespace(fit_markdown="", markdown="", text="public content", html="", success=True)

    monkeypatch.setattr(agno_crawl4ai, "AsyncWebCrawler", FakeAsyncWebCrawler)
    tool = crawl4ai_tools()()

    result = await tool._async_crawl("https://example.com")

    assert result == "public content"


def test_plugin_validation_uses_sys_modules_snapshot(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    """Plugin validation should snapshot sys.modules before iterating over it."""

    class SnapshotOnlyModules(dict[str, ModuleType]):
        def items(self) -> Never:
            msg = "live sys.modules.items() should not be used"
            raise RuntimeError(msg)

        def copy(self) -> dict[str, ModuleType]:
            return dict(self)

    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    module_path = plugin_root / "tools.py"
    module_path.write_text("VALUE = 1\n", encoding="utf-8")

    snapshot_modules = SnapshotOnlyModules(sys.modules.copy())
    monkeypatch.setattr(metadata_module.sys, "modules", snapshot_modules)

    snapshot = capture_tool_registry_snapshot()
    assert isinstance(snapshot.plugin_modules, dict)

    module_name = _execute_validation_plugin_module("demo", plugin_root, module_path, {})
    assert module_name
    assert "demo" in module_name
    assert "__validation__" in module_name


def test_module_origin_within_root_caches_path_resolution(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Repeated origin checks for one module file should not re-resolve the path."""
    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    module_path = plugin_root / "helper.py"
    module_path.write_text("VALUE = 1\n", encoding="utf-8")
    module = ModuleType("demo.helper")
    module.__file__ = str(module_path)
    resolve_calls = 0
    original_resolve = Path.resolve

    def counted_resolve(self: Path, *args: object, **kwargs: object) -> Path:
        nonlocal resolve_calls
        if self == Path(str(module_path)):
            resolve_calls += 1
        return original_resolve(self, *args, **kwargs)

    _clear_module_origin_caches()
    monkeypatch.setattr(metadata_module.Path, "resolve", counted_resolve)
    try:
        assert metadata_module._module_origin_within_root(module, plugin_root)
        assert metadata_module._module_origin_within_root(module, plugin_root)
    finally:
        _clear_module_origin_caches()

    assert resolve_calls == 1


def test_module_origin_within_root_caches_containment_checks(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Plugin validation rescans all of sys.modules, so containment must resolve once."""
    plugin_root = tmp_path / "plugins" / "demo"
    plugin_root.mkdir(parents=True)
    module_path = plugin_root / "helper.py"
    module_path.write_text("VALUE = 1\n", encoding="utf-8")
    module = ModuleType("demo.helper")
    module.__file__ = str(module_path)
    outside_module = ModuleType("outside.helper")
    outside_module.__file__ = str(tmp_path / "outside.py")
    containment_calls = 0
    original_is_relative_to = Path.is_relative_to

    def counted_is_relative_to(self: Path, *args: object, **kwargs: object) -> bool:
        nonlocal containment_calls
        containment_calls += 1
        return original_is_relative_to(self, *args, **kwargs)

    _clear_module_origin_caches()
    monkeypatch.setattr(metadata_module.Path, "is_relative_to", counted_is_relative_to)
    try:
        for _ in range(5):
            assert metadata_module._module_origin_within_root(module, plugin_root)
            assert not metadata_module._module_origin_within_root(outside_module, plugin_root)
    finally:
        _clear_module_origin_caches()

    assert containment_calls == 2


def test_restore_tool_registry_snapshot_uses_sys_modules_snapshot(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Restoring registry state should snapshot sys.modules before iterating over it."""

    class SnapshotOnlyModules(dict[str, ModuleType]):
        def __iter__(self) -> Never:
            msg = "live sys.modules iteration should not be used"
            raise RuntimeError(msg)

        def copy(self) -> dict[str, ModuleType]:
            return dict(self)

    snapshot = capture_tool_registry_snapshot()
    leaked_module_name = f"{PLUGIN_MODULE_PREFIX}leaked"
    assert leaked_module_name not in snapshot.plugin_modules

    leaked_module = ModuleType(leaked_module_name)
    leaked_module.__file__ = str(tmp_path / "leaked.py")

    snapshot_modules = SnapshotOnlyModules(sys.modules.copy())
    snapshot_modules[leaked_module_name] = leaked_module
    monkeypatch.setattr(metadata_module.sys, "modules", snapshot_modules)

    restore_tool_registry_snapshot(snapshot)

    assert leaked_module_name not in metadata_module.sys.modules


def test_tool_metadata_consistency() -> None:
    """Verify that all tool metadata is properly configured."""
    for tool_name, metadata in TOOL_METADATA.items():
        # Check that all required fields are present
        assert metadata.name == tool_name, f"Tool name mismatch: {tool_name} != {metadata.name}"
        assert metadata.display_name, f"Tool {tool_name} missing display_name"
        assert metadata.description, f"Tool {tool_name} missing description"
        assert metadata.category, f"Tool {tool_name} missing category"
        assert metadata.status, f"Tool {tool_name} missing status"
        assert metadata.setup_type, f"Tool {tool_name} missing setup_type"
        if tool_name not in TOOL_REGISTRY:
            assert metadata.managed_init_args == (), (
                f"{tool_name} is metadata-only and should not declare managed init args: "
                f"{[managed_arg.value for managed_arg in metadata.managed_init_args]}"
            )


def test_dynamic_tools_is_durable_metadata_only_builtin(tmp_path: Path) -> None:
    """Dynamic tools metadata should survive runtime registry rebuilding without a factory."""
    metadata = TOOL_METADATA["dynamic_tools"]

    assert BUILTIN_TOOL_METADATA["dynamic_tools"] == metadata
    assert "dynamic_tools" not in BUILTIN_TOOL_REGISTRY
    assert "dynamic_tools" not in TOOL_REGISTRY

    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "mindroom_data",
        process_env={},
    )

    assert metadata_module.resolved_tool_metadata_for_runtime(runtime_paths, Config())["dynamic_tools"] == metadata


def test_tool_metadata_does_not_advertise_env_var_fallbacks() -> None:
    """Tool metadata should describe explicit config, not resurrect env fallback docs."""
    forbidden_phrases = (
        "falls back to",
        "can also be set via",
    )

    for tool_name, metadata in TOOL_METADATA.items():
        text_snippets = [metadata.description, metadata.helper_text]
        text_snippets.extend(field.description for field in metadata.config_fields or [])

        for text in filter(None, text_snippets):
            lowered = text.lower()
            assert not any(phrase in lowered for phrase in forbidden_phrases), (
                f"Tool metadata for {tool_name} still advertises env fallback: {text}"
            )


def test_get_tool_by_name_does_not_infer_hidden_constructor_kwargs(tmp_path: Path) -> None:
    """Undeclared MindRoom-managed kwargs should not be inferred from parameter names."""
    tool_name = "test_hidden_runtime_tool"

    class HiddenRuntimeToolkit(Toolkit):
        def __init__(self, *, runtime_paths: object) -> None:
            self.runtime_paths = runtime_paths
            super().__init__(name=tool_name, tools=[])

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Hidden Runtime Tool",
        description="Test-only toolkit for constructor contract coverage.",
        category=ToolCategory.DEVELOPMENT,
    )
    def _hidden_runtime_tool_factory() -> type[HiddenRuntimeToolkit]:
        return HiddenRuntimeToolkit

    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={},
    )

    try:
        with pytest.raises(TypeError, match="runtime_paths"):
            get_tool_by_name(
                tool_name,
                runtime_paths,
                runtime_overrides={"runtime_paths": runtime_paths},
                worker_target=None,
            )
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_get_tool_by_name_passes_declared_managed_init_args(tmp_path: Path) -> None:
    """Declared MindRoom-managed kwargs should reach the constructor directly."""
    tool_name = "test_explicit_runtime_tool"

    class ExplicitRuntimeToolkit(Toolkit):
        def __init__(
            self,
            *,
            runtime_paths: object,
            worker_target: object,
            current_room_id: str | None,
        ) -> None:
            self.runtime_paths = runtime_paths
            self.worker_target = worker_target
            self.current_room_id = current_room_id
            super().__init__(name=tool_name, tools=[])

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Explicit Runtime Tool",
        description="Test-only toolkit for explicit constructor contract coverage.",
        category=ToolCategory.DEVELOPMENT,
        managed_init_args=(
            ToolManagedInitArg.RUNTIME_PATHS,
            ToolManagedInitArg.WORKER_TARGET,
            ToolManagedInitArg.CURRENT_ROOM_ID,
        ),
    )
    def _explicit_runtime_tool_factory() -> type[ExplicitRuntimeToolkit]:
        return ExplicitRuntimeToolkit

    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={},
    )

    try:
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@user:localhost",
            room_id="!room:localhost",
            thread_id="$thread:localhost",
            resolved_thread_id="$thread:localhost",
            session_id="session",
        )
        worker_target = resolve_worker_target(
            "shared",
            "general",
            execution_identity=execution_identity,
            tenant_id=runtime_paths.env_value("CUSTOMER_ID"),
            account_id=runtime_paths.env_value("ACCOUNT_ID"),
        )
        tool = get_tool_by_name(
            tool_name,
            runtime_paths,
            worker_target=worker_target,
        )
        assert isinstance(tool, ExplicitRuntimeToolkit)
        assert tool.runtime_paths == runtime_paths
        assert tool.worker_target == worker_target
        assert tool.current_room_id == execution_identity.room_id
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_validate_authored_overrides_accepts_declared_field_types_and_nulls() -> None:
    """Authored overrides should accept declared scalar types and optional nulls."""
    tool_name = "test_authored_override_tool"

    class _FakeToolkit(Toolkit):
        def __init__(self, **_kwargs: object) -> None:
            super().__init__(name=tool_name, tools=[])

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Authored Override Tool",
        description="Test-only toolkit for authored override validation.",
        category=ToolCategory.DEVELOPMENT,
        config_fields=[
            ConfigField(name="enabled", label="Enabled", type="boolean", required=False),
            ConfigField(name="count", label="Count", type="number", required=False),
            ConfigField(name="label", label="Label", type="text", required=False),
            ConfigField(name="endpoint", label="Endpoint", type="url", required=False),
        ],
    )
    def _fake_tool_factory() -> type[_FakeToolkit]:
        return _FakeToolkit

    try:
        assert _validate_authored_overrides(
            tool_name,
            {
                "enabled": True,
                "count": 3.5,
                "label": None,
                "endpoint": "https://example.com",
            },
            config_path_prefix="agents.code.tools[0]",
        ) == {
            "enabled": True,
            "count": 3.5,
            "label": None,
            "endpoint": "https://example.com",
        }
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_validate_authored_overrides_accepts_inherit_sentinel_for_required_fields() -> None:
    """The inherit sentinel should be allowed even when the field itself is required."""
    tool_name = "test_authored_override_inherit_required"

    class _FakeToolkit(Toolkit):
        def __init__(self, **_kwargs: object) -> None:
            super().__init__(name=tool_name, tools=[])

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Authored Override Inherit Required",
        description="Test-only toolkit for inherit sentinel coverage.",
        category=ToolCategory.DEVELOPMENT,
        config_fields=[
            ConfigField(name="workspace_id", label="Workspace ID", type="text", required=True),
        ],
    )
    def _fake_tool_factory() -> type[_FakeToolkit]:
        return _FakeToolkit

    try:
        assert _validate_authored_overrides(
            tool_name,
            {"workspace_id": _AUTHORED_OVERRIDE_INHERIT},
            config_path_prefix="agents.code.tools[0]",
        ) == {"workspace_id": _AUTHORED_OVERRIDE_INHERIT}
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_validate_authored_overrides_accepts_string_lists_for_text_fields_with_agent_override_arrays() -> None:
    """Text config fields may accept list-form values when the agent override schema exposes string arrays."""
    tool_name = "test_authored_override_string_array_compat"

    class _FakeToolkit(Toolkit):
        def __init__(self, **_kwargs: object) -> None:
            super().__init__(name=tool_name, tools=[])

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Authored Override String Array Compat",
        description="Test-only toolkit for string-array compatibility coverage.",
        category=ToolCategory.DEVELOPMENT,
        config_fields=[
            ConfigField(name="patterns", label="Patterns", type="text", required=False),
        ],
        agent_override_fields=[
            ConfigField(name="patterns", label="Patterns", type="string[]", required=False),
        ],
    )
    def _fake_tool_factory() -> type[_FakeToolkit]:
        return _FakeToolkit

    try:
        assert _validate_authored_overrides(
            tool_name,
            {"patterns": ["GITEA_*", "WHISPER_URL"]},
            config_path_prefix="agents.code.tools[0]",
        ) == {"patterns": "GITEA_*, WHISPER_URL"}
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_validate_authored_overrides_rejects_bad_types_and_password_fields() -> None:
    """Authored overrides should reject bad types, runtime-only fields, and password fields."""
    tool_name = "test_authored_override_errors"

    class _FakeToolkit(Toolkit):
        def __init__(self, **_kwargs: object) -> None:
            super().__init__(name=tool_name, tools=[])

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Authored Override Errors",
        description="Test-only toolkit for override error coverage.",
        category=ToolCategory.DEVELOPMENT,
        config_fields=[
            ConfigField(name="flag", label="Flag", type="boolean", required=False),
            ConfigField(name="base_dir", label="Base Dir", type="text", required=False, authored_override=False),
            ConfigField(name="api_key", label="API Key", type="password", required=False),
        ],
    )
    def _fake_tool_factory() -> type[_FakeToolkit]:
        return _FakeToolkit

    try:
        with pytest.raises(
            ToolConfigOverrideError,
            match=r"agents.code.tools\[0\].test_authored_override_errors.flag",
        ):
            _validate_authored_overrides(
                tool_name,
                {"flag": "yes"},
                config_path_prefix="agents.code.tools[0]",
            )

        with pytest.raises(ToolConfigOverrideError, match="authored overrides are not allowed for this field"):
            _validate_authored_overrides(
                tool_name,
                {"base_dir": "/workspace"},
                config_path_prefix="agents.code.tools[0]",
            )

        with pytest.raises(ToolConfigOverrideError, match="password fields"):
            _validate_authored_overrides(
                tool_name,
                {"api_key": "sk-test"},
                config_path_prefix="agents.code.tools[0]",
            )

        with pytest.raises(ToolConfigOverrideError, match="unknown authored override field"):
            _validate_authored_overrides(
                tool_name,
                {"missing": True},
                config_path_prefix="agents.code.tools[0]",
            )
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_searxng_include_tools_override_filters_registered_functions(tmp_path: Path) -> None:
    """Universal Agno toolkit filters should retain selected functions."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
    )

    tool = get_tool_by_name(
        "searxng",
        runtime_paths,
        credential_overrides={"host": "https://search.example.com"},
        tool_config_overrides={
            "include_tools": ["search_web", "news_search", "image_search"],
        },
        disable_sandbox_proxy=True,
        worker_target=None,
    )

    assert set(tool.functions) == {"search_web", "news_search", "image_search"}


def test_searxng_empty_include_tools_override_filters_all_functions(tmp_path: Path) -> None:
    """An explicit empty universal allowlist should expose no toolkit functions."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
    )

    tool = get_tool_by_name(
        "searxng",
        runtime_paths,
        credential_overrides={"host": "https://search.example.com"},
        tool_config_overrides={"include_tools": []},
        disable_sandbox_proxy=True,
        worker_target=None,
    )

    assert not tool.functions


def test_file_empty_exclude_patterns_override_reaches_constructor(tmp_path: Path) -> None:
    """Declared string-array fields should preserve an explicit empty list."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
    )

    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_config_overrides={"exclude_patterns": []},
        disable_sandbox_proxy=True,
        worker_target=None,
    )

    assert tool.exclude_patterns == []


def test_custom_toolkit_exclude_tools_override_filters_async_functions(tmp_path: Path) -> None:
    """Universal filters should work when a Toolkit subclass omits filter constructor kwargs."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
    )

    tool = get_tool_by_name(
        "scheduler",
        runtime_paths,
        tool_config_overrides={"exclude_tools": ["cancel_schedule"]},
        disable_sandbox_proxy=True,
        worker_target=None,
    )

    assert set(tool.async_functions) == {
        "schedule",
        "edit_schedule",
        "list_schedules",
    }


@pytest.mark.parametrize("tool_name", ["composio", "memory"])
def test_non_toolkit_registration_rejects_universal_filters(tool_name: str) -> None:
    """Universal filters should not validate for non-Toolkit catalog entries."""
    with pytest.raises(ToolConfigOverrideError, match="unknown authored override field"):
        _validate_authored_overrides(
            tool_name,
            {"include_tools": ["GITHUB_CREATE_ISSUE"]},
            config_path_prefix="agents.code.tools[0]",
        )


def test_config_load_rejects_unknown_tool_override_key(tmp_path: Path) -> None:
    """Config runtime validation should name the tool and unknown override key."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
    )

    with pytest.raises(ConfigRuntimeValidationError) as exc_info:
        Config.validate_with_runtime(
            {
                "models": {
                    "default": {
                        "provider": "openai",
                        "id": "gpt-5.6",
                    },
                },
                "router": {"model": "default"},
                "agents": {
                    "research": {
                        "display_name": "Research",
                        "role": "Search the web",
                        "model": "default",
                        "tools": [
                            {
                                "searxng": {
                                    "host": "https://search.example.com",
                                    "fixed_max_results": 10,
                                    "unknown_filter": ["search_web"],
                                },
                            },
                        ],
                    },
                },
            },
            runtime_paths,
        )

    message = str(exc_info.value)
    assert "searxng.unknown_filter" in message
    assert "unknown authored override field" in message


def test_tool_validation_snapshot_round_trips_mcp_override_validation(tmp_path: Path) -> None:
    """Validation snapshots should preserve explicit MCP override-validator semantics."""
    runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config.yaml")
    config = Config.model_validate(
        {
            "models": {
                "default": {
                    "provider": "openai",
                    "id": "gpt-5.4",
                },
            },
            "agents": {},
            "router": {"model": "default"},
            "mcp_servers": {
                "demo": {
                    "transport": "stdio",
                    "command": "npx",
                    "args": ["-y", "@modelcontextprotocol/server-everything"],
                },
            },
        },
        context={"runtime_paths": runtime_paths},
    )

    snapshot = resolved_tool_validation_snapshot_for_runtime(runtime_paths, config)
    payload = serialize_tool_validation_snapshot(snapshot)
    restored_snapshot = deserialize_tool_validation_snapshot(payload)

    assert restored_snapshot["mcp_demo"].authored_override_validator == ToolAuthoredOverrideValidator.MCP
    assert restored_snapshot["mcp_demo"].agent_override_fields is not None


def test_deserialize_tool_validation_snapshot_rejects_non_boolean_runtime_loadable() -> None:
    """Validation snapshot payloads should type-check runtime_loadable strictly."""
    with pytest.raises(TypeError, match="runtime_loadable to a boolean"):
        deserialize_tool_validation_snapshot(
            {
                "shell": {
                    "config_fields": [],
                    "agent_override_fields": [],
                    "authored_override_validator": "default",
                    "runtime_loadable": "yes",
                },
            },
        )


def test_deserialize_tool_validation_snapshot_rejects_non_boolean_room_context() -> None:
    """Validation snapshot payloads should type-check room-context requirements strictly."""
    with pytest.raises(TypeError, match="requires_room_context to a boolean"):
        deserialize_tool_validation_snapshot(
            {
                "todo": {
                    "config_fields": [],
                    "agent_override_fields": [],
                    "authored_override_validator": "default",
                    "requires_room_context": "yes",
                    "runtime_loadable": True,
                },
            },
        )


def test_deserialize_tool_validation_snapshot_rejects_non_boolean_toolkit_filter_support() -> None:
    """Validation snapshot payloads should type-check toolkit-filter support strictly."""
    with pytest.raises(TypeError, match="supports_toolkit_filters to a boolean"):
        deserialize_tool_validation_snapshot(
            {
                "todo": {
                    "config_fields": [],
                    "agent_override_fields": [],
                    "authored_override_validator": "default",
                    "supports_toolkit_filters": "yes",
                    "runtime_loadable": True,
                },
            },
        )


def test_get_tool_by_name_rejects_invalid_mcp_assignment_overrides(tmp_path: Path) -> None:
    """Direct tool construction must enforce the same MCP-specific override rules as config loading."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n"
        "  default:\n"
        "    provider: openai\n"
        "    id: gpt-5.4\n"
        "router:\n"
        "  model: default\n"
        "mcp_servers:\n"
        "  demo:\n"
        "    transport: stdio\n"
        "    command: python\n"
        "    args:\n"
        "      - -c\n"
        "      - print(0)\n"
        "agents:\n"
        "  code:\n"
        "    display_name: Code\n"
        "    role: test\n"
        "    model: default\n"
        "    tools:\n"
        "      - mcp_demo\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(config_path=config_path, storage_path=tmp_path / "storage")
    config = load_config(runtime_paths)
    try:
        ensure_tool_registry_loaded(runtime_paths, config)

        with pytest.raises(ToolConfigOverrideError, match="include_tools and exclude_tools overlap"):
            get_tool_by_name(
                "mcp_demo",
                runtime_paths,
                tool_config_overrides={
                    "include_tools": ["echo"],
                    "exclude_tools": ["echo"],
                },
                disable_sandbox_proxy=True,
                worker_target=None,
            )
    finally:
        TOOL_REGISTRY.pop("mcp_demo", None)
        TOOL_METADATA.pop("mcp_demo", None)


def test_secret_like_config_fields_are_marked_password() -> None:
    """Secret-like tool config fields should be declared as password inputs."""
    suspicious_suffixes = ("_api_key", "_password", "_secret", "_token")
    suspicious_exact = {
        "api_key",
        "password",
        "secret",
        "token",
        "access_token",
        "refresh_token",
        "auth_token",
        "bearer_token",
    }

    for tool_name, metadata in TOOL_METADATA.items():
        for field in metadata.config_fields or []:
            lowered = field.name.lower()
            if "url" in lowered or lowered.endswith("_id") or lowered == "client_id":
                continue
            if lowered in suspicious_exact or lowered.endswith(suspicious_suffixes):
                assert field.type == "password", f"{tool_name}.{field.name} should use type='password'"


def test_resolved_tool_state_cached_per_config_object(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """One config object should resolve its tool state once across all consumers."""
    compute_calls = 0
    dummy_state = metadata_module._ResolvedToolState({}, {}, {})

    def counted_compute(*_args: object, **_kwargs: object) -> metadata_module._ResolvedToolState:
        nonlocal compute_calls
        compute_calls += 1
        return dummy_state

    monkeypatch.setattr(metadata_module, "_compute_resolved_tool_state_for_runtime", counted_compute)
    metadata_module.clear_resolved_tool_state_cache()
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={},
    )
    config = Config.model_validate({})
    try:
        first = metadata_module._resolved_tool_state_for_runtime(runtime_paths, config)
        second = metadata_module._resolved_tool_state_for_runtime(runtime_paths, config)
        assert first is dummy_state
        assert second is dummy_state
        assert compute_calls == 1

        # A different tolerate flag, config object, or runtime context recomputes.
        metadata_module._resolved_tool_state_for_runtime(runtime_paths, config, tolerate_plugin_load_errors=True)
        assert compute_calls == 2
        metadata_module._resolved_tool_state_for_runtime(runtime_paths, Config.model_validate({}))
        assert compute_calls == 3
        other_runtime_paths = resolve_runtime_paths(
            config_path=tmp_path / "other" / "config.yaml",
            storage_path=tmp_path / "other" / "storage",
            process_env={},
        )
        metadata_module._resolved_tool_state_for_runtime(other_runtime_paths, config)
        assert compute_calls == 4

        metadata_module.clear_resolved_tool_state_cache()
        metadata_module._resolved_tool_state_for_runtime(runtime_paths, config)
        assert compute_calls == 5
    finally:
        metadata_module.clear_resolved_tool_state_cache()


def test_resolved_tool_state_cache_evicts_on_config_gc(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Cached tool state must not outlive its config object (id-reuse safety)."""
    monkeypatch.setattr(
        metadata_module,
        "_compute_resolved_tool_state_for_runtime",
        lambda *_args, **_kwargs: metadata_module._ResolvedToolState({}, {}, {}),
    )
    metadata_module.clear_resolved_tool_state_cache()
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={},
    )
    config = Config.model_validate({})
    try:
        metadata_module._resolved_tool_state_for_runtime(runtime_paths, config)
        assert metadata_module._RESOLVED_TOOL_STATE_CACHE
        del config
        gc.collect()
        assert not metadata_module._RESOLVED_TOOL_STATE_CACHE
    finally:
        metadata_module.clear_resolved_tool_state_cache()
