mirror of
https://github.com/ggozad/oterm.git
synced 2026-10-17 20:53:21 +02:00
28206897ac- Drop throttle and debounce; LogViewer polls with set_interval. - OTerm actions share an _active_chat() lookup. - ChatEdit dismisses with a ChatModel instead of JSON, drops the supported-settings filter (every parameter it offers is in the base ModelSettings, so the filter never removed anything), and holds its state in plain attributes instead of unwatched reactives. - ToolMeta is a TypedDict; resolve_api_key and BUILTIN_OPENAI_COMPAT are public since agent.py imports them. - Ollama URL helpers derive from one another and share one client constructor. - Store maps rows to ChatModel in one place; build_user_prompt decodes images in one place. - Remove mutable default arguments, unused attributes and parameters, and four tcss rules that match nothing. - Rename supress_logging to suppress_logging. - Delete comments that restate the code; correct ones that were false (the logging comment in cli, the pydantic-ai 2.0 note in the 0.18.0 upgrade, a stale Textual version in cancel_streams).
277 lines
11 KiB
Python
277 lines
11 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from pydantic_ai import Tool as PydanticTool
|
|
from pydantic_ai.messages import ToolReturnPart
|
|
|
|
from oterm.app.widgets.chat import ChatContainer, _resolve_tools
|
|
from oterm.types import ChatModel
|
|
|
|
|
|
class _FakeServer:
|
|
def __init__(self) -> None:
|
|
self.filtered_args: list = []
|
|
|
|
def filtered(self, predicate):
|
|
self.filtered_args.append(predicate)
|
|
return f"toolset-{id(predicate)}"
|
|
|
|
|
|
class TestResolveTools:
|
|
def test_empty_returns_empty(self):
|
|
tools, toolsets, capabilities = _resolve_tools([])
|
|
assert tools == []
|
|
assert toolsets == []
|
|
assert capabilities == []
|
|
|
|
def test_unknown_tool_logged_and_dropped(self, monkeypatch):
|
|
import oterm.log
|
|
|
|
monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", [])
|
|
monkeypatch.setattr("oterm.app.widgets.chat.mcp_tool_meta", {})
|
|
before = len(oterm.log.log_lines)
|
|
tools, toolsets, capabilities = _resolve_tools(["ghost"])
|
|
assert tools == []
|
|
assert toolsets == []
|
|
assert capabilities == []
|
|
messages = [msg for _, msg in oterm.log.log_lines[before:]]
|
|
assert any("unavailable tools" in m and "ghost" in m for m in messages)
|
|
|
|
def test_known_tool_returned(self, monkeypatch):
|
|
def sample() -> str:
|
|
return "x"
|
|
|
|
tool = PydanticTool(sample, takes_ctx=False)
|
|
monkeypatch.setattr(
|
|
"oterm.app.widgets.chat.builtin_tools",
|
|
[{"name": "sample", "description": "", "tool": tool}],
|
|
)
|
|
monkeypatch.setattr("oterm.app.widgets.chat.mcp_tool_meta", {})
|
|
tools, toolsets, capabilities = _resolve_tools(["sample"])
|
|
assert tools == [tool]
|
|
assert toolsets == []
|
|
assert capabilities == []
|
|
|
|
def test_capability_resolved(self, monkeypatch):
|
|
from pydantic_ai.capabilities import WebSearch
|
|
|
|
import oterm.log
|
|
|
|
monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", [])
|
|
monkeypatch.setattr("oterm.app.widgets.chat.mcp_tool_meta", {})
|
|
before = len(oterm.log.log_lines)
|
|
tools, toolsets, capabilities = _resolve_tools(["web_search"])
|
|
assert tools == []
|
|
assert toolsets == []
|
|
assert len(capabilities) == 1
|
|
assert isinstance(capabilities[0], WebSearch)
|
|
messages = [msg for _, msg in oterm.log.log_lines[before:]]
|
|
assert not any("unavailable tools" in m for m in messages)
|
|
|
|
def test_mcp_tools_become_prefixed_filtered_toolset(self, monkeypatch):
|
|
from pydantic_ai.toolsets import PrefixedToolset
|
|
|
|
server = _FakeServer()
|
|
monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", [])
|
|
monkeypatch.setattr(
|
|
"oterm.app.widgets.chat.mcp_tool_meta",
|
|
{"oracle": [{"name": "ask_oracle", "description": ""}]},
|
|
)
|
|
monkeypatch.setattr("oterm.app.widgets.chat.mcp_servers", {"oracle": server})
|
|
tools, toolsets, capabilities = _resolve_tools(["oracle_ask_oracle"])
|
|
assert tools == []
|
|
assert capabilities == []
|
|
assert len(toolsets) == 1
|
|
assert isinstance(toolsets[0], PrefixedToolset)
|
|
assert toolsets[0].prefix == "oracle"
|
|
# The filter predicate sees unprefixed names, as it wraps the raw toolset.
|
|
predicate = server.filtered_args[0]
|
|
assert predicate(None, SimpleNamespace(name="ask_oracle")) is True
|
|
assert predicate(None, SimpleNamespace(name="other")) is False
|
|
|
|
def test_same_tool_name_on_two_servers_resolves_both_independently(
|
|
self, monkeypatch
|
|
):
|
|
"""Regression for #321: a name shared by two servers must not conflict."""
|
|
k8s, grafana = _FakeServer(), _FakeServer()
|
|
monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", [])
|
|
monkeypatch.setattr(
|
|
"oterm.app.widgets.chat.mcp_tool_meta",
|
|
{
|
|
"k8s": [{"name": "query_prometheus", "description": ""}],
|
|
"grafana": [{"name": "query_prometheus", "description": ""}],
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
"oterm.app.widgets.chat.mcp_servers", {"k8s": k8s, "grafana": grafana}
|
|
)
|
|
_, toolsets, _ = _resolve_tools(["k8s_query_prometheus"])
|
|
assert [ts.prefix for ts in toolsets] == ["k8s"]
|
|
assert grafana.filtered_args == []
|
|
|
|
_, toolsets, _ = _resolve_tools(
|
|
["k8s_query_prometheus", "grafana_query_prometheus"]
|
|
)
|
|
assert sorted(ts.prefix for ts in toolsets) == ["grafana", "k8s"]
|
|
|
|
def test_unqualified_mcp_tool_name_is_unavailable(self, monkeypatch):
|
|
"""A bare MCP tool name resolves to no server."""
|
|
import oterm.log
|
|
|
|
server = _FakeServer()
|
|
monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", [])
|
|
monkeypatch.setattr(
|
|
"oterm.app.widgets.chat.mcp_tool_meta",
|
|
{"oracle": [{"name": "ask_oracle", "description": ""}]},
|
|
)
|
|
monkeypatch.setattr("oterm.app.widgets.chat.mcp_servers", {"oracle": server})
|
|
before = len(oterm.log.log_lines)
|
|
_, toolsets, _ = _resolve_tools(["ask_oracle"])
|
|
assert toolsets == []
|
|
messages = [msg for _, msg in oterm.log.log_lines[before:]]
|
|
assert any("unavailable tools" in m and "ask_oracle" in m for m in messages)
|
|
|
|
|
|
class TestMcpToolNameCollisions:
|
|
"""Regression for #321, driving real toolsets through a real agent."""
|
|
|
|
@staticmethod
|
|
def _two_servers(monkeypatch):
|
|
from pydantic_ai.toolsets import FunctionToolset
|
|
|
|
def k8s_query_prometheus() -> str:
|
|
return "k8s"
|
|
|
|
def grafana_query_prometheus() -> str:
|
|
return "grafana"
|
|
|
|
k8s = FunctionToolset(id="k8s")
|
|
k8s.add_function(k8s_query_prometheus, name="query_prometheus")
|
|
grafana = FunctionToolset(id="grafana")
|
|
grafana.add_function(grafana_query_prometheus, name="query_prometheus")
|
|
|
|
monkeypatch.setattr("oterm.app.widgets.chat.builtin_tools", [])
|
|
monkeypatch.setattr(
|
|
"oterm.app.widgets.chat.mcp_tool_meta",
|
|
{
|
|
"k8s": [{"name": "query_prometheus", "description": ""}],
|
|
"grafana": [{"name": "query_prometheus", "description": ""}],
|
|
},
|
|
)
|
|
monkeypatch.setattr(
|
|
"oterm.app.widgets.chat.mcp_servers", {"k8s": k8s, "grafana": grafana}
|
|
)
|
|
|
|
async def test_both_servers_expose_the_shared_name_without_conflict(
|
|
self, monkeypatch
|
|
):
|
|
from pydantic_ai.models.test import TestModel
|
|
|
|
from oterm.agent import get_agent
|
|
|
|
self._two_servers(monkeypatch)
|
|
_, toolsets, _ = _resolve_tools(
|
|
["k8s_query_prometheus", "grafana_query_prometheus"]
|
|
)
|
|
agent = get_agent(model="llama3", toolsets=toolsets)
|
|
result = await agent.run("go", model=TestModel())
|
|
|
|
called = {
|
|
part.tool_name
|
|
for message in result.all_messages()
|
|
for part in message.parts
|
|
if hasattr(part, "tool_name")
|
|
}
|
|
assert {"k8s_query_prometheus", "grafana_query_prometheus"} <= called
|
|
|
|
async def test_selecting_one_server_calls_only_that_server(self, monkeypatch):
|
|
from pydantic_ai.models.test import TestModel
|
|
|
|
from oterm.agent import get_agent
|
|
|
|
self._two_servers(monkeypatch)
|
|
_, toolsets, _ = _resolve_tools(["grafana_query_prometheus"])
|
|
agent = get_agent(model="llama3", toolsets=toolsets)
|
|
result = await agent.run("go", model=TestModel())
|
|
|
|
returns = [
|
|
(part.tool_name, part.content)
|
|
for message in result.all_messages()
|
|
for part in message.parts
|
|
if isinstance(part, ToolReturnPart)
|
|
]
|
|
assert returns == [("grafana_query_prometheus", "grafana")]
|
|
|
|
|
|
class TestRebuildAgent:
|
|
def test_success_populates_agent(self):
|
|
cm = ChatModel(model="llama3", provider="ollama")
|
|
container = ChatContainer(chat_model=cm, messages=[])
|
|
assert container.agent is not None
|
|
assert container._agent_error is None
|
|
|
|
def test_selected_capability_reaches_agent(self):
|
|
from pydantic_ai.capabilities import WebSearch
|
|
|
|
cm = ChatModel(model="llama3", provider="ollama", tools=["web_search"])
|
|
container = ChatContainer(chat_model=cm, messages=[])
|
|
assert container.agent is not None
|
|
capabilities = []
|
|
container.agent._root_capability.apply(capabilities.append)
|
|
assert any(isinstance(c, WebSearch) for c in capabilities)
|
|
|
|
def test_failure_captures_error_message(self, app_config):
|
|
cm = ChatModel(model="m", provider="openai-compat/does-not-exist")
|
|
container = ChatContainer(chat_model=cm, messages=[])
|
|
assert container.agent is None
|
|
assert container._agent_error is not None
|
|
assert "not configured" in container._agent_error
|
|
|
|
|
|
class TestStreamAgentGuards:
|
|
async def test_stream_agent_raises_when_agent_is_none(self, app_config):
|
|
cm = ChatModel(model="m", provider="openai-compat/missing")
|
|
container = ChatContainer(chat_model=cm, messages=[])
|
|
assert container.agent is None
|
|
|
|
with pytest.raises(RuntimeError, match="not configured"):
|
|
async for _ in container.stream_agent("hello"):
|
|
pass
|
|
|
|
|
|
class TestResponseTaskErrors:
|
|
async def test_model_http_error_notifies(self, store, chat_model):
|
|
from collections.abc import AsyncIterator
|
|
|
|
from pydantic_ai import Agent
|
|
from pydantic_ai.exceptions import ModelHTTPError
|
|
from pydantic_ai.messages import ModelMessage
|
|
from pydantic_ai.models.function import AgentInfo, FunctionModel
|
|
from textual.app import App, ComposeResult
|
|
|
|
chat_id = await store.save_chat(chat_model)
|
|
chat_model.id = chat_id
|
|
|
|
async def stream_fn(
|
|
messages: list[ModelMessage], info: AgentInfo
|
|
) -> AsyncIterator[str]:
|
|
raise ModelHTTPError(status_code=500, model_name="x", body="boom")
|
|
yield # pragma: no cover
|
|
|
|
class _Host(App):
|
|
def compose(self) -> ComposeResult:
|
|
yield ChatContainer(chat_model=chat_model, messages=[])
|
|
|
|
app = _Host()
|
|
async with app.run_test() as pilot:
|
|
container = app.query_one(ChatContainer)
|
|
container.agent = Agent(FunctionModel(stream_function=stream_fn))
|
|
await container.response_task("hello")
|
|
await pilot.pause()
|
|
assert any(
|
|
"error running your request" in n.message
|
|
for n in list(app._notifications)
|
|
)
|
|
stored = await store.get_messages(chat_id)
|
|
assert stored == []
|
|
assert container.messages == []
|