"""Tests for meeting_room.engine — DiscussionEngine.""" from __future__ import annotations import json from unittest.mock import MagicMock, patch import pytest from meeting_room.api_client import APIClient from meeting_room.engine import DiscussionEngine from meeting_room.events import ( AGENT_ERROR, AGENT_MESSAGE, AGENT_TURN_START, BOSS_TURN, DISCUSSION_END, DISCUSSION_START, ROUND_END, ROUND_START, TOOL_CALL, TOOL_RESULT, Event, EventEmitter, ) from meeting_room.models import ( DefaultsConfig, DiscussionConfig, ProviderConfig, RoleConfig, ScenarioConfig, ) from meeting_room.tools import ToolRegistry # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- PROVIDERS = { "openai": ProviderConfig(base_url="https://api.openai.com/v1", api_key="sk-test"), } ROLES = { "analyst": RoleConfig( name="Analyst", provider="openai", model="gpt-4o-mini", temperature=0.7, tools="none", system_prompt="You are an analyst.", ), "skeptic": RoleConfig( name="Skeptic", provider="openai", model="gpt-4o-mini", temperature=0.5, tools="none", system_prompt="You are a skeptic.", ), "researcher": RoleConfig( name="Researcher", provider="openai", model="gpt-4o-mini", temperature=0.7, tools="all", system_prompt="You research things.", ), } BASIC_CONFIG = DiscussionConfig( providers=PROVIDERS, roles=ROLES, defaults=DefaultsConfig(max_rounds=3), ) BASIC_SCENARIO = ScenarioConfig( name="Test Discussion", participants=["analyst", "skeptic"], max_rounds=2, problem="Should we use Python or Rust?", ) def _make_mock_api( stream_responses: dict[str, str] | None = None, chat_responses: dict[str, dict] | None = None, ) -> MagicMock: """Create a mock APIClient. stream_responses: maps model -> text response for chat_stream chat_responses: maps model -> dict response for chat """ api = MagicMock(spec=APIClient) default_stream = stream_responses or {"gpt-4o-mini": "I think Python is great."} default_chat = chat_responses or {} def chat_side_effect(provider, model, messages, temperature=0.7, tools=None): if model in default_chat: return default_chat[model] return {"content": "Chat response", "tool_calls": None} def stream_side_effect(provider, model, messages, temperature=0.7): return default_stream.get(model, "Stream response") api.chat.side_effect = chat_side_effect api.chat_stream.side_effect = stream_side_effect return api def _make_engine( config: DiscussionConfig | None = None, api: APIClient | None = None, events: EventEmitter | None = None, ) -> tuple[DiscussionEngine, EventEmitter]: """Create a DiscussionEngine with sensible defaults. Returns (engine, event_bus).""" cfg = config or BASIC_CONFIG bus = events or EventEmitter() api_client = api or _make_mock_api() tools = ToolRegistry(workdir=".") engine = DiscussionEngine( config=cfg, api_client=api_client, tool_registry=tools, events=bus, ) return engine, bus # --------------------------------------------------------------------------- # DiscussionEngine — construction # --------------------------------------------------------------------------- class TestDiscussionEngineInit: def test_stores_config(self) -> None: engine, _ = _make_engine() assert engine.config is BASIC_CONFIG def test_stores_dependencies(self) -> None: api = _make_mock_api() bus = EventEmitter() tools = ToolRegistry() engine = DiscussionEngine(BASIC_CONFIG, api, tools, bus) assert engine.api is api assert engine.tools is tools assert engine.events is bus # --------------------------------------------------------------------------- # DiscussionEngine.run — basic lifecycle # --------------------------------------------------------------------------- class TestDiscussionEngineRunLifecycle: def test_emits_discussion_start(self) -> None: engine, bus = _make_engine() events: list[Event] = [] bus.on(DISCUSSION_START, lambda e: events.append(e)) engine.run(BASIC_SCENARIO) assert len(events) == 1 assert events[0].data["name"] == "Test Discussion" assert events[0].data["rounds"] == 2 def test_emits_round_start_and_end(self) -> None: engine, bus = _make_engine() starts: list[Event] = [] ends: list[Event] = [] bus.on(ROUND_START, lambda e: starts.append(e)) bus.on(ROUND_END, lambda e: ends.append(e)) engine.run(BASIC_SCENARIO) assert len(starts) == 2 # 2 rounds assert len(ends) == 2 assert starts[0].data["round_num"] == 1 assert starts[1].data["round_num"] == 2 assert ends[0].data["total_rounds"] == 2 def test_emits_agent_turn_start_for_each_participant(self) -> None: engine, bus = _make_engine() turn_starts: list[Event] = [] bus.on(AGENT_TURN_START, lambda e: turn_starts.append(e)) engine.run(BASIC_SCENARIO) # 2 participants * 2 rounds = 4 turns assert len(turn_starts) == 4 assert turn_starts[0].data["role_id"] == "analyst" assert turn_starts[0].data["name"] == "Analyst" assert turn_starts[1].data["role_id"] == "skeptic" def test_emits_agent_message_for_each_turn(self) -> None: engine, bus = _make_engine() messages: list[Event] = [] bus.on(AGENT_MESSAGE, lambda e: messages.append(e)) engine.run(BASIC_SCENARIO) assert len(messages) == 4 # 2 participants * 2 rounds def test_emits_discussion_end(self) -> None: engine, bus = _make_engine() end_events: list[Event] = [] bus.on(DISCUSSION_END, lambda e: end_events.append(e)) engine.run(BASIC_SCENARIO) assert len(end_events) == 1 assert end_events[0].data["total_messages"] == 4 def test_returns_conversation(self) -> None: engine, _ = _make_engine() result = engine.run(BASIC_SCENARIO) assert len(result) == 4 assert all(isinstance(m, __import__("meeting_room.models", fromlist=["Message"]).Message) for m in result) assert result[0].role_id == "analyst" assert result[1].role_id == "skeptic" def test_full_event_order(self) -> None: """Events must be emitted in the correct order for a 1-round, 2-participant discussion.""" config = DiscussionConfig( providers=PROVIDERS, roles=ROLES, defaults=DefaultsConfig(max_rounds=3), ) scenario = ScenarioConfig( name="Order Test", participants=["analyst", "skeptic"], max_rounds=1, problem="Test problem", ) engine, bus = _make_engine(config=config) log: list[str] = [] bus.on(DISCUSSION_START, lambda e: log.append("DISCUSSION_START")) bus.on(ROUND_START, lambda e: log.append(f"ROUND_START:{e.data['round_num']}")) bus.on(AGENT_TURN_START, lambda e: log.append(f"TURN_START:{e.data['role_id']}")) bus.on(AGENT_MESSAGE, lambda e: log.append(f"AGENT_MESSAGE:{e.data['role_id']}")) bus.on(ROUND_END, lambda e: log.append(f"ROUND_END:{e.data['round_num']}")) bus.on(DISCUSSION_END, lambda e: log.append("DISCUSSION_END")) engine.run(scenario) assert log == [ "DISCUSSION_START", "ROUND_START:1", "TURN_START:analyst", "AGENT_MESSAGE:analyst", "TURN_START:skeptic", "AGENT_MESSAGE:skeptic", "ROUND_END:1", "DISCUSSION_END", ] # --------------------------------------------------------------------------- # DiscussionEngine.run — backward compatibility # --------------------------------------------------------------------------- class TestDiscussionEngineRunDict: def test_accepts_dict_scenario(self) -> None: engine, bus = _make_engine() events: list[Event] = [] bus.on(DISCUSSION_START, lambda e: events.append(e)) scenario_dict = { "name": "Dict Test", "participants": ["analyst"], "max_rounds": 1, "problem": "Dict problem?", } result = engine.run(scenario_dict) assert events[0].data["name"] == "Dict Test" assert len(result) == 1 def test_uses_default_participants_when_empty(self) -> None: """If scenario.participants is empty, use all roles from config.""" engine, bus = _make_engine() turn_starts: list[Event] = [] bus.on(AGENT_TURN_START, lambda e: turn_starts.append(e)) scenario = ScenarioConfig( name="Default Participants", participants=[], # empty max_rounds=1, problem="Test", ) result = engine.run(scenario) # Should use all 3 roles from config: analyst, skeptic, researcher # (but researcher has tools="all" — mock the chat call) assert len(turn_starts) == 3 assert len(result) == 3 def test_uses_config_default_max_rounds_when_zero(self) -> None: engine, bus = _make_engine() end_events: list[Event] = [] bus.on(DISCUSSION_END, lambda e: end_events.append(e)) scenario = ScenarioConfig( name="Rounds Test", participants=["analyst"], max_rounds=0, # zero → should fall back to config default (3) problem="Test", ) engine.run(scenario) assert end_events[0].data["total_messages"] == 3 # 1 participant * 3 rounds # --------------------------------------------------------------------------- # DiscussionEngine.run — unknown role # --------------------------------------------------------------------------- class TestDiscussionEngineRunUnknownRole: def test_unknown_role_id_raises(self) -> None: engine, _ = _make_engine() scenario = ScenarioConfig( name="Bad Role", participants=["nonexistent_role"], max_rounds=1, problem="Test", ) with pytest.raises(ValueError, match="Unknown role_id"): engine.run(scenario) # --------------------------------------------------------------------------- # DiscussionEngine._run_agent_turn — tool calls # --------------------------------------------------------------------------- class TestDiscussionEngineToolCalls: def test_tool_call_events_are_emitted(self) -> None: """When an agent makes a tool call, TOOL_CALL and TOOL_RESULT events must fire.""" api = _make_mock_api() # First call returns a tool call, second call returns text api.chat.side_effect = None api.chat.side_effect = [ { "content": None, "tool_calls": [ { "id": "call_001", "name": "read_file", "arguments": '{"path": "/tmp/test.txt"}', } ], }, {"content": "I read the file.", "tool_calls": None}, ] engine, bus = _make_engine(api=api) tool_calls: list[Event] = [] tool_results: list[Event] = [] bus.on(TOOL_CALL, lambda e: tool_calls.append(e)) bus.on(TOOL_RESULT, lambda e: tool_results.append(e)) scenario = ScenarioConfig( name="Tool Test", participants=["researcher"], # has tools="all" max_rounds=1, problem="Read a file", ) result = engine.run(scenario) assert len(tool_calls) == 1 assert tool_calls[0].data["role_id"] == "researcher" assert tool_calls[0].data["tool_name"] == "read_file" assert len(tool_results) == 1 assert tool_results[0].data["role_id"] == "researcher" assert tool_results[0].data["tool_name"] == "read_file" # The final message should contain the text response assert "I read the file." in result[0].content def test_multiple_tool_calls_in_single_turn(self) -> None: """Multiple tool calls in one response should all emit events.""" api = _make_mock_api() api.chat.side_effect = None api.chat.side_effect = [ { "content": None, "tool_calls": [ {"id": "call_001", "name": "read_file", "arguments": '{"path": "/a"}'}, {"id": "call_002", "name": "list_files", "arguments": '{"path": "/b"}'}, ], }, {"content": "Done researching.", "tool_calls": None}, ] engine, bus = _make_engine(api=api) tool_calls: list[Event] = [] bus.on(TOOL_CALL, lambda e: tool_calls.append(e)) scenario = ScenarioConfig( name="Multi-Tool", participants=["researcher"], max_rounds=1, problem="Research", ) engine.run(scenario) assert len(tool_calls) == 2 assert tool_calls[0].data["tool_name"] == "read_file" assert tool_calls[1].data["tool_name"] == "list_files" def test_tool_result_preview_truncated(self) -> None: """Tool results longer than 200 chars should have truncated preview in event.""" long_result = "x" * 500 api = _make_mock_api() api.chat.side_effect = None # We patch execute_tool to return a long string engine, bus = _make_engine(api=api) tool_results: list[Event] = [] bus.on(TOOL_RESULT, lambda e: tool_results.append(e)) original_execute = engine.tools.execute_tool def mock_execute(name, args): if name == "read_file": return long_result return original_execute(name, args) engine.tools.execute_tool = mock_execute api.chat.side_effect = [ { "content": None, "tool_calls": [ {"id": "call_001", "name": "read_file", "arguments": '{"path": "/x"}'}, ], }, {"content": "Got it.", "tool_calls": None}, ] scenario = ScenarioConfig( name="Truncation Test", participants=["researcher"], max_rounds=1, problem="Read big file", ) engine.run(scenario) # Preview is first 200 chars with newlines replaced by spaces assert len(tool_results[0].data["result_preview"]) <= 200 assert "\n" not in tool_results[0].data["result_preview"] # Full result should be preserved assert tool_results[0].data["full_result"] == long_result # --------------------------------------------------------------------------- # DiscussionEngine.run — error handling # --------------------------------------------------------------------------- class TestDiscussionEngineErrors: def test_api_error_emits_agent_error(self) -> None: """If the API call raises, the engine should emit AGENT_ERROR and continue.""" api = _make_mock_api() api.chat_stream.side_effect = None api.chat_stream.side_effect = RuntimeError("API timeout") engine, bus = _make_engine(api=api) error_events: list[Event] = [] bus.on(AGENT_ERROR, lambda e: error_events.append(e)) scenario = ScenarioConfig( name="Error Test", participants=["analyst"], # no tools, uses chat_stream max_rounds=1, problem="Test", ) result = engine.run(scenario) assert len(error_events) == 1 assert "API timeout" in error_events[0].data["error"] # The message content should contain the error assert "[Error:" in result[0].content def test_all_agents_error_still_completes_discussion(self) -> None: """Even if every agent errors, the discussion should complete with DISCUSSION_END.""" api = _make_mock_api() api.chat_stream.side_effect = None api.chat_stream.side_effect = RuntimeError("fail") engine, bus = _make_engine(api=api) end_events: list[Event] = [] bus.on(DISCUSSION_END, lambda e: end_events.append(e)) scenario = ScenarioConfig( name="All Error", participants=["analyst", "skeptic"], max_rounds=1, problem="Test", ) result = engine.run(scenario) assert len(end_events) == 1 assert end_events[0].data["total_messages"] == 2 assert all("[Error:" in m.content for m in result) # --------------------------------------------------------------------------- # DiscussionEngine.inject_message # --------------------------------------------------------------------------- class TestDiscussionEngineInjectMessage: def test_inject_emits_boss_turn(self) -> None: engine, bus = _make_engine() boss_events: list[Event] = [] bus.on(BOSS_TURN, lambda e: boss_events.append(e)) msg = engine.inject_message("analyst", "Stop discussing and wrap up.") assert len(boss_events) == 1 assert boss_events[0].data["role_id"] == "analyst" assert boss_events[0].data["content"] == "Stop discussing and wrap up." assert boss_events[0].data["name"] == "Analyst" def test_inject_returns_message(self) -> None: engine, _ = _make_engine() msg = engine.inject_message("skeptic", "I disagree.") assert msg.role_id == "skeptic" assert msg.name == "Skeptic" assert msg.content == "I disagree." def test_inject_unknown_role_raises(self) -> None: engine, _ = _make_engine() with pytest.raises(ValueError, match="Unknown role_id"): engine.inject_message("nonexistent", "test") # --------------------------------------------------------------------------- # DiscussionEngine._format_conversation # --------------------------------------------------------------------------- class TestFormatConversation: def test_formats_messages(self) -> None: conversation = [ __import__("meeting_room.models", fromlist=["Message"]).Message( role_id="a", name="Alice", content="Hello" ), __import__("meeting_room.models", fromlist=["Message"]).Message( role_id="b", name="Bob", content="Hi there" ), ] result = DiscussionEngine._format_conversation(conversation) assert "[Alice]: Hello" in result assert "[Bob]: Hi there" in result def test_empty_conversation(self) -> None: result = DiscussionEngine._format_conversation([]) assert result == "" # --------------------------------------------------------------------------- # DiscussionEngine._build_messages # --------------------------------------------------------------------------- class TestBuildMessages: def test_first_message_contains_problem(self) -> None: engine, _ = _make_engine() role = ROLES["analyst"] messages = engine._build_messages(role, "Should we use Go?", [], is_first_message=True) assert messages[0]["role"] == "system" assert messages[0]["content"] == role.system_prompt.strip() assert "Should we use Go?" in messages[1]["content"] def test_subsequent_message_contains_conversation(self) -> None: engine, _ = _make_engine() role = ROLES["analyst"] conversation = [ __import__("meeting_room.models", fromlist=["Message"]).Message( role_id="analyst", name="Analyst", content="I think Python is great." ), ] messages = engine._build_messages(role, "The problem", conversation, is_first_message=False) assert "I think Python is great." in messages[1]["content"] assert "Continue the discussion" in messages[1]["content"] # --------------------------------------------------------------------------- # Integration: streaming vs non-streaming path # --------------------------------------------------------------------------- class TestStreamingPath: def test_no_tools_uses_chat_stream(self) -> None: """Agents with tools='none' should use chat_stream (streaming).""" api = _make_mock_api(stream_responses={"gpt-4o-mini": "Streamed response"}) engine, bus = _make_engine(api=api) msg_events: list[Event] = [] bus.on(AGENT_MESSAGE, lambda e: msg_events.append(e)) scenario = ScenarioConfig( name="Stream Test", participants=["analyst"], # tools="none" max_rounds=1, problem="Test", ) result = engine.run(scenario) api.chat_stream.assert_called_once() api.chat.assert_not_called() assert result[0].content == "Streamed response" def test_with_tools_uses_chat(self) -> None: """Agents with tools should use chat (non-streaming).""" api = _make_mock_api() api.chat.side_effect = None api.chat.return_value = {"content": "Research done.", "tool_calls": None} engine, bus = _make_engine(api=api) scenario = ScenarioConfig( name="Chat Test", participants=["researcher"], # tools="all" max_rounds=1, problem="Test", ) result = engine.run(scenario) api.chat.assert_called_once() api.chat_stream.assert_not_called() assert result[0].content == "Research done."