ed
81d8bce419
Per spec FR2 + Phase 2.1: VendorCapabilities + register + get_capabilities + list_models_for_vendor + the ~40 vendor registrations move into ai_client.py as a region block. Renamed internal _REGISTRY to _VENDOR_REGISTRY to avoid collision with mcp_tool_specs._REGISTRY. Importers (in src/) updated: - src/ai_client.py: removed top-level import; removed 4 local imports of list_models_for_vendor/get_capabilities (symbol now in module namespace) - src/app_controller.py: 2 sites updated to 'from src.ai_client import get_capabilities' - src/gui_2.py: 1 site updated to 'from src.ai_client import VendorCapabilities, get_capabilities' Tests updated: - 8 test_*.py files: changed 'from src.vendor_capabilities import' to 'from src.ai_client import' - tests/test_vendor_capabilities.py: _clean_registry fixture updated to reference src.ai_client._VENDOR_REGISTRY (was src.vendor_capabilities._REGISTRY) Verification: 157 tests pass across the affected files (vendor_capabilities, ai_client_tool_loop variants, openai_compatible, command_palette, diff_viewer, patch_modal, app_controller_result, app_controller_sigint, handle_reset_session, ai_loop_regressions, grok/llama/minimax provider tests).
111 lines
5.3 KiB
Python
111 lines
5.3 KiB
Python
"""Tests for src.ai_client.run_with_tool_loop (shared tool-loop helper).
|
|
|
|
5 Red tests. They verify:
|
|
1. No-tool-call path: returns immediately after one send.
|
|
2. Tool-call dispatch: dispatches via _execute_tool_calls_concurrently and
|
|
continues the loop.
|
|
3. Max-rounds safety: bails out after MAX_TOOL_ROUNDS + 2 iterations.
|
|
4. History append: appends an assistant message to the caller's history.
|
|
5. Error tolerance: continues even if a tool errors.
|
|
|
|
The helper lives in src.ai_client (per the AGENTS.md HARD RULE: no new
|
|
src/<thing>.py files). The tests patch src.ai_client.send_openai_compatible
|
|
because that's the symbol the function uses internally.
|
|
"""
|
|
from __future__ import annotations
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
from src.result_types import Result
|
|
from src.openai_compatible import NormalizedResponse, OpenAICompatibleRequest
|
|
from src.openai_schemas import UsageStats
|
|
from src.ai_client import run_with_tool_loop
|
|
from src.ai_client import VendorCapabilities
|
|
|
|
@pytest.fixture
|
|
def caps() -> VendorCapabilities:
|
|
return VendorCapabilities(vendor="test", model="test-model", tool_calling=True, context_window=8192)
|
|
|
|
def _make_normalized_response(text: str = "ok", tool_calls: list[dict[str, Any]] | None = None) -> Result[NormalizedResponse]:
|
|
return Result(data=NormalizedResponse(
|
|
text=text, tool_calls=tool_calls or [],
|
|
usage=UsageStats(input_tokens=10, output_tokens=5, cache_read_tokens=0, cache_creation_tokens=0),
|
|
raw_response=None,
|
|
))
|
|
|
|
def test_run_with_tool_loop_no_tool_calls_returns_immediately(caps: VendorCapabilities) -> None:
|
|
client = MagicMock()
|
|
with patch("src.openai_compatible.send_openai_compatible", return_value=_make_normalized_response("hello")) as call:
|
|
result = run_with_tool_loop(
|
|
client, OpenAICompatibleRequest(messages=[{"role": "user", "content": "x"}], model="m"),
|
|
capabilities=caps,
|
|
pre_tool_callback=None, qa_callback=None, patch_callback=None,
|
|
base_dir=".", vendor_name="test", history_lock=None, history=None,
|
|
)
|
|
assert result == "hello"
|
|
assert call.call_count == 1
|
|
|
|
def test_run_with_tool_loop_dispatches_tool_calls(caps: VendorCapabilities) -> None:
|
|
client = MagicMock()
|
|
tool_response = _make_normalized_response(
|
|
"first response", tool_calls=[{"id": "c1", "type": "function", "function": {"name": "read_file", "arguments": "{}"}}]
|
|
)
|
|
final_response = _make_normalized_response("after tool")
|
|
with patch("src.openai_compatible.send_openai_compatible", side_effect=[tool_response, final_response]) as call, \
|
|
patch("src.ai_client._execute_tool_calls_concurrently", return_value=[("read_file", "c1", "result", "")]) as dispatch:
|
|
result = run_with_tool_loop(
|
|
client, OpenAICompatibleRequest(messages=[{"role": "user", "content": "x"}], model="m"),
|
|
capabilities=caps,
|
|
pre_tool_callback=None, qa_callback=None, patch_callback=None,
|
|
base_dir=".", vendor_name="test", history_lock=None, history=None,
|
|
)
|
|
assert result == "after tool"
|
|
assert call.call_count == 2
|
|
assert dispatch.call_count == 1
|
|
|
|
def test_run_with_tool_loop_respects_max_rounds(caps: VendorCapabilities) -> None:
|
|
client = MagicMock()
|
|
infinite_tool_response = _make_normalized_response(
|
|
"loop", tool_calls=[{"id": "c1", "type": "function", "function": {"name": "noop", "arguments": "{}"}}]
|
|
)
|
|
with patch("src.openai_compatible.send_openai_compatible", return_value=infinite_tool_response), \
|
|
patch("src.ai_client._execute_tool_calls_concurrently", return_value=[("noop", "c1", "result", "")]):
|
|
result = run_with_tool_loop(
|
|
client, OpenAICompatibleRequest(messages=[{"role": "user", "content": "x"}], model="m"),
|
|
capabilities=caps,
|
|
pre_tool_callback=None, qa_callback=None, patch_callback=None,
|
|
base_dir=".", vendor_name="test", history_lock=None, history=None,
|
|
)
|
|
assert result == "loop"
|
|
|
|
def test_run_with_tool_loop_appends_to_history(caps: VendorCapabilities) -> None:
|
|
client = MagicMock()
|
|
history: list[dict[str, Any]] = []
|
|
history_lock = MagicMock()
|
|
history_lock.__enter__ = MagicMock(return_value=history_lock)
|
|
history_lock.__exit__ = MagicMock(return_value=False)
|
|
with patch("src.openai_compatible.send_openai_compatible", return_value=_make_normalized_response("hi")):
|
|
run_with_tool_loop(
|
|
client, OpenAICompatibleRequest(messages=[{"role": "user", "content": "x"}], model="m"),
|
|
capabilities=caps,
|
|
pre_tool_callback=None, qa_callback=None, patch_callback=None,
|
|
base_dir=".", vendor_name="test", history_lock=history_lock, history=history,
|
|
)
|
|
assert any(msg.get("role") == "assistant" and msg.get("content") == "hi" for msg in history)
|
|
|
|
def test_run_with_tool_loop_does_not_crash_on_tool_error(caps: VendorCapabilities) -> None:
|
|
client = MagicMock()
|
|
tool_response = _make_normalized_response(
|
|
"err", tool_calls=[{"id": "c1", "type": "function", "function": {"name": "fail", "arguments": "{}"}}]
|
|
)
|
|
final_response = _make_normalized_response("recovered")
|
|
with patch("src.openai_compatible.send_openai_compatible", side_effect=[tool_response, final_response]), \
|
|
patch("src.ai_client._execute_tool_calls_concurrently", return_value=[("fail", "c1", "", "ToolExecutionError")]):
|
|
result = run_with_tool_loop(
|
|
client, OpenAICompatibleRequest(messages=[{"role": "user", "content": "x"}], model="m"),
|
|
capabilities=caps,
|
|
pre_tool_callback=None, qa_callback=None, patch_callback=None,
|
|
base_dir=".", vendor_name="test", history_lock=None, history=None,
|
|
)
|
|
assert result == "recovered"
|