"""DiscussionEngine — event-driven round-robin discussion runner. Replaces the monolithic backends/custom/run.py with a clean, testable architecture that emits typed events instead of printing to stdout. """ from __future__ import annotations import json import logging from typing import Union from meeting_room.api_client import APIClient 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, EventEmitter, ) from meeting_room.models import ( DiscussionConfig, Message, RoleConfig, ScenarioConfig, ToolCallRecord, ) from meeting_room.tools import ToolRegistry logger = logging.getLogger(__name__) MAX_TOOL_ITERATIONS = 5 class DiscussionEngine: """Orchestrates a multi-agent discussion using events for output. Parameters ---------- config : DiscussionConfig Top-level configuration (providers, roles, defaults). api_client : APIClient Client for making LLM API calls. tool_registry : ToolRegistry Registry for tool schema lookup and execution. events : EventEmitter Event bus for emitting lifecycle events. """ def __init__( self, config: DiscussionConfig, api_client: APIClient, tool_registry: ToolRegistry, events: EventEmitter, ) -> None: self.config = config self.api = api_client self.tools = tool_registry self.events = events # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def run( self, scenario: Union[ScenarioConfig, dict], ) -> list[Message]: """Run a full discussion and return the conversation as Message objects. Parameters ---------- scenario : ScenarioConfig or dict The scenario to run. A dict is accepted for backward compatibility and is coerced to ScenarioConfig. Returns ------- list[Message] The ordered list of messages produced during the discussion. """ if isinstance(scenario, dict): scenario = ScenarioConfig(**scenario) participants = scenario.participants or list(self.config.roles.keys()) max_rounds = scenario.max_rounds or self.config.defaults.max_rounds # Eagerly resolve role configs so missing role_id fails fast. roles: dict[str, RoleConfig] = {} for rid in participants: if rid not in self.config.roles: raise ValueError(f"Unknown role_id '{rid}' — not in config.roles") roles[rid] = self.config.roles[rid] conversation: list[Message] = [] self.events.emit( DISCUSSION_START, name=scenario.name, participants=[roles[rid].name for rid in participants], rounds=max_rounds, problem=scenario.problem, ) for round_num in range(1, max_rounds + 1): self.events.emit(ROUND_START, round_num=round_num, total_rounds=max_rounds) for role_id in participants: role = roles[role_id] is_first = len(conversation) == 0 self.events.emit( AGENT_TURN_START, role_id=role_id, name=role.name, ) try: content = self._run_agent_turn( role=role, role_id=role_id, problem=scenario.problem, conversation=conversation, is_first_message=is_first, ) except Exception as exc: content = f"[Error: {exc}]" self.events.emit( AGENT_ERROR, role_id=role_id, name=role.name, error=str(exc), ) conversation.append( Message(role_id=role_id, name=role.name, content=content) ) self.events.emit( AGENT_MESSAGE, role_id=role_id, name=role.name, content=content, ) self.events.emit(ROUND_END, round_num=round_num, total_rounds=max_rounds) self.events.emit( DISCUSSION_END, total_messages=len(conversation), ) return conversation def inject_message(self, role_id: str, content: str) -> Message: """Inject an arbitrary message from a Boss/CSO role. This does **not** make an LLM call — it simply appends a message and emits a BOSS_TURN event so subscribers can log or relay it. Parameters ---------- role_id : str Identifier of the injecting role (e.g. "boss", "cso"). content : str The message content to inject. Returns ------- Message The created Message object. Raises ------ ValueError If *role_id* is not in config.roles. """ if role_id not in self.config.roles: raise ValueError(f"Unknown role_id '{role_id}' — not in config.roles") role = self.config.roles[role_id] msg = Message(role_id=role_id, name=role.name, content=content) self.events.emit( BOSS_TURN, role_id=role_id, name=role.name, content=content, ) return msg # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ @staticmethod def _format_conversation(conversation: list[Message]) -> str: """Format a list of Message objects into a prompt-friendly string.""" return "\n\n".join( f"[{msg.name}]: {msg.content}" for msg in conversation ) def _build_messages( self, role: RoleConfig, problem: str, conversation: list[Message], is_first_message: bool, ) -> list[dict]: """Build the API message list for one agent turn.""" system_msg = {"role": "system", "content": role.system_prompt.strip()} if is_first_message: context_msg = { "role": "user", "content": ( f"Problem for discussion:\n{problem}\n\n" "Discuss it with other participants. Express your position. " "Use tools if necessary." ), } else: context_msg = { "role": "user", "content": ( f"Here is what has been discussed so far:\n\n" f"{self._format_conversation(conversation)}\n\n" "Continue the discussion. Respond to arguments, offer your own. " "Use tools if necessary." ), } return [system_msg, context_msg] def _run_agent_turn( self, role: RoleConfig, role_id: str, problem: str, conversation: list[Message], is_first_message: bool, ) -> str: """Run one agent turn with tool support. Returns the final text content produced by the agent. """ messages = self._build_messages(role, problem, conversation, is_first_message) tool_schemas = self.tools.get_tool_schemas(role.tools) has_tools = len(tool_schemas) > 0 for _ in range(MAX_TOOL_ITERATIONS): if has_tools: result = self.api.chat( provider=role.provider, model=role.model, messages=messages, temperature=role.temperature, tools=tool_schemas, ) else: content = self.api.chat_stream( provider=role.provider, model=role.model, messages=messages, temperature=role.temperature, ) return content if result["tool_calls"]: # Append assistant message with tool calls assistant_msg: dict = { "role": "assistant", "content": result["content"] or None, "tool_calls": [ { "id": tc["id"], "type": "function", "function": { "name": tc["name"], "arguments": tc["arguments"], }, } for tc in result["tool_calls"] ], } messages.append(assistant_msg) # Execute each tool call and append results for tc in result["tool_calls"]: self.events.emit( TOOL_CALL, role_id=role_id, tool_name=tc["name"], arguments=tc["arguments"], ) try: args = ( json.loads(tc["arguments"]) if isinstance(tc["arguments"], str) else tc["arguments"] ) except json.JSONDecodeError: args = {} tool_result = self.tools.execute_tool(tc["name"], args) preview = tool_result[:200].replace("\n", " ") self.events.emit( TOOL_RESULT, role_id=role_id, tool_name=tc["name"], result_preview=preview, full_result=tool_result, ) messages.append( { "role": "tool", "tool_call_id": tc["id"], "content": tool_result, } ) continue # No tool calls — return the content if result["content"]: return result["content"] # Exhausted tool iterations — return whatever we have return result.get("content", "")