"""Tests for the meeting event loop.""" from __future__ import annotations import json from pathlib import Path from src.meeting import run_meeting, run_meeting_interactive, save_report, save_transcript from src.models import AgentConfig, MeetingResult SAMPLE_AGENTS: dict[str, AgentConfig] = { "architect": AgentConfig(name="architect", role="Software Architect", system_prompt="You are an architect."), "devops": AgentConfig(name="devops", role="DevOps Engineer", system_prompt="You are DevOps."), } class SequentialMockClient: """Mock client that returns different responses in sequence.""" def __init__(self, responses: list[str]) -> None: self.responses = responses self.call_index = 0 def chat(self, system: str, messages: list[dict[str, str]]) -> str: if self.call_index < len(self.responses): response = self.responses[self.call_index] self.call_index += 1 return response return json.dumps({ "analysis": "Forced finish.", "next_action": "FINISH", "final_report": "# Fallback Report", }) class TestRunMeeting: """Tests for run_meeting function.""" def test_simple_meeting_flow(self) -> None: """Meeting calls one agent then finishes, returns MeetingResult.""" responses = [ json.dumps({ "analysis": "Need architect input.", "next_action": "CALL_AGENT", "target_agent": "architect", "prompt_for_agent": "What architecture?", }), "Use microservices for scalability.", json.dumps({ "analysis": "Got architect input.", "next_action": "FINISH", "final_report": "# Meeting Report\n\nArchitect recommends microservices.", }), ] mock_client = SequentialMockClient(responses) result = run_meeting(topic="Database migration", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5) assert isinstance(result, MeetingResult) assert "Meeting Report" in result.final_report assert "microservices" in result.final_report assert result.topic == "Database migration" assert len(result.transcript) > 0 def test_run_meeting_returns_meeting_result(self) -> None: """run_meeting returns a MeetingResult with full metadata.""" responses = [ json.dumps({ "analysis": "Done.", "next_action": "FINISH", "final_report": "# Quick Report", }), ] mock_client = SequentialMockClient(responses) result = run_meeting(topic="Quick test", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5) assert isinstance(result, MeetingResult) assert result.topic == "Quick test" assert result.final_report == "# Quick Report" assert result.started_at is not None assert result.finished_at is not None assert result.started_at <= result.finished_at assert set(result.participants) == set(SAMPLE_AGENTS.keys()) def test_max_rounds_forces_finish(self) -> None: """Meeting forces finish when max_rounds is reached.""" responses = [] for i in range(10): responses.append(json.dumps({ "analysis": f"Round {i}.", "next_action": "CALL_AGENT", "target_agent": "architect", "prompt_for_agent": "Thoughts?", })) responses.append("Some response.") responses.append(json.dumps({ "analysis": "Forced to finish.", "next_action": "FINISH", "final_report": "# Forced Report\n\nMax rounds reached.", })) mock_client = SequentialMockClient(responses) result = run_meeting(topic="Test", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=2) assert isinstance(result, MeetingResult) assert "Forced Report" in result.final_report or "Report" in result.final_report def test_meeting_with_multiple_agents(self) -> None: """Meeting calls multiple agents before finishing.""" responses = [ json.dumps({ "analysis": "Start with architect.", "next_action": "CALL_AGENT", "target_agent": "architect", "prompt_for_agent": "Architecture thoughts?", }), "Use event-driven architecture.", json.dumps({ "analysis": "Now DevOps.", "next_action": "CALL_AGENT", "target_agent": "devops", "prompt_for_agent": "Deployment concerns?", }), "Need Kubernetes.", json.dumps({ "analysis": "All covered.", "next_action": "FINISH", "final_report": "# Full Report\n\nArch: event-driven. DevOps: K8s.", }), ] mock_client = SequentialMockClient(responses) result = run_meeting(topic="New system", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5) assert isinstance(result, MeetingResult) assert "Full Report" in result.final_report class TestSaveReport: """Tests for save_report function.""" def test_save_report_creates_file(self, tmp_path: Path) -> None: report = "# Test Report\n\nContent here." path = save_report(report, topic="test-topic", output_dir=tmp_path) assert path.exists() assert path.suffix == ".md" assert "test-topic" in path.name assert path.read_text(encoding="utf-8") == report class TestSaveTranscript: """Tests for save_transcript function.""" def test_save_transcript_creates_md_file(self, tmp_path: Path) -> None: """save_transcript writes a .transcript.md file with round-by-round content.""" from datetime import datetime from src.models import ScratchpadEntry transcript = [ ScratchpadEntry(agent_name="architect", content="Use microservices.", timestamp=datetime(2026, 3, 30, 10, 0)), ] result = MeetingResult( topic="Test topic", context="Some background", participants=["architect"], transcript=transcript, final_report="# Report", started_at=datetime(2026, 3, 30, 10, 0), finished_at=datetime(2026, 3, 30, 10, 30), ) path = save_transcript(result, output_dir=tmp_path) assert path.exists() assert path.name.endswith(".transcript.md") content = path.read_text(encoding="utf-8") assert "Test topic" in content assert "architect" in content assert "Use microservices." in content class TestRunMeetingInteractive: """Tests for run_meeting_interactive function.""" def test_interactive_empty_input_finishes_immediately(self, tmp_path: Path) -> None: """Empty input from user ends the interactive loop after one meeting run.""" responses = [ json.dumps({ "analysis": "Done.", "next_action": "FINISH", "final_report": "# Report V1", }), ] mock_client = SequentialMockClient(responses) result = run_meeting_interactive( topic="Test", context="", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5, output_dir=tmp_path, input_fn=lambda _: "", ) assert isinstance(result, MeetingResult) assert result.final_report == "# Report V1" def test_interactive_with_feedback_runs_more_rounds(self, tmp_path: Path) -> None: """User feedback triggers additional meeting rounds.""" feedback_count = 0 def mock_input(prompt: str) -> str: nonlocal feedback_count feedback_count += 1 if feedback_count == 1: return "What about security?" return "" # First loop: finish. Second loop (after feedback): call agent then finish. responses = [ json.dumps({ "analysis": "Done round 1.", "next_action": "FINISH", "final_report": "# Report V1", }), # After human feedback, PM calls agent then finishes json.dumps({ "analysis": "Human asked about security.", "next_action": "CALL_AGENT", "target_agent": "architect", "prompt_for_agent": "Address security concerns.", }), "Security is handled via TLS and auth.", json.dumps({ "analysis": "Security addressed.", "next_action": "FINISH", "final_report": "# Report V2\n\nSecurity addressed.", }), ] mock_client = SequentialMockClient(responses) result = run_meeting_interactive( topic="Test", context="", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5, output_dir=tmp_path, input_fn=mock_input, ) assert isinstance(result, MeetingResult) assert "Report V2" in result.final_report def test_human_entry_appears_in_transcript(self, tmp_path: Path) -> None: """Human feedback appears in the final transcript as [HUMAN].""" feedback_given = False def mock_input(prompt: str) -> str: nonlocal feedback_given if not feedback_given: feedback_given = True return "Consider cost implications" return "" responses = [ json.dumps({ "analysis": "First pass done.", "next_action": "FINISH", "final_report": "# V1", }), json.dumps({ "analysis": "Addressing cost.", "next_action": "FINISH", "final_report": "# V2 with cost", }), ] mock_client = SequentialMockClient(responses) result = run_meeting_interactive( topic="Test", context="", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5, output_dir=tmp_path, input_fn=mock_input, ) human_entries = [e for e in result.transcript if e.agent_name == "HUMAN"] assert len(human_entries) == 1 assert "Consider cost implications" in human_entries[0].content class TestMeetingWithContextManager: """Tests for meeting loop with context compression.""" def test_meeting_with_context_manager(self) -> None: """Meeting works with context_manager integration.""" from src.context_manager import ContextManager responses = [ json.dumps({ "analysis": "Need architect.", "next_action": "CALL_AGENT", "target_agent": "architect", "prompt_for_agent": "Architecture?", }), "Use microservices.", json.dumps({ "analysis": "Done.", "next_action": "FINISH", "final_report": "# Report with compression", }), ] mock_client = SequentialMockClient(responses) cm = ContextManager(budget_tokens=6000) result = run_meeting( topic="Compression test", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5, context="Some background", context_manager=cm, ) assert isinstance(result, MeetingResult) assert "Report with compression" in result.final_report def test_meeting_without_context_manager_backward_compatible(self) -> None: """Meeting works without context_manager (backward compatible).""" responses = [ json.dumps({ "analysis": "Done.", "next_action": "FINISH", "final_report": "# No compression report", }), ] mock_client = SequentialMockClient(responses) result = run_meeting( topic="No compress test", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5, ) assert result.final_report == "# No compression report" def test_context_summary_generated_for_long_context(self) -> None: """When context is long and context_manager present, context_summary is set.""" from src.context_manager import ContextManager long_context = "D" * 20000 responses = [ json.dumps({ "analysis": "Done.", "next_action": "FINISH", "final_report": "# Summary test", }), ] class SummaryMockClient: """Returns different responses for summarization vs PM calls.""" def __init__(self) -> None: self.call_index = 0 self.responses = responses def chat(self, system: str, messages: list[dict[str, str]]) -> str: if "Summarize" in system: return "Context summary generated." if self.call_index < len(self.responses): r = self.responses[self.call_index] self.call_index += 1 return r return json.dumps({ "analysis": "Forced.", "next_action": "FINISH", "final_report": "# Fallback", }) mock_client = SummaryMockClient() cm = ContextManager(budget_tokens=6000) result = run_meeting( topic="Summary test", agent_configs=SAMPLE_AGENTS, client=mock_client, max_rounds=5, context=long_context, context_manager=cm, ) assert isinstance(result, MeetingResult)