# From: apps/api/app/agents/core/agent.py:117-161

async def call_agent(
    request: MessageRequestWithHistory,
    conversation_id: str,
    user: dict,
    user_time: datetime,
    user_model_config: Optional[ModelConfig] = None,
) -> AsyncGenerator[str, None]:
    """
    Execute agent in streaming mode for interactive chat.
    
    The executor agent:
    - Maintains conversation context
    - Analyzes user intent
    - Decides on tool usage
    - Delegates to subagents when needed
    - Coordinates multi-step workflows
    """

# From: apps/api/app/agents/core/subagents/handoff_tools.py

async def handoff(
    subagent_id: str,     # Which subagent to use
    task: str,            # What they should do
    config: RunnableConfig
) -> str:
    """
    Delegate a task to a specialized subagent.
    
    Example:
    handoff(
        subagent_id="gmail",
        task="Search for emails from [email protected] in last 7 days"
    )
    
    The subagent:
    1. Receives full context
    2. Uses its specialized tools
    3. Returns results to executor
    """

# From: apps/api/app/agents/core/graph_builder/build_graph.py

def create_agent(
    llm: LanguageModelLike,
    tool_registry: dict,
    agent_name: str,
    pre_model_hooks: List[Callable],
    end_graph_hooks: List[Callable]
):
    """
    Build an agent graph with:
    - Pre-processing nodes (message filtering, prompt management)
    - LLM execution node
    - Tool execution nodes
    - Post-processing nodes (memory learning, follow-up actions)
    
    Returns compiled StateGraph ready for execution.
    """

# From: apps/api/app/agents/core/subagents/base_subagent.py:88-94

# Every agent has this node pipeline:
pre_model_hooks = [
    filter_messages_node,         # Remove irrelevant messages
    manage_system_prompts_node,   # Inject memories & skills
    trim_messages_node,           # Keep context within limits
]

end_graph_hooks = [
    memory_learning_node,         # Learn from execution
    follow_up_actions_node,       # Suggest next steps (main agent only)
]

# From: apps/api/app/agents/core/state.py

class State(BaseModel):
    query: str                        # User's current request
    messages: List[AnyMessage]        # Conversation history
    current_datetime: str             # For time-aware operations
    mem0_user_id: str                 # Memory namespace
    memories: List[str]               # Retrieved user memories
    memories_stored: bool             # Tracking flag
    conversation_id: str              # Thread identifier

# From: apps/api/app/agents/core/graph_builder/checkpointer_manager.py

class CheckpointerManager:
    """
    Manages PostgreSQL-backed state persistence.
    
    Enables:
    - Resume interrupted conversations
    - Time-travel debugging
    - Branching conversations
    - State recovery after errors
    """
    
    async def get_checkpointer(self):
        """Returns PostgreSQL checkpointer for state storage."""

# From: apps/api/app/agents/core/graph_manager.py

class GraphManager:
    """Singleton manager for agent graphs."""
    
    _graphs: Dict[str, CompiledGraph] = {}
    
    @classmethod
    async def get_graph(cls, agent_name: str) -> CompiledGraph:
        """
        Get or create agent graph.
        
        Graphs are compiled once and reused for efficiency.
        Each user execution uses same graph with different config.
        """

# From: apps/api/app/helpers/agent_helpers.py

def build_agent_config(
    conversation_id: str,
    user: dict,
    user_time: datetime,
    user_model_config: Optional[ModelConfig],
    agent_name: str,
) -> dict:
    """
    Build configuration for agent execution.
    
    Config includes:
    - user_id: For memory and permissions
    - thread_id: For conversation tracking
    - user_time: For time-aware operations
    - user_timezone: For scheduling
    - model_config: LLM preferences
    - agent_name: Which agent is executing
    """

# From: apps/api/app/agents/tools/core/registry.py:159-192

class ToolRegistry:
    """Central registry managing all agent tools."""
    
    def _add_category(
        self,
        name: str,              # "gmail"
        tools: List[BaseTool],
        space: str = "general", # Namespace for isolation
        require_integration: bool = False,
        is_delegated: bool = False  # Has subagent?
    ):
        """
        Register a category of tools.
        
        Delegated categories route to subagents.
        Non-delegated categories use direct tool execution.
        """

# From: apps/api/app/agents/prompts/

# Each subagent has custom prompts
PROMPTS = {
    "gmail_agent": {
        "system": "You are a Gmail expert. You understand email etiquette, ...",
        "skills": "Email composition patterns, threading, ...",
        "examples": "When user says 'email client', search for..."
    },
    
    "calendar_agent": {
        "system": "You are a scheduling expert. You understand availability, ...",
        "skills": "Time conflict resolution, recurring events, ...",
        "examples": "When user says 'next Tuesday', calculate..."
    },
    
    # ... prompts for each subagent
}

# From: apps/api/app/agents/core/nodes/manage_system_prompts.py

async def manage_system_prompts_node(state, config):
    """
    Enhance base prompt with:
    
    1. User Context:
       - Retrieved memories
       - Current datetime
       - User timezone
    
    2. Agent Skills:
       - Learned procedures for this agent
       - Optimal approaches
       - Known gotchas
    
    3. Tool Availability:
       - Connected integrations
       - Available tool list
    """

# From: apps/api/app/agents/core/agent.py:117-161

async def call_agent(...) -> AsyncGenerator[str, None]:
    """
    Streaming mode for real-time user interactions.
    
    Yields:
    - Intermediate thinking
    - Tool call events
    - Tool results
    - Final response
    - Follow-up suggestions
    
    Used for: Chat interface, voice agents
    """

# From: apps/api/app/agents/core/agent.py:163-193

async def call_agent_silent(...) -> tuple[str, dict]:
    """
    Silent mode for background processing.
    
    Returns:
    - Complete message
    - Tool usage data
    
    Used for: Workflows, scheduled tasks, webhooks
    """

# Agents can execute independent tools in parallel

# Sequential (slow)
gmail_result = await handoff("gmail", "search emails")
calendar_result = await handoff("calendar", "list events")

# Parallel (fast)
results = await asyncio.gather(
    handoff("gmail", "search emails"),
    handoff("calendar", "list events")
)

# Tool registry is cached
@cache
async def get_tool_registry() -> ToolRegistry:
    """Singleton registry - built once, used many times."""

# User capabilities are cached
@cache(ttl=300)  # 5 minutes
async def get_user_integration_capabilities(user_id: str):
    """Avoid repeated DB queries for same user."""

# From: apps/api/app/agents/core/nodes/trim_messages_node.py

async def trim_messages_node(state, config):
    """
    Smart message trimming:
    - Keep system messages (always needed)
    - Keep recent messages (context)
    - Keep tool results (for coherence)
    - Summarize old history (reduce tokens)
    
    Target: Stay under model's context window
    """

# From: apps/api/app/agents/evals/

class AgentEvaluator:
    """
    Test agents against benchmark datasets.
    
    For each integration (Gmail, Slack, GitHub):
    - Load test scenarios
    - Execute agent
    - Compare output to expected results
    - Score: correctness, efficiency, safety
    """
    
    async def evaluate_agent(
        agent_name: str,
        test_dataset: str
    ) -> EvaluationReport:
        """Run full evaluation suite."""

// From: apps/api/app/agents/evals/datasets/gmail.json

[
  {
    "scenario": "Send email to team",
    "user_input": "Email the team about tomorrow's meeting",
    "expected_tools": ["GMAIL_SEND_MESSAGE"],
    "expected_behavior": "Sends to team distribution list",
    "constraints": ["Must include meeting time", "Professional tone"]
  },
  // ... 50+ test scenarios per integration
]

// From: apps/web/src/features/chat/

// Every conversation has full execution trace:
interface ConversationTrace {
  messages: Message[];
  tool_calls: ToolCall[];
  subagent_handoffs: Handoff[];
  state_checkpoints: Checkpoint[];
  memory_updates: MemoryUpdate[];
}

// Inspect exactly what happened at each step

# Agents have robust error handling

try:
    result = await handoff("gmail", task)
except ToolExecutionError as e:
    # Log error, inform user, suggest alternatives
    logger.error(f"Tool execution failed: {e}")
    return "I had trouble with Gmail. Would you like me to try another approach?"

except RateLimitError as e:
    # Automatic retry with backoff
    await asyncio.sleep(e.retry_after)
    result = await handoff("gmail", task)