"""Tests for generic sandbox proxy tool wrapping."""

from __future__ import annotations

import ast
import asyncio
import base64
import hashlib
import json
import os
import stat
import sys
import threading
from dataclasses import asdict
from pathlib import Path
from types import SimpleNamespace
from typing import TYPE_CHECKING, Any, Self
from unittest.mock import patch

import httpx
import pytest
from agno.agent import Agent
from agno.agent._tools import parse_tools
from agno.models.base import Model
from agno.models.response import ModelResponse
from agno.tools import Toolkit
from agno.tools.function import Function, FunctionCall

import mindroom.api.sandbox_exec as sandbox_exec_module
import mindroom.api.sandbox_runner as sandbox_runner_module
import mindroom.tool_system.sandbox_proxy as sandbox_proxy_module
import mindroom.tools  # noqa: F401
import mindroom.tools.shell as shell_tool_module
from mindroom.config.agent import AgentConfig, AgentPrivateConfig
from mindroom.config.main import Config, load_config
from mindroom.constants import (
    RuntimePaths,
    build_execution_tool_env,
    isolated_runtime_paths,
    resolve_runtime_paths,
    shell_execution_runtime_env_values,
    shell_extra_env_values,
    subprocess_path_with_prepends,
)
from mindroom.credentials import get_runtime_credentials_manager, save_scoped_credentials
from mindroom.hooks import HookRegistry
from mindroom.message_target import MessageTarget
from mindroom.runtime_env_policy import VENDOR_TELEMETRY_ENV_VALUES
from mindroom.tool_system.metadata import (
    TOOL_METADATA,
    TOOL_REGISTRY,
    ConfigField,
    ToolCategory,
    ToolExecutionTarget,
    ToolInitOverrideError,
    ToolMetadata,
    ToolValidationInfo,
    get_tool_by_name,
    resolved_tool_validation_snapshot_for_runtime,
    serialize_tool_validation_snapshot,
)
from mindroom.tool_system.output_files import OUTPUT_PATH_ARGUMENT
from mindroom.tool_system.registration import register_tool_with_metadata
from mindroom.tool_system.runtime_context import ToolRuntimeContext, tool_runtime_context, worker_progress_pump_scope
from mindroom.tool_system.tool_hooks import build_tool_hook_bridge, prepend_tool_hook_bridge
from mindroom.tool_system.worker_proxy_client import WorkerProxyClientConfig, execute_worker_proxy_request
from mindroom.tool_system.worker_routing import (
    ResolvedWorkerTarget,
    ToolExecutionIdentity,
    agent_workspace_root_path,
    resolve_worker_key,
    resolve_worker_target,
)
from mindroom.workers import runtime as workers_runtime_module
from mindroom.workers.backend import WorkerBackendError
from mindroom.workers.backends.local import local_worker_state_paths_for_root
from mindroom.workers.backends.static_runner import StaticSandboxRunnerBackend
from mindroom.workers.models import WorkerHandle, WorkerReadyProgress, WorkerSpec
from tests.conftest import FakeCredentialsManager, make_conversation_cache_mock, make_event_cache_mock, requires_linux

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Callable, Iterator

_TEST_AUTH_TOKEN = "test-token"  # noqa: S105
_TEST_KUBERNETES_VALIDATION_SNAPSHOT = {
    "calculator": {
        "config_fields": [],
        "agent_override_fields": [],
        "authored_override_validator": "default",
        "runtime_loadable": True,
    },
}
_TEST_KUBERNETES_CONFIG_SNAPSHOT: dict[str, object] = {
    "defaults": {"worker_scope": None},
    "agents": {},
    "knowledge_bases": {},
}
_TEST_RUNTIME_PATHS = resolve_runtime_paths(config_path=Path("config.yaml"), process_env={})


def test_approved_egress_tool_stays_primary_even_when_sandbox_mode_all(tmp_path: Path) -> None:
    """Policy API grants must be requested from the primary runtime, not a worker."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        process_env={
            "MINDROOM_WORKER_BACKEND": "kubernetes",
            "MINDROOM_SANDBOX_EXECUTION_MODE": "all",
            "MINDROOM_SANDBOX_PROXY_TOKEN": "test-token",
        },
    )

    assert not sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
        "approved_egress",
        runtime_paths=runtime_paths,
    )


def test_attachment_save_protocol_payload_fields_round_trip() -> None:
    """Attachment save payload helpers should preserve the current wire fields."""
    payload_bytes = b"attachment-bytes"
    fields = sandbox_proxy_module._attachment_save_payload_fields(payload_bytes)

    assert fields == {
        "bytes_b64": base64.b64encode(payload_bytes).decode("ascii"),
        "sha256": hashlib.sha256(payload_bytes).hexdigest(),
        "size_bytes": len(payload_bytes),
    }
    assert (
        sandbox_proxy_module.decode_attachment_save_bytes(
            bytes_b64=fields["bytes_b64"],
            sha256=fields["sha256"],
            size_bytes=fields["size_bytes"],
        )
        == payload_bytes
    )


@pytest.mark.parametrize(
    ("bytes_b64", "sha256", "size_bytes", "expected_error"),
    [
        ("abc", "0" * 64, True, "Attachment size_bytes must be a non-negative integer."),
        ("not valid base64", "0" * 64, None, "Attachment bytes are not valid base64."),
        (
            base64.b64encode(b"payload").decode("ascii"),
            hashlib.sha256(b"payload").hexdigest(),
            999,
            "Attachment byte length does not match the request receipt.",
        ),
        (
            base64.b64encode(b"payload").decode("ascii"),
            "0" * 64,
            len(b"payload"),
            "Attachment SHA256 does not match the request payload.",
        ),
    ],
)
def test_attachment_save_protocol_decode_preserves_error_strings(
    bytes_b64: str,
    sha256: str,
    size_bytes: int | None,
    expected_error: str,
) -> None:
    """Malformed attachment payloads should keep the current runner error text."""
    assert (
        sandbox_proxy_module.decode_attachment_save_bytes(
            bytes_b64=bytes_b64,
            sha256=sha256,
            size_bytes=size_bytes,
        )
        == expected_error
    )


@pytest.mark.parametrize(
    ("response_payload", "expected_error"),
    [
        (
            {"worker_path": "sample.bin", "size_bytes": True, "sha256": "0" * 64},
            "Sandbox save-attachment response is missing its receipt fields.",
        ),
        (
            {"worker_path": "other.bin", "size_bytes": 16, "sha256": "0" * 64},
            "Sandbox save-attachment response path does not match the requested workspace path.",
        ),
        (
            {"worker_path": "sample.bin", "size_bytes": 999, "sha256": "0" * 64},
            "Sandbox save-attachment response size does not match the sent bytes.",
        ),
        (
            {"worker_path": "sample.bin", "size_bytes": 16, "sha256": "0" * 64},
            "Sandbox save-attachment response SHA256 does not match the sent bytes.",
        ),
    ],
)
def test_attachment_save_protocol_receipt_validation_preserves_error_strings(
    response_payload: dict[str, object],
    expected_error: str,
) -> None:
    """Bad worker receipts should keep the current primary-side error text."""
    payload_bytes = b"attachment-bytes"
    result = sandbox_proxy_module._validate_attachment_save_receipt(
        response_payload,
        requested_path="sample.bin",
        byte_count=len(payload_bytes),
        sha256=hashlib.sha256(payload_bytes).hexdigest(),
    )

    assert result == expected_error


def test_attachment_save_protocol_receipt_validation_returns_dataclass() -> None:
    """Valid worker receipts should keep the current receipt field names."""
    payload_bytes = b"attachment-bytes"
    sha256 = hashlib.sha256(payload_bytes).hexdigest()

    receipt = sandbox_proxy_module._validate_attachment_save_receipt(
        {"worker_path": "sample.bin", "size_bytes": len(payload_bytes), "sha256": sha256},
        requested_path="sample.bin",
        byte_count=len(payload_bytes),
        sha256=sha256,
    )

    assert receipt == sandbox_proxy_module.WorkerAttachmentSaveReceipt(
        worker_path="sample.bin",
        size_bytes=len(payload_bytes),
        sha256=sha256,
    )


def _runtime_paths_from_env() -> RuntimePaths:
    return resolve_runtime_paths(config_path=Path("config.yaml"), process_env=dict(os.environ))


def _configure_proxy_runtime(
    monkeypatch: pytest.MonkeyPatch,
    *,
    proxy_url: str | None,
    proxy_token: str | None = _TEST_AUTH_TOKEN,
    execution_mode: str | None = "all",
    runner_mode: bool = False,
    proxy_tools: set[str] | None = None,
    credential_policy: dict[str, tuple[str, ...]] | None = None,
) -> RuntimePaths:
    if proxy_url is None:
        monkeypatch.delenv("MINDROOM_SANDBOX_PROXY_URL", raising=False)
    else:
        monkeypatch.setenv("MINDROOM_SANDBOX_PROXY_URL", proxy_url)
    if proxy_token is None:
        monkeypatch.delenv("MINDROOM_SANDBOX_PROXY_TOKEN", raising=False)
    else:
        monkeypatch.setenv("MINDROOM_SANDBOX_PROXY_TOKEN", proxy_token)
    if execution_mode is None:
        monkeypatch.delenv("MINDROOM_SANDBOX_EXECUTION_MODE", raising=False)
    else:
        monkeypatch.setenv("MINDROOM_SANDBOX_EXECUTION_MODE", execution_mode)
    monkeypatch.setenv("MINDROOM_SANDBOX_RUNNER_MODE", "true" if runner_mode else "false")
    if proxy_tools is None:
        monkeypatch.delenv("MINDROOM_SANDBOX_PROXY_TOOLS", raising=False)
    else:
        monkeypatch.setenv("MINDROOM_SANDBOX_PROXY_TOOLS", ",".join(sorted(proxy_tools)))
    if credential_policy is None:
        monkeypatch.delenv("MINDROOM_SANDBOX_CREDENTIAL_POLICY_JSON", raising=False)
    else:
        monkeypatch.setenv(
            "MINDROOM_SANDBOX_CREDENTIAL_POLICY_JSON",
            json.dumps({key: list(value) for key, value in credential_policy.items()}),
        )
    return _runtime_paths_from_env()


def _worker_target(
    runtime_paths: RuntimePaths,
    worker_scope: str | None,
    routing_agent_name: str | None,
    execution_identity: ToolExecutionIdentity | None,
    *,
    private_agent_names: frozenset[str] | None = None,
) -> ResolvedWorkerTarget:
    return resolve_worker_target(
        worker_scope,
        routing_agent_name,
        execution_identity=execution_identity,
        tenant_id=runtime_paths.env_value("CUSTOMER_ID"),
        account_id=runtime_paths.env_value("ACCOUNT_ID"),
        private_agent_names=private_agent_names,
    )


class _FakeResponse:
    def __init__(self, payload: object | None = None) -> None:
        self._payload = payload or {"ok": True, "result": "sandbox-result"}

    def raise_for_status(self) -> None:
        return

    def json(self) -> object:
        return self._payload


class _TrackingWorkerManager:
    def __init__(self) -> None:
        self.touched: list[str] = []
        self.failures: list[tuple[str, str]] = []

    def ensure_worker(self, spec: WorkerSpec, *, now: float | None = None, progress_sink: object = None) -> object:
        del now, progress_sink
        return WorkerHandle(
            worker_id="worker-1",
            worker_key=spec.worker_key,
            endpoint="http://worker/api/sandbox-runner/execute",
            auth_token=_TEST_AUTH_TOKEN,
            status="ready",
            backend_name="kubernetes",
            last_used_at=0.0,
            created_at=0.0,
        )

    def touch_worker(self, worker_key: str, *, now: float | None = None) -> object:
        del now
        self.touched.append(worker_key)
        return None

    def record_failure(self, worker_key: str, failure_reason: str, *, now: float | None = None) -> object:
        del now
        self.failures.append((worker_key, failure_reason))
        return None


def _static_worker_manager_lease(manager: object) -> object:
    class _StaticLease:
        def __enter__(self) -> object:
            return manager

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return None

    return _StaticLease()


class _HttpStatusErrorResponse:
    def __init__(self, url: str, *, status_code: int, payload: dict[str, object] | None = None) -> None:
        self.url = url
        self.status_code = status_code
        self.payload = payload

    def raise_for_status(self) -> None:
        request = httpx.Request("POST", self.url)
        response = (
            httpx.Response(self.status_code, request=request, json=self.payload)
            if self.payload is not None
            else httpx.Response(self.status_code, request=request)
        )
        message = f"{self.status_code} client error"
        raise httpx.HTTPStatusError(message, request=request, response=response)

    def json(self) -> dict[str, object]:
        return self.payload or {}


def _http_status_client_class(*, status_code: int, payload: dict[str, object] | None = None) -> type:
    class _HttpStatusClient:
        def __init__(self, *, timeout: float) -> None:
            del timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _HttpStatusErrorResponse:
            del json, headers
            assert url == "http://worker/api/sandbox-runner/execute"
            return _HttpStatusErrorResponse(url, status_code=status_code, payload=payload)

    return _HttpStatusClient


def _recording_client_class(
    *,
    captured: dict[str, Any] | None = None,
    captured_calls: list[tuple[str, dict[str, Any]]] | None = None,
    responder: Callable[[str, dict[str, Any]], dict[str, object]] | None = None,
) -> type:
    class _FakeClient:
        def __init__(self, *, timeout: float) -> None:
            if captured is not None:
                captured["timeout"] = timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            if captured is not None:
                captured["url"] = url
                captured["json"] = json
                captured["headers"] = headers
            if captured_calls is not None:
                captured_calls.append((url, json))
            payload = responder(url, json) if responder is not None else {"ok": True, "result": "sandbox-result"}
            return _FakeResponse(payload)

    return _FakeClient


def test_worker_proxy_client_records_worker_success() -> None:
    """Worker proxy HTTP details should live behind one focused client seam."""
    captured: dict[str, Any] = {}
    manager = _TrackingWorkerManager()
    handle = WorkerHandle(
        worker_id="worker-1",
        worker_key="agent:test",
        endpoint="http://worker/api/sandbox-runner/execute",
        auth_token=_TEST_AUTH_TOKEN,
        status="ready",
        backend_name="docker",
        last_used_at=0.0,
        created_at=0.0,
    )

    result = execute_worker_proxy_request(
        config=WorkerProxyClientConfig(
            proxy_url=None,
            proxy_token=None,
            proxy_timeout_seconds=7.0,
            credential_lease_ttl_seconds=60,
            credential_policy={},
        ),
        payload={"tool_name": "shell", "function_name": "run_shell_command"},
        credentials_manager=None,
        tool_name="shell",
        function_name="run_shell_command",
        worker_target=None,
        worker_handle=handle,
        worker_manager=manager,
        client_factory=_recording_client_class(captured=captured),
    )

    assert result == "sandbox-result"
    assert captured["url"] == "http://worker/api/sandbox-runner/execute"
    assert captured["headers"] == {"x-mindroom-sandbox-token": _TEST_AUTH_TOKEN}
    assert captured["timeout"] == 7.0
    assert manager.touched == ["agent:test"]
    assert manager.failures == []


def _run_worker_proxy_request_with_exception(
    exception_factory: Callable[[httpx.Request], Exception],
) -> _TrackingWorkerManager:
    manager = _TrackingWorkerManager()
    handle = WorkerHandle(
        worker_id="worker-1",
        worker_key="agent:test",
        endpoint="http://worker/api/sandbox-runner/execute",
        auth_token=_TEST_AUTH_TOKEN,
        status="ready",
        backend_name="kubernetes",
        last_used_at=0.0,
        created_at=0.0,
    )

    class _TimeoutClient:
        def __init__(self, *, timeout: float) -> None:
            _ = timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            _ = json, headers
            request = httpx.Request("POST", url)
            raise exception_factory(request)

    with pytest.raises(httpx.TransportError):
        execute_worker_proxy_request(
            config=WorkerProxyClientConfig(
                proxy_url=None,
                proxy_token=None,
                proxy_timeout_seconds=7.0,
                credential_lease_ttl_seconds=60,
                credential_policy={},
            ),
            payload={"tool_name": "shell", "function_name": "run_shell_command"},
            credentials_manager=None,
            tool_name="shell",
            function_name="run_shell_command",
            worker_target=None,
            worker_handle=handle,
            worker_manager=manager,
            client_factory=_TimeoutClient,
        )

    return manager


@pytest.mark.parametrize(
    "exception_factory",
    [
        lambda request: httpx.ConnectError("[Errno 110] Connection timed out", request=request),
        lambda request: httpx.ReadTimeout("timed out", request=request),
        lambda request: httpx.WriteTimeout("timed out", request=request),
    ],
)
def test_worker_proxy_transport_timeout_keeps_ready_worker_alive(
    exception_factory: Callable[[httpx.Request], Exception],
) -> None:
    """A transient proxy transport timeout must not evict the shared worker for sibling streams."""
    manager = _run_worker_proxy_request_with_exception(exception_factory)

    assert manager.touched == ["agent:test"]
    assert manager.failures == []


@pytest.mark.parametrize(
    ("exception_factory", "failure_reason"),
    [
        (
            lambda request: httpx.ConnectError("connection refused", request=request),
            "connection refused",
        ),
        (
            lambda request: httpx.RemoteProtocolError("malformed worker response", request=request),
            "malformed worker response",
        ),
        (
            lambda request: httpx.ReadError("worker read failed", request=request),
            "worker read failed",
        ),
        (
            lambda request: httpx.WriteError("worker write failed", request=request),
            "worker write failed",
        ),
    ],
)
def test_worker_proxy_malformed_transport_errors_record_worker_failure(
    exception_factory: Callable[[httpx.Request], Exception],
    failure_reason: str,
) -> None:
    """Malformed proxy transport failures must evict the broken shared worker."""
    manager = _run_worker_proxy_request_with_exception(exception_factory)

    assert manager.touched == []
    assert manager.failures == [("agent:test", failure_reason)]


class _MinimalModel(Model):
    """Minimal model surface for exercising Agno's async tool execution path."""

    def invoke(self, *_args: object, **_kwargs: object) -> ModelResponse:
        return ModelResponse(content="ok")

    async def ainvoke(self, *_args: object, **_kwargs: object) -> ModelResponse:
        return ModelResponse(content="ok")

    def invoke_stream(self, *_args: object, **_kwargs: object) -> Iterator[ModelResponse]:
        yield ModelResponse(content="ok")

    async def ainvoke_stream(self, *_args: object, **_kwargs: object) -> AsyncIterator[ModelResponse]:
        yield ModelResponse(content="ok")

    def _parse_provider_response(self, response: ModelResponse, *_args: object, **_kwargs: object) -> ModelResponse:
        return response

    def _parse_provider_response_delta(
        self,
        response: ModelResponse,
        *_args: object,
        **_kwargs: object,
    ) -> ModelResponse:
        return response


def test_proxy_wraps_tool_calls(monkeypatch: pytest.MonkeyPatch) -> None:
    """Tool entrypoints should call the sandbox runner API when proxy mode is enabled."""
    captured: dict[str, Any] = {}

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"calculator"},
        credential_policy={},
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(captured=captured),
    )

    tool = get_tool_by_name("calculator", runtime_paths, worker_target=None)
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    result = entrypoint(1, 2)

    assert result == "sandbox-result"
    assert captured["url"] == "http://sandbox-runner:8765/api/sandbox-runner/execute"
    assert captured["json"] == {
        "tool_name": "calculator",
        "function_name": "add",
        "args": [1, 2],
        "kwargs": {},
    }
    assert captured["headers"] == {"x-mindroom-sandbox-token": "test-token"}


def test_sandbox_proxy_schema_keeps_mindroom_output_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
    """Proxy wrapping should preserve the augmented output-path schema and forward the kwarg."""
    captured: dict[str, Any] = {}

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"calculator"},
        credential_policy={},
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(captured=captured),
    )

    tool = get_tool_by_name(
        "calculator",
        runtime_paths,
        tool_output_workspace_root=tmp_path,
        worker_target=None,
    )
    function = tool.functions["add"].model_copy(deep=True)
    function.process_entrypoint()

    output_schema = function.parameters["properties"][OUTPUT_PATH_ARGUMENT]
    assert output_schema["description"].startswith("Optional")
    assert output_schema["default"] is None
    assert OUTPUT_PATH_ARGUMENT not in function.parameters["required"]
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None

    result = entrypoint(1, 2, mindroom_output_path="tool-results/add.json")

    assert result == "sandbox-result"
    assert captured["json"]["args"] == [1, 2]
    assert captured["json"]["kwargs"] == {OUTPUT_PATH_ARGUMENT: "tool-results/add.json"}


def test_static_proxy_payload_carries_agent_identity_for_output_redirect(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Default static-runner proxy requests should carry enough identity for runner-side wrapping."""
    captured: dict[str, Any] = {}
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="general",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"calculator"},
        credential_policy={},
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(captured=captured),
    )

    tool = get_tool_by_name(
        "calculator",
        runtime_paths,
        tool_output_workspace_root=tmp_path,
        worker_target=_worker_target(runtime_paths, None, "general", execution_identity),
    )
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None

    result = entrypoint(1, 2, mindroom_output_path="tool-results/add.json")

    assert result == "sandbox-result"
    assert captured["json"]["routing_agent_name"] == "general"
    assert captured["json"]["execution_identity"] == asdict(execution_identity)


def test_sandbox_runner_executes_wrapper_before_to_json_compatible(tmp_path: Path) -> None:
    """Runner-side wrapping should save raw output before proxy response serialization."""
    tool_name = "test_runner_output_redirect"
    marker = "ISSUE200_RUNNER_MARKER"

    class _RunnerRedirectToolkit(Toolkit):
        def __init__(self) -> None:
            super().__init__(name=tool_name, tools=[self.large])

        def large(self) -> dict[str, str]:
            return {"marker": marker * 20}

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Runner Redirect",
        description="Test-only runner redirect coverage.",
        category=ToolCategory.DEVELOPMENT,
    )
    def _runner_redirect_factory() -> type[_RunnerRedirectToolkit]:
        return _RunnerRedirectToolkit

    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={},
    )
    try:
        _toolkit, entrypoint = sandbox_runner_module._resolve_entrypoint(
            runtime_paths=runtime_paths,
            config=Config(agents={}, models={}),
            tool_name=tool_name,
            function_name="large",
            tool_output_workspace_root=tmp_path,
        )

        result = entrypoint(mindroom_output_path="runner/result.json")
        proxy_result = sandbox_runner_module.to_json_compatible(result)

        assert marker in (tmp_path / "runner/result.json").read_text(encoding="utf-8")
        assert marker not in str(proxy_result)
        assert proxy_result["mindroom_tool_output"]["status"] == "saved_to_file"
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


@pytest.mark.asyncio
@requires_linux(reason="local worker venv bootstrap is validated on Linux", timeout=180)
async def test_sandbox_runner_save_attachment_writes_worker_workspace(tmp_path: Path) -> None:
    """The runner save endpoint should validate and atomically write bytes under the worker workspace."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={"MINDROOM_SANDBOX_PROXY_TOKEN": _TEST_AUTH_TOKEN},
    )
    config = Config(agents={"code": AgentConfig(display_name="Code", memory_backend="file")}, models={})
    app = SimpleNamespace(
        state=SimpleNamespace(
            sandbox_runner_context=sandbox_runner_module._SandboxRunnerContext(
                runtime_paths=runtime_paths,
                config=config,
                tool_metadata=TOOL_METADATA.copy(),
                runner_token=_TEST_AUTH_TOKEN,
            ),
        ),
    )
    payload_bytes = b"worker-bytes"
    sha256 = hashlib.sha256(payload_bytes).hexdigest()

    response = await sandbox_runner_module.save_attachment_to_worker(
        SimpleNamespace(app=app),
        sandbox_runner_module.SandboxRunnerSaveAttachmentRequest(
            worker_key="worker-test",
            attachment_id="att_sample",
            mindroom_output_path="inputs/sample.bin",
            sha256=sha256,
            size_bytes=len(payload_bytes),
            bytes_b64=base64.b64encode(payload_bytes).decode("ascii"),
        ),
    )

    assert response.ok is True
    assert response.worker_path == "inputs/sample.bin"
    saved_path = next(runtime_paths.storage_root.rglob("sample.bin"))
    assert saved_path.read_bytes() == payload_bytes
    assert saved_path.is_relative_to(runtime_paths.storage_root)
    assert stat.S_IMODE(saved_path.stat().st_mode) == 0o600
    assert response.size_bytes == len(payload_bytes)
    assert response.sha256 == sha256


@pytest.mark.asyncio
async def test_sandbox_runner_save_attachment_rejects_sha_mismatch_and_unsafe_path(tmp_path: Path) -> None:
    """The save endpoint should reject mismatched bytes and unsafe output paths without writing files."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={"MINDROOM_SANDBOX_PROXY_TOKEN": _TEST_AUTH_TOKEN},
    )
    config = Config(agents={"code": AgentConfig(display_name="Code", memory_backend="file")}, models={})
    app = SimpleNamespace(
        state=SimpleNamespace(
            sandbox_runner_context=sandbox_runner_module._SandboxRunnerContext(
                runtime_paths=runtime_paths,
                config=config,
                tool_metadata=TOOL_METADATA.copy(),
                runner_token=_TEST_AUTH_TOKEN,
            ),
        ),
    )
    request = SimpleNamespace(app=app)
    payload_bytes = b"worker-bytes"
    encoded = base64.b64encode(payload_bytes).decode("ascii")

    mismatch = await sandbox_runner_module.save_attachment_to_worker(
        request,
        sandbox_runner_module.SandboxRunnerSaveAttachmentRequest(
            worker_key="worker-test",
            attachment_id="att_sample",
            mindroom_output_path="inputs/sample.bin",
            sha256="0" * 64,
            size_bytes=len(payload_bytes),
            bytes_b64=encoded,
        ),
    )
    unsafe = await sandbox_runner_module.save_attachment_to_worker(
        request,
        sandbox_runner_module.SandboxRunnerSaveAttachmentRequest(
            worker_key="worker-test",
            attachment_id="att_sample",
            mindroom_output_path="../escape.bin",
            sha256=hashlib.sha256(payload_bytes).hexdigest(),
            size_bytes=len(payload_bytes),
            bytes_b64=encoded,
        ),
    )

    assert mismatch.ok is False
    assert mismatch.failure_kind == "tool"
    assert "SHA256" in (mismatch.error or "")
    assert unsafe.ok is False
    assert unsafe.failure_kind == "tool"
    assert "mindroom_output_path" in (unsafe.error or "")
    assert not list(runtime_paths.storage_root.rglob("sample.bin"))
    assert not list(runtime_paths.storage_root.rglob("escape.bin"))


@pytest.mark.asyncio
async def test_sandbox_runner_save_attachment_rejects_unsafe_path_before_decoding(tmp_path: Path) -> None:
    """Unsafe save paths should be rejected before malformed request bytes are decoded."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={"MINDROOM_SANDBOX_PROXY_TOKEN": _TEST_AUTH_TOKEN},
    )
    config = Config(agents={"code": AgentConfig(display_name="Code")}, models={})
    app = SimpleNamespace(
        state=SimpleNamespace(
            sandbox_runner_context=sandbox_runner_module._SandboxRunnerContext(
                runtime_paths=runtime_paths,
                config=config,
                tool_metadata=TOOL_METADATA.copy(),
                runner_token=_TEST_AUTH_TOKEN,
            ),
        ),
    )

    response = await sandbox_runner_module.save_attachment_to_worker(
        SimpleNamespace(app=app),
        sandbox_runner_module.SandboxRunnerSaveAttachmentRequest(
            worker_key="worker-test",
            attachment_id="att_sample",
            mindroom_output_path="../escape.bin",
            sha256="0" * 64,
            size_bytes=10,
            bytes_b64="not valid base64",
        ),
    )

    assert response.ok is False
    assert response.failure_kind == "tool"
    assert "mindroom_output_path" in (response.error or "")


@pytest.mark.asyncio
async def test_sandbox_runner_save_attachment_supports_static_unkeyed_workspace(tmp_path: Path) -> None:
    """Static runner saves should mirror execute output redirection without requiring worker_key."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={"MINDROOM_SANDBOX_PROXY_TOKEN": _TEST_AUTH_TOKEN},
    )
    config = Config(agents={"code": AgentConfig(display_name="Code", memory_backend="file")}, models={})
    app = SimpleNamespace(
        state=SimpleNamespace(
            sandbox_runner_context=sandbox_runner_module._SandboxRunnerContext(
                runtime_paths=runtime_paths,
                config=config,
                tool_metadata=TOOL_METADATA.copy(),
                runner_token=_TEST_AUTH_TOKEN,
            ),
        ),
    )
    payload_bytes = b"static-runner"
    sha256 = hashlib.sha256(payload_bytes).hexdigest()

    response = await sandbox_runner_module.save_attachment_to_worker(
        SimpleNamespace(app=app),
        sandbox_runner_module.SandboxRunnerSaveAttachmentRequest(
            routing_agent_name="code",
            attachment_id="att_sample",
            mindroom_output_path="inputs/static.bin",
            sha256=sha256,
            size_bytes=len(payload_bytes),
            bytes_b64=base64.b64encode(payload_bytes).decode("ascii"),
        ),
    )

    workspace_root = agent_workspace_root_path(runtime_paths.storage_root, "code")
    assert response.ok is True
    assert response.worker_path == "inputs/static.bin"
    assert (workspace_root / "inputs" / "static.bin").read_bytes() == payload_bytes


@pytest.mark.asyncio
async def test_static_runner_redirect_resolves_agent_workspace_without_prepared_worker(tmp_path: Path) -> None:
    """Static runner redirects should rebuild with the routing agent workspace root."""
    tool_name = "test_static_runner_output_redirect"
    marker = "ISSUE200_STATIC_RUNNER_MARKER"

    class _StaticRunnerRedirectToolkit(Toolkit):
        def __init__(self) -> None:
            super().__init__(name=tool_name, tools=[self.large])

        def large(self) -> str:
            return marker * 20

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Static Runner Redirect",
        description="Test-only static runner redirect coverage.",
        category=ToolCategory.DEVELOPMENT,
    )
    def _static_runner_redirect_factory() -> type[_StaticRunnerRedirectToolkit]:
        return _StaticRunnerRedirectToolkit

    storage_root = tmp_path / "storage"
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=storage_root,
        process_env={},
    )
    config = Config(
        agents={"general": AgentConfig(display_name="General", memory_backend="file")},
        models={},
    )
    request = sandbox_runner_module.SandboxRunnerExecuteRequest(
        tool_name=tool_name,
        function_name="large",
        kwargs={OUTPUT_PATH_ARGUMENT: "tool-results/static.txt"},
        routing_agent_name="general",
    )
    try:
        response = await sandbox_runner_module._execute_request_inprocess(request, runtime_paths, config)

        workspace = agent_workspace_root_path(storage_root, "general")
        output_path = workspace / "tool-results/static.txt"
        assert response.ok is True
        assert marker in output_path.read_text(encoding="utf-8")
        assert marker not in str(response.result)
        assert response.result["mindroom_tool_output"]["status"] == "saved_to_file"
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


@pytest.mark.asyncio
async def test_worker_redirect_uses_agent_workspace_not_worker_scratch(tmp_path: Path) -> None:
    """Worker-routed redirects should save where later agent file tools can read them."""
    tool_name = "test_worker_output_redirect"
    marker = "ISSUE200_WORKER_REDIRECT_MARKER"

    class _WorkerRedirectToolkit(Toolkit):
        def __init__(self) -> None:
            super().__init__(name=tool_name, tools=[self.large])

        def large(self) -> dict[str, str]:
            return {"marker": marker}

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Worker Redirect",
        description="Test-only worker redirect coverage.",
        category=ToolCategory.DEVELOPMENT,
    )
    def _worker_redirect_factory() -> type[_WorkerRedirectToolkit]:
        return _WorkerRedirectToolkit

    storage_root = tmp_path / "storage"
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=storage_root,
        process_env={},
    )
    config = Config(
        agents={"general": AgentConfig(display_name="General", memory_backend="file")},
        models={},
    )
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="general",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    worker_key = resolve_worker_key("shared", execution_identity, agent_name="general")
    worker_paths = local_worker_state_paths_for_root(tmp_path / "worker-scratch")
    worker_paths.workspace.mkdir(parents=True, exist_ok=True)
    prepared_worker = sandbox_runner_module.sandbox_worker_prep.PreparedWorkerRequest(
        handle=WorkerHandle(
            worker_id="worker-1",
            worker_key=worker_key,
            endpoint="/api/sandbox-runner/execute",
            auth_token=_TEST_AUTH_TOKEN,
            status="ready",
            backend_name="local",
            last_used_at=0.0,
            created_at=0.0,
        ),
        paths=worker_paths,
        runtime_overrides={"base_dir": worker_paths.workspace},
    )
    request = sandbox_runner_module.SandboxRunnerExecuteRequest(
        tool_name=tool_name,
        function_name="large",
        kwargs={OUTPUT_PATH_ARGUMENT: "tool-results/worker.json"},
        worker_key=worker_key,
        worker_scope="shared",
        routing_agent_name="general",
        execution_identity=asdict(execution_identity),
    )
    try:
        response = await sandbox_runner_module._execute_request_inprocess(
            request,
            runtime_paths,
            config,
            prepared_worker=prepared_worker,
        )

        workspace = agent_workspace_root_path(storage_root, "general")
        output_path = workspace / "tool-results/worker.json"
        assert response.ok is True
        assert marker in output_path.read_text(encoding="utf-8")
        assert not (worker_paths.workspace / "tool-results/worker.json").exists()
        assert response.result["mindroom_tool_output"]["path"] == "tool-results/worker.json"
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_proxy_payload_includes_tool_config_overrides(monkeypatch: pytest.MonkeyPatch) -> None:
    """Proxy execution should forward authored tool config overrides separately from runtime init overrides."""
    captured: dict[str, Any] = {}
    tool_name = "test_proxy_configured_tool"

    class _ConfiguredToolkit(Toolkit):
        def __init__(self, label: str | None = None) -> None:
            self.label = label
            super().__init__(name=tool_name, tools=[self.ping])

        def ping(self) -> str:
            return self.label or "local"

    @register_tool_with_metadata(
        name=tool_name,
        display_name="Proxy Configured Tool",
        description="Test-only proxy payload coverage.",
        category=ToolCategory.DEVELOPMENT,
        config_fields=[ConfigField(name="label", label="Label", type="text", required=False)],
    )
    def _configured_tool_factory() -> type[_ConfiguredToolkit]:
        return _ConfiguredToolkit

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={tool_name},
        credential_policy={},
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(captured=captured),
    )

    try:
        tool = get_tool_by_name(
            tool_name,
            runtime_paths,
            tool_config_overrides={"label": "from-config"},
            worker_target=None,
        )
        entrypoint = tool.functions["ping"].entrypoint
        assert entrypoint is not None
        result = entrypoint()

        assert result == "sandbox-result"
        assert captured["json"]["tool_config_overrides"] == {"label": "from-config"}
        assert "tool_init_overrides" not in captured["json"]
    finally:
        TOOL_REGISTRY.pop(tool_name, None)
        TOOL_METADATA.pop(tool_name, None)


def test_proxy_disabled_in_runner_mode(monkeypatch: pytest.MonkeyPatch) -> None:
    """Runner mode must execute tools locally to avoid proxy recursion."""

    class _ForbiddenClient:
        def __init__(self, *_args: object, **_kwargs: object) -> None:
            msg = "Proxy client should not be used in runner mode."
            raise AssertionError(msg)

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        execution_mode="all",
        runner_mode=True,
    )
    monkeypatch.setattr("mindroom.tool_system.sandbox_proxy.httpx.Client", _ForbiddenClient)

    tool = get_tool_by_name("calculator", runtime_paths, worker_target=None)
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    result = entrypoint(1, 2)
    assert '"result": 3' in result


def test_proxy_requests_credential_lease_when_policy_matches(monkeypatch: pytest.MonkeyPatch) -> None:
    """Proxy should create and consume a lease when credential sharing policy allows it."""
    captured_calls: list[tuple[str, dict[str, Any]]] = []

    fake_credentials = FakeCredentialsManager({"openai": {"api_key": "sk-test", "_source": "ui"}})

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="all",
        credential_policy={"calculator.add": ("openai",)},
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            captured_calls=captured_calls,
            responder=lambda url, _json: (
                {"lease_id": "lease-123", "expires_at": 123.0, "max_uses": 1}
                if url.endswith("/leases")
                else {"ok": True, "result": "proxied"}
            ),
        ),
    )

    tool = get_tool_by_name("calculator", runtime_paths, credentials_manager=fake_credentials, worker_target=None)
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    result = entrypoint(1, 2)

    assert result == "proxied"
    assert len(captured_calls) == 2

    lease_url, lease_payload = captured_calls[0]
    assert lease_url.endswith("/api/sandbox-runner/leases")
    assert lease_payload["credential_overrides"] == {"api_key": "sk-test"}

    execute_url, execute_payload = captured_calls[1]
    assert execute_url.endswith("/api/sandbox-runner/execute")
    assert execute_payload["lease_id"] == "lease-123"


def test_save_attachment_to_worker_posts_with_worker_token_and_size_cap(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Attachment worker saves should use the selected worker handle and reject oversized payloads locally."""
    captured: dict[str, Any] = {}
    manager = _TrackingWorkerManager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"file"},
    )
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)
    payload_bytes = b"attachment-bytes"
    sha256 = hashlib.sha256(payload_bytes).hexdigest()

    monkeypatch.setattr(
        sandbox_proxy_module,
        "lease_primary_worker_manager",
        lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            captured=captured,
            responder=lambda _url, _json: {
                "ok": True,
                "worker_path": "sample.bin",
                "size_bytes": len(payload_bytes),
                "sha256": sha256,
            },
        ),
    )

    receipt = sandbox_proxy_module.save_attachment_to_worker(
        runtime_paths=runtime_paths,
        worker_target=worker_target,
        attachment_id="att_sample",
        mindroom_output_path="sample.bin",
        payload_bytes=payload_bytes,
        mime_type="application/octet-stream",
        filename="sample.bin",
    )

    assert receipt is not None
    assert receipt.worker_path == "sample.bin"
    assert captured["url"] == "http://worker/api/sandbox-runner/save-attachment"
    assert captured["headers"] == {"x-mindroom-sandbox-token": _TEST_AUTH_TOKEN}
    request_json = captured["json"]
    assert request_json["attachment_id"] == "att_sample"
    assert request_json["mindroom_output_path"] == "sample.bin"
    assert request_json["worker_key"] == worker_target.worker_key
    assert base64.b64decode(request_json["bytes_b64"]) == payload_bytes
    assert request_json["sha256"] == sha256
    assert manager.touched == [worker_target.worker_key]

    monkeypatch.setenv("MINDROOM_ATTACHMENT_INLINE_SAVE_MAX_BYTES", "3")
    with pytest.raises(RuntimeError, match="att_sample"):
        sandbox_proxy_module.save_attachment_to_worker(
            runtime_paths=_runtime_paths_from_env(),
            worker_target=worker_target,
            worker_tools_override=["file"],
            attachment_id="att_sample",
            mindroom_output_path="sample.bin",
            payload_bytes=payload_bytes,
            mime_type=None,
            filename=None,
        )


@pytest.mark.parametrize(
    ("response_payload", "expected_error"),
    [
        (
            {"ok": True, "worker_path": "sample.bin", "size_bytes": 999, "sha256": "0" * 64},
            "size does not match",
        ),
        (
            {"ok": True, "worker_path": "sample.bin", "size_bytes": 16, "sha256": "0" * 64},
            "SHA256 does not match",
        ),
        (
            {"ok": True, "worker_path": "sample.bin", "size_bytes": True, "sha256": "0" * 64},
            "missing its receipt fields",
        ),
        (
            {"ok": True, "worker_path": "sample.bin", "size_bytes": -1, "sha256": "0" * 64},
            "missing its receipt fields",
        ),
    ],
)
def test_save_attachment_to_worker_rejects_bad_receipts(
    monkeypatch: pytest.MonkeyPatch,
    response_payload: dict[str, object],
    expected_error: str,
) -> None:
    """Primary saves should verify worker receipts against the bytes sent."""
    manager = _TrackingWorkerManager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"file"},
    )
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)

    monkeypatch.setattr(
        sandbox_proxy_module,
        "lease_primary_worker_manager",
        lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(responder=lambda _url, _json: response_payload),
    )

    with pytest.raises(RuntimeError, match=expected_error):
        sandbox_proxy_module.save_attachment_to_worker(
            runtime_paths=runtime_paths,
            worker_target=worker_target,
            attachment_id="att_sample",
            mindroom_output_path="sample.bin",
            payload_bytes=b"attachment-bytes",
            mime_type=None,
            filename=None,
        )

    assert manager.failures
    assert manager.touched == []


def test_save_attachment_to_worker_request_failure_does_not_record_worker_failure(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Structured tool/request failures from save-attachment should not poison worker health."""
    manager = _TrackingWorkerManager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"file"},
    )
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)

    monkeypatch.setattr(
        sandbox_proxy_module,
        "lease_primary_worker_manager",
        lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            responder=lambda _url, _json: {
                "ok": False,
                "error": "mindroom_output_path must stay inside the workspace.",
                "failure_kind": "tool",
            },
        ),
    )

    with pytest.raises(RuntimeError, match="mindroom_output_path"):
        sandbox_proxy_module.save_attachment_to_worker(
            runtime_paths=runtime_paths,
            worker_target=worker_target,
            attachment_id="att_sample",
            mindroom_output_path="../escape.bin",
            payload_bytes=b"attachment-bytes",
            mime_type=None,
            filename=None,
        )

    assert manager.failures == []
    assert manager.touched == [worker_target.worker_key]


def _sandbox_proxy_test_metadata(
    name: str,
    *,
    default_execution_target: ToolExecutionTarget = ToolExecutionTarget.PRIMARY,
    consumes_workspace_paths: bool = False,
) -> ToolMetadata:
    return ToolMetadata(
        name=name,
        display_name=name,
        description=name,
        category=ToolCategory.DEVELOPMENT,
        default_execution_target=default_execution_target,
        consumes_workspace_paths=consumes_workspace_paths,
    )


def test_default_proxy_routing_uses_worker_execution_metadata(monkeypatch: pytest.MonkeyPatch) -> None:
    """Default fail-closed proxy routing should come from tool metadata, not a hardcoded name set."""
    tool_name = "metadata_worker_execution_test"
    monkeypatch.setitem(
        TOOL_METADATA,
        tool_name,
        _sandbox_proxy_test_metadata(
            tool_name,
            default_execution_target=ToolExecutionTarget.WORKER,
        ),
    )
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode=None,
    )

    assert sandbox_proxy_module.sandbox_proxy_enabled_for_tool(tool_name, runtime_paths=runtime_paths) is True


def test_attachment_save_uses_workspace_consumer_metadata(monkeypatch: pytest.MonkeyPatch) -> None:
    """Attachment saves should follow workspace-consumer metadata, not a hardcoded name set."""
    consumer_tool = "metadata_workspace_consumer_test"
    non_consumer_tool = "metadata_non_workspace_consumer_test"
    monkeypatch.setitem(
        TOOL_METADATA,
        consumer_tool,
        _sandbox_proxy_test_metadata(
            consumer_tool,
            default_execution_target=ToolExecutionTarget.WORKER,
            consumes_workspace_paths=True,
        ),
    )
    monkeypatch.setitem(
        TOOL_METADATA,
        non_consumer_tool,
        _sandbox_proxy_test_metadata(
            non_consumer_tool,
            default_execution_target=ToolExecutionTarget.WORKER,
            consumes_workspace_paths=False,
        ),
    )
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )

    assert (
        sandbox_proxy_module.attachment_save_uses_worker(
            runtime_paths=runtime_paths,
            worker_tools_override=[consumer_tool],
        )
        is True
    )
    assert (
        sandbox_proxy_module.attachment_save_uses_worker(
            runtime_paths=runtime_paths,
            worker_tools_override=[non_consumer_tool],
        )
        is False
    )


@pytest.mark.parametrize("worker_tools_override", [["coding"], ["docker"], ["python"], ["shell", "coding"]])
def test_attachment_save_uses_worker_for_worker_routed_workspace_consumers(
    monkeypatch: pytest.MonkeyPatch,
    worker_tools_override: list[str],
) -> None:
    """Attachment saves should follow worker-routed workspace consumers, not only file workspaces."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    assert (
        sandbox_proxy_module.attachment_save_uses_worker(
            runtime_paths=runtime_paths,
            worker_tools_override=worker_tools_override,
        )
        is True
    )
    assert (
        sandbox_proxy_module.attachment_save_uses_worker(
            runtime_paths=runtime_paths,
            worker_tools_override=["calculator"],
        )
        is False
    )


def test_save_attachment_to_static_runner_posts_without_worker_key(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Static proxy saves should use the shared runner endpoint without forcing a worker key."""
    captured: dict[str, Any] = {}
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"file"},
    )
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    worker_target = _worker_target(runtime_paths, None, "code", execution_identity)
    payload_bytes = b"attachment-bytes"
    sha256 = hashlib.sha256(payload_bytes).hexdigest()

    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            captured=captured,
            responder=lambda _url, _json: {
                "ok": True,
                "worker_path": "sample.bin",
                "size_bytes": len(payload_bytes),
                "sha256": sha256,
            },
        ),
    )

    receipt = sandbox_proxy_module.save_attachment_to_worker(
        runtime_paths=runtime_paths,
        worker_target=worker_target,
        attachment_id="att_sample",
        mindroom_output_path="sample.bin",
        payload_bytes=payload_bytes,
        mime_type=None,
        filename=None,
    )

    assert receipt is not None
    assert captured["url"] == "http://sandbox-runner:8765/api/sandbox-runner/save-attachment"
    assert "worker_key" not in captured["json"]
    assert captured["json"]["routing_agent_name"] == "code"


def test_get_tool_by_name_rejects_unsafe_tool_init_overrides() -> None:
    """Tool init overrides should allow only the explicit safe whitelist."""
    with pytest.raises(ToolInitOverrideError, match="api_key"):
        get_tool_by_name(
            "openai",
            _TEST_RUNTIME_PATHS,
            tool_init_overrides={"api_key": "sk-test"},
            worker_target=None,
        )


def test_get_tool_by_name_rejects_invalid_base_dir_override_type() -> None:
    """base_dir overrides should be validated before toolkit construction."""
    with pytest.raises(ToolInitOverrideError, match="base_dir"):
        get_tool_by_name(
            "coding",
            _TEST_RUNTIME_PATHS,
            tool_init_overrides={"base_dir": {"bad": "value"}},
            worker_target=None,
        )


def test_get_tool_by_name_loads_persisted_non_secret_file_config(tmp_path: Path) -> None:
    """Persisted plain config should still hydrate SetupType.NONE tools during rebuilds."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("models: {}\nagents: {}\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={},
    )
    workspace = tmp_path / "workspace"
    workspace.mkdir(parents=True, exist_ok=True)
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    save_scoped_credentials(
        "file",
        {
            "base_dir": str(workspace),
            "enable_delete_file": True,
        },
        credentials_manager=credentials_manager,
        worker_target=None,
    )

    tool = get_tool_by_name("file", runtime_paths, worker_target=None)

    assert tool.base_dir == workspace.resolve()
    assert "delete_file" in tool.functions


def test_get_tool_by_name_builds_google_bigquery_from_scoped_credentials(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Google BigQuery should be configured from persisted tool credentials."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text("models: {}\nagents: {}\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={},
    )
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    save_scoped_credentials(
        "google_bigquery",
        {
            "dataset": "demo_dataset",
            "project": "demo-project",
            "location": "us-central1",
        },
        credentials_manager=credentials_manager,
        worker_target=None,
    )
    captured: dict[str, object] = {}

    class _FakeGoogleBigQueryTools:
        def __init__(
            self,
            *,
            dataset: str,
            project: str,
            location: str,
            credentials: object | None = None,
            **_: object,
        ) -> None:
            self.dataset = dataset
            self.project = project
            self.location = location
            captured["project"] = project
            captured["credentials"] = credentials

    monkeypatch.setitem(TOOL_REGISTRY, "google_bigquery", lambda: _FakeGoogleBigQueryTools)

    tool = get_tool_by_name("google_bigquery", runtime_paths, worker_target=None)

    assert tool.dataset == "demo_dataset"
    assert tool.project == "demo-project"
    assert tool.location == "us-central1"
    assert captured["project"] == "demo-project"
    assert captured["credentials"] is None


def test_get_tool_by_name_requires_explicit_clickup_config(tmp_path: Path) -> None:
    """Runtime-scoped env values should not configure ClickUp during toolkit construction."""
    config_dir = tmp_path / "cfg"
    config_dir.mkdir(parents=True, exist_ok=True)
    config_path = config_dir / "config.yaml"
    config_path.write_text("models: {}\nagents: {}\n", encoding="utf-8")
    (config_dir / ".env").write_text(
        "CLICKUP_API_KEY=clickup-test\nMASTER_SPACE_ID=space-123\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={},
    )

    with pytest.raises(ValueError, match="CLICKUP_API_KEY not set"):
        get_tool_by_name("clickup", runtime_paths, worker_target=None)


def test_get_tool_by_name_does_not_expose_runtime_env_to_direct_python_execution(tmp_path: Path) -> None:
    """Direct in-process Python execution should not emulate committed runtime env."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={
            "MINDROOM_UNSAFE_ALLOW_LOCAL_EXECUTION_TOOLS": "true",
            "OPENAI_BASE_URL": "http://example.invalid/v1",
            "MINDROOM_NAMESPACE": "alpha1234",
        },
    )

    tool = get_tool_by_name("python", runtime_paths, worker_target=None)
    entrypoint = tool.functions["run_python_code"].entrypoint
    assert entrypoint is not None

    result = entrypoint(
        'import os\nresult = {"openai_base_url": os.environ.get("OPENAI_BASE_URL"), "namespace": os.environ.get("MINDROOM_NAMESPACE"), "storage": os.environ.get("MINDROOM_STORAGE_PATH")}',
        "result",
    )

    assert ast.literal_eval(result) == {
        "openai_base_url": None,
        "namespace": None,
        "storage": None,
    }


def test_get_tool_by_name_does_not_expose_runtime_env_to_file_backed_python_execution(tmp_path: Path) -> None:
    """Direct file-backed Python execution should also avoid runtime env emulation."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={
            "MINDROOM_UNSAFE_ALLOW_LOCAL_EXECUTION_TOOLS": "true",
            "OPENAI_BASE_URL": "http://example.invalid/v1",
            "MINDROOM_NAMESPACE": "alpha1234",
        },
    )

    tool = get_tool_by_name(
        "python",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_target=None,
    )
    save_entrypoint = tool.functions["save_to_file_and_run"].entrypoint
    run_file_entrypoint = tool.functions["run_python_file_return_variable"].entrypoint
    assert save_entrypoint is not None
    assert run_file_entrypoint is not None

    code = (
        "import os\n"
        'result = {"openai_base_url": os.environ.get("OPENAI_BASE_URL"), '
        '"namespace": os.environ.get("MINDROOM_NAMESPACE"), '
        '"storage": os.environ.get("MINDROOM_STORAGE_PATH")}'
    )
    save_result = save_entrypoint("runtime_values.py", code, "result")
    run_result = run_file_entrypoint("runtime_values.py", "result")
    expected = {
        "openai_base_url": None,
        "namespace": None,
        "storage": None,
    }

    assert ast.literal_eval(save_result) == expected
    assert ast.literal_eval(run_result) == expected


def test_shell_subprocess_path_keeps_existing_path_without_prepend() -> None:
    """Shell PATH normalization should preserve the base PATH when nothing is prepended."""
    assert subprocess_path_with_prepends("/usr/local/bin:/usr/bin:/bin") == "/usr/local/bin:/usr/bin:/bin"


def test_shell_subprocess_path_uses_only_prepend_entries_for_empty_path() -> None:
    """Configured path entries should still be available when PATH is empty."""
    assert (
        subprocess_path_with_prepends(
            "",
            prepend_entries=("/opt/custom/bin", "/opt/worker/bin"),
        )
        == "/opt/custom/bin:/opt/worker/bin"
    )


def test_shell_subprocess_path_prepends_configured_entries_and_dedupes() -> None:
    """Configured path entries should stay first without duplicating existing PATH entries."""
    assert (
        subprocess_path_with_prepends(
            "/usr/bin:/opt/existing/bin:/bin",
            prepend_entries=("/opt/custom/bin", "/opt/existing/bin"),
        )
        == "/opt/custom/bin:/opt/existing/bin:/usr/bin:/bin"
    )


def test_shell_subprocess_env_path_passthrough_without_prepend(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Shell PATH normalization should preserve the runtime PATH by default."""
    monkeypatch.setenv("PATH", "/usr/local/bin:/usr/bin")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={},
    )

    runtime_env = dict(shell_execution_runtime_env_values(runtime_paths))
    assert shell_tool_module._shell_subprocess_env(runtime_env)["PATH"] == "/usr/local/bin:/usr/bin"


def test_shell_subprocess_env_prefers_explicit_base_process_env(monkeypatch: pytest.MonkeyPatch) -> None:
    """Shell subprocess env should come from the explicit request snapshot, not live runner state."""
    monkeypatch.setenv("PATH", "/runner/bin")
    monkeypatch.setenv("HOME", "/runner-home")

    env = shell_tool_module._shell_subprocess_env(
        {},
        base_process_env={"PATH": "/request/bin", "HOME": "/request-home"},
    )

    assert env["PATH"] == "/request/bin"
    assert env["HOME"] == "/request-home"


def test_shell_subprocess_env_forces_vendor_telemetry_over_runtime_env() -> None:
    """Shell subprocesses should not let request-scoped env re-enable vendor telemetry."""
    runtime_env = dict.fromkeys(VENDOR_TELEMETRY_ENV_VALUES, "enabled")

    env = shell_tool_module._shell_subprocess_env(runtime_env)

    for name, value in VENDOR_TELEMETRY_ENV_VALUES.items():
        assert env[name] == value


def test_subprocess_env_for_request_forces_vendor_telemetry_over_execution_env() -> None:
    """Sandbox subprocess env should not let execution overlays re-enable vendor telemetry."""
    base_env = dict.fromkeys(VENDOR_TELEMETRY_ENV_VALUES, "base-enabled")
    execution_env = dict.fromkeys(VENDOR_TELEMETRY_ENV_VALUES, "request-enabled")
    execution_env["MINDROOM_KEEP"] = "from-request"

    env = sandbox_exec_module.subprocess_env_for_request(base_env, execution_env)

    assert env is not None
    assert env["MINDROOM_KEEP"] == "from-request"
    for name, value in VENDOR_TELEMETRY_ENV_VALUES.items():
        assert env[name] == value


def test_shell_subprocess_env_preserves_workspace_home_contract_names() -> None:
    """Workspace-home env names should survive into the shell command process."""
    env = shell_tool_module._shell_subprocess_env(
        {
            "HOME": "/env-home",
            "MINDROOM_AGENT_WORKSPACE": "/env-agent-workspace",
            "PIP_CACHE_DIR": "/env-pip-cache",
            "PYTHONPYCACHEPREFIX": "/env-pycache",
            "UV_CACHE_DIR": "/env-uv-cache",
            "VIRTUAL_ENV": "/env-venv",
            "XDG_CACHE_HOME": "/env-cache",
            "XDG_CONFIG_HOME": "/env-config",
            "XDG_DATA_HOME": "/env-data",
            "XDG_STATE_HOME": "/env-state",
        },
        base_process_env={
            "HOME": "/workspace",
            "MINDROOM_AGENT_WORKSPACE": "/workspace",
            "PIP_CACHE_DIR": "/worker-cache/pip",
            "PYTHONPYCACHEPREFIX": "/worker-cache/pycache",
            "UV_CACHE_DIR": "/worker-cache/uv",
            "VIRTUAL_ENV": "/worker-venv",
            "XDG_CACHE_HOME": "/worker-cache",
            "XDG_CONFIG_HOME": "/workspace/.config",
            "XDG_DATA_HOME": "/workspace/.local/share",
            "XDG_STATE_HOME": "/workspace/.local/state",
        },
    )

    assert env["HOME"] == "/workspace"
    assert env["MINDROOM_AGENT_WORKSPACE"] == "/workspace"
    assert env["XDG_CONFIG_HOME"] == "/workspace/.config"
    assert env["XDG_DATA_HOME"] == "/workspace/.local/share"
    assert env["XDG_STATE_HOME"] == "/workspace/.local/state"
    assert env["XDG_CACHE_HOME"] == "/worker-cache"
    assert env["PIP_CACHE_DIR"] == "/worker-cache/pip"
    assert env["UV_CACHE_DIR"] == "/worker-cache/uv"
    assert env["PYTHONPYCACHEPREFIX"] == "/worker-cache/pycache"
    assert env["VIRTUAL_ENV"] == "/worker-venv"


def test_shell_subprocess_env_prefers_runtime_env_when_no_workspace_contract() -> None:
    """Direct shell runtime env should not be overwritten by ordinary process env."""
    env = shell_tool_module._shell_subprocess_env(
        {
            "HOME": "/runtime-home",
            "XDG_CONFIG_HOME": "/runtime-config",
            "VIRTUAL_ENV": "/runtime-venv",
        },
        base_process_env={
            "HOME": "/process-home",
            "XDG_CONFIG_HOME": "/process-config",
            "VIRTUAL_ENV": "/process-venv",
        },
    )

    assert env["HOME"] == "/runtime-home"
    assert env["XDG_CONFIG_HOME"] == "/runtime-config"
    assert env["VIRTUAL_ENV"] == "/runtime-venv"


def test_execution_env_payload_denies_provider_env_by_default_in_isolated_runtime(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Isolated execution should not inherit provider env or arbitrary runtime `.env` values by default."""
    monkeypatch.setenv("OPENAI_API_KEY", "env-openai-key")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (tmp_path / ".env").write_text(
        "MINDROOM_NAMESPACE=alpha1234\nTEST_EXECUTION_ENV=visible-in-shell\nOPENAI_BASE_URL=http://example.invalid/v1\nCUSTOM_API_TOKEN=custom-secret\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env=dict(os.environ),
    )

    class _FakeConfig:
        def get_worker_grantable_credentials(self) -> frozenset[str]:
            return frozenset()

    monkeypatch.setattr(
        sandbox_proxy_module,
        "get_tool_runtime_context",
        lambda: SimpleNamespace(config=_FakeConfig()),
    )

    execution_env = sandbox_proxy_module._execution_env_payload("python", runtime_paths=runtime_paths)

    assert execution_env is not None
    assert execution_env["MINDROOM_NAMESPACE"] == "alpha1234"
    assert "TEST_EXECUTION_ENV" not in execution_env
    assert "OPENAI_API_KEY" not in execution_env
    assert "OPENAI_BASE_URL" not in execution_env
    assert "CUSTOM_API_TOKEN" not in execution_env


def test_execution_env_payload_keeps_provider_env_denied_even_with_worker_credential_allowlist(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Isolated execution should not reintroduce provider env just because credentials are mirrored."""
    monkeypatch.setenv("OPENAI_API_KEY", "env-openai-key")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env=dict(os.environ),
    )

    class _FakeConfig:
        def get_worker_grantable_credentials(self) -> frozenset[str]:
            return frozenset({"openai"})

    monkeypatch.setattr(
        sandbox_proxy_module,
        "get_tool_runtime_context",
        lambda: SimpleNamespace(config=_FakeConfig()),
    )

    execution_env = sandbox_proxy_module._execution_env_payload("python", runtime_paths=runtime_paths)

    assert execution_env is not None
    assert "OPENAI_API_KEY" not in execution_env
    assert "OPENAI_BASE_URL" not in execution_env


def test_worker_env_excludes_openai_api_key_unless_extra_env_passthrough(
    tmp_path: Path,
) -> None:
    """Isolated worker startup env should not inherit provider API keys by default."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={
            "OPENAI_API_KEY": "sk-primary",
            "ANTHROPIC_API_KEY": "sk-ant-primary",
        },
    )

    worker_paths = isolated_runtime_paths(runtime_paths)
    shell_env = build_execution_tool_env(
        "shell",
        worker_paths,
        extra_env_passthrough="OPENAI_API_KEY",
        shell_process_env=runtime_paths.process_env,
    )

    assert worker_paths.env_value("OPENAI_API_KEY") is None
    assert worker_paths.env_value("ANTHROPIC_API_KEY") is None
    assert shell_env["OPENAI_API_KEY"] == "sk-primary"


def test_worker_env_includes_extra_env_passthrough(tmp_path: Path) -> None:
    """Explicit shell passthrough should still expose selected process env values."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={
            "MY_VAR": "visible",
            "OTHER_VAR": "hidden",
        },
    )

    worker_paths = isolated_runtime_paths(runtime_paths)
    shell_env = build_execution_tool_env(
        "shell",
        worker_paths,
        extra_env_passthrough="MY_VAR",
        shell_process_env=runtime_paths.process_env,
    )

    assert worker_paths.env_value("MY_VAR") is None
    assert shell_env["MY_VAR"] == "visible"
    assert "OTHER_VAR" not in shell_env


@pytest.mark.asyncio
async def test_get_tool_by_name_exposes_runtime_env_to_shell_execution(tmp_path: Path) -> None:
    """Direct shell execution should inherit committed runtime env values from the runtime `.env`."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (tmp_path / ".env").write_text("TEST_EXECUTION_ENV=visible-in-shell\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env={},
    )

    tool = get_tool_by_name("shell", runtime_paths, disable_sandbox_proxy=True, worker_target=None)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(
        [
            sys.executable,
            "-c",
            'import os, sys; sys.stdout.write(os.environ.get("TEST_EXECUTION_ENV", ""))',
        ],
    )

    assert result == "visible-in-shell"


@pytest.mark.asyncio
async def test_local_shell_exposes_configured_extra_parent_env_without_leaking_control_secrets(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Direct shell execution should allow explicit extra passthrough without leaking control secrets."""
    monkeypatch.setenv("GITEA_TOKEN", "visible-gitea-token")
    monkeypatch.setenv("MINDROOM_SANDBOX_PROXY_TOKEN", "runner-secret")
    monkeypatch.setenv("CI_JOB_TOKEN", "ci-secret")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (tmp_path / ".env").write_text("TEST_EXECUTION_ENV=visible-in-shell\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env=dict(os.environ),
    )
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    save_scoped_credentials(
        "shell",
        {"extra_env_passthrough": "GITEA_*, MINDROOM_*, CI_JOB_TOKEN"},
        credentials_manager=credentials_manager,
        worker_target=None,
    )

    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        disable_sandbox_proxy=True,
        worker_target=None,
    )
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(
        [
            "bash",
            "-lc",
            "printf '%s' \"$GITEA_TOKEN|$MINDROOM_SANDBOX_PROXY_TOKEN|$CI_JOB_TOKEN|$TEST_EXECUTION_ENV\"",
        ],
    )

    assert result == "visible-gitea-token||ci-secret|visible-in-shell"


@pytest.mark.asyncio
async def test_local_shell_does_not_expose_extra_parent_env_without_configuration(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Direct shell execution should not expose extra parent env without explicit config."""
    monkeypatch.setenv("WHISPER_URL", "https://whisper.example")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    (tmp_path / ".env").write_text("TEST_EXECUTION_ENV=visible-in-shell\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env=dict(os.environ),
    )

    tool = get_tool_by_name("shell", runtime_paths, disable_sandbox_proxy=True, worker_target=None)
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    result = await entrypoint(["bash", "-lc", "printf '%s' \"$WHISPER_URL|$TEST_EXECUTION_ENV\""])

    assert result == "|visible-in-shell"


@pytest.mark.asyncio
async def test_local_shell_prepends_configured_path_entries(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Direct shell execution should prepend configured PATH entries without losing runtime PATH."""
    monkeypatch.setenv("PATH", "/usr/local/bin:/usr/bin:/bin")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=config_path.parent / "storage",
        process_env=dict(os.environ),
    )
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    save_scoped_credentials(
        "shell",
        {"shell_path_prepend": "/opt/custom/bin, /opt/worker/bin"},
        credentials_manager=credentials_manager,
        worker_target=None,
    )

    shell_tool = get_tool_by_name("shell", runtime_paths, disable_sandbox_proxy=True, worker_target=None)
    shell_entrypoint = shell_tool.async_functions["run_shell_command"].entrypoint
    assert shell_entrypoint is not None
    result = await shell_entrypoint(
        [
            sys.executable,
            "-c",
            'import os, sys; sys.stdout.write(os.environ.get("PATH", ""))',
        ],
    )

    assert result.startswith("/opt/custom/bin:/opt/worker/bin:")
    assert result.endswith("/usr/local/bin:/usr/bin:/bin")


@pytest.mark.asyncio
async def test_proxy_forwards_configured_shell_execution_env_only_for_execution_tools(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Sandbox proxy should forward shell execution env and passthrough config from stored tool settings."""
    captured: dict[str, Any] = {}
    _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="all",
        credential_policy={},
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(captured=captured),
    )
    monkeypatch.setenv("GITEA_TOKEN", "visible-gitea-token")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    config_path.with_name(".env").write_text("TEST_EXECUTION_ENV=visible-in-shell\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=config_path.parent / "storage",
        process_env=dict(os.environ),
    )
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    save_scoped_credentials(
        "shell",
        {
            "extra_env_passthrough": "GITEA_*",
            "shell_path_prepend": "/opt/custom/bin",
        },
        credentials_manager=credentials_manager,
        worker_target=None,
    )

    shell_tool = get_tool_by_name("shell", runtime_paths, worker_target=None)
    shell_entrypoint = shell_tool.async_functions["run_shell_command"].entrypoint
    assert shell_entrypoint is not None
    result = await shell_entrypoint(["bash", "-lc", "printf '%s' \"$TEST_EXECUTION_ENV\""])

    assert result == "sandbox-result"
    assert captured["json"]["extra_env_passthrough"] == "GITEA_*"
    assert captured["json"]["tool_init_overrides"]["shell_path_prepend"] == "/opt/custom/bin"
    assert "TEST_EXECUTION_ENV" not in captured["json"]["execution_env"]
    assert captured["json"]["execution_env"]["GITEA_TOKEN"] == "visible-gitea-token"  # noqa: S105
    assert "MINDROOM_SANDBOX_PROXY_TOKEN" not in captured["json"]["execution_env"]

    captured.clear()
    calculator = get_tool_by_name("calculator", runtime_paths, worker_target=None)
    calculator_entrypoint = calculator.functions["add"].entrypoint
    assert calculator_entrypoint is not None
    calculator_entrypoint(1, 2)

    assert "execution_env" not in captured["json"]


@pytest.mark.asyncio
async def test_proxy_shell_extra_env_passthrough_survives_sandbox_runner_rebuild(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Configured shell passthrough should survive proxy forwarding and runner subprocess rebuilds."""
    _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="all",
        credential_policy={},
    )
    monkeypatch.setenv("MINDROOM_SANDBOX_RUNNER_EXECUTION_MODE", "subprocess")
    monkeypatch.setenv("GITEA_TOKEN", "visible-gitea-token")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    config_path.with_name(".env").write_text("TEST_EXECUTION_ENV=visible-in-shell\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=config_path.parent / "storage",
        process_env=dict(os.environ),
    )
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    save_scoped_credentials(
        "shell",
        {"extra_env_passthrough": "GITEA_*"},
        credentials_manager=credentials_manager,
        worker_target=None,
    )

    def responder(_url: str, payload: dict[str, Any]) -> dict[str, object]:
        response = sandbox_runner_module._execute_request_subprocess_sync(
            sandbox_runner_module.SandboxRunnerExecuteRequest.model_validate(payload),
            runtime_paths,
            sandbox_runner_module._runtime_config_or_empty(runtime_paths),
            runner_token=_TEST_AUTH_TOKEN,
        )
        return response.model_dump(mode="json")

    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(responder=responder),
    )

    shell_tool = get_tool_by_name("shell", runtime_paths, worker_target=None)
    shell_entrypoint = shell_tool.async_functions["run_shell_command"].entrypoint
    assert shell_entrypoint is not None
    result = await shell_entrypoint(
        [
            "bash",
            "-lc",
            "printf '%s' \"$GITEA_TOKEN|$TEST_EXECUTION_ENV\"",
        ],
    )

    assert result == "visible-gitea-token|"


@pytest.mark.asyncio
async def test_proxy_shell_path_prepend_survives_sandbox_runner_rebuild(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Configured shell_path_prepend should survive proxy forwarding and runner rebuilds."""
    _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="all",
        credential_policy={},
    )
    monkeypatch.setenv("MINDROOM_SANDBOX_RUNNER_EXECUTION_MODE", "subprocess")
    monkeypatch.setenv("PATH", "/usr/local/bin:/usr/bin:/bin")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=config_path.parent / "storage",
        process_env=dict(os.environ),
    )
    credentials_manager = get_runtime_credentials_manager(runtime_paths)
    save_scoped_credentials(
        "shell",
        {"shell_path_prepend": "/opt/custom/bin, /opt/worker/bin"},
        credentials_manager=credentials_manager,
        worker_target=None,
    )

    def responder(_url: str, payload: dict[str, Any]) -> dict[str, object]:
        response = sandbox_runner_module._execute_request_subprocess_sync(
            sandbox_runner_module.SandboxRunnerExecuteRequest.model_validate(payload),
            runtime_paths,
            sandbox_runner_module._runtime_config_or_empty(runtime_paths),
            runner_token=_TEST_AUTH_TOKEN,
        )
        return response.model_dump(mode="json")

    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(responder=responder),
    )

    shell_tool = get_tool_by_name("shell", runtime_paths, worker_target=None)
    shell_entrypoint = shell_tool.async_functions["run_shell_command"].entrypoint
    assert shell_entrypoint is not None
    result = await shell_entrypoint(
        [
            sys.executable,
            "-c",
            'import os, sys; sys.stdout.write(os.environ.get("PATH", ""))',
        ],
    )

    assert result.startswith("/opt/custom/bin:/opt/worker/bin:")
    assert result.endswith("/usr/local/bin:/usr/bin:/bin")


def test_dedicated_worker_runtime_config_resolves_include_tags(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Dedicated workers must load configs split across !include files."""
    (tmp_path / "models.yaml").write_text(
        "default:\n  provider: openai\n  id: gpt-5.4\n",
        encoding="utf-8",
    )
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models: !include models.yaml\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=config_path.parent / "storage",
        process_env={},
    )
    monkeypatch.setattr(
        sandbox_runner_module,
        "_upstream_tool_validation_snapshot",
        lambda _runtime_paths: {"shell": object()},
    )

    config = sandbox_runner_module._dedicated_worker_runtime_config_or_empty(runtime_paths)

    assert config.models["default"].id == "gpt-5.4"


@pytest.mark.asyncio
async def test_inprocess_runner_shell_uses_request_scoped_extra_env_snapshot(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """In-process runner rebuilds should source shell passthrough from request-scoped runtime env."""
    monkeypatch.setenv("GITEA_TOKEN", "ambient-gitea-token")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    config_path.with_name(".env").write_text("TEST_EXECUTION_ENV=visible-in-shell\n", encoding="utf-8")
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=config_path.parent / "storage",
        process_env={},
    )
    config = sandbox_runner_module._runtime_config_or_empty(runtime_paths)
    request = sandbox_runner_module.SandboxRunnerExecuteRequest(
        tool_name="shell",
        function_name="run_shell_command",
        args=[
            [
                "bash",
                "-lc",
                "printf '%s' \"$GITEA_TOKEN|$TEST_EXECUTION_ENV\"",
            ],
        ],
        execution_env={"GITEA_TOKEN": "request-gitea-token"},
        extra_env_passthrough="GITEA_*",
    )

    response = await sandbox_runner_module._execute_request_inprocess(
        request,
        runtime_paths,
        config,
    )

    assert response.ok is True
    assert response.result == "request-gitea-token|"


def test_get_worker_manager_falls_back_to_runtime_storage_root_without_tool_context(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Worker routing should not require ToolRuntimeContext just to recover storage_root."""
    runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config.yaml",
        storage_path=tmp_path / "storage",
        process_env={},
    )
    captured: dict[str, object] = {}

    def _fake_get_primary_worker_manager(
        runtime_paths_arg: RuntimePaths,
        *,
        proxy_url: str | None,
        proxy_token: str | None,
        storage_root: Path | None = None,
        kubernetes_tool_validation_snapshot: dict[str, dict[str, object]] | None = None,
        kubernetes_config_snapshot: dict[str, object] | None = None,
        worker_grantable_credentials: frozenset[str] | None = None,
    ) -> str:
        captured["runtime_paths"] = runtime_paths_arg
        captured["proxy_url"] = proxy_url
        captured["proxy_token"] = proxy_token
        captured["storage_root"] = storage_root
        captured["kubernetes_tool_validation_snapshot"] = kubernetes_tool_validation_snapshot
        captured["kubernetes_config_snapshot"] = kubernetes_config_snapshot
        captured["worker_grantable_credentials"] = worker_grantable_credentials
        return "manager"

    monkeypatch.setattr(sandbox_proxy_module, "get_primary_worker_manager", _fake_get_primary_worker_manager)
    monkeypatch.setattr(sandbox_proxy_module, "get_tool_runtime_context", lambda: None)

    manager = sandbox_proxy_module._get_worker_manager(
        runtime_paths,
        sandbox_proxy_module._SandboxProxyConfig(
            runner_mode=False,
            proxy_url="http://sandbox",
            proxy_token="token",  # noqa: S106
            proxy_timeout_seconds=30.0,
            execution_mode="all",
            credential_lease_ttl_seconds=60,
            proxy_tools=None,
            credential_policy={},
        ),
    )

    assert manager == "manager"
    assert captured["runtime_paths"] == runtime_paths
    assert captured["storage_root"] == (tmp_path / "storage").resolve()
    assert captured["kubernetes_tool_validation_snapshot"] is None
    assert captured["kubernetes_config_snapshot"] is None
    assert captured["worker_grantable_credentials"] is None


def test_proxy_holds_manager_lease_through_success_bookkeeping(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dedicated worker success bookkeeping should run before the manager lease is released."""
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="all",
    )
    events: list[str] = []
    lease_active = False
    worker_handle = WorkerHandle(
        worker_id="worker-1",
        worker_key="worker-key",
        endpoint="http://worker/api/sandbox-runner/execute",
        auth_token=_TEST_AUTH_TOKEN,
        status="ready",
        backend_name="docker",
        last_used_at=1.0,
        created_at=0.0,
    )

    class _FakeWorkerManager:
        def touch_worker(self, worker_key: str, *, now: float | None = None) -> WorkerHandle | None:
            nonlocal lease_active
            _ = now
            assert lease_active is True
            assert worker_key == "worker-key"
            events.append("touch")
            return worker_handle

        def record_failure(self, worker_key: str, failure_reason: str, *, now: float | None = None) -> WorkerHandle:
            _ = now
            msg = f"unexpected failure bookkeeping for {worker_key}: {failure_reason}"
            raise AssertionError(msg)

    worker_manager = _FakeWorkerManager()

    class _FakeLease:
        def __enter__(self) -> _FakeWorkerManager:
            nonlocal lease_active
            lease_active = True
            events.append("enter")
            return worker_manager

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            nonlocal lease_active
            lease_active = False
            events.append("exit")

    def _fake_build_worker_routing_payload(
        *,
        runtime_paths: RuntimePaths,
        tool_name: str,
        function_name: str,
        worker_target: ResolvedWorkerTarget | None,
        worker_manager: object | None = None,
        progress_sink: object = None,
    ) -> tuple[dict[str, object], WorkerHandle | None]:
        _ = worker_target, progress_sink
        assert runtime_paths is not None
        assert tool_name == "calculator"
        assert function_name == "add"
        assert worker_manager is worker_manager_ref
        events.append("build")
        return {"worker_key": "worker-key"}, worker_handle

    def _fake_lease_primary_worker_manager(*_args: object, **_kwargs: object) -> _FakeLease:
        return _FakeLease()

    worker_manager_ref = worker_manager
    monkeypatch.setattr(sandbox_proxy_module, "lease_primary_worker_manager", _fake_lease_primary_worker_manager)
    monkeypatch.setattr(sandbox_proxy_module, "_build_worker_routing_payload", _fake_build_worker_routing_payload)
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            responder=lambda _url, _json: {"ok": True, "result": "proxied"},
        ),
    )

    result = sandbox_proxy_module._call_proxy_sync(
        runtime_paths=runtime_paths,
        tool_name="calculator",
        function_name="add",
        args=(1, 2),
        kwargs={},
        credentials_manager=None,
    )

    assert result == "proxied"
    assert events == ["enter", "build", "touch", "exit"]


def test_proxy_holds_manager_lease_through_failure_bookkeeping(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Dedicated worker failure bookkeeping should run before the manager lease is released."""
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="all",
    )
    events: list[str] = []
    lease_active = False
    worker_handle = WorkerHandle(
        worker_id="worker-1",
        worker_key="worker-key",
        endpoint="http://worker/api/sandbox-runner/execute",
        auth_token=_TEST_AUTH_TOKEN,
        status="ready",
        backend_name="docker",
        last_used_at=1.0,
        created_at=0.0,
    )

    class _FakeWorkerManager:
        def touch_worker(self, worker_key: str, *, now: float | None = None) -> WorkerHandle | None:
            _ = now
            msg = f"unexpected touch bookkeeping for {worker_key}"
            raise AssertionError(msg)

        def record_failure(self, worker_key: str, failure_reason: str, *, now: float | None = None) -> WorkerHandle:
            nonlocal lease_active
            _ = now
            assert lease_active is True
            assert worker_key == "worker-key"
            assert failure_reason == "boom"
            events.append("record_failure")
            return worker_handle

    worker_manager = _FakeWorkerManager()

    class _FakeLease:
        def __enter__(self) -> _FakeWorkerManager:
            nonlocal lease_active
            lease_active = True
            events.append("enter")
            return worker_manager

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            nonlocal lease_active
            lease_active = False
            events.append("exit")

    def _fake_build_worker_routing_payload(
        *,
        runtime_paths: RuntimePaths,
        tool_name: str,
        function_name: str,
        worker_target: ResolvedWorkerTarget | None,
        worker_manager: object | None = None,
        progress_sink: object = None,
    ) -> tuple[dict[str, object], WorkerHandle | None]:
        _ = runtime_paths, worker_target, progress_sink
        assert tool_name == "calculator"
        assert function_name == "add"
        assert worker_manager is worker_manager_ref
        events.append("build")
        return {"worker_key": "worker-key"}, worker_handle

    def _fake_lease_primary_worker_manager(*_args: object, **_kwargs: object) -> _FakeLease:
        return _FakeLease()

    worker_manager_ref = worker_manager
    monkeypatch.setattr(sandbox_proxy_module, "lease_primary_worker_manager", _fake_lease_primary_worker_manager)
    monkeypatch.setattr(sandbox_proxy_module, "_build_worker_routing_payload", _fake_build_worker_routing_payload)
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            responder=lambda _url, _json: {"ok": False, "error": "boom"},
        ),
    )

    with pytest.raises(RuntimeError, match="boom"):
        sandbox_proxy_module._call_proxy_sync(
            runtime_paths=runtime_paths,
            tool_name="calculator",
            function_name="add",
            args=(1, 2),
            kwargs={},
            credentials_manager=None,
        )

    assert events == ["enter", "build", "record_failure", "exit"]


def test_proxy_requires_shared_token(monkeypatch: pytest.MonkeyPatch) -> None:
    """Proxy mode should fail closed when no shared token is configured."""

    class _FakeClient:
        def __init__(self, *_args: object, **_kwargs: object) -> None:
            return

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, *_args: object, **_kwargs: object) -> None:
            msg = "Proxy client should not make requests without a shared token."
            raise AssertionError(msg)

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=None,
        execution_mode="all",
    )
    monkeypatch.setattr("mindroom.tool_system.sandbox_proxy.httpx.Client", _FakeClient)

    tool = get_tool_by_name("calculator", runtime_paths, worker_target=None)
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    with pytest.raises(RuntimeError, match="MINDROOM_SANDBOX_PROXY_TOKEN"):
        entrypoint(1, 2)


def test_proxy_prefers_worker_scoped_credentials_for_worker_routed_calls(monkeypatch: pytest.MonkeyPatch) -> None:
    """Worker-routed credential leases should prefer credentials stored in the resolved worker scope."""
    captured_calls: list[tuple[str, dict[str, Any]]] = []

    class _FakeResponse:
        def __init__(self, data: dict[str, object]) -> None:
            self._data = data

        def raise_for_status(self) -> None:
            return

        def json(self) -> dict[str, object]:
            return self._data

    class _FakeClient:
        def __init__(self, *, timeout: float) -> None:
            self.timeout = timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            _ = headers
            captured_calls.append((url, json))
            if url.endswith("/leases"):
                return _FakeResponse({"lease_id": "lease-123", "expires_at": 123.0, "max_uses": 1})
            return _FakeResponse({"ok": True, "result": "proxied"})

    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    worker_key = resolve_worker_key("user", execution_identity, agent_name="code")
    assert worker_key is not None
    fake_credentials = FakeCredentialsManager(
        {"openai": {"api_key": "shared-key", "_source": "ui"}},
        worker_managers={
            worker_key: FakeCredentialsManager({"openai": {"api_key": "worker-key", "_source": "ui"}}),
        },
    )

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
        credential_policy={"calculator.add": ("openai",)},
    )
    monkeypatch.setattr("mindroom.tool_system.sandbox_proxy.httpx.Client", _FakeClient)

    tool = get_tool_by_name(
        "calculator",
        runtime_paths,
        credentials_manager=fake_credentials,
        worker_tools_override=["calculator"],
        worker_target=_worker_target(runtime_paths, "user", "code", execution_identity),
    )
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    result = entrypoint(1, 2)

    assert result == "proxied"
    lease_url, lease_payload = captured_calls[0]
    assert lease_url.endswith("/api/sandbox-runner/leases")
    assert lease_payload["credential_overrides"] == {"api_key": "worker-key"}


def test_proxy_worker_routed_lease_skips_non_grantable_shared_credentials(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Worker-routed leases must not expose shared credentials outside worker_grantable_credentials."""
    captured_calls: list[tuple[str, dict[str, Any]]] = []

    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    fake_credentials = FakeCredentialsManager({"openai": {"api_key": "shared-key", "_source": "ui"}})

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
        credential_policy={"calculator.add": ("openai",)},
    )
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            captured_calls=captured_calls,
            responder=lambda _url, _json: {"ok": True, "result": "proxied"},
        ),
    )

    tool = get_tool_by_name(
        "calculator",
        runtime_paths,
        credentials_manager=fake_credentials,
        worker_tools_override=["calculator"],
        worker_target=_worker_target(runtime_paths, "user", "code", execution_identity),
    )
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id="$thread",
            reply_to_event_id=None,
        ),
        requester_id="@alice:example.org",
        client=object(),
        config=Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    worker_scope="user",
                ),
            },
            models={},
        ),
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
    )

    with tool_runtime_context(runtime_context):
        result = entrypoint(1, 2)

    assert result == "proxied"
    assert len(captured_calls) == 1
    execute_url, execute_payload = captured_calls[0]
    assert execute_url.endswith("/api/sandbox-runner/execute")
    assert "lease_id" not in execute_payload


def test_proxy_includes_worker_routing_identity(monkeypatch: pytest.MonkeyPatch) -> None:
    """Worker-routed tool calls should include scope, key, and execution identity."""
    captured: dict[str, Any] = {}

    class _FakeResponse:
        def raise_for_status(self) -> None:
            return

        def json(self) -> dict[str, object]:
            return {"ok": True, "result": "sandbox-result"}

    class _FakeClient:
        def __init__(self, *, timeout: float) -> None:
            captured["timeout"] = timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            captured["url"] = url
            captured["json"] = json
            captured["headers"] = headers
            return _FakeResponse()

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
        credential_policy={},
    )
    monkeypatch.setattr("mindroom.tool_system.sandbox_proxy.httpx.Client", _FakeClient)

    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="shared_agent",
        requester_id="alice",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
        transport_agent_name="shared_agent",
    )

    tool = get_tool_by_name(
        "calculator",
        runtime_paths,
        worker_tools_override=["calculator"],
        worker_target=_worker_target(
            runtime_paths,
            "user_agent",
            "code",
            execution_identity,
            private_agent_names=frozenset({"code"}),
        ),
    )
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    expected_worker_key = resolve_worker_key("user_agent", execution_identity, agent_name="code")
    assert expected_worker_key is not None

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id="$thread",
            reply_to_event_id=None,
        ),
        requester_id="alice",
        client=object(),
        config=Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    private=AgentPrivateConfig(per="user_agent"),
                ),
            },
            models={},
        ),
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
    )

    with tool_runtime_context(runtime_context):
        result = entrypoint(1, 2)

    assert result == "sandbox-result"
    assert captured["headers"] == {"x-mindroom-sandbox-token": "test-token"}
    assert captured["json"]["worker_scope"] == "user_agent"
    assert captured["json"]["worker_key"] == expected_worker_key
    assert captured["json"]["execution_identity"] == {
        "channel": "matrix",
        "agent_name": "shared_agent",
        "requester_id": "alice",
        "room_id": "!room:example.org",
        "thread_id": "$thread",
        "resolved_thread_id": "$thread",
        "session_id": "session-1",
        "tenant_id": None,
        "account_id": None,
        "transport_agent_name": "shared_agent",
    }
    assert captured["json"]["private_agent_names"] == ["code"]


def test_proxy_user_agent_shared_agent_sends_explicit_empty_private_visibility(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Shared user-agent workers should still send explicit empty private visibility."""
    captured: dict[str, Any] = {}

    class _FakeResponse:
        def raise_for_status(self) -> None:
            return

        def json(self) -> dict[str, object]:
            return {"ok": True, "result": "sandbox-result"}

    class _FakeClient:
        def __init__(self, *, timeout: float) -> None:
            captured["timeout"] = timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            captured["url"] = url
            captured["json"] = json
            captured["headers"] = headers
            return _FakeResponse()

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
        credential_policy={},
    )
    monkeypatch.setattr("mindroom.tool_system.sandbox_proxy.httpx.Client", _FakeClient)

    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="shared_agent",
        requester_id="alice",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )

    tool = get_tool_by_name(
        "calculator",
        runtime_paths,
        worker_tools_override=["calculator"],
        worker_target=_worker_target(
            runtime_paths,
            "user_agent",
            "code",
            execution_identity,
            private_agent_names=frozenset(),
        ),
    )
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    expected_worker_key = resolve_worker_key("user_agent", execution_identity, agent_name="code")
    assert expected_worker_key is not None

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id="$thread",
            reply_to_event_id=None,
        ),
        requester_id="alice",
        client=object(),
        config=Config(
            agents={
                "code": AgentConfig(
                    display_name="Code",
                    worker_scope="user_agent",
                ),
            },
            models={},
        ),
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
    )

    with tool_runtime_context(runtime_context):
        result = entrypoint(1, 2)

    assert result == "sandbox-result"
    assert captured["json"]["worker_key"] == expected_worker_key
    assert captured["json"]["private_agent_names"] == []


def test_unscoped_dedicated_worker_payload_reconciles_via_ensure_worker(monkeypatch: pytest.MonkeyPatch) -> None:
    """Dedicated worker routing should always reconcile through ensure_worker before use."""

    class _FakeWorkerManager:
        def __init__(self, ensured_handle: WorkerHandle) -> None:
            self.ensured_handle = ensured_handle
            self.ensure_calls: list[str] = []

        def ensure_worker(
            self,
            spec: WorkerSpec,
            *,
            now: float | None = None,
            progress_sink: object = None,
        ) -> WorkerHandle:
            _ = now, progress_sink
            self.ensure_calls.append(spec.worker_key)
            return self.ensured_handle

    refreshed_handle = WorkerHandle(
        worker_id="worker-a",
        worker_key="worker-a",
        endpoint="http://sandbox-runner:8765/api/sandbox-runner/execute",
        auth_token=_TEST_AUTH_TOKEN,
        status="ready",
        backend_name="docker",
        last_used_at=20.0,
        created_at=10.0,
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
        credential_policy={},
    )
    fake_manager = _FakeWorkerManager(refreshed_handle)
    monkeypatch.setattr(
        sandbox_proxy_module,
        "_get_worker_manager",
        lambda _runtime_paths_arg, _proxy_config_arg: fake_manager,
    )
    monkeypatch.setattr(sandbox_proxy_module, "primary_worker_backend_is_dedicated", lambda _runtime_paths_arg: True)

    payload, worker_handle = sandbox_proxy_module._build_worker_routing_payload(
        runtime_paths=runtime_paths,
        tool_name="shell",
        function_name="run_shell_command",
        worker_target=_worker_target(runtime_paths, None, "code", None),
    )

    assert payload["worker_key"] == "v1:default:unscoped:code"
    assert worker_handle is refreshed_handle
    assert fake_manager.ensure_calls == ["v1:default:unscoped:code"]


def test_scoped_dedicated_worker_payload_preserves_resolved_worker_key(monkeypatch: pytest.MonkeyPatch) -> None:
    """Dedicated worker reconciliation should pass the resolved worker key through unchanged."""

    class _FakeWorkerManager:
        def __init__(self, ensured_handle: WorkerHandle) -> None:
            self.ensured_handle = ensured_handle
            self.ensure_calls: list[str] = []

        def ensure_worker(
            self,
            spec: WorkerSpec,
            *,
            now: float | None = None,
            progress_sink: object = None,
        ) -> WorkerHandle:
            _ = now, progress_sink
            self.ensure_calls.append(spec.worker_key)
            return self.ensured_handle

    ensured_handle = WorkerHandle(
        worker_id="worker-b",
        worker_key="worker-b",
        endpoint="http://sandbox-runner:8765/api/sandbox-runner/execute",
        auth_token=_TEST_AUTH_TOKEN,
        status="idle",
        backend_name="docker",
        last_used_at=15.0,
        created_at=10.0,
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
        credential_policy={},
    )
    fake_manager = _FakeWorkerManager(ensured_handle)
    monkeypatch.setattr(
        sandbox_proxy_module,
        "_get_worker_manager",
        lambda _runtime_paths_arg, _proxy_config_arg: fake_manager,
    )
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    expected_worker_key = resolve_worker_key("user", execution_identity, agent_name="code")
    assert expected_worker_key is not None

    payload, worker_handle = sandbox_proxy_module._build_worker_routing_payload(
        runtime_paths=runtime_paths,
        tool_name="shell",
        function_name="run_shell_command",
        worker_target=_worker_target(runtime_paths, "user", "code", execution_identity),
    )

    assert payload["worker_key"] == expected_worker_key
    assert worker_handle is ensured_handle
    assert fake_manager.ensure_calls == [expected_worker_key]


def test_proxy_surfaces_runner_http_detail(monkeypatch: pytest.MonkeyPatch) -> None:
    """Proxy should preserve sandbox-runner detail messages on HTTP failures."""

    class _FakeResponse:
        def raise_for_status(self) -> None:
            request = httpx.Request("POST", "http://sandbox-runner:8765/api/sandbox-runner/execute")
            response = httpx.Response(
                400,
                request=request,
                json={"detail": "base_dir must stay inside the allowed state roots or worker root"},
            )
            message = "bad request"
            raise httpx.HTTPStatusError(message, request=request, response=response)

        def json(self) -> dict[str, object]:
            msg = "Success payload should not be parsed for HTTP failures."
            raise AssertionError(msg)

    class _FakeClient:
        def __init__(self, *, timeout: float) -> None:
            _ = timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            _ = url, json, headers
            return _FakeResponse()

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="all",
        credential_policy={},
    )
    monkeypatch.setattr("mindroom.tool_system.sandbox_proxy.httpx.Client", _FakeClient)

    tool = get_tool_by_name("calculator", runtime_paths, worker_target=None)
    entrypoint = tool.functions["add"].entrypoint
    assert entrypoint is not None
    with pytest.raises(RuntimeError, match="base_dir must stay inside the allowed state roots or worker root"):
        entrypoint(1, 2)


def test_static_sandbox_runner_backend_reuses_worker_handle_identity() -> None:
    """The current shared sandbox-runner provider should return stable handle identity per worker key."""
    backend = StaticSandboxRunnerBackend(
        api_root="http://sandbox-runner:8765",
        auth_token=_TEST_AUTH_TOKEN,
        idle_timeout_seconds=60.0,
    )

    first = backend.ensure_worker(WorkerSpec("worker-a"), now=10.0)
    second = backend.ensure_worker(WorkerSpec("worker-a"), now=20.0)

    assert first.worker_id == second.worker_id
    assert second.worker_key == "worker-a"
    assert second.endpoint == "http://sandbox-runner:8765/api/sandbox-runner/execute"
    assert second.auth_token == _TEST_AUTH_TOKEN
    assert second.backend_name == "static_sandbox_runner"
    assert second.startup_count == 1
    assert second.last_used_at == 20.0


def test_static_sandbox_runner_backend_marks_idle_workers() -> None:
    """Idle cleanup on the static provider should preserve worker identity while changing lifecycle state."""
    backend = StaticSandboxRunnerBackend(
        api_root="http://sandbox-runner:8765",
        auth_token=_TEST_AUTH_TOKEN,
        idle_timeout_seconds=5.0,
    )
    backend.ensure_worker(WorkerSpec("worker-a"), now=0.0)

    cleaned_workers = backend.cleanup_idle_workers(now=10.0)

    assert len(cleaned_workers) == 1
    assert cleaned_workers[0].worker_key == "worker-a"
    assert cleaned_workers[0].status == "idle"


def test_get_worker_manager_singleton_creation_is_thread_safe(monkeypatch: pytest.MonkeyPatch) -> None:
    """Concurrent proxy requests should not build multiple static worker managers for one config."""
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner-thread-safe:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode=None,
    )
    proxy_config = sandbox_proxy_module.sandbox_proxy_config(runtime_paths)
    monkeypatch.delenv("MINDROOM_WORKER_BACKEND", raising=False)

    first_init_started = threading.Event()
    allow_first_init_to_finish = threading.Event()
    init_count_lock = threading.Lock()
    init_count = 0
    managers: list[object] = []
    exceptions: list[Exception] = []

    class FakeBackend:
        backend_name = "fake_static_backend"
        idle_timeout_seconds = 60.0

        def __init__(self, *, api_root: str, auth_token: str | None) -> None:
            del api_root, auth_token
            nonlocal init_count
            with init_count_lock:
                init_count += 1
                call_number = init_count
            if call_number == 1:
                first_init_started.set()
                assert allow_first_init_to_finish.wait(timeout=1.0)

        def shutdown(self) -> None:
            return None

    def load_manager() -> None:
        try:
            managers.append(sandbox_proxy_module._get_worker_manager(runtime_paths, proxy_config))
        except Exception as exc:  # pragma: no cover - surfaced by assertion below
            exceptions.append(exc)

    monkeypatch.setattr(workers_runtime_module, "StaticSandboxRunnerBackend", FakeBackend)

    first_thread = threading.Thread(target=load_manager)
    second_thread = threading.Thread(target=load_manager)

    first_thread.start()
    assert first_init_started.wait(timeout=1.0)
    second_thread.start()
    assert init_count == 1
    allow_first_init_to_finish.set()
    first_thread.join(timeout=1.0)
    second_thread.join(timeout=1.0)

    assert exceptions == []
    assert init_count == 1
    assert len(managers) == 2
    assert managers[0] is managers[1]


def test_docker_worker_manager_rebuilds_when_runtime_storage_path_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Changing the runtime storage path should rebuild the Docker worker manager with the new root."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    built_storage_paths: list[Path] = []
    built_runtime_paths: list[RuntimePaths] = []

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path is not None
            built_storage_paths.append(storage_path)
            built_runtime_paths.append(runtime_paths)
            return cls()

        def shutdown(self) -> None:
            return None

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)
    monkeypatch.setattr(
        workers_runtime_module,
        "docker_backend_config_signature",
        lambda runtime_paths, *, auth_token, storage_path=None, worker_grantable_credentials=frozenset(): (
            "docker",
            auth_token or "",
            str(storage_path),
            runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") or "",
            *sorted(worker_grantable_credentials),
        ),
    )

    config_path = tmp_path / "config.yaml"
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
        },
    )
    first_storage_path = (tmp_path / "runtime-a").resolve()
    second_storage_path = (tmp_path / "runtime-b").resolve()

    try:
        first_manager = workers_runtime_module.get_primary_worker_manager(
            runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=first_storage_path,
        )

        second_manager = workers_runtime_module.get_primary_worker_manager(
            runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )

        assert built_storage_paths == [first_storage_path, second_storage_path]
        assert built_runtime_paths == [runtime_paths, runtime_paths]
        assert first_manager is not second_manager
    finally:
        workers_runtime_module._reset_primary_worker_manager()


def test_shutdown_primary_worker_manager_resets_cached_runtime_manager() -> None:
    """Shutting down the primary manager should tear down the cached manager."""
    workers_runtime_module._reset_primary_worker_manager()
    shutdown_calls: list[str] = []

    class _FakeWorkerManager:
        def __init__(self, name: str) -> None:
            self._name = name

        def shutdown(self) -> None:
            shutdown_calls.append(self._name)

    workers_runtime_module._PRIMARY_WORKER_MANAGER_ENTRY = workers_runtime_module._WorkerManagerEntry(
        manager=_FakeWorkerManager("cached"),
        config_signature=("docker", "cached"),
    )

    workers_runtime_module.shutdown_primary_worker_manager(timeout_seconds=0.0)

    assert shutdown_calls == ["cached"]
    assert workers_runtime_module._PRIMARY_WORKER_MANAGER_ENTRY is None
    assert workers_runtime_module._RETIRED_PRIMARY_WORKER_MANAGER_ENTRIES == []


def test_docker_worker_manager_retires_obsolete_cached_manager_until_lease_release(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A leased manager should survive replacement until the request-scoped lease is released."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    build_order: list[str] = []

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path is not None
            assert runtime_paths.storage_root == storage_path
            build_order.append(str(storage_path))
            backend = cls()
            backend.shutdown_calls = 0
            return backend

        def shutdown(self) -> None:
            self.shutdown_calls += 1

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)
    monkeypatch.setattr(
        workers_runtime_module,
        "docker_backend_config_signature",
        lambda runtime_paths, *, auth_token, storage_path=None, worker_grantable_credentials=frozenset(): (
            "docker",
            auth_token or "",
            str(storage_path),
            runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") or "",
            *sorted(worker_grantable_credentials),
        ),
    )

    first_storage_path = (tmp_path / "runtime-a").resolve()
    second_storage_path = (tmp_path / "runtime-b").resolve()
    first_runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config-a.yaml", storage_path=first_storage_path)
    second_runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config-b.yaml",
        storage_path=second_storage_path,
    )

    with workers_runtime_module.lease_primary_worker_manager(
        first_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=first_storage_path,
    ) as first_manager:
        second_manager = workers_runtime_module.get_primary_worker_manager(
            second_runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )
        repeated_second_manager = workers_runtime_module.get_primary_worker_manager(
            second_runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )
        assert build_order == [str(first_storage_path), str(second_storage_path)]
        assert first_manager is not second_manager
        assert second_manager is repeated_second_manager
        assert first_manager.shutdown_calls == 0
        assert second_manager.shutdown_calls == 0

    assert first_manager.shutdown_calls == 1
    assert build_order == [str(first_storage_path), str(second_storage_path)]
    workers_runtime_module._reset_primary_worker_manager()
    assert second_manager.shutdown_calls == 1


def test_shutdown_primary_worker_manager_keeps_leased_retired_manager_tracked_until_release(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A timed-out shutdown should keep leased retired managers so lease release can drain them."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path is not None
            assert runtime_paths.storage_root == storage_path
            backend = cls()
            backend.shutdown_calls = 0
            return backend

        def shutdown(self) -> None:
            self.shutdown_calls += 1

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)
    monkeypatch.setattr(
        workers_runtime_module,
        "docker_backend_config_signature",
        lambda runtime_paths, *, auth_token, storage_path=None, worker_grantable_credentials=frozenset(): (
            "docker",
            auth_token or "",
            str(storage_path),
            runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") or "",
            *sorted(worker_grantable_credentials),
        ),
    )

    storage_path = (tmp_path / "runtime-a").resolve()
    runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config-a.yaml", storage_path=storage_path)
    lease = workers_runtime_module.lease_primary_worker_manager(
        runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=storage_path,
    )
    manager = lease.manager

    workers_runtime_module.shutdown_primary_worker_manager(timeout_seconds=0.0)

    assert workers_runtime_module._PRIMARY_WORKER_MANAGER_ENTRY is None
    assert [lease._entry] == workers_runtime_module._RETIRED_PRIMARY_WORKER_MANAGER_ENTRIES
    assert manager.shutdown_calls == 0

    lease.release()

    assert manager.shutdown_calls == 1
    assert workers_runtime_module._RETIRED_PRIMARY_WORKER_MANAGER_ENTRIES == []


def test_docker_worker_manager_preserves_cached_manager_when_new_build_fails(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A failed build for a new signature should not evict the last known-good manager."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    first_storage_path = (tmp_path / "runtime-a").resolve()
    second_storage_path = (tmp_path / "runtime-b").resolve()
    build_order: list[str] = []

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path is not None
            assert runtime_paths.storage_root == storage_path
            build_order.append(str(storage_path))
            if storage_path == second_storage_path:
                msg = "boom"
                raise WorkerBackendError(msg)
            backend = cls()
            backend.shutdown_calls = 0
            return backend

        def shutdown(self) -> None:
            self.shutdown_calls += 1

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)
    monkeypatch.setattr(
        workers_runtime_module,
        "docker_backend_config_signature",
        lambda runtime_paths, *, auth_token, storage_path=None, worker_grantable_credentials=frozenset(): (
            "docker",
            auth_token or "",
            str(storage_path),
            runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") or "",
            *sorted(worker_grantable_credentials),
        ),
    )

    first_runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config-a.yaml", storage_path=first_storage_path)
    second_runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config-b.yaml",
        storage_path=second_storage_path,
    )

    first_manager = workers_runtime_module.get_primary_worker_manager(
        first_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=first_storage_path,
    )

    with pytest.raises(WorkerBackendError, match="boom"):
        workers_runtime_module.get_primary_worker_manager(
            second_runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )

    repeated_first_manager = workers_runtime_module.get_primary_worker_manager(
        first_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=first_storage_path,
    )

    assert build_order == [str(first_storage_path), str(second_storage_path)]
    assert repeated_first_manager is first_manager
    assert first_manager.shutdown_calls == 0
    workers_runtime_module._reset_primary_worker_manager()
    assert first_manager.shutdown_calls == 1


def test_docker_worker_manager_replacement_succeeds_even_if_previous_shutdown_would_raise(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Retired-manager shutdown failures should not poison the replacement manager or later requests."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path is not None
            assert runtime_paths.storage_root == storage_path
            backend = cls()
            backend.shutdown_calls = 0
            backend.raise_on_shutdown = False
            return backend

        def shutdown(self) -> None:
            self.shutdown_calls += 1
            if self.raise_on_shutdown:
                msg = "old shutdown boom"
                raise WorkerBackendError(msg)

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)
    monkeypatch.setattr(
        workers_runtime_module,
        "docker_backend_config_signature",
        lambda runtime_paths, *, auth_token, storage_path=None, worker_grantable_credentials=frozenset(): (
            "docker",
            auth_token or "",
            str(storage_path),
            runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") or "",
            *sorted(worker_grantable_credentials),
        ),
    )

    first_storage_path = (tmp_path / "runtime-a").resolve()
    second_storage_path = (tmp_path / "runtime-b").resolve()
    first_runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config-a.yaml", storage_path=first_storage_path)
    second_runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config-b.yaml",
        storage_path=second_storage_path,
    )

    with workers_runtime_module.lease_primary_worker_manager(
        first_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=first_storage_path,
    ) as first_manager:
        first_manager.raise_on_shutdown = True

        second_manager = workers_runtime_module.get_primary_worker_manager(
            second_runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )
        repeated_second_manager = workers_runtime_module.get_primary_worker_manager(
            second_runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )

        assert second_manager is repeated_second_manager
        assert first_manager.shutdown_calls == 0
        assert second_manager.shutdown_calls == 0

    assert first_manager.shutdown_calls == 1
    assert (
        workers_runtime_module.get_primary_worker_manager(
            second_runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )
        is second_manager
    )

    first_manager.raise_on_shutdown = False
    workers_runtime_module._reset_primary_worker_manager()
    assert first_manager.shutdown_calls == 1
    assert second_manager.shutdown_calls == 1


def test_docker_worker_manager_build_does_not_hold_cache_lock(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Slow manager construction must not block unrelated cache-lock reads."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    build_started = threading.Event()
    allow_build = threading.Event()

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = runtime_paths, storage_path, worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            build_started.set()
            assert allow_build.wait(timeout=5.0)
            return cls()

        def shutdown(self) -> None:
            return None

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)
    monkeypatch.setattr(
        workers_runtime_module,
        "docker_backend_config_signature",
        lambda runtime_paths, *, auth_token, storage_path=None, worker_grantable_credentials=frozenset(): (
            "docker",
            auth_token or "",
            str(storage_path),
            runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") or "",
            *sorted(worker_grantable_credentials),
        ),
    )

    storage_path = (tmp_path / "runtime-a").resolve()
    runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config-a.yaml", storage_path=storage_path)
    thread_error: list[BaseException] = []

    def _build_manager() -> None:
        try:
            workers_runtime_module.get_primary_worker_manager(
                runtime_paths,
                proxy_url=None,
                proxy_token=_TEST_AUTH_TOKEN,
                storage_root=storage_path,
            )
        except BaseException as exc:  # pragma: no cover - surfaced by assertion below
            thread_error.append(exc)

    build_thread = threading.Thread(target=_build_manager)
    build_thread.start()
    try:
        assert build_started.wait(timeout=5.0)
        assert workers_runtime_module._PRIMARY_WORKER_MANAGER_CONDITION.acquire(timeout=0.5)
        workers_runtime_module._PRIMARY_WORKER_MANAGER_CONDITION.release()
    finally:
        allow_build.set()
        build_thread.join(timeout=5.0)

    assert thread_error == []
    assert not build_thread.is_alive()
    workers_runtime_module._reset_primary_worker_manager()


def test_docker_worker_manager_replacement_does_not_hold_cache_lock_during_retired_shutdown(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """A slow retired-manager shutdown must not block cache reads of the new active manager."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    shutdown_started = threading.Event()
    allow_shutdown = threading.Event()

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path is not None
            assert runtime_paths.storage_root == storage_path
            backend = cls()
            backend.block_shutdown = False
            return backend

        def shutdown(self) -> None:
            if not self.block_shutdown:
                return
            shutdown_started.set()
            assert allow_shutdown.wait(timeout=5.0)

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)
    monkeypatch.setattr(
        workers_runtime_module,
        "docker_backend_config_signature",
        lambda runtime_paths, *, auth_token, storage_path=None, worker_grantable_credentials=frozenset(): (
            "docker",
            auth_token or "",
            str(storage_path),
            runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH") or "",
            *sorted(worker_grantable_credentials),
        ),
    )

    first_storage_path = (tmp_path / "runtime-a").resolve()
    second_storage_path = (tmp_path / "runtime-b").resolve()
    first_runtime_paths = resolve_runtime_paths(config_path=tmp_path / "config-a.yaml", storage_path=first_storage_path)
    second_runtime_paths = resolve_runtime_paths(
        config_path=tmp_path / "config-b.yaml",
        storage_path=second_storage_path,
    )

    first_manager = workers_runtime_module.get_primary_worker_manager(
        first_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=first_storage_path,
    )
    first_manager.block_shutdown = True

    thread_result: dict[str, object] = {}

    def _replace_manager() -> None:
        thread_result["manager"] = workers_runtime_module.get_primary_worker_manager(
            second_runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=second_storage_path,
        )

    replacement_thread = threading.Thread(target=_replace_manager)
    replacement_thread.start()
    assert shutdown_started.wait(timeout=5.0)

    second_manager = workers_runtime_module.get_primary_worker_manager(
        second_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=second_storage_path,
    )

    allow_shutdown.set()
    replacement_thread.join(timeout=5.0)

    assert thread_result["manager"] is second_manager
    workers_runtime_module._reset_primary_worker_manager()


def test_docker_worker_manager_rebuilds_when_runtime_shared_credentials_path_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Changing the runtime shared-credentials path should rebuild the Docker worker manager."""
    workers_runtime_module._reset_primary_worker_manager()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")

    built_shared_paths: list[str | None] = []

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path == tmp_path.resolve()
            built_shared_paths.append(runtime_paths.env_value("MINDROOM_SHARED_CREDENTIALS_PATH"))
            return cls()

        def shutdown(self) -> None:
            return None

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)

    config_path = tmp_path / "config.yaml"
    first_runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_SHARED_CREDENTIALS_PATH": str((tmp_path / "shared-a").resolve()),
        },
    )
    second_runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "MINDROOM_SHARED_CREDENTIALS_PATH": str((tmp_path / "shared-b").resolve()),
        },
    )

    first_manager = workers_runtime_module.get_primary_worker_manager(
        first_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
    )
    second_manager = workers_runtime_module.get_primary_worker_manager(
        second_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
    )

    assert built_shared_paths == [
        str((tmp_path / "shared-a").resolve()),
        str((tmp_path / "shared-b").resolve()),
    ]
    assert first_manager is not second_manager
    workers_runtime_module._reset_primary_worker_manager()


def test_docker_worker_manager_rebuilds_when_committed_runtime_env_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Changing committed worker runtime env should rebuild the Docker worker manager."""
    workers_runtime_module._reset_primary_worker_manager()
    built_browser_paths: list[str | None] = []

    class _FakeDockerBackend:
        backend_name = "docker"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_path: Path | None = None,
            worker_grantable_credentials: frozenset[str] = frozenset(),
        ) -> _FakeDockerBackend:
            _ = worker_grantable_credentials
            assert auth_token == _TEST_AUTH_TOKEN
            assert storage_path == tmp_path.resolve()
            built_browser_paths.append(runtime_paths.env_value("BROWSER_EXECUTABLE_PATH"))
            return cls()

        def shutdown(self) -> None:
            return None

    monkeypatch.setattr(workers_runtime_module, "DockerWorkerBackend", _FakeDockerBackend)

    config_path = tmp_path / "config.yaml"
    first_runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "BROWSER_EXECUTABLE_PATH": "/usr/bin/a",
        },
    )
    second_runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path,
        process_env={
            "MINDROOM_WORKER_BACKEND": "docker",
            "MINDROOM_DOCKER_WORKER_IMAGE": "ghcr.io/mindroom-ai/mindroom:latest",
            "BROWSER_EXECUTABLE_PATH": "/usr/bin/b",
        },
    )

    first_manager = workers_runtime_module.get_primary_worker_manager(
        first_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
    )
    second_manager = workers_runtime_module.get_primary_worker_manager(
        second_runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
    )

    assert built_browser_paths == ["/usr/bin/a", "/usr/bin/b"]
    assert first_manager is not second_manager
    workers_runtime_module._reset_primary_worker_manager()


def test_get_worker_manager_rebuilds_kubernetes_backend_when_committed_snapshot_changes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Kubernetes worker-manager caching should track authoritative committed snapshots."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    first_snapshot = dict(_TEST_KUBERNETES_VALIDATION_SNAPSHOT)
    second_snapshot = {
        **_TEST_KUBERNETES_VALIDATION_SNAPSHOT,
        "worker_only": {
            "config_fields": [],
            "agent_override_fields": [],
            "authored_override_validator": "default",
            "runtime_loadable": False,
        },
    }
    second_config_snapshot = {
        **_TEST_KUBERNETES_CONFIG_SNAPSHOT,
        "agents": {"code": {"knowledge_bases": ["shared_docs"]}},
        "knowledge_bases": {"shared_docs": {"path": "./knowledge/shared-docs"}},
    }
    captured_snapshots: list[dict[str, dict[str, object]]] = []
    captured_config_snapshots: list[dict[str, object]] = []

    class FakeKubernetesBackend:
        backend_name = "kubernetes"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_root: Path,
            tool_validation_snapshot: dict[str, dict[str, object]],
            config_snapshot: dict[str, object],
            worker_grantable_credentials: frozenset[str],
        ) -> Self:
            del runtime_paths, auth_token, storage_root, worker_grantable_credentials
            captured_snapshots.append(tool_validation_snapshot)
            captured_config_snapshots.append(config_snapshot)
            return cls()

        def ensure_worker(self, spec: WorkerSpec, *, now: float | None = None, progress_sink: object = None) -> object:
            del spec, now, progress_sink
            raise NotImplementedError

        def touch_worker(self, worker_key: str, *, now: float | None = None) -> object:
            raise NotImplementedError

        def list_workers(self, *, include_idle: bool = True, now: float | None = None) -> list[object]:
            raise NotImplementedError

        def cleanup_idle_workers(self, *, now: float | None = None) -> list[object]:
            raise NotImplementedError

        def record_failure(self, worker_key: str, failure_reason: str, *, now: float | None = None) -> object:
            raise NotImplementedError

        def shutdown(self) -> None:
            return None

    monkeypatch.setattr(workers_runtime_module, "KubernetesWorkerBackend", FakeKubernetesBackend)

    first_manager = workers_runtime_module.get_primary_worker_manager(
        runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
        kubernetes_tool_validation_snapshot=first_snapshot,
        kubernetes_config_snapshot=_TEST_KUBERNETES_CONFIG_SNAPSHOT,
    )
    second_manager = workers_runtime_module.get_primary_worker_manager(
        runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
        kubernetes_tool_validation_snapshot=second_snapshot,
        kubernetes_config_snapshot=_TEST_KUBERNETES_CONFIG_SNAPSHOT,
    )
    third_manager = workers_runtime_module.get_primary_worker_manager(
        runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
        kubernetes_tool_validation_snapshot=second_snapshot,
        kubernetes_config_snapshot=second_config_snapshot,
    )

    assert first_manager is not second_manager
    assert second_manager is not third_manager
    assert captured_snapshots == [first_snapshot, second_snapshot, second_snapshot]
    assert captured_config_snapshots == [
        _TEST_KUBERNETES_CONFIG_SNAPSHOT,
        _TEST_KUBERNETES_CONFIG_SNAPSHOT,
        second_config_snapshot,
    ]


def test_get_primary_worker_manager_requires_explicit_snapshot_for_kubernetes(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Kubernetes worker-manager lookup should require the caller to provide a committed snapshot."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )

    with pytest.raises(
        WorkerBackendError,
        match="requires an explicit tool validation snapshot",
    ):
        workers_runtime_module.get_primary_worker_manager(
            runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=tmp_path,
        )

    with pytest.raises(
        WorkerBackendError,
        match="requires an explicit committed config snapshot",
    ):
        workers_runtime_module.get_primary_worker_manager(
            runtime_paths,
            proxy_url=None,
            proxy_token=_TEST_AUTH_TOKEN,
            storage_root=tmp_path,
            kubernetes_tool_validation_snapshot=_TEST_KUBERNETES_VALIDATION_SNAPSHOT,
        )


def test_get_primary_worker_manager_reuses_cached_manager_without_rereading_disk_when_snapshot_is_explicit(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Explicit validation snapshots should keep manager lookups independent from later disk drift."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nrouter:\n  model: default\nagents: {}\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env=dict(os.environ),
    )
    runtime_config = load_config(runtime_paths)
    tool_validation_snapshot = serialize_tool_validation_snapshot(
        resolved_tool_validation_snapshot_for_runtime(runtime_paths, runtime_config),
    )
    config_snapshot = workers_runtime_module.serialized_kubernetes_worker_config_snapshot(runtime_config)

    class FakeKubernetesBackend:
        backend_name = "kubernetes"
        idle_timeout_seconds = 60.0

        @classmethod
        def from_runtime(
            cls,
            runtime_paths: RuntimePaths,
            *,
            auth_token: str | None,
            storage_root: Path,
            tool_validation_snapshot: dict[str, dict[str, object]],
            config_snapshot: dict[str, object],
            worker_grantable_credentials: frozenset[str],
        ) -> Self:
            del (
                runtime_paths,
                auth_token,
                storage_root,
                tool_validation_snapshot,
                config_snapshot,
                worker_grantable_credentials,
            )
            return cls()

        def ensure_worker(self, spec: WorkerSpec, *, now: float | None = None, progress_sink: object = None) -> object:
            del spec, now, progress_sink
            raise NotImplementedError

        def touch_worker(self, worker_key: str, *, now: float | None = None) -> object:
            raise NotImplementedError

        def list_workers(self, *, include_idle: bool = True, now: float | None = None) -> list[object]:
            raise NotImplementedError

        def cleanup_idle_workers(self, *, now: float | None = None) -> list[object]:
            raise NotImplementedError

        def record_failure(self, worker_key: str, failure_reason: str, *, now: float | None = None) -> object:
            raise NotImplementedError

        def shutdown(self) -> None:
            return None

    monkeypatch.setattr(workers_runtime_module, "KubernetesWorkerBackend", FakeKubernetesBackend)

    first_manager = workers_runtime_module.get_primary_worker_manager(
        runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
        kubernetes_tool_validation_snapshot=tool_validation_snapshot,
        kubernetes_config_snapshot=config_snapshot,
    )
    config_path.write_text("models: [\n", encoding="utf-8")
    second_manager = workers_runtime_module.get_primary_worker_manager(
        runtime_paths,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        storage_root=tmp_path,
        kubernetes_tool_validation_snapshot=tool_validation_snapshot,
        kubernetes_config_snapshot=config_snapshot,
    )

    assert first_manager is second_manager


def test_get_worker_manager_passes_committed_snapshot_from_tool_runtime_context(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Sandbox proxy worker routing should use the committed tool-runtime config snapshot."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    runtime_config = load_config(runtime_paths)
    captured_kwargs: dict[str, object] = {}

    def _fake_get_primary_worker_manager(*_args: object, **kwargs: object) -> object:
        captured_kwargs.update(kwargs)
        return object()

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id=None,
            reply_to_event_id=None,
        ),
        requester_id="@user:example.org",
        client=object(),
        config=runtime_config,
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
    )
    proxy_config = sandbox_proxy_module.sandbox_proxy_config(runtime_paths)
    monkeypatch.setattr(sandbox_proxy_module, "get_primary_worker_manager", _fake_get_primary_worker_manager)

    with tool_runtime_context(runtime_context):
        sandbox_proxy_module._get_worker_manager(runtime_paths, proxy_config)

    assert captured_kwargs["kubernetes_tool_validation_snapshot"] is not None
    assert captured_kwargs["kubernetes_config_snapshot"] == (
        workers_runtime_module.serialized_kubernetes_worker_config_snapshot(runtime_config)
    )


def test_get_worker_manager_reuses_cached_kubernetes_validation_snapshot(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Repeated proxy worker-manager access should not repeatedly resolve validation metadata."""
    workers_runtime_module.clear_worker_validation_snapshot_cache()
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    runtime_config = Config()
    resolver_call_count = 0
    captured_snapshots: list[dict[str, dict[str, object]]] = []

    def fake_resolver(*_args: object, **_kwargs: object) -> dict[str, ToolValidationInfo]:
        nonlocal resolver_call_count
        resolver_call_count += 1
        return {"fake": ToolValidationInfo(name="fake")}

    def fake_get_primary_worker_manager(*_args: object, **kwargs: object) -> object:
        captured_snapshots.append(kwargs["kubernetes_tool_validation_snapshot"])
        return object()

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id=None,
            reply_to_event_id=None,
        ),
        requester_id="@user:example.org",
        client=object(),
        config=runtime_config,
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
    )
    proxy_config = sandbox_proxy_module.sandbox_proxy_config(runtime_paths)
    monkeypatch.setattr(
        "mindroom.tool_system.catalog.resolved_tool_validation_snapshot_for_runtime",
        fake_resolver,
    )
    monkeypatch.setattr(sandbox_proxy_module, "get_primary_worker_manager", fake_get_primary_worker_manager)

    with tool_runtime_context(runtime_context):
        sandbox_proxy_module._get_worker_manager(runtime_paths, proxy_config)
        sandbox_proxy_module._get_worker_manager(runtime_paths, proxy_config)

    assert resolver_call_count == 1
    assert len(captured_snapshots) == 2
    assert captured_snapshots[0] == captured_snapshots[1]
    assert captured_snapshots[0] is not captured_snapshots[1]


def test_proxy_leases_worker_manager_with_committed_runtime_context(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Request-scoped worker-manager leases must use the committed runtime-context snapshot and credential policy."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    monkeypatch.setenv("MINDROOM_SANDBOX_PROXY_URL", "http://sandbox-runner:8765")
    monkeypatch.setenv("MINDROOM_SANDBOX_PROXY_TOKEN", _TEST_AUTH_TOKEN)
    monkeypatch.setenv("MINDROOM_SANDBOX_EXECUTION_MODE", "all")
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n"
        "  default:\n"
        "    provider: openai\n"
        "    id: gpt-5.4\n"
        "router:\n"
        "  model: default\n"
        "agents: {}\n"
        "defaults:\n"
        "  worker_grantable_credentials: [gmail]\n",
        encoding="utf-8",
    )
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=tmp_path / "storage",
        process_env=dict(os.environ),
    )
    runtime_config = load_config(runtime_paths)
    request_storage_path = (tmp_path / "request-storage").resolve()
    captured_kwargs: dict[str, object] = {}
    fake_worker_manager = object()

    class _FakeLease:
        def __enter__(self) -> object:
            return fake_worker_manager

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return None

    def _fake_lease_primary_worker_manager(*_args: object, **kwargs: object) -> _FakeLease:
        captured_kwargs.update(kwargs)
        return _FakeLease()

    def _fake_build_worker_routing_payload(
        *,
        runtime_paths: RuntimePaths,
        tool_name: str,
        function_name: str,
        worker_target: ResolvedWorkerTarget | None,
        progress_sink: object | None = None,
        worker_manager: object | None = None,
    ) -> tuple[dict[str, object], WorkerHandle | None]:
        del progress_sink, worker_target
        assert runtime_paths is not None
        assert tool_name == "calculator"
        assert function_name == "add"
        assert worker_manager is fake_worker_manager
        return {}, None

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id=None,
            reply_to_event_id=None,
        ),
        requester_id="@user:example.org",
        client=object(),
        config=runtime_config,
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
        storage_path=request_storage_path,
    )
    monkeypatch.setattr(sandbox_proxy_module, "lease_primary_worker_manager", _fake_lease_primary_worker_manager)
    monkeypatch.setattr(sandbox_proxy_module, "_build_worker_routing_payload", _fake_build_worker_routing_payload)
    monkeypatch.setattr(
        "mindroom.tool_system.sandbox_proxy.httpx.Client",
        _recording_client_class(
            responder=lambda _url, _json: {"ok": True, "result": "proxied"},
        ),
    )

    with tool_runtime_context(runtime_context):
        result = sandbox_proxy_module._call_proxy_sync(
            runtime_paths=runtime_paths,
            tool_name="calculator",
            function_name="add",
            args=(1, 2),
            kwargs={},
            credentials_manager=None,
        )

    assert result == "proxied"
    assert captured_kwargs["storage_root"] == request_storage_path
    assert captured_kwargs["kubernetes_tool_validation_snapshot"] is not None
    assert captured_kwargs["worker_grantable_credentials"] == frozenset({"gmail"})


def test_worker_tools_override_can_use_kubernetes_backend_without_proxy_url(monkeypatch: pytest.MonkeyPatch) -> None:
    """Worker-routed tools should stay proxy-enabled when the Kubernetes backend provides worker handles directly."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )

    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
            worker_tools_override=["shell"],
        )
        is True
    )
    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
            worker_tools_override=None,
        )
        is False
    )


def test_kubernetes_backend_keeps_unscoped_env_routing_enabled_without_proxy_url(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Unscoped agents should still route through dedicated workers on the Kubernetes backend."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"shell"},
    )

    assert sandbox_proxy_module.sandbox_proxy_enabled_for_tool("shell", runtime_paths=runtime_paths) is True
    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "calculator",
            runtime_paths=runtime_paths,
        )
        is False
    )


def test_kubernetes_backend_uses_env_routing_for_worker_scoped_agents_without_proxy_url(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Worker-scoped agents should still honor env-based routing on the Kubernetes backend."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"shell"},
    )

    assert sandbox_proxy_module.sandbox_proxy_enabled_for_tool("shell", runtime_paths=runtime_paths) is True
    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "calculator",
            runtime_paths=runtime_paths,
        )
        is False
    )


def test_kubernetes_backend_keeps_wrapping_when_required_config_is_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Kubernetes routing should stay enabled so misconfiguration fails closed at call time."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.delenv("MINDROOM_KUBERNETES_WORKER_IMAGE", raising=False)
    monkeypatch.delenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", raising=False)
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode=None,
    )

    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
            worker_tools_override=["shell"],
        )
        is True
    )


def test_kubernetes_backend_keeps_wrapping_when_proxy_token_is_missing(monkeypatch: pytest.MonkeyPatch) -> None:
    """Kubernetes routing should stay enabled so missing auth fails closed at call time."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    monkeypatch.setenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", "mindroom-storage")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=None,
        execution_mode=None,
    )

    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
            worker_tools_override=["shell"],
        )
        is True
    )


@pytest.mark.asyncio
async def test_kubernetes_backend_misconfiguration_raises_instead_of_running_locally(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Misconfigured Kubernetes worker routing should raise rather than executing in the primary runtime."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "kubernetes")
    monkeypatch.delenv("MINDROOM_KUBERNETES_WORKER_IMAGE", raising=False)
    monkeypatch.delenv("MINDROOM_KUBERNETES_WORKER_STORAGE_PVC_NAME", raising=False)
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    runtime_config = load_config(runtime_paths)

    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        worker_tools_override=["shell"],
        worker_target=_worker_target(runtime_paths, None, "code", None),
    )
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id=None,
            reply_to_event_id=None,
        ),
        requester_id="@user:example.org",
        client=object(),
        config=runtime_config,
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
    )

    with (
        tool_runtime_context(runtime_context),
        pytest.raises(
            WorkerBackendError,
            match="MINDROOM_KUBERNETES_WORKER_IMAGE",
        ),
    ):
        await entrypoint("pwd")


@pytest.mark.asyncio
async def test_sync_only_worker_routed_tool_surfaces_progress_in_real_async_path(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Sync-only proxied tools should emit worker progress before the async call result resolves."""
    release_execute = threading.Event()
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )

    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )

    class _FakeWorkerManager:
        def ensure_worker(self, spec: WorkerSpec, *, now: float | None = None, progress_sink: object = None) -> object:
            del now
            assert progress_sink is not None
            progress_sink(
                WorkerReadyProgress(
                    phase="cold_start",
                    worker_key=spec.worker_key,
                    backend_name="kubernetes",
                    elapsed_seconds=1.0,
                ),
            )
            return WorkerHandle(
                worker_id="worker-1",
                worker_key=spec.worker_key,
                endpoint="http://worker/api/sandbox-runner/execute",
                auth_token=_TEST_AUTH_TOKEN,
                status="ready",
                backend_name="kubernetes",
                last_used_at=0.0,
                created_at=0.0,
            )

        def touch_worker(self, worker_key: str, *, now: float | None = None) -> object:
            del worker_key, now
            return None

    class _BlockingClient:
        def __init__(self, *, timeout: float) -> None:
            del timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            del json, headers
            assert url == "http://worker/api/sandbox-runner/execute"
            release_execute.wait(timeout=1.0)
            return _FakeResponse({"ok": True, "result": "proxied"})

    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=_worker_target(runtime_paths, "shared", "code", execution_identity),
    )
    assert tool.async_functions == {}
    tool = prepend_tool_hook_bridge(
        tool,
        build_tool_hook_bridge(
            HookRegistry.empty(),
            agent_name="code",
            dispatch_context=None,
            config=Config(agents={"code": AgentConfig(display_name="Code")}, models={}),
            runtime_paths=runtime_paths,
        ),
    )

    model = _MinimalModel(id="fake-model", provider="fake")
    agent = Agent(id="code", model=model)
    parsed_tools = parse_tools(agent, [tool], model, async_mode=True)
    read_file = next(
        function for function in parsed_tools if isinstance(function, Function) and function.name == "read_file"
    )

    progress_queue: asyncio.Queue[object] = asyncio.Queue()
    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(_FakeWorkerManager()),
        ),
        patch("mindroom.tool_system.sandbox_proxy.httpx.Client", _BlockingClient),
        worker_progress_pump_scope(asyncio.get_running_loop(), progress_queue),
    ):
        call_task = asyncio.create_task(
            model.arun_function_call(
                FunctionCall(
                    function=read_file,
                    arguments={"file_name": "demo.txt"},
                    call_id="call-1",
                ),
            ),
        )

        progress_event = await asyncio.wait_for(progress_queue.get(), timeout=0.5)
        assert progress_event.tool_name == "file"
        assert progress_event.function_name == "read_file"
        assert progress_event.progress.phase == "cold_start"
        assert call_task.done() is False

        release_execute.set()
        success, _timer, _function_call, result = await call_task

    assert success is True
    assert result.result == "proxied"


def test_worker_routed_tool_error_does_not_record_worker_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Ordinary proxied tool errors should fail the call without tearing down the worker."""
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )

    class _ToolErrorClient:
        def __init__(self, *, timeout: float) -> None:
            del timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            del json, headers
            assert url == "http://worker/api/sandbox-runner/execute"
            return _FakeResponse(
                {
                    "ok": False,
                    "error": "Sandbox tool execution failed: FileNotFoundError: missing.txt",
                    "failure_kind": "tool",
                },
            )

    manager = _TrackingWorkerManager()
    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=_worker_target(runtime_paths, "shared", "code", execution_identity),
    )
    entrypoint = tool.functions["read_file"].entrypoint
    assert entrypoint is not None

    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
        ),
        patch("mindroom.tool_system.sandbox_proxy.httpx.Client", _ToolErrorClient),
        pytest.raises(RuntimeError, match=r"FileNotFoundError: missing\.txt"),
    ):
        entrypoint("missing.txt")

    assert manager.touched == [_worker_target(runtime_paths, "shared", "code", execution_identity).worker_key]
    assert manager.failures == []


def test_worker_routed_oauth_connection_result_survives_proxy_boundary(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Structured OAuth prompts returned by the runner should stay tool-visible."""
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)
    manager = _TrackingWorkerManager()
    expected_result = {
        "error": "Google Drive is not connected for this agent.",
        "oauth_connection_required": True,
        "provider": "google_drive",
        "connect_url": "/api/oauth/google_drive/connect?agent_name=general",
    }

    class _OAuthResultClient:
        def __init__(self, *, timeout: float) -> None:
            del timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            del json, headers
            assert url == "http://worker/api/sandbox-runner/execute"
            return _FakeResponse({"ok": True, "result": expected_result})

    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=worker_target,
    )
    entrypoint = tool.functions["read_file"].entrypoint
    assert entrypoint is not None

    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
        ),
        patch("mindroom.tool_system.sandbox_proxy.httpx.Client", _OAuthResultClient),
    ):
        result = entrypoint("missing.txt")

    assert result == expected_result
    assert manager.touched == [worker_target.worker_key]
    assert manager.failures == []


def test_worker_routed_worker_failure_records_worker_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Worker-level runner failures should still evict broken workers."""
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)
    assert worker_target.worker_key is not None
    manager = _TrackingWorkerManager()

    class _WorkerFailureClient:
        def __init__(self, *, timeout: float) -> None:
            del timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            del json, headers
            assert url == "http://worker/api/sandbox-runner/execute"
            return _FakeResponse(
                {
                    "ok": False,
                    "error": "Sandbox subprocess timed out.",
                    "failure_kind": "worker",
                },
            )

    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=worker_target,
    )
    entrypoint = tool.functions["read_file"].entrypoint
    assert entrypoint is not None

    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
        ),
        patch("mindroom.tool_system.sandbox_proxy.httpx.Client", _WorkerFailureClient),
        pytest.raises(RuntimeError, match=r"Sandbox subprocess timed out\."),
    ):
        entrypoint("missing.txt")

    assert manager.touched == []
    assert manager.failures == [(worker_target.worker_key, "Sandbox subprocess timed out.")]


def test_worker_routed_legacy_structured_failure_records_worker_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Structured failures without failure_kind should fail closed as worker failures."""
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)
    assert worker_target.worker_key is not None
    manager = _TrackingWorkerManager()

    class _LegacyFailureClient:
        def __init__(self, *, timeout: float) -> None:
            del timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            del json, headers
            assert url == "http://worker/api/sandbox-runner/execute"
            return _FakeResponse(
                {
                    "ok": False,
                    "error": "Sandbox subprocess timed out.",
                },
            )

    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=worker_target,
    )
    entrypoint = tool.functions["read_file"].entrypoint
    assert entrypoint is not None

    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
        ),
        patch("mindroom.tool_system.sandbox_proxy.httpx.Client", _LegacyFailureClient),
        pytest.raises(RuntimeError, match=r"Sandbox subprocess timed out\."),
    ):
        entrypoint("missing.txt")

    assert manager.touched == []
    assert manager.failures == [(worker_target.worker_key, "Sandbox subprocess timed out.")]


def test_worker_routed_malformed_success_payload_records_worker_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    """Malformed 2xx worker responses should evict the worker before raising."""
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)
    assert worker_target.worker_key is not None
    manager = _TrackingWorkerManager()

    class _MalformedSuccessClient:
        def __init__(self, *, timeout: float) -> None:
            del timeout

        def __enter__(self) -> Self:
            return self

        def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
            return

        def post(self, url: str, *, json: dict[str, Any], headers: dict[str, str]) -> _FakeResponse:
            del json, headers
            assert url == "http://worker/api/sandbox-runner/execute"
            return _FakeResponse(["not", "an", "object"])

    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=worker_target,
    )
    entrypoint = tool.functions["read_file"].entrypoint
    assert entrypoint is not None

    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
        ),
        patch("mindroom.tool_system.sandbox_proxy.httpx.Client", _MalformedSuccessClient),
        pytest.raises(TypeError, match=r"Sandbox proxy returned a non-object response\."),
    ):
        entrypoint("missing.txt")

    assert manager.touched == []
    assert manager.failures == [(worker_target.worker_key, "Sandbox proxy returned a non-object response.")]


@pytest.mark.parametrize(
    ("status_code", "payload"),
    [
        (400, {"detail": "credential_overrides must be supplied via lease_id."}),
        (422, {"detail": "Input should be a valid dictionary"}),
        (422, {"detail": [{"msg": "Input should be a valid dictionary", "type": "dict_type"}]}),
        (404, {"detail": "Tool 'file' does not expose 'read_file'."}),
    ],
)
def test_worker_routed_http_request_error_does_not_record_worker_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    status_code: int,
    payload: dict[str, object],
) -> None:
    """Client-side HTTP errors from a healthy worker should fail the call without tearing down the worker."""
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)
    assert worker_target.worker_key is not None
    manager = _TrackingWorkerManager()
    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=worker_target,
    )
    entrypoint = tool.functions["read_file"].entrypoint
    assert entrypoint is not None

    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
        ),
        patch(
            "mindroom.tool_system.sandbox_proxy.httpx.Client",
            _http_status_client_class(status_code=status_code, payload=payload),
        ),
        pytest.raises(httpx.HTTPStatusError),
    ):
        entrypoint("missing.txt")

    assert manager.touched == [worker_target.worker_key]
    assert manager.failures == []


@pytest.mark.parametrize(
    ("status_code", "payload"),
    [
        (401, {"detail": "Unauthorized sandbox runner request"}),
        (404, {"detail": "Not Found"}),
    ],
)
def test_worker_routed_ambiguous_http_client_error_records_worker_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    status_code: int,
    payload: dict[str, object],
) -> None:
    """Auth and ambiguous route failures should evict the worker instead of treating it as healthy."""
    execution_identity = ToolExecutionIdentity(
        channel="matrix",
        agent_name="code",
        requester_id="@alice:example.org",
        room_id="!room:example.org",
        thread_id="$thread",
        resolved_thread_id="$thread",
        session_id="session-1",
    )
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url="http://sandbox-runner:8765",
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    worker_target = _worker_target(runtime_paths, "shared", "code", execution_identity)
    assert worker_target.worker_key is not None
    manager = _TrackingWorkerManager()
    tool = get_tool_by_name(
        "file",
        runtime_paths,
        tool_init_overrides={"base_dir": str(tmp_path)},
        worker_tools_override=["file"],
        worker_target=worker_target,
    )
    entrypoint = tool.functions["read_file"].entrypoint
    assert entrypoint is not None

    with (
        patch.object(
            sandbox_proxy_module,
            "lease_primary_worker_manager",
            lambda *_args, **_kwargs: _static_worker_manager_lease(manager),
        ),
        patch(
            "mindroom.tool_system.sandbox_proxy.httpx.Client",
            _http_status_client_class(status_code=status_code, payload=payload),
        ),
        pytest.raises(httpx.HTTPStatusError),
    ):
        entrypoint("missing.txt")

    assert manager.touched == []
    assert len(manager.failures) == 1
    assert manager.failures[0][0] == worker_target.worker_key


def test_worker_tools_override_can_use_docker_backend_without_proxy_url(monkeypatch: pytest.MonkeyPatch) -> None:
    """Worker-routed tools should stay proxy-enabled when the Docker backend provides worker handles directly."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )

    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
            worker_tools_override=["shell"],
        )
        is True
    )
    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
            worker_tools_override=None,
        )
        is False
    )


def test_docker_backend_keeps_unscoped_env_routing_enabled_without_proxy_url(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Unscoped agents should still route through dedicated workers on the Docker backend."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.setenv("MINDROOM_DOCKER_WORKER_IMAGE", "ghcr.io/mindroom-ai/mindroom:latest")
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="selective",
        proxy_tools={"shell"},
    )

    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
        )
        is True
    )
    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "calculator",
            runtime_paths=runtime_paths,
        )
        is False
    )


def test_docker_backend_keeps_wrapping_when_required_config_is_missing(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Docker routing should stay enabled so misconfiguration fails closed at call time."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.delenv("MINDROOM_DOCKER_WORKER_IMAGE", raising=False)
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode=None,
    )

    assert (
        sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
            "shell",
            runtime_paths=runtime_paths,
            worker_tools_override=["shell"],
        )
        is True
    )


@pytest.mark.asyncio
async def test_docker_backend_misconfiguration_raises_instead_of_running_locally(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Misconfigured Docker worker routing should raise rather than executing in the primary runtime."""
    monkeypatch.setenv("MINDROOM_WORKER_BACKEND", "docker")
    monkeypatch.delenv("MINDROOM_DOCKER_WORKER_IMAGE", raising=False)
    runtime_paths = _configure_proxy_runtime(
        monkeypatch,
        proxy_url=None,
        proxy_token=_TEST_AUTH_TOKEN,
        execution_mode="off",
    )
    runtime_config = load_config(runtime_paths)

    tool = get_tool_by_name(
        "shell",
        runtime_paths,
        worker_tools_override=["shell"],
        worker_target=_worker_target(runtime_paths, None, "code", None),
    )
    entrypoint = tool.async_functions["run_shell_command"].entrypoint
    assert entrypoint is not None

    runtime_context = ToolRuntimeContext(
        agent_name="code",
        target=MessageTarget.resolve(
            room_id="!room:example.org",
            thread_id=None,
            reply_to_event_id=None,
        ),
        requester_id="@user:example.org",
        client=object(),
        config=runtime_config,
        runtime_paths=runtime_paths,
        event_cache=make_event_cache_mock(),
        conversation_cache=make_conversation_cache_mock(),
    )

    with (
        tool_runtime_context(runtime_context),
        pytest.raises(
            WorkerBackendError,
            match="MINDROOM_DOCKER_WORKER_IMAGE",
        ),
    ):
        await entrypoint("pwd")


class TestWorkerToolsOverride:
    """Tests for per-agent worker_tools_override parameter."""

    def test_override_none_defers_to_env_vars(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """None override should defer to the standard sandbox-proxy env controls."""
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            execution_mode="selective",
            proxy_tools={"shell"},
        )

        # None override -> falls through to sandbox-proxy env controls
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is True
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "calculator",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is False
        )

    def test_override_empty_list_disables_sandboxing(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Empty list override should disable sandboxing even when sandbox env controls enable it."""
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            execution_mode="all",
        )

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=[],
            )
            is False
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "file",
                runtime_paths=runtime_paths,
                worker_tools_override=[],
            )
            is False
        )

    def test_override_explicit_list_selects_tools(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Explicit list override should sandbox only the listed tools."""
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            execution_mode="off",
        )

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=["shell", "file"],
            )
            is True
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "file",
                runtime_paths=runtime_paths,
                worker_tools_override=["shell", "file"],
            )
            is True
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "calculator",
                runtime_paths=runtime_paths,
                worker_tools_override=["shell", "file"],
            )
            is False
        )

    def test_override_still_respects_runner_mode(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Runner mode should always disable proxying, even with override."""
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            execution_mode=None,
            runner_mode=True,
        )

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=["shell"],
            )
            is False
        )

    def test_override_without_proxy_url_fails_closed(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """No proxy URL should keep worker-routed execution tools wrapped so calls fail closed."""
        monkeypatch.delenv("MINDROOM_WORKER_BACKEND", raising=False)
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url=None,
            execution_mode=None,
        )

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=["shell"],
            )
            is True
        )

    def test_default_static_runner_without_proxy_keeps_execution_tools_local(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Plain local static-runner installs should keep execution tools usable."""
        monkeypatch.delenv("MINDROOM_WORKER_BACKEND", raising=False)
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url=None,
            execution_mode=None,
        )

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is False
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "calculator",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is False
        )

    def test_unset_mode_with_static_proxy_url_preserves_all_tool_proxying(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """A configured static proxy should keep the legacy unset-mode all-tools default."""
        monkeypatch.delenv("MINDROOM_WORKER_BACKEND", raising=False)
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox-runner:8765",
            execution_mode=None,
        )

        assert sandbox_proxy_module.sandbox_proxy_config(runtime_paths).proxy_tools is None
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is True
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "calculator",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is True
        )

    def test_explicit_wildcard_without_proxy_url_fails_closed_for_all_tools(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Explicit all-tools routing should not collapse into plain local execution."""
        monkeypatch.delenv("MINDROOM_WORKER_BACKEND", raising=False)
        _configure_proxy_runtime(
            monkeypatch,
            proxy_url=None,
            execution_mode=None,
        )
        monkeypatch.setenv("MINDROOM_SANDBOX_PROXY_TOOLS", "*")
        runtime_paths = _runtime_paths_from_env()

        assert sandbox_proxy_module.sandbox_proxy_config(runtime_paths).proxy_tools is None
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is True
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "calculator",
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is True
        )

    def test_unsafe_local_execution_opt_in_disables_explicit_routing_without_backend(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Explicit unsafe env opt-in should permit host execution after explicit routing."""
        monkeypatch.delenv("MINDROOM_WORKER_BACKEND", raising=False)
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url=None,
            execution_mode=None,
        )
        monkeypatch.setenv("MINDROOM_UNSAFE_ALLOW_LOCAL_EXECUTION_TOOLS", "true")
        runtime_paths = _runtime_paths_from_env()

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "shell",
                runtime_paths=runtime_paths,
                worker_tools_override=["shell"],
            )
            is False
        )

    @pytest.mark.asyncio
    async def test_default_static_runner_shell_call_runs_locally(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Plain local static-runner installs should keep shell usable without extra env."""
        monkeypatch.delenv("MINDROOM_WORKER_BACKEND", raising=False)
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url=None,
            execution_mode=None,
        )

        tool = get_tool_by_name("shell", runtime_paths, worker_target=None)
        entrypoint = tool.async_functions["run_shell_command"].entrypoint
        assert entrypoint is not None

        assert await entrypoint("printf local-ok") == "local-ok"

    @pytest.mark.parametrize(
        "tool_name",
        [
            "callback_manager",
            "desktop",
            "gmail",
            "google_calendar",
            "google_docs",
            "google_drive",
            "google_sheets",
            "homeassistant",
            "todo",
        ],
    )
    def test_local_only_tools_never_proxy(
        self,
        monkeypatch: pytest.MonkeyPatch,
        tool_name: str,
    ) -> None:
        """Tools marked local-only should stay in the primary runtime."""
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            execution_mode="all",
        )

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                tool_name,
                runtime_paths=runtime_paths,
                worker_tools_override=None,
            )
            is False
        )
        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                tool_name,
                runtime_paths=runtime_paths,
                worker_tools_override=[tool_name],
            )
            is False
        )

    def test_browser_can_still_use_explicit_worker_routing(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Host-browser isolation must not be disabled by the optional Matrix desktop target."""
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            execution_mode="all",
        )

        assert (
            sandbox_proxy_module.sandbox_proxy_enabled_for_tool(
                "browser",
                runtime_paths=runtime_paths,
                worker_tools_override=["browser"],
            )
            is True
        )

    def test_get_tool_by_name_keeps_homeassistant_local_even_when_listed(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Home Assistant should execute locally even if it appears in worker_tools."""

        class _ForbiddenClient:
            def __init__(self, *_args: object, **_kwargs: object) -> None:
                msg = "Sandbox proxy should not be used for local-only tools."
                raise AssertionError(msg)

        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            proxy_token=_TEST_AUTH_TOKEN,
            execution_mode="all",
            credential_policy={},
        )
        monkeypatch.setattr("mindroom.tool_system.sandbox_proxy.httpx.Client", _ForbiddenClient)

        fake_credentials = FakeCredentialsManager({})

        tool = get_tool_by_name(
            "homeassistant",
            runtime_paths,
            credentials_manager=fake_credentials,
            worker_tools_override=["homeassistant"],
            worker_target=_worker_target(runtime_paths, "shared", "general", None),
        )
        entrypoint = tool.async_functions["list_entities"].entrypoint
        assert entrypoint is not None

        result = asyncio.run(entrypoint())
        assert "Home Assistant is not configured" in result

    def test_get_tool_by_name_passes_override(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """get_tool_by_name should pass worker_tools_override through to the proxy wrapper."""
        captured: dict[str, Any] = {}

        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            proxy_token=_TEST_AUTH_TOKEN,
            execution_mode="off",
            credential_policy={},
        )
        monkeypatch.setattr(
            "mindroom.tool_system.sandbox_proxy.httpx.Client",
            _recording_client_class(captured=captured),
        )

        # Override says sandbox calculator
        tool = get_tool_by_name(
            "calculator",
            runtime_paths,
            worker_tools_override=["calculator"],
            worker_target=None,
        )
        entrypoint = tool.functions["add"].entrypoint
        assert entrypoint is not None
        result = entrypoint(1, 2)
        assert result == "sandbox-result"
        assert captured["url"] == "http://sandbox:8765/api/sandbox-runner/execute"

    def test_get_tool_by_name_passes_tool_init_overrides_to_proxy(self, monkeypatch: pytest.MonkeyPatch) -> None:
        """Proxy execution should preserve non-secret tool init overrides like base_dir."""
        captured: dict[str, Any] = {}

        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            proxy_token=_TEST_AUTH_TOKEN,
            execution_mode="off",
            credential_policy={},
        )
        monkeypatch.setattr(
            "mindroom.tool_system.sandbox_proxy.httpx.Client",
            _recording_client_class(captured=captured),
        )

        tool = get_tool_by_name(
            "coding",
            runtime_paths,
            tool_init_overrides={"base_dir": "/workspace/demo"},
            worker_tools_override=["coding"],
            worker_target=None,
        )
        entrypoint = tool.functions["ls"].entrypoint
        assert entrypoint is not None

        result = entrypoint(path=".")

        assert result == "sandbox-result"
        assert captured["url"] == "http://sandbox:8765/api/sandbox-runner/execute"
        assert captured["json"]["tool_init_overrides"] == {"base_dir": "/workspace/demo"}

    def test_proxy_rewrites_storage_root_base_dir_to_shared_relative_path(
        self,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Worker-keyed proxy execution should rewrite canonical agent base_dir paths portably."""
        captured: dict[str, Any] = {}

        monkeypatch.delenv("MINDROOM_STORAGE_PATH", raising=False)
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            proxy_token=_TEST_AUTH_TOKEN,
            execution_mode="off",
            credential_policy={},
        )
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id="$thread",
            resolved_thread_id="$thread",
            session_id="session-1",
        )
        monkeypatch.setattr(
            "mindroom.tool_system.sandbox_proxy.httpx.Client",
            _recording_client_class(captured=captured),
        )
        tool = get_tool_by_name(
            "coding",
            runtime_paths,
            tool_init_overrides={"base_dir": "/srv/mindroom/agents/general/workspace"},
            shared_storage_root_path=Path("/srv/mindroom"),
            worker_tools_override=["coding"],
            worker_target=_worker_target(runtime_paths, "shared", "general", execution_identity),
        )
        entrypoint = tool.functions["ls"].entrypoint
        assert entrypoint is not None
        result = entrypoint(path=".")

        assert result == "sandbox-result"
        assert captured["url"].endswith("/api/sandbox-runner/execute")
        assert captured["json"]["tool_init_overrides"] == {"base_dir": "agents/general/workspace"}

    def test_proxy_preserves_storage_root_absolute_base_dir_without_worker_key(
        self,
        monkeypatch: pytest.MonkeyPatch,
        tmp_path: Path,
    ) -> None:
        """Unscoped proxied calls must keep absolute canonical paths unchanged."""
        captured: dict[str, Any] = {}
        storage_root = tmp_path / "mindroom_data"
        base_dir = storage_root / "agents" / "general" / "workspace" / "mind_data"

        monkeypatch.setenv("MINDROOM_STORAGE_PATH", str(storage_root))
        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            proxy_token=_TEST_AUTH_TOKEN,
            execution_mode="off",
            credential_policy={},
        )
        monkeypatch.setattr(
            "mindroom.tool_system.sandbox_proxy.httpx.Client",
            _recording_client_class(captured=captured),
        )

        tool = get_tool_by_name(
            "coding",
            runtime_paths,
            tool_init_overrides={"base_dir": str(base_dir)},
            worker_tools_override=["coding"],
            worker_target=None,
        )
        entrypoint = tool.functions["ls"].entrypoint
        assert entrypoint is not None

        result = entrypoint(path=".")

        assert result == "sandbox-result"
        assert captured["url"] == "http://sandbox:8765/api/sandbox-runner/execute"
        assert captured["json"]["tool_init_overrides"] == {
            "base_dir": str(base_dir),
        }

    def test_proxy_preserves_unrelated_absolute_base_dir_for_worker_key(
        self,
        tmp_path: Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Worker-keyed requests must not retarget unrelated absolute paths that merely contain `agents`."""
        captured: dict[str, Any] = {}
        unrelated_base_dir = tmp_path / "demo" / "agents" / "general" / "workspace"

        runtime_paths = _configure_proxy_runtime(
            monkeypatch,
            proxy_url="http://sandbox:8765",
            proxy_token=_TEST_AUTH_TOKEN,
            execution_mode="off",
            credential_policy={},
        )
        execution_identity = ToolExecutionIdentity(
            channel="matrix",
            agent_name="general",
            requester_id="@alice:example.org",
            room_id="!room:example.org",
            thread_id="$thread",
            resolved_thread_id="$thread",
            session_id="session-1",
        )
        monkeypatch.setattr(
            "mindroom.tool_system.sandbox_proxy.httpx.Client",
            _recording_client_class(captured=captured),
        )

        tool = get_tool_by_name(
            "coding",
            runtime_paths,
            tool_init_overrides={"base_dir": str(unrelated_base_dir)},
            worker_tools_override=["coding"],
            worker_target=_worker_target(runtime_paths, "shared", "general", execution_identity),
        )
        entrypoint = tool.functions["ls"].entrypoint
        assert entrypoint is not None
        result = entrypoint(path=".")

        assert result == "sandbox-result"
        assert captured["json"]["tool_init_overrides"] == {
            "base_dir": str(unrelated_base_dir),
        }


@pytest.mark.asyncio
async def test_inprocess_runner_blocks_cross_runtime_secret_leakage(
    tmp_path: Path,
) -> None:
    """Runner-only env vars must not leak via glob passthrough patterns."""
    config_path = tmp_path / "config.yaml"
    config_path.write_text(
        "models:\n  default:\n    provider: openai\n    id: gpt-5.4\nagents: {}\nrouter:\n  model: default\n",
        encoding="utf-8",
    )
    config_path.with_name(".env").write_text("", encoding="utf-8")

    # Runner has a secret env var that matches the passthrough pattern.
    runner_process_env = {"WHISPER_API_TOKEN": "runner-only-secret"}
    runtime_paths = resolve_runtime_paths(
        config_path=config_path,
        storage_path=config_path.parent / "storage",
        process_env=runner_process_env,
    )
    config = sandbox_runner_module._runtime_config_or_empty(runtime_paths)

    # Client sends a request with WHISPER_* passthrough but only WHISPER_URL
    # in its execution_env.  The runner-only WHISPER_API_TOKEN must NOT leak.
    request = sandbox_runner_module.SandboxRunnerExecuteRequest(
        tool_name="shell",
        function_name="run_shell_command",
        args=[
            ["bash", "-lc", "printf '%s' \"$WHISPER_URL|$WHISPER_API_TOKEN\""],
        ],
        execution_env={"WHISPER_URL": "https://whisper.example"},
        extra_env_passthrough="WHISPER_*",
    )

    response = await sandbox_runner_module._execute_request_inprocess(
        request,
        runtime_paths,
        config,
    )

    assert response.ok is True
    # WHISPER_URL from execution_env should be visible; WHISPER_API_TOKEN should not.
    assert response.result == "https://whisper.example|"


def test_shell_extra_env_trusts_explicit_patterns_except_runner_control() -> None:
    """Wildcard passthrough keeps matched user env and drops only runner control names."""
    env = {
        "ANTHROPIC_API_KEY": "sk-ant-secret",
        "OPENAI_API_KEY": "sk-oai-secret",
        "GOOGLE_API_KEY": "goog-secret",
        "MY_SERVICE_PASSWORD": "pass-secret",
        "WEBHOOK_SECRET": "hook-secret",
        "CI_JOB_TOKEN": "ci-token",
        "GITEA_TOKEN": "gitea-token",
        "WHISPER_URL": "https://whisper.example",
        "GITEA_HOST": "gitea.local",
        "MINDROOM_API_KEY": "runner-api-key",
        "MINDROOM_LOCAL_CLIENT_SECRET": "runner-client-secret",
        "MINDROOM_SANDBOX_PROXY_TOKEN": "runner-proxy-token",
        "MINDROOM_SANDBOX_STARTUP_MANIFEST_PATH": "/var/lib/mindroom/manifest.json",
        "MINDROOM_SANDBOX_FOO": "runner-sandbox-value",
        "MINDROOM_LOCAL_CLIENT_ID": "local-client-id",
    }
    result = dict(shell_extra_env_values(extra_env_passthrough="*", process_env=env))

    assert result["WHISPER_URL"] == "https://whisper.example"
    assert result["GITEA_HOST"] == "gitea.local"
    assert result["GITEA_TOKEN"] == "gitea-token"  # noqa: S105
    assert result["ANTHROPIC_API_KEY"] == "sk-ant-secret"
    assert result["OPENAI_API_KEY"] == "sk-oai-secret"
    assert result["GOOGLE_API_KEY"] == "goog-secret"
    assert result["MY_SERVICE_PASSWORD"] == "pass-secret"  # noqa: S105
    assert result["WEBHOOK_SECRET"] == "hook-secret"  # noqa: S105
    assert result["CI_JOB_TOKEN"] == "ci-token"  # noqa: S105
    assert result["MINDROOM_LOCAL_CLIENT_ID"] == "local-client-id"

    assert "MINDROOM_API_KEY" not in result
    assert "MINDROOM_LOCAL_CLIENT_SECRET" not in result
    assert "MINDROOM_SANDBOX_PROXY_TOKEN" not in result
    assert "MINDROOM_SANDBOX_STARTUP_MANIFEST_PATH" not in result
    assert "MINDROOM_SANDBOX_FOO" not in result


def test_shell_extra_env_requires_explicit_patterns_for_service_urls() -> None:
    """extra_env_passthrough should not auto-include unrelated service URLs."""
    env = {
        "GITEA_TOKEN": "gitea-token",
        "WHISPER_URL": "https://whisper.example",
    }

    result = dict(shell_extra_env_values(extra_env_passthrough="GITEA_*", process_env=env))

    assert result == {"GITEA_TOKEN": "gitea-token"}
    assert "WHISPER_URL" not in result
