import json import asyncio import logging from collections.abc import AsyncIterator 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 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.event( 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.system_prompt == "" 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_handoff_note_metadata(): 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", ] 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, }, } 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"), ) 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, ) assert [message["content"] for message in client.get( f"/api/sessions/{session_id}/messages" ).json()] == ["debug this", "answer"] 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_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" 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, ) -> 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.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 payload["tool_choice"] == { "type": "function", "function": {"name": "handoff_note"}, } 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_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_rejects_internal_tool_messages_before_network(): network_called = False def handler(request: httpx.Request) -> httpx.Response: nonlocal network_called network_called = True 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, ) with pytest.raises(ValueError, match="tool messages are internal"): [ item async for item in client.stream_chat( messages=[ ChatMessage( role="tool", content="internal tool result", tool_call_id="call_1", ) ], tools=[], params=AgentParams(model="provider-default", temperature=0.3, max_tokens=50), ) ] assert network_called is False 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(' 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, include_usage=False, http_client=http_client, ) [ item async for item in client.stream_chat( messages=[ChatMessage(role="user", content="hi")], tools=[], params=AgentParams(model="provider-default", temperature=0.3, max_tokens=50), ) ] assert "stream_options" not in captured_payloads[0] @pytest.mark.asyncio async def test_openai_chat_client_merges_agent_extra_body_into_payload(): 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="provider-default", temperature=0.3, max_tokens=50, extra_body={ "thinking": {"type": "disabled"}, "enable_search": False, "search_options": {"forced_search": False}, }, ), ) ] assert captured_payloads[0]["thinking"] == {"type": "disabled"} assert captured_payloads[0]["enable_search"] is False assert captured_payloads[0]["search_options"] == {"forced_search": False}