"""Tests for omnigent.tools.manager (ToolManager).""" from __future__ import annotations import json import logging from pathlib import Path from typing import Any import pytest from omnigent.errors import OmnigentError from omnigent.spec.types import ( AgentSpec, BuiltinToolConfig, LLMConfig, LocalToolInfo, MCPServerConfig, SharePolicy, SkillSpec, ToolRuntime, ToolsConfig, ) from omnigent.tools import ToolManager from omnigent.tools.base import ToolContext from omnigent.tools.client_specified import ClientSideTool, ClientSideToolSpec from omnigent.tools.mcp import clear_discovery_cache _TEST_CTX = ToolContext(task_id="task_test", agent_id="agent_test") # Tools that ToolManager registers unconditionally on every # spec these tests construct: the lifecycle ``sys_cancel_task`` # plus the async-inbox trio (``sys_call_async``, # ``sys_read_inbox``, ``sys_cancel_async``) which register # whenever ``async_enabled`` is True (the default in # ``_make_spec``). Filtered out in tests that assert on # *specific* tool sets so these don't alter the other checks. # Adding a tool here requires a deliberate decision — this is # a hard-coded allowlist, not a filter on ``_register_*``. # ``check_task`` was dropped per design step 11. _ALWAYS_PRESENT_TOOLS: frozenset[str] = frozenset( { "sys_cancel_task", "sys_call_async", "sys_read_inbox", "sys_cancel_async", # ``load_skill`` is always registered — it discovers # host-scope skills at init time even when the agent # has no bundled skills. "load_skill", # Comment tools are always auto-registered so agents can # list and update review comments without the spec opting in. "list_comments", "update_comment", # Read-only session discovery tools are registered for every # agent (a user-added agent that declares no sub-agents still # needs to list/peek/get-info on its session-mates); the # mutating tools (sys_session_send/close/create/share) are # opt-in via tools.agents or the top-level ``spawn: true``. "sys_session_get_history", "sys_session_list", "sys_session_get_info", # Read-only agent discovery tools are likewise always available # (global, permission-bounded reads of any accessible session's # agent / bundle). "sys_agent_get", "sys_agent_download", "sys_agent_list", # Policy tools are always auto-registered so agents can # browse the registry and add policies at runtime. "sys_add_policy", "sys_policy_registry", # Embedded-browser tools are always auto-registered (framework- # owned) so any agent can drive the desktop app's browser without # the spec opting in. Schema-only; runner-dispatched. "browser_navigate", "browser_snapshot", "browser_click", "browser_type", "browser_screenshot", } ) def _non_lifecycle_schemas( mgr: ToolManager, ) -> list[dict[str, object]]: """ Return tool schemas minus the always-registered lifecycle tool (``sys_cancel_task``). Tests that assert "no tools registered" or "only tool X" should filter out the lifecycle tools — they're part of every ToolManager but orthogonal to whatever the test is actually checking. :param mgr: The :class:`ToolManager` under test. :returns: Schema list with lifecycle entries filtered out. """ return [ s for s in mgr.get_tool_schemas() if not ( isinstance(s, dict) and isinstance(fn := s.get("function"), dict) and fn.get("name") in _ALWAYS_PRESENT_TOOLS ) ] @pytest.fixture() def skill_with_resources(tmp_path: Path) -> SkillSpec: """ A skill with a ``references/`` directory containing a file, for testing ``read_skill_file`` registration. :returns: A ``SkillSpec`` pointing at a real directory with a reference file. """ skill_dir = tmp_path / "skills" / "code-review" skill_dir.mkdir(parents=True) refs_dir = skill_dir / "references" refs_dir.mkdir() (refs_dir / "style-guide.md").write_text("# Style Guide\n\nUse snake_case.") return SkillSpec( name="code-review", description="Reviews code.", content="Review the code.", skill_dir=skill_dir, ) @pytest.fixture() def skill_no_resources() -> SkillSpec: """ A skill with no ``skill_dir`` (in-memory only). :returns: A ``SkillSpec`` with ``skill_dir=None``. """ return SkillSpec( name="summarize", description="Summarizes text.", content="Summarize the input concisely.", ) @pytest.fixture(autouse=True) def _clean_mcp_cache() -> None: """ Clear the MCP discovery cache before each test. """ clear_discovery_cache() def _make_spec( skills: list[SkillSpec] | None = None, mcp_servers: list[MCPServerConfig] | None = None, local_tools: list[LocalToolInfo] | None = None, ) -> AgentSpec: """ Create a minimal ``AgentSpec`` with the given skills, MCP servers, and local tools. :param skills: Skills to include, or ``None`` for no skills. :param mcp_servers: MCP server configs, or ``None`` for no MCP servers. :param local_tools: Local tool infos, or ``None`` for no local tools. :returns: An ``AgentSpec`` with ``spec_version=1``. """ return AgentSpec( spec_version=1, skills=skills or [], mcp_servers=mcp_servers or [], local_tools=local_tools or [], ) # ── Registry dispatch ───────────────────────────────── def test_registry_dispatches_to_load_skill( skill_no_resources: SkillSpec, ) -> None: """ ToolManager.call_tool dispatches to LoadSkillTool via the registry. """ mgr = ToolManager( _make_spec([skill_no_resources]), ) result = mgr.call_tool( "load_skill", json.dumps({"name": "summarize"}), _TEST_CTX, ) assert result == "Summarize the input concisely." def test_registry_dispatches_to_read_skill_file( skill_with_resources: SkillSpec, ) -> None: """ ToolManager.call_tool dispatches to ReadSkillFileTool via the registry. """ mgr = ToolManager( _make_spec([skill_with_resources]), ) result = mgr.call_tool( "read_skill_file", json.dumps( { "skill_name": "code-review", "path": "references/style-guide.md", } ), _TEST_CTX, ) assert "# Style Guide" in result def test_registry_unknown_tool_returns_error( skill_no_resources: SkillSpec, ) -> None: """ ToolManager.call_tool returns error for unregistered tools. """ mgr = ToolManager( _make_spec([skill_no_resources]), ) result = mgr.call_tool("nonexistent", json.dumps({}), _TEST_CTX) assert "not found" in result assert "load_skill" in result # ── get_tool_schemas ────────────────────────────────── def test_schemas_include_load_skill_when_skills_exist( skill_no_resources: SkillSpec, ) -> None: """ get_tool_schemas includes load_skill when the agent has skills, and the schema description enumerates the available skill names. The enumeration is the discovery mechanism for harnesses with no native skill loader (openai-agents-sdk, and the in-process loop): they receive ``load_skill`` via ``request.tools`` and learn which skills exist from its description alone — they do not depend on the prompt's skill listing. If the enumeration regressed, those harnesses could call ``load_skill`` but wouldn't know any skill names to pass. """ mgr = ToolManager( _make_spec([skill_no_resources]), ) schemas = mgr.get_tool_schemas() by_name = {s["function"]["name"]: s for s in schemas} assert "load_skill" in by_name # The bundled skill ("summarize") is named in the tool description so a # loaderless harness can discover and load it. assert "summarize" in by_name["load_skill"]["function"]["description"] def test_schemas_include_read_skill_file_with_resources( skill_with_resources: SkillSpec, ) -> None: """ get_tool_schemas includes read_skill_file when a skill has bundled resource files. """ mgr = ToolManager( _make_spec([skill_with_resources]), ) schemas = mgr.get_tool_schemas() names = [s["function"]["name"] for s in schemas] assert "read_skill_file" in names def test_schemas_exclude_read_skill_file_without_resources( skill_no_resources: SkillSpec, ) -> None: """ get_tool_schemas does NOT include read_skill_file when no skill has bundled resources. """ mgr = ToolManager( _make_spec([skill_no_resources]), ) schemas = mgr.get_tool_schemas() names = [s["function"]["name"] for s in schemas] assert "read_skill_file" not in names def test_schemas_empty_when_no_skills() -> None: """ get_tool_schemas returns empty when agent has no skills, excluding the always-registered lifecycle tool (``sys_cancel_task``). """ mgr = ToolManager(_make_spec([])) assert _non_lifecycle_schemas(mgr) == [] def test_schemas_isolate_a_failing_tool( caplog: pytest.LogCaptureFixture, ) -> None: """ A tool whose ``get_schema`` raises is skipped, not allowed to drop the entire toolset: every other tool's schema is still returned and a warning names the offending tool. Regression test for the silent total-toolset drop in #378. A single unimportable ``type: function`` tool (dotted ``callable`` path that fails to import) used to abort the whole list comprehension, leaving the model with zero declared tools and only a swallowed WARNING. """ class _BoomTool: def get_schema(self) -> dict[str, Any]: raise ImportError("No module named 'boom'") mgr = ToolManager(_make_spec([])) healthy = {s["function"]["name"] for s in mgr.get_tool_schemas()} assert healthy # sanity: always-present lifecycle tools exist mgr._tools["boom"] = _BoomTool() # type: ignore[assignment] with caplog.at_level(logging.WARNING, logger="omnigent.tools.manager"): schemas = mgr.get_tool_schemas() names = {s["function"]["name"] for s in schemas} assert "boom" not in names assert names == healthy # every healthy tool survived the bad one assert any("boom" in record.getMessage() for record in caplog.records) def test_client_schemas_isolate_a_failing_tool( caplog: pytest.LogCaptureFixture, ) -> None: """ A client-side tool whose ``get_schema`` raises is skipped, not allowed to drop the entire client toolset: every other client tool's schema is still returned and a warning names the offending tool. Mirrors ``test_schemas_isolate_a_failing_tool`` for the ``get_client_tool_schemas`` path that ``SpawnTool`` uses to propagate client tools to sub-agents (#378). """ class _BoomClientTool(ClientSideTool): def get_schema(self) -> dict[str, Any]: raise ImportError("No module named 'boom'") mgr = ToolManager(_make_spec([])) healthy_tool = ClientSideTool( ClientSideToolSpec( name="weather", schema={"type": "function", "function": {"name": "weather"}}, ) ) mgr._tools["weather"] = healthy_tool healthy = {s["function"]["name"] for s in mgr.get_client_tool_schemas()} assert healthy == {"weather"} # sanity: the good client tool is advertised mgr._tools["boom"] = _BoomClientTool(ClientSideToolSpec(name="boom", schema={})) with caplog.at_level(logging.WARNING, logger="omnigent.tools.manager"): schemas = mgr.get_client_tool_schemas() names = {s["function"]["name"] for s in schemas} assert "boom" not in names assert names == healthy # the healthy client tool survived the bad one assert any("boom" in record.getMessage() for record in caplog.records) def test_session_reads_registered_but_writes_gated_without_opt_in() -> None: """ Read-only session discovery (``sys_session_get_history`` / ``sys_session_list`` / ``sys_session_get_info``) is registered for **every** agent, even one that declares no sub-agents — so a user-added agent can read its session-mates for context. The mutating session tools (``sys_session_send`` / ``sys_session_close`` / ``sys_session_create`` / ``sys_session_share``) are NOT registered without an opt-in (``tools.agents`` or top-level ``spawn: true``). A regression that registered the writes by default would expose the child-session spawn surface — and, for share, the ability to expose the session to a third party or ``__public__`` — to every custom agent. """ mgr = ToolManager(_make_spec([])) names = {s["function"]["name"] for s in mgr.get_tool_schemas()} assert "sys_session_get_history" in names assert "sys_session_list" in names # get_info is a read like peek/list — always available so any agent # can inspect a session it can access. If absent, the registration # in _register_sub_agent_tools regressed and the orchestrator can't # check session status. assert "sys_session_get_info" in names assert "sys_session_send" not in names assert "sys_session_close" not in names assert "sys_session_create" not in names # Sharing has its OWN dedicated `share:` flag (default `none`), so it # is absent here even though this spec also lacks spawn/agents. A # regression registering it by default would let any prompt-injected # agent expose its session (incl. via __public__). assert "sys_session_share" not in names # Model awareness pairs with the dispatch grant — without send there # is no args.model to pick, so the listing tool must stay gated too. assert "sys_list_models" not in names # The read-only sys_agent_get/list/download stay always-on. assert "sys_agent_get" in names assert "sys_agent_list" in names def test_spawn_flag_registers_write_tools_without_sub_agents() -> None: """ Top-level ``spawn: true`` registers the spawn writes without any declared sub-agents — the opt-in for agents that author a config locally and launch it via ``sys_session_create(config_path=...)``. If this gate arm regressed, ``spawn: true`` in YAML would have no runtime effect and the author-and-launch flow would be impossible without bundling sub-agent directories. """ spec = AgentSpec(spec_version=1, spawn=True) names = {s["function"]["name"] for s in ToolManager(spec).get_tool_schemas()} assert "sys_session_send" in names assert "sys_session_close" in names assert "sys_session_create" in names # The dispatch grant brings model awareness along with it. assert "sys_list_models" in names # Sharing is DECOUPLED from spawn — its own `share:` flag governs it, # so `spawn: true` alone (share defaulting to `none`) does NOT # register it. A regression coupling them would re-expose sharing to # every spawn-capable agent. assert "sys_session_share" not in names def test_session_send_schema_drops_named_mode_without_sub_agents() -> None: """ With the ``spawn: true`` opt-in but no declared sub-agents, ``sys_session_send``'s schema omits the named-mode ``agent`` / ``title`` parameters (an empty enum is unusable and invalid for some providers) and advertises only the ``session_id`` mode; with declared sub-agents, the named mode is present with the declared-type enum. If the empty case regressed to emitting ``"enum": []``, providers that validate schemas would reject the tool list outright. """ bare = ToolManager(AgentSpec(spec_version=1, spawn=True)) bare_schema = next( s for s in bare.get_tool_schemas() if s["function"]["name"] == "sys_session_send" ) bare_params = bare_schema["function"]["parameters"] assert set(bare_params["properties"]) == {"session_id", "args"} spec = AgentSpec( spec_version=1, tools=ToolsConfig(agents=["researcher"]), sub_agents=[AgentSpec(spec_version=1, name="researcher")], ) named = ToolManager(spec) named_schema = next( s for s in named.get_tool_schemas() if s["function"]["name"] == "sys_session_send" ) named_params = named_schema["function"]["parameters"] assert set(named_params["properties"]) == {"agent", "title", "session_id", "args"} # The enum carries exactly the declared sub-agent names. assert named_params["properties"]["agent"]["enum"] == ["researcher"] def test_declared_agents_grant_send_close_but_not_create() -> None: """ Declaring ``tools.agents`` permits spawning ONLY the specified sub-agent list: ``sys_session_send`` (named mode over the declared enum) and ``sys_session_close`` register, but ``sys_session_create`` does NOT — launching arbitrary agents (by id or custom bundle) requires the separate ``spawn: true`` grant. If create leaked in here, a spec that whitelisted two sub-agent types could launch any accessible agent or upload arbitrary bundles. """ spec = AgentSpec( spec_version=1, tools=ToolsConfig(agents=["researcher"]), sub_agents=[AgentSpec(spec_version=1, name="researcher")], ) mgr = ToolManager(spec) names = {s["function"]["name"] for s in mgr.get_tool_schemas()} assert "sys_session_send" in names assert "sys_session_close" in names assert "sys_session_create" not in names # Model awareness rides the same grant as send. assert "sys_list_models" in names # Declaring sub-agents does NOT enable sharing — that is the separate # `share:` flag's job, decoupled from the spawn/agents grant. assert "sys_session_share" not in names def test_share_non_public_registers_share_tool_without_public() -> None: """ ``agent_session_sharing: non-public`` alone (no spawn / declared agents) registers ``sys_session_share`` — proving the flag is independently sufficient AND does not drag in the spawn-lifecycle tools. The advertised ``user_id`` schema must NOT mention ``__public__``, so the model is not offered a grantee the runner would reject. If the gating regressed to the spawn opt-in, share would be absent here. """ mgr = ToolManager(AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.NON_PUBLIC)) schemas = {s["function"]["name"]: s for s in mgr.get_tool_schemas()} assert "sys_session_share" in schemas # Sharing is decoupled from spawn — none of the spawn writes ride along. assert "sys_session_send" not in schemas assert "sys_session_close" not in schemas assert "sys_session_create" not in schemas # non-public must not advertise the public sentinel. user_id_desc = schemas["sys_session_share"]["function"]["parameters"]["properties"]["user_id"][ "description" ] assert "__public__" not in user_id_desc def test_share_public_registers_share_tool_advertising_public() -> None: """ ``agent_session_sharing: public`` registers ``sys_session_share`` and the advertised ``user_id`` schema DOES mention ``__public__`` — the only tier where anonymous-read grants are permitted. If ``allow_public`` weren't threaded from the flag into the tool, the public option would be hidden (or, worse, advertised under non-public). """ mgr = ToolManager(AgentSpec(spec_version=1, agent_session_sharing=SharePolicy.PUBLIC)) schemas = {s["function"]["name"]: s for s in mgr.get_tool_schemas()} assert "sys_session_share" in schemas user_id_desc = schemas["sys_session_share"]["function"]["parameters"]["properties"]["user_id"][ "description" ] assert "__public__" in user_id_desc def test_both_grants_compose() -> None: """ A spec with BOTH ``tools.agents`` and ``spawn: true`` (the nessie/polly shape) gets the union: named-mode send over the declared enum, close, AND create for self-defined children. If composition regressed (e.g. one arm short-circuiting the other), an orchestrator would lose either its declared sub-agents or its author-and-launch flow. """ spec = AgentSpec( spec_version=1, spawn=True, tools=ToolsConfig(agents=["researcher"]), sub_agents=[AgentSpec(spec_version=1, name="researcher")], ) schemas = {s["function"]["name"]: s["function"] for s in ToolManager(spec).get_tool_schemas()} assert "sys_session_close" in schemas assert "sys_session_create" in schemas # Named mode survives alongside the spawn grant. assert schemas["sys_session_send"]["parameters"]["properties"]["agent"]["enum"] == [ "researcher" ] def test_session_get_info_schema_has_optional_session_id() -> None: """ ``sys_session_get_info`` advertises a single optional ``session_id`` parameter (omitting it defaults to the caller's own session). If ``session_id`` became required, an agent inspecting its own session would be forced to look up its own id first. """ mgr = ToolManager(_make_spec([])) schema = next( s for s in mgr.get_tool_schemas() if s["function"]["name"] == "sys_session_get_info" ) params = schema["function"]["parameters"] assert set(params["properties"]) == {"session_id"} # No required fields — session_id is optional by design. assert params["required"] == [] def test_agent_read_tools_registered_for_every_agent() -> None: """ ``sys_agent_get`` and ``sys_agent_download`` are registered for **every** agent — they are global, permission-bounded reads of any accessible session's agent/bundle, like the session reads. If they regress out of registration, an orchestrator can't inspect or fork agents it can see. """ mgr = ToolManager(_make_spec([])) names = {s["function"]["name"] for s in mgr.get_tool_schemas()} assert "sys_agent_get" in names assert "sys_agent_download" in names assert "sys_agent_list" in names # get/download require a session_id (an agent is only inspectable # while running in some session); list takes no parameters. for tool_name in ("sys_agent_get", "sys_agent_download"): schema = next(s for s in mgr.get_tool_schemas() if s["function"]["name"] == tool_name) assert "session_id" in schema["function"]["parameters"]["required"] list_schema = next( s for s in mgr.get_tool_schemas() if s["function"]["name"] == "sys_agent_list" ) assert list_schema["function"]["parameters"]["properties"] == {} # ── MCP integration ────────────────────────────────────── def test_shutdown_safe_without_start() -> None: """ ``shutdown()`` is safe to call without ``start()``. """ spec = _make_spec() mgr = ToolManager(spec) mgr.shutdown() def test_shutdown_idempotent() -> None: """Calling ``shutdown()`` twice does not raise.""" spec = _make_spec() mgr = ToolManager(spec) mgr.start() mgr.shutdown() mgr.shutdown() def test_shutdown_closes_os_env(monkeypatch: pytest.MonkeyPatch) -> None: """``shutdown()`` closes ``_os_env`` when it was self-created.""" spec = _make_spec() mgr = ToolManager(spec) mgr.start() class _FakeOSEnv: closed = False def close(self) -> None: self.closed = True fake_env = _FakeOSEnv() mgr._os_env = fake_env # type: ignore[assignment] mgr._pre_resolved_os_env = None mgr.shutdown() assert fake_env.closed assert mgr._os_env is None def test_shutdown_skips_pre_resolved_os_env() -> None: """``shutdown()`` does NOT close a pre-resolved (shared) OS env.""" spec = _make_spec() class _FakeOSEnv: closed = False def close(self) -> None: self.closed = True shared_env = _FakeOSEnv() mgr = ToolManager(spec, os_env=shared_env) # type: ignore[arg-type] mgr.start() mgr.shutdown() assert not shared_env.closed def test_shutdown_calls_tool_shutdown() -> None: """``shutdown()`` calls ``shutdown()`` on every registered tool.""" from omnigent.tools.base import Tool class _TrackingTool(Tool): shut_down = False @classmethod def name(cls) -> str: return "_tracking" @classmethod def description(cls) -> str: return "test" def get_schema(self) -> dict[str, Any]: return {"type": "function", "function": {"name": "_tracking"}} def shutdown(self) -> None: self.shut_down = True spec = _make_spec() mgr = ToolManager(spec) tracker = _TrackingTool() mgr._tools["_tracking"] = tracker mgr.start() mgr.shutdown() assert tracker.shut_down # ── Client-specified tools ──────────────────────────────── def _make_client_side_spec(name: str) -> ClientSideToolSpec: """ Build a minimal :class:`ClientSideToolSpec` for use in manager tests. :param name: Tool function name, e.g. ``"get_weather"``. :returns: A :class:`ClientSideToolSpec` with a minimal schema. """ return ClientSideToolSpec( name=name, schema={ "type": "function", "function": {"name": name, "description": "A test tool.", "parameters": {}}, }, ) def test_client_tools_registered_in_schemas() -> None: """ Client-specified tools appear in get_tool_schemas() alongside built-in tools without calling start(). A failure here means the LLM never sees client tools — the client_tool_specs constructor arg is not being wired up. """ spec = _make_spec() mgr = ToolManager( spec, client_tool_specs=[ _make_client_side_spec("get_weather"), _make_client_side_spec("send_email"), ], ) # Filter out the always-present lifecycle tool # (sys_cancel_task) — orthogonal to the client-tool # registration being tested. schemas = _non_lifecycle_schemas(mgr) names = [s["function"]["name"] for s in schemas] # Both client tools appear in schemas — 2 registered, 2 returned assert len(schemas) == 2, ( f"Expected 2 schemas (2 client tools), got {len(schemas)}. " "If 0, client_tool_specs are not being registered." ) assert "get_weather" in names assert "send_email" in names def test_is_client_side_tool_returns_true_for_registered_client_tools() -> None: """ is_client_side_tool returns True for registered ClientSideTool entries and False for built-in tools and unknown names. The agent loop uses this to detect when to complete the response instead of executing tools server-side. A failure here would cause client-side tools to be dispatched through call_tool, triggering RuntimeError from ClientSideTool.invoke. """ spec = _make_spec(skills=[SkillSpec(name="summarize", description=".", content=".")]) mgr = ToolManager( spec, client_tool_specs=[ _make_client_side_spec("get_weather"), _make_client_side_spec("send_email"), ], ) # Client tools are detected as client-side assert mgr.is_client_side_tool("get_weather") is True, ( "Expected True for registered ClientSideTool 'get_weather'. " "If False, is_client_side_tool is not checking isinstance(tool, ClientSideTool)." ) assert mgr.is_client_side_tool("send_email") is True # Built-in tool is not client-side assert mgr.is_client_side_tool("load_skill") is False, ( "Expected False for built-in 'load_skill'. " "If True, is_client_side_tool is not type-checking correctly." ) # Unregistered tool is not client-side assert mgr.is_client_side_tool("nonexistent") is False def test_client_tool_shadows_skill_tool( skill_no_resources: SkillSpec, ) -> None: """ A client tool with the same name as a skill tool overwrites the skill tool (last registered wins, with a warning). This ensures the override behavior is intentional — clients can replace spec-defined tools at request time. """ spec = _make_spec(skills=[skill_no_resources]) mgr = ToolManager( spec, # 'load_skill' is the built-in skill tool name client_tool_specs=[_make_client_side_spec("load_skill")], ) schemas = mgr.get_tool_schemas() # Only one 'load_skill' — client version overwrote built-in names = [s["function"]["name"] for s in schemas] assert names.count("load_skill") == 1, ( f"Expected exactly one 'load_skill' (client overwrite), got {names.count('load_skill')}." ) # The registered tool is the client's ClientSideTool, not LoadSkillTool from omnigent.tools.client_specified import ClientSideTool assert isinstance(mgr._tools["load_skill"], ClientSideTool), ( "Expected ClientSideTool after client override, " f"got {type(mgr._tools['load_skill']).__name__}." ) def test_client_tools_none_equivalent_to_empty() -> None: """ Passing client_tool_specs=None and client_tool_specs=[] produce the same result: no client tools registered. """ spec = _make_spec() mgr_none = ToolManager(spec, client_tool_specs=None) mgr_empty = ToolManager(spec, client_tool_specs=[]) assert _non_lifecycle_schemas(mgr_none) == [] assert _non_lifecycle_schemas(mgr_empty) == [] # ── Spec-declared client tools (runtime: client) ────────── def _spec_with_local( name: str, runtime: ToolRuntime = ToolRuntime.SERVER, parameters: dict[str, Any] | None = None, ) -> AgentSpec: """ Build an :class:`AgentSpec` with a single local-tool entry. Mirrors the per-entry ``runtime`` field that :class:`LocalToolInfo` (and only :class:`LocalToolInfo` — builtins are intrinsically server-side) carries. Server-runtime entries declare a fake ``path``/``language`` so the existing loaders skip them gracefully when ``workdir`` is ``None``; client-runtime entries declare ``parameters`` (required by the validator and by the manager's :class:`ClientSideTool` registration). :param name: The local tool name, e.g. ``"open_in_editor"``. :param runtime: Either :attr:`ToolRuntime.SERVER` (default) or :attr:`ToolRuntime.CLIENT`. ``CLIENT`` opts the tool into the spec-declared client-side dispatch path under test. :param parameters: JSON-Schema ``parameters`` block. Required for ``CLIENT`` entries (no callable to introspect); ignored for ``SERVER`` entries. :returns: A minimal :class:`AgentSpec` with that single local tool. """ if runtime == ToolRuntime.CLIENT: info = LocalToolInfo( name=name, path=None, language="python", runtime=runtime, parameters=parameters or {"type": "object", "properties": {}}, ) else: info = LocalToolInfo( name=name, path=f"tools/python/{name}.py", language="python", runtime=runtime, ) return AgentSpec( spec_version=1, local_tools=[info], ) def test_is_client_side_tool_true_for_spec_declared_client_local_tool() -> None: """ A local tool declared with ``runtime: client`` in the spec is reported as client-side by :meth:`ToolManager.is_client_side_tool`, even when no request-time ``client_tool_specs`` are passed. Claim: spec-declared client tools take the same dispatch path (``action_required``) as request-supplied client tools. Failure here would mean ``runtime: client`` is silently ignored and the workflow would try to execute the local tool server-side. """ spec = _spec_with_local("open_in_editor", runtime=ToolRuntime.CLIENT) mgr = ToolManager(spec) # Spec-declared client tool is detected as client-side. assert mgr.is_client_side_tool("open_in_editor") is True, ( "Expected True for 'open_in_editor' declared as runtime: client. " "If False, ToolManager is not consulting the spec's runtime field." ) def test_is_client_side_tool_false_for_default_server_local_tool() -> None: """ A local tool without ``runtime: client`` (default :attr:`ToolRuntime.SERVER`) is NOT reported as client-side. Claim: the default runtime ``SERVER`` preserves the existing server-side dispatch path. Failure would mean every local tool is treated as client-side regardless of its declared runtime — catastrophic regression of the existing path. Note: the manager is created with ``workdir=None`` so the underlying file-based loader doesn't try to open the nonexistent path on disk. Only the ``runtime`` field matters for what we're asserting here. """ spec = _spec_with_local("echo_tool", runtime=ToolRuntime.SERVER) mgr = ToolManager(spec) assert mgr.is_client_side_tool("echo_tool") is False, ( "Expected False for 'echo_tool' with default runtime: server. " "If True, the default runtime is being misread." ) def test_spec_client_local_tool_schema_still_visible_to_llm() -> None: """ The schema for a spec-declared client tool is still emitted by :meth:`ToolManager.get_tool_schemas`, so the LLM sees the tool and can call it. Claim: marking a tool ``runtime: client`` does not hide its schema from the LLM — it only changes the dispatch path. Failure would mean the LLM never learns the tool exists and can't invoke it, defeating the feature. """ spec = _spec_with_local( "open_in_editor", runtime=ToolRuntime.CLIENT, parameters={ "type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"], }, ) mgr = ToolManager(spec) schemas = mgr.get_tool_schemas() names = [s["function"]["name"] for s in schemas] assert "open_in_editor" in names, ( f"Expected 'open_in_editor' in tool schemas even when runtime=client. " f"Got {names}. If missing, the spec-declared client local tool " f"is not being registered as a ClientSideTool." ) # The parameters block from the spec must be carried through to # the LLM-facing schema; otherwise the LLM emits empty arguments. fn = next(s["function"] for s in schemas if s["function"]["name"] == "open_in_editor") assert fn["parameters"]["required"] == ["path"], ( f"Expected parameters.required=['path'] from the spec to flow " f"through to the schema, got {fn.get('parameters')!r}." ) def test_spec_client_local_tool_in_get_client_tool_schemas() -> None: """ :meth:`ToolManager.get_client_tool_schemas` includes spec-declared client tools alongside request-time :class:`ClientSideTool` instances. Claim: sub-agents launched via SpawnTool inherit spec-declared client tools the same way they inherit request-supplied ones — both go through ``get_client_tool_schemas``. Failure would mean sub-agents lose the spec-declared client tool surface. """ spec = _spec_with_local("open_in_editor", runtime=ToolRuntime.CLIENT) mgr = ToolManager( spec, client_tool_specs=[_make_client_side_spec("get_weather")], ) names = [s["function"]["name"] for s in mgr.get_client_tool_schemas()] # Both paths represented: 1 spec-declared + 1 request-supplied. assert sorted(names) == ["get_weather", "open_in_editor"], ( f"Expected both client paths in get_client_tool_schemas, " f"got {sorted(names)}. If 'open_in_editor' is missing, " f"spec-declared client tools are not propagated to sub-agents. " f"If 'get_weather' is missing, the request-time path regressed." ) def test_spec_client_and_request_client_coexist() -> None: """ Spec-declared and request-supplied client tools both register and both are reported as client-side. Claim: the two paths are additive — a single ToolManager can expose tools from both sources, and ``is_client_side_tool`` returns True for both. Failure would mean one path overrides or hides the other, breaking the integration story. """ spec = _spec_with_local("open_in_editor", runtime=ToolRuntime.CLIENT) mgr = ToolManager( spec, client_tool_specs=[_make_client_side_spec("get_weather")], ) # Spec-declared path. assert mgr.is_client_side_tool("open_in_editor") is True # Request-time path (existing behavior preserved). assert mgr.is_client_side_tool("get_weather") is True schema_names = [s["function"]["name"] for s in mgr.get_tool_schemas()] assert "open_in_editor" in schema_names assert "get_weather" in schema_names def test_is_client_side_tool_false_for_unregistered_name() -> None: """ :meth:`ToolManager.is_client_side_tool` returns False for an unknown name even after introducing the spec-declared path. Guards against an off-by-one where a name typo'd in the spec (or arbitrary string) gets misclassified as client-side just because it's in some set. """ spec = _spec_with_local("open_in_editor", runtime=ToolRuntime.CLIENT) mgr = ToolManager(spec) assert mgr.is_client_side_tool("definitely_not_a_real_tool") is False @pytest.mark.parametrize( "name", [ "tool with spaces", "tool:colon", "", "a" * 257, ], ids=[ "spaces", "colon", "empty", "too_long", ], ) def test_client_tool_invalid_name_raises( name: str, ) -> None: """ Client-specified tools with invalid names raise ``OmnigentError`` at registration time. """ spec = _make_spec() with pytest.raises(OmnigentError, match="Invalid client tool name"): ToolManager( spec, client_tool_specs=[_make_client_side_spec(name)], ) # ── Local tool registration ────────────────────────── def _write_local_tool( workdir: Path, filename: str, schema_name: str, ) -> None: """ Write a minimal local Python tool file to ``workdir/tools/python/``. :param workdir: Agent image root directory. :param filename: File name, e.g. ``"web_fetch.py"``. :param schema_name: The ``@tool``-decorated function's name — becomes the LLM-facing tool name. """ py_dir = workdir / "tools" / "python" py_dir.mkdir(parents=True, exist_ok=True) code = ( '"""Test tool."""\n' "from omnigent_client import tool\n" "\n" "\n" "@tool\n" f"def {schema_name}() -> str:\n" ' """Execute."""\n' ' return "local_tool_result"\n' ) (py_dir / filename).write_text(code) def test_local_tools_registered_and_callable( tmp_path: Path, ) -> None: """ ToolManager registers local Python tools from the workdir and dispatches calls to them. """ _write_local_tool(tmp_path, "echo_tool.py", "echo_tool") info = LocalToolInfo( name="echo_tool", path="tools/python/echo_tool.py", language="python", ) spec = _make_spec(local_tools=[info]) mgr = ToolManager(spec, workdir=tmp_path) schemas = mgr.get_tool_schemas() # Local tool appears in the schema list. names = [s["function"]["name"] for s in schemas] assert "echo_tool" in names, ( f"Expected 'echo_tool' in schemas, got {names}. " f"If missing, _register_local_tools did not register the tool." ) # Dispatching works through call_tool. result = mgr.call_tool("echo_tool", json.dumps({}), _TEST_CTX) assert result == "local_tool_result", ( f"Expected 'local_tool_result', got {result!r}. " f"If 'Error: tool not found', the tool was not registered." ) def test_local_tools_skipped_without_workdir() -> None: """ ToolManager with workdir=None skips local tool registration without error, even if spec has local_tools. """ info = LocalToolInfo( name="some_tool", path="tools/python/some_tool.py", language="python", ) spec = _make_spec(local_tools=[info]) # workdir=None (default) — should not raise. mgr = ToolManager(spec) # Excluding the always-present lifecycle tool # (sys_cancel_task), no tools should register when workdir # is None and no skills / builtins are set. assert _non_lifecycle_schemas(mgr) == [], ( "No tools (apart from the always-present sys_cancel_task) " "should be registered when workdir is None." ) # ── web_search builtin: Databricks model does not emit web_search_preview ─── def test_web_search_does_not_emit_web_search_preview_for_databricks_model() -> None: """ When the agent's model is a ``databricks-*`` model, the ``web_search`` builtin must NOT emit ``{"type": "web_search_preview"}`` in its schema. Databricks rejects that tool type at the API level with HTTP 400, killing the request before the agent runs. A failure here means the ``databricks-*`` guard in ``ToolManager._create_web_search`` was removed or the prefix changed. The wrong schema would be ``{"type": "web_search_preview"}``. """ spec = AgentSpec( spec_version=1, llm=LLMConfig(model="databricks-gpt-5-4"), tools=ToolsConfig(builtins=[BuiltinToolConfig(name="web_search")]), ) mgr = ToolManager(spec) tool = mgr.get_tool("web_search") assert tool is not None, "web_search should be registered when declared in tools.builtins" schema = tool.get_schema() assert schema.get("type") != "web_search_preview", ( f'web_search emitted {{"type": "web_search_preview"}} for ' f"databricks-gpt-5-4 — Databricks does not support this tool type " f"and rejects the request with HTTP 400. Got schema: {schema!r}" )