server/tests/test_runtime_service.py

Rama main · réplica de lectura del repositorio original.

import asyncio
import json
import logging
from types import SimpleNamespace
from typing import Any

import pytest

from bumblehive import BumblehiveConfig, SkillsManager
from bumblehive.config import ProviderConfig
from bumblehive.observability.events import make_event
from bumblehive.observability.emitter import EventEmitter
from bumblehive.protocols import ToolCall, ToolResult
from bumblehive.protocols.errors import AgentError
from bumblehive.tools import ToolManager
from bumblehive.tools.scope import (
    bind_tool_session,
    reset_tool_session,
)
from bumblehive_server.runtime_service import RuntimeBusyError, RuntimeService
from bumblehive_server.subagents import SUBAGENT_INSTRUCTIONS, observe_subagents
from bumblehive_server.tool_context import capture_tool_call, current_task_id


@pytest.mark.asyncio
async def test_repeated_desktop_tool_calls_use_the_current_call_id() -> None:
    tools = ToolManager()

    @tools.tool(name="pending", description="Return this call's pending result.")
    async def pending() -> dict:
        await asyncio.sleep(0)
        return {"status": "pending", "task_id": current_task_id()}

    emitter = EventEmitter.from_hooks(capture_tool_call)
    for call_id in ("first-call", "second-call"):
        result = await tools.execute_call(ToolCall(call_id, "pending", {}), emitter=emitter)
        assert result.error is None
        assert result.content == {"status": "pending", "task_id": call_id}
        with pytest.raises(RuntimeError, match="active tool call"):
            current_task_id()


class FakeRuntime:
    def __init__(
        self,
        config: BumblehiveConfig,
        *,
        initialize_error: Exception | None = None,
    ) -> None:
        self.config = config
        self.tools = ToolManager()
        self.initialize_error = initialize_error
        self.initialize_calls = 0
        self.initialized = False
        self.closed = False
        self.deleted_sessions: list[str] = []

    async def initialize_tools(self) -> None:
        self.initialize_calls += 1
        if self.initialize_error is not None:
            raise self.initialize_error
        self.initialized = True

    async def close(self) -> None:
        self.closed = True

    async def delete_session(self, session_id: str) -> bool:
        self.deleted_sessions.append(session_id)
        return True


class FakeModels:
    def __init__(self, response: Any = None, error: Exception | None = None) -> None:
        self.response = response
        self.error = error
        self.calls = 0

    async def list(self) -> Any:
        self.calls += 1
        if self.error is not None:
            raise self.error
        return self.response


class FakeOpenAIClient:
    def __init__(self, models: FakeModels) -> None:
        self.models = models
        self.closed = False

    async def close(self) -> None:
        self.closed = True


class FakeOpenAIClientFactory:
    def __init__(self, models: FakeModels) -> None:
        self.models = models
        self.calls: list[dict[str, str]] = []
        self.clients: list[FakeOpenAIClient] = []

    def __call__(self, *, api_key: str, base_url: str) -> FakeOpenAIClient:
        self.calls.append({"api_key": api_key, "base_url": base_url})
        client = FakeOpenAIClient(self.models)
        self.clients.append(client)
        return client


class RecordingSessionReader:
    def __init__(self, session_id: str = "child-session") -> None:
        self.session_id = session_id
        self.created_workspaces: list[Any] = []
        self.created_titles: list[str | None] = []
        self.created_parent_session_ids: list[str | None] = []

    async def create_child(
        self,
        workspace: Any,
        *,
        title: str,
        parent_session_id: str,
    ) -> str:
        self.created_workspaces.append(workspace)
        self.created_titles.append(title)
        self.created_parent_session_ids.append(parent_session_id)
        return self.session_id


class SequentialSessionReader(RecordingSessionReader):
    async def create_child(
        self,
        workspace: Any,
        *,
        title: str,
        parent_session_id: str,
    ) -> str:
        self.created_workspaces.append(workspace)
        self.created_titles.append(title)
        self.created_parent_session_ids.append(parent_session_id)
        return f"child-session-{len(self.created_workspaces)}"


class RecordingSubagentObserver:
    def __init__(self) -> None:
        self.created: list[dict[str, str]] = []
        self.events: list[Any] = []
        self.results: list[dict[str, Any]] = []
        self.errors: list[dict[str, str]] = []
        self.cancelled: list[str] = []

    async def on_created(
        self,
        *,
        session_id: str,
        workspace: str,
        title: str,
        content: str,
    ) -> None:
        self.created.append(
            {
                "session_id": session_id,
                "workspace": workspace,
                "title": title,
                "content": content,
            }
        )

    async def on_event(self, event: Any) -> None:
        self.events.append(event)

    async def on_result(
        self,
        *,
        session_id: str,
        result: Any,
        duration_s: float,
    ) -> None:
        self.results.append(
            {
                "session_id": session_id,
                "result": result,
                "duration_s": duration_s,
            }
        )

    async def on_error(self, *, session_id: str, message: str) -> None:
        self.errors.append({"session_id": session_id, "message": message})

    async def on_cancelled(self, *, session_id: str) -> None:
        self.cancelled.append(session_id)


class SubagentFakeStream:
    def __init__(
        self,
        runtime: "SubagentFakeRuntime",
        session_id: str,
    ) -> None:
        self.runtime = runtime
        self.session_id = session_id

    def __aiter__(self) -> Any:
        return self._iterate()

    async def _iterate(self) -> Any:
        yield make_event(
            "model.stream.content_delta",
            run_id=f"run-{self.session_id}",
            session_id=self.session_id,
            delta=f"Streaming {self.session_id}",
        )
        await self.runtime.run_release.wait()
        if self.runtime.run_exception is not None:
            raise self.runtime.run_exception

    async def result(self) -> Any:
        return self.runtime.run_result

    async def aclose(self) -> None:
        return None


class SubagentFakeRuntime(FakeRuntime):
    def __init__(self, config: BumblehiveConfig) -> None:
        super().__init__(config)
        self.tools = ToolManager()
        self.run_started = asyncio.Event()
        self.run_release = asyncio.Event()
        self.expected_run_count = 1
        self.run_calls: list[dict[str, Any]] = []
        self.run_result = SimpleNamespace(final_content="Child answer", error=None)
        self.run_exception: Exception | None = None

    def stream(
        self,
        message: str,
        *,
        session_id: str,
        config: dict[str, Any],
    ) -> SubagentFakeStream:
        self.run_calls.append(
            {
                "message": message,
                "session_id": session_id,
                "config": config,
            }
        )
        if len(self.run_calls) >= self.expected_run_count:
            self.run_started.set()
        return SubagentFakeStream(self, session_id)


def _start_subagent_tool(
    runtime: SubagentFakeRuntime,
    *,
    workspace: Any,
    title: str,
    task: str,
) -> asyncio.Task[ToolResult]:
    session_token = bind_tool_session("parent-session")
    try:
        return asyncio.create_task(runtime.tools.execute_call(
            ToolCall("delegate", "sub_agent", {"title": title, "task": task}),
            workspace=workspace,
            emitter=EventEmitter.from_hooks(capture_tool_call),
        ))
    finally:
        reset_tool_session(session_token)


async def _finish_subagents(service: RuntimeService) -> None:
    async with asyncio.timeout(2):
        await asyncio.gather(*(entry.task for entry in service.subagents._tasks.values()))


@pytest.mark.asyncio
async def test_runtime_service_starts_without_a_config_file(
    tmp_path,
    caplog,
) -> None:
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(tmp_path / "missing.json", runtime_factory=factory)

    with caplog.at_level(logging.INFO, logger="uvicorn.error.bumblehive"):
        await service.startup()
        deleted = await service.delete_session("session-1")

    assert service.ready is True
    assert runtimes[0].initialized is True
    assert runtimes[0].initialize_calls == 1
    assert deleted is True
    assert service.config.provider.model is None
    assert service.public_config()["provider"] == {
        "type": "openai_chat_completions",
        "api_key_configured": False,
    }
    assert service.public_config()["generation"] == {
        "max_completion_tokens": 16_384,
    }
    assert service.public_config()["runtime"] == {
        "workspace": str(service.workspace),
        "context_window_tokens": 200_000,
        "max_tool_result_chars": 20_000,
        "max_iterations": 300,
    }
    assert service.public_config()["agent"] == {}
    assert service.public_config()["mcp_servers"] == []
    assert "[config] loaded | source=defaults | duration=" in caplog.text
    assert (
        "[session] delete completed | session_id=session-1 | deleted=true"
        in caplog.text
    )

    await service.shutdown()
    assert runtimes[0].closed is True


@pytest.mark.asyncio
async def test_runtime_service_updates_config_without_exposing_api_key(
    tmp_path,
    caplog,
) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig(
        provider=ProviderConfig(model="old-model", api_key="secret")
    ).to_json_file(config_path)
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(config_path, runtime_factory=factory)
    with caplog.at_level(logging.INFO, logger="uvicorn.error.bumblehive"):
        await service.startup()

        public = service.public_config()
        assert "api_key" not in public["provider"]
        assert public["provider"]["api_key_configured"] is True

        await service.update_config({"provider": {"model": "new-model"}})
    saved = json.loads(config_path.read_text(encoding="utf-8"))
    assert saved == {
        "provider": {
            "type": "openai_chat_completions",
            "model": "new-model",
            "api_key": "secret",
            "base_url": None,
        },
        "generation": {
            "max_completion_tokens": 16_384,
            "temperature": None,
            "reasoning_effort": None,
            "extra_body": None,
        },
        "agent": {
            "instructions": None,
            "dynamic_context": {},
            "skill_names": None,
            "tool_names": None,
        },
        "runtime": {
            "workspace": None,
            "timezone": None,
            "context_window_tokens": 200_000,
            "max_tool_result_chars": 20_000,
            "max_iterations": 300,
        },
        "mcp_servers": [],
        "skills_dir": None,
    }
    assert runtimes[0].closed is True
    assert runtimes[1].initialized is True
    assert runtimes[1].initialize_calls == 1

    with caplog.at_level(logging.INFO, logger="uvicorn.error.bumblehive"):
        async with service.lease():
            with pytest.raises(RuntimeBusyError):
                await service.update_config({"provider": {"model": "busy"}})

    assert "[config] loaded | source=file | duration=" in caplog.text
    assert "[config] update completed | duration=" in caplog.text
    assert "[config] update rejected | reason=active_run" in caplog.text
    assert "secret" not in caplog.text

    await service.shutdown()
    assert runtimes[-1].closed is True


@pytest.mark.asyncio
async def test_runtime_service_can_clear_dynamic_context(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig.from_mapping(
        {
            "provider": {"model": "test-model"},
            "agent": {
                "instructions": "Keep this instruction.",
                "dynamic_context": {"project": "bumblehive"},
            },
        }
    ).to_json_file(config_path)
    service = RuntimeService(config_path, runtime_factory=FakeRuntime)
    await service.startup()

    await service.update_config({"agent": {"dynamic_context": {}}})

    assert service.config.agent.instructions == "Keep this instruction."
    assert service.config.agent.dynamic_context == {}
    saved = json.loads(config_path.read_text(encoding="utf-8"))
    assert saved["agent"]["dynamic_context"] == {}

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_masks_and_preserves_mcp_header_secrets(
    tmp_path,
) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig.from_mapping(
        {
            "provider": {"model": "test-model"},
            "mcp_servers": [
                {
                    "name": "private-server",
                    "url": "https://mcp.example.test",
                    "headers": {
                        "Authorization": "Bearer secret-token",
                        "X-Tenant": "secret-tenant",
                    },
                }
            ],
        }
    ).to_json_file(config_path)
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(config_path, runtime_factory=factory)
    await service.startup()

    public = service.public_config()
    assert public["mcp_servers"] == [
        {
            "name": "private-server",
            "url": "https://mcp.example.test",
            "headers": {
                "Authorization": "",
                "X-Tenant": "",
            },
        }
    ]

    await service.update_config(
        {
            "mcp_servers": [
                {
                    "name": "renamed-server",
                    "url": "https://mcp.example.test/v2",
                    "headers": {
                        "Authorization": "",
                        "X-Tenant": "",
                        "X-New": "new-secret",
                    },
                }
            ]
        }
    )

    saved = json.loads(config_path.read_text(encoding="utf-8"))
    saved_server = saved["mcp_servers"][0]
    assert saved_server["name"] == "renamed-server"
    assert saved_server["url"] == "https://mcp.example.test/v2"
    assert saved_server["headers"] == {
        "Authorization": "Bearer secret-token",
        "X-Tenant": "secret-tenant",
        "X-New": "new-secret",
    }
    assert service.public_config()["mcp_servers"][0]["headers"] == {
        "Authorization": "",
        "X-Tenant": "",
        "X-New": "",
    }

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_tests_mcp_in_an_isolated_manager_and_restores_secrets(
    tmp_path,
) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig.from_mapping(
        {
            "mcp_servers": [
                {
                    "name": "private-server",
                    "url": "https://old.example.test/mcp",
                    "headers": {
                        "Authorization": "Bearer saved-secret",
                    },
                }
            ],
        }
    ).to_json_file(config_path)
    runtimes: list[FakeRuntime] = []
    managers: list[Any] = []

    def runtime_factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    class IsolatedManager:
        def __init__(self, *, mcp_servers: list[Any]) -> None:
            self.servers = mcp_servers
            self.closed = False
            managers.append(self)

        async def connect_mcp_server(self, server_name: str) -> list[str]:
            assert server_name == "renamed-server"
            return ["mcp_renamed-server_search"]

        async def close(self) -> None:
            self.closed = True

    service = RuntimeService(
        config_path,
        runtime_factory=runtime_factory,
        tool_manager_factory=IsolatedManager,  # type: ignore[arg-type]
    )
    await service.startup()

    registered = await service.test_mcp_server(
        {
            "name": "renamed-server",
            "url": "https://new.example.test/mcp",
            "headers": {"Authorization": ""},
        },
        original_name="private-server",
    )

    assert registered == ["mcp_renamed-server_search"]
    assert len(managers) == 1
    assert managers[0].servers[0].headers == {
        "Authorization": "Bearer saved-secret"
    }
    assert managers[0].servers[0].url == "https://new.example.test/mcp"
    assert managers[0].servers[0].tool_timeout is None
    assert managers[0].servers[0].enabled_tools == ["*"]
    assert managers[0].closed is True
    assert service.config.mcp_servers[0].name == "private-server"

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_reloads_one_or_all_mcp_servers(tmp_path) -> None:
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(tmp_path / "config.json", runtime_factory=factory)
    await service.startup()
    calls: list[str] = []

    async def reload_one(server_name: str) -> list[str]:
        calls.append(server_name)
        return [f"mcp_{server_name}_search"]

    async def reload_all() -> list[str]:
        calls.append("*")
        return ["mcp_docs_search", "mcp_github_search"]

    runtimes[0].tools = SimpleNamespace(
        reload_mcp_server=reload_one,
        reload_mcp=reload_all,
    )

    assert await service.reload_mcp_servers("docs") == ["mcp_docs_search"]
    assert await service.reload_mcp_servers() == [
        "mcp_docs_search",
        "mcp_github_search",
    ]
    assert calls == ["docs", "*"]

    async with service.lease():
        with pytest.raises(RuntimeBusyError):
            await service.reload_mcp_servers("docs")

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_exposes_settings_choices(tmp_path) -> None:
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(tmp_path / "config.json", runtime_factory=factory)
    await service.startup()
    runtime = runtimes[0]
    runtime.skills = SimpleNamespace(
        list_skills=lambda: SimpleNamespace(
            skills=[
                SimpleNamespace(
                    name="review",
                    description="Review project changes",
                )
            ],
            errors=[
                SimpleNamespace(
                    path=tmp_path / "broken" / "SKILL.md",
                    message="invalid frontmatter",
                )
            ],
        )
    )
    runtime.tools = SimpleNamespace(
        list_tools=lambda: [
            SimpleNamespace(
                name="read_file",
                description="Read a file",
                source="local",
                parallel_safe=True,
            )
        ],
        list_mcp_server_statuses=lambda: [
            SimpleNamespace(
                name="github",
                connected=True,
                registered_tools=["mcp_github_search"],
            )
        ],
    )

    assert service.settings_options() == {
        "skills": [
            {
                "name": "review",
                "description": "Review project changes",
            }
        ],
        "skill_errors": [
            {
                "path": str(tmp_path / "broken" / "SKILL.md"),
                "message": "invalid frontmatter",
            }
        ],
        "tools": [
            {
                "name": "read_file",
                "description": "Read a file",
                "source": "local",
                "parallel_safe": True,
            }
        ],
        "mcp_statuses": [
            {
                "name": "github",
                "connected": True,
                "registered_tools": ["mcp_github_search"],
            }
        ],
    }

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_manages_skills_only_while_idle(tmp_path) -> None:
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(tmp_path / "config.json", runtime_factory=factory)
    await service.startup()
    runtime = runtimes[0]
    runtime.skills = SkillsManager(tmp_path / "skills")
    source = tmp_path / "review"
    source.mkdir()
    (source / "SKILL.md").write_text(
        "---\nname: review\ndescription: Review changes.\n---\n",
        encoding="utf-8",
    )

    installed = await service.install_skills([source])
    refreshed = await service.reload_skills()

    assert [skill.name for skill in installed.skills] == ["review"]
    assert [skill.name for skill in refreshed.skills] == ["review"]

    async with service.lease():
        with pytest.raises(RuntimeBusyError):
            await service.remove_skill("review")

    removed = await service.remove_skill("review")
    assert removed.skills == []

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_startup_closes_failed_runtime(tmp_path) -> None:
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(
            config,
            initialize_error=RuntimeError("MCP unavailable"),
        )
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(
        tmp_path / "missing.json",
        runtime_factory=factory,
    )

    with pytest.raises(RuntimeError, match="MCP unavailable"):
        await service.startup()

    assert service.ready is False
    assert runtimes[0].initialize_calls == 1
    assert runtimes[0].closed is True


@pytest.mark.asyncio
async def test_runtime_service_keeps_current_when_replacement_init_fails(
    tmp_path,
) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig(provider=ProviderConfig(model="old-model")).to_json_file(
        config_path
    )
    runtimes: list[FakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = FakeRuntime(
            config,
            initialize_error=(RuntimeError("MCP unavailable") if runtimes else None),
        )
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(config_path, runtime_factory=factory)
    await service.startup()

    with pytest.raises(RuntimeError, match="MCP unavailable"):
        await service.update_config({"provider": {"model": "new-model"}})

    assert service.config.provider.model == "old-model"
    assert (
        json.loads(config_path.read_text(encoding="utf-8"))["provider"]["model"]
        == "old-model"
    )
    assert runtimes[0].closed is False
    assert runtimes[1].initialize_calls == 1
    assert runtimes[1].closed is True

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_lists_models_with_new_url_and_key_without_saving(
    tmp_path,
) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig(
        provider=ProviderConfig(
            model="saved-model",
            api_key="saved-secret",
            base_url="https://saved.example/v1",
        )
    ).to_json_file(config_path)
    models = FakeModels(
        SimpleNamespace(
            data=[
                SimpleNamespace(id=" model-a "),
                SimpleNamespace(id="model-b"),
                SimpleNamespace(id="model-a"),
                SimpleNamespace(id="  "),
                SimpleNamespace(id=None),
            ]
        )
    )
    client_factory = FakeOpenAIClientFactory(models)
    service = RuntimeService(
        config_path,
        runtime_factory=FakeRuntime,
        openai_client_factory=client_factory,  # type: ignore[arg-type]
    )
    await service.startup()

    models = await service.list_models(
        api_key=" temporary-secret ",
        base_url=" https://DRAFT.example/v1/ ",
    )

    assert models == ["model-a", "model-b"]
    assert client_factory.calls == [
        {
            "api_key": "temporary-secret",
            "base_url": "https://draft.example/v1",
        }
    ]
    assert client_factory.clients[0].closed is True
    assert service.config.provider.api_key == "saved-secret"
    saved = json.loads(config_path.read_text(encoding="utf-8"))
    assert saved["provider"]["api_key"] == "saved-secret"
    assert saved["provider"]["base_url"] == "https://saved.example/v1"

    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_reuses_saved_key_for_the_same_url(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig(
        provider=ProviderConfig(
            api_key="saved-secret",
            base_url="https://saved.example/v1/",
        )
    ).to_json_file(config_path)
    client_factory = FakeOpenAIClientFactory(
        FakeModels(SimpleNamespace(data=[SimpleNamespace(id="saved-model")]))
    )
    service = RuntimeService(
        config_path,
        runtime_factory=FakeRuntime,
        openai_client_factory=client_factory,  # type: ignore[arg-type]
    )
    await service.startup()

    models = await service.list_models(base_url="https://SAVED.example/v1")

    assert models == ["saved-model"]
    assert client_factory.calls == [
        {
            "api_key": "saved-secret",
            "base_url": "https://saved.example/v1",
        }
    ]
    assert client_factory.clients[0].closed is True
    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_rejects_new_url_without_a_new_key(tmp_path) -> None:
    config_path = tmp_path / "config.json"
    BumblehiveConfig(
        provider=ProviderConfig(
            api_key="saved-secret",
            base_url="https://saved.example/v1",
        )
    ).to_json_file(config_path)
    client_factory = FakeOpenAIClientFactory(FakeModels())
    service = RuntimeService(
        config_path,
        runtime_factory=FakeRuntime,
        openai_client_factory=client_factory,  # type: ignore[arg-type]
    )
    await service.startup()

    with pytest.raises(ValueError, match="different Base URL"):
        await service.list_models(base_url="https://new.example/v1")

    assert client_factory.calls == []
    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_wraps_model_provider_failures(tmp_path) -> None:
    from bumblehive_server.runtime_service import ModelListError

    client_factory = FakeOpenAIClientFactory(
        FakeModels(error=RuntimeError("provider unavailable"))
    )
    service = RuntimeService(
        tmp_path / "missing.json",
        runtime_factory=FakeRuntime,
        openai_client_factory=client_factory,  # type: ignore[arg-type]
    )
    await service.startup()

    with pytest.raises(ModelListError, match="third-party model query failed"):
        await service.list_models(
            base_url="https://new.example/v1",
            api_key="temporary-secret",
        )

    assert client_factory.clients[0].closed is True
    await service.shutdown()


@pytest.mark.asyncio
async def test_runtime_service_logs_config_load_failure(
    tmp_path,
    caplog,
) -> None:
    config_path = tmp_path / "config.json"
    config_path.write_text("{", encoding="utf-8")
    service = RuntimeService(config_path)

    with caplog.at_level(logging.INFO, logger="uvicorn.error.bumblehive"):
        with pytest.raises(json.JSONDecodeError):
            await service.startup()

    failure_records = [
        record
        for record in caplog.records
        if record.getMessage().startswith("[config] load failed")
    ]
    assert len(failure_records) == 1
    assert failure_records[0].exc_info is not None


@pytest.mark.asyncio
async def test_runtime_service_registers_subagent_on_startup_and_replacement(
    tmp_path,
) -> None:
    reader = RecordingSessionReader()
    runtimes: list[SubagentFakeRuntime] = []

    def factory(config: BumblehiveConfig) -> Any:
        runtime = SubagentFakeRuntime(config)
        runtimes.append(runtime)
        return runtime

    service = RuntimeService(
        tmp_path / "config.json",
        session_reader=reader,  # type: ignore[arg-type]
        runtime_factory=factory,
    )

    await service.startup()
    await service.update_config({"provider": {"model": "replacement"}})

    assert len(runtimes) == 2
    assert all(runtime.tools.get_tool("sub_agent") is not None for runtime in runtimes)
    tool = runtimes[0].tools.get_tool("sub_agent")
    assert tool is not None
    assert tool.parameters["required"] == ["title", "task"]
    assert tool.parameters["properties"]["title"]["maxLength"] == 80
    assert set(tool.parameters["properties"]) == {"title", "task"}
    assert "background=false" not in tool.description
    assert "independent read-only sub-agent" in tool.description
    await service.shutdown()


@pytest.mark.asyncio
async def test_subagent_tool_runs_a_read_only_child_session(tmp_path) -> None:
    reader = RecordingSessionReader()
    observer = RecordingSubagentObserver()
    runtime: SubagentFakeRuntime | None = None

    def factory(config: BumblehiveConfig) -> Any:
        nonlocal runtime
        runtime = SubagentFakeRuntime(config)
        return runtime

    service = RuntimeService(
        tmp_path / "config.json",
        session_reader=reader,  # type: ignore[arg-type]
        runtime_factory=factory,
    )
    await service.startup()
    assert runtime is not None

    workspace = tmp_path / "workspace"
    with observe_subagents(observer):
        tool_task = _start_subagent_tool(
            runtime,
            workspace=workspace,
            title="Inspect the project",
            task="  Inspect the project  ",
        )
    await runtime.run_started.wait()

    assert reader.created_workspaces == [workspace.resolve()]
    assert reader.created_titles == ["Inspect the project"]
    assert reader.created_parent_session_ids == ["parent-session"]
    assert observer.created == [
        {
            "session_id": "child-session",
            "workspace": str(workspace.resolve()),
            "title": "Inspect the project",
            "content": "Inspect the project",
        }
    ]
    assert [event.session_id for event in observer.events] == ["child-session"]
    assert runtime.run_calls == [
        {
            "message": [{
                "role": "user",
                "content": SUBAGENT_INSTRUCTIONS + "Inspect the project",
                "_bumblehive_ui": {"content": "Inspect the project"},
            }],
            "session_id": "child-session",
            "config": {
                "runtime": {"workspace": str(workspace.resolve())},
                "agent": {
                    "tool_names": (
                        "read_file",
                        "list_dir",
                        "find_files",
                        "grep",
                    )
                },
            },
        }
    ]

    result = await asyncio.wait_for(tool_task, 1)
    assert result.error is None
    assert result.content == {"status": "pending", "task_id": "delegate"}
    assert not runtime.run_release.is_set()
    runtime.run_release.set()
    await _finish_subagents(service)
    assert [item["session_id"] for item in observer.results] == [
        "child-session"
    ]
    await service.shutdown()


@pytest.mark.asyncio
async def test_subagent_tool_runs_multiple_calls_in_parallel(tmp_path) -> None:
    reader = SequentialSessionReader()
    observer = RecordingSubagentObserver()
    runtime: SubagentFakeRuntime | None = None

    def factory(config: BumblehiveConfig) -> Any:
        nonlocal runtime
        runtime = SubagentFakeRuntime(config)
        return runtime

    service = RuntimeService(
        tmp_path / "config.json",
        session_reader=reader,  # type: ignore[arg-type]
        runtime_factory=factory,
    )
    await service.startup()
    assert runtime is not None
    runtime.expected_run_count = 2

    session_token = bind_tool_session("parent-session")
    try:
        with observe_subagents(observer):
            calls_task = asyncio.create_task(
                runtime.tools.execute_many(
                    [
                        ToolCall(
                            id="call-1",
                            name="sub_agent",
                            arguments={
                                "title": "Inspect A",
                                "task": "Inspect task A",
                            },
                        ),
                        ToolCall(
                            id="call-2",
                            name="sub_agent",
                            arguments={
                                "title": "Inspect B",
                                "task": "Inspect task B",
                            },
                        ),
                    ],
                    tool_names=["sub_agent"],
                    workspace=tmp_path / "workspace",
                    emitter=EventEmitter.from_hooks(capture_tool_call),
                )
            )
    finally:
        reset_tool_session(session_token)

    await asyncio.wait_for(runtime.run_started.wait(), timeout=1)

    assert [call["message"][0]["content"] for call in runtime.run_calls] == [
        SUBAGENT_INSTRUCTIONS + "Inspect task A",
        SUBAGENT_INSTRUCTIONS + "Inspect task B",
    ]
    assert reader.created_titles == ["Inspect A", "Inspect B"]
    assert {
        call["session_id"] for call in runtime.run_calls
    } == {"child-session-1", "child-session-2"}
    assert {
        event.session_id for event in observer.events
    } == {"child-session-1", "child-session-2"}

    results = await asyncio.wait_for(calls_task, 1)
    assert [result.content for result in results] == [
        {"status": "pending", "task_id": "call-1"},
        {"status": "pending", "task_id": "call-2"},
    ]
    runtime.run_release.set()
    await _finish_subagents(service)
    assert {
        item["session_id"] for item in observer.results
    } == {"child-session-1", "child-session-2"}
    await service.shutdown()


@pytest.mark.asyncio
async def test_subagent_tool_cleans_running_state_on_error_and_cancellation(
    tmp_path,
) -> None:
    reader = RecordingSessionReader()
    observer = RecordingSubagentObserver()
    runtime: SubagentFakeRuntime | None = None

    def factory(config: BumblehiveConfig) -> Any:
        nonlocal runtime
        runtime = SubagentFakeRuntime(config)
        return runtime

    service = RuntimeService(
        tmp_path / "config.json",
        session_reader=reader,  # type: ignore[arg-type]
        runtime_factory=factory,
    )
    await service.startup()
    assert runtime is not None

    runtime.run_result = SimpleNamespace(
        final_content=None,
        error=AgentError(code="model_error", message="child failed"),
    )
    with observe_subagents(observer):
        failed_task = _start_subagent_tool(
            runtime,
            workspace=tmp_path / "workspace",
            title="Test failure",
            task="Fail",
        )
    await runtime.run_started.wait()
    runtime.run_release.set()

    assert (await failed_task).content == {"status": "pending", "task_id": "delegate"}
    await _finish_subagents(service)
    assert not service.subagents.is_running("child-session")
    assert [item["session_id"] for item in observer.results] == [
        "child-session"
    ]

    runtime.run_started.clear()
    runtime.run_release.clear()
    runtime.run_result = SimpleNamespace(final_content="unused", error=None)
    runtime.run_exception = RuntimeError("stream failed")
    with observe_subagents(observer):
        errored_task = _start_subagent_tool(
            runtime,
            workspace=tmp_path / "workspace",
            title="Test stream error",
            task="Error",
        )
    await runtime.run_started.wait()
    runtime.run_release.set()

    assert (await errored_task).error is None
    await _finish_subagents(service)
    assert not service.subagents.is_running("child-session")
    assert observer.errors == [
        {
            "session_id": "child-session",
            "message": "stream failed",
        }
    ]

    runtime.run_started.clear()
    runtime.run_release.clear()
    runtime.run_exception = None
    runtime.run_result = SimpleNamespace(final_content="unused", error=None)
    with observe_subagents(observer):
        cancelled_task = _start_subagent_tool(
            runtime,
            workspace=tmp_path / "workspace",
            title="Test cancellation",
            task="Wait",
        )
    await runtime.run_started.wait()
    assert (await cancelled_task).error is None
    await service.subagents.cancel("child-session")
    assert not service.subagents.is_running("child-session")
    assert observer.cancelled == ["child-session"]
    await service.shutdown()