import json
import asyncio
import logging
import subprocess
from collections.abc import AsyncIterator
from html.parser import HTMLParser
from pathlib import Path
import httpx
import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError
from agent_lab.application.contracts import AgentParams, DebugRunRequest, EventAgentParams
from agent_lab.application.queues import RuntimeQueues
from agent_lab.application.runtime import DebugRuntime
from agent_lab.domain.events import ToolCallEvent
from agent_lab.domain.messages import ChatMessage, StreamItem, TokenUsage
from agent_lab.infrastructure.chat_client import OpenAICompatibleChatClient
from agent_lab.infrastructure.sqlite_store import SQLiteSessionStore
from agent_lab.presentation.web import create_app
from agent_lab.settings import Settings
def _css_rule(css: str, selector: str) -> str:
start = css.index(f"{selector} {{")
end = css.index("}", start)
return css[start:end]
class _HtmlNode:
def __init__(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
self.tag = tag
self.attrs = dict(attrs)
self.children: list["_HtmlNode"] = []
self.text_parts: list[str] = []
@property
def classes(self) -> set[str]:
return set(self.attrs.get("class", "").split())
def descendants(self) -> list["_HtmlNode"]:
nodes: list["_HtmlNode"] = []
for child in self.children:
nodes.append(child)
nodes.extend(child.descendants())
return nodes
def text_content(self) -> str:
return "".join(
[*self.text_parts, *(child.text_content() for child in self.children)]
)
class _HtmlTreeParser(HTMLParser):
_void_tags = {
"area",
"base",
"br",
"col",
"embed",
"hr",
"img",
"input",
"link",
"meta",
"param",
"source",
"track",
"wbr",
}
def __init__(self) -> None:
super().__init__()
self.root = _HtmlNode("document", [])
self.stack = [self.root]
def handle_starttag(
self,
tag: str,
attrs: list[tuple[str, str | None]],
) -> None:
node = _HtmlNode(tag, attrs)
self.stack[-1].children.append(node)
if tag not in self._void_tags:
self.stack.append(node)
def handle_startendtag(
self,
tag: str,
attrs: list[tuple[str, str | None]],
) -> None:
self.stack[-1].children.append(_HtmlNode(tag, attrs))
def handle_endtag(self, tag: str) -> None:
for index in range(len(self.stack) - 1, 0, -1):
if self.stack[index].tag == tag:
del self.stack[index:]
return
def handle_data(self, data: str) -> None:
self.stack[-1].text_parts.append(data)
def _parse_html(html: str) -> _HtmlNode:
parser = _HtmlTreeParser()
parser.feed(html)
return parser.root
def _html_node_by_id(root: _HtmlNode, element_id: str) -> _HtmlNode:
node = next(
(
node
for node in [root, *root.descendants()]
if node.attrs.get("id") == element_id
),
None,
)
assert node is not None, f"missing HTML element #{element_id}"
return node
def _run_node_json(source: str) -> object:
completed = subprocess.run(
["node", "--input-type=module", "-e", source],
capture_output=True,
text=True,
check=False,
)
assert completed.returncode == 0, completed.stderr
return json.loads(completed.stdout)
def _console_ui_source(js: str) -> str:
assert 'const CONSOLE_UI_STORAGE_KEY = "agent-lab-console-ui-v1";' in js
for function_name in [
"readConsoleUiState",
"writeConsoleUiState",
"setControlSectionExpanded",
"setDebugDrawerOpen",
"initializeConsoleUi",
"updateTopbarStatus",
]:
assert f"function {function_name}(" in js
return js[
js.index("function readConsoleUiState(") :
js.index("function openChatConfig(")
]
def _run_load_session_replay_scenario(scenario: str) -> object:
js = Path("src/agent_lab/presentation/static/app.js").read_text()
load_replay = js[
js.index("async function loadSessionReplay(") :
js.index("function setSessionStatus(")
]
harness = """
let currentSessionId = null;
let sessionReplayLoadToken = 0;
const sessionTitle = { value: "" };
const sessionList = { value: "" };
const mutations = {
workspace: [],
transcript: [],
audit: [],
usage: [],
status: [],
};
const pending = new Map();
function fetchJson(url) {
let resolve;
let reject;
const promise = new Promise((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
pending.set(url, { resolve, reject });
return promise;
}
function resolveReplay(sessionId) {
pending.get(`/api/sessions/${sessionId}`).resolve({
title: `${sessionId} title`,
config: { marker: sessionId },
});
pending.get(`/api/sessions/${sessionId}/messages`).resolve([
{ content: sessionId },
]);
pending.get(`/api/sessions/${sessionId}/audit`).resolve([
{ event: sessionId },
]);
pending.get(`/api/sessions/${sessionId}/usage`).resolve({
session: { marker: sessionId },
});
}
function restoreWorkspaceSnapshot(config) {
mutations.workspace.push(config.marker);
}
function renderReplayMessages(messages) {
mutations.transcript.push(messages[0].content);
}
function renderAuditReplay(audit) {
mutations.audit.push(audit[0].event);
}
function renderSessionUsage(usage) {
mutations.usage.push(usage.session.marker);
}
function setSessionStatus(status) {
mutations.status.push(status);
}
function isSocketOpen() {
return false;
}
"""
return _run_node_json("\n".join([harness, load_replay, scenario]))
class FakeRuntime:
def __init__(self) -> None:
self.requests: list[DebugRunRequest] = []
async def run(self, request: DebugRunRequest) -> AsyncIterator[dict]:
self.requests.append(request)
yield {"type": "session_started"}
yield {"type": "message_delta", "content": "hello"}
yield {"type": "done"}
class QueueAwareRuntime:
def __init__(self) -> None:
self.requests: list[DebugRunRequest] = []
self.queues: RuntimeQueues | None = None
self.task: asyncio.Task | None = None
def start(self, request: DebugRunRequest) -> RuntimeQueues:
self.requests.append(request)
self.queues = RuntimeQueues()
self.task = asyncio.create_task(self._run())
return self.queues
async def _run(self) -> None:
assert self.queues is not None
await self.queues.output.put({"type": "session_started"})
message = await self.queues.input.get()
await self.queues.output.put(
{"type": "message_delta", "content": message.content}
)
await self.queues.output.put({"type": "done"})
async def aclose(self) -> None:
if self.task is not None and not self.task.done():
self.task.cancel()
await asyncio.gather(self.task, return_exceptions=True)
class PersistentSessionRuntime:
def __init__(self) -> None:
self.requests: list[DebugRunRequest] = []
self.queues: RuntimeQueues | None = None
self.task: asyncio.Task | None = None
def start_session(self, request: DebugRunRequest) -> RuntimeQueues:
self.requests.append(request)
self.queues = RuntimeQueues()
self.task = asyncio.create_task(self._run(request))
return self.queues
async def _run(self, request: DebugRunRequest) -> None:
assert self.queues is not None
await self.queues.output.put({"type": "session_started"})
await self._emit_turn(1, request.user_message)
turn_index = 1
while True:
message = await self.queues.input.get()
turn_index += 1
await self._emit_turn(turn_index, message.content)
async def _emit_turn(self, turn_index: int, content: str) -> None:
assert self.queues is not None
await self.queues.output.put({"type": "turn_started", "turn_index": turn_index})
await self.queues.output.put({"type": "message_delta", "content": content})
await self.queues.output.put(
{"type": "turn_completed", "turn_index": turn_index}
)
async def aclose(self) -> None:
if self.task is not None and not self.task.done():
self.task.cancel()
await asyncio.gather(self.task, return_exceptions=True)
def _request_payload() -> dict:
return {
"user_message": "debug this",
"system_prompts": ["You are a debugger."],
"pre_messages": [{"role": "user", "content": "previous turn"}],
"chat_agent": {
"model": "fake-model",
"temperature": 0.1,
"max_tokens": 200,
},
"event_agent": {
"enabled_tools": ["handoff_note"],
"max_event_loops": 2,
},
}
def _event_tool_call_from_tools(
tools: list[dict],
messages: list[ChatMessage],
) -> StreamItem:
tool_name = tools[0]["function"]["name"]
content = ""
for role in ("assistant", "user"):
content = next(
(
message.content
for message in reversed(messages)
if message.role == role and message.content.strip()
),
"",
)
if content:
break
arguments = {
"message": content,
"query": content,
"title": content,
}
return StreamItem.provider_tool_call(
ToolCallEvent(
id="event_agent_call_1",
name=tool_name,
arguments=arguments,
raw_arguments=json.dumps(arguments),
)
)
def test_agent_params_defaults_include_extra_body_and_event_agent_defaults_to_one_round():
chat_params = AgentParams()
event_params = EventAgentParams()
assert chat_params.extra_body == {
"thinking": {"type": "disabled"},
"enable_search": False,
"search_options": {"forced_search": False},
}
assert event_params.extra_body == chat_params.extra_body
assert event_params.max_event_loops == 1
assert event_params.max_parallel_events == 4
assert event_params.batch_timeout_seconds == 15.0
assert event_params.system_prompt == ""
@pytest.mark.parametrize("value", [0, 17])
def test_event_agent_params_rejects_out_of_range_parallelism(value: int):
with pytest.raises(ValidationError, match="max_parallel_events"):
EventAgentParams(max_parallel_events=value)
@pytest.mark.parametrize("value", [0, -1, 121])
def test_event_agent_params_rejects_invalid_batch_timeout(value: float):
with pytest.raises(ValidationError, match="batch_timeout_seconds"):
EventAgentParams(batch_timeout_seconds=value)
def test_debug_run_request_defaults_to_dual_agent_tool_invocation_mode():
request = DebugRunRequest.model_validate(_request_payload())
assert request.tool_invocation_mode == "dual_agent"
@pytest.mark.parametrize("mode", ["dual_agent", "chat_agent_tools"])
def test_debug_run_request_accepts_supported_tool_invocation_modes(mode: str):
payload = _request_payload()
payload["tool_invocation_mode"] = mode
request = DebugRunRequest.model_validate(payload)
assert request.tool_invocation_mode == mode
def test_debug_run_request_rejects_unknown_tool_invocation_mode():
payload = _request_payload()
payload["tool_invocation_mode"] = "unknown"
with pytest.raises(ValidationError, match="tool_invocation_mode"):
DebugRunRequest.model_validate(payload)
def test_health_returns_ok():
app = create_app(runtime_factory=FakeRuntime)
client = TestClient(app)
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_api_tools_returns_default_tool_catalog():
app = create_app(runtime_factory=FakeRuntime)
client = TestClient(app)
response = client.get("/api/tools")
assert response.status_code == 200
tools = response.json()
assert [tool["name"] for tool in tools] == [
"handoff_note",
"mock_search",
"mock_ticket",
"session.terminate",
"device.volume.adjust",
"calendar.schedule.create",
"knowledge.web.search",
]
assert tools[0]["parameters"]["required"] == ["message"]
def test_session_api_creates_lists_and_returns_empty_replay_data(tmp_path):
settings = Settings(database_path=str(tmp_path / "agent_lab.sqlite3"))
app = create_app(settings=settings, runtime_factory=FakeRuntime)
client = TestClient(app)
created = client.post(
"/api/sessions",
json={
"title": "Debug session",
"config": {"chat_agent": {"model": "chat-model"}},
},
)
assert created.status_code == 200
session_id = created.json()["id"]
assert created.json()["title"] == "Debug session"
sessions = client.get("/api/sessions")
assert sessions.status_code == 200
assert sessions.json()[0]["id"] == session_id
assert sessions.json()[0]["turn_count"] == 0
detail = client.get(f"/api/sessions/{session_id}")
assert detail.status_code == 200
assert detail.json()["config"]["chat_agent"]["model"] == "chat-model"
assert client.get(f"/api/sessions/{session_id}/messages").json() == []
assert client.get(f"/api/sessions/{session_id}/audit").json() == []
usage = client.get(f"/api/sessions/{session_id}/usage").json()
assert usage == {
"calls": [],
"turns": [],
"session": {
"prompt_tokens": 0,
"completion_tokens": 0,
"total_tokens": 0,
"cached_tokens": 0,
"elapsed_ms": 0,
"call_count": 0,
"fallback_count": 0,
"tool_count": 0,
"turn_wall_time_ms": None,
},
}
def test_session_api_returns_persisted_replay_data(tmp_path):
database_path = tmp_path / "agent_lab.sqlite3"
settings = Settings(database_path=str(database_path))
app = create_app(settings=settings, runtime_factory=FakeRuntime)
client = TestClient(app)
session_id = client.post("/api/sessions", json={"title": "Replay"}).json()["id"]
store = SQLiteSessionStore(database_path)
store.start_turn(session_id, turn_index=1, user_message="debug this")
store.append_message(
session_id,
turn_index=1,
message=ChatMessage(role="user", content="debug this"),
)
store.append_message(
session_id,
turn_index=1,
message=ChatMessage(
role="assistant",
content="answer",
tool_calls=[
ToolCallEvent(
id="call-1",
name="mock_search",
arguments={"query": "docs"},
raw_arguments='{"query":"docs"}',
)
],
),
)
store.append_message(
session_id,
turn_index=1,
message=ChatMessage(
role="tool",
content='{"results":[]}',
name="mock_search",
tool_call_id="call-1",
),
)
store.append_audit(
session_id,
event="chat_agent_request",
details={"model": "chat-model"},
turn_index=1,
round_index=1,
)
store.append_usage(
session_id,
turn_index=1,
round_index=1,
usage=TokenUsage(prompt_tokens=1, completion_tokens=2, total_tokens=3),
ttft_ms=10,
elapsed_ms=20,
)
messages = client.get(f"/api/sessions/{session_id}/messages").json()
assert [message["content"] for message in messages] == [
"debug this",
"answer",
'{"results":[]}',
]
assert messages[0]["tool_calls"] == []
assert messages[1]["tool_calls"] == [
{
"id": "call-1",
"name": "mock_search",
"arguments": {"query": "docs"},
"raw_arguments": '{"query":"docs"}',
}
]
assert messages[2]["tool_call_id"] == "call-1"
assert messages[2]["tool_calls"] == []
assert client.get(f"/api/sessions/{session_id}/audit").json()[0]["details"] == {
"model": "chat-model"
}
usage = client.get(f"/api/sessions/{session_id}/usage").json()
assert usage["calls"][0]["total_tokens"] == 3
assert usage["turns"][0]["total_tokens"] == 3
assert usage["session"]["total_tokens"] == 3
def test_default_websocket_runtime_factory_accepts_session_store(tmp_path):
settings = Settings(database_path=str(tmp_path / "agent_lab.sqlite3"))
app = create_app(settings=settings)
client = TestClient(app)
with client.websocket_connect("/ws/debug") as websocket:
websocket.send_json({"chat_agent": {"model": "fake-model"}})
message = websocket.receive_json()
assert message["type"] == "error"
assert "user_message" in message["message"]
def test_websocket_debug_streams_runtime_messages():
runtime = FakeRuntime()
app = create_app(runtime_factory=lambda: runtime)
client = TestClient(app)
with client.websocket_connect("/ws/debug") as websocket:
websocket.send_json(_request_payload())
assert websocket.receive_json() == {"type": "session_started"}
assert websocket.receive_json() == {
"type": "message_delta",
"content": "hello",
}
assert websocket.receive_json() == {"type": "done"}
assert runtime.requests[0].user_message == "debug this"
assert runtime.requests[0].pre_messages[0].content == "previous turn"
def test_websocket_debug_logs_session_lifecycle(caplog):
runtime = FakeRuntime()
app = create_app(runtime_factory=lambda: runtime)
client = TestClient(app)
with caplog.at_level(logging.INFO, logger="agent_lab.presentation.web"):
with client.websocket_connect("/ws/debug") as websocket:
websocket.send_json(_request_payload())
assert websocket.receive_json() == {"type": "session_started"}
assert websocket.receive_json() == {
"type": "message_delta",
"content": "hello",
}
assert websocket.receive_json() == {"type": "done"}
assert "websocket session accepted" in caplog.text
assert "websocket request accepted" in caplog.text
assert "websocket session closed" in caplog.text
def test_websocket_debug_enqueues_user_messages_during_running_session():
runtime = QueueAwareRuntime()
app = create_app(runtime_factory=lambda: runtime)
client = TestClient(app)
with client.websocket_connect("/ws/debug") as websocket:
websocket.send_json(_request_payload())
assert websocket.receive_json() == {"type": "session_started"}
websocket.send_json({"type": "user_message", "content": "follow-up"})
assert websocket.receive_json() == {
"type": "message_delta",
"content": "follow-up",
}
assert websocket.receive_json() == {"type": "done"}
assert runtime.requests[0].user_message == "debug this"
def test_websocket_debug_keeps_session_open_for_multiple_turns():
runtime = PersistentSessionRuntime()
app = create_app(runtime_factory=lambda: runtime)
client = TestClient(app)
with client.websocket_connect("/ws/debug") as websocket:
websocket.send_json(_request_payload())
assert websocket.receive_json() == {"type": "session_started"}
assert websocket.receive_json() == {"type": "turn_started", "turn_index": 1}
assert websocket.receive_json() == {
"type": "message_delta",
"content": "debug this",
}
assert websocket.receive_json() == {"type": "turn_completed", "turn_index": 1}
websocket.send_json({"type": "user_message", "content": "follow-up"})
assert websocket.receive_json() == {"type": "turn_started", "turn_index": 2}
assert websocket.receive_json() == {
"type": "message_delta",
"content": "follow-up",
}
assert websocket.receive_json() == {"type": "turn_completed", "turn_index": 2}
websocket.close()
assert runtime.requests[0].user_message == "debug this"
def test_websocket_debug_sends_error_for_invalid_request():
app = create_app(runtime_factory=FakeRuntime)
client = TestClient(app)
with client.websocket_connect("/ws/debug") as websocket:
websocket.send_json({"chat_agent": {"model": "fake-model"}})
message = websocket.receive_json()
assert message["type"] == "error"
assert "user_message" in message["message"]
def test_debug_run_request_rejects_invalid_pre_message_role():
payload = _request_payload()
payload["pre_messages"] = [{"role": "developer", "content": "invalid"}]
with pytest.raises(ValidationError) as exc_info:
DebugRunRequest.model_validate(payload)
assert "role" in str(exc_info.value)
def test_debug_run_request_rejects_tool_pre_messages():
payload = _request_payload()
payload["pre_messages"] = [
{
"role": "tool",
"content": "orphan result",
"tool_call_id": "call_1",
}
]
with pytest.raises(ValidationError, match="pre_messages cannot include tool messages"):
DebugRunRequest.model_validate(payload)
def test_debug_run_request_rejects_assistant_tool_calls_in_pre_messages():
payload = _request_payload()
payload["pre_messages"] = [
{
"role": "assistant",
"content": "I will inspect this.",
"tool_calls": [
{
"id": "call_1",
"name": "mock_search",
"arguments": {"query": "latency"},
"raw_arguments": '{"query":"latency"}',
}
],
}
]
with pytest.raises(
ValidationError,
match="pre_messages cannot include assistant tool calls",
):
DebugRunRequest.model_validate(payload)
def test_chat_message_allows_internal_tool_replies():
message = ChatMessage(
role="tool",
content='{"message":"handled"}',
name="handoff_note",
tool_call_id="call_1",
)
assert message.role == "tool"
assert message.tool_call_id == "call_1"
def test_chat_message_preserves_assistant_tool_calls():
message = ChatMessage(
role="assistant",
content="I will inspect both sources.",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={"query": "latency docs"},
raw_arguments='{"query":"latency docs"}',
),
ToolCallEvent(
id="call_2",
name="handoff_note",
arguments={"message": "inspect provider behavior"},
raw_arguments='{"message":"inspect provider behavior"}',
),
],
)
assert message.model_dump()["tool_calls"] == [
{
"id": "call_1",
"name": "mock_search",
"arguments": {"query": "latency docs"},
"raw_arguments": '{"query":"latency docs"}',
},
{
"id": "call_2",
"name": "handoff_note",
"arguments": {"message": "inspect provider behavior"},
"raw_arguments": '{"message":"inspect provider behavior"}',
},
]
def test_chat_message_rejects_tool_calls_on_non_assistant_role():
with pytest.raises(ValidationError, match="tool_calls require assistant role"):
ChatMessage(
role="user",
content="invalid",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={},
raw_arguments="{}",
)
],
)
def test_chat_message_rejects_tool_role_without_tool_call_id():
with pytest.raises(ValidationError, match="tool messages require tool_call_id"):
ChatMessage(role="tool", content="orphan result")
@pytest.mark.parametrize("role", ["system", "user", "assistant"])
def test_chat_message_rejects_tool_call_id_on_non_tool_roles(role: str):
with pytest.raises(ValidationError, match="tool_call_id requires tool role"):
ChatMessage(role=role, content="invalid", tool_call_id="call_1")
def test_chat_message_rejects_duplicate_assistant_tool_call_ids():
with pytest.raises(ValidationError, match="duplicate assistant tool-call ID: call_1"):
ChatMessage(
role="assistant",
content="checking twice",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={"query": "first"},
raw_arguments='{"query":"first"}',
),
ToolCallEvent(
id="call_1",
name="handoff_note",
arguments={"message": "second"},
raw_arguments='{"message":"second"}',
),
],
)
class HistoryCapturingChatClient:
def __init__(self) -> None:
self.calls = 0
self.second_call_messages: list[ChatMessage] = []
async def stream_chat(
self,
messages: list[ChatMessage],
tools: list[dict],
params: AgentParams,
tool_choice: dict | None = None,
) -> AsyncIterator[StreamItem]:
if tools:
yield _event_tool_call_from_tools(tools, messages)
return
self.calls += 1
if self.calls == 1:
yield StreamItem.message_delta("Need event help.")
yield StreamItem.text_event(
ToolCallEvent(
id="call_1",
name="handoff_note",
arguments={},
raw_arguments="{}",
)
)
return
self.second_call_messages = list(messages)
yield StreamItem.message_delta("Final answer.")
@pytest.mark.asyncio
async def test_runtime_appends_event_summary_without_tool_reply_history():
request = DebugRunRequest(
user_message="debug this",
system_prompts=["You are a debugger."],
pre_messages=[],
chat_agent=AgentParams(model="fake-model", temperature=0.1, max_tokens=200),
event_agent=EventAgentParams(enabled_tools=["handoff_note"], max_event_loops=2),
)
client = HistoryCapturingChatClient()
runtime = DebugRuntime(client)
outputs = [message async for message in runtime.run(request)]
assert client.calls == 2
assert [message.role for message in client.second_call_messages] == [
"system",
"system",
"user",
"assistant",
"user",
]
assert "Available events:" in client.second_call_messages[1].content
assert client.second_call_messages[3].content == "Need event help."
assert not any(message.role == "tool" for message in client.second_call_messages)
assert client.second_call_messages[4].content.startswith("EventAgent results:\n")
assert client.second_call_messages[4].name == "event_agent"
assert outputs[-1] == {"type": "done"}
@pytest.mark.asyncio
async def test_openai_chat_client_streams_sse_chunks_through_parser():
requests: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
payload = json.loads(request.content)
assert payload["stream"] is True
assert payload["model"] == "model-x"
assert payload["messages"] == [{"role": "user", "content": "hi"}]
assert payload["tools"][0]["function"]["name"] == "handoff_note"
assert "tool_choice" not in payload
assert payload["temperature"] == 0.3
assert payload["max_tokens"] == 50
assert request.headers["authorization"] == "Bearer test-key"
return httpx.Response(
200,
content=(
b'data: {"choices":[{"delta":{"content":"hi"},"finish_reason":null}]}\n\n'
b"data: [DONE]\n\n"
),
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://llm.test/v1",
) as http_client:
client = OpenAICompatibleChatClient(
api_key="test-key",
base_url="https://llm.test/v1",
default_model="default-model",
request_timeout_seconds=5,
http_client=http_client,
)
items = [
item
async for item in client.stream_chat(
messages=[ChatMessage(role="user", content="hi")],
tools=[
{
"type": "function",
"function": {"name": "handoff_note", "parameters": {}},
}
],
params=AgentParams(model="model-x", temperature=0.3, max_tokens=50),
)
]
assert requests[0].url.path == "/v1/chat/completions"
assert [item.content for item in items if item.kind == "message_delta"] == ["hi"]
assert [item.raw_chunk for item in items if item.kind == "raw_chunk"] == [
{"choices": [{"delta": {"content": "hi"}, "finish_reason": None}]}
]
@pytest.mark.asyncio
async def test_openai_chat_client_serializes_explicit_tool_choice():
captured_payloads: list[dict] = []
forced_choice = {
"type": "function",
"function": {"name": "handoff_note"},
}
def handler(request: httpx.Request) -> httpx.Response:
captured_payloads.append(json.loads(request.content))
return httpx.Response(200, content=b"data: [DONE]\n\n")
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://llm.test/v1",
) as http_client:
client = OpenAICompatibleChatClient(
api_key="",
base_url="https://llm.test/v1",
default_model="provider-default",
request_timeout_seconds=5,
http_client=http_client,
)
[
item
async for item in client.stream_chat(
messages=[ChatMessage(role="user", content="prepare handoff")],
tools=[
{
"type": "function",
"function": {"name": "handoff_note", "parameters": {}},
}
],
params=AgentParams(model="provider-default"),
tool_choice=forced_choice,
)
]
assert captured_payloads[0]["tool_choice"] == forced_choice
@pytest.mark.asyncio
async def test_openai_chat_client_flushes_one_provider_tool_call_at_done():
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
content=(
b'data: {"choices":[{"delta":{"tool_calls":[{"index":0,'
b'"id":"call_1","function":{"name":"mock_search",'
b'"arguments":"{\\"query\\":\\""}}]},"finish_reason":null}]}\n\n'
b'data: {"choices":[{"delta":{"tool_calls":[{"index":0,'
b'"function":{"arguments":"latency docs\\"}"}}]},'
b'"finish_reason":null}]}\n\n'
b"data: [DONE]\n\n"
),
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://llm.test/v1",
) as http_client:
client = OpenAICompatibleChatClient(
api_key="",
base_url="https://llm.test/v1",
default_model="provider-default",
request_timeout_seconds=5,
http_client=http_client,
)
items = [
item
async for item in client.stream_chat(
messages=[ChatMessage(role="user", content="find docs")],
tools=[
{
"type": "function",
"function": {"name": "mock_search", "parameters": {}},
}
],
params=AgentParams(model="provider-default"),
)
]
provider_calls = [
item.event for item in items if item.kind == "provider_tool_call"
]
assert provider_calls == [
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={"query": "latency docs"},
raw_arguments='{"query":"latency docs"}',
)
]
@pytest.mark.asyncio
async def test_openai_chat_client_uses_default_model_when_request_model_is_blank():
captured_payloads: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
captured_payloads.append(json.loads(request.content))
return httpx.Response(
200,
content=b'data: {"choices":[{"delta":{"content":"hi"},"finish_reason":null}]}\n\n',
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://llm.test/v1",
) as http_client:
client = OpenAICompatibleChatClient(
api_key="",
base_url="https://llm.test/v1",
default_model="provider-default",
request_timeout_seconds=5,
http_client=http_client,
)
[
item
async for item in client.stream_chat(
messages=[ChatMessage(role="user", content="hi")],
tools=[],
params=AgentParams(model=" ", temperature=0.3, max_tokens=50),
)
]
assert captured_payloads[0]["model"] == "provider-default"
@pytest.mark.asyncio
async def test_openai_chat_client_serializes_provider_tool_transcript():
captured_payloads: list[dict] = []
def handler(request: httpx.Request) -> httpx.Response:
captured_payloads.append(json.loads(request.content))
return httpx.Response(
200,
content=b'data: {"choices":[{"delta":{"content":"done"},"finish_reason":null}]}\n\n',
)
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://llm.test/v1",
) as http_client:
client = OpenAICompatibleChatClient(
api_key="",
base_url="https://llm.test/v1",
default_model="provider-default",
request_timeout_seconds=5,
http_client=http_client,
)
[
item
async for item in client.stream_chat(
messages=[
ChatMessage(role="user", content="inspect both"),
ChatMessage(
role="assistant",
content="I will inspect both sources.",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={"query": "latency docs"},
raw_arguments='{"query":"latency docs"}',
),
ToolCallEvent(
id="call_2",
name="handoff_note",
arguments={"message": "inspect provider behavior"},
raw_arguments='{"message":"inspect provider behavior"}',
),
],
),
ChatMessage(
role="tool",
content='{"results":[]}',
name="mock_search",
tool_call_id="call_1",
),
ChatMessage(
role="tool",
content='{"message":"handled"}',
name="handoff_note",
tool_call_id="call_2",
),
ChatMessage(role="user", content="continue"),
],
tools=[],
params=AgentParams(
model="provider-default",
temperature=0.3,
max_tokens=50,
),
)
]
assert captured_payloads[0]["messages"] == [
{"role": "user", "content": "inspect both"},
{
"role": "assistant",
"content": "I will inspect both sources.",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "mock_search",
"arguments": '{"query":"latency docs"}',
},
},
{
"id": "call_2",
"type": "function",
"function": {
"name": "handoff_note",
"arguments": '{"message":"inspect provider behavior"}',
},
},
],
},
{
"role": "tool",
"content": '{"results":[]}',
"tool_call_id": "call_1",
},
{
"role": "tool",
"content": '{"message":"handled"}',
"tool_call_id": "call_2",
},
{"role": "user", "content": "continue"},
]
async def _assert_tool_transcript_rejected(
messages: list[ChatMessage],
error_match: str,
) -> None:
network_called = False
def handler(request: httpx.Request) -> httpx.Response:
nonlocal network_called
network_called = True
return httpx.Response(200)
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://llm.test/v1",
) as http_client:
client = OpenAICompatibleChatClient(
api_key="",
base_url="https://llm.test/v1",
default_model="provider-default",
request_timeout_seconds=5,
http_client=http_client,
)
with pytest.raises(ValueError, match=error_match):
[
item
async for item in client.stream_chat(
messages=messages,
tools=[],
params=AgentParams(model="provider-default"),
)
]
assert network_called is False
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_orphan_tool_reply_before_network():
await _assert_tool_transcript_rejected(
[
ChatMessage(
role="tool",
content="orphan",
tool_call_id="call_1",
)
],
"orphan tool reply: call_1",
)
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_mismatched_tool_call_id_before_network():
await _assert_tool_transcript_rejected(
[
ChatMessage(
role="assistant",
content="",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={},
raw_arguments="{}",
)
],
),
ChatMessage(role="tool", content="wrong", tool_call_id="call_2"),
],
"mismatched tool_call_id: call_2",
)
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_duplicate_tool_reply_before_network():
await _assert_tool_transcript_rejected(
[
ChatMessage(
role="assistant",
content="",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={},
raw_arguments="{}",
)
],
),
ChatMessage(role="tool", content="first", tool_call_id="call_1"),
ChatMessage(role="tool", content="duplicate", tool_call_id="call_1"),
],
"duplicate tool reply: call_1",
)
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_non_tool_message_with_unresolved_calls():
await _assert_tool_transcript_rejected(
[
ChatMessage(
role="assistant",
content="checking",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={},
raw_arguments="{}",
),
ToolCallEvent(
id="call_2",
name="handoff_note",
arguments={},
raw_arguments="{}",
),
],
),
ChatMessage(role="tool", content="done", tool_call_id="call_1"),
ChatMessage(role="user", content="continue too early"),
],
"unresolved tool calls before user message: call_2",
)
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_unfinished_tool_calls_at_end():
await _assert_tool_transcript_rejected(
[
ChatMessage(
role="assistant",
content="checking",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={},
raw_arguments="{}",
)
],
)
],
"unresolved tool calls at end: call_1",
)
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_reused_tool_call_id_across_transcript():
await _assert_tool_transcript_rejected(
[
ChatMessage(
role="assistant",
content="first round",
tool_calls=[
ToolCallEvent(
id="call_1",
name="mock_search",
arguments={},
raw_arguments="{}",
)
],
),
ChatMessage(role="tool", content="done", tool_call_id="call_1"),
ChatMessage(
role="assistant",
content="second round",
tool_calls=[
ToolCallEvent(
id="call_1",
name="handoff_note",
arguments={},
raw_arguments="{}",
)
],
),
],
"duplicate assistant tool-call ID across transcript: call_1",
)
async def _assert_tool_choice_rejected(
tools: list[dict],
tool_choice: dict,
error_match: str,
) -> None:
network_called = False
def handler(request: httpx.Request) -> httpx.Response:
nonlocal network_called
network_called = True
return httpx.Response(200)
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://llm.test/v1",
) as http_client:
client = OpenAICompatibleChatClient(
api_key="",
base_url="https://llm.test/v1",
default_model="provider-default",
request_timeout_seconds=5,
http_client=http_client,
)
with pytest.raises(ValueError, match=error_match):
[
item
async for item in client.stream_chat(
messages=[ChatMessage(role="user", content="use a tool")],
tools=tools,
params=AgentParams(model="provider-default"),
tool_choice=tool_choice,
)
]
assert network_called is False
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_forced_choice_without_tools():
await _assert_tool_choice_rejected(
tools=[],
tool_choice={
"type": "function",
"function": {"name": "handoff_note"},
},
error_match="forced tool choice requires tools",
)
@pytest.mark.asyncio
async def test_openai_chat_client_rejects_forced_choice_for_unknown_tool():
await _assert_tool_choice_rejected(
tools=[
{
"type": "function",
"function": {"name": "mock_search", "parameters": {}},
}
],
tool_choice={
"type": "function",
"function": {"name": "handoff_note"},
},
error_match="forced tool choice references unknown tool: handoff_note",
)
def test_static_pre_message_role_selector_does_not_offer_tool():
html = Path("src/agent_lab/presentation/static/index.html").read_text()
assert '' not in html
def test_static_tools_ui_is_dynamic_and_not_hardcoded_to_handoff_note_checkbox():
html = Path("src/agent_lab/presentation/static/index.html").read_text()
js = Path("src/agent_lab/presentation/static/app.js").read_text()
assert 'id="tool-handoff-note"' not in html
assert "#tool-handoff-note" not in js
assert 'id="tool-list"' in html
assert '"/api/tools"' in js
assert "fetch" in js
def test_static_workspace_snapshot_controls_are_available_outside_agent_config():
html = Path("src/agent_lab/presentation/static/index.html").read_text()
chat_dialog_start = html.index('