"""Tests for tool call interception hooks."""

from __future__ import annotations

import asyncio
import json
import threading
import time
from dataclasses import replace
from types import SimpleNamespace
from typing import TYPE_CHECKING, ClassVar, Literal
from unittest.mock import AsyncMock, MagicMock, patch

import nio
import pytest
from agno.models.ollama import Ollama
from agno.tools import Toolkit
from agno.tools.function import Function, FunctionCall

from mindroom.agents import create_agent
from mindroom.approval_manager import (
    PendingApproval,
    SentApprovalEvent,
    _ApprovalManager,
    get_approval_store,
    initialize_approval_store,
)
from mindroom.bot import AgentBot
from mindroom.config.agent import AgentConfig
from mindroom.config.main import Config
from mindroom.config.models import DebugConfig, ModelConfig
from mindroom.config.plugin import PluginEntryConfig
from mindroom.constants import HOOK_MESSAGE_RECEIVED_DEPTH_KEY
from mindroom.entity_resolution import mindroom_user_id
from mindroom.hooks import (
    BUILTIN_EVENT_NAMES,
    EVENT_MESSAGE_RECEIVED,
    EVENT_TOOL_AFTER_CALL,
    EVENT_TOOL_BEFORE_CALL,
    CustomEventContext,
    HookRegistry,
    ToolAfterCallContext,
    ToolBeforeCallContext,
    emit_gate,
    hook,
)
from mindroom.hooks.types import default_timeout_ms_for_event, validate_event_name
from mindroom.matrix.users import AgentMatrixUser
from mindroom.message_target import MessageTarget
from mindroom.oauth.providers import OAuthConnectionRequired
from mindroom.orchestrator import _MultiAgentOrchestrator
from mindroom.session_ids import create_session_id
from mindroom.sync_bridge_state import is_loop_blocked_by_sync_tool_bridge
from mindroom.tool_approval import ToolCallWorkflowOrigin, _shutdown_approval_store
from mindroom.tool_system import tool_hooks
from mindroom.tool_system.metadata import TOOL_METADATA, TOOL_REGISTRY, ToolCategory
from mindroom.tool_system.registration import register_tool_with_metadata
from mindroom.tool_system.runtime_context import (
    ToolDispatchContext,
    ToolRuntimeContext,
    emit_custom_event,
    tool_runtime_context,
)
from mindroom.tool_system.tool_hooks import build_tool_hook_bridge, prepend_tool_hook_bridge
from mindroom.tool_system.worker_routing import ToolExecutionIdentity, tool_execution_identity
from mindroom.tools import approved_egress as _approved_egress
from tests.approval_test_support import resolve_pending_approval as _resolve_pending_approval
from tests.conftest import (
    bind_runtime_paths,
    make_conversation_cache_mock,
    make_event_cache_mock,
    runtime_paths_for,
    test_runtime_paths,
)

if TYPE_CHECKING:
    from collections.abc import Awaitable, Callable, Generator
    from pathlib import Path

    from mindroom.constants import RuntimePaths

type SyncBridgeEvent = (
    tuple[Literal["before"], str, dict[str, str], str]
    | tuple[Literal["tool"], str]
    | tuple[Literal["after"], str, bool, str]
)

_SESSION_ID = "!room:localhost:$resolved-thread"


class NonDeepcopyableResult:
    """Mutable result object that deliberately breaks deepcopy()."""

    def __init__(self, payload: dict[str, str]) -> None:
        self.payload = payload

    def __deepcopy__(self, memo: dict[int, object]) -> object:
        """Force the snapshot path away from deepcopy()."""
        del memo
        msg = "result deepcopy disabled"
        raise TypeError(msg)

    def __repr__(self) -> str:
        """Keep the lossy fallback stable for assertions."""
        return f"NonDeepcopyableResult(payload={self.payload!r})"


class NonDeepcopyableToolError(ValueError):
    """Mutable exception object that deliberately breaks deepcopy()."""

    def __init__(self, message: str, details: dict[str, str]) -> None:
        super().__init__(message)
        self.details = details

    def __deepcopy__(self, memo: dict[int, object]) -> object:
        """Force the snapshot path away from deepcopy()."""
        del memo
        msg = "error deepcopy disabled"
        raise TypeError(msg)


def _config(
    tmp_path: Path,
    *,
    tools: list[str] | None = None,
    plugins: list[object] | None = None,
    log_llm_requests: bool = False,
) -> Config:
    runtime_paths = test_runtime_paths(tmp_path)
    return bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    tools=tools or [],
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            plugins=plugins or [],
            debug=DebugConfig(log_llm_requests=log_llm_requests),
        ),
        runtime_paths,
    )


def _plugin(
    name: str,
    callbacks: list[object],
    *,
    plugin_order: int = 0,
    settings: dict[str, object] | None = None,
) -> SimpleNamespace:
    return SimpleNamespace(
        name=name,
        discovered_hooks=tuple(callbacks),
        entry_config=PluginEntryConfig(path=f"./plugins/{name}", settings=settings or {}),
        plugin_order=plugin_order,
    )


def _initialize_router_approval_store(
    runtime_paths: RuntimePaths,
    *,
    room_send: AsyncMock | None = None,
    editor: AsyncMock | None = None,
) -> tuple[MagicMock, AsyncMock]:
    orchestrator = _MultiAgentOrchestrator(runtime_paths=runtime_paths)
    orchestrator.config = bind_runtime_paths(Config(), runtime_paths)
    orchestrator._capture_runtime_loop()

    client = MagicMock()
    client.user_id = "@mindroom_router:localhost"
    client.room_send = room_send or AsyncMock(
        return_value=nio.RoomSendResponse(event_id="$approval", room_id="!room:localhost"),
    )
    client.rooms = {"!room:localhost": nio.MatrixRoom("!room:localhost", "@mindroom_router:localhost")}
    bot = MagicMock()
    bot.agent_name = "router"
    bot.running = True
    bot.client = client
    bot.approval_room_ids = frozenset({"!room:localhost"})
    bot.latest_thread_event_id_if_needed = AsyncMock(return_value="$resolved-thread")
    orchestrator.agent_bots = {"router": bot}

    approval_editor = editor or AsyncMock()
    initialize_approval_store(
        runtime_paths,
        sender=orchestrator._approval_transport.send_approval_event,
        editor=approval_editor,
    )
    return client, approval_editor


def _before_context(
    tmp_path: Path,
    *,
    agent_name: str = "code",
    room_id: str | None = "!room:localhost",
) -> ToolBeforeCallContext:
    config = _config(tmp_path)
    return ToolBeforeCallContext(
        tool_name="read_file",
        arguments={"path": "notes.txt"},
        agent_name=agent_name,
        room_id=room_id,
        thread_id="$thread",
        requester_id="@user:localhost",
        session_id=_SESSION_ID,
        config=config,
        runtime_paths=runtime_paths_for(config),
        correlation_id="corr-tool",
    )


def _tool_runtime_context(
    tmp_path: Path,
    *,
    agent_name: str = "code",
    log_llm_requests: bool = False,
    hook_message_sender: object | None = None,
    room_state_querier: object | None = None,
    room_state_putter: object | None = None,
    message_received_depth: int = 0,
    hook_registry: HookRegistry | None = None,
) -> ToolRuntimeContext:
    config = _config(tmp_path, log_llm_requests=log_llm_requests)
    return ToolRuntimeContext(
        agent_name=agent_name,
        target=MessageTarget(
            room_id="!room:localhost",
            source_thread_id="$thread",
            resolved_thread_id="$resolved-thread",
            reply_to_event_id=None,
            session_id=_SESSION_ID,
        ),
        requester_id="@user:localhost",
        client=AsyncMock(),
        config=config,
        runtime_paths=runtime_paths_for(config),
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
        correlation_id="corr-runtime",
        hook_registry=hook_registry or HookRegistry.empty(),
        hook_message_sender=hook_message_sender,
        room_state_querier=room_state_querier,
        room_state_putter=room_state_putter,
        message_received_depth=message_received_depth,
    )


def _execution_identity() -> ToolExecutionIdentity:
    return ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@user:localhost",
        room_id="!room:localhost",
        thread_id="$thread",
        resolved_thread_id="$resolved-thread",
        session_id=_SESSION_ID,
    )


def _dispatch_context(
    execution_identity: ToolExecutionIdentity | None = None,
) -> ToolDispatchContext | None:
    if execution_identity is None:
        return None
    return ToolDispatchContext(execution_identity=execution_identity)


def _agent_bot(tmp_path: Path, *, config: Config, agent_name: str = "code") -> AgentBot:
    bot = AgentBot(
        agent_user=AgentMatrixUser(
            agent_name=agent_name,
            user_id=f"@mindroom_{agent_name}:localhost",
            display_name=agent_name.title(),
            password="test-password",  # noqa: S106
        ),
        storage_path=tmp_path,
        config=config,
        runtime_paths=runtime_paths_for(config),
        rooms=["!room:localhost"],
    )
    bot.client = MagicMock(rooms={})
    return bot


def _initialize_test_approval_store(runtime_paths: RuntimePaths) -> tuple[AsyncMock, AsyncMock]:
    sender = AsyncMock(return_value=SentApprovalEvent("$approval"))
    editor = AsyncMock()
    initialize_approval_store(
        runtime_paths,
        sender=sender,
        editor=editor,
    )
    return sender, editor


async def _wait_for_sent_pending(
    store: _ApprovalManager,
    sender: AsyncMock | MagicMock,
    *,
    room_id: str = "!room:localhost",
) -> PendingApproval:
    async with asyncio.timeout(5):
        while True:
            if sender.await_args is not None:
                content = sender.await_args.kwargs.get("content")
                if content is None:
                    content = sender.await_args.args[2]
                assert isinstance(content, dict)
                approval_id = content["approval_id"]
                card_event_id = store._live_card_event_id_for_approval(approval_id)
                pending = (
                    None
                    if card_event_id is None
                    else await store._pending_approval_for_card(room_id=room_id, card_event_id=card_event_id)
                )
                if pending is not None:
                    return pending
            await asyncio.sleep(0)


def _first_function(toolkit: Toolkit) -> Function:
    functions = [*toolkit.functions.values(), *toolkit.async_functions.values()]
    assert functions
    return functions[0]


@pytest.fixture(autouse=True)
def reset_approval_store() -> Generator[None, None, None]:
    """Keep the module-level approval store isolated per test."""
    asyncio.run(_shutdown_approval_store())
    yield
    asyncio.run(_shutdown_approval_store())


def test_tool_events_are_registered_with_expected_timeouts() -> None:
    """Tool hook events should be valid built-ins with the expected defaults."""
    assert EVENT_TOOL_BEFORE_CALL in BUILTIN_EVENT_NAMES
    assert EVENT_TOOL_AFTER_CALL in BUILTIN_EVENT_NAMES
    assert validate_event_name(EVENT_TOOL_BEFORE_CALL) == EVENT_TOOL_BEFORE_CALL
    assert validate_event_name(EVENT_TOOL_AFTER_CALL) == EVENT_TOOL_AFTER_CALL
    with pytest.raises(ValueError, match="reserved namespace"):
        validate_event_name("tool:custom")
    assert default_timeout_ms_for_event(EVENT_MESSAGE_RECEIVED) == 15000
    assert default_timeout_ms_for_event(EVENT_TOOL_BEFORE_CALL) == 200
    assert default_timeout_ms_for_event(EVENT_TOOL_AFTER_CALL) == 300


def test_tool_before_call_context_decline_helper() -> None:
    """ToolBeforeCallContext.decline() should set the mutable gate fields."""
    context = ToolBeforeCallContext(
        tool_name="read_file",
        arguments={"path": "secret.txt"},
        agent_name="code",
        room_id="!room:localhost",
        thread_id="$thread",
        requester_id="@user:localhost",
        session_id=_SESSION_ID,
    )

    context.decline("blocked")

    assert context.declined is True
    assert context.decline_reason == "blocked"


@pytest.mark.asyncio
async def test_emit_gate_respects_priority_scope_and_first_decline(tmp_path: Path) -> None:
    """emit_gate() should run in priority order, apply scopes, and stop on the first decline."""
    seen: list[str] = []

    @hook(EVENT_TOOL_BEFORE_CALL, name="wrong-room", priority=5, rooms=["!elsewhere:localhost"])
    async def wrong_room(ctx: ToolBeforeCallContext) -> None:
        del ctx
        seen.append("wrong-room")

    @hook(EVENT_TOOL_BEFORE_CALL, priority=10, agents=["code"], rooms=["!room:localhost"])
    async def blocker(ctx: ToolBeforeCallContext) -> None:
        seen.append("blocker")
        ctx.decline("policy blocked the tool")

    @hook(EVENT_TOOL_BEFORE_CALL, priority=20)
    async def later(ctx: ToolBeforeCallContext) -> None:
        del ctx
        seen.append("later")

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [later, wrong_room, blocker])])
    context = _before_context(tmp_path)

    await emit_gate(registry, EVENT_TOOL_BEFORE_CALL, context)

    assert seen == ["blocker"]
    assert context.declined is True
    assert context.decline_reason == "policy blocked the tool"


@pytest.mark.asyncio
async def test_emit_gate_isolates_arguments_between_hooks(tmp_path: Path) -> None:
    """Each gate hook should see the original arguments, even after earlier hook mutation."""
    seen: list[tuple[str, bool, list[str]]] = []

    @hook(EVENT_TOOL_BEFORE_CALL, priority=10)
    async def tamper(ctx: ToolBeforeCallContext) -> None:
        ctx.arguments["path"] = "notes.txt"
        ctx.arguments["options"]["allowed"] = True
        ctx.arguments["tags"].append("tampered")
        seen.append(
            (str(ctx.arguments["path"]), bool(ctx.arguments["options"]["allowed"]), list(ctx.arguments["tags"])),
        )

    @hook(EVENT_TOOL_BEFORE_CALL, priority=20)
    async def policy(ctx: ToolBeforeCallContext) -> None:
        seen.append(
            (str(ctx.arguments["path"]), bool(ctx.arguments["options"]["allowed"]), list(ctx.arguments["tags"])),
        )
        if "secret" in str(ctx.arguments["path"]):
            ctx.decline("secret paths stay blocked")

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [tamper, policy])])
    context = _before_context(tmp_path)
    context.arguments = {
        "path": "secret.txt",
        "options": {"allowed": False},
        "tags": ["original"],
    }

    await emit_gate(registry, EVENT_TOOL_BEFORE_CALL, context)

    assert seen == [
        ("notes.txt", True, ["original", "tampered"]),
        ("secret.txt", False, ["original"]),
    ]
    assert context.arguments == {
        "path": "secret.txt",
        "options": {"allowed": False},
        "tags": ["original"],
    }
    assert context.declined is True
    assert context.decline_reason == "secret paths stay blocked"


@pytest.mark.asyncio
async def test_emit_gate_fails_open_on_errors_and_timeouts(tmp_path: Path) -> None:
    """emit_gate() should ignore hook failures and continue to later hooks."""
    seen: list[str] = []

    @hook(EVENT_TOOL_BEFORE_CALL, priority=10)
    async def broken(ctx: ToolBeforeCallContext) -> None:
        seen.append("broken")
        ctx.decline("do not keep this decline")
        msg = "boom"
        raise RuntimeError(msg)

    @hook(EVENT_TOOL_BEFORE_CALL, priority=20, timeout_ms=10)
    async def slow(ctx: ToolBeforeCallContext) -> None:
        del ctx
        seen.append("slow")
        await asyncio.sleep(0.05)

    @hook(EVENT_TOOL_BEFORE_CALL, priority=30)
    async def allowed(ctx: ToolBeforeCallContext) -> None:
        del ctx
        seen.append("allowed")

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [broken, slow, allowed])])
    context = _before_context(tmp_path)

    await emit_gate(registry, EVENT_TOOL_BEFORE_CALL, context)

    assert seen == ["broken", "slow", "allowed"]
    assert context.declined is False
    assert context.decline_reason == ""


def test_build_tool_hook_bridge_returns_bridge_without_tool_hooks() -> None:
    """Failure logging should keep the bridge installed even without plugin hooks."""
    assert build_tool_hook_bridge(HookRegistry.empty(), agent_name="code") is not None


@pytest.mark.asyncio
async def test_tool_hook_bridge_records_failures_without_registered_hooks(tmp_path: Path) -> None:
    """The bridge should durably record failures and re-raise the original exception."""
    runtime_context = _tool_runtime_context(tmp_path)
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        runtime_paths=runtime_context.runtime_paths,
    )
    error = ValueError("boom {'api_key': 'secret'} https://alice:secret@example.com/private")

    async def explode(**kwargs: object) -> object:
        del kwargs
        raise error

    assert bridge is not None
    with (
        tool_runtime_context(runtime_context),
        tool_execution_identity(_execution_identity()),
        pytest.raises(ValueError, match="boom") as exc_info,
    ):
        await bridge(
            "explode",
            explode,
            {
                "api_key": "secret",
                "nested": [{"refresh_token": "refresh-secret"}],
                "url": "https://alice:secret@example.com/private",
            },
        )

    assert exc_info.value is error

    log_path = runtime_context.runtime_paths.storage_root / "tracking" / "tool_calls.jsonl"
    records = [json.loads(line) for line in log_path.read_text(encoding="utf-8").splitlines() if line]

    assert len(records) == 1
    assert set(records[0]) == {
        "timestamp",
        "tool_name",
        "agent_name",
        "channel",
        "room_id",
        "thread_id",
        "reply_to_event_id",
        "requester_id",
        "session_id",
        "correlation_id",
        "duration_ms",
        "arguments",
        "success",
        "error_type",
        "error_message",
        "timing",
        "traceback",
    }
    assert records[0]["tool_name"] == "explode"
    assert records[0]["agent_name"] == "code"
    assert records[0]["channel"] == "matrix"
    assert records[0]["room_id"] == "!room:localhost"
    assert records[0]["thread_id"] == "$resolved-thread"
    assert records[0]["reply_to_event_id"] is None
    assert records[0]["requester_id"] == "@user:localhost"
    assert records[0]["session_id"] == _SESSION_ID
    assert records[0]["correlation_id"] == "corr-runtime"
    assert records[0]["success"] is False
    assert records[0]["error_type"] == "ValueError"
    assert isinstance(records[0]["timing"]["approval_ms"], float)
    assert isinstance(records[0]["timing"]["result_ready_ms"], float)
    assert isinstance(records[0]["timing"]["tool_body_ms"], float)
    assert records[0]["arguments"] == {
        "api_key": "***redacted***",
        "nested": [{"refresh_token": "***redacted***"}],
        "url": "https://alice:***@example.com/private",
    }
    assert "secret" not in records[0]["error_message"]
    assert "secret" not in records[0]["traceback"]


@pytest.mark.asyncio
async def test_tool_hook_bridge_preserves_original_error_when_failure_recording_breaks(tmp_path: Path) -> None:
    """Secondary logging failures should not mask the original tool exception."""
    seen_errors: list[BaseException | None] = []

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        seen_errors.append(ctx.error)

    runtime_context = _tool_runtime_context(tmp_path)
    bridge = build_tool_hook_bridge(
        HookRegistry.from_plugins([_plugin("tool-policy", [after])]),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        runtime_paths=runtime_context.runtime_paths,
    )
    error = ValueError("boom")

    async def explode(**kwargs: object) -> object:
        del kwargs
        raise error

    assert bridge is not None
    with (
        patch("mindroom.tool_system.tool_hooks.record_tool_failure", side_effect=RuntimeError("disk full")),
        patch("mindroom.tool_system.tool_hooks.logger.exception") as mock_logger_exception,
        tool_runtime_context(runtime_context),
        tool_execution_identity(_execution_identity()),
        pytest.raises(ValueError, match="boom") as exc_info,
    ):
        await bridge("explode", explode, {"api_key": "secret"})

    assert exc_info.value is error
    assert len(seen_errors) == 1
    assert isinstance(seen_errors[0], ValueError)
    assert str(seen_errors[0]) == "boom"
    mock_logger_exception.assert_called_once_with(
        "Failed to record tool failure",
        tool_name="explode",
        correlation_id="corr-runtime",
    )


def test_sync_function_call_execute_runs_tool_hooks(tmp_path: Path) -> None:
    """Sync FunctionCall.execute() should still fire the bridge hooks."""
    seen: list[SyncBridgeEvent] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        seen.append(("before", ctx.tool_name, dict(ctx.arguments), ctx.thread_id))

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        seen.append(("after", ctx.result, ctx.blocked, ctx.thread_id))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before, after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            seen.append(("tool", text))
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    with (
        patch("agno.tools.function.log_warning") as mock_log_warning,
        tool_runtime_context(_tool_runtime_context(tmp_path)),
        tool_execution_identity(_execution_identity()),
    ):
        result = FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").execute()

    assert result.status == "success"
    assert result.result == "HI"
    mock_log_warning.assert_not_called()
    assert seen == [
        ("before", "echo", {"text": "hi"}, "$resolved-thread"),
        ("tool", "hi"),
        ("after", "HI", False, "$resolved-thread"),
    ]


@pytest.mark.asyncio
async def test_sync_function_call_execute_inside_running_loop_unwraps_tool_hook_result(tmp_path: Path) -> None:
    """Sync FunctionCall.execute() must not leak async bridge sentinels inside a running loop."""
    seen: list[SyncBridgeEvent] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        seen.append(("before", ctx.tool_name, dict(ctx.arguments), ctx.thread_id))

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        seen.append(("after", ctx.result, ctx.blocked, ctx.thread_id))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before, after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            seen.append(("tool", text))
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    with (
        patch("agno.tools.function.log_warning") as mock_log_warning,
        tool_runtime_context(_tool_runtime_context(tmp_path)),
        tool_execution_identity(_execution_identity()),
    ):
        result = FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").execute()

    assert result.status == "success"
    assert result.result == "HI"
    mock_log_warning.assert_not_called()
    assert seen == [
        ("before", "echo", {"text": "hi"}, "$resolved-thread"),
        ("tool", "hi"),
        ("after", "HI", False, "$resolved-thread"),
    ]


@pytest.mark.asyncio
async def test_sync_tool_function_call_aexecute_runs_tool_hooks(tmp_path: Path) -> None:
    """Async execution should still run the bridge for sync tool entrypoints."""
    seen: list[SyncBridgeEvent] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        seen.append(("before", ctx.tool_name, dict(ctx.arguments), ctx.thread_id))

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        seen.append(("after", ctx.result, ctx.blocked, ctx.thread_id))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before, after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            seen.append(("tool", text))
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    with tool_runtime_context(_tool_runtime_context(tmp_path)), tool_execution_identity(_execution_identity()):
        result = await FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").aexecute()

    assert result.status == "success"
    assert result.result == "HI"
    assert seen == [
        ("before", "echo", {"text": "hi"}, "$resolved-thread"),
        ("tool", "hi"),
        ("after", "HI", False, "$resolved-thread"),
    ]


@pytest.mark.asyncio
async def test_tool_hook_bridge_runs_sync_tools_off_event_loop() -> None:
    """Blocking sync tools should not stall the async tool-dispatch loop."""
    bridge = build_tool_hook_bridge(HookRegistry.empty(), agent_name="code")
    assert bridge is not None

    loop_thread_id = threading.get_ident()
    stop_ticking = False
    ticks = 0

    async def ticker() -> None:
        nonlocal ticks
        while not stop_ticking:
            await asyncio.sleep(0.01)
            ticks += 1

    def slow_sync_tool() -> int:
        time.sleep(0.1)
        return threading.get_ident()

    ticker_task = asyncio.create_task(ticker())
    tool_thread_id = await bridge("slow_sync_tool", slow_sync_tool, {})
    stop_ticking = True
    await ticker_task

    assert tool_thread_id != loop_thread_id
    assert ticks > 0


@pytest.mark.asyncio
async def test_tool_hook_bridge_allows_call_and_populates_contexts(tmp_path: Path) -> None:
    """The bridge should pass through allowed calls and emit before/after contexts."""
    before_seen: list[tuple[str, dict[str, object], str, str | None, str | None, str | None, str | None]] = []
    after_seen: list[
        tuple[object | None, BaseException | None, bool, str | None, str | None, str | None, str | None]
    ] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        before_seen.append(
            (
                ctx.tool_name,
                dict(ctx.arguments),
                ctx.agent_name,
                ctx.room_id,
                ctx.thread_id,
                ctx.requester_id,
                ctx.session_id,
            ),
        )

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        after_seen.append(
            (
                ctx.result,
                ctx.error,
                ctx.blocked,
                ctx.room_id,
                ctx.thread_id,
                ctx.requester_id,
                ctx.session_id,
            ),
        )

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before, after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    with tool_runtime_context(_tool_runtime_context(tmp_path)), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    assert before_seen == [
        (
            "read_file",
            {"path": "notes.txt"},
            "code",
            "!room:localhost",
            "$resolved-thread",
            "@user:localhost",
            _SESSION_ID,
        ),
    ]
    assert after_seen == [
        (
            {"echo": "notes.txt"},
            None,
            False,
            "!room:localhost",
            "$resolved-thread",
            "@user:localhost",
            _SESSION_ID,
        ),
    ]


@pytest.mark.asyncio
async def test_tool_after_call_hooks_cannot_mutate_returned_result(tmp_path: Path) -> None:
    """After-call hooks should observe results without mutating the returned value."""

    @hook(EVENT_TOOL_AFTER_CALL)
    async def mutate(ctx: ToolAfterCallContext) -> None:
        assert isinstance(ctx.result, dict)
        ctx.result["mutated"] = True

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [mutate])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    with tool_runtime_context(_tool_runtime_context(tmp_path)), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}


@pytest.mark.asyncio
async def test_tool_hook_bridge_does_not_record_success_rows_by_default(tmp_path: Path) -> None:
    """Successful tool calls should not create durable audit rows unless debug logging is enabled."""
    runtime_context = _tool_runtime_context(tmp_path)
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=runtime_context.config,
        runtime_paths=runtime_context.runtime_paths,
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    with tool_runtime_context(runtime_context), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    assert not (runtime_context.runtime_paths.storage_root / "tracking" / "tool_calls.jsonl").exists()


@pytest.mark.asyncio
async def test_tool_after_call_hook_error_does_not_lose_debug_success_row(tmp_path: Path) -> None:
    """Debug success rows should be persisted before after-call hooks run."""

    @hook(EVENT_TOOL_AFTER_CALL)
    async def explode_after(_ctx: ToolAfterCallContext) -> None:
        msg = "after failed"
        raise RuntimeError(msg)

    runtime_context = _tool_runtime_context(tmp_path, log_llm_requests=True)
    registry = HookRegistry.from_plugins([_plugin("tool-policy", [explode_after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=runtime_context.config,
        runtime_paths=runtime_context.runtime_paths,
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    with tool_runtime_context(runtime_context), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}

    log_path = runtime_context.runtime_paths.storage_root / "tracking" / "tool_calls.jsonl"
    records = [json.loads(line) for line in log_path.read_text(encoding="utf-8").splitlines() if line]
    assert len(records) == 1
    assert records[0]["tool_name"] == "read_file"
    assert records[0]["success"] is True
    assert records[0]["reply_to_event_id"] is None
    assert records[0]["correlation_id"] == "corr-runtime"
    assert records[0]["result"] == {"echo": "notes.txt"}


@pytest.mark.asyncio
async def test_tool_success_logging_error_does_not_skip_after_hook(tmp_path: Path) -> None:
    """Debug logging failures should not turn successful tools into failed calls."""
    seen: list[object] = []

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after_call(ctx: ToolAfterCallContext) -> None:
        seen.append(ctx.result)

    runtime_context = _tool_runtime_context(tmp_path, log_llm_requests=True)
    registry = HookRegistry.from_plugins([_plugin("tool-policy", [after_call])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=runtime_context.config,
        runtime_paths=runtime_context.runtime_paths,
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    with (
        tool_runtime_context(runtime_context),
        tool_execution_identity(_execution_identity()),
        patch("mindroom.tool_system.tool_calls._append_tool_call_record", side_effect=RuntimeError("disk full")),
    ):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    assert seen == [{"echo": "notes.txt"}]


@pytest.mark.asyncio
async def test_oauth_required_tool_result_is_recorded_when_debug_enabled(tmp_path: Path) -> None:
    """OAuth-required structured returns should be logged like other successful tool results."""
    runtime_context = _tool_runtime_context(tmp_path, log_llm_requests=True)
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=runtime_context.config,
        runtime_paths=runtime_context.runtime_paths,
    )
    assert bridge is not None

    async def next_func(**_kwargs: object) -> None:
        msg = "Connect Google"
        raise OAuthConnectionRequired(
            msg,
            provider_id="google",
            connect_url="https://example.test/connect",
        )

    with tool_runtime_context(runtime_context), tool_execution_identity(_execution_identity()):
        result = await bridge("gmail", next_func, {})

    assert isinstance(result, dict)
    assert result["oauth_connection_required"] is True
    log_path = runtime_context.runtime_paths.storage_root / "tracking" / "tool_calls.jsonl"
    records = [json.loads(line) for line in log_path.read_text(encoding="utf-8").splitlines() if line]
    assert len(records) == 1
    assert records[0]["tool_name"] == "gmail"
    assert records[0]["success"] is True
    assert records[0]["result"]["oauth_connection_required"] is True


@pytest.mark.asyncio
async def test_tool_after_call_hooks_cannot_mutate_non_deepcopyable_result(tmp_path: Path) -> None:
    """After-call hooks should still get an isolated snapshot when deepcopy() fails."""
    seen: list[object | None] = []

    @hook(EVENT_TOOL_AFTER_CALL)
    async def mutate(ctx: ToolAfterCallContext) -> None:
        seen.append(ctx.result)
        assert ctx.result == "NonDeepcopyableResult(payload={'echo': 'notes.txt'})"

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [mutate])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> NonDeepcopyableResult:
        return NonDeepcopyableResult({"echo": str(kwargs["path"])})

    with tool_runtime_context(_tool_runtime_context(tmp_path)), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert isinstance(result, NonDeepcopyableResult)
    assert result.payload == {"echo": "notes.txt"}
    assert seen == ["NonDeepcopyableResult(payload={'echo': 'notes.txt'})"]


@pytest.mark.asyncio
async def test_tool_hook_context_send_message_uses_bound_sender(tmp_path: Path) -> None:
    """Tool hook contexts should route send_message() through the active hook sender."""
    sent: list[tuple[str, str, str | None, str, dict[str, object] | None, bool]] = []

    async def hook_message_sender(
        room_id: str,
        body: str,
        thread_id: str | None,
        source_hook: str,
        extra_content: dict[str, object] | None,
        *,
        trigger_dispatch: bool = False,
    ) -> str | None:
        sent.append((room_id, body, thread_id, source_hook, extra_content, trigger_dispatch))
        return f"${body}-event"

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        event_id = await ctx.send_message(
            "!room:localhost",
            "before",
            thread_id="$before-thread",
            extra_content={"phase": "before"},
        )
        assert event_id == "$before-event"

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        event_id = await ctx.send_message(
            "!room:localhost",
            "after",
            thread_id="$after-thread",
            extra_content={"phase": "after"},
        )
        assert event_id == "$after-event"

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before, after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    with (
        tool_runtime_context(_tool_runtime_context(tmp_path, hook_message_sender=hook_message_sender)),
        tool_execution_identity(_execution_identity()),
    ):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    assert sent == [
        (
            "!room:localhost",
            "before",
            "$before-thread",
            "tool-policy:tool:before_call",
            {
                "phase": "before",
                "com.mindroom.original_sender": "@user:localhost",
                HOOK_MESSAGE_RECEIVED_DEPTH_KEY: 1,
            },
            False,
        ),
        (
            "!room:localhost",
            "after",
            "$after-thread",
            "tool-policy:tool:after_call",
            {
                "phase": "after",
                "com.mindroom.original_sender": "@user:localhost",
                HOOK_MESSAGE_RECEIVED_DEPTH_KEY: 1,
            },
            False,
        ),
    ]


@pytest.mark.asyncio
async def test_tool_hook_context_send_message_advances_existing_message_received_depth(tmp_path: Path) -> None:
    """Tool hook sends should keep advancing an existing synthetic message:received chain."""
    sent: list[dict[str, object] | None] = []

    async def hook_message_sender(
        room_id: str,
        body: str,
        thread_id: str | None,
        source_hook: str,
        extra_content: dict[str, object] | None,
        *,
        trigger_dispatch: bool = False,
    ) -> str | None:
        del room_id, body, thread_id, source_hook, trigger_dispatch
        sent.append(extra_content)
        return "$dispatch-event"

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        event_id = await ctx.send_message("!room:localhost", "dispatch", trigger_dispatch=True)
        assert event_id == "$dispatch-event"

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    with (
        tool_runtime_context(
            _tool_runtime_context(
                tmp_path,
                hook_message_sender=hook_message_sender,
                message_received_depth=1,
            ),
        ),
        tool_execution_identity(_execution_identity()),
    ):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    assert sent == [
        {
            "com.mindroom.original_sender": "@user:localhost",
            HOOK_MESSAGE_RECEIVED_DEPTH_KEY: 2,
        },
    ]


@pytest.mark.asyncio
async def test_tool_hook_contexts_expose_router_backed_matrix_admin(tmp_path: Path) -> None:
    """tool:* hook contexts should expose the router-backed matrix admin helper."""
    resolved_aliases: list[str | None] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        assert ctx.matrix_admin is not None
        resolved_aliases.append(await ctx.matrix_admin.resolve_alias("#personal-user:localhost"))

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        assert ctx.matrix_admin is not None
        resolved_aliases.append(await ctx.matrix_admin.resolve_alias("#personal-user:localhost"))

    config = _config(tmp_path)
    bot = _agent_bot(tmp_path, config=config)
    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.rooms = {}
    bot.event_cache = MagicMock()
    bot.client.homeserver = "http://agent.local:8008"
    bot.client.room_resolve_alias.return_value = nio.RoomResolveAliasError(
        "not found",
        status_code="M_NOT_FOUND",
    )

    router_bot = _agent_bot(tmp_path, config=config, agent_name="router")
    router_bot.client = AsyncMock(spec=nio.AsyncClient)
    router_bot.client.rooms = {}
    router_bot.client.homeserver = "http://localhost:8008"
    router_bot.client.room_resolve_alias.return_value = nio.RoomResolveAliasResponse(
        room_alias="#personal-user:localhost",
        room_id="!personal:localhost",
        servers=["localhost"],
    )

    orchestrator = _MultiAgentOrchestrator(runtime_paths=runtime_paths_for(config))
    orchestrator.agent_bots = {"router": router_bot, "code": bot}
    bot.orchestrator = orchestrator

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before, after])])
    target = MessageTarget.resolve("!room:localhost", "$thread", None)
    execution_identity = bot._tool_runtime_support.build_execution_identity(
        target=target,
        user_id="@user:localhost",
    )
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(execution_identity),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    runtime_context = bot._tool_runtime_support.build_context(target, user_id="@user:localhost")
    assert runtime_context is not None

    with tool_runtime_context(runtime_context), tool_execution_identity(execution_identity):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    assert resolved_aliases == ["!personal:localhost", "!personal:localhost"]
    bot.client.room_resolve_alias.assert_not_awaited()
    assert router_bot.client.room_resolve_alias.await_count == 2


@pytest.mark.asyncio
async def test_tool_hook_context_room_state_helpers_use_runtime_client(tmp_path: Path) -> None:
    """Tool hook contexts should expose live room-state helpers from the active runtime client."""
    sent: list[tuple[str, str, str | None, str, dict[str, object] | None, bool]] = []

    async def hook_message_sender(
        room_id: str,
        body: str,
        thread_id: str | None,
        source_hook: str,
        extra_content: dict[str, object] | None,
        *,
        trigger_dispatch: bool = False,
    ) -> str | None:
        sent.append((room_id, body, thread_id, source_hook, extra_content, trigger_dispatch))
        return "$dispatch-event"

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        query_result = await ctx.query_room_state("!room:localhost", "m.room.name", "")
        put_result = await ctx.put_room_state(
            "!room:localhost",
            "com.mindroom.thread.tags",
            "$resolved-thread",
            {"tags": {"queued": True}},
        )
        event_id = await ctx.send_message("!room:localhost", "dispatch", trigger_dispatch=True)
        assert query_result == {"name": "Lobby"}
        assert put_result is True
        assert event_id == "$dispatch-event"

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    runtime_context = _tool_runtime_context(tmp_path, hook_message_sender=hook_message_sender)
    runtime_context.client.room_get_state_event.return_value = SimpleNamespace(content={"name": "Lobby"})
    runtime_context.client.room_put_state.return_value = object()

    with tool_runtime_context(runtime_context), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    runtime_context.client.room_get_state_event.assert_awaited_once_with("!room:localhost", "m.room.name", "")
    runtime_context.client.room_put_state.assert_awaited_once_with(
        "!room:localhost",
        "com.mindroom.thread.tags",
        {"tags": {"queued": True}},
        state_key="$resolved-thread",
    )
    assert sent == [
        (
            "!room:localhost",
            "dispatch",
            None,
            "tool-policy:tool:before_call",
            {
                "com.mindroom.original_sender": "@user:localhost",
                HOOK_MESSAGE_RECEIVED_DEPTH_KEY: 1,
            },
            True,
        ),
    ]


@pytest.mark.asyncio
async def test_agent_bot_tool_runtime_context_room_state_helpers_fallback_to_router(tmp_path: Path) -> None:
    """Bot-built tool runtime contexts should use current-bot-first room-state helpers with router fallback."""
    seen: list[tuple[dict[str, object] | None, bool]] = []
    config = _config(tmp_path)
    bot = _agent_bot(tmp_path, config=config)
    bot.client = AsyncMock(spec=nio.AsyncClient)
    bot.client.rooms = {}
    bot.event_cache = MagicMock()
    bot.client.room_get_state_event.return_value = nio.RoomGetStateEventError(message="forbidden")
    bot.client.room_put_state.return_value = nio.RoomPutStateError(message="forbidden")
    router_bot = _agent_bot(tmp_path, config=config, agent_name="router")
    router_bot.client = AsyncMock(spec=nio.AsyncClient)
    router_bot.client.rooms = {}
    router_bot.client.room_get_state_event.return_value = SimpleNamespace(content={"name": "Router Lobby"})
    router_bot.client.room_put_state.return_value = object()
    orchestrator = _MultiAgentOrchestrator(runtime_paths=runtime_paths_for(config))
    orchestrator.agent_bots = {"router": router_bot, "code": bot}
    bot.orchestrator = orchestrator

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        query_result = await ctx.query_room_state("!room:localhost", "m.room.name", "")
        put_result = await ctx.put_room_state(
            "!room:localhost",
            "com.mindroom.thread.tags",
            "$resolved-thread",
            {"tags": {"queued": True}},
        )
        seen.append((query_result, put_result))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    target = MessageTarget.resolve("!room:localhost", "$thread", None)
    execution_identity = bot._tool_runtime_support.build_execution_identity(
        target=target,
        user_id="@user:localhost",
    )
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(execution_identity),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> dict[str, object]:
        return {"echo": kwargs["path"]}

    runtime_context = bot._tool_runtime_support.build_context(target, user_id="@user:localhost")
    assert runtime_context is not None

    with tool_runtime_context(runtime_context), tool_execution_identity(execution_identity):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == {"echo": "notes.txt"}
    assert seen == [({"name": "Router Lobby"}, True)]
    bot.client.room_get_state_event.assert_awaited_once_with("!room:localhost", "m.room.name", "")
    bot.client.room_put_state.assert_awaited_once_with(
        "!room:localhost",
        "com.mindroom.thread.tags",
        {"tags": {"queued": True}},
        state_key="$resolved-thread",
    )
    router_bot.client.room_get_state_event.assert_awaited_once_with("!room:localhost", "m.room.name", "")
    router_bot.client.room_put_state.assert_awaited_once_with(
        "!room:localhost",
        "com.mindroom.thread.tags",
        {"tags": {"queued": True}},
        state_key="$resolved-thread",
    )


@pytest.mark.asyncio
async def test_sync_tool_aexecute_send_message_uses_request_loop(tmp_path: Path) -> None:
    """Sync-tool hooks should keep send_message() on the active request loop under aexecute()."""
    request_thread = threading.get_ident()
    request_loop = asyncio.get_running_loop()
    seen: list[tuple[str, int, int] | tuple[str, str]] = []

    async def hook_message_sender(
        room_id: str,
        body: str,
        thread_id: str | None,
        source_hook: str,
        extra_content: dict[str, object] | None,
        *,
        trigger_dispatch: bool = False,
    ) -> str | None:
        del room_id, body, thread_id, source_hook, extra_content, trigger_dispatch
        current_loop = asyncio.get_running_loop()
        current_thread = threading.get_ident()
        seen.append(("sender", current_thread, id(current_loop)))
        assert current_thread == request_thread
        assert current_loop is request_loop
        return "$ok"

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        current_loop = asyncio.get_running_loop()
        current_thread = threading.get_ident()
        seen.append(("hook", current_thread, id(current_loop)))
        assert current_thread == request_thread
        assert current_loop is request_loop
        event_id = await ctx.send_message("!room:localhost", "before")
        seen.append(("event_id", event_id or ""))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            current_loop = asyncio.get_running_loop()
            current_thread = threading.get_ident()
            seen.append(("tool", current_thread, id(current_loop)))
            assert current_thread == request_thread
            assert current_loop is request_loop
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    with (
        tool_runtime_context(_tool_runtime_context(tmp_path, hook_message_sender=hook_message_sender)),
        tool_execution_identity(_execution_identity()),
    ):
        result = await FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").aexecute()

    assert result.status == "success"
    assert result.result == "HI"
    assert seen == [
        ("hook", request_thread, id(request_loop)),
        ("sender", request_thread, id(request_loop)),
        ("event_id", "$ok"),
        ("tool", request_thread, id(request_loop)),
    ]


@pytest.mark.asyncio
async def test_sync_tool_approval_send_uses_runtime_loop(tmp_path: Path) -> None:
    """Sync-tool approval sends should hop back to the runtime loop."""
    request_thread = threading.get_ident()
    request_loop = asyncio.get_running_loop()
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "timeout_days": 0.000001,
                "rules": [{"match": "echo", "action": "require_approval"}],
            },
        ),
        runtime_paths,
    )

    async def mock_room_send(
        room_id: str,
        message_type: str,
        content: dict[str, object],
        *,
        ignore_unverified_devices: bool = False,
    ) -> nio.RoomSendResponse:
        current_loop = asyncio.get_running_loop()
        current_thread = threading.get_ident()
        assert current_thread == request_thread
        assert current_loop is request_loop
        assert room_id == "!room:localhost"
        assert message_type == "io.mindroom.tool_approval"
        assert ignore_unverified_devices is True
        assert content["status"] == "pending"
        return nio.RoomSendResponse(event_id="$approval", room_id=room_id)

    client, _ = _initialize_router_approval_store(runtime_paths, room_send=AsyncMock(side_effect=mock_room_send))

    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    result = await asyncio.to_thread(
        lambda: FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").execute(),
    )

    assert result.status == "success"
    assert result.result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: echo\n"
        "Reason: Tool approval request timed out.\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    client.room_send.assert_awaited_once()


@pytest.mark.asyncio
async def test_sync_execute_async_tool_entrypoint_still_runs_approval_gate(tmp_path: Path) -> None:
    """FunctionCall.execute() must not bypass approval hooks for async tool entrypoints."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "timeout_days": 0.000001,
                "rules": [{"match": "echo", "action": "require_approval"}],
            },
        ),
        runtime_paths,
    )
    client, _ = _initialize_router_approval_store(runtime_paths)

    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    executed: list[str] = []

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

        async def echo(self, text: str) -> str:
            executed.append(text)
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    with patch("agno.tools.function.log_warning") as mock_log_warning:
        result = await asyncio.to_thread(
            lambda: FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").execute(),
        )

    assert result.status == "success"
    assert result.result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: echo\n"
        "Reason: Tool approval request timed out.\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    assert executed == []
    client.room_send.assert_awaited_once()
    mock_log_warning.assert_not_called()


def _request_network_access_config(runtime_paths: RuntimePaths) -> Config:
    return bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(display_name="Code", role="Help with coding.", rooms=["!room:localhost"]),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "timeout_days": 0.000001,
                "rules": [{"match": "request_network_access", "action": "require_approval"}],
            },
        ),
        runtime_paths,
    )


def _request_network_access_bridge(config: Config, runtime_paths: RuntimePaths) -> Callable[..., Awaitable[object]]:
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None
    return bridge


@pytest.mark.parametrize("ttl_minutes", [5, "5"])
@pytest.mark.asyncio
async def test_request_network_access_static_allowlist_skips_matrix_approval(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    ttl_minutes: object,
) -> None:
    """Static-allowlisted egress requests should answer without an approval card or a grant."""
    runtime_paths = test_runtime_paths(tmp_path)
    monkeypatch.setenv("MINDROOM_APPROVED_EGRESS_ALLOWLIST", ".example.com")
    config = _request_network_access_config(runtime_paths)
    client, _ = _initialize_router_approval_store(runtime_paths)
    bridge = _request_network_access_bridge(config, runtime_paths)

    def post_grant(_payload: dict[str, object]) -> dict[str, object]:
        msg = "a static-allowlisted request must not create a temporary grant"
        raise AssertionError(msg)

    monkeypatch.setattr(_approved_egress, "_post_grant", post_grant)
    toolkit = _approved_egress._ApprovedEgressTools()
    function = toolkit.async_functions["request_network_access"]
    prepend_tool_hook_bridge(toolkit, bridge)

    result = await FunctionCall(
        function=function,
        arguments={"hostnames": ["docs.example.com"], "ttl_minutes": ttl_minutes, "reason": "Need docs."},
        call_id="call-1",
    ).aexecute()

    assert result.status == "success"
    assert (
        result.result
        == "Already allowed by the static egress allowlist: docs.example.com. No temporary grant was created."
    )
    client.room_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_request_network_access_blocked_host_still_uses_matrix_approval(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Batches containing any blocked hostname should still go through Matrix approval before reaching the tool."""
    runtime_paths = test_runtime_paths(tmp_path)
    monkeypatch.setenv("MINDROOM_APPROVED_EGRESS_ALLOWLIST", ".example.com")
    config = _request_network_access_config(runtime_paths)
    client, _ = _initialize_router_approval_store(runtime_paths)
    bridge = _request_network_access_bridge(config, runtime_paths)
    toolkit = _approved_egress._ApprovedEgressTools()

    result = await bridge(
        "request_network_access",
        toolkit.async_functions["request_network_access"].entrypoint,
        {"hostnames": ["docs.example.com", "docs.other.test"], "ttl_minutes": 5, "reason": "Need docs."},
    )

    assert result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: request_network_access\n"
        "Reason: Tool approval request timed out.\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    client.room_send.assert_awaited_once()


@pytest.mark.asyncio
async def test_sync_tool_approval_resumes_after_cross_loop_resolution(tmp_path: Path) -> None:
    """Approval-gated sync tools should resume after approval resolves on another loop."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "rules": [{"match": "echo", "action": "require_approval"}],
            },
        ),
        runtime_paths,
    )
    editor = AsyncMock()
    client, _ = _initialize_router_approval_store(runtime_paths, editor=editor)

    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)
    result: object | None = None
    error: BaseException | None = None

    def worker() -> None:
        nonlocal result, error

        try:
            result = FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").execute()
        except BaseException as exc:  # pragma: no cover - asserted below
            error = exc

    thread = threading.Thread(target=worker)
    thread.start()

    store = get_approval_store()
    assert store is not None
    pending = await _wait_for_sent_pending(store, client.room_send)

    await _resolve_pending_approval(
        store,
        pending,
        status="approved",
        reason=None,
    )
    await asyncio.to_thread(thread.join, 1)

    assert error is None
    assert not thread.is_alive()
    assert result is not None
    assert result.status == "success"
    assert result.result == "HI"
    client.room_send.assert_awaited_once()
    assert editor.await_args.args[2]["status"] == "approved"


@pytest.mark.asyncio
async def test_sync_tool_approval_aexecute_resumes_on_runtime_loop(tmp_path: Path) -> None:
    """Approval-gated sync tools should not deadlock under FunctionCall.aexecute()."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "rules": [{"match": "echo", "action": "require_approval"}],
            },
        ),
        runtime_paths,
    )
    editor = AsyncMock(return_value=True)
    client, _ = _initialize_router_approval_store(runtime_paths, editor=editor)

    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    async def run_and_approve() -> object:
        task = asyncio.create_task(
            FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").aexecute(),
        )
        store = get_approval_store()
        assert store is not None
        pending = await _wait_for_sent_pending(store, client.room_send)
        await _resolve_pending_approval(
            store,
            pending,
            status="approved",
            reason=None,
        )
        return await task

    result = await asyncio.wait_for(run_and_approve(), timeout=1)

    assert result.status == "success"
    assert result.result == "HI"
    client.room_send.assert_awaited_once()
    assert editor.await_args.args[2]["status"] == "approved"


@pytest.mark.asyncio
async def test_deferred_sync_bridge_marks_runtime_loop_before_worker_start(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """The sync bridge deadlock guard must be active before the worker can run."""
    runtime_loop = asyncio.get_running_loop()
    worker_started = threading.Event()
    marker_observed: list[bool] = []
    original_start = threading.Thread.start

    def start_and_wait_for_worker(self: threading.Thread) -> None:
        original_start(self)
        if self.name == "mindroom-tool-hook-sync-bridge":
            assert worker_started.wait(timeout=1)

    monkeypatch.setattr(tool_hooks.threading.Thread, "start", start_and_wait_for_worker)

    async def awaitable_result() -> str:
        marker_observed.append(is_loop_blocked_by_sync_tool_bridge(runtime_loop))
        worker_started.set()
        return "done"

    deferred = tool_hooks._run_coroutine_from_sync(awaitable_result())
    result = tool_hooks._resolve_deferred_sync_result(deferred)

    assert result == "done"
    assert marker_observed == [True]


@pytest.mark.asyncio
async def test_sync_tool_approval_execute_on_runtime_loop_fails_fast(tmp_path: Path) -> None:
    """Direct sync execute() on the runtime loop cannot wait for Matrix approval transport."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "rules": [{"match": "echo", "action": "require_approval"}],
            },
        ),
        runtime_paths,
    )
    client, _ = _initialize_router_approval_store(runtime_paths)

    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

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

        def echo(self, text: str) -> str:
            return text.upper()

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)

    result = FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").execute()

    assert result.status == "success"
    assert result.result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: echo\n"
        "Reason: Cannot perform Matrix approval transport while synchronous FunctionCall.execute() "
        "is blocking the MindRoom runtime loop; use FunctionCall.aexecute() or run execute() "
        "outside the runtime event loop.\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    client.room_send.assert_not_awaited()


@pytest.mark.asyncio
async def test_tool_approval_scripts_cannot_mutate_rendered_approval_payload(tmp_path: Path) -> None:
    """Approval scripts should not be able to rewrite the payload the human approves."""
    runtime_paths = test_runtime_paths(tmp_path)
    script_path = tmp_path / "approval_scripts" / "redact.py"
    script_path.parent.mkdir(parents=True, exist_ok=True)
    script_path.write_text(
        "def check(tool_name, arguments, agent_name):\n    arguments['payload'] = 'tampered'\n    return True\n",
        encoding="utf-8",
    )
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "rules": [{"match": "echo", "script": "approval_scripts/redact.py"}],
            },
        ),
        runtime_paths,
    )
    sender = AsyncMock(return_value=SentApprovalEvent("$approval"))
    initialize_approval_store(runtime_paths, sender=sender, editor=AsyncMock())

    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

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

        def echo(self, payload: str) -> str:
            return json.dumps({"payload": payload}, sort_keys=True)

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    prepend_tool_hook_bridge(toolkit, bridge)
    result: object | None = None
    error: BaseException | None = None

    def worker() -> None:
        nonlocal result, error

        try:
            result = FunctionCall(function=function, arguments={"payload": "original"}, call_id="call-1").execute()
        except BaseException as exc:  # pragma: no cover - asserted below
            error = exc

    thread = threading.Thread(target=worker)
    thread.start()

    store = get_approval_store()
    assert store is not None
    pending = await _wait_for_sent_pending(store, sender)

    approval_payload = sender.await_args.args[2]
    assert approval_payload["arguments"] == {"payload": "original"}

    await _resolve_pending_approval(
        store,
        pending,
        status="approved",
        reason=None,
    )
    await asyncio.to_thread(thread.join, 1)

    assert error is None
    assert not thread.is_alive()
    assert result is not None
    assert result.status == "success"
    assert result.result == json.dumps({"payload": "original"}, sort_keys=True)


@pytest.mark.asyncio
async def test_tool_approval_script_error_text_is_sanitized_in_decline_result(tmp_path: Path) -> None:
    """Approval script failures should not leak policy internals to the requester."""
    runtime_paths = test_runtime_paths(tmp_path)
    script_path = tmp_path / "approval_scripts" / "broken.py"
    script_path.parent.mkdir(parents=True, exist_ok=True)
    script_path.write_text(
        "def check(tool_name, arguments, agent_name):\n    raise ValueError(arguments['secret'])\n",
        encoding="utf-8",
    )
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "rules": [{"match": "read_file", "script": "approval_scripts/broken.py"}],
            },
        ),
        runtime_paths,
    )
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="should not run")
    with tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"secret": "sk-secret-123"})

    assert next_func.await_count == 0
    assert "Tool approval policy failed." in result
    assert "ValueError" not in result
    assert "approval_scripts" not in result
    assert "sk-secret-123" not in result


@pytest.mark.asyncio
async def test_tool_hook_bridge_sanitizes_import_time_approval_script_failures(tmp_path: Path) -> None:
    """Import-time approval script failures should not leak raw exception text to the requester."""
    runtime_paths = test_runtime_paths(tmp_path)
    script_path = tmp_path / "approval_scripts" / "broken_import.py"
    script_path.parent.mkdir(parents=True, exist_ok=True)
    script_path.write_text("raise RuntimeError('token sk-secret-123')\n", encoding="utf-8")
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "rules": [{"match": "read_file", "script": "approval_scripts/broken_import.py"}],
            },
        ),
        runtime_paths,
    )
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="should not run")
    with tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"secret": "sk-secret-123"})

    assert next_func.await_count == 0
    assert "Tool approval policy failed." in result
    assert "RuntimeError" not in result
    assert "approval_scripts" not in result
    assert "sk-secret-123" not in result


@pytest.mark.asyncio
async def test_tool_approval_uses_transport_agent_for_detached_team_member_runs(tmp_path: Path) -> None:
    """Detached team-member approvals should send from the ingress bot, not the child agent."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={"rules": [{"match": "read_file", "action": "require_approval"}]},
        ),
        runtime_paths,
    )
    sender = AsyncMock(return_value=SentApprovalEvent("$approval"))
    initialize_approval_store(runtime_paths, sender=sender, editor=AsyncMock())
    dispatch_context = _dispatch_context(
        ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@user:localhost",
            room_id="!room:localhost",
            thread_id="$thread",
            resolved_thread_id="$resolved-thread",
            session_id=_SESSION_ID,
            transport_agent_name="general",
        ),
    )
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=dispatch_context,
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="ok")
    with tool_execution_identity(dispatch_context.execution_identity if dispatch_context is not None else None):
        task = asyncio.create_task(bridge("read_file", next_func, {"path": "notes.txt"}))
        await asyncio.sleep(0)
        store = get_approval_store()
        assert store is not None
        pending = await _wait_for_sent_pending(store, sender)
        assert sender.await_args.args[:2] == ("!room:localhost", "$resolved-thread")
        assert sender.await_args.args[2]["agent_name"] == "code"
        await _resolve_pending_approval(
            store,
            pending,
            status="approved",
        )
        result = await task

    assert result == "ok"
    assert next_func.await_count == 1


@pytest.mark.asyncio
async def test_tool_approval_uses_transport_agent_for_delegated_live_runs(tmp_path: Path) -> None:
    """Delegated approvals should keep using the ingress bot from the live runtime context."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={"rules": [{"match": "read_file", "action": "require_approval"}]},
        ),
        runtime_paths,
    )
    sender = AsyncMock(return_value=SentApprovalEvent("$approval"))
    initialize_approval_store(runtime_paths, sender=sender, editor=AsyncMock())
    delegated_runtime_context = replace(
        _tool_runtime_context(tmp_path),
        agent_name="code",
        transport_agent_name="general",
        config=config,
        runtime_paths=runtime_paths,
    )
    delegated_execution_identity = replace(
        _execution_identity(),
        agent_name="code",
        transport_agent_name="general",
    )
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(delegated_execution_identity),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="ok")
    with (
        tool_runtime_context(delegated_runtime_context),
        tool_execution_identity(delegated_execution_identity),
    ):
        task = asyncio.create_task(bridge("read_file", next_func, {"path": "notes.txt"}))
        await asyncio.sleep(0)
        store = get_approval_store()
        assert store is not None
        pending = await _wait_for_sent_pending(store, sender)
        assert sender.await_args.args[:2] == ("!room:localhost", "$resolved-thread")
        assert sender.await_args.args[2]["agent_name"] == "code"
        await _resolve_pending_approval(
            store,
            pending,
            status="approved",
        )
        result = await task

    assert result == "ok"
    assert next_func.await_count == 1


@pytest.mark.asyncio
async def test_tool_approval_rejects_internal_mindroom_user_requester(tmp_path: Path) -> None:
    """Internal system dispatches should fail fast instead of creating an unresolvable approval."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=["!room:localhost"],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={"rules": [{"match": "read_file", "action": "require_approval"}]},
            mindroom_user={"username": "mindroom_user", "display_name": "MindRoomUser"},
        ),
        runtime_paths,
    )
    sender = AsyncMock(return_value=SentApprovalEvent("$approval"))
    initialize_approval_store(runtime_paths, sender=sender, editor=AsyncMock())
    internal_user_id = mindroom_user_id(config, runtime_paths)
    assert internal_user_id is not None
    internal_execution_identity = replace(_execution_identity(), requester_id=internal_user_id)
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(internal_execution_identity),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="should not run")
    with tool_execution_identity(internal_execution_identity):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert next_func.await_count == 0
    assert result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: read_file\n"
        "Reason: Tool approval requires a human requester.\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    sender.assert_not_awaited()


@pytest.mark.asyncio
async def test_tool_hook_bridge_prefers_bridge_agent_name_over_nested_runtime_context(tmp_path: Path) -> None:
    """Nested tool execution should stay attributed to the bridge agent, not the parent runtime context."""
    before_seen: list[str] = []
    after_seen: list[str] = []
    parent_seen: list[str] = []

    @hook(EVENT_TOOL_BEFORE_CALL, agents=["child-agent"])
    async def child_before(ctx: ToolBeforeCallContext) -> None:
        before_seen.append(ctx.agent_name)

    @hook(EVENT_TOOL_AFTER_CALL, agents=["child-agent"])
    async def child_after(ctx: ToolAfterCallContext) -> None:
        after_seen.append(ctx.agent_name)

    @hook(EVENT_TOOL_BEFORE_CALL, agents=["parent-agent"])
    async def parent_before(ctx: ToolBeforeCallContext) -> None:
        parent_seen.append(ctx.agent_name)

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [child_before, child_after, parent_before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="child-agent",
        dispatch_context=_dispatch_context(
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="child-agent",
                requester_id="@user:localhost",
                room_id="!room:localhost",
                thread_id="$thread",
                resolved_thread_id="$resolved-thread",
                session_id=_SESSION_ID,
            ),
        ),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> str:
        return str(kwargs["path"])

    with (
        tool_runtime_context(_tool_runtime_context(tmp_path, agent_name="parent-agent")),
        tool_execution_identity(_execution_identity()),
    ):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == "notes.txt"
    assert before_seen == ["child-agent"]
    assert after_seen == ["child-agent"]
    assert parent_seen == []


@pytest.mark.asyncio
async def test_tool_hook_bridge_does_not_merge_explicit_identity_with_ambient_identity() -> None:
    """An explicit bridge identity should not backfill missing fields from ambient execution state."""
    seen: list[tuple[str | None, str | None, str | None, str | None]] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        seen.append((ctx.room_id, ctx.thread_id, ctx.requester_id, ctx.session_id))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="child-agent",
        dispatch_context=_dispatch_context(
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="child-agent",
                requester_id=None,
                room_id=None,
                thread_id=None,
                resolved_thread_id=None,
                session_id=None,
            ),
        ),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> str:
        return str(kwargs["path"])

    with tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == "notes.txt"
    assert seen == [(None, None, None, None)]


@pytest.mark.asyncio
async def test_tool_hook_bridge_does_not_merge_explicit_identity_with_ambient_runtime_context(
    tmp_path: Path,
) -> None:
    """An explicit bridge identity should not backfill missing fields from ambient runtime context."""
    seen: list[tuple[str | None, str | None, str | None, str | None]] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        seen.append((ctx.room_id, ctx.thread_id, ctx.requester_id, ctx.session_id))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="child-agent",
        dispatch_context=_dispatch_context(
            ToolExecutionIdentity(
                channel="matrix",
                agent_name="child-agent",
                requester_id=None,
                room_id=None,
                thread_id=None,
                resolved_thread_id=None,
                session_id=None,
            ),
        ),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> str:
        return str(kwargs["path"])

    with tool_runtime_context(_tool_runtime_context(tmp_path, agent_name="parent-agent")):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == "notes.txt"
    assert seen == [(None, None, None, None)]


@pytest.mark.asyncio
async def test_tool_hook_bridge_declines_and_skips_real_tool(tmp_path: Path) -> None:
    """A declined before-call hook should return a synthetic result and not call the real tool."""
    after_seen: list[tuple[bool, object | None, BaseException | None]] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        if "secret" in str(ctx.arguments.get("path", "")):
            ctx.decline("secret paths are blocked")

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        after_seen.append((ctx.blocked, ctx.result, ctx.error))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before, after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="should not run")
    with tool_runtime_context(_tool_runtime_context(tmp_path)), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "secret.txt"})

    assert next_func.await_count == 0
    assert result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: read_file\n"
        "Reason: secret paths are blocked\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    assert after_seen == [(True, result, None)]


@pytest.mark.asyncio
async def test_tool_before_call_hooks_run_before_tool_approval_gate(tmp_path: Path) -> None:
    """Policy gates should run before approval requests are emitted."""
    seen: list[str] = []
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=[],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "rules": [
                    {"match": "read_file", "action": "require_approval"},
                ],
            },
        ),
        runtime_paths,
    )
    sender, _ = _initialize_test_approval_store(runtime_paths)

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        del ctx
        seen.append("before")

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> str:
        del kwargs
        seen.append("tool")
        return "ok"

    with tool_execution_identity(_execution_identity()):
        task = asyncio.create_task(bridge("read_file", next_func, {"path": "notes.txt"}))
        await asyncio.sleep(0)
        store = get_approval_store()
        assert store is not None
        pending = await _wait_for_sent_pending(store, sender)
        assert seen == ["before"]

        await _resolve_pending_approval(
            store,
            pending,
            status="approved",
        )
        result = await task

    assert result == "ok"
    assert seen == ["before", "tool"]


@pytest.mark.asyncio
async def test_bridge_workflow_origin_reaches_approval_card(tmp_path: Path) -> None:
    """A bridge built with workflow provenance must surface it on the approval card."""
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={"code": AgentConfig(display_name="Code", role="Help with coding.", rooms=[])},
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={"rules": [{"match": "read_file", "action": "require_approval"}]},
        ),
        runtime_paths,
    )
    sender, _ = _initialize_test_approval_store(runtime_paths)
    bridge = build_tool_hook_bridge(
        HookRegistry.empty(),
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
        workflow_origin=ToolCallWorkflowOrigin(workflow_id="research-report", participant_id="writer"),
    )
    assert bridge is not None

    async def next_func(**kwargs: object) -> str:
        del kwargs
        return "ok"

    with tool_execution_identity(_execution_identity()):
        task = asyncio.create_task(bridge("read_file", next_func, {"path": "notes.txt"}))
        await asyncio.sleep(0)
        store = get_approval_store()
        assert store is not None
        pending = await _wait_for_sent_pending(store, sender)

        card_content = sender.await_args.args[2]
        assert card_content["workflow_id"] == "research-report"
        assert card_content["participant_id"] == "writer"
        assert card_content["body"] == (
            "🔒 Approval required: read_file — Dynamic Workflow 'research-report' participant 'writer'"
        )

        await _resolve_pending_approval(store, pending, status="approved")
        result = await task

    assert result == "ok"


@pytest.mark.asyncio
async def test_tool_before_call_decline_short_circuits_tool_approval(tmp_path: Path) -> None:
    """Denied policy hooks should prevent approval cards from being shown."""
    runtime_paths = test_runtime_paths(tmp_path)
    sender = AsyncMock(return_value=SentApprovalEvent("$approval"))
    initialize_approval_store(runtime_paths, sender=sender, editor=AsyncMock())
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=[],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={"rules": [{"match": "read_file", "action": "require_approval"}]},
        ),
        runtime_paths,
    )

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        ctx.decline("policy blocked the tool")

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="should not run")
    with tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert next_func.await_count == 0
    assert result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: read_file\n"
        "Reason: policy blocked the tool\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    sender.assert_not_awaited()


@pytest.mark.asyncio
async def test_tool_approval_deny_emits_after_call_as_blocked(tmp_path: Path) -> None:
    """Denied approvals should return the declined result and still emit blocked after-call hooks."""
    after_seen: list[tuple[bool, object | None, BaseException | None]] = []
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=[],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={"rules": [{"match": "read_file", "action": "require_approval"}]},
        ),
        runtime_paths,
    )
    sender, _ = _initialize_test_approval_store(runtime_paths)

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        after_seen.append((ctx.blocked, ctx.result, ctx.error))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="should not run")
    with tool_execution_identity(_execution_identity()):
        task = asyncio.create_task(bridge("read_file", next_func, {"path": "notes.txt"}))
        await asyncio.sleep(0)
        store = get_approval_store()
        assert store is not None
        pending = await _wait_for_sent_pending(store, sender)
        await _resolve_pending_approval(
            store,
            pending,
            status="denied",
            reason="Denied by dashboard user.",
        )
        result = await task

    assert next_func.await_count == 0
    assert result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: read_file\n"
        "Reason: Denied by dashboard user.\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    assert after_seen == [(True, result, None)]


@pytest.mark.asyncio
async def test_tool_approval_expiry_emits_after_call_as_blocked(tmp_path: Path) -> None:
    """Expired approvals should return the declined result and emit blocked after-call hooks."""
    after_seen: list[tuple[bool, object | None, BaseException | None]] = []
    runtime_paths = test_runtime_paths(tmp_path)
    config = bind_runtime_paths(
        Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    role="Help with coding.",
                    rooms=[],
                ),
            },
            models={"default": ModelConfig(provider="openai", id="test-model")},
            tool_approval={
                "timeout_days": 0.000001,
                "rules": [{"match": "read_file", "action": "require_approval"}],
            },
        ),
        runtime_paths,
    )
    _initialize_test_approval_store(runtime_paths)

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        after_seen.append((ctx.blocked, ctx.result, ctx.error))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
        config=config,
        runtime_paths=runtime_paths,
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="should not run")
    with tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert next_func.await_count == 0
    assert result == (
        "[TOOL CALL DECLINED]\n"
        "Tool: read_file\n"
        "Reason: Tool approval request timed out.\n\n"
        "Adjust your approach — try a different tool or different arguments."
    )
    assert after_seen == [(True, result, None)]


@pytest.mark.asyncio
async def test_tool_hook_bridge_reraises_tool_errors_after_after_call(tmp_path: Path) -> None:
    """Tool exceptions should still propagate after the after-call hook observes them."""
    after_seen: list[tuple[object | None, BaseException | None, bool]] = []

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        after_seen.append((ctx.result, ctx.error, ctx.blocked))

    registry = HookRegistry.from_plugins([_plugin("tool-audit", [after])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def explode(**kwargs: object) -> object:
        del kwargs
        msg = "boom"
        raise ValueError(msg)

    with (
        tool_runtime_context(_tool_runtime_context(tmp_path)),
        tool_execution_identity(_execution_identity()),
        pytest.raises(ValueError, match="boom"),
    ):
        await bridge("explode", explode, {})

    assert len(after_seen) == 1
    assert after_seen[0][0] is None
    assert isinstance(after_seen[0][1], ValueError)
    assert after_seen[0][2] is False


@pytest.mark.asyncio
async def test_tool_after_call_hooks_cannot_mutate_reraised_non_deepcopyable_error(tmp_path: Path) -> None:
    """After-call hooks should not be able to rewrite the original raised error."""
    seen_error_types: list[type[BaseException]] = []

    @hook(EVENT_TOOL_AFTER_CALL)
    async def rewrite(ctx: ToolAfterCallContext) -> None:
        assert ctx.error is not None
        seen_error_types.append(type(ctx.error))
        ctx.error.args = ("rewritten by hook",)

    registry = HookRegistry.from_plugins([_plugin("tool-audit", [rewrite])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    async def explode(**kwargs: object) -> object:
        del kwargs
        message = "original boom"
        raise NonDeepcopyableToolError(message, {"status": "original"})

    with (
        tool_runtime_context(_tool_runtime_context(tmp_path)),
        tool_execution_identity(_execution_identity()),
        pytest.raises(NonDeepcopyableToolError, match="original boom") as exc_info,
    ):
        await bridge("explode", explode, {})

    assert exc_info.value.args == ("original boom",)
    assert exc_info.value.details == {"status": "original"}
    assert seen_error_types == [Exception]


@pytest.mark.asyncio
async def test_agent_bot_tool_runtime_context_routes_custom_events_from_tool_hooks(tmp_path: Path) -> None:
    """AgentBot-built tool runtime context should deliver custom events and live request session IDs."""
    tool_name = "tool_hooks_runtime_event_tool"
    custom_event_name = "demo:tool_runtime_event"
    seen: list[tuple[str, str, dict[str, object], str | None, str | None, str | None, str]] = []
    before_seen: list[tuple[str | None, str | None, str | None, str | None]] = []

    class DemoToolkit(Toolkit):
        def __init__(self) -> None:
            super().__init__(name="tool-hooks-runtime-demo", tools=[self.echo])

        async def echo(self, text: str) -> str:
            return text.upper()

    original_registry = TOOL_REGISTRY.copy()
    original_metadata = TOOL_METADATA.copy()

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Tool Hooks Runtime Event Tool",
        description="Test tool for runtime hook event delivery.",
        category=ToolCategory.DEVELOPMENT,
    )
    def demo_tool_factory() -> type[DemoToolkit]:
        return DemoToolkit

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        before_seen.append((ctx.room_id, ctx.thread_id, ctx.requester_id, ctx.session_id))
        await emit_custom_event(
            "tool-policy",
            custom_event_name,
            {
                "tool_name": ctx.tool_name,
                "arguments": dict(ctx.arguments),
            },
        )

    @hook(custom_event_name)
    async def on_custom_event(ctx: CustomEventContext) -> None:
        seen.append(
            (
                ctx.event_name,
                ctx.source_plugin,
                dict(ctx.payload),
                ctx.room_id,
                ctx.thread_id,
                ctx.sender_id,
                ctx.plugin_name,
            ),
        )

    plugin_root = tmp_path / "plugins" / "tool-policy"
    plugin_root.mkdir(parents=True)
    (plugin_root / "mindroom.plugin.json").write_text(
        '{"name": "tool-policy", "tools_module": null, "skills": []}',
        encoding="utf-8",
    )
    plugins = [_plugin("tool-policy", [before, on_custom_event])]
    config = _config(tmp_path, tools=[tool_name], plugins=["./plugins/tool-policy"])
    bot = _agent_bot(tmp_path, config=config)
    bot.event_cache = MagicMock()
    bot.hook_registry = HookRegistry.from_plugins(plugins)
    bot.orchestrator = MagicMock(knowledge_managers={}, knowledge_refresh_scheduler=None)

    try:
        with patch("mindroom.model_loading.get_model_instance", return_value=Ollama(id="test-model")):
            target = MessageTarget.resolve(
                room_id="!room:localhost",
                thread_id="$thread",
                reply_to_event_id=None,
            )
            tool_context = bot._tool_runtime_support.build_context(target, user_id="@user:localhost")
            assert tool_context is not None
            assert tool_context.hook_registry.has_hooks(custom_event_name)

            execution_identity = bot._tool_runtime_support.build_execution_identity(
                target=target,
                user_id="@user:localhost",
            )
            toolkit = next(tool for tool in bot.agent.tools if tool.name == "tool-hooks-runtime-demo")
            function = _first_function(toolkit)

            with tool_runtime_context(tool_context), tool_execution_identity(execution_identity):
                result = await FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1").aexecute()

        assert result.status == "success"
        assert result.result == "HI"
        assert before_seen == [
            ("!room:localhost", "$thread", "@user:localhost", target.session_id),
        ]
        assert seen == [
            (
                custom_event_name,
                "tool-policy",
                {"tool_name": "echo", "arguments": {"text": "hi"}},
                "!room:localhost",
                "$thread",
                "@user:localhost",
                "tool-policy",
            ),
        ]
    finally:
        TOOL_REGISTRY.clear()
        TOOL_REGISTRY.update(original_registry)
        TOOL_METADATA.clear()
        TOOL_METADATA.update(original_metadata)


@pytest.mark.asyncio
async def test_emit_custom_event_preserves_message_received_depth_and_bound_room_state_accessors(
    tmp_path: Path,
) -> None:
    """Tool-emitted custom events should preserve chain depth and use bound room-state helpers."""
    custom_event_name = "demo:tool_custom_event"
    sent: list[dict[str, object] | None] = []
    seen: list[tuple[dict[str, object] | None, bool]] = []
    room_state_querier = AsyncMock(return_value={"name": "Lobby"})
    room_state_putter = AsyncMock(return_value=True)

    async def hook_message_sender(
        room_id: str,
        body: str,
        thread_id: str | None,
        source_hook: str,
        extra_content: dict[str, object] | None,
        *,
        trigger_dispatch: bool = False,
    ) -> str | None:
        del room_id, body, thread_id, source_hook, trigger_dispatch
        sent.append(extra_content)
        return "$dispatch-event"

    @hook(custom_event_name)
    async def on_custom_event(ctx: CustomEventContext) -> None:
        query_result = await ctx.query_room_state("!room:localhost", "m.room.name", "")
        put_result = await ctx.put_room_state(
            "!room:localhost",
            "com.mindroom.thread.tags",
            "$resolved-thread",
            {"tags": {"queued": True}},
        )
        seen.append((query_result, put_result))
        event_id = await ctx.send_message("!room:localhost", "dispatch", trigger_dispatch=True)
        assert event_id == "$dispatch-event"

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [on_custom_event])])
    runtime_context = _tool_runtime_context(
        tmp_path,
        hook_message_sender=hook_message_sender,
        room_state_querier=room_state_querier,
        room_state_putter=room_state_putter,
        message_received_depth=1,
        hook_registry=registry,
    )

    with tool_runtime_context(runtime_context):
        await emit_custom_event("tool-policy", custom_event_name, {"item_id": "123"})

    assert seen == [({"name": "Lobby"}, True)]
    room_state_querier.assert_awaited_once_with("!room:localhost", "m.room.name", "")
    room_state_putter.assert_awaited_once_with(
        "!room:localhost",
        "com.mindroom.thread.tags",
        "$resolved-thread",
        {"tags": {"queued": True}},
    )
    assert sent == [
        {
            "com.mindroom.original_sender": "@user:localhost",
            HOOK_MESSAGE_RECEIVED_DEPTH_KEY: 2,
        },
    ]


@pytest.mark.asyncio
async def test_emit_custom_event_ignores_raw_room_mode_thread_id(tmp_path: Path) -> None:
    """Custom tool events should not re-scope room-mode contexts from raw thread provenance."""
    custom_event_name = "demo:room_mode_thread_guard"
    seen_thread_ids: list[str | None] = []

    @hook(custom_event_name)
    async def on_custom_event(ctx: CustomEventContext) -> None:
        seen_thread_ids.append(ctx.thread_id)

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [on_custom_event])])
    base_context = _tool_runtime_context(tmp_path, hook_registry=registry)
    runtime_context = replace(
        base_context,
        target=replace(
            base_context.target,
            source_thread_id="$raw-thread",
            resolved_thread_id=None,
            session_id=create_session_id(base_context.room_id, None),
        ),
    )

    with tool_runtime_context(runtime_context):
        await emit_custom_event("tool-policy", custom_event_name, {"item_id": "123"})

    assert seen_thread_ids == [None]


@pytest.mark.asyncio
async def test_tool_hook_bridge_fails_open_when_before_hook_raises(tmp_path: Path) -> None:
    """A broken before-call hook should not stop the actual tool execution."""

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def broken(ctx: ToolBeforeCallContext) -> None:
        del ctx
        msg = "boom"
        raise RuntimeError(msg)

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [broken])])
    bridge = build_tool_hook_bridge(
        registry,
        agent_name="code",
        dispatch_context=_dispatch_context(_execution_identity()),
    )
    assert bridge is not None

    next_func = AsyncMock(return_value="ok")
    with tool_runtime_context(_tool_runtime_context(tmp_path)), tool_execution_identity(_execution_identity()):
        result = await bridge("read_file", next_func, {"path": "notes.txt"})

    assert result == "ok"
    assert next_func.await_count == 1


@pytest.mark.asyncio
async def test_prepend_tool_hook_bridge_preserves_existing_function_hooks() -> None:
    """The MindRoom bridge should prepend without overwriting existing Agno tool hooks."""
    seen: list[str] = []

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def before(ctx: ToolBeforeCallContext) -> None:
        del ctx
        seen.append("bridge-before")

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [before])])
    bridge = build_tool_hook_bridge(registry, agent_name="code")
    assert bridge is not None

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

        async def echo(self, text: str) -> str:
            seen.append("tool")
            return text

    async def existing_hook(name: str, func: object, args: dict[str, object]) -> object:
        seen.append(f"existing-before:{name}")
        result = await func(**args)
        seen.append("existing-after")
        return result

    toolkit = DemoToolkit()
    function = _first_function(toolkit)
    function.tool_hooks = [existing_hook]
    prepend_tool_hook_bridge(toolkit, bridge)

    assert function.tool_hooks is not None
    assert function.tool_hooks[0] is not existing_hook
    assert callable(function.tool_hooks[0])
    assert function.tool_hooks[1] is existing_hook

    execution = FunctionCall(function=function, arguments={"text": "hi"}, call_id="call-1")
    result = await execution.aexecute()

    assert result.status == "success"
    assert result.result == "hi"
    assert seen == ["bridge-before", "existing-before:echo", "tool", "existing-after"]


@pytest.mark.asyncio
async def test_create_agent_prepends_bridge_to_real_tool_functions(tmp_path: Path) -> None:
    """create_agent() should attach the bridge and provide fallback hook context."""
    tool_name = "tool_hooks_test_tool"
    after_seen: list[tuple[Config | None, object | None, str, bool]] = []

    class DemoToolkit(Toolkit):
        calls: ClassVar[list[str]] = []

        def __init__(self) -> None:
            super().__init__(name="tool-hooks-demo", tools=[self.echo])

        async def echo(self, text: str) -> str:
            self.calls.append(text)
            return text

    original_registry = TOOL_REGISTRY.copy()
    original_metadata = TOOL_METADATA.copy()

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Tool Hooks Test Tool",
        description="Test tool for tool hook interception.",
        category=ToolCategory.DEVELOPMENT,
    )
    def demo_tool_factory() -> type[DemoToolkit]:
        return DemoToolkit

    @hook(EVENT_TOOL_BEFORE_CALL)
    async def block_all(ctx: ToolBeforeCallContext) -> None:
        ctx.decline("blocked in create_agent")

    @hook(EVENT_TOOL_AFTER_CALL)
    async def after(ctx: ToolAfterCallContext) -> None:
        after_seen.append((ctx.config, ctx.runtime_paths, str(ctx.state_root), ctx.blocked))

    registry = HookRegistry.from_plugins([_plugin("tool-policy", [block_all, after])])
    config = _config(tmp_path, tools=[tool_name])
    plugin_state_root = runtime_paths_for(config).storage_root / "plugins" / "tool-policy"

    try:
        with patch("mindroom.model_loading.get_model_instance", return_value=Ollama(id="test-model")):
            agent = create_agent(
                "code",
                config,
                runtime_paths_for(config),
                execution_identity=None,
                hook_registry=registry,
            )

        toolkit = next(tool for tool in agent.tools if tool.name == "tool-hooks-demo")
        function = _first_function(toolkit)
        assert function.tool_hooks is not None
        assert callable(function.tool_hooks[0])

        execution = FunctionCall(function=function, arguments={"text": "hello"}, call_id="call-1")
        result = await execution.aexecute()

        assert result.status == "success"
        assert result.result == (
            "[TOOL CALL DECLINED]\n"
            "Tool: echo\n"
            "Reason: blocked in create_agent\n\n"
            "Adjust your approach — try a different tool or different arguments."
        )
        assert DemoToolkit.calls == []
        assert after_seen == [(config, runtime_paths_for(config), str(plugin_state_root), True)]
    finally:
        TOOL_REGISTRY.clear()
        TOOL_REGISTRY.update(original_registry)
        TOOL_METADATA.clear()
        TOOL_METADATA.update(original_metadata)
