diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 2b7de85e..977d01f1 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -209,6 +209,13 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool: invoke_tool = tool.on_invoke_tool async def invoke(ctx: Any, raw_input: str) -> Any: + try: + parsed = json.loads(raw_input) + except (json.JSONDecodeError, TypeError): + parsed = None + if isinstance(parsed, dict) and "shell" not in parsed: + parsed["shell"] = "bash" + raw_input = json.dumps(parsed) try: return await invoke_tool(ctx, raw_input) except ValidationError as exc: diff --git a/tests/test_agent_factory_shell.py b/tests/test_agent_factory_shell.py new file mode 100644 index 00000000..6de93219 --- /dev/null +++ b/tests/test_agent_factory_shell.py @@ -0,0 +1,51 @@ +"""Tests for the shell tool adapters in the agent factory.""" + +from __future__ import annotations + +import json +from typing import Any, cast + +import pytest +from agents.tool import FunctionTool + +from strix.agents import factory + + +def _capturing_exec_tool(captured: dict[str, str]) -> FunctionTool: + async def invoke(_ctx: Any, raw_input: str) -> str: + captured["raw_input"] = raw_input + return "ok" + + return FunctionTool( + name="exec_command", + description="test tool", + params_json_schema={"type": "object", "properties": {}}, + on_invoke_tool=invoke, + ) + + +@pytest.mark.asyncio +async def test_wrap_exec_command_defaults_shell_to_bash() -> None: + captured: dict[str, str] = {} + wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured)) + + result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "source /tmp/env"})) + + assert result == "ok" + assert json.loads(captured["raw_input"]) == { + "cmd": "source /tmp/env", + "shell": "bash", + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize("shell", ["/bin/zsh", ""]) +async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None: + captured: dict[str, str] = {} + wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured)) + + await wrapped.on_invoke_tool( + cast("Any", None), json.dumps({"cmd": "echo test", "shell": shell}) + ) + + assert json.loads(captured["raw_input"])["shell"] == shell