manual_slop/tests/test_ai_client_tool_loop.py

"""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.openai_compatible import NormalizedResponse, OpenAICompatibleRequest
from src.ai_client import run_with_tool_loop
from src.vendor_capabilities 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) -> NormalizedResponse:
 return NormalizedResponse(
  text=text, tool_calls=tool_calls or [],
  usage_input_tokens=10, usage_output_tokens=5,
  usage_cache_read_tokens=0, usage_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.ai_client.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.ai_client.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.ai_client.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.ai_client.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.ai_client.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"