From 4c1b62de014610f4ccaf96054c68c8764109b6d8 Mon Sep 17 00:00:00 2001 From: Mathew Date: Tue, 23 Dec 2025 13:55:41 -0800 Subject: [PATCH 1/2] refactor(agents): Implement thin orchestrator pattern for Design Assistant MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Splits the monolithic design_assistant.py (~1030 lines) into a modular architecture: - DesignOrchestrator: Thin routing layer that handles phase transitions only - BasePhaseAgent: Abstract base class with common hook handling for UI events - GoalUnderstandingAgent: Phase 1 - goal discovery and confirmation - AgentConfigurationAgent: Phase 2 - specialist agent design - BlueprintDesignAgent: Phase 3 - interactive blueprint creation Benefits: - Each phase agent owns its prompt, tools, and transition logic - Shared session state via existing tools.py mechanism - Backward compatible - DesignAssistantSession wraps the orchestrator - Reduced orchestrator complexity from ~850 to ~100 lines 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- docs/design-thin-orchestrator.md | 250 +++++ src/backend/clara/agents/design_assistant.py | 913 ++---------------- src/backend/clara/agents/orchestrator.py | 422 ++++++++ .../clara/agents/phase_agents/__init__.py | 16 + .../phase_agents/agent_configuration.py | 68 ++ src/backend/clara/agents/phase_agents/base.py | 229 +++++ .../agents/phase_agents/blueprint_design.py | 90 ++ .../agents/phase_agents/goal_understanding.py | 61 ++ 8 files changed, 1200 insertions(+), 849 deletions(-) create mode 100644 docs/design-thin-orchestrator.md create mode 100644 src/backend/clara/agents/orchestrator.py create mode 100644 src/backend/clara/agents/phase_agents/__init__.py create mode 100644 src/backend/clara/agents/phase_agents/agent_configuration.py create mode 100644 src/backend/clara/agents/phase_agents/base.py create mode 100644 src/backend/clara/agents/phase_agents/blueprint_design.py create mode 100644 src/backend/clara/agents/phase_agents/goal_understanding.py diff --git a/docs/design-thin-orchestrator.md b/docs/design-thin-orchestrator.md new file mode 100644 index 0000000..e2b6887 --- /dev/null +++ b/docs/design-thin-orchestrator.md @@ -0,0 +1,250 @@ +# Thin Orchestrator Architecture + +## Overview + +This document describes the refactored orchestrator pattern for Clara's Design Assistant. +The key insight is that **the orchestrator should only route between phases** - all +conversation logic lives in the individual phase agents. + +## Current vs. Proposed Architecture + +### Current (Monolithic) + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ DesignAssistantSession │ +│ (~1000 lines) │ +├─────────────────────────────────────────────────────────────────┤ +│ - Session state management │ +│ - Subagent definitions (inline prompts) │ +│ - Hook handling for UI events │ +│ - Router/structured output decision logic │ +│ - Message processing & streaming │ +│ - Phase transition detection │ +└─────────────────────────────────────────────────────────────────┘ +``` + +### Proposed (Thin Orchestrator + Phase Agents) + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ Orchestrator │ +│ (~100 lines) │ +├─────────────────────────────────────────────────────────────────┤ +│ ONLY: │ +│ 1. Look up current phase from session state │ +│ 2. Route message to the correct phase agent │ +│ 3. Stream events from the phase agent │ +│ 4. Detect phase_transition events and update session │ +└─────────────────────────────────────────────────────────────────┘ + │ + ▼ + ┌────────────────────┼────────────────────┐ + │ │ │ + ▼ ▼ ▼ +┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ +│ Phase 1 Agent │ │ Phase 2 Agent │ │ Phase 3 Agent │ +│ Goal Discovery │ │ Agent Config │ │ Blueprint │ +├─────────────────┤ ├─────────────────┤ ├─────────────────┤ +│ - Own prompt │ │ - Own prompt │ │ - Own prompt │ +│ - Own tools │ │ - Own tools │ │ - Own tools │ +│ - Own hooks │ │ - Own hooks │ │ - Own hooks │ +│ - Transition │ │ - Transition │ │ - Transition │ +│ logic │ │ logic │ │ logic │ +└─────────────────┘ └─────────────────┘ └─────────────────┘ + │ │ │ + └────────────────────┼────────────────────┘ + │ + ▼ + ┌─────────────────┐ + │ Shared Context │ + │ (Session State)│ + ├─────────────────┤ + │ - phase │ + │ - goal_summary │ + │ - agent_config │ + │ - blueprint │ + │ - entities │ + │ - agents │ + └─────────────────┘ +``` + +## Benefits + +1. **Single Responsibility**: Orchestrator only routes, agents only converse +2. **Testability**: Each phase agent can be tested in isolation +3. **Maintainability**: Changes to one phase don't affect others +4. **Readability**: Clear separation of concerns +5. **Flexibility**: Easy to add new phases or modify existing ones + +## File Structure + +``` +clara/agents/ +├── orchestrator.py # Thin routing layer (~100 lines) +├── phase_agents/ +│ ├── __init__.py +│ ├── base.py # BasePhaseAgent class +│ ├── goal_understanding.py # Phase 1 +│ ├── agent_configuration.py # Phase 2 +│ └── blueprint_design.py # Phase 3 +├── tools.py # MCP tools (shared) +├── router.py # UI routing logic (optional fallback) +├── structured_output.py # Structured output parsing +└── prompts/ # Prompt templates (unchanged) +``` + +## Implementation Details + +### BasePhaseAgent + +```python +class BasePhaseAgent(ABC): + """Base class for phase agents.""" + + phase: str # Phase identifier + tools: list[str] # Tools available to this phase + + @abstractmethod + def get_prompt(self, session_state: dict) -> str: + """Get the hydrated prompt for this phase.""" + pass + + async def handle_message( + self, + message: str, + session: DesignSessionState, + client: ClaudeSDKClient, + ) -> AsyncIterator[AGUIEvent]: + """Process a message and yield AG-UI events.""" + pass + + def should_transition(self, event: AGUIEvent) -> str | None: + """Check if we should transition to another phase. + Returns the target phase name or None. + """ + pass +``` + +### Orchestrator + +```python +class DesignOrchestrator: + """Thin orchestrator that routes to phase agents.""" + + agents: dict[str, BasePhaseAgent] = { + "goal_understanding": GoalUnderstandingAgent(), + "agent_configuration": AgentConfigurationAgent(), + "blueprint_design": BlueprintDesignAgent(), + } + + async def handle_message( + self, + message: str, + session: DesignSessionState, + ) -> AsyncIterator[AGUIEvent]: + """Route message to the appropriate phase agent.""" + + # 1. Get current phase agent + agent = self.agents[session.phase.value] + + # 2. Stream events from the agent + async for event in agent.handle_message(message, session, self.client): + # 3. Check for phase transition + if target := agent.should_transition(event): + session.phase = DesignPhase(target) + yield AGUIEvent(type="STATE_SNAPSHOT", data={...}) + + yield event +``` + +### Phase Transition Flow + +Phase transitions happen via the `mcp__clara__phase` tool: + +1. Phase agent calls `mcp__clara__phase({"phase": "next_phase"})` +2. Tool updates session state and returns success +3. Agent's turn ends naturally +4. Next message comes in +5. Orchestrator reads session.phase and routes to the new agent + +The orchestrator does NOT need to detect transitions mid-stream because: +- Tools update session state synchronously +- The phase change is already persisted before the agent turn ends +- Next turn naturally goes to the new phase + +### Hooks + +Each phase agent can define its own hooks: + +```python +class GoalUnderstandingAgent(BasePhaseAgent): + def get_hooks(self) -> dict: + """Get hooks for this phase.""" + return { + 'PreToolUse': [ + HookMatcher( + matcher="mcp__clara__ask", + hooks=[self._on_ask_tool] + ) + ] + } +``` + +The orchestrator merges hooks from the active phase agent when creating the client. + +### Session State Flow + +Session state is managed by `tools.py` and flows through MCP tools: + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ Message Flow │ +├─────────────────────────────────────────────────────────────────┤ +│ │ +│ 1. User message arrives │ +│ │ +│ 2. Orchestrator reads session.phase │ +│ │ +│ 3. Routes to phase agent │ +│ │ +│ 4. Agent calls MCP tools: │ +│ - mcp__clara__ask → emits CUSTOM event │ +│ - mcp__clara__save_goal → updates session state │ +│ - mcp__clara__phase → updates session.phase │ +│ │ +│ 5. Agent turn ends │ +│ │ +│ 6. Orchestrator emits STATE_SNAPSHOT │ +│ │ +│ 7. Next message uses updated session.phase │ +│ │ +└─────────────────────────────────────────────────────────────────┘ +``` + +## Migration Strategy + +1. **Create BasePhaseAgent** - Define the interface +2. **Extract Phase 1** - Move goal understanding logic +3. **Extract Phase 2** - Move agent configuration logic +4. **Extract Phase 3** - Move blueprint design logic +5. **Create Orchestrator** - Wire everything together +6. **Update DesignAssistantSession** - Delegate to orchestrator +7. **Test each phase** - Verify behavior preserved +8. **Remove old code** - Clean up DesignAssistantSession + +## Open Questions + +1. **Should each phase agent own its SDK client instance?** + - Option A: Share one client, reconfigure for each phase + - Option B: Each phase agent creates its own client + - Recommendation: Option A for simplicity + +2. **How to handle restored sessions?** + - Restoration context is already built in orchestrator + - Phase agents don't need special handling + +3. **Should router.py be per-phase?** + - Currently router.py handles UI component inference + - Could be moved into phase agents or kept shared + - Recommendation: Keep shared for now, refactor later if needed diff --git a/src/backend/clara/agents/design_assistant.py b/src/backend/clara/agents/design_assistant.py index 7f6d216..6508ad4 100644 --- a/src/backend/clara/agents/design_assistant.py +++ b/src/backend/clara/agents/design_assistant.py @@ -1,888 +1,105 @@ """Design Assistant using Claude Agent SDK. -This module implements Clara's Design Architect agent that helps users -create Interview Blueprints through collaborative conversation. +This module provides the public interface for Clara's Design Architect agent. +The actual implementation is delegated to the DesignOrchestrator which routes +messages to specialized phase agents. + +This file maintains backward compatibility with existing code that uses +DesignAssistantSession and session_manager. """ -import asyncio import logging from collections.abc import AsyncIterator -from dataclasses import dataclass, field -from enum import Enum -from pathlib import Path +from dataclasses import dataclass from typing import Any -from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, ClaudeSDKClient, HookMatcher - -from clara.agents.router import ( - RouterState, - UIRouter, - build_ui_component, - infer_selection_from_assistant_output, - is_cancel_intent, - is_tool_reply, - parse_ui_submission, - strip_selection_list_from_text, - summarize_ui_submission, -) -from clara.agents.structured_output import StructuredOutputParser, ui_component_to_payload -from clara.agents.tools import ( - CLARA_TOOL_NAMES, - clear_session_state, - create_clara_tools, - ensure_other_option, - get_session_state, - sanitize_ask_options, +from clara.agents.orchestrator import ( + AgentCapabilities, + AGUIEvent, + BlueprintPreview, + DesignOrchestrator, + DesignPhase, + DesignSessionState, ) +from clara.agents.tools import get_session_state logger = logging.getLogger(__name__) -# Paths to prompt files -PROMPTS_DIR = Path(__file__).parent / "prompts" - - -def load_prompt(filename: str) -> str: - """Load a prompt from the prompts directory.""" - prompt_path = PROMPTS_DIR / filename - with open(prompt_path, encoding="utf-8") as f: - return f.read().strip() - - -class DesignPhase(str, Enum): - """Phases of the blueprint design process.""" - GOAL_UNDERSTANDING = "goal_understanding" - AGENT_CONFIGURATION = "agent_configuration" - BLUEPRINT_DESIGN = "blueprint_design" - COMPLETE = "complete" - -@dataclass -class BlueprintPreview: - """Preview of the blueprint being designed.""" - project_name: str | None = None - project_type: str | None = None - entity_types: list[str] = field(default_factory=list) - agent_count: int = 0 - topics: list[str] = field(default_factory=list) - - -@dataclass -class AgentCapabilities: - """Configured specialist agent capabilities from Phase 2.""" - role: str | None = None - capabilities: list[str] = field(default_factory=list) - expertise_areas: list[str] = field(default_factory=list) - interaction_style: str | None = None - focus_areas: list[str] = field(default_factory=list) - - -@dataclass -class DesignSessionState: - """State for a design assistant session.""" - session_id: str - project_id: str - phase: DesignPhase = DesignPhase.GOAL_UNDERSTANDING - blueprint_preview: BlueprintPreview = field(default_factory=BlueprintPreview) - agent_capabilities: AgentCapabilities = field(default_factory=AgentCapabilities) - goal_summary: str | None = None - inferred_domain: str | None = None - domain_confidence: float = 0.0 - turn_count: int = 0 - message_count: int = 0 - discussed_topics: list[str] = field(default_factory=list) - - -@dataclass -class AGUIEvent: - """Base AG-UI event structure.""" - type: str - data: dict[str, Any] +# Re-export types for backward compatibility +__all__ = [ + "DesignPhase", + "BlueprintPreview", + "AgentCapabilities", + "DesignSessionState", + "AGUIEvent", + "DesignAssistantSession", + "DesignAssistantManager", + "session_manager", +] class DesignAssistantSession: - """Manages a single design assistant session using Claude Agent SDK.""" - - def __init__(self, session_id: str, project_id: str): - self.session_id = session_id - self.project_id = project_id - self.state = DesignSessionState( - session_id=session_id, - project_id=project_id, - ) - self.client: ClaudeSDKClient | None = None - self._message_queue: asyncio.Queue[str] = asyncio.Queue() - self._response_queue: asyncio.Queue[AGUIEvent] = asyncio.Queue() - self._running = False - self._restored = False # True if this session was restored from DB - self._first_message_sent = False # Track if we've sent the first message after restoration - self.router_state = RouterState() - self.router = UIRouter() - self._structured_output_parser = StructuredOutputParser() - self._ui_emitted_in_turn = False - - def _sync_state_from_tools(self) -> None: - """Sync session state from tool state. - - This ensures session state (used for STATE_SNAPSHOT events and DB persistence) - stays in sync with tool state (modified by MCP tools like phase, project, etc). - """ - tool_state = get_session_state(self.session_id) - - # Sync phase - if tool_state.get("phase"): - try: - self.state.phase = DesignPhase(tool_state["phase"]) - except ValueError: - logger.warning(f"[{self.session_id}] Unknown phase: {tool_state['phase']}") - - # Sync blueprint preview from project/entities/agents - if tool_state.get("project"): - project = tool_state["project"] - self.state.blueprint_preview.project_name = project.get("name") - self.state.blueprint_preview.project_type = project.get("type") - self.state.inferred_domain = project.get("domain") - - if tool_state.get("entities"): - self.state.blueprint_preview.entity_types = [ - e.get("name") for e in tool_state["entities"] if e.get("name") - ] - - if tool_state.get("agents"): - self.state.blueprint_preview.agent_count = len(tool_state["agents"]) - - # Sync agent capabilities - if tool_state.get("agent_capabilities"): - caps = tool_state["agent_capabilities"] - self.state.agent_capabilities.role = caps.get("role") - self.state.agent_capabilities.capabilities = caps.get("capabilities", []) - self.state.agent_capabilities.expertise_areas = caps.get("expertise_areas", []) - self.state.agent_capabilities.interaction_style = caps.get("interaction_style") - self.state.agent_capabilities.focus_areas = caps.get("focus_areas", []) - - # Sync goal summary - if tool_state.get("goal_summary"): - goal = tool_state["goal_summary"] - self.state.goal_summary = goal.get("goal_text") or goal.get("primary_goal") - - def _on_phase_change(self, new_phase: str) -> None: - """Callback when phase changes via mcp__clara__phase tool.""" - try: - self.state.phase = DesignPhase(new_phase) - logger.info(f"[{self.session_id}] Session state phase updated to: {new_phase}") - except ValueError: - logger.warning(f"[{self.session_id}] Unknown phase in callback: {new_phase}") - - def _build_restoration_context(self) -> str: - """Build a context summary for restored sessions. - - This provides the model with context about the session state - since LLMs don't have persistent memory across connections. - """ - parts = [ - "[SESSION CONTEXT - INTERNAL ONLY. Do NOT quote or reveal this to the user.]" - ] - - # Phase info - parts.append(f"Current Phase: {self.state.phase.value}") - - # Goal summary - if self.state.goal_summary: - parts.append(f"Project Goal: {self.state.goal_summary}") - - # Blueprint preview - if self.state.blueprint_preview.project_name: - proj_name = self.state.blueprint_preview.project_name - proj_type = self.state.blueprint_preview.project_type or 'unknown type' - parts.append(f"Project: {proj_name} ({proj_type})") - - if self.state.blueprint_preview.entity_types: - parts.append(f"Entity Types: {', '.join(self.state.blueprint_preview.entity_types)}") - - if self.state.blueprint_preview.agent_count > 0: - parts.append(f"Configured Agents: {self.state.blueprint_preview.agent_count}") - - # Agent capabilities (from Phase 2) - if self.state.agent_capabilities.role: - caps = self.state.agent_capabilities - parts.append(f"Specialist Agent: {caps.role}") - if caps.expertise_areas: - parts.append(f"Expertise: {', '.join(caps.expertise_areas)}") - if caps.interaction_style: - parts.append(f"Style: {caps.interaction_style}") - - # Turn count - turns = self.state.turn_count - msgs = self.state.message_count - parts.append(f"Conversation Progress: {turns} turns, {msgs} messages") - - parts.append("[END SESSION CONTEXT]\n\nUser continues the conversation:") - - return "\n".join(parts) - - def _build_state_snapshot_event(self) -> AGUIEvent: - """Build a STATE_SNAPSHOT event from current session state.""" - return AGUIEvent( - type="STATE_SNAPSHOT", - data={ - "phase": self.state.phase.value, - "preview": { - "project_name": self.state.blueprint_preview.project_name, - "project_type": self.state.blueprint_preview.project_type, - "entity_types": self.state.blueprint_preview.entity_types, - "agent_count": self.state.blueprint_preview.agent_count, - "topics": self.state.blueprint_preview.topics, - }, - "inferred_domain": self.state.inferred_domain, - "debug": { - "thinking": None, - "approach": None, - "turn_count": self.state.turn_count, - "message_count": self.state.message_count, - "domain_confidence": self.state.domain_confidence, - "discussed_topics": self.state.discussed_topics, - } - } - ) - - def _create_subagents(self) -> dict[str, AgentDefinition]: - """Define phase-based subagents. - - All subagents fetch their hydrated prompts dynamically via mcp__clara__get_prompt. - This ensures placeholders like {{goal}} and {{role}} are filled from session state. - """ - # Phase 1 can use static prompt since no prior context needed - phase1_prompt = load_prompt("phase1_goal_understanding.txt") - - # Phase 2 fetches hydrated prompt to get {{goal}} from Phase 1 - phase2_prompt = """You are configuring a specialist interview agent. - -First, call mcp__clara__get_prompt with phase="agent_configuration" to get your -full instructions with the project goal. - -Then follow those instructions to: -1. Analyze the goal -2. Design the specialist agent configuration -3. Call mcp__clara__agent_summary to save the config -4. Call mcp__clara__hydrate_phase3 with the config -5. Transition to blueprint_design phase -""" - - # Phase 3 fetches hydrated prompt and MUST use ask tool interactively - phase3_prompt = """You are Clara, designing an Interview Blueprint interactively. - -CRITICAL INSTRUCTION: You MUST use mcp__clara__ask for EVERY decision point. -Do NOT proceed to build anything until you have collected user input via ask tool. - -Step 1: First, call mcp__clara__get_prompt with phase="blueprint_design" to get -your full context. -Step 2: Then call mcp__clara__ask to ask the user what knowledge they want to -extract from interviews. -Step 3: Wait for user response. Do NOT call entity/agent/project tools until user confirms via ask. + """Thin wrapper around DesignOrchestrator for backward compatibility. -Your ONLY job in this turn is: -1. Call get_prompt to get context -2. Call ask tool to collect what knowledge the user wants to gather -3. Stop and wait for response + This class maintains the same public interface as the original + DesignAssistantSession but delegates all work to the orchestrator. + """ -DO NOT build entities, agents, or projects until the user has responded to your ask tool questions. -""" - - return { - "phase1-goal-discovery": AgentDefinition( - description=( - "Handles Phase 1: Goal Confirmation. " - "Use this agent to discover the user's project through " - "natural conversation. It will explore the core discovery areas " - "and use mcp__clara__ask for structured choices. " - "After completing, call mcp__clara__hydrate_phase2." - ), - tools=[ - "mcp__clara__ask", - "mcp__clara__request_selection_list", - "mcp__clara__request_data_table", - "mcp__clara__request_process_map", - "mcp__clara__project", - "mcp__clara__save_goal_summary", - "mcp__clara__hydrate_phase2", - "mcp__clara__phase", - "mcp__clara__get_prompt", - ], - prompt=phase1_prompt, - model="sonnet" - ), - "phase2-agent-config": AgentDefinition( - description=( - "Handles Phase 2: Agent Configuration. " - "First call mcp__clara__get_prompt to get hydrated instructions with the goal. " - "Then configure the specialist agent and call mcp__clara__hydrate_phase3." - ), - tools=[ - "mcp__clara__agent_summary", - "mcp__clara__phase", - "mcp__clara__get_prompt", - "mcp__clara__hydrate_phase3", - "mcp__clara__request_selection_list", - "mcp__clara__request_data_table", - "mcp__clara__request_process_map", - ], - prompt=phase2_prompt, - model="sonnet" - ), - "phase3-blueprint-design": AgentDefinition( - description=( - "Handles Phase 3: Blueprint Design INTERACTIVELY. " - "First call mcp__clara__get_prompt to get hydrated instructions. " - "This agent MUST use mcp__clara__ask to collect user input " - "before building. It should NOT build entities/agents until " - "user confirms via ask tool responses. " - "Use mcp__clara__prompt_editor for editing prompts. " - "Use mcp__clara__get_agent_context for context files." - ), - tools=[ - "mcp__clara__project", - "mcp__clara__entity", - "mcp__clara__agent", - "mcp__clara__ask", - "mcp__clara__request_selection_list", - "mcp__clara__request_data_table", - "mcp__clara__request_process_map", - "mcp__clara__preview", - "mcp__clara__phase", - "mcp__clara__get_prompt", - "mcp__clara__prompt_editor", - "mcp__clara__get_agent_context", - ], - prompt=phase3_prompt, - model="sonnet" - ), - } - - def _create_hooks(self) -> dict: - """Create hooks for tracking agent activity.""" - response_queue = self._response_queue - - async def pre_tool_hook( - input_data: dict[str, Any], - tool_use_id: str | None, - context: Any - ) -> dict[str, Any]: - """Track tool usage before execution.""" - # Log full input_data to understand structure - logger.info(f"[PreToolUse] input_data keys: {list(input_data.keys())}") - logger.info(f"[PreToolUse] full input_data: {input_data}") - tool_name = input_data.get("tool_name", input_data.get("name", "unknown")) - tool_input = input_data.get("tool_input", input_data.get("input", {})) - logger.info(f"[PreToolUse] {tool_name}: {tool_input}") - await response_queue.put(AGUIEvent( - type="TOOL_CALL_START", - data={"tool": tool_name, "input": tool_input} - )) - - if tool_name in { - "mcp__clara__request_data_table", - "mcp__clara__request_process_map", - "mcp__clara__request_selection_list", - }: - normalized_tool = tool_name.replace("mcp__clara__", "") - self.router_state.pending_tool = normalized_tool - self.router_state.last_tool = normalized_tool - self.router_state.last_tool_status = "open" - - # Special handling for ask tool - emit CUSTOM event with UI component - if tool_name == "mcp__clara__ask": - options = sanitize_ask_options(tool_input.get("options", [])) - ui_component = { - "type": "user_input_required", - "question": tool_input.get("question", ""), - "options": options, - "multi_select": tool_input.get("multi_select", False), - } - # Emit as CUSTOM AG-UI event for reliable rendering - await response_queue.put(AGUIEvent( - type="CUSTOM", - data={"name": "clara:ask", "value": ui_component} - )) - self._ui_emitted_in_turn = True - logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:ask") + def __init__(self, session_id: str, project_id: str): + self._orchestrator = DesignOrchestrator(session_id, project_id) - if tool_name == "mcp__clara__request_selection_list": - options = ensure_other_option(sanitize_ask_options(tool_input.get("options", []))) - ui_component = { - "type": "user_input_required", - "question": tool_input.get("question", ""), - "options": options, - "multi_select": tool_input.get("multi_select", False), - } - await response_queue.put(AGUIEvent( - type="CUSTOM", - data={"name": "clara:ask", "value": ui_component} - )) - self._ui_emitted_in_turn = True - logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:ask (selection list)") + @property + def session_id(self) -> str: + return self._orchestrator.session_id - # Special handling for prompt_editor tool - emit CUSTOM event for editable prompt - if tool_name == "mcp__clara__prompt_editor": - ui_component = { - "type": "prompt_editor", - "title": tool_input.get("title", "System Prompt"), - "prompt": tool_input.get("prompt", ""), - "description": tool_input.get("description", ""), - } - # Emit as CUSTOM AG-UI event for editable prompt UI - await response_queue.put(AGUIEvent( - type="CUSTOM", - data={"name": "clara:prompt_editor", "value": ui_component} - )) - self._ui_emitted_in_turn = True - logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:prompt_editor") + @property + def project_id(self) -> str: + return self._orchestrator.project_id - if tool_name == "mcp__clara__request_data_table": - ui_component = { - "type": "data_table", - "title": tool_input.get("title", "Data Table"), - "columns": tool_input.get("columns", []), - "min_rows": tool_input.get("min_rows", 3), - "starter_rows": tool_input.get("starter_rows", 3), - "input_modes": tool_input.get("input_modes", ["paste", "inline"]), - "summary_prompt": tool_input.get("summary_prompt", ""), - } - await response_queue.put(AGUIEvent( - type="CUSTOM", - data={"name": "clara:data_table", "value": ui_component} - )) - self._ui_emitted_in_turn = True - logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:data_table") + @property + def state(self) -> DesignSessionState: + return self._orchestrator.state - if tool_name == "mcp__clara__request_process_map": - ui_component = { - "type": "process_map", - "title": tool_input.get("title", "Process Map"), - "required_fields": tool_input.get("required_fields", []), - "edge_types": tool_input.get("edge_types", []), - "min_steps": tool_input.get("min_steps", 3), - "seed_nodes": tool_input.get("seed_nodes", []), - } - await response_queue.put(AGUIEvent( - type="CUSTOM", - data={"name": "clara:process_map", "value": ui_component} - )) - self._ui_emitted_in_turn = True - logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:process_map") + @state.setter + def state(self, value: DesignSessionState) -> None: + self._orchestrator.state = value - # Note: agent_summary tool stores config in state but no longer emits UI card + @property + def _restored(self) -> bool: + return self._orchestrator._restored - return {} # No modifications to tool behavior + @_restored.setter + def _restored(self, value: bool) -> None: + self._orchestrator._restored = value - async def post_tool_hook( - input_data: dict[str, Any], - tool_use_id: str | None, - context: Any - ) -> dict[str, Any]: - """Track tool completion.""" - tool_name = input_data.get("tool_name", "unknown") - logger.debug(f"[PostToolUse] {tool_name} completed") - await response_queue.put(AGUIEvent( - type="TOOL_CALL_END", - data={"tool": tool_name} - )) - return {} + @property + def _first_message_sent(self) -> bool: + return self._orchestrator._first_message_sent - return { - 'PreToolUse': [ - HookMatcher( - matcher=None, # Match all tools - hooks=[pre_tool_hook] - ) - ], - 'PostToolUse': [ - HookMatcher( - matcher=None, - hooks=[post_tool_hook] - ) - ] - } + @_first_message_sent.setter + def _first_message_sent(self, value: bool) -> None: + self._orchestrator._first_message_sent = value async def start(self) -> None: """Start the design assistant session.""" - if self._running: - return - - orchestrator_prompt = load_prompt("interview_orchestrator.txt") - agents = self._create_subagents() - hooks = self._create_hooks() - - # Create MCP server with custom tools bound to this session - clara_mcp_server = create_clara_tools(self.session_id) - - # Register phase change callback to keep session state in sync with tool state - tool_state = get_session_state(self.session_id) - tool_state["_on_phase_change"] = self._on_phase_change - - options = ClaudeAgentOptions( - permission_mode="bypassPermissions", - system_prompt=orchestrator_prompt, - allowed_tools=["Task"] + CLARA_TOOL_NAMES, # Task for subagents + custom tools - agents=agents, - hooks=hooks, - mcp_servers={"clara": clara_mcp_server}, # Register our custom MCP server - model="sonnet" - ) - - self.client = ClaudeSDKClient(options=options) - await self.client.__aenter__() - self._running = True - logger.info(f"Design session {self.session_id} started with Clara tools") + await self._orchestrator.start() async def stop(self) -> None: """Stop the design assistant session.""" - if self.client and self._running: - await self.client.__aexit__(None, None, None) - self._running = False - # Clean up in-memory tool state - clear_session_state(self.session_id) - logger.info(f"Design session {self.session_id} stopped") + await self._orchestrator.stop() async def send_message(self, message: str) -> AsyncIterator[AGUIEvent]: """Send a message and stream the response as AG-UI events.""" - if not self._running or not self.client: - raise RuntimeError("Session not started") - - self._ui_emitted_in_turn = False - self.state.message_count += 1 - self.state.turn_count += 1 - - # Sync state from tools before emitting snapshot - self._sync_state_from_tools() - - submission = parse_ui_submission(message) - if submission: - self.router_state.last_tool = submission.kind - self.router_state.last_tool_status = "completed" - self.router_state.pending_tool = None - self.router_state.pending_payload = None - message = summarize_ui_submission(submission) - - if ( - not submission - and self.router_state.pending_tool == "request_selection_list" - and is_tool_reply(message) - ): - self.router_state.last_tool_status = "completed" - self.router_state.pending_tool = None - self.router_state.pending_payload = None - - if ( - not submission - and self.router_state.pending_tool - and is_cancel_intent(message) - ): - self.router_state.last_tool_status = "canceled" - self.router_state.pending_tool = None - self.router_state.pending_payload = None - - # For restored sessions, prepend context on first message - actual_message = message - if self._restored and not self._first_message_sent: - context = self._build_restoration_context() - actual_message = f"{context}\n\n{message}" - self._first_message_sent = True - logger.info(f"[{self.session_id}] Prepending restoration context to first message") - - # Emit state snapshot at start of turn - yield self._build_state_snapshot_event() - - if not submission: - decision = None - if self.state.phase != DesignPhase.GOAL_UNDERSTANDING: - decision = await self.router.decide( - message=message, - state=self.router_state, - phase=self.state.phase.value, - flow="design_assistant", - allow_selection=False, - ) - - if decision and decision.action in {"tool", "clarify"}: - preamble = "" - if decision.action == "tool": - ui_component = build_ui_component(decision) - if not ui_component: - decision = None - else: - tool_name_map = { - "request_data_table": "mcp__clara__request_data_table", - "request_process_map": "mcp__clara__request_process_map", - "request_selection_list": "mcp__clara__request_selection_list", - } - tool_name = tool_name_map.get( - decision.tool_name, "mcp__clara__request_data_table" - ) - if decision.tool_name == "request_data_table": - preamble = "Let's capture that in a table." - elif decision.tool_name == "request_process_map": - preamble = "Let's map the steps so we capture the workflow accurately." - elif decision.tool_name == "request_selection_list": - preamble = "Pick the options that apply." - tool_state = get_session_state(self.session_id) - tool_state["pending_ui_component"] = ui_component - self.router_state.pending_tool = decision.tool_name - self.router_state.pending_payload = decision.params - self.router_state.last_tool = decision.tool_name - self.router_state.last_tool_status = "open" - - yield AGUIEvent( - type="TOOL_CALL_START", - data={"tool": tool_name, "input": decision.params or {}} - ) - yield AGUIEvent( - type="TEXT_MESSAGE_CONTENT", - data={"delta": preamble} - ) - yield AGUIEvent( - type="CUSTOM", - data={ - "name": ( - "clara:data_table" - if decision.tool_name == "request_data_table" - else "clara:process_map" - if decision.tool_name == "request_process_map" - else "clara:ask" - ), - "value": ui_component, - } - ) - yield AGUIEvent( - type="TOOL_CALL_END", - data={"tool": tool_name} - ) - yield AGUIEvent(type="TEXT_MESSAGE_END", data={}) - yield self._build_state_snapshot_event() - return - - if decision and decision.action == "clarify": - self.router_state.last_clarify = decision.clarifying_question - yield AGUIEvent( - type="TEXT_MESSAGE_CONTENT", - data={"delta": decision.clarifying_question or "Can you clarify?"} - ) - yield AGUIEvent(type="TEXT_MESSAGE_END", data={}) - yield self._build_state_snapshot_event() - return - - # Send message to agent (uses actual_message which may include restoration context) - await self.client.query(prompt=actual_message) - - # Helper to drain queued events from hooks - async def drain_queue(): - """Yield any events queued by hooks.""" - while not self._response_queue.empty(): - try: - event = self._response_queue.get_nowait() - yield event - except asyncio.QueueEmpty: - break - - # Stream response - use_structured_output = self._structured_output_parser.is_available() - current_text = "" - async for msg in self.client.receive_response(): - # First, drain any events queued by hooks - async for event in drain_queue(): - yield event - - msg_type = type(msg).__name__ - logger.debug(f"[{self.session_id}] Message type: {msg_type}, attrs: {dir(msg)}") - - if msg_type == 'AssistantMessage': - # Extract text content from the message - if hasattr(msg, 'content'): - for block in msg.content: - if hasattr(block, 'text'): - new_text = block.text - if new_text and new_text != current_text: - # Check if this is a continuation or a new message - # If new_text starts with current_text, it's a continuation - # Otherwise, it's a new message (e.g., after a tool call) - if current_text and new_text.startswith(current_text): - # Continuation - emit just the delta - delta = new_text[len(current_text):] - else: - # New message - emit the full text - delta = new_text - current_text = new_text - if delta and not use_structured_output: - yield AGUIEvent( - type="TEXT_MESSAGE_CONTENT", - data={"delta": delta} - ) - - elif msg_type == 'ToolUseMessage': - # Tool being used - if hasattr(msg, 'name') and hasattr(msg, 'input'): - yield AGUIEvent( - type="TOOL_CALL_START", - data={"tool": msg.name, "input": msg.input} - ) - - elif msg_type == 'ToolResultMessage': - # Tool completed - extract and stream any text content - tool_text = "" - logger.info(f"[{self.session_id}] ToolResultMessage: {msg}") - if hasattr(msg, 'content'): - # Content can be a list of blocks or a string - content = msg.content - logger.info( - f"[{self.session_id}] Tool content: {type(content)}" - ) - if isinstance(content, list): - for block in content: - if hasattr(block, 'text'): - tool_text += block.text - elif isinstance(block, dict) and 'text' in block: - tool_text += block['text'] - elif isinstance(content, str): - tool_text = content - # Also check for 'result' attribute - elif hasattr(msg, 'result'): - result = msg.result - logger.info(f"[{self.session_id}] Tool result: {type(result)}") - if isinstance(result, str): - tool_text = result - elif isinstance(result, dict) and 'content' in result: - for block in result['content']: - if isinstance(block, dict) and 'text' in block: - tool_text += block['text'] - - tool_preview = tool_text[:100] if tool_text else 'empty' - logger.info(f"[{self.session_id}] Extracted tool_text: {tool_preview}") - - # Note: UI components are now handled via CUSTOM events in pre_tool_hook - # No need to parse [UI_COMPONENT] markers from tool results - - yield AGUIEvent( - type="TOOL_CALL_END", - data={} - ) - - # Final drain of any remaining queued events - async for event in drain_queue(): + async for event in self._orchestrator.send_message(message): yield event - if use_structured_output and current_text: - structured_response = None - ui_payload: dict[str, Any] | None = None - selection_decision = None - if not self._ui_emitted_in_turn and not self.router_state.pending_tool: - structured_response = await self._structured_output_parser.parse( - message=current_text, - phase=self.state.phase.value, - flow="design_assistant", - ) - if structured_response: - ui_payload = ui_component_to_payload(structured_response.ui_component) - - if ( - not ui_payload - and not self._ui_emitted_in_turn - and not self.router_state.pending_tool - ): - selection_decision = infer_selection_from_assistant_output(current_text) - if selection_decision: - ui_payload = build_ui_component(selection_decision) - - display_text = ( - structured_response.display_text - if structured_response - else current_text - ) - if ui_payload and display_text and ui_payload.get("type") == "user_input_required": - cleaned_text = strip_selection_list_from_text(display_text) - if cleaned_text: - display_text = cleaned_text - elif selection_decision and selection_decision.params: - display_text = ( - selection_decision.params.get("question") or display_text - ) - if display_text: - yield AGUIEvent( - type="TEXT_MESSAGE_CONTENT", - data={"delta": display_text} - ) - - if ui_payload: - tool_type = ui_payload.get("type") - tool_name = "request_selection_list" - custom_name = "clara:ask" - if tool_type == "data_table": - tool_name = "request_data_table" - custom_name = "clara:data_table" - elif tool_type == "process_map": - tool_name = "request_process_map" - custom_name = "clara:process_map" - - tool_state = get_session_state(self.session_id) - tool_state["pending_ui_component"] = ui_payload - self.router_state.pending_tool = tool_name - self.router_state.pending_payload = ui_payload - self.router_state.last_tool = tool_name - self.router_state.last_tool_status = "open" - self._ui_emitted_in_turn = True - - yield AGUIEvent( - type="TOOL_CALL_START", - data={ - "tool": f"mcp__clara__{tool_name}", - "input": ui_payload, - } - ) - yield AGUIEvent( - type="CUSTOM", - data={"name": custom_name, "value": ui_payload} - ) - yield AGUIEvent( - type="TOOL_CALL_END", - data={"tool": f"mcp__clara__{tool_name}"} - ) - - if not use_structured_output and current_text and not self.router_state.pending_tool: - selection_decision = infer_selection_from_assistant_output(current_text) - if selection_decision: - ui_component = build_ui_component(selection_decision) - if ui_component: - tool_state = get_session_state(self.session_id) - tool_state["pending_ui_component"] = ui_component - self.router_state.pending_tool = selection_decision.tool_name - self.router_state.pending_payload = selection_decision.params - self.router_state.last_tool = selection_decision.tool_name - self.router_state.last_tool_status = "open" - yield AGUIEvent( - type="TOOL_CALL_START", - data={ - "tool": "mcp__clara__request_selection_list", - "input": selection_decision.params or {}, - } - ) - yield AGUIEvent( - type="CUSTOM", - data={"name": "clara:ask", "value": ui_component} - ) - yield AGUIEvent( - type="TOOL_CALL_END", - data={"tool": "mcp__clara__request_selection_list"} - ) - - # Sync state from tools after all tool calls complete - self._sync_state_from_tools() - - # Emit final state snapshot with any changes from this turn - yield self._build_state_snapshot_event() - - # Emit end of message - yield AGUIEvent( - type="TEXT_MESSAGE_END", - data={} - ) +@dataclass +class _SessionWrapper: + """Internal wrapper to track both orchestrator and session.""" + session: DesignAssistantSession + orchestrator: DesignOrchestrator class DesignAssistantManager: @@ -953,8 +170,6 @@ async def restore_session( This creates a new in-memory session and populates it with the state from the database record. """ - from clara.agents.tools import get_session_state - # If session already exists in memory, return it if session_id in self._sessions: return self._sessions[session_id] diff --git a/src/backend/clara/agents/orchestrator.py b/src/backend/clara/agents/orchestrator.py new file mode 100644 index 0000000..a790880 --- /dev/null +++ b/src/backend/clara/agents/orchestrator.py @@ -0,0 +1,422 @@ +"""Thin Orchestrator for the Design Assistant. + +The orchestrator is responsible for: +1. Looking up the current phase from session state +2. Routing messages to the correct phase agent +3. Streaming events from the phase agent +4. Handling phase transitions + +It does NOT contain conversation logic - that lives in the phase agents. +""" + +import asyncio +import logging +from collections.abc import AsyncIterator +from dataclasses import dataclass, field +from enum import Enum +from typing import Any + +from claude_agent_sdk import AgentDefinition, ClaudeAgentOptions, ClaudeSDKClient + +from clara.agents.phase_agents import ( + AgentConfigurationAgent, + BasePhaseAgent, + BlueprintDesignAgent, + GoalUnderstandingAgent, +) +from clara.agents.phase_agents.base import AGUIEvent +from clara.agents.tools import ( + CLARA_TOOL_NAMES, + create_clara_tools, + get_session_state, +) + +logger = logging.getLogger(__name__) + + +class DesignPhase(str, Enum): + """Phases of the blueprint design process.""" + GOAL_UNDERSTANDING = "goal_understanding" + AGENT_CONFIGURATION = "agent_configuration" + BLUEPRINT_DESIGN = "blueprint_design" + COMPLETE = "complete" + + +@dataclass +class BlueprintPreview: + """Preview of the blueprint being designed.""" + project_name: str | None = None + project_type: str | None = None + entity_types: list[str] = field(default_factory=list) + agent_count: int = 0 + topics: list[str] = field(default_factory=list) + + +@dataclass +class AgentCapabilities: + """Configured specialist agent capabilities from Phase 2.""" + role: str | None = None + capabilities: list[str] = field(default_factory=list) + expertise_areas: list[str] = field(default_factory=list) + interaction_style: str | None = None + focus_areas: list[str] = field(default_factory=list) + + +@dataclass +class DesignSessionState: + """State for a design assistant session.""" + session_id: str + project_id: str + phase: DesignPhase = DesignPhase.GOAL_UNDERSTANDING + blueprint_preview: BlueprintPreview = field(default_factory=BlueprintPreview) + agent_capabilities: AgentCapabilities = field(default_factory=AgentCapabilities) + goal_summary: str | None = None + inferred_domain: str | None = None + domain_confidence: float = 0.0 + turn_count: int = 0 + message_count: int = 0 + discussed_topics: list[str] = field(default_factory=list) + + +class DesignOrchestrator: + """Thin orchestrator that routes messages to phase agents. + + The orchestrator is intentionally simple - it just routes to agents + and handles phase transitions. All conversation logic lives in the + phase agents themselves. + """ + + def __init__(self, session_id: str, project_id: str): + self.session_id = session_id + self.project_id = project_id + self.state = DesignSessionState( + session_id=session_id, + project_id=project_id, + ) + + # Event queue for hooks to emit events + self._event_queue: asyncio.Queue[AGUIEvent] = asyncio.Queue() + + # Phase agents (lazy-initialized) + self._phase_agents: dict[str, BasePhaseAgent] = {} + self._client: ClaudeSDKClient | None = None + self._running = False + self._restored = False + self._first_message_sent = False + + def _get_phase_agent(self, phase: DesignPhase) -> BasePhaseAgent: + """Get or create the agent for a phase.""" + phase_str = phase.value + if phase_str not in self._phase_agents: + if phase == DesignPhase.GOAL_UNDERSTANDING: + self._phase_agents[phase_str] = GoalUnderstandingAgent( + self.session_id, self._event_queue + ) + elif phase == DesignPhase.AGENT_CONFIGURATION: + self._phase_agents[phase_str] = AgentConfigurationAgent( + self.session_id, self._event_queue + ) + elif phase == DesignPhase.BLUEPRINT_DESIGN: + self._phase_agents[phase_str] = BlueprintDesignAgent( + self.session_id, self._event_queue + ) + else: + raise ValueError(f"No agent for phase: {phase}") + + return self._phase_agents[phase_str] + + def _create_subagents(self) -> dict[str, AgentDefinition]: + """Create SDK agent definitions from phase agents.""" + agents = {} + tool_state = get_session_state(self.session_id) + + for phase in [ + DesignPhase.GOAL_UNDERSTANDING, + DesignPhase.AGENT_CONFIGURATION, + DesignPhase.BLUEPRINT_DESIGN, + ]: + agent = self._get_phase_agent(phase) + prompt = agent.get_prompt(tool_state) + + # Map phase to agent name + name_map = { + DesignPhase.GOAL_UNDERSTANDING: "phase1-goal-discovery", + DesignPhase.AGENT_CONFIGURATION: "phase2-agent-config", + DesignPhase.BLUEPRINT_DESIGN: "phase3-blueprint-design", + } + + agents[name_map[phase]] = AgentDefinition( + description=agent.get_description(), + tools=agent.tools, + prompt=prompt, + model="sonnet" + ) + + return agents + + def _create_hooks(self) -> dict: + """Create hooks that delegate to the current phase agent.""" + async def pre_tool_hook( + input_data: dict[str, Any], + tool_use_id: str | None, + context: Any + ) -> dict[str, Any]: + agent = self._get_phase_agent(self.state.phase) + return await agent._pre_tool_hook(input_data, tool_use_id, context) + + async def post_tool_hook( + input_data: dict[str, Any], + tool_use_id: str | None, + context: Any + ) -> dict[str, Any]: + agent = self._get_phase_agent(self.state.phase) + return await agent._post_tool_hook(input_data, tool_use_id, context) + + from claude_agent_sdk import HookMatcher + return { + 'PreToolUse': [ + HookMatcher(matcher=None, hooks=[pre_tool_hook]) + ], + 'PostToolUse': [ + HookMatcher(matcher=None, hooks=[post_tool_hook]) + ] + } + + def _sync_state_from_tools(self) -> None: + """Sync session state from tool state.""" + tool_state = get_session_state(self.session_id) + + # Sync phase + if tool_state.get("phase"): + try: + self.state.phase = DesignPhase(tool_state["phase"]) + except ValueError: + logger.warning(f"[{self.session_id}] Unknown phase: {tool_state['phase']}") + + # Sync blueprint preview + if tool_state.get("project"): + project = tool_state["project"] + self.state.blueprint_preview.project_name = project.get("name") + self.state.blueprint_preview.project_type = project.get("type") + self.state.inferred_domain = project.get("domain") + + if tool_state.get("entities"): + self.state.blueprint_preview.entity_types = [ + e.get("name") for e in tool_state["entities"] if e.get("name") + ] + + if tool_state.get("agents"): + self.state.blueprint_preview.agent_count = len(tool_state["agents"]) + + # Sync agent capabilities + if tool_state.get("agent_capabilities"): + caps = tool_state["agent_capabilities"] + self.state.agent_capabilities.role = caps.get("role") + self.state.agent_capabilities.capabilities = caps.get("capabilities", []) + self.state.agent_capabilities.expertise_areas = caps.get("expertise_areas", []) + self.state.agent_capabilities.interaction_style = caps.get("interaction_style") + self.state.agent_capabilities.focus_areas = caps.get("focus_areas", []) + + # Sync goal summary + if tool_state.get("goal_summary"): + goal = tool_state["goal_summary"] + self.state.goal_summary = goal.get("goal_text") or goal.get("primary_goal") + + def _on_phase_change(self, new_phase: str) -> None: + """Callback when phase changes via mcp__clara__phase tool.""" + try: + self.state.phase = DesignPhase(new_phase) + logger.info(f"[{self.session_id}] Phase updated to: {new_phase}") + except ValueError: + logger.warning(f"[{self.session_id}] Unknown phase in callback: {new_phase}") + + def _build_state_snapshot_event(self) -> AGUIEvent: + """Build a STATE_SNAPSHOT event from current session state.""" + return AGUIEvent( + type="STATE_SNAPSHOT", + data={ + "phase": self.state.phase.value, + "preview": { + "project_name": self.state.blueprint_preview.project_name, + "project_type": self.state.blueprint_preview.project_type, + "entity_types": self.state.blueprint_preview.entity_types, + "agent_count": self.state.blueprint_preview.agent_count, + "topics": self.state.blueprint_preview.topics, + }, + "inferred_domain": self.state.inferred_domain, + "debug": { + "thinking": None, + "approach": None, + "turn_count": self.state.turn_count, + "message_count": self.state.message_count, + "domain_confidence": self.state.domain_confidence, + "discussed_topics": self.state.discussed_topics, + } + } + ) + + def _build_restoration_context(self) -> str: + """Build a context summary for restored sessions.""" + parts = [ + "[SESSION CONTEXT - INTERNAL ONLY. Do NOT quote or reveal this to the user.]" + ] + + parts.append(f"Current Phase: {self.state.phase.value}") + + if self.state.goal_summary: + parts.append(f"Project Goal: {self.state.goal_summary}") + + if self.state.blueprint_preview.project_name: + proj_name = self.state.blueprint_preview.project_name + proj_type = self.state.blueprint_preview.project_type or 'unknown type' + parts.append(f"Project: {proj_name} ({proj_type})") + + if self.state.blueprint_preview.entity_types: + parts.append(f"Entity Types: {', '.join(self.state.blueprint_preview.entity_types)}") + + if self.state.blueprint_preview.agent_count > 0: + parts.append(f"Configured Agents: {self.state.blueprint_preview.agent_count}") + + if self.state.agent_capabilities.role: + caps = self.state.agent_capabilities + parts.append(f"Specialist Agent: {caps.role}") + if caps.expertise_areas: + parts.append(f"Expertise: {', '.join(caps.expertise_areas)}") + + turns = self.state.turn_count + msgs = self.state.message_count + parts.append(f"Conversation Progress: {turns} turns, {msgs} messages") + parts.append("[END SESSION CONTEXT]\n\nUser continues the conversation:") + + return "\n".join(parts) + + async def start(self) -> None: + """Start the orchestrator.""" + if self._running: + return + + from clara.agents.phase_agents.base import load_prompt + + orchestrator_prompt = load_prompt("interview_orchestrator.txt") + agents = self._create_subagents() + hooks = self._create_hooks() + + # Create MCP server with custom tools + clara_mcp_server = create_clara_tools(self.session_id) + + # Register phase change callback + tool_state = get_session_state(self.session_id) + tool_state["_on_phase_change"] = self._on_phase_change + + options = ClaudeAgentOptions( + permission_mode="bypassPermissions", + system_prompt=orchestrator_prompt, + allowed_tools=["Task"] + CLARA_TOOL_NAMES, + agents=agents, + hooks=hooks, + mcp_servers={"clara": clara_mcp_server}, + model="sonnet" + ) + + self._client = ClaudeSDKClient(options=options) + await self._client.__aenter__() + self._running = True + logger.info(f"[{self.session_id}] Orchestrator started") + + async def stop(self) -> None: + """Stop the orchestrator.""" + if self._client and self._running: + await self._client.__aexit__(None, None, None) + self._running = False + from clara.agents.tools import clear_session_state + clear_session_state(self.session_id) + logger.info(f"[{self.session_id}] Orchestrator stopped") + + async def send_message(self, message: str) -> AsyncIterator[AGUIEvent]: + """Send a message and stream the response as AG-UI events. + + This is the main entry point for processing messages. + The orchestrator routes to the current phase agent and streams events. + """ + if not self._running or not self._client: + raise RuntimeError("Orchestrator not started") + + # Reset turn state + agent = self._get_phase_agent(self.state.phase) + agent.reset_turn_state() + + self.state.message_count += 1 + self.state.turn_count += 1 + + # Sync state from tools + self._sync_state_from_tools() + + # Build message with restoration context if needed + actual_message = message + if self._restored and not self._first_message_sent: + context = self._build_restoration_context() + actual_message = f"{context}\n\n{message}" + self._first_message_sent = True + logger.info(f"[{self.session_id}] Prepending restoration context") + + # Emit state snapshot at start + yield self._build_state_snapshot_event() + + # Send to SDK + await self._client.query(prompt=actual_message) + + # Helper to drain queued events + async def drain_queue(): + while not self._event_queue.empty(): + try: + event = self._event_queue.get_nowait() + yield event + except asyncio.QueueEmpty: + break + + # Stream response + current_text = "" + async for msg in self._client.receive_response(): + # Drain hook events + async for event in drain_queue(): + yield event + + msg_type = type(msg).__name__ + + if msg_type == 'AssistantMessage': + if hasattr(msg, 'content'): + for block in msg.content: + if hasattr(block, 'text'): + new_text = block.text + if new_text and new_text != current_text: + if current_text and new_text.startswith(current_text): + delta = new_text[len(current_text):] + else: + delta = new_text + current_text = new_text + if delta: + yield AGUIEvent( + type="TEXT_MESSAGE_CONTENT", + data={"delta": delta} + ) + + elif msg_type == 'ToolUseMessage': + if hasattr(msg, 'name') and hasattr(msg, 'input'): + yield AGUIEvent( + type="TOOL_CALL_START", + data={"tool": msg.name, "input": msg.input} + ) + + elif msg_type == 'ToolResultMessage': + yield AGUIEvent(type="TOOL_CALL_END", data={}) + + # Final drain + async for event in drain_queue(): + yield event + + # Sync state after tool calls + self._sync_state_from_tools() + + # Emit final state snapshot + yield self._build_state_snapshot_event() + yield AGUIEvent(type="TEXT_MESSAGE_END", data={}) diff --git a/src/backend/clara/agents/phase_agents/__init__.py b/src/backend/clara/agents/phase_agents/__init__.py new file mode 100644 index 0000000..621ce6c --- /dev/null +++ b/src/backend/clara/agents/phase_agents/__init__.py @@ -0,0 +1,16 @@ +"""Phase agents for the Design Assistant. + +Each phase agent handles a specific phase of the blueprint design process. +""" + +from clara.agents.phase_agents.agent_configuration import AgentConfigurationAgent +from clara.agents.phase_agents.base import BasePhaseAgent +from clara.agents.phase_agents.blueprint_design import BlueprintDesignAgent +from clara.agents.phase_agents.goal_understanding import GoalUnderstandingAgent + +__all__ = [ + "BasePhaseAgent", + "GoalUnderstandingAgent", + "AgentConfigurationAgent", + "BlueprintDesignAgent", +] diff --git a/src/backend/clara/agents/phase_agents/agent_configuration.py b/src/backend/clara/agents/phase_agents/agent_configuration.py new file mode 100644 index 0000000..5b13aa9 --- /dev/null +++ b/src/backend/clara/agents/phase_agents/agent_configuration.py @@ -0,0 +1,68 @@ +"""Phase 2: Agent Configuration Agent. + +This agent handles the automatic agent configuration phase where we: +1. Analyze the confirmed goal from Phase 1 +2. Design a specialist agent persona (role, capabilities, expertise) +3. Display the agent card to the user +4. Transition to Phase 3 +""" + +import asyncio +import logging +from typing import Any + +from clara.agents.phase_agents.base import AGUIEvent, BasePhaseAgent, load_prompt + +logger = logging.getLogger(__name__) + + +class AgentConfigurationAgent(BasePhaseAgent): + """Agent for Phase 2: Agent Configuration. + + This is a mostly automatic phase: + 1. LLM analyzes the goal + 2. LLM designs a specialist agent + 3. Agent card is displayed to user + 4. Transition to Phase 3 + """ + + phase = "agent_configuration" + + tools = [ + "mcp__clara__agent_summary", + "mcp__clara__phase", + "mcp__clara__get_prompt", + "mcp__clara__hydrate_phase3", + "mcp__clara__request_selection_list", + "mcp__clara__request_data_table", + "mcp__clara__request_process_map", + ] + + def __init__(self, session_id: str, event_queue: asyncio.Queue[AGUIEvent]): + super().__init__(session_id, event_queue) + self._base_prompt: str | None = None + + def get_prompt(self, session_state: dict[str, Any]) -> str: + """Get the Phase 2 prompt, hydrated with the goal from Phase 1. + + The prompt has a {{goal}} placeholder that gets filled in. + """ + if self._base_prompt is None: + self._base_prompt = load_prompt("phase2_agent_configuration.txt") + + # Get the goal from session state + goal = "" + if session_state.get("goal_summary"): + goal_summary = session_state["goal_summary"] + goal = goal_summary.get("goal_text") or goal_summary.get("primary_goal", "") + + # Hydrate the prompt + return self._base_prompt.replace("{{goal}}", goal) + + def get_description(self) -> str: + """Get the agent description for the SDK.""" + return ( + "Handles Phase 2: Agent Configuration. " + "First call mcp__clara__get_prompt to get hydrated instructions with the goal. " + "Then configure the specialist agent and call mcp__clara__hydrate_phase3." + ) diff --git a/src/backend/clara/agents/phase_agents/base.py b/src/backend/clara/agents/phase_agents/base.py new file mode 100644 index 0000000..015c0df --- /dev/null +++ b/src/backend/clara/agents/phase_agents/base.py @@ -0,0 +1,229 @@ +"""Base class for phase agents. + +Each phase agent encapsulates: +- The prompt for that phase +- The tools available in that phase +- Hook handlers for UI event emission +- Transition logic to detect when to move to the next phase +""" + +import asyncio +import logging +from abc import ABC, abstractmethod +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from claude_agent_sdk import HookMatcher + +logger = logging.getLogger(__name__) + +# Paths to prompt files +PROMPTS_DIR = Path(__file__).parent.parent / "prompts" + + +def load_prompt(filename: str) -> str: + """Load a prompt from the prompts directory.""" + prompt_path = PROMPTS_DIR / filename + with open(prompt_path, encoding="utf-8") as f: + return f.read().strip() + + +@dataclass +class AGUIEvent: + """Base AG-UI event structure.""" + type: str + data: dict[str, Any] = field(default_factory=dict) + + +class BasePhaseAgent(ABC): + """Base class for phase agents. + + Each phase agent is responsible for: + 1. Providing the prompt for its phase + 2. Defining the tools available in its phase + 3. Handling hooks for UI event emission + 4. Processing messages and yielding AG-UI events + """ + + # Phase identifier (e.g., "goal_understanding") + phase: str + + # Tools available to this phase + tools: list[str] + + def __init__(self, session_id: str, event_queue: asyncio.Queue[AGUIEvent]): + """Initialize the phase agent. + + Args: + session_id: The session ID for logging and state access + event_queue: Queue for emitting AG-UI events from hooks + """ + self.session_id = session_id + self._event_queue = event_queue + self._ui_emitted_in_turn = False + + @abstractmethod + def get_prompt(self, session_state: dict[str, Any]) -> str: + """Get the hydrated prompt for this phase. + + Args: + session_state: Current session state (from tools.py) + + Returns: + The prompt string, with any placeholders filled in + """ + pass + + def get_hooks(self) -> dict[str, list[HookMatcher]]: + """Get hooks for this phase. + + Returns a dictionary mapping hook types to lists of HookMatcher objects. + Default implementation provides common UI tool hooks. + """ + return { + 'PreToolUse': [ + HookMatcher( + matcher=None, # Match all tools + hooks=[self._pre_tool_hook] + ) + ], + 'PostToolUse': [ + HookMatcher( + matcher=None, + hooks=[self._post_tool_hook] + ) + ] + } + + async def _pre_tool_hook( + self, + input_data: dict[str, Any], + tool_use_id: str | None, + context: Any + ) -> dict[str, Any]: + """Default pre-tool hook that handles UI component emission. + + Subclasses can override this to add phase-specific handling. + """ + from clara.agents.tools import ensure_other_option, sanitize_ask_options + + tool_name = input_data.get("tool_name", input_data.get("name", "unknown")) + tool_input = input_data.get("tool_input", input_data.get("input", {})) + + logger.info(f"[{self.session_id}] PreToolUse: {tool_name}") + + await self._event_queue.put(AGUIEvent( + type="TOOL_CALL_START", + data={"tool": tool_name, "input": tool_input} + )) + + # Handle ask tool - emit CUSTOM event with UI component + if tool_name == "mcp__clara__ask": + options = sanitize_ask_options(tool_input.get("options", [])) + ui_component = { + "type": "user_input_required", + "question": tool_input.get("question", ""), + "options": options, + "multi_select": tool_input.get("multi_select", False), + } + await self._event_queue.put(AGUIEvent( + type="CUSTOM", + data={"name": "clara:ask", "value": ui_component} + )) + self._ui_emitted_in_turn = True + logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:ask") + + # Handle selection list + if tool_name == "mcp__clara__request_selection_list": + options = ensure_other_option( + sanitize_ask_options(tool_input.get("options", [])) + ) + ui_component = { + "type": "user_input_required", + "question": tool_input.get("question", ""), + "options": options, + "multi_select": tool_input.get("multi_select", False), + } + await self._event_queue.put(AGUIEvent( + type="CUSTOM", + data={"name": "clara:ask", "value": ui_component} + )) + self._ui_emitted_in_turn = True + logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:ask (selection)") + + # Handle data table + if tool_name == "mcp__clara__request_data_table": + ui_component = { + "type": "data_table", + "title": tool_input.get("title", "Data Table"), + "columns": tool_input.get("columns", []), + "min_rows": tool_input.get("min_rows", 3), + "starter_rows": tool_input.get("starter_rows", 3), + "input_modes": tool_input.get("input_modes", ["paste", "inline"]), + "summary_prompt": tool_input.get("summary_prompt", ""), + } + await self._event_queue.put(AGUIEvent( + type="CUSTOM", + data={"name": "clara:data_table", "value": ui_component} + )) + self._ui_emitted_in_turn = True + logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:data_table") + + # Handle process map + if tool_name == "mcp__clara__request_process_map": + ui_component = { + "type": "process_map", + "title": tool_input.get("title", "Process Map"), + "required_fields": tool_input.get("required_fields", []), + "edge_types": tool_input.get("edge_types", []), + "min_steps": tool_input.get("min_steps", 3), + "seed_nodes": tool_input.get("seed_nodes", []), + } + await self._event_queue.put(AGUIEvent( + type="CUSTOM", + data={"name": "clara:process_map", "value": ui_component} + )) + self._ui_emitted_in_turn = True + logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:process_map") + + # Handle prompt editor + if tool_name == "mcp__clara__prompt_editor": + ui_component = { + "type": "prompt_editor", + "title": tool_input.get("title", "System Prompt"), + "prompt": tool_input.get("prompt", ""), + "description": tool_input.get("description", ""), + } + await self._event_queue.put(AGUIEvent( + type="CUSTOM", + data={"name": "clara:prompt_editor", "value": ui_component} + )) + self._ui_emitted_in_turn = True + logger.info(f"[{self.session_id}] Emitted CUSTOM event clara:prompt_editor") + + return {} # No modifications to tool behavior + + async def _post_tool_hook( + self, + input_data: dict[str, Any], + tool_use_id: str | None, + context: Any + ) -> dict[str, Any]: + """Default post-tool hook.""" + tool_name = input_data.get("tool_name", "unknown") + logger.debug(f"[{self.session_id}] PostToolUse: {tool_name} completed") + await self._event_queue.put(AGUIEvent( + type="TOOL_CALL_END", + data={"tool": tool_name} + )) + return {} + + def reset_turn_state(self) -> None: + """Reset per-turn state. Called at the start of each message.""" + self._ui_emitted_in_turn = False + + @property + def ui_emitted(self) -> bool: + """Whether a UI component was emitted this turn.""" + return self._ui_emitted_in_turn diff --git a/src/backend/clara/agents/phase_agents/blueprint_design.py b/src/backend/clara/agents/phase_agents/blueprint_design.py new file mode 100644 index 0000000..333d310 --- /dev/null +++ b/src/backend/clara/agents/phase_agents/blueprint_design.py @@ -0,0 +1,90 @@ +"""Phase 3: Blueprint Design Agent. + +This agent handles the interactive blueprint design phase where we: +1. Ask the user what knowledge to extract +2. Ask about details for each knowledge area +3. Ask about interview topics and style +4. Build the blueprint with entities and agents +5. Show the prompt editor for the interviewer prompt +6. Get final approval and transition to complete +""" + +import asyncio +import logging +from typing import Any + +from clara.agents.phase_agents.base import AGUIEvent, BasePhaseAgent, load_prompt + +logger = logging.getLogger(__name__) + + +class BlueprintDesignAgent(BasePhaseAgent): + """Agent for Phase 3: Blueprint Design. + + This is an interactive phase with multiple tasks: + 1. What knowledge to extract (multi-select) + 2. Details for each knowledge area + 3. Interview topics + 4. Interview style + 5. Build blueprint (project, entities, agent) + 6. Prompt editor + 7. Final approval + """ + + phase = "blueprint_design" + + tools = [ + "mcp__clara__project", + "mcp__clara__entity", + "mcp__clara__agent", + "mcp__clara__ask", + "mcp__clara__request_selection_list", + "mcp__clara__request_data_table", + "mcp__clara__request_process_map", + "mcp__clara__preview", + "mcp__clara__phase", + "mcp__clara__get_prompt", + "mcp__clara__prompt_editor", + "mcp__clara__get_agent_context", + ] + + def __init__(self, session_id: str, event_queue: asyncio.Queue[AGUIEvent]): + super().__init__(session_id, event_queue) + self._base_prompt: str | None = None + + def get_prompt(self, session_state: dict[str, Any]) -> str: + """Get the Phase 3 prompt, hydrated with goal and agent config. + + The prompt has {{goal}} and {{role}} placeholders that get filled in. + """ + if self._base_prompt is None: + self._base_prompt = load_prompt("phase3_blueprint_design.txt") + + # Get the goal from session state + goal = "" + if session_state.get("goal_summary"): + goal_summary = session_state["goal_summary"] + goal = goal_summary.get("goal_text") or goal_summary.get("primary_goal", "") + + # Get the agent role from session state + role = "" + if session_state.get("agent_capabilities"): + agent_caps = session_state["agent_capabilities"] + role = agent_caps.get("role", "") + + # Hydrate the prompt + prompt = self._base_prompt.replace("{{goal}}", goal) + prompt = prompt.replace("{{role}}", role) + return prompt + + def get_description(self) -> str: + """Get the agent description for the SDK.""" + return ( + "Handles Phase 3: Blueprint Design INTERACTIVELY. " + "First call mcp__clara__get_prompt to get hydrated instructions. " + "This agent MUST use mcp__clara__ask to collect user input " + "before building. It should NOT build entities/agents until " + "user confirms via ask tool responses. " + "Use mcp__clara__prompt_editor for editing prompts. " + "Use mcp__clara__get_agent_context for context files." + ) diff --git a/src/backend/clara/agents/phase_agents/goal_understanding.py b/src/backend/clara/agents/phase_agents/goal_understanding.py new file mode 100644 index 0000000..3556174 --- /dev/null +++ b/src/backend/clara/agents/phase_agents/goal_understanding.py @@ -0,0 +1,61 @@ +"""Phase 1: Goal Understanding Agent. + +This agent handles the initial goal discovery phase where we: +1. Ask the user for their project goal +2. Confirm the goal with a simple Yes/No +3. Transition to Phase 2 once confirmed +""" + +import asyncio +import logging +from typing import Any + +from clara.agents.phase_agents.base import AGUIEvent, BasePhaseAgent, load_prompt + +logger = logging.getLogger(__name__) + + +class GoalUnderstandingAgent(BasePhaseAgent): + """Agent for Phase 1: Goal Understanding. + + This is a simple two-turn phase: + 1. User provides goal → Agent confirms + 2. User confirms Yes/No → Agent transitions or asks again + """ + + phase = "goal_understanding" + + tools = [ + "mcp__clara__ask", + "mcp__clara__request_selection_list", + "mcp__clara__request_data_table", + "mcp__clara__request_process_map", + "mcp__clara__project", + "mcp__clara__save_goal_summary", + "mcp__clara__hydrate_phase2", + "mcp__clara__phase", + "mcp__clara__get_prompt", + ] + + def __init__(self, session_id: str, event_queue: asyncio.Queue[AGUIEvent]): + super().__init__(session_id, event_queue) + self._prompt: str | None = None + + def get_prompt(self, session_state: dict[str, Any]) -> str: + """Get the Phase 1 prompt. + + Phase 1 uses a static prompt since there's no prior context needed. + """ + if self._prompt is None: + self._prompt = load_prompt("phase1_goal_understanding.txt") + return self._prompt + + def get_description(self) -> str: + """Get the agent description for the SDK.""" + return ( + "Handles Phase 1: Goal Confirmation. " + "Use this agent to discover the user's project through " + "natural conversation. It will explore the core discovery areas " + "and use mcp__clara__ask for structured choices. " + "After completing, call mcp__clara__hydrate_phase2." + ) From 970088208476a5eddc8a0007097b10f6f3a2158c Mon Sep 17 00:00:00 2001 From: Mathew Date: Tue, 23 Dec 2025 14:36:11 -0800 Subject: [PATCH 2/2] refactor(agents): Remove design_assistant.py, consolidate SessionManager into orchestrator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Delete design_assistant.py legacy wrapper module - Move SessionManager class and session_manager singleton to orchestrator.py - Update imports in main.py and design_sessions.py to use orchestrator module - No functional changes, just module consolidation 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- src/backend/clara/agents/design_assistant.py | 244 ------------------- src/backend/clara/agents/orchestrator.py | 134 +++++++++- src/backend/clara/api/design_sessions.py | 2 +- src/backend/clara/main.py | 2 +- 4 files changed, 131 insertions(+), 251 deletions(-) delete mode 100644 src/backend/clara/agents/design_assistant.py diff --git a/src/backend/clara/agents/design_assistant.py b/src/backend/clara/agents/design_assistant.py deleted file mode 100644 index 6508ad4..0000000 --- a/src/backend/clara/agents/design_assistant.py +++ /dev/null @@ -1,244 +0,0 @@ -"""Design Assistant using Claude Agent SDK. - -This module provides the public interface for Clara's Design Architect agent. -The actual implementation is delegated to the DesignOrchestrator which routes -messages to specialized phase agents. - -This file maintains backward compatibility with existing code that uses -DesignAssistantSession and session_manager. -""" - -import logging -from collections.abc import AsyncIterator -from dataclasses import dataclass -from typing import Any - -from clara.agents.orchestrator import ( - AgentCapabilities, - AGUIEvent, - BlueprintPreview, - DesignOrchestrator, - DesignPhase, - DesignSessionState, -) -from clara.agents.tools import get_session_state - -logger = logging.getLogger(__name__) - - -# Re-export types for backward compatibility -__all__ = [ - "DesignPhase", - "BlueprintPreview", - "AgentCapabilities", - "DesignSessionState", - "AGUIEvent", - "DesignAssistantSession", - "DesignAssistantManager", - "session_manager", -] - - -class DesignAssistantSession: - """Thin wrapper around DesignOrchestrator for backward compatibility. - - This class maintains the same public interface as the original - DesignAssistantSession but delegates all work to the orchestrator. - """ - - def __init__(self, session_id: str, project_id: str): - self._orchestrator = DesignOrchestrator(session_id, project_id) - - @property - def session_id(self) -> str: - return self._orchestrator.session_id - - @property - def project_id(self) -> str: - return self._orchestrator.project_id - - @property - def state(self) -> DesignSessionState: - return self._orchestrator.state - - @state.setter - def state(self, value: DesignSessionState) -> None: - self._orchestrator.state = value - - @property - def _restored(self) -> bool: - return self._orchestrator._restored - - @_restored.setter - def _restored(self, value: bool) -> None: - self._orchestrator._restored = value - - @property - def _first_message_sent(self) -> bool: - return self._orchestrator._first_message_sent - - @_first_message_sent.setter - def _first_message_sent(self, value: bool) -> None: - self._orchestrator._first_message_sent = value - - async def start(self) -> None: - """Start the design assistant session.""" - await self._orchestrator.start() - - async def stop(self) -> None: - """Stop the design assistant session.""" - await self._orchestrator.stop() - - async def send_message(self, message: str) -> AsyncIterator[AGUIEvent]: - """Send a message and stream the response as AG-UI events.""" - async for event in self._orchestrator.send_message(message): - yield event - - -@dataclass -class _SessionWrapper: - """Internal wrapper to track both orchestrator and session.""" - session: DesignAssistantSession - orchestrator: DesignOrchestrator - - -class DesignAssistantManager: - """Manages multiple design assistant sessions.""" - - def __init__(self): - self._sessions: dict[str, DesignAssistantSession] = {} - # Import here to avoid circular imports - from clara.db.session import async_session_maker - self._db_session_maker = async_session_maker - - async def get_or_create_session( - self, - session_id: str, - project_id: str, - initial_blueprint_state: dict | None = None - ) -> DesignAssistantSession: - """Get an existing session or create a new one. - - Args: - session_id: The session ID - project_id: The project ID - initial_blueprint_state: Optional blueprint state to initialize with - (used for add-agent mode to preserve existing agents) - """ - if session_id not in self._sessions: - session = DesignAssistantSession(session_id, project_id) - await session.start() - - # If initial blueprint state provided, populate the tools state - if initial_blueprint_state: - tool_state = get_session_state(session_id) - tool_state["project"] = initial_blueprint_state.get("project") - tool_state["entities"] = initial_blueprint_state.get("entities", []) - tool_state["agents"] = initial_blueprint_state.get("agents", []) - tool_state["phase"] = DesignPhase.AGENT_CONFIGURATION.value - - # Update session state to reflect the blueprint - if initial_blueprint_state.get("project"): - proj = initial_blueprint_state["project"] - session.state.blueprint_preview.project_name = proj.get("name") - session.state.blueprint_preview.project_type = proj.get("type") - session.state.inferred_domain = proj.get("domain") - session.state.blueprint_preview.agent_count = len( - initial_blueprint_state.get("agents", []) - ) - session.state.blueprint_preview.entity_types = [ - e.get("name") for e in initial_blueprint_state.get("entities", []) - ] - session.state.phase = DesignPhase.AGENT_CONFIGURATION - - logger.info( - f"Initialized session {session_id} with existing blueprint " - f"({len(initial_blueprint_state.get('agents', []))} agents)" - ) - - self._sessions[session_id] = session - return self._sessions[session_id] - - async def restore_session( - self, - session_id: str, - project_id: str, - db_session: Any # DesignSession model instance - ) -> DesignAssistantSession: - """Restore a session from database state. - - This creates a new in-memory session and populates it with - the state from the database record. - """ - # If session already exists in memory, return it - if session_id in self._sessions: - return self._sessions[session_id] - - # Create new in-memory session - session = DesignAssistantSession(session_id, project_id) - await session.start() - - # Mark as restored so context will be prepended on first message - session._restored = True - - # Restore state from DB - if db_session.phase: - session.state.phase = DesignPhase(db_session.phase) - session.state.turn_count = db_session.turn_count or 0 - session.state.message_count = db_session.message_count or 0 - - # Restore blueprint preview state - blueprint_state = db_session.blueprint_state or {} - if blueprint_state.get("project"): - project = blueprint_state["project"] - session.state.blueprint_preview.project_name = project.get("name") - session.state.blueprint_preview.project_type = project.get("type") - session.state.blueprint_preview.entity_types = [ - e.get("name") for e in blueprint_state.get("entities", []) - ] - session.state.blueprint_preview.agent_count = len(blueprint_state.get("agents", [])) - - # Restore goal summary (handle both goal_text and primary_goal keys) - if db_session.goal_summary: - goal = db_session.goal_summary - session.state.goal_summary = goal.get("goal_text") or goal.get("primary_goal") - - # Restore agent capabilities - if db_session.agent_capabilities: - caps = db_session.agent_capabilities - session.state.agent_capabilities.role = caps.get("role") - session.state.agent_capabilities.capabilities = caps.get("capabilities", []) - session.state.agent_capabilities.expertise_areas = caps.get("expertise_areas", []) - session.state.agent_capabilities.interaction_style = caps.get("interaction_style") - - # Restore tools state - tool_state = get_session_state(session_id) - tool_state["project"] = blueprint_state.get("project") - tool_state["entities"] = blueprint_state.get("entities", []) - tool_state["agents"] = blueprint_state.get("agents", []) - tool_state["phase"] = db_session.phase - tool_state["goal_summary"] = db_session.goal_summary - tool_state["agent_capabilities"] = db_session.agent_capabilities - - self._sessions[session_id] = session - logger.info(f"Restored session {session_id} from database (phase: {db_session.phase})") - return session - - async def get_session(self, session_id: str) -> DesignAssistantSession | None: - """Get an existing session.""" - return self._sessions.get(session_id) - - async def close_session(self, session_id: str) -> None: - """Close and remove a session.""" - if session_id in self._sessions: - session = self._sessions.pop(session_id) - await session.stop() - - async def close_all(self) -> None: - """Close all active sessions.""" - for session_id in list(self._sessions.keys()): - await self.close_session(session_id) - - -# Global session manager -session_manager = DesignAssistantManager() diff --git a/src/backend/clara/agents/orchestrator.py b/src/backend/clara/agents/orchestrator.py index a790880..4d824ca 100644 --- a/src/backend/clara/agents/orchestrator.py +++ b/src/backend/clara/agents/orchestrator.py @@ -1,10 +1,15 @@ -"""Thin Orchestrator for the Design Assistant. +"""Design Assistant Orchestrator and Session Management. + +This module provides: +1. DesignOrchestrator - Routes messages to phase agents +2. SessionManager - Manages multiple design sessions +3. session_manager - Global singleton for session management The orchestrator is responsible for: -1. Looking up the current phase from session state -2. Routing messages to the correct phase agent -3. Streaming events from the phase agent -4. Handling phase transitions +- Looking up the current phase from session state +- Routing messages to the correct phase agent +- Streaming events from the phase agent +- Handling phase transitions It does NOT contain conversation logic - that lives in the phase agents. """ @@ -420,3 +425,122 @@ async def drain_queue(): # Emit final state snapshot yield self._build_state_snapshot_event() yield AGUIEvent(type="TEXT_MESSAGE_END", data={}) + + +class SessionManager: + """Manages multiple design assistant sessions.""" + + def __init__(self): + self._sessions: dict[str, DesignOrchestrator] = {} + from clara.db.session import async_session_maker + self._db_session_maker = async_session_maker + + async def get_or_create_session( + self, + session_id: str, + project_id: str, + initial_blueprint_state: dict | None = None + ) -> DesignOrchestrator: + """Get an existing session or create a new one.""" + if session_id not in self._sessions: + session = DesignOrchestrator(session_id, project_id) + await session.start() + + if initial_blueprint_state: + tool_state = get_session_state(session_id) + tool_state["project"] = initial_blueprint_state.get("project") + tool_state["entities"] = initial_blueprint_state.get("entities", []) + tool_state["agents"] = initial_blueprint_state.get("agents", []) + tool_state["phase"] = DesignPhase.AGENT_CONFIGURATION.value + + if initial_blueprint_state.get("project"): + proj = initial_blueprint_state["project"] + session.state.blueprint_preview.project_name = proj.get("name") + session.state.blueprint_preview.project_type = proj.get("type") + session.state.inferred_domain = proj.get("domain") + session.state.blueprint_preview.agent_count = len( + initial_blueprint_state.get("agents", []) + ) + session.state.blueprint_preview.entity_types = [ + e.get("name") for e in initial_blueprint_state.get("entities", []) + ] + session.state.phase = DesignPhase.AGENT_CONFIGURATION + + logger.info( + f"Initialized session {session_id} with existing blueprint " + f"({len(initial_blueprint_state.get('agents', []))} agents)" + ) + + self._sessions[session_id] = session + return self._sessions[session_id] + + async def restore_session( + self, + session_id: str, + project_id: str, + db_session: Any + ) -> DesignOrchestrator: + """Restore a session from database state.""" + if session_id in self._sessions: + return self._sessions[session_id] + + session = DesignOrchestrator(session_id, project_id) + await session.start() + session._restored = True + + if db_session.phase: + session.state.phase = DesignPhase(db_session.phase) + session.state.turn_count = db_session.turn_count or 0 + session.state.message_count = db_session.message_count or 0 + + blueprint_state = db_session.blueprint_state or {} + if blueprint_state.get("project"): + project = blueprint_state["project"] + session.state.blueprint_preview.project_name = project.get("name") + session.state.blueprint_preview.project_type = project.get("type") + session.state.blueprint_preview.entity_types = [ + e.get("name") for e in blueprint_state.get("entities", []) + ] + session.state.blueprint_preview.agent_count = len(blueprint_state.get("agents", [])) + + if db_session.goal_summary: + goal = db_session.goal_summary + session.state.goal_summary = goal.get("goal_text") or goal.get("primary_goal") + + if db_session.agent_capabilities: + caps = db_session.agent_capabilities + session.state.agent_capabilities.role = caps.get("role") + session.state.agent_capabilities.capabilities = caps.get("capabilities", []) + session.state.agent_capabilities.expertise_areas = caps.get("expertise_areas", []) + session.state.agent_capabilities.interaction_style = caps.get("interaction_style") + + tool_state = get_session_state(session_id) + tool_state["project"] = blueprint_state.get("project") + tool_state["entities"] = blueprint_state.get("entities", []) + tool_state["agents"] = blueprint_state.get("agents", []) + tool_state["phase"] = db_session.phase + tool_state["goal_summary"] = db_session.goal_summary + tool_state["agent_capabilities"] = db_session.agent_capabilities + + self._sessions[session_id] = session + logger.info(f"Restored session {session_id} from database (phase: {db_session.phase})") + return session + + async def get_session(self, session_id: str) -> DesignOrchestrator | None: + """Get an existing session.""" + return self._sessions.get(session_id) + + async def close_session(self, session_id: str) -> None: + """Close and remove a session.""" + if session_id in self._sessions: + session = self._sessions.pop(session_id) + await session.stop() + + async def close_all(self) -> None: + """Close all active sessions.""" + for session_id in list(self._sessions.keys()): + await self.close_session(session_id) + + +# Global session manager singleton +session_manager = SessionManager() diff --git a/src/backend/clara/api/design_sessions.py b/src/backend/clara/api/design_sessions.py index 87d733c..b880d93 100644 --- a/src/backend/clara/api/design_sessions.py +++ b/src/backend/clara/api/design_sessions.py @@ -16,7 +16,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession -from clara.agents.design_assistant import AGUIEvent, session_manager +from clara.agents.orchestrator import AGUIEvent, session_manager from clara.db.models import DesignPhase, DesignSession, DesignSessionStatus from clara.db.session import get_db diff --git a/src/backend/clara/main.py b/src/backend/clara/main.py index 0342010..0a8dca3 100644 --- a/src/backend/clara/main.py +++ b/src/backend/clara/main.py @@ -6,7 +6,7 @@ from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware -from clara.agents.design_assistant import session_manager +from clara.agents.orchestrator import session_manager from clara.agents.simulation_agent import simulation_manager from clara.api.context_files import router as context_files_router from clara.api.design_sessions import router as design_sessions_router