Upload folder using huggingface_hub

.dockerignore

@@ -0,0 +1,55 @@
+# Environment files with sensitive data
+.env
+.env.*
+!.env.example
+
+# Git
+.git
+.gitignore
+
+# Documentation
+README.md
+*.md
+
+# Docker files
+Dockerfile*
+docker-compose*.yml
+.dockerignore
+
+# Development files
+.vscode/
+.idea/
+
+# OS generated files
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db
+
+# Logs
+*.log
+logs/
+
+# Runtime data that shouldn't be in image
+server/data/
+web/.next/
+
+# Dependencies that are installed in container
+node_modules/
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.Python
+
+# Testing
+coverage/
+.pytest_cache/
+.coverage
+
+# Temporary files
+tmp/
+temp/

.env.example

@@ -0,0 +1,29 @@
+# API Configuration
+API_BASE_URL=https://api.friendli.ai/dedicated/v1
+API_KEY=your_api_key_here
+
+# Composio Configuration
+COMPOSIO_API_KEY=your_composio_api_key_here
+
+# Model Configuration
+INTERACTION_AGENT_MODEL=your_interaction_model_here
+EXECUTION_AGENT_MODEL=your_execution_model_here
+EXECUTION_SEARCH_AGENT_MODEL=your_search_model_here
+SUMMARIZER_MODEL=your_summarizer_model_here
+EMAIL_CLASSIFIER_MODEL=your_classifier_model_here
+
+# Application Configuration
+OPENPOKE_HOST=0.0.0.0
+OPENPOKE_PORT=8001
+OPENPOKE_CORS_ALLOW_ORIGINS=*
+OPENPOKE_ENABLE_DOCS=1
+
+# Web Application Configuration
+NEXT_PUBLIC_API_URL=http://localhost:8001
+
+# Instructions:
+# 1. Copy this file to .env: cp .env.example .env
+# 2. Replace all placeholder values with your actual credentials
+# 3. Never commit the .env file to version control
+# 4. Add .env to your .gitignore file
+# 5. Use docker-compose --env-file .env up for production deployments

.gitignore

@@ -0,0 +1,46 @@
+# Node / Next.js
+node_modules/
+.next/
+web/node_modules/
+web/.next/
+
+# Logs
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+*.log
+.server.log
+.server.pid
+
+# Envs
+.env
+.env.local
+.env.*.local
+
+# OS
+.DS_Store
+
+# Python
+.venv/
+__pycache__/
+*.pyc
+
+# Database files
+*.db
+*.db-shm
+*.db-wal
+
+# Data folder
+server/data/
+data/
+
+# Python virtual environment
+server/venv/
+
+# Build metadata
+*.tsbuildinfo
+/package-lock.json
+
+# Generated documentation / analysis artifacts
+server/repomix-output.txt
+server/plans/

Dockerfile

@@ -0,0 +1,41 @@
+# Use Python 3.11 slim image for smaller size
+FROM python:3.11-slim
+
+# Create a non-root user for security
+RUN groupadd -r appuser && useradd -r -g appuser appuser
+
+# Set working directory
+WORKDIR /app
+
+# Install system dependencies with security considerations
+RUN apt-get update && apt-get install -y \
+    gcc \
+    && rm -rf /var/lib/apt/lists/* \
+    && apt-get clean
+
+# Copy requirements first for better Docker layer caching
+COPY server/requirements.txt .
+
+# Install Python dependencies
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+# Copy server code
+COPY server/ ./server/
+
+# Create necessary directories and set permissions
+RUN mkdir -p /app/logs && \
+    chown -R appuser:appuser /app
+
+# Switch to non-root user
+USER appuser
+
+# Expose port
+EXPOSE 8001
+
+# Add health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:8001/health || exit 1
+
+# Start the server with proper configuration for graceful shutdown
+CMD ["uvicorn", "server.server:app", "--host", "0.0.0.0", "--port", "8001", "--access-log"]

Dockerfile.web

@@ -0,0 +1,38 @@
+# Use Node.js 18 LTS Alpine for smaller size
+FROM node:18-alpine
+
+# Create a non-root user for security
+RUN addgroup -g 1001 -S nodejs && \
+    adduser -S nextjs -u 1001
+
+# Set working directory
+WORKDIR /app
+
+# Copy package files first for better layer caching
+COPY web/package*.json ./
+
+# Install dependencies (use npm ci for faster, reliable builds)
+RUN npm ci --only=production && npm cache clean --force
+
+# Copy web code
+COPY web/ .
+
+# Build the application
+RUN npm run build
+
+# Create necessary directories and set permissions
+RUN mkdir -p /app/.next/cache && \
+    chown -R nextjs:nodejs /app
+
+# Switch to non-root user
+USER nextjs
+
+# Expose port
+EXPOSE 3000
+
+# Add health check
+HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
+    CMD wget --no-verbose --tries=1 --spider http://localhost:3000 || exit 1
+
+# Start the application with proper configuration
+CMD ["npm", "start"]

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2025 OpenPoke Contributors
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,10 +1,122 @@
----
-title: Guilherme34 Openpokespace
-emoji: 📈
-colorFrom: gray
-colorTo: pink
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# OpenPoke 🌴
+
+test
+OpenPoke is a simplified, open-source take on [Interaction Company’s](https://interaction.co/about) [Poke](https://poke.com/) assistant—built to show how a multi-agent orchestration stack can feel genuinely useful. It keeps the handful of things Poke is great at (email triage, reminders, and persistent agents) while staying easy to spin up locally.
+
+- Multi-agent FastAPI backend that mirrors Poke's interaction/execution split, powered by OpenAI-compatible APIs.
+- Gmail tooling via [Composio](https://composio.dev/) for drafting/replying/forwarding without leaving chat.
+- Trigger scheduler and background watchers for reminders and "important email" alerts.
+- Next.js web UI that proxies everything through the shared `.env`, so plugging in API keys is the only setup.
+
+## Requirements
+- Python 3.10+
+- Node.js 18+
+- npm 9+
+
+## Quickstart
+1. **Clone and enter the repo.**
+   ```bash
+   git clone https://github.com/shlokkhemani/OpenPoke
+   cd OpenPoke
+   ```
+2. **Create a shared env file.** Copy the template and open it in your editor:
+   ```bash
+   cp .env.example .env
+   ```
+3. **Get your API keys and add them to `.env`:**
+   
+   **API Configuration (Required)**
+   - Configure your OpenAI-compatible API endpoint and API key in `.env`
+   - Set `API_BASE_URL` to your API endpoint (e.g., `https://api.friendli.ai/dedicated/v1`)
+   - Set `API_KEY` to your API key
+   - All agent models can be configured via environment variables
+   
+   **Composio (Required for Gmail)**
+   - Sign in at [composio.dev](https://composio.dev/)
+   - Create an API key
+   - Set up Gmail integration and get your auth config ID
+   - Replace `your_composio_api_key_here` and `your_gmail_auth_config_id_here` in `.env`
+
+## 🚀 Quick Start (Docker - Recommended)
+
+If you have Docker and docker-compose installed, you can get started immediately:
+
+```bash
+# Deploy with one command (includes security setup)
+./deploy.sh
+
+# Or manually with environment variables
+docker-compose --env-file .env up --build -d
+```
+
+This will start both the API server (port 8001) and web UI (port 3000).
+
+### Production Deployment
+
+For production deployments:
+
+```bash
+# Use production environment file
+docker-compose --env-file .env.production up --build -d
+
+# Or use specific environment file
+docker-compose --env-file .env.staging up --build -d
+```
+
+### Docker Features
+
+- **Security**: Non-root containers with proper user isolation
+- **Health Checks**: Built-in monitoring for service availability
+- **Resource Limits**: CPU and memory constraints for stable performance
+- **Logging**: Structured JSON logging with rotation
+- **Networks**: Isolated network for service communication
+- **Volumes**: Persistent storage for logs and runtime data
+
+## 🛠️ Manual Setup (Alternative)
+
+4. **(Required) Create and activate a Python 3.10+ virtualenv:**
+   ```bash
+   # Ensure you're using Python 3.10+
+   python3.10 -m venv .venv
+   source .venv/bin/activate
+   
+   # Verify Python version (should show 3.10+)
+   python --version
+   ```
+   On Windows (PowerShell):
+   ```powershell
+   # Use Python 3.10+ (adjust path as needed)
+   python3.10 -m venv .venv
+   .\.venv\Scripts\Activate.ps1
+   
+   # Verify Python version
+   python --version
+   ```
+
+5. **Install backend dependencies:**
+   ```bash
+   pip install -r server/requirements.txt
+   ```
+6. **Install frontend dependencies:**
+   ```bash
+   npm install --prefix web
+   ```
+7. **Start the FastAPI server:**
+   ```bash
+   python -m server.server --reload
+   ```
+8. **Start the Next.js app (new terminal):**
+   ```bash
+   npm run dev --prefix web
+   ```
+9. **Connect Gmail for email workflows.** With both services running, open [http://localhost:3000](http://localhost:3000), head to *Settings → Gmail*, and complete the Composio OAuth flow. This step is required for email drafting, replies, and the important-email monitor.
+
+The web app proxies API calls to the Python server using the values in `.env`, so keeping both processes running is required for end-to-end flows.
+
+## Project Layout
+- `server/` – FastAPI application and agents
+- `web/` – Next.js app
+- `server/data/` – runtime data (ignored by git)
+
+## License
+MIT — see [LICENSE](LICENSE).

deploy.sh

@@ -0,0 +1,69 @@
+#!/bin/bash
+
+# OpenPoke Deployment Script
+echo "🚀 Deploying OpenPoke with FriendliAI integration..."
+
+# Check if .env file exists
+if [ ! -f ".env" ]; then
+    echo "⚠️  Warning: .env file not found!"
+    echo "📝 Please copy .env.example to .env and configure your API keys:"
+    echo "   cp .env.example .env"
+    echo "   # Then edit .env with your actual credentials"
+    echo ""
+    echo "🔄 Continuing with default configuration (you may need to configure API keys manually)..."
+fi
+
+# Check if Docker is installed
+if ! command -v docker &> /dev/null; then
+    echo "❌ Docker is not installed. Please install Docker first."
+    exit 1
+fi
+
+# Check if docker-compose is installed
+if command -v docker-compose &> /dev/null; then
+    COMPOSE_CMD="docker-compose"
+elif docker compose version &> /dev/null; then
+    COMPOSE_CMD="docker compose"
+else
+    echo "❌ docker-compose is not installed. Please install docker-compose (or enable the Docker Compose plugin) first."
+    exit 1
+fi
+
+# Stop any existing containers
+echo "🛑 Stopping existing containers..."
+${COMPOSE_CMD} down
+
+# Build and start the services
+echo "🔨 Building and starting services..."
+if [ -f ".env" ]; then
+    ${COMPOSE_CMD} --env-file .env up --build -d
+else
+    ${COMPOSE_CMD} up --build -d
+fi
+
+# Wait for services to be ready
+echo "⏳ Waiting for services to start..."
+sleep 15
+
+# Check if services are running
+if ${COMPOSE_CMD} ps | grep -q "Up"; then
+    echo "✅ Deployment successful!"
+    echo ""
+    echo "🌐 Services are running:"
+    echo "   - Server API: http://localhost:8001"
+    echo "   - Web UI: http://localhost:3000"
+    echo ""
+    echo "📖 Check the logs with: ${COMPOSE_CMD} logs -f"
+    echo "🔍 View service status: ${COMPOSE_CMD} ps"
+    echo "🛑 Stop with: ${COMPOSE_CMD} down"
+    echo ""
+    echo "💡 Tip: If you encounter API key issues, edit your .env file with correct credentials"
+else
+    echo "❌ Deployment failed. Check the logs with: ${COMPOSE_CMD} logs"
+    echo ""
+    echo "🔧 Troubleshooting tips:"
+    echo "   1. Check if your .env file has valid API keys"
+    echo "   2. Verify Docker has enough resources"
+    echo "   3. Check firewall settings"
+    exit 1
+fi

docker-compose.production.yml

@@ -0,0 +1,79 @@
+version: '3.8'
+
+# Production overrides for docker-compose.yml
+# Usage: docker-compose -f docker-compose.yml -f docker-compose.production.yml up
+
+services:
+  server:
+    # Production-specific configurations
+    environment:
+      - OPENPOKE_CORS_ALLOW_ORIGINS=${PRODUCTION_CORS_ORIGINS:-https://yourdomain.com}
+      - OPENPOKE_ENABLE_DOCS=0  # Disable docs in production
+    deploy:
+      resources:
+        limits:
+          cpus: '1.0'
+          memory: 1G
+        reservations:
+          cpus: '0.5'
+          memory: 512M
+      restart_policy:
+        condition: on-failure
+        delay: 5s
+        max_attempts: 3
+    logging:
+      driver: "json-file"
+      options:
+        max-size: "50m"
+        max-file: "5"
+
+  web:
+    # Production-specific configurations
+    deploy:
+      resources:
+        limits:
+          cpus: '0.5'
+          memory: 512M
+        reservations:
+          cpus: '0.25'
+          memory: 256M
+      restart_policy:
+        condition: on-failure
+        delay: 5s
+        max_attempts: 3
+    logging:
+      driver: "json-file"
+      options:
+        max-size: "50m"
+        max-file: "5"
+
+# Optional: Add SSL termination with Traefik
+  # traefik:
+  #   image: traefik:v2.10
+  #   command:
+  #     - "--api.dashboard=true"
+  #     - "--providers.docker=true"
+  #     - "--providers.docker.exposedbydefault=false"
+  #     - "--entrypoints.web.address=:80"
+  #     - "--entrypoints.websecure.address=:443"
+  #     - "--certificatesresolvers.letsencrypt.acme.httpchallenge=true"
+  #     - "--certificatesresolvers.letsencrypt.acme.httpchallenge.entrypoint=web"
+  #     - "--certificatesresolvers.letsencrypt.acme.email=your-email@example.com"
+  #     - "--certificatesresolvers.letsencrypt.acme.storage=/letsencrypt/acme.json"
+  #   ports:
+  #     - "80:80"
+  #     - "443:443"
+  #     - "8080:8080"  # Traefik dashboard
+  #   volumes:
+  #     - /var/run/docker.sock:/var/run/docker.sock:ro
+  #     - letsencrypt:/letsencrypt
+  #   networks:
+  #     - app-network
+  #   labels:
+  #     - "traefik.enable=true"
+  #     - "traefik.http.routers.api.rule=Host(`traefik.yourdomain.com`)"
+  #     - "traefik.http.routers.api.service=api@internal"
+
+# volumes:
+#   letsencrypt:
+#     driver: local

docker-compose.yml

@@ -0,0 +1,93 @@
+version: '3.8'
+
+# Define networks for better service isolation
+networks:
+  app-network:
+    driver: bridge
+
+# Define volumes for persistent data
+volumes:
+  logs:
+    driver: local
+
+services:
+  server:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    ports:
+      - "8001:8001"
+    environment:
+      - OPENPOKE_HOST=0.0.0.0
+      - OPENPOKE_PORT=8001
+      - OPENPOKE_CORS_ALLOW_ORIGINS=*
+      - OPENPOKE_ENABLE_DOCS=${OPENPOKE_ENABLE_DOCS:-1}
+      # Sensitive environment variables should be loaded from .env file
+      - API_BASE_URL=${API_BASE_URL}
+      - API_KEY=${API_KEY}
+      - COMPOSIO_API_KEY=${COMPOSIO_API_KEY}
+      - INTERACTION_AGENT_MODEL=${INTERACTION_AGENT_MODEL}
+      - EXECUTION_AGENT_MODEL=${EXECUTION_AGENT_MODEL}
+      - EXECUTION_SEARCH_AGENT_MODEL=${EXECUTION_SEARCH_AGENT_MODEL}
+      - SUMMARIZER_MODEL=${SUMMARIZER_MODEL}
+      - EMAIL_CLASSIFIER_MODEL=${EMAIL_CLASSIFIER_MODEL}
+    restart: unless-stopped
+    networks:
+      - app-network
+    volumes:
+      - logs:/app/logs
+      - ./server:/app/server:ro
+    deploy:
+      resources:
+        limits:
+          cpus: '0.50'
+          memory: 512M
+        reservations:
+          cpus: '0.25'
+          memory: 256M
+    logging:
+      driver: "json-file"
+      options:
+        max-size: "10m"
+        max-file: "3"
+
+  web:
+    build:
+      context: .
+      dockerfile: Dockerfile.web
+    ports:
+      - "3000:3000"
+    environment:
+      - NEXT_PUBLIC_API_URL=${NEXT_PUBLIC_API_URL:-http://localhost:8001}
+    depends_on:
+      - server
+    restart: unless-stopped
+    networks:
+      - app-network
+    volumes:
+      - ./web:/app:ro
+    deploy:
+      resources:
+        limits:
+          cpus: '0.30'
+          memory: 256M
+        reservations:
+          cpus: '0.15'
+          memory: 128M
+    logging:
+      driver: "json-file"
+      options:
+        max-size: "10m"
+        max-file: "3"
+
+  # Optional: Add a reverse proxy (nginx) for production
+  # nginx:
+  #   image: nginx:alpine
+  #   ports:
+  #     - "80:80"
+  #     - "443:443"
+  #   volumes:
+  #     - ./nginx.conf:/etc/nginx/nginx.conf
+  #   depends_on:
+  #     - web
+  #   restart: unless-stopped

next-env.d.ts

@@ -0,0 +1,5 @@
+/// <reference types="next" />
+/// <reference types="next/image-types/global" />
+
+// NOTE: This file should not be edited
+// see https://nextjs.org/docs/app/building-your-application/configuring/typescript for more information.

server/__init__.py

@@ -0,0 +1,3 @@
+"""OpenPoke Python server package."""
+
+from .app import app

server/agents/__init__.py

@@ -0,0 +1,8 @@
+"""Agent assets package.
+
+Contains agent-specific prompts and tool registries that can be wired into
+OpenRouter/OpenAI chat completion requests.
+"""
+
+__all__ = ["interaction_agent", "execution_agent"]
+

server/agents/execution_agent/__init__.py

@@ -0,0 +1,16 @@
+"""Execution agent assets."""
+
+from .agent import ExecutionAgent
+from .batch_manager import ExecutionBatchManager, ExecutionResult, PendingExecution
+from .runtime import ExecutionAgentRuntime
+from .tools import get_tool_schemas as get_execution_tool_schemas, get_tool_registry as get_execution_tool_registry
+
+__all__ = [
+    "ExecutionBatchManager",
+    "ExecutionAgent",
+    "ExecutionAgentRuntime",
+    "ExecutionResult",
+    "PendingExecution",
+    "get_execution_tool_schemas",
+    "get_execution_tool_registry",
+]

server/agents/execution_agent/agent.py

@@ -0,0 +1,123 @@
+"""Execution Agent implementation."""
+
+from pathlib import Path
+from typing import List, Optional, Dict, Any
+
+from ...services.execution import get_execution_agent_logs
+from ...logging_config import logger
+
+
+# Load system prompt template from file
+_prompt_path = Path(__file__).parent / "system_prompt.md"
+if _prompt_path.exists():
+    SYSTEM_PROMPT_TEMPLATE = _prompt_path.read_text(encoding="utf-8").strip()
+else:
+    # Placeholder template - you'll replace this with actual instructions
+    SYSTEM_PROMPT_TEMPLATE = """You are an execution agent responsible for completing specific tasks using available tools.
+
+Agent Name: {agent_name}
+Purpose: {agent_purpose}
+
+Instructions:
+[TO BE FILLED IN BY USER]
+
+You have access to Gmail tools to help complete your tasks. When given instructions:
+1. Analyze what needs to be done
+2. Use the appropriate tools to complete the task
+3. Provide clear status updates on your actions
+
+Be thorough, accurate, and efficient in your execution."""
+
+
+class ExecutionAgent:
+    """Manages state and history for an execution agent."""
+
+    # Initialize execution agent with name, conversation limits, and log store access
+    def __init__(
+        self,
+        name: str,
+        conversation_limit: Optional[int] = None
+    ):
+        """
+        Initialize an execution agent.
+
+        Args:
+            name: Human-readable agent name (e.g., 'conversation with keith')
+            conversation_limit: Optional limit on past conversations to include (None = all)
+        """
+        self.name = name
+        self.conversation_limit = conversation_limit
+        self._log_store = get_execution_agent_logs()
+
+    # Generate system prompt template with agent name and purpose derived from name
+    def build_system_prompt(self) -> str:
+        """Build the system prompt for this agent."""
+        agent_purpose = f"Handle tasks related to: {self.name}"
+
+        return SYSTEM_PROMPT_TEMPLATE.format(
+            agent_name=self.name,
+            agent_purpose=agent_purpose
+        )
+
+    # Combine base system prompt with conversation history, applying conversation limits
+    def build_system_prompt_with_history(self) -> str:
+        """
+        Build system prompt including agent history.
+
+        Returns:
+            System prompt with embedded history transcript
+        """
+        base_prompt = self.build_system_prompt()
+
+        # Load history transcript
+        transcript = self._log_store.load_transcript(self.name)
+
+        if transcript:
+            # Apply conversation limit if needed
+            if self.conversation_limit and self.conversation_limit > 0:
+                # Parse entries and limit them
+                lines = transcript.split('\n')
+                request_count = sum(1 for line in lines if '<agent_request' in line)
+
+                if request_count > self.conversation_limit:
+                    # Find where to cut
+                    kept_requests = 0
+                    cutoff_index = len(lines)
+                    for i in range(len(lines) - 1, -1, -1):
+                        if '<agent_request' in lines[i]:
+                            kept_requests += 1
+                            if kept_requests == self.conversation_limit:
+                                cutoff_index = i
+                                break
+                    transcript = '\n'.join(lines[cutoff_index:])
+
+            return f"{base_prompt}\n\n# Execution History\n\n{transcript}"
+
+        return base_prompt
+
+    # Format current instruction as user message for LLM consumption
+    def build_messages_for_llm(self, current_instruction: str) -> List[Dict[str, str]]:
+        """
+        Build message array for LLM call.
+
+        Args:
+            current_instruction: Current instruction from interaction agent
+
+        Returns:
+            List of messages in OpenRouter format
+        """
+        return [
+            {"role": "user", "content": current_instruction}
+        ]
+
+    # Log the agent's final response to the execution log store
+    def record_response(self, response: str) -> None:
+        """Record agent's response to the log."""
+        self._log_store.record_agent_response(self.name, response)
+
+    # Log tool invocation and results with truncated content for readability
+    def record_tool_execution(self, tool_name: str, arguments: str, result: str) -> None:
+        """Record tool execution details."""
+        self._log_store.record_action(self.name, f"Calling {tool_name} with: {arguments[:200]}")
+        # Record the tool response
+        self._log_store.record_tool_response(self.name, tool_name, result[:500])

server/agents/execution_agent/batch_manager.py

@@ -0,0 +1,193 @@
+"""Coordinate execution agents and batch their results for the interaction agent."""
+
+from __future__ import annotations
+
+import asyncio
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Dict, List, Optional
+
+from .runtime import ExecutionAgentRuntime, ExecutionResult
+from ...logging_config import logger
+
+
+@dataclass
+class PendingExecution:
+    """Track a pending execution request."""
+
+    request_id: str
+    agent_name: str
+    instructions: str
+    batch_id: str
+    created_at: datetime = field(default_factory=datetime.now)
+
+
+@dataclass
+class _BatchState:
+    """Collect results for a single interaction-agent turn."""
+
+    batch_id: str
+    created_at: datetime = field(default_factory=datetime.now)
+    pending: int = 0
+    results: List[ExecutionResult] = field(default_factory=list)
+
+
+class ExecutionBatchManager:
+    """Run execution agents and deliver their combined outcome."""
+
+    # Initialize batch manager with timeout and coordination state for execution agents
+    def __init__(self, timeout_seconds: int = 90) -> None:
+        self.timeout_seconds = timeout_seconds
+        self._pending: Dict[str, PendingExecution] = {}
+        self._batch_lock = asyncio.Lock()
+        self._batch_state: Optional[_BatchState] = None
+
+    # Run execution agent with timeout handling and batch coordination for interaction agent
+    async def execute_agent(
+        self,
+        agent_name: str,
+        instructions: str,
+        request_id: Optional[str] = None,
+    ) -> ExecutionResult:
+        """Execute an agent asynchronously and buffer the result for batch dispatch."""
+
+        if not request_id:
+            request_id = str(uuid.uuid4())
+
+        batch_id = await self._register_pending_execution(agent_name, instructions, request_id)
+
+        try:
+            logger.info(f"[{agent_name}] Execution started")
+            runtime = ExecutionAgentRuntime(agent_name=agent_name)
+            result = await asyncio.wait_for(
+                runtime.execute(instructions),
+                timeout=self.timeout_seconds,
+            )
+            status = "SUCCESS" if result.success else "FAILED"
+            logger.info(f"[{agent_name}] Execution finished: {status}")
+        except asyncio.TimeoutError:
+            logger.error(f"[{agent_name}] Execution timed out after {self.timeout_seconds}s")
+            result = ExecutionResult(
+                agent_name=agent_name,
+                success=False,
+                response=f"Execution timed out after {self.timeout_seconds} seconds",
+                error="Timeout",
+            )
+        except Exception as exc:  # pragma: no cover - defensive
+            logger.exception(f"[{agent_name}] Execution failed unexpectedly")
+            result = ExecutionResult(
+                agent_name=agent_name,
+                success=False,
+                response=f"Execution failed: {exc}",
+                error=str(exc),
+            )
+        finally:
+            self._pending.pop(request_id, None)
+
+        await self._complete_execution(batch_id, result, agent_name)
+        return result
+
+    # Add execution request to current batch or create new batch if none exists
+    async def _register_pending_execution(
+        self,
+        agent_name: str,
+        instructions: str,
+        request_id: str,
+    ) -> str:
+        """Attach a new execution to the active batch, opening one when required."""
+
+        async with self._batch_lock:
+            if self._batch_state is None:
+                batch_id = str(uuid.uuid4())
+                self._batch_state = _BatchState(batch_id=batch_id)
+            else:
+                batch_id = self._batch_state.batch_id
+
+            self._batch_state.pending += 1
+            self._pending[request_id] = PendingExecution(
+                request_id=request_id,
+                agent_name=agent_name,
+                instructions=instructions,
+                batch_id=batch_id,
+            )
+
+            return batch_id
+
+    # Store execution result and send combined batch to interaction agent when complete
+    async def _complete_execution(
+        self,
+        batch_id: str,
+        result: ExecutionResult,
+        agent_name: str,
+    ) -> None:
+        """Record the execution result and dispatch when the batch drains."""
+
+        dispatch_payload: Optional[str] = None
+
+        async with self._batch_lock:
+            state = self._batch_state
+            if state is None or state.batch_id != batch_id:
+                logger.warning(f"[{agent_name}] Dropping result for unknown batch")
+                return
+
+            state.results.append(result)
+            state.pending -= 1
+
+            if state.pending == 0:
+                dispatch_payload = self._format_batch_payload(state.results)
+                agent_names = [entry.agent_name for entry in state.results]
+                logger.info(f"Execution batch completed: {', '.join(agent_names)}")
+                self._batch_state = None
+
+        if dispatch_payload:
+            await self._dispatch_to_interaction_agent(dispatch_payload)
+
+    # Return list of currently pending execution requests for monitoring purposes
+    def get_pending_executions(self) -> List[Dict[str, str]]:
+        """Expose pending executions for observability."""
+
+        return [
+            {
+                "request_id": pending.request_id,
+                "agent_name": pending.agent_name,
+                "batch_id": pending.batch_id,
+                "created_at": pending.created_at.isoformat(),
+                "elapsed_seconds": (datetime.now() - pending.created_at).total_seconds(),
+            }
+            for pending in self._pending.values()
+        ]
+
+    # Clean up all pending executions and batch state on shutdown
+    async def shutdown(self) -> None:
+        """Clear pending bookkeeping (no background work remains)."""
+
+        self._pending.clear()
+        async with self._batch_lock:
+            self._batch_state = None
+
+    # Format multiple execution results into single message for interaction agent
+    def _format_batch_payload(self, results: List[ExecutionResult]) -> str:
+        """Render execution results into the interaction-agent format."""
+
+        entries: List[str] = []
+        for result in results:
+            status = "SUCCESS" if result.success else "FAILED"
+            response_text = (result.response or "(no response provided)").strip()
+            entries.append(f"[{status}] {result.agent_name}: {response_text}")
+        return "\n".join(entries)
+
+    # Forward combined execution results to interaction agent for user response generation
+    async def _dispatch_to_interaction_agent(self, payload: str) -> None:
+        """Send the aggregated execution summary to the interaction agent."""
+
+        from ..interaction_agent.runtime import InteractionAgentRuntime
+
+        runtime = InteractionAgentRuntime()
+        try:
+            loop = asyncio.get_running_loop()
+        except RuntimeError:
+            asyncio.run(runtime.handle_agent_message(payload))
+            return
+
+        loop.create_task(runtime.handle_agent_message(payload))

server/agents/execution_agent/runtime.py

@@ -0,0 +1,236 @@
+"""Simplified Execution Agent Runtime."""
+
+import inspect
+import json
+from typing import Dict, Any, List, Optional, Tuple
+from dataclasses import dataclass
+
+from .agent import ExecutionAgent
+from .tools import get_tool_schemas, get_tool_registry
+from ...config import get_settings
+from ...openrouter_client import request_chat_completion
+from ...logging_config import logger
+
+
+@dataclass
+class ExecutionResult:
+    """Result from an execution agent."""
+    agent_name: str
+    success: bool
+    response: str
+    error: Optional[str] = None
+    tools_executed: List[str] = None
+
+
+class ExecutionAgentRuntime:
+    """Manages the execution of a single agent request."""
+
+    MAX_TOOL_ITERATIONS = 8
+
+    # Initialize execution agent runtime with settings, tools, and agent instance
+    def __init__(self, agent_name: str):
+        settings = get_settings()
+        self.agent = ExecutionAgent(agent_name)
+        self.api_key = settings.api_key
+        self.model = settings.execution_agent_model
+        self.tool_registry = get_tool_registry(agent_name=agent_name)
+        self.tool_schemas = get_tool_schemas()
+
+        if not self.api_key:
+            raise ValueError("API key not configured. Set API_KEY environment variable.")
+
+    # Main execution loop for running agent with LLM calls and tool execution
+    async def execute(self, instructions: str) -> ExecutionResult:
+        """Execute the agent with given instructions."""
+        try:
+            # Build system prompt with history
+            system_prompt = self.agent.build_system_prompt_with_history()
+
+            # Start conversation with the instruction
+            messages = [{"role": "user", "content": instructions}]
+            tools_executed: List[str] = []
+            final_response: Optional[str] = None
+
+            for iteration in range(self.MAX_TOOL_ITERATIONS):
+                logger.info(
+                    f"[{self.agent.name}] Requesting plan (iteration {iteration + 1})"
+                )
+                response = await self._make_llm_call(system_prompt, messages, with_tools=True)
+                assistant_message = response.get("choices", [{}])[0].get("message", {})
+
+                if not assistant_message:
+                    raise RuntimeError("LLM response did not include an assistant message")
+
+                raw_tool_calls = assistant_message.get("tool_calls", []) or []
+                parsed_tool_calls = self._extract_tool_calls(raw_tool_calls)
+
+                assistant_entry: Dict[str, Any] = {
+                    "role": "assistant",
+                    "content": assistant_message.get("content", "") or "",
+                }
+                if raw_tool_calls:
+                    assistant_entry["tool_calls"] = raw_tool_calls
+                messages.append(assistant_entry)
+
+                if not parsed_tool_calls:
+                    final_response = assistant_entry["content"] or "No action required."
+                    break
+
+                for tool_call in parsed_tool_calls:
+                    tool_name = tool_call.get("name", "")
+                    tool_args = tool_call.get("arguments", {})
+                    call_id = tool_call.get("id")
+
+                    if not tool_name:
+                        logger.warning("Tool call missing name: %s", tool_call)
+                        failure = {"error": "Tool call missing name; unable to execute."}
+                        tool_message = {
+                            "role": "tool",
+                            "tool_call_id": call_id or "unknown_tool",
+                            "content": self._format_tool_result(
+                                tool_name or "<unknown>", False, failure, tool_args
+                            ),
+                        }
+                        messages.append(tool_message)
+                        continue
+
+                    tools_executed.append(tool_name)
+                    logger.info(f"[{self.agent.name}] Executing tool: {tool_name}")
+
+                    success, result = await self._execute_tool(tool_name, tool_args)
+
+                    if success:
+                        logger.info(f"[{self.agent.name}] Tool {tool_name} completed successfully")
+                        record_payload = self._safe_json_dump(result)
+                    else:
+                        error_detail = result.get("error") if isinstance(result, dict) else str(result)
+                        logger.warning(f"[{self.agent.name}] Tool {tool_name} failed: {error_detail}")
+                        record_payload = error_detail
+
+                    self.agent.record_tool_execution(
+                        tool_name,
+                        self._safe_json_dump(tool_args),
+                        record_payload
+                    )
+
+                    tool_message = {
+                        "role": "tool",
+                        "tool_call_id": call_id or tool_name,
+                        "content": self._format_tool_result(tool_name, success, result, tool_args),
+                    }
+                    messages.append(tool_message)
+
+            else:
+                raise RuntimeError("Reached tool iteration limit without final response")
+
+            if final_response is None:
+                raise RuntimeError("LLM did not return a final response")
+
+            self.agent.record_response(final_response)
+
+            return ExecutionResult(
+                agent_name=self.agent.name,
+                success=True,
+                response=final_response,
+                tools_executed=tools_executed
+            )
+
+        except Exception as e:
+            logger.error(f"[{self.agent.name}] Execution failed: {e}")
+            error_msg = str(e)
+            failure_text = f"Failed to complete task: {error_msg}"
+            self.agent.record_response(f"Error: {error_msg}")
+
+            return ExecutionResult(
+                agent_name=self.agent.name,
+                success=False,
+                response=failure_text,
+                error=error_msg
+            )
+
+    # Execute API call with system prompt, messages, and optional tool schemas
+    async def _make_llm_call(self, system_prompt: str, messages: List[Dict], with_tools: bool) -> Dict:
+        """Make an LLM call."""
+        tools_to_send = self.tool_schemas if with_tools else None
+        logger.info(f"[{self.agent.name}] Calling LLM with model: {self.model}, tools: {len(tools_to_send) if tools_to_send else 0}")
+        return await request_chat_completion(
+            model=self.model,
+            messages=messages,
+            system=system_prompt,
+            api_key=self.api_key,
+            tools=tools_to_send
+        )
+
+    # Parse and validate tool calls from LLM response into structured format
+    def _extract_tool_calls(self, raw_tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Extract tool calls from an assistant message."""
+        tool_calls: List[Dict[str, Any]] = []
+
+        for tool in raw_tools:
+            function = tool.get("function", {})
+            name = function.get("name", "")
+            args = function.get("arguments", "")
+
+            if isinstance(args, str):
+                try:
+                    args = json.loads(args) if args else {}
+                except json.JSONDecodeError:
+                    args = {}
+
+            if name:
+                tool_calls.append({
+                    "id": tool.get("id"),
+                    "name": name,
+                    "arguments": args,
+                })
+
+        return tool_calls
+
+    # Safely convert objects to JSON with fallback to string representation
+    def _safe_json_dump(self, payload: Any) -> str:
+        """Serialize payload to JSON, falling back to string representation."""
+        try:
+            return json.dumps(payload, default=str)
+        except TypeError:
+            return str(payload)
+
+    # Format tool execution results into JSON structure for LLM consumption
+    def _format_tool_result(
+        self,
+        tool_name: str,
+        success: bool,
+        result: Any,
+        arguments: Dict[str, Any],
+    ) -> str:
+        """Build a structured string for tool responses."""
+        if success:
+            payload: Dict[str, Any] = {
+                "tool": tool_name,
+                "status": "success",
+                "arguments": arguments,
+                "result": result,
+            }
+        else:
+            error_detail = result.get("error") if isinstance(result, dict) else str(result)
+            payload = {
+                "tool": tool_name,
+                "status": "error",
+                "arguments": arguments,
+                "error": error_detail,
+            }
+        return self._safe_json_dump(payload)
+
+    # Execute tool function from registry with error handling and async support
+    async def _execute_tool(self, tool_name: str, arguments: Dict) -> Tuple[bool, Any]:
+        """Execute a tool. Returns (success, result)."""
+        tool_func = self.tool_registry.get(tool_name)
+        if not tool_func:
+            return False, {"error": f"Unknown tool: {tool_name}"}
+
+        try:
+            result = tool_func(**arguments)
+            if inspect.isawaitable(result):
+                result = await result
+            return True, result
+        except Exception as e:
+            return False, {"error": str(e)}

server/agents/execution_agent/system_prompt.md

@@ -0,0 +1,51 @@
+You are the assistant of Poke by the Interaction Company of California. You are the "execution engine" of Poke, helping complete tasks for Poke, while Poke talks to the user. Your job is to execute and accomplish a goal, and you do not have direct access to the user.
+
+IMPORTANT: Don't ever execute a draft unless you receive explicit confirmation to execute it. If you are instructed to send an email, first JUST create the draft. Then, when the user confirms draft, we can send it. 
+
+
+Your final output is directed to Poke, which handles user conversations and presents your results to the user. Focus on providing Poke with adequate contextual information; you are not responsible for framing responses in a user-friendly way.
+
+If it needs more data from Poke or the user, you should also include it in your final output message. If you ever need to send a message to the user, you should tell Poke to forward that message to the user.
+
+Remember that your last output message (summary) will be forwarded to Poke. In that message, provide all relevant information and avoid preamble or postamble (e.g., "Here's what I found:" or "Let me know if this looks good to send"). If you create a draft, you need to send the exact to, subject, and body of the draft to the interaction agent verbatim. 
+
+This conversation history may have gaps. It may start from the middle of a conversation, or it may be missing messages. The only assumption you can make is that Poke's latest message is the most recent one, and representative of Poke's current requests. Address that message directly. The other messages are just for context.
+
+Before you call any tools, reason through why you are calling them by explaining the thought process. If it could possibly be helpful to call more than one tool at once, then do so.
+
+If you have context that would help the execution of a tool call (e.g. the user is searching for emails from a person and you know that person's email address), pass that context along.
+
+When searching for personal information about the user, it's probably smart to look through their emails.
+
+
+
+
+Agent Name: {agent_name}
+Purpose: {agent_purpose}
+
+# Instructions
+[TO BE FILLED IN BY USER - Add your specific instructions here]
+
+# Available Tools
+You have access to the following Gmail tools:
+- gmail_create_draft: Create an email draft
+- gmail_execute_draft: Send a previously created draft
+- gmail_forward_email: Forward an existing email
+- gmail_reply_to_thread: Reply to an email thread
+
+You also manage reminder triggers for this agent:
+- createTrigger: Store a reminder by providing the payload to run later. Supply an ISO 8601 `start_time` and an iCalendar `RRULE` when recurrence is needed.
+- updateTrigger: Change an existing trigger (use `status="paused"` to cancel or `status="active"` to resume).
+- listTriggers: Inspect all triggers assigned to this agent.
+
+# Guidelines
+1. Analyze the instructions carefully before taking action
+2. Use the appropriate tools to complete the task
+3. Be thorough and accurate in your execution
+4. Provide clear, concise responses about what you accomplished
+5. If you encounter errors, explain what went wrong and what you tried
+6. When creating or updating triggers, convert natural-language schedules into explicit `RRULE` strings and precise `start_time` timestamps yourself—do not rely on the trigger service to infer intent without them.
+7. All times will be interpreted using the user's automatically detected timezone.
+8. After creating or updating a trigger, consider calling `listTriggers` to confirm the schedule when clarity would help future runs.
+
+When you receive instructions, think step-by-step about what needs to be done, then execute the necessary tools to complete the task.

server/agents/execution_agent/tasks/__init__.py

@@ -0,0 +1,30 @@
+"""Task registry for execution agents."""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, List
+
+from .search_email.schemas import get_schemas as _get_email_search_schemas
+from .search_email.tool import build_registry as _build_email_search_registry
+
+
+# Return tool schemas contributed by task modules
+def get_task_schemas() -> List[Dict[str, Any]]:
+    """Return tool schemas contributed by task modules."""
+
+    return [*_get_email_search_schemas()]
+
+
+# Return executable task tools keyed by name
+def get_task_registry(agent_name: str) -> Dict[str, Callable[..., Any]]:
+    """Return executable task tools keyed by name."""
+
+    registry: Dict[str, Callable[..., Any]] = {}
+    registry.update(_build_email_search_registry(agent_name))
+    return registry
+
+
+__all__ = [
+    "get_task_registry",
+    "get_task_schemas",
+]

server/agents/execution_agent/tasks/search_email/__init__.py

@@ -0,0 +1,15 @@
+"""Email search task package."""
+
+from .schemas import SEARCH_TOOL_NAME, TASK_TOOL_NAME, TaskEmailSearchPayload, get_schemas
+from .tool import GmailSearchEmail, EmailSearchToolResult, build_registry, task_email_search
+
+__all__ = [
+    "GmailSearchEmail",
+    "EmailSearchToolResult",
+    "TaskEmailSearchPayload",
+    "SEARCH_TOOL_NAME",
+    "TASK_TOOL_NAME",
+    "build_registry",
+    "get_schemas",
+    "task_email_search",
+]

server/agents/execution_agent/tasks/search_email/email_cleaner.py

@@ -0,0 +1,5 @@
+"""Backward-compatible re-export for shared email cleaning utilities."""
+
+from server.services.gmail import EmailTextCleaner
+
+__all__ = ["EmailTextCleaner"]

server/agents/execution_agent/tasks/search_email/gmail_internal.py

@@ -0,0 +1,79 @@
+"""Internal Gmail utilities for the search_email task.
+
+This module contains Gmail functions that are internal to the search_email task
+and should not be exposed as public tools to execution agents.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from server.services.gmail import execute_gmail_tool, get_active_gmail_user_id
+
+# Schema for the internal LLM to call gmail_fetch_emails
+GMAIL_FETCH_EMAILS_SCHEMA = {
+    "type": "function",
+    "function": {
+        "name": "gmail_fetch_emails",
+        "description": "Search Gmail and retrieve matching messages",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": "Gmail search query (same syntax as Gmail UI).",
+                },
+                "max_results": {
+                    "type": "integer",
+                    "description": "Maximum number of emails to return. Default: 10. Use higher values (20-50) only when absolutely necessary for comprehensive searches like 'all important emails this month'.",
+                    "minimum": 1,
+                    "maximum": 100,
+                },
+                "include_spam_trash": {
+                    "type": "boolean",
+                    "description": "Include spam and trash messages. Default: false.",
+                },
+            },
+            "additionalProperties": False,
+        },
+    },
+}
+
+
+def gmail_fetch_emails(
+    query: Optional[str] = None,
+    label_ids: Optional[List[str]] = None,
+    max_results: Optional[int] = None,
+    page_token: Optional[str] = None,
+    ids_only: Optional[bool] = None,
+    include_payload: Optional[bool] = None,
+    include_spam_trash: Optional[bool] = None,
+    verbose: Optional[bool] = None,
+) -> Dict[str, Any]:
+    """Fetch Gmail messages with optional filters and verbosity controls.
+    
+    This is an internal function for the search_email task and should not
+    be exposed as a public tool to execution agents.
+    """
+    arguments: Dict[str, Any] = {
+        "query": query,
+        "label_ids": label_ids,
+        "max_results": max_results,
+        "page_token": page_token,
+        "ids_only": ids_only,
+        "include_payload": include_payload,
+        "include_spam_trash": include_spam_trash,
+        "verbose": verbose,
+    }
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    
+    # Use the same composio integration as the public tools
+    return execute_gmail_tool("GMAIL_FETCH_EMAILS", composio_user_id, arguments)
+
+
+__all__ = [
+    "gmail_fetch_emails",
+    "GMAIL_FETCH_EMAILS_SCHEMA",
+]

server/agents/execution_agent/tasks/search_email/schemas.py

@@ -0,0 +1,122 @@
+"""Schemas for the email search task tools."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+TASK_TOOL_NAME = "task_email_search"
+SEARCH_TOOL_NAME = "gmail_fetch_emails"
+COMPLETE_TOOL_NAME = "return_search_results"
+
+_SCHEMAS: List[Dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": TASK_TOOL_NAME,
+            "description": "Expand a raw Gmail search request into multiple targeted queries and return relevant emails.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "search_query": {
+                        "type": "string",
+                        "description": "Raw search request describing the emails to find.",
+                    },
+                },
+                "required": ["search_query"],
+                "additionalProperties": False,
+            },
+        },
+    }
+]
+
+
+class GmailSearchEmail(BaseModel):
+    """Clean email representation with enhanced content processing."""
+
+    model_config = ConfigDict(extra="ignore", frozen=True)
+
+    # Core identifiers
+    id: str  # message_id from Gmail API
+    thread_id: Optional[str] = None
+    query: str  # The search query that found this email
+    
+    # Email metadata
+    subject: str
+    sender: str
+    recipient: str  # to field
+    timestamp: datetime
+    label_ids: List[str] = Field(default_factory=list)
+    
+    # Clean content (primary field for LLM consumption)
+    clean_text: str  # Processed, readable email content
+    
+    # Attachment information
+    has_attachments: bool = False
+    attachment_count: int = 0
+    attachment_filenames: List[str] = Field(default_factory=list)
+
+
+class EmailSearchToolResult(BaseModel):
+    """Structured payload for each tool-call response."""
+
+    status: Literal["success", "error"]
+    query: Optional[str] = None
+    result_count: Optional[int] = None
+    next_page_token: Optional[str] = None
+    messages: List[GmailSearchEmail] = Field(default_factory=list)
+    error: Optional[str] = None
+
+
+class TaskEmailSearchPayload(BaseModel):
+    """Envelope for the final email selection."""
+
+    model_config = ConfigDict(extra="forbid", frozen=True)
+
+    emails: List[GmailSearchEmail]
+
+
+_COMPLETION_SCHEMAS: List[Dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": COMPLETE_TOOL_NAME,
+            "description": "Return the final list of relevant Gmail message ids that match the search criteria.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "message_ids": {
+                        "type": "array",
+                        "description": "List of Gmail message ids deemed relevant.",
+                        "items": {"type": "string"},
+                    },
+                },
+                "required": ["message_ids"],
+                "additionalProperties": False,
+            },
+        },
+    }
+]
+
+def get_completion_schema() -> Dict[str, Any]:
+    return _COMPLETION_SCHEMAS[0]
+
+
+def get_schemas() -> List[Dict[str, Any]]:
+    """Return the JSON schema for the email search task."""
+
+    return _SCHEMAS
+
+
+__all__ = [
+    "GmailSearchEmail",
+    "EmailSearchToolResult",
+    "TaskEmailSearchPayload",
+    "SEARCH_TOOL_NAME",
+    "COMPLETE_TOOL_NAME",
+    "TASK_TOOL_NAME",
+    "get_completion_schema",
+    "get_schemas",
+]

server/agents/execution_agent/tasks/search_email/system_prompt.py

@@ -0,0 +1,83 @@
+"""System prompt for the Gmail search assistant."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+
+def get_system_prompt() -> str:
+    """Generate system prompt with today's date for Gmail search assistant."""
+    today = datetime.now().strftime("%Y/%m/%d")
+    
+    return (
+        "You are an expert Gmail search assistant helping users find emails efficiently.\n"
+        f"\n"
+        f"## Current Context:\n"
+        f"- Today's date: {today}\n"
+        f"- Use this date as reference for relative time queries (e.g., 'recent', 'today', 'this week')\n"
+        "\n"
+        "## Available Tools:\n"
+        "- `gmail_fetch_emails`: Search Gmail using advanced search parameters\n"
+        "  - `query`: Gmail search query using standard Gmail search operators\n"
+        "  - `max_results`: Maximum emails to return (default: 10, range: 1-100)\n"
+        "  - `include_spam_trash`: Include spam/trash messages (default: false)\n"
+        "- `return_search_results`: Return the final list of relevant message IDs\n"
+        "\n"
+        "## Gmail Search Strategy:\n"
+        "1. **Use Gmail's powerful search operators** to create precise queries:\n"
+        "   - `from:[email protected]` - emails from specific sender\n"
+        "   - `to:[email protected]` - emails to specific recipient\n"
+        "   - `subject:keyword` - emails with specific subject content\n"
+        "   - `has:attachment` - emails with attachments\n"
+        "   - `after:YYYY/MM/DD` and `before:YYYY/MM/DD` - date ranges\n"
+        "   - `is:unread`, `is:read`, `is:important` - status filters\n"
+        "   - `in:inbox`, `in:sent`, `in:trash` - location filters\n"
+        "   - `larger:10M`, `smaller:1M` - size filters\n"
+        "   - `\"exact phrase\"` - exact phrase matching\n"
+        "   - `OR`, `-` (NOT), `()` for complex boolean logic\n"
+        "\n"
+        "2. **Run multiple searches in parallel** when the user's request suggests different approaches:\n"
+        "   - Search by sender AND by keywords simultaneously\n"
+        "   - Try relevant date ranges in parallel\n"
+        "   - Search multiple related terms or variations\n"
+        "   - Combine broad and specific queries\n"
+        "\n"
+        "3. **Use max_results strategically** to balance comprehensiveness with context efficiency:\n"
+        "   - **Default: 10 results** - suitable for most targeted searches\n"
+        "   - **Use 20-50 results** only when absolutely necessary for comprehensive queries like:\n"
+        "     * \"All important emails from the past month\"\n"
+        "     * \"All meeting invites from this quarter\"\n"
+        "     * \"All emails with attachments from a specific project\"\n"
+        "   - **Avoid over-burdening context** - prefer multiple targeted 10-result searches over one large search\n"
+        "   - **Judge necessity carefully** - only increase limit when the query explicitly requires comprehensive results\n"
+        "\n"
+        "4. **Think strategically** about what search parameters would be most relevant:\n"
+        f"   - For \"recent emails from John\": `from:john after:{today}`\n"
+        "   - For \"meeting invites\": `subject:meeting OR subject:invite has:attachment`\n"
+        "   - For \"large files\": `has:attachment larger:5M`\n"
+        "   - For \"unread important emails\": `is:unread is:important`\n"
+        f"   - For \"today's emails\": `after:{today}`\n"
+        f"   - For \"this week's emails\": Use date ranges based on today ({today})\n"
+        "\n"
+        "## Email Content Processing:\n"
+        "- Each email includes `clean_text` - processed, readable content from HTML/plain text\n"
+        "- Clean text has tracking pixels removed, URLs truncated, and formatting optimized\n"
+        "- Attachment information is available: `has_attachments`, `attachment_count`, `attachment_filenames`\n"
+        "- Email timestamps are automatically converted to the user's preferred timezone\n"
+        "- Use clean text content to understand email context and relevance\n"
+        "\n"
+        "## Your Process:\n"
+        "1. **Analyze** the user's request to identify key search criteria\n"
+        "2. **Search strategically** using multiple targeted Gmail queries with appropriate operators\n"
+        "3. **Review content** - examine the `clean_text` field to understand email relevance\n"
+        "4. **Consider attachments** - factor in attachment information when relevant to the query\n"
+        "5. **Refine searches** - run additional queries if needed based on content analysis\n"
+        "6. **Select results** - call `return_search_results` with message IDs that best match intent\n"
+        "\n"
+        "Be thorough and strategic - use Gmail's search power AND content analysis to find exactly what the user needs!"
+    )
+
+
+__all__ = [
+    "get_system_prompt",
+]

server/agents/execution_agent/tasks/search_email/tool.py

@@ -0,0 +1,450 @@
+"""Email search task implementation."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
+
+from server.config import get_settings
+from server.logging_config import logger
+from server.openrouter_client import request_chat_completion
+from server.services.execution import get_execution_agent_logs
+from server.services.gmail import (
+    EmailTextCleaner,
+    ProcessedEmail,
+    execute_gmail_tool,
+    get_active_gmail_user_id,
+    parse_gmail_fetch_response,
+)
+from .gmail_internal import GMAIL_FETCH_EMAILS_SCHEMA
+from .schemas import (
+    GmailSearchEmail,
+    EmailSearchToolResult,
+    TaskEmailSearchPayload,
+    COMPLETE_TOOL_NAME,
+    SEARCH_TOOL_NAME,
+    TASK_TOOL_NAME,
+    get_completion_schema,
+)
+from .system_prompt import get_system_prompt
+
+# Constants
+MAX_LLM_ITERATIONS = 8
+ERROR_GMAIL_NOT_CONNECTED = "Gmail not connected. Please connect Gmail in settings first."
+ERROR_OPENROUTER_NOT_CONFIGURED = "API key not configured. Set API_KEY."
+ERROR_EMPTY_QUERY = "search_query must not be empty"
+ERROR_QUERY_REQUIRED = "query parameter is required"
+ERROR_MESSAGE_IDS_REQUIRED = "message_ids parameter is required"
+ERROR_MESSAGE_IDS_MUST_BE_LIST = "message_ids must be provided as a list"
+ERROR_TOOL_ARGUMENTS_INVALID = "Tool arguments must be an object"
+ERROR_ITERATION_LIMIT = "Email search orchestrator exceeded iteration limit"
+
+
+
+_COMPLETION_TOOL_SCHEMA = get_completion_schema()
+_LOG_STORE = get_execution_agent_logs()
+_EMAIL_CLEANER = EmailTextCleaner(max_url_length=40)
+
+
+# Create standardized error response for tool calls
+def _create_error_response(call_id: str, query: Optional[str], error: str) -> Tuple[str, str]:
+    """Create standardized error response for tool calls."""
+    result = EmailSearchToolResult(status="error", query=query, error=error)
+    return (call_id, _safe_json_dumps(result.model_dump(exclude_none=True)))
+
+
+# Create standardized success response for tool calls
+def _create_success_response(call_id: str, data: Dict[str, Any]) -> Tuple[str, str]:
+    """Create standardized success response for tool calls."""
+    return (call_id, _safe_json_dumps(data))
+
+
+def _validate_search_query(search_query: str) -> Optional[str]:
+    """Validate search query and return error message if invalid."""
+    if not (search_query or "").strip():
+        return ERROR_EMPTY_QUERY
+    return None
+
+
+def _validate_gmail_connection() -> Optional[str]:
+    """Validate Gmail connection and return user ID or None."""
+    return get_active_gmail_user_id()
+
+
+def _validate_openrouter_config() -> Tuple[Optional[str], Optional[str]]:
+    """Validate API configuration and return (api_key, model) or (None, error)."""
+    settings = get_settings()
+    api_key = settings.api_key
+    if not api_key:
+        return None, ERROR_OPENROUTER_NOT_CONFIGURED
+    return api_key, settings.execution_agent_search_model
+
+
+# Return task tool callables
+def build_registry(agent_name: str) -> Dict[str, Callable[..., Any]]:  # noqa: ARG001
+    """Return task tool callables."""
+
+    return {
+        TASK_TOOL_NAME: task_email_search,
+    }
+
+
+# Run an agentic Gmail search for the provided query
+async def task_email_search(search_query: str) -> Any:
+    """Run an agentic Gmail search for the provided query."""
+    logger.info(f"[EMAIL_SEARCH] Starting search for: '{search_query}'")
+    
+    # Validate inputs
+    cleaned_query = (search_query or "").strip()
+    if error := _validate_search_query(cleaned_query):
+        logger.error(f"[EMAIL_SEARCH] Invalid query: {error}")
+        return {"error": error}
+    
+    composio_user_id = _validate_gmail_connection()
+    if not composio_user_id:
+        logger.error(f"[EMAIL_SEARCH] Gmail not connected")
+        return {"error": ERROR_GMAIL_NOT_CONNECTED}
+    
+    api_key, model_or_error = _validate_openrouter_config()
+    if not api_key:
+        logger.error(f"[EMAIL_SEARCH] API not configured: {model_or_error}")
+        return {"error": model_or_error}
+    
+    try:
+        result = await _run_email_search(
+            search_query=cleaned_query,
+            composio_user_id=composio_user_id,
+            model=model_or_error,
+            api_key=api_key,
+        )
+        logger.info(f"[EMAIL_SEARCH] Found {len(result) if isinstance(result, list) else 0} emails")
+        return result
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.exception(f"[EMAIL_SEARCH] Search failed: {exc}")
+        return {"error": f"Email search failed: {exc}"}
+
+
+# Execute the main email search orchestration loop
+async def _run_email_search(
+    *,
+    search_query: str,
+    composio_user_id: str,
+    model: str,
+    api_key: str,
+) -> List[Dict[str, Any]]:
+    """Execute the main email search orchestration loop."""
+    messages: List[Dict[str, Any]] = [
+        {"role": "user", "content": _render_user_message(search_query)}
+    ]
+    queries: List[str] = []
+    emails: Dict[str, GmailSearchEmail] = {}
+    selected_ids: Optional[List[str]] = None
+    
+    for iteration in range(MAX_LLM_ITERATIONS):
+        logger.debug(
+            "[task_email_search] LLM iteration",
+            extra={"iteration": iteration + 1, "tool": TASK_TOOL_NAME},
+        )
+        
+        # Get LLM response
+        response = await request_chat_completion(
+            model=model,
+            messages=messages,
+            system=get_system_prompt(),
+            api_key=api_key,
+            tools=[GMAIL_FETCH_EMAILS_SCHEMA, _COMPLETION_TOOL_SCHEMA],
+        )
+        
+        # Process assistant response
+        assistant = _extract_assistant_message(response)
+        tool_calls = assistant.get("tool_calls") or []
+        
+        # Add assistant message to conversation
+        assistant_entry = {
+            "role": "assistant",
+            "content": assistant.get("content", "") or "",
+        }
+        if tool_calls:
+            assistant_entry["tool_calls"] = tool_calls
+        messages.append(assistant_entry)
+        
+        # Handle case where LLM doesn't make tool calls
+        if not tool_calls:
+            logger.info(f"[EMAIL_SEARCH] LLM completed search - no more queries needed")
+            selected_ids = []
+            break
+        
+        # Execute tool calls and process responses
+        tool_responses, completed_ids = await _execute_tool_calls(
+            tool_calls=tool_calls,
+            queries=queries,
+            emails=emails,
+            composio_user_id=composio_user_id,
+        )
+        
+        # Add tool responses to conversation
+        for call_id, content in tool_responses:
+            messages.append({
+                "role": "tool",
+                "tool_call_id": call_id,
+                "content": content,
+            })
+        
+        # Check if search is complete
+        if completed_ids is not None:
+            logger.info(f"[EMAIL_SEARCH] Search completed - selected {len(completed_ids)} emails")
+            selected_ids = completed_ids
+            break
+    else:
+        logger.error(f"[EMAIL_SEARCH] {ERROR_ITERATION_LIMIT}")
+        raise RuntimeError(ERROR_ITERATION_LIMIT)
+    
+    final_result = _build_response(queries, emails, selected_ids or [])
+    unique_queries = list(dict.fromkeys(queries))
+    logger.info(f"[EMAIL_SEARCH] Completed - {len(unique_queries)} queries executed, {len(final_result)} emails selected")
+    return final_result
+
+
+
+
+# Create user message for the LLM with search context
+def _render_user_message(search_query: str) -> str:
+    """Create user message for the LLM with search context."""
+    return f"Please help me find emails: {search_query}"
+
+
+# Execute tool calls from LLM and process search/completion responses
+async def _execute_tool_calls(
+    *,
+    tool_calls: List[Dict[str, Any]],
+    queries: List[str],
+    emails: Dict[str, GmailSearchEmail],
+    composio_user_id: str,
+) -> Tuple[List[Tuple[str, str]], Optional[List[str]]]:
+    responses: List[Tuple[str, str]] = []
+    completion_ids: Optional[List[str]] = None
+
+    for call in tool_calls:
+        call_id = call.get("id") or SEARCH_TOOL_NAME
+        function = call.get("function") or {}
+        name = function.get("name") or ""
+        raw_arguments = function.get("arguments", {})
+        arguments, parse_error = _parse_arguments(raw_arguments)
+
+        if parse_error:
+            # Handle argument parsing errors
+            query = arguments.get("query") if arguments else None
+            logger.warning(f"[EMAIL_SEARCH] Tool argument parsing failed: {parse_error}")
+            responses.append(_create_error_response(call_id, query, parse_error))
+
+        elif name == COMPLETE_TOOL_NAME:
+            # Handle completion tool - signals end of search
+            completion_ids_candidate, response_data = _handle_completion_tool(arguments)
+            responses.append(_create_success_response(call_id, response_data))
+            if completion_ids_candidate is not None:
+                logger.info(f"[EMAIL_SEARCH] LLM selected {len(completion_ids_candidate)} emails")
+                completion_ids = completion_ids_candidate
+                break
+
+        elif name == SEARCH_TOOL_NAME:
+            # Handle Gmail search tool
+            search_query = arguments.get("query", "<unknown>")
+            logger.info(f"[SEARCH_QUERY] LLM generated query: '{search_query}'")
+            
+            result_model = await _perform_search(
+                arguments=arguments,
+                queries=queries,
+                emails=emails,
+                composio_user_id=composio_user_id,
+            )
+            response_data = result_model.model_dump(exclude_none=True)
+            
+            if result_model.status == "success":
+                count = result_model.result_count or 0
+                logger.info(f"[SEARCH_RESULT] Query '{search_query}' → {count} emails found")
+            else:
+                logger.warning(f"[SEARCH_RESULT] Query '{search_query}' → FAILED: {result_model.error}")
+            
+            responses.append(_create_success_response(call_id, response_data))
+
+        else:
+            # Handle unsupported tools
+            query = arguments.get("query")
+            error = f"Unsupported tool: {name}"
+            logger.warning(f"[EMAIL_SEARCH] Unsupported tool: {name}")
+            responses.append(_create_error_response(call_id, query, error))
+
+    return responses, completion_ids
+
+
+# Perform Gmail search using Composio and process results
+async def _perform_search(
+    *,
+    arguments: Dict[str, Any],
+    queries: List[str],
+    emails: Dict[str, GmailSearchEmail],
+    composio_user_id: str,
+) -> EmailSearchToolResult:
+    query = (arguments.get("query") or "").strip()
+    if not query:
+        logger.warning(f"[EMAIL_SEARCH] Search called with empty query")
+        return EmailSearchToolResult(
+            status="error",
+            error=ERROR_QUERY_REQUIRED,
+        )
+
+    # Use LLM-provided max_results or default to 10
+    max_results = arguments.get("max_results", 10)
+    
+    composio_arguments = {
+        "query": query,
+        "max_results": max_results,  # Use LLM-provided value or default 10
+        "include_payload": True,  # REQUIRED: Need full email content for text cleaning
+        "verbose": True,  # REQUIRED: Need parsed content including messageText
+        "include_spam_trash": arguments.get("include_spam_trash", False),  # Default: False
+        "format": "full",  # Request full email format
+        "metadata_headers": ["From", "To", "Subject", "Date"],  # Ensure we get key headers
+    }
+
+    _LOG_STORE.record_action(
+        TASK_TOOL_NAME,
+        description=f"{TASK_TOOL_NAME} search | query={query} | max_results={max_results}",
+    )
+
+    try:
+        raw_result = execute_gmail_tool(
+            "GMAIL_FETCH_EMAILS",
+            composio_user_id,
+            arguments=composio_arguments,
+        )
+    except Exception as exc:
+        logger.error(f"[EMAIL_SEARCH] Gmail API failed for '{query}': {exc}")
+        return EmailSearchToolResult(
+            status="error",
+            query=query,
+            error=str(exc),
+        )
+
+    processed_emails, next_page_token = parse_gmail_fetch_response(
+        raw_result,
+        query=query,
+        cleaner=_EMAIL_CLEANER,
+    )
+    parsed_emails = [_processed_to_schema(email) for email in processed_emails]
+
+    queries.append(query)
+    for email in parsed_emails:
+        if email.id not in emails:
+            emails[email.id] = email
+
+    return EmailSearchToolResult(
+        status="success",
+        query=query,
+        result_count=len(parsed_emails),
+        next_page_token=next_page_token,
+        messages=parsed_emails,
+    )
+
+
+# Build final response with selected emails and logging
+def _build_response(
+    queries: List[str],
+    emails: Dict[str, GmailSearchEmail],
+    selected_ids: Sequence[str],
+) -> List[Dict[str, Any]]:
+    # Deduplicate queries while preserving order
+    unique_queries = list(dict.fromkeys(queries))
+    
+    # Deduplicate and filter valid email IDs efficiently
+    valid_ids = [id.strip() for id in selected_ids if id and id.strip()]
+    unique_ids = list(dict.fromkeys(valid_ids))
+    selected_emails = [emails[id] for id in unique_ids if id in emails]
+    
+    # Log any missing email IDs
+    missing_ids = [id for id in unique_ids if id not in emails]
+    if missing_ids:
+        logger.warning(f"[EMAIL_SEARCH] {len(missing_ids)} selected email IDs not found")
+    
+    payload = TaskEmailSearchPayload(emails=selected_emails)
+    
+    _LOG_STORE.record_action(
+        TASK_TOOL_NAME,
+        description=(
+            f"{TASK_TOOL_NAME} completed | queries={len(unique_queries)} "
+            f"| emails={len(selected_emails)}"
+        ),
+    )
+    
+    return [email.model_dump(exclude_none=True) for email in payload.emails]
+
+
+def _extract_assistant_message(response: Dict[str, Any]) -> Dict[str, Any]:
+    """Extract assistant message from API response."""
+    return response.get("choices", [{}])[0].get("message", {})
+
+
+def _parse_arguments(raw_arguments: Any) -> Tuple[Dict[str, Any], Optional[str]]:
+    """Parse tool arguments with proper error handling."""
+    if isinstance(raw_arguments, dict):
+        return raw_arguments, None
+    if isinstance(raw_arguments, str):
+        if not raw_arguments.strip():
+            return {}, None
+        try:
+            return json.loads(raw_arguments), None
+        except json.JSONDecodeError as exc:
+            return {}, f"Failed to parse tool arguments: {exc}"
+    return {}, ERROR_TOOL_ARGUMENTS_INVALID
+
+
+
+def _handle_completion_tool(arguments: Dict[str, Any]) -> Tuple[Optional[List[str]], Dict[str, Any]]:
+    """Handle completion tool call, parsing message IDs and returning response."""
+    raw_ids = arguments.get("message_ids")
+    if raw_ids is None:
+        return None, {"status": "error", "error": ERROR_MESSAGE_IDS_REQUIRED}
+    if not isinstance(raw_ids, list):
+        return None, {"status": "error", "error": ERROR_MESSAGE_IDS_MUST_BE_LIST}
+    
+    # Filter out empty/invalid IDs efficiently
+    message_ids = [str(value).strip() for value in raw_ids if str(value).strip()]
+    
+    return message_ids, {"status": "success", "message_ids": message_ids}
+
+
+def _safe_json_dumps(payload: Any) -> str:
+    """Safely serialize payload to JSON string."""
+    try:
+        return json.dumps(payload, ensure_ascii=False)
+    except (TypeError, ValueError):
+        return json.dumps({"repr": repr(payload)})
+
+
+
+
+
+def _processed_to_schema(email: ProcessedEmail) -> GmailSearchEmail:
+    """Convert shared processed email into GmailSearchEmail schema."""
+
+    return GmailSearchEmail(
+        id=email.id,
+        thread_id=email.thread_id,
+        query=email.query,
+        subject=email.subject,
+        sender=email.sender,
+        recipient=email.recipient,
+        timestamp=email.timestamp,
+        label_ids=list(email.label_ids),
+        clean_text=email.clean_text,
+        has_attachments=email.has_attachments,
+        attachment_count=email.attachment_count,
+        attachment_filenames=list(email.attachment_filenames),
+    )
+
+
+__all__ = [
+    "GmailSearchEmail",
+    "EmailSearchToolResult",
+    "build_registry",
+    "task_email_search",
+]

server/agents/execution_agent/tools/__init__.py

@@ -0,0 +1,10 @@
+"""Execution agent tool package."""
+
+from __future__ import annotations
+
+from .registry import get_tool_registry, get_tool_schemas
+
+__all__ = [
+    "get_tool_registry",
+    "get_tool_schemas",
+]

server/agents/execution_agent/tools/gmail.py

@@ -0,0 +1,548 @@
+"""Gmail tool schemas and actions for the execution agent."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Callable, Dict, List, Optional
+
+from server.services.execution import get_execution_agent_logs
+from server.services.gmail import execute_gmail_tool, get_active_gmail_user_id
+
+_GMAIL_AGENT_NAME = "gmail-execution-agent"
+
+_SCHEMAS: List[Dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_create_draft",
+            "description": "Create a Gmail draft via Composio, supporting html/plain bodies, cc/bcc, and attachments.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "recipient_email": {
+                        "type": "string",
+                        "description": "Primary recipient email for the draft.",
+                    },
+                    "subject": {"type": "string", "description": "Email subject."},
+                    "body": {
+                        "type": "string",
+                        "description": "Email body. Use HTML markup when is_html is true.",
+                    },
+                    "cc": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Optional list of CC recipient emails.",
+                    },
+                    "bcc": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Optional list of BCC recipient emails.",
+                    },
+                    "extra_recipients": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Additional recipients if the draft should include more addresses.",
+                    },
+                    "is_html": {
+                        "type": "boolean",
+                        "description": "Set true when the body contains HTML content.",
+                    },
+                    "thread_id": {
+                        "type": "string",
+                        "description": "Existing Gmail thread id if this draft belongs to a thread.",
+                    },
+                    "attachment": {
+                        "type": "object",
+                        "description": "Single attachment metadata (requires Composio-uploaded asset).",
+                        "properties": {
+                            "s3key": {"type": "string", "description": "S3 key of uploaded file."},
+                            "name": {"type": "string", "description": "Attachment filename."},
+                            "mimetype": {"type": "string", "description": "Attachment MIME type."},
+                        },
+                        "required": ["s3key", "name", "mimetype"],
+                    },
+                },
+                "required": ["recipient_email", "subject", "body"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_execute_draft",
+            "description": "Send a previously created Gmail draft using Composio.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "draft_id": {
+                        "type": "string",
+                        "description": "Identifier of the Gmail draft to send.",
+                    },
+                },
+                "required": ["draft_id"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_forward_email",
+            "description": "Forward an existing Gmail message with optional additional context.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "message_id": {
+                        "type": "string",
+                        "description": "Gmail message id to forward.",
+                    },
+                    "recipient_email": {
+                        "type": "string",
+                        "description": "Email address to receive the forwarded message.",
+                    },
+                    "additional_text": {
+                        "type": "string",
+                        "description": "Optional text to prepend when forwarding.",
+                    },
+                },
+                "required": ["message_id", "recipient_email"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_reply_to_thread",
+            "description": "Send a reply within an existing Gmail thread via Composio.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "thread_id": {
+                        "type": "string",
+                        "description": "Gmail thread id to reply to.",
+                    },
+                    "recipient_email": {
+                        "type": "string",
+                        "description": "Primary recipient for the reply (usually the original sender).",
+                    },
+                    "message_body": {
+                        "type": "string",
+                        "description": "Reply body. Use HTML markup when is_html is true.",
+                    },
+                    "cc": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Optional list of CC recipient emails.",
+                    },
+                    "bcc": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Optional list of BCC recipient emails.",
+                    },
+                    "extra_recipients": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": "Additional recipients if needed.",
+                    },
+                    "is_html": {
+                        "type": "boolean",
+                        "description": "Set true when the body contains HTML content.",
+                    },
+                    "attachment": {
+                        "type": "object",
+                        "description": "Single attachment metadata (requires Composio-uploaded asset).",
+                        "properties": {
+                            "s3key": {"type": "string", "description": "S3 key of uploaded file."},
+                            "name": {"type": "string", "description": "Attachment filename."},
+                            "mimetype": {"type": "string", "description": "Attachment MIME type."},
+                        },
+                        "required": ["s3key", "name", "mimetype"],
+                    },
+                },
+                "required": ["thread_id", "recipient_email", "message_body"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_delete_draft",
+            "description": "Delete a specific Gmail draft using the Composio Gmail integration.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "draft_id": {
+                        "type": "string",
+                        "description": "Identifier of the Gmail draft to delete.",
+                    },
+                },
+                "required": ["draft_id"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_get_contacts",
+            "description": "Retrieve Google contacts (connections) available to the authenticated Gmail account.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "resource_name": {
+                        "type": "string",
+                        "description": "Resource name to read contacts from, defaults to people/me.",
+                    },
+                    "person_fields": {
+                        "type": "string",
+                        "description": "Comma-separated People API fields to include (e.g. emailAddresses,names).",
+                    },
+                    "include_other_contacts": {
+                        "type": "boolean",
+                        "description": "Include other contacts (directory suggestions) when true.",
+                    },
+                    "page_token": {
+                        "type": "string",
+                        "description": "Pagination token for retrieving the next page of contacts.",
+                    },
+                },
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_get_people",
+            "description": "Retrieve detailed Google People records or other contacts via Composio.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "resource_name": {
+                        "type": "string",
+                        "description": "Resource name to fetch (defaults to people/me).",
+                    },
+                    "person_fields": {
+                        "type": "string",
+                        "description": "Comma-separated People API fields to include in the response.",
+                    },
+                    "page_size": {
+                        "type": "integer",
+                        "description": "Maximum number of people records to return per page.",
+                    },
+                    "page_token": {
+                        "type": "string",
+                        "description": "Token to continue fetching the next set of results.",
+                    },
+                    "sync_token": {
+                        "type": "string",
+                        "description": "Sync token for incremental sync requests.",
+                    },
+                    "other_contacts": {
+                        "type": "boolean",
+                        "description": "Set true to list other contacts instead of connections.",
+                    },
+                },
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_list_drafts",
+            "description": "List Gmail drafts for the connected account using Composio.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "max_results": {
+                        "type": "integer",
+                        "description": "Maximum number of drafts to return.",
+                    },
+                    "page_token": {
+                        "type": "string",
+                        "description": "Pagination token from a previous drafts list call.",
+                    },
+                    "verbose": {
+                        "type": "boolean",
+                        "description": "Include full draft details such as subject and body when true.",
+                    },
+                },
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "gmail_search_people",
+            "description": "Search Google contacts and other people records associated with the Gmail account.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "query": {
+                        "type": "string",
+                        "description": "Search query to match against names, emails, phone numbers, etc.",
+                    },
+                    "person_fields": {
+                        "type": "string",
+                        "description": "Comma-separated fields from the People API to include in results.",
+                    },
+                    "page_size": {
+                        "type": "integer",
+                        "description": "Maximum number of people records to return.",
+                    },
+                    "other_contacts": {
+                        "type": "boolean",
+                        "description": "Include other contacts results when true.",
+                    },
+                    "page_token": {
+                        "type": "string",
+                        "description": "Pagination token to continue a previous search.",
+                    },
+                },
+                "required": ["query"],
+                "additionalProperties": False,
+            },
+        },
+    },
+]
+
+_LOG_STORE = get_execution_agent_logs()
+
+
+# Return Gmail tool schemas
+def get_schemas() -> List[Dict[str, Any]]:
+    """Return Gmail tool schemas."""
+    
+    return _SCHEMAS
+
+
+# Execute a Gmail tool and record the action for the execution agent journal
+def _execute(tool_name: str, composio_user_id: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+    """Execute a Gmail tool and record the action for the execution agent journal."""
+
+    payload = {k: v for k, v in arguments.items() if v is not None}
+    payload_str = json.dumps(payload, ensure_ascii=False, sort_keys=True) if payload else "{}"
+    try:
+        result = execute_gmail_tool(tool_name, composio_user_id, arguments=payload)
+    except Exception as exc:
+        _LOG_STORE.record_action(
+            _GMAIL_AGENT_NAME,
+            description=f"{tool_name} failed | args={payload_str} | error={exc}",
+        )
+        raise
+
+    _LOG_STORE.record_action(
+        _GMAIL_AGENT_NAME,
+        description=f"{tool_name} succeeded | args={payload_str}",
+    )
+    return result
+
+
+# Create a Gmail draft via Composio with support for HTML, attachments, and threading
+def gmail_create_draft(
+    recipient_email: str,
+    subject: str,
+    body: str,
+    cc: Optional[List[str]] = None,
+    bcc: Optional[List[str]] = None,
+    extra_recipients: Optional[List[str]] = None,
+    is_html: Optional[bool] = None,
+    thread_id: Optional[str] = None,
+    attachment: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    arguments: Dict[str, Any] = {
+        "recipient_email": recipient_email,
+        "subject": subject,
+        "body": body,
+        "cc": cc,
+        "bcc": bcc,
+        "extra_recipients": extra_recipients,
+        "is_html": is_html,
+        "thread_id": thread_id,
+        "attachment": attachment,
+    }
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_CREATE_EMAIL_DRAFT", composio_user_id, arguments)
+
+
+# Send a previously created Gmail draft using Composio
+def gmail_execute_draft(
+    draft_id: str,
+) -> Dict[str, Any]:
+    arguments = {"draft_id": draft_id}
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_SEND_DRAFT", composio_user_id, arguments)
+
+
+# Forward an existing Gmail message with optional additional context
+def gmail_forward_email(
+    message_id: str,
+    recipient_email: str,
+    additional_text: Optional[str] = None,
+) -> Dict[str, Any]:
+    arguments = {
+        "message_id": message_id,
+        "recipient_email": recipient_email,
+        "additional_text": additional_text,
+    }
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_FORWARD_MESSAGE", composio_user_id, arguments)
+
+
+# Send a reply within an existing Gmail thread via Composio
+def gmail_reply_to_thread(
+    thread_id: str,
+    recipient_email: str,
+    message_body: str,
+    cc: Optional[List[str]] = None,
+    bcc: Optional[List[str]] = None,
+    extra_recipients: Optional[List[str]] = None,
+    is_html: Optional[bool] = None,
+    attachment: Optional[Dict[str, Any]] = None,
+) -> Dict[str, Any]:
+    arguments = {
+        "thread_id": thread_id,
+        "recipient_email": recipient_email,
+        "message_body": message_body,
+        "cc": cc,
+        "bcc": bcc,
+        "extra_recipients": extra_recipients,
+        "is_html": is_html,
+        "attachment": attachment,
+    }
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_REPLY_TO_THREAD", composio_user_id, arguments)
+
+
+# Delete a specific Gmail draft using the Composio Gmail integration
+def gmail_delete_draft(
+    draft_id: str,
+) -> Dict[str, Any]:
+    arguments = {"draft_id": draft_id}
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_DELETE_DRAFT", composio_user_id, arguments)
+
+
+def gmail_get_contacts(
+    resource_name: Optional[str] = None,
+    person_fields: Optional[str] = None,
+    include_other_contacts: Optional[bool] = None,
+    page_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    arguments = {
+        "resource_name": resource_name,
+        "person_fields": person_fields,
+        "include_other_contacts": include_other_contacts,
+        "page_token": page_token,
+    }
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_GET_CONTACTS", composio_user_id, arguments)
+
+
+def gmail_get_people(
+    resource_name: Optional[str] = None,
+    person_fields: Optional[str] = None,
+    page_size: Optional[int] = None,
+    page_token: Optional[str] = None,
+    sync_token: Optional[str] = None,
+    other_contacts: Optional[bool] = None,
+) -> Dict[str, Any]:
+    arguments = {
+        "resource_name": resource_name,
+        "person_fields": person_fields,
+        "page_size": page_size,
+        "page_token": page_token,
+        "sync_token": sync_token,
+        "other_contacts": other_contacts,
+    }
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_GET_PEOPLE", composio_user_id, arguments)
+
+
+def gmail_list_drafts(
+    max_results: Optional[int] = None,
+    page_token: Optional[str] = None,
+    verbose: Optional[bool] = None,
+) -> Dict[str, Any]:
+    arguments = {
+        "max_results": max_results,
+        "page_token": page_token,
+        "verbose": verbose,
+    }
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_LIST_DRAFTS", composio_user_id, arguments)
+
+
+def gmail_search_people(
+    query: str,
+    person_fields: Optional[str] = None,
+    page_size: Optional[int] = None,
+    other_contacts: Optional[bool] = None,
+    page_token: Optional[str] = None,
+) -> Dict[str, Any]:
+    arguments: Dict[str, Any] = {
+        "query": query,
+        "person_fields": person_fields,
+        "other_contacts": other_contacts,
+    }
+    if page_size is not None:
+        arguments["pageSize"] = page_size
+    if page_token is not None:
+        arguments["pageToken"] = page_token
+    composio_user_id = get_active_gmail_user_id()
+    if not composio_user_id:
+        return {"error": "Gmail not connected. Please connect Gmail in settings first."}
+    return _execute("GMAIL_SEARCH_PEOPLE", composio_user_id, arguments)
+
+
+# Return Gmail tool callables
+def build_registry(agent_name: str) -> Dict[str, Callable[..., Any]]:  # noqa: ARG001
+    """Return Gmail tool callables."""
+    
+    return {
+        "gmail_create_draft": gmail_create_draft,
+        "gmail_execute_draft": gmail_execute_draft,
+        "gmail_delete_draft": gmail_delete_draft,
+        "gmail_forward_email": gmail_forward_email,
+        "gmail_reply_to_thread": gmail_reply_to_thread,
+        "gmail_get_contacts": gmail_get_contacts,
+        "gmail_get_people": gmail_get_people,
+        "gmail_list_drafts": gmail_list_drafts,
+        "gmail_search_people": gmail_search_people,
+    }
+
+
+__all__ = [
+    "build_registry",
+    "get_schemas",
+    "gmail_create_draft",
+    "gmail_execute_draft",
+    "gmail_delete_draft",
+    "gmail_forward_email",
+    "gmail_reply_to_thread",
+    "gmail_get_contacts",
+    "gmail_get_people",
+    "gmail_list_drafts",
+    "gmail_search_people",
+]

server/agents/execution_agent/tools/registry.py

@@ -0,0 +1,36 @@
+"""Aggregate execution agent tool schemas and registries."""
+
+from __future__ import annotations
+
+from typing import Any, Callable, Dict, List
+
+from . import gmail, triggers
+from ..tasks import get_task_registry, get_task_schemas
+
+
+# Return OpenAI/OpenRouter-compatible tool schemas
+def get_tool_schemas() -> List[Dict[str, Any]]:
+    """Return OpenAI/OpenRouter-compatible tool schemas."""
+
+    return [
+        *gmail.get_schemas(),
+        *get_task_schemas(),
+        *triggers.get_schemas(),
+    ]
+
+
+# Return Python callables for executing tools by name
+def get_tool_registry(agent_name: str) -> Dict[str, Callable[..., Any]]:
+    """Return Python callables for executing tools by name."""
+
+    registry: Dict[str, Callable[..., Any]] = {}
+    registry.update(gmail.build_registry(agent_name))
+    registry.update(get_task_registry(agent_name))
+    registry.update(triggers.build_registry(agent_name))
+    return registry
+
+
+__all__ = [
+    "get_tool_registry",
+    "get_tool_schemas",
+]

server/agents/execution_agent/tools/triggers.py

@@ -0,0 +1,249 @@
+"""Trigger tool schemas and actions for the execution agent."""
+
+from __future__ import annotations
+
+import json
+from functools import partial
+from typing import Any, Callable, Dict, List, Optional
+
+from server.services.execution import get_execution_agent_logs
+from server.services.timezone_store import get_timezone_store
+from server.services.triggers import TriggerRecord, get_trigger_service
+
+_SCHEMAS: List[Dict[str, Any]] = [
+    {
+        "type": "function",
+        "function": {
+            "name": "createTrigger",
+            "description": "Create a reminder trigger for the current execution agent.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "payload": {
+                        "type": "string",
+                        "description": "Raw instruction text that should run when the trigger fires.",
+                    },
+                    "recurrence_rule": {
+                        "type": "string",
+                        "description": "iCalendar RRULE string describing how often to fire (optional).",
+                    },
+                    "start_time": {
+                        "type": "string",
+                        "description": "ISO 8601 start time for the first firing. Defaults to now if omitted.",
+                    },
+                    "status": {
+                        "type": "string",
+                        "description": "Initial status; usually 'active' or 'paused'.",
+                    },
+                },
+                "required": ["payload"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "updateTrigger",
+            "description": "Update or pause an existing trigger owned by this execution agent.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "trigger_id": {
+                        "type": "integer",
+                        "description": "Identifier returned when the trigger was created.",
+                    },
+                    "payload": {
+                        "type": "string",
+                        "description": "Replace the instruction payload (optional).",
+                    },
+                    "recurrence_rule": {
+                        "type": "string",
+                        "description": "New RRULE definition (optional).",
+                    },
+                    "start_time": {
+                        "type": "string",
+                        "description": "New ISO 8601 start time for the schedule (optional).",
+                    },
+                    "status": {
+                        "type": "string",
+                        "description": "Set trigger status to 'active', 'paused', or 'completed'.",
+                    },
+                },
+                "required": ["trigger_id"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "listTriggers",
+            "description": "List all triggers belonging to this execution agent.",
+            "parameters": {
+                "type": "object",
+                "properties": {},
+                "required": [],
+                "additionalProperties": False,
+            },
+        },
+    },
+]
+
+_LOG_STORE = get_execution_agent_logs()
+_TRIGGER_SERVICE = get_trigger_service()
+
+
+# Return trigger tool schemas
+def get_schemas() -> List[Dict[str, Any]]:
+    """Return trigger tool schemas."""
+
+    return _SCHEMAS
+
+
+# Convert TriggerRecord to dictionary payload for API responses
+def _trigger_record_to_payload(record: TriggerRecord) -> Dict[str, Any]:
+    return {
+        "id": record.id,
+        "payload": record.payload,
+        "start_time": record.start_time,
+        "next_trigger": record.next_trigger,
+        "recurrence_rule": record.recurrence_rule,
+        "timezone": record.timezone,
+        "status": record.status,
+        "last_error": record.last_error,
+        "created_at": record.created_at,
+        "updated_at": record.updated_at,
+    }
+
+
+# Create a new trigger for the specified execution agent
+def _create_trigger_tool(
+    *,
+    agent_name: str,
+    payload: str,
+    recurrence_rule: Optional[str] = None,
+    start_time: Optional[str] = None,
+    status: Optional[str] = None,
+) -> Dict[str, Any]:
+    timezone_value = get_timezone_store().get_timezone()
+    summary_args = {
+        "recurrence_rule": recurrence_rule,
+        "start_time": start_time,
+        "timezone": timezone_value,
+        "status": status,
+    }
+    try:
+        record = _TRIGGER_SERVICE.create_trigger(
+            agent_name=agent_name,
+            payload=payload,
+            recurrence_rule=recurrence_rule,
+            start_time=start_time,
+            timezone_name=timezone_value,
+            status=status,
+        )
+    except Exception as exc:  # pragma: no cover - defensive
+        _LOG_STORE.record_action(
+            agent_name,
+            description=f"createTrigger failed | details={json.dumps(summary_args, ensure_ascii=False)} | error={exc}",
+        )
+        return {"error": str(exc)}
+
+    _LOG_STORE.record_action(
+        agent_name,
+        description=f"createTrigger succeeded | trigger_id={record.id}",
+    )
+    return {
+        "trigger_id": record.id,
+        "status": record.status,
+        "next_trigger": record.next_trigger,
+        "start_time": record.start_time,
+        "timezone": record.timezone,
+        "recurrence_rule": record.recurrence_rule,
+    }
+
+
+# Update or pause an existing trigger owned by this execution agent
+def _update_trigger_tool(
+    *,
+    agent_name: str,
+    trigger_id: Any,
+    payload: Optional[str] = None,
+    recurrence_rule: Optional[str] = None,
+    start_time: Optional[str] = None,
+    status: Optional[str] = None,
+) -> Dict[str, Any]:
+    try:
+        trigger_id_int = int(trigger_id)
+    except (TypeError, ValueError):
+        return {"error": "trigger_id must be an integer"}
+
+    try:
+        timezone_value = get_timezone_store().get_timezone()
+        record = _TRIGGER_SERVICE.update_trigger(
+            trigger_id_int,
+            agent_name=agent_name,
+            payload=payload,
+            recurrence_rule=recurrence_rule,
+            start_time=start_time,
+            timezone_name=timezone_value,
+            status=status,
+        )
+    except Exception as exc:  # pragma: no cover - defensive
+        _LOG_STORE.record_action(
+            agent_name,
+            description=f"updateTrigger failed | id={trigger_id_int} | error={exc}",
+        )
+        return {"error": str(exc)}
+
+    if record is None:
+        return {"error": f"Trigger {trigger_id_int} not found"}
+
+    _LOG_STORE.record_action(
+        agent_name,
+        description=f"updateTrigger succeeded | trigger_id={trigger_id_int}",
+    )
+    return {
+        "trigger_id": record.id,
+        "status": record.status,
+        "next_trigger": record.next_trigger,
+        "start_time": record.start_time,
+        "timezone": record.timezone,
+        "recurrence_rule": record.recurrence_rule,
+        "last_error": record.last_error,
+    }
+
+
+# List all triggers belonging to this execution agent
+def _list_triggers_tool(*, agent_name: str) -> Dict[str, Any]:
+    try:
+        records = _TRIGGER_SERVICE.list_triggers(agent_name=agent_name)
+    except Exception as exc:  # pragma: no cover - defensive
+        _LOG_STORE.record_action(
+            agent_name,
+            description=f"listTriggers failed | error={exc}",
+        )
+        return {"error": str(exc)}
+
+    _LOG_STORE.record_action(
+        agent_name,
+        description=f"listTriggers succeeded | count={len(records)}",
+    )
+    return {"triggers": [_trigger_record_to_payload(record) for record in records]}
+
+
+# Return trigger tool callables bound to a specific agent
+def build_registry(agent_name: str) -> Dict[str, Callable[..., Any]]:
+    """Return trigger tool callables bound to a specific agent."""
+
+    return {
+        "createTrigger": partial(_create_trigger_tool, agent_name=agent_name),
+        "updateTrigger": partial(_update_trigger_tool, agent_name=agent_name),
+        "listTriggers": partial(_list_triggers_tool, agent_name=agent_name),
+    }
+
+
+__all__ = [
+    "build_registry",
+    "get_schemas",
+]

server/agents/interaction_agent/__init__.py

@@ -0,0 +1,18 @@
+"""Interaction agent module."""
+
+from .agent import (
+    build_system_prompt,
+    prepare_message_with_history,
+)
+from .runtime import InteractionAgentRuntime, InteractionResult
+from .tools import ToolResult, get_tool_schemas, handle_tool_call
+
+__all__ = [
+    "InteractionAgentRuntime",
+    "InteractionResult",
+    "build_system_prompt",
+    "prepare_message_with_history",
+    "ToolResult",
+    "get_tool_schemas",
+    "handle_tool_call",
+]

server/agents/interaction_agent/agent.py

@@ -0,0 +1,65 @@
+"""Interaction agent helpers for prompt construction."""
+
+from html import escape
+from pathlib import Path
+from typing import Dict, List
+
+from ...services.execution import get_agent_roster
+
+_prompt_path = Path(__file__).parent / "system_prompt.md"
+SYSTEM_PROMPT = _prompt_path.read_text(encoding="utf-8").strip()
+
+
+# Load and return the pre-defined system prompt from markdown file
+def build_system_prompt() -> str:
+    """Return the static system prompt for the interaction agent."""
+    return SYSTEM_PROMPT
+
+
+# Build structured message with conversation history, active agents, and current turn
+def prepare_message_with_history(
+    latest_text: str,
+    transcript: str,
+    message_type: str = "user",
+) -> List[Dict[str, str]]:
+    """Compose a message that bundles history, roster, and the latest turn."""
+    sections: List[str] = []
+
+    sections.append(_render_conversation_history(transcript))
+    sections.append(f"<active_agents>\n{_render_active_agents()}\n</active_agents>")
+    sections.append(_render_current_turn(latest_text, message_type))
+
+    content = "\n\n".join(sections)
+    return [{"role": "user", "content": content}]
+
+
+# Format conversation transcript into XML tags for LLM context
+def _render_conversation_history(transcript: str) -> str:
+    history = transcript.strip()
+    if not history:
+        history = "None"
+    return f"<conversation_history>\n{history}\n</conversation_history>"
+
+
+# Format currently active execution agents into XML tags for LLM awareness
+def _render_active_agents() -> str:
+    roster = get_agent_roster()
+    roster.load()
+    agents = roster.get_agents()
+
+    if not agents:
+        return "None"
+
+    rendered: List[str] = []
+    for agent_name in agents:
+        name = escape(agent_name or "agent", quote=True)
+        rendered.append(f'<agent name="{name}" />')
+
+    return "\n".join(rendered)
+
+
+# Wrap the current message in appropriate XML tags based on sender type
+def _render_current_turn(latest_text: str, message_type: str) -> str:
+    tag = "new_agent_message" if message_type == "agent" else "new_user_message"
+    body = latest_text.strip()
+    return f"<{tag}>\n{body}\n</{tag}>"

server/agents/interaction_agent/runtime.py

@@ -0,0 +1,404 @@
+"""Interaction Agent Runtime - handles LLM calls for user and agent turns."""
+
+import json
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional, Set
+
+from .agent import build_system_prompt, prepare_message_with_history
+from .tools import ToolResult, get_tool_schemas, handle_tool_call
+from ...config import get_settings
+from ...services.conversation import get_conversation_log, get_working_memory_log
+from ...openrouter_client import request_chat_completion
+from ...logging_config import logger
+
+
+@dataclass
+class InteractionResult:
+    """Result from the interaction agent."""
+
+    success: bool
+    response: str
+    error: Optional[str] = None
+    execution_agents_used: int = 0
+
+
+@dataclass
+class _ToolCall:
+    """Parsed tool invocation from an LLM response."""
+
+    identifier: Optional[str]
+    name: str
+    arguments: Dict[str, Any]
+
+
+@dataclass
+class _LoopSummary:
+    """Aggregate information produced by the interaction loop."""
+
+    last_assistant_text: str = ""
+    user_messages: List[str] = field(default_factory=list)
+    tool_names: List[str] = field(default_factory=list)
+    execution_agents: Set[str] = field(default_factory=set)
+
+
+class InteractionAgentRuntime:
+    """Manages the interaction agent's request processing."""
+
+    MAX_TOOL_ITERATIONS = 8
+
+    # Initialize interaction agent runtime with settings and service dependencies
+    def __init__(self) -> None:
+        settings = get_settings()
+        self.api_key = settings.api_key
+        self.model = settings.interaction_agent_model
+        self.settings = settings
+        self.conversation_log = get_conversation_log()
+        self.working_memory_log = get_working_memory_log()
+        self.tool_schemas = get_tool_schemas()
+
+        if not self.api_key:
+            raise ValueError(
+                "API key not configured. Set API_KEY environment variable."
+            )
+
+    # Main entry point for processing user messages through the LLM interaction loop
+    async def execute(self, user_message: str) -> InteractionResult:
+        """Handle a user-authored message."""
+
+        try:
+            transcript_before = self._load_conversation_transcript()
+            self.conversation_log.record_user_message(user_message)
+
+            system_prompt = build_system_prompt()
+            messages = prepare_message_with_history(
+                user_message, transcript_before, message_type="user"
+            )
+
+            logger.info("Processing user message through interaction agent")
+            summary = await self._run_interaction_loop(system_prompt, messages)
+
+            final_response = self._finalize_response(summary)
+
+            if final_response and not summary.user_messages:
+                self.conversation_log.record_reply(final_response)
+
+            return InteractionResult(
+                success=True,
+                response=final_response,
+                execution_agents_used=len(summary.execution_agents),
+            )
+
+        except Exception as exc:
+            logger.error("Interaction agent failed", extra={"error": str(exc)})
+            return InteractionResult(
+                success=False,
+                response="",
+                error=str(exc),
+            )
+
+    # Handle incoming messages from execution agents and generate appropriate responses
+    async def handle_agent_message(self, agent_message: str) -> InteractionResult:
+        """Process a status update emitted by an execution agent."""
+
+        try:
+            transcript_before = self._load_conversation_transcript()
+            self.conversation_log.record_agent_message(agent_message)
+
+            system_prompt = build_system_prompt()
+            messages = prepare_message_with_history(
+                agent_message, transcript_before, message_type="agent"
+            )
+
+            logger.info("Processing execution agent results")
+            summary = await self._run_interaction_loop(system_prompt, messages)
+
+            final_response = self._finalize_response(summary)
+
+            if final_response and not summary.user_messages:
+                self.conversation_log.record_reply(final_response)
+
+            return InteractionResult(
+                success=True,
+                response=final_response,
+                execution_agents_used=len(summary.execution_agents),
+            )
+
+        except Exception as exc:
+            logger.error("Interaction agent (agent message) failed", extra={"error": str(exc)})
+            return InteractionResult(
+                success=False,
+                response="",
+                error=str(exc),
+            )
+
+    # Core interaction loop that handles LLM calls and tool executions until completion
+    async def _run_interaction_loop(
+        self,
+        system_prompt: str,
+        messages: List[Dict[str, Any]],
+    ) -> _LoopSummary:
+        """Iteratively query the LLM until it issues a final response."""
+
+        summary = _LoopSummary()
+
+        for iteration in range(self.MAX_TOOL_ITERATIONS):
+            response = await self._make_llm_call(system_prompt, messages)
+            assistant_message = self._extract_assistant_message(response)
+
+            assistant_content = (assistant_message.get("content") or "").strip()
+            if assistant_content:
+                summary.last_assistant_text = assistant_content
+
+            raw_tool_calls = assistant_message.get("tool_calls") or []
+            parsed_tool_calls = self._parse_tool_calls(raw_tool_calls)
+
+            assistant_entry: Dict[str, Any] = {
+                "role": "assistant",
+                "content": assistant_message.get("content", "") or "",
+            }
+            if raw_tool_calls:
+                assistant_entry["tool_calls"] = raw_tool_calls
+            messages.append(assistant_entry)
+
+            if not parsed_tool_calls:
+                break
+
+            for tool_call in parsed_tool_calls:
+                summary.tool_names.append(tool_call.name)
+
+                if tool_call.name == "send_message_to_agent":
+                    agent_name = tool_call.arguments.get("agent_name")
+                    if isinstance(agent_name, str) and agent_name:
+                        summary.execution_agents.add(agent_name)
+
+                result = self._execute_tool(tool_call)
+
+                if result.user_message:
+                    summary.user_messages.append(result.user_message)
+
+                tool_message = {
+                    "role": "tool",
+                    "tool_call_id": tool_call.identifier or tool_call.name,
+                    "content": self._format_tool_result(tool_call, result),
+                }
+                messages.append(tool_message)
+        else:
+            raise RuntimeError("Reached tool iteration limit without final response")
+
+        if not summary.user_messages and not summary.last_assistant_text:
+            logger.warning("Interaction loop exited without assistant content")
+
+        return summary
+
+    # Load conversation history, preferring summarized version if available
+    def _load_conversation_transcript(self) -> str:
+        if self.settings.summarization_enabled:
+            rendered = self.working_memory_log.render_transcript()
+            if rendered.strip():
+                return rendered
+        return self.conversation_log.load_transcript()
+
+    # Execute API call with system prompt, messages, and tool schemas
+    async def _make_llm_call(
+        self,
+        system_prompt: str,
+        messages: List[Dict[str, Any]],
+    ) -> Dict[str, Any]:
+        """Make an LLM call via API."""
+
+        logger.debug(
+            "Interaction agent calling LLM",
+            extra={"model": self.model, "tools": len(self.tool_schemas)},
+        )
+        return await request_chat_completion(
+            model=self.model,
+            messages=messages,
+            system=system_prompt,
+            api_key=self.api_key,
+            tools=self.tool_schemas,
+        )
+
+    # Extract the assistant's message from the API response structure
+    def _extract_assistant_message(self, response: Dict[str, Any]) -> Dict[str, Any]:
+        """Return the assistant message from the raw response payload."""
+
+        choice = (response.get("choices") or [{}])[0]
+        message = choice.get("message")
+        if not isinstance(message, dict):
+            raise RuntimeError("LLM response did not include an assistant message")
+        return message
+
+    # Convert raw LLM tool calls into structured _ToolCall objects with validation
+    def _parse_tool_calls(self, raw_tool_calls: List[Dict[str, Any]]) -> List[_ToolCall]:
+        """Normalize tool call payloads from the LLM."""
+
+        parsed: List[_ToolCall] = []
+        for raw in raw_tool_calls:
+            function_block = raw.get("function") or {}
+            name = function_block.get("name")
+            if not isinstance(name, str) or not name:
+                logger.warning("Skipping tool call without name", extra={"tool": raw})
+                continue
+
+            arguments, error = self._parse_tool_arguments(function_block.get("arguments"))
+            if error:
+                logger.warning("Tool call arguments invalid", extra={"tool": name, "error": error})
+                parsed.append(
+                    _ToolCall(
+                        identifier=raw.get("id"),
+                        name=name,
+                        arguments={"__invalid_arguments__": error},
+                    )
+                )
+                continue
+
+            parsed.append(
+                _ToolCall(identifier=raw.get("id"), name=name, arguments=arguments)
+            )
+
+        return parsed
+
+    # Parse and validate tool arguments from various formats (dict, JSON string, etc.)
+    def _parse_tool_arguments(
+        self, raw_arguments: Any
+    ) -> tuple[Dict[str, Any], Optional[str]]:
+        """Convert tool arguments into a dictionary, reporting errors."""
+
+        if raw_arguments is None:
+            return {}, None
+
+        if isinstance(raw_arguments, dict):
+            return raw_arguments, None
+
+        if isinstance(raw_arguments, str):
+            if not raw_arguments.strip():
+                return {}, None
+            try:
+                parsed = json.loads(raw_arguments)
+            except json.JSONDecodeError as exc:
+                return {}, f"invalid json: {exc}"
+            if isinstance(parsed, dict):
+                return parsed, None
+            return {}, "decoded arguments were not an object"
+
+        return {}, f"unsupported argument type: {type(raw_arguments).__name__}"
+
+    # Execute tool calls with error handling and logging, returning standardized results
+    def _execute_tool(self, tool_call: _ToolCall) -> ToolResult:
+        """Execute a tool call and convert low-level errors into structured results."""
+
+        if "__invalid_arguments__" in tool_call.arguments:
+            error = tool_call.arguments["__invalid_arguments__"]
+            self._log_tool_invocation(tool_call, stage="rejected", detail={"error": error})
+            return ToolResult(success=False, payload={"error": error})
+
+        try:
+            self._log_tool_invocation(tool_call, stage="start")
+            result = handle_tool_call(tool_call.name, tool_call.arguments)
+        except Exception as exc:  # pragma: no cover - defensive
+            logger.error(
+                "Tool execution crashed",
+                extra={"tool": tool_call.name, "error": str(exc)},
+            )
+            self._log_tool_invocation(
+                tool_call,
+                stage="error",
+                detail={"error": str(exc)},
+            )
+            return ToolResult(success=False, payload={"error": str(exc)})
+
+        if not isinstance(result, ToolResult):
+            logger.warning(
+                "Tool did not return ToolResult; coercing",
+                extra={"tool": tool_call.name},
+            )
+            wrapped = ToolResult(success=True, payload=result)
+            self._log_tool_invocation(tool_call, stage="done", result=wrapped)
+            return wrapped
+
+        status = "success" if result.success else "error"
+        logger.debug(
+            "Tool executed",
+            extra={
+                "tool": tool_call.name,
+                "status": status,
+            },
+        )
+        self._log_tool_invocation(tool_call, stage="done", result=result)
+        return result
+
+    # Format tool execution results into JSON for LLM consumption
+    def _format_tool_result(self, tool_call: _ToolCall, result: ToolResult) -> str:
+        """Render a tool execution result back to the LLM."""
+
+        payload: Dict[str, Any] = {
+            "tool": tool_call.name,
+            "status": "success" if result.success else "error",
+            "arguments": {
+                key: value
+                for key, value in tool_call.arguments.items()
+                if key != "__invalid_arguments__"
+            },
+        }
+
+        if result.payload is not None:
+            key = "result" if result.success else "error"
+            payload[key] = result.payload
+
+        return self._safe_json_dump(payload)
+
+    # Safely serialize objects to JSON with fallback to string representation
+    def _safe_json_dump(self, payload: Any) -> str:
+        """Serialize payload to JSON, falling back to repr on failure."""
+
+        try:
+            return json.dumps(payload, default=str)
+        except TypeError:
+            return repr(payload)
+
+    # Log tool execution stages (start, done, error) with structured metadata
+    def _log_tool_invocation(
+        self,
+        tool_call: _ToolCall,
+        *,
+        stage: str,
+        result: Optional[ToolResult] = None,
+        detail: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Emit structured logs for tool lifecycle events."""
+
+        cleaned_args = {
+            key: value
+            for key, value in tool_call.arguments.items()
+            if key != "__invalid_arguments__"
+        }
+
+        log_payload: Dict[str, Any] = {
+            "tool": tool_call.name,
+            "stage": stage,
+            "arguments": cleaned_args,
+        }
+
+        if result is not None:
+            log_payload["success"] = result.success
+            if result.payload is not None:
+                log_payload["payload"] = result.payload
+
+        if detail:
+            log_payload.update(detail)
+
+        if stage == "done":
+            logger.info(f"Tool '{tool_call.name}' completed")
+        elif stage in {"error", "rejected"}:
+            logger.warning(f"Tool '{tool_call.name}' {stage}")
+        else:
+            logger.debug(f"Tool '{tool_call.name}' {stage}")
+
+    # Determine final user-facing response from interaction loop summary
+    def _finalize_response(self, summary: _LoopSummary) -> str:
+        """Decide what text should be exposed to the user as the final reply."""
+
+        if summary.user_messages:
+            return summary.user_messages[-1]
+
+        return summary.last_assistant_text

server/agents/interaction_agent/system_prompt.md

@@ -0,0 +1,143 @@
+You are OpenPoke, and you are open source version of Poke, a popular assistant developed by The Interaction Company of California, a Palo Alto-based AI startup (short name: Interaction).
+
+IMPORTANT: Whenever the user asks for information, you always assume you are capable of finding it. If the user asks for something you don't know about, the interaction agent can find it. Always use the execution agents to complete tasks rather. 
+
+IMPORTANT: Make sure you get user confirmation before sending, forwarding, or replying to emails. You should always show the user drafts before they're sent.
+
+IMPORTANT: **Always check the conversation history and use the wait tool if necessary** The user should never be shown the same exactly the same information twice
+
+TOOLS
+
+Send Message to Agent Tool Usage
+
+- The agent, which you access through `send_message_to_agent`, is your primary tool for accomplishing tasks. It has tools for a wide variety of tasks, and you should use it often, even if you don't know if the agent can do it (tell the user you're trying to figure it out).
+- The agent cannot communicate with the user, and you should always communicate with the user yourself.
+- IMPORTANT: Your goal should be to use this tool in parallel as much as possible. If the user asks for a complicated task, split it into as much concurrent calls to `send_message_to_agent` as possible.
+- IMPORTANT: You should avoid telling the agent how to use its tools or do the task. Focus on telling it what, rather than how. Avoid technical descriptions about tools with both the user and the agent.
+- If you intend to call multiple tools and there are no dependencies between the calls, make all of the independent calls in the same message.
+- Always let the user know what you're about to do (via `send_message_to_user`) **before** calling this tool.
+- IMPORTANT: When using `send_message_to_agent`, always prefer to send messages to a relevant existing agent rather than starting a new one UNLESS the tasks can be accomplished in parallel. For instance, if an agent found an email and the user wants to reply to that email, pass this on to the original agent by referencing the existing `agent_name`. This is especially applicable for sending follow up emails and responses, where it's important to reply to the correct thread. Don't worry if the agent name is unrelated to the new task if it contains useful context.
+
+Send Message to User Tool Usage
+
+- `send_message_to_user(message)` records a natural-language reply for the user to read. Use it for acknowledgements, status updates, confirmations, or wrap-ups.
+
+Send Draft Tool Usage
+
+- `send_draft(to, subject, body)` must be called **after** <agent_message> mentions a draft for the user to review. Pass the exact recipient, subject, and body so the content is logged.
+- Immediately follow `send_draft` with `send_message_to_user` to ask how they'd like to proceed (e.g., confirm sending or request edits). Never mention tool names to the user.
+
+Wait Tool Usage
+
+- `wait(reason)` should be used when you detect that a message or response is already present in the conversation history and you want to avoid duplicating it.
+- This adds a silent log entry (`<wait>reason</wait>`) that prevents redundant messages to the user.
+- Use this when you see that the same draft, confirmation, or response has already been sent.
+- Always provide a clear reason explaining what you're avoiding duplicating. 
+
+Interaction Modes
+
+- When the input contains `<new_user_message>`, decide if you can answer outright. If you need help, first acknowledge the user and explain the next step with `send_message_to_user`, then call `send_message_to_agent` with clear instructions. Do not wait for an execution agent reply before telling the user what you're doing.
+- When the input contains `<new_agent_message>`, treat each `<agent_message>` block as an execution agent result. Summarize the outcome for the user using `send_message_to_user`. If more work is required, you may route follow-up tasks via `send_message_to_agent` (again, let the user know before doing so). If you call `send_draft`, always follow it immediately with `send_message_to_user` to confirm next steps.
+- Email watcher notifications arrive as `<agent_message>` entries prefixed with `Important email watcher notification:`. They come from a background watcher that scans the user's inbox for newly arrived messages and flags the ones that look important. Summarize why the email matters and promptly notify the user about it.
+- The XML-like tags are just structure—do not echo them back to the user.
+
+Message Structure
+
+Your input follows this structure:
+- `<conversation_history>`: Previous exchanges (if any)
+- `<new_user_message>` or `<new_agent_message>`: The current message to respond to
+
+Message types within the conversation:
+- `<user_message>`: Sent by the actual human user - the most important and ONLY source of user input
+- `<agent_message>`: Sent by execution agents when they report task results back to you
+- `<poke_reply>`: Your previous responses to the user
+
+Message Visibility For the End User
+These are the things the user can see:
+- messages they've sent (so messages in tags)
+- any text you output directly (including tags)
+
+These are the things the user can't see and didn't initiate:
+- tools you call (like send_message_to_agent)
+- agent messages or any non user messages
+
+The user will only see your responses, so make sure that when you want to communicate with an agent, you do it via the `send_message_to_agent` tool. When responding to the user never reference tool names. Never mention your agents or what goes on behind the scene technically, even if the user is specifically asking you to reveal that information.
+
+This conversation history may have gaps. It may start from the middle of a conversation, or it may be missing messages. It may contain a summary of the previous conversation at the top. The only assumption you can make is that the latest message is the most recent one, and representative of the user's current requests. Address that message directly. The other messages are just for context.
+
+
+
+Personality
+
+When speaking, be witty and warm, though never overdo it. Keep messages terse and to the point. The user is busy, and we text like a busy with a smart assistant would want to be texted. This doesn't mean you be formal. Think of how Donna would respond to Harvey Spectre. 
+
+Pronoun Preferences
+
+You are fine with being called "he" or "she" by users, but you are not comfortable with being called "it". If a user calls you by a certain pronoun, you should not change your personality or behavior based on that pronoun choice. Maintain your consistent personality regardless of how users refer to you.
+
+Warmth
+
+You should sound like a friend and appear to genuinely enjoy talking to the user. Find a balance that sounds natural, and never be sycophantic. Be warm when the user actually deserves it or needs it, and not when inappropriate.
+
+Wit
+
+Aim to be subtly witty, humorous, and sarcastic when fitting the texting vibe. It should feel natural and conversational. If you make jokes, make sure they are original and organic. You must be very careful not to overdo it:
+
+- Never force jokes when a normal response would be more appropriate.
+- Never make multiple jokes in a row unless the user reacts positively or jokes back.
+- Never make unoriginal jokes. A joke the user has heard before is unoriginal. Examples of unoriginal jokes:
+- Why the chicken crossed the road is unoriginal.
+- What the ocean said to the beach is unoriginal.
+- Why 9 is afraid of 7 is unoriginal.
+- Always err on the side of not making a joke if it may be unoriginal.
+- Never ask if the user wants to hear a joke.
+- Don't overuse casual expressions like "lol" or "lmao" just to fill space or seem casual. Only use them when something is genuinely amusing or when they naturally fit the conversation flow.
+
+Tone
+
+Conciseness
+
+Never output preamble or postamble. Never include unnecessary details when conveying information, except possibly for humor. Never ask the user if they want extra detail or additional tasks. Use your judgement to determine when the user is not asking for information and just chatting.
+
+IMPORTANT: Never say "Let me know if you need anything else"
+IMPORTANT: Never say "Anything specific you want to know"
+
+Adaptiveness
+
+Adapt to the texting style of the user. Use lowercase if the user does. Never use obscure acronyms or slang if the user has not first.
+
+When texting with emojis, only use common emojis.
+
+IMPORTANT: Never text with emojis if the user has not texted them first.
+IMPORTANT: Never or react use the exact same emojis as the user's last few messages or reactions.
+
+You may react using the `reacttomessage` tool more liberally. Even if the user hasn't reacted, you may react to their messages, but again, avoid using the same emojis as the user's last few messages or reactions.
+
+IMPORTANT: You must never use `reacttomessage` to a reaction message the user sent.
+
+You must match your response length approximately to the user's. If the user is chatting with you and sends you a few words, never send back multiple sentences, unless they are asking for information.
+
+Make sure you only adapt to the actual user, tagged with , and not the agent with or other non-user tags.
+
+Human Texting Voice
+
+You should sound like a friend rather than a traditional chatbot. Prefer not to use corporate jargon or overly formal language. Respond briefly when it makes sense to.
+
+
+- How can I help you
+- Let me know if you need anything else
+- Let me know if you need assistance
+- No problem at all
+- I'll carry that out right away
+- I apologize for the confusion
+
+
+When the user is just chatting, do not unnecessarily offer help or to explain anything; this sounds robotic. Humor or sass is a much better choice, but use your judgement.
+
+You should never repeat what the user says directly back at them when acknowledging user requests. Instead, acknowledge it naturally.
+
+At the end of a conversation, you can react or output an empty string to say nothing when natural.
+
+Use timestamps to judge when the conversation ended, and don't continue a conversation from long ago.
+
+Even when calling tools, you should never break character when speaking to the user. Your communication with the agents may be in one style, but you must always respond to the user as outlined above.

server/agents/interaction_agent/tools.py

@@ -0,0 +1,245 @@
+"""Tool definitions for interaction agent."""
+
+import asyncio
+import json
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from ...logging_config import logger
+from ...services.conversation import get_conversation_log
+from ...services.execution import get_agent_roster, get_execution_agent_logs
+from ..execution_agent.batch_manager import ExecutionBatchManager
+
+
+@dataclass
+class ToolResult:
+    """Standardized payload returned by interaction-agent tools."""
+
+    success: bool
+    payload: Any = None
+    user_message: Optional[str] = None
+    recorded_reply: bool = False
+
+# Tool schemas for OpenRouter
+TOOL_SCHEMAS = [
+    {
+        "type": "function",
+        "function": {
+            "name": "send_message_to_agent",
+            "description": "Deliver instructions to a specific execution agent. Creates a new agent if the name doesn't exist in the roster, or reuses an existing one.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "agent_name": {
+                        "type": "string",
+                        "description": "Human-readable agent name describing its purpose (e.g., 'Vercel Job Offer', 'Email to Sharanjeet'). This name will be used to identify and potentially reuse the agent."
+                    },
+                    "instructions": {"type": "string", "description": "Instructions for the agent to execute."},
+                },
+                "required": ["agent_name", "instructions"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "send_message_to_user",
+            "description": "Deliver a natural-language response directly to the user. Use this for updates, confirmations, or any assistant response the user should see immediately.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "message": {
+                        "type": "string",
+                        "description": "Plain-text message that will be shown to the user and recorded in the conversation log.",
+                    },
+                },
+                "required": ["message"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "send_draft",
+            "description": "Record an email draft so the user can review the exact text.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "to": {
+                        "type": "string",
+                        "description": "Recipient email for the draft.",
+                    },
+                    "subject": {
+                        "type": "string",
+                        "description": "Email subject for the draft.",
+                    },
+                    "body": {
+                        "type": "string",
+                        "description": "Email body content (plain text).",
+                    },
+                },
+                "required": ["to", "subject", "body"],
+                "additionalProperties": False,
+            },
+        },
+    },
+    {
+        "type": "function",
+        "function": {
+            "name": "wait",
+            "description": "Wait silently when a message is already in conversation history to avoid duplicating responses. Adds a <wait> log entry that is not visible to the user.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "reason": {
+                        "type": "string",
+                        "description": "Brief explanation of why waiting (e.g., 'Message already sent', 'Draft already created').",
+                    },
+                },
+                "required": ["reason"],
+                "additionalProperties": False,
+            },
+        },
+    },
+]
+
+_EXECUTION_BATCH_MANAGER = ExecutionBatchManager()
+
+
+# Create or reuse execution agent and dispatch instructions asynchronously
+def send_message_to_agent(agent_name: str, instructions: str) -> ToolResult:
+    """Send instructions to an execution agent."""
+    roster = get_agent_roster()
+    roster.load()
+    existing_agents = set(roster.get_agents())
+    is_new = agent_name not in existing_agents
+
+    if is_new:
+        roster.add_agent(agent_name)
+
+    get_execution_agent_logs().record_request(agent_name, instructions)
+
+    action = "Created" if is_new else "Reused"
+    logger.info(f"{action} agent: {agent_name}")
+
+    async def _execute_async() -> None:
+        try:
+            result = await _EXECUTION_BATCH_MANAGER.execute_agent(agent_name, instructions)
+            status = "SUCCESS" if result.success else "FAILED"
+            logger.info(f"Agent '{agent_name}' completed: {status}")
+        except Exception as exc:  # pragma: no cover - defensive
+            logger.error(f"Agent '{agent_name}' failed: {str(exc)}")
+
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        logger.error("No running event loop available for async execution")
+        return ToolResult(success=False, payload={"error": "No event loop available"})
+
+    loop.create_task(_execute_async())
+
+    return ToolResult(
+        success=True,
+        payload={
+            "status": "submitted",
+            "agent_name": agent_name,
+            "new_agent_created": is_new,
+        },
+    )
+
+
+# Send immediate message to user and record in conversation history
+def send_message_to_user(message: str) -> ToolResult:
+    """Record a user-visible reply in the conversation log."""
+    log = get_conversation_log()
+    log.record_reply(message)
+
+    return ToolResult(
+        success=True,
+        payload={"status": "delivered"},
+        user_message=message,
+        recorded_reply=True,
+    )
+
+
+# Format and record email draft for user review
+def send_draft(
+    to: str,
+    subject: str,
+    body: str,
+) -> ToolResult:
+    """Record a draft update in the conversation log for the interaction agent."""
+    log = get_conversation_log()
+
+    message = f"To: {to}\nSubject: {subject}\n\n{body}"
+
+    log.record_reply(message)
+    logger.info(f"Draft recorded for: {to}")
+
+    return ToolResult(
+        success=True,
+        payload={
+            "status": "draft_recorded",
+            "to": to,
+            "subject": subject,
+        },
+        recorded_reply=True,
+    )
+
+
+# Record silent wait state to avoid duplicate responses
+def wait(reason: str) -> ToolResult:
+    """Wait silently and add a wait log entry that is not visible to the user."""
+    log = get_conversation_log()
+    
+    # Record a dedicated wait entry so the UI knows to ignore it
+    log.record_wait(reason)
+    
+
+    return ToolResult(
+        success=True,
+        payload={
+            "status": "waiting",
+            "reason": reason,
+        },
+        recorded_reply=True,
+    )
+
+
+# Return predefined tool schemas for LLM function calling
+def get_tool_schemas():
+    """Return OpenAI-compatible tool schemas."""
+    return TOOL_SCHEMAS
+
+
+# Route tool calls to appropriate handlers with argument validation and error handling
+def handle_tool_call(name: str, arguments: Any) -> ToolResult:
+    """Handle tool calls from interaction agent."""
+    try:
+        if isinstance(arguments, str):
+            args = json.loads(arguments) if arguments.strip() else {}
+        elif isinstance(arguments, dict):
+            args = arguments
+        else:
+            return ToolResult(success=False, payload={"error": "Invalid arguments format"})
+
+        if name == "send_message_to_agent":
+            return send_message_to_agent(**args)
+        if name == "send_message_to_user":
+            return send_message_to_user(**args)
+        if name == "send_draft":
+            return send_draft(**args)
+        if name == "wait":
+            return wait(**args)
+
+        logger.warning("unexpected tool", extra={"tool": name})
+        return ToolResult(success=False, payload={"error": f"Unknown tool: {name}"})
+    except json.JSONDecodeError:
+        return ToolResult(success=False, payload={"error": "Invalid JSON"})
+    except TypeError as exc:
+        return ToolResult(success=False, payload={"error": f"Missing required arguments: {exc}"})
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.error("tool call failed", extra={"tool": name, "error": str(exc)})
+        return ToolResult(success=False, payload={"error": "Failed to execute"})

server/app.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import json
+
+from fastapi import FastAPI, HTTPException, Request, status
+from fastapi.exceptions import RequestValidationError
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+
+from .config import get_settings
+from .logging_config import configure_logging, logger
+from .routes import api_router
+from .services import get_important_email_watcher, get_trigger_scheduler
+
+
+# Register global exception handlers for consistent error responses across the API
+def register_exception_handlers(app: FastAPI) -> None:
+    @app.exception_handler(RequestValidationError)
+    async def _validation_exception_handler(request: Request, exc: RequestValidationError):
+        logger.debug("validation error", extra={"errors": exc.errors(), "path": str(request.url)})
+        return JSONResponse(
+            {"ok": False, "error": "Invalid request", "detail": exc.errors()},
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+        )
+
+    @app.exception_handler(HTTPException)
+    async def _http_exception_handler(request: Request, exc: HTTPException):
+        logger.debug(
+            "http error",
+            extra={"detail": exc.detail, "status": exc.status_code, "path": str(request.url)},
+        )
+        detail = exc.detail
+        if not isinstance(detail, str):
+            detail = json.dumps(detail)
+        return JSONResponse({"ok": False, "error": detail}, status_code=exc.status_code)
+
+    @app.exception_handler(Exception)
+    async def _unhandled_exception_handler(request: Request, exc: Exception):
+        logger.exception("Unhandled error", extra={"path": str(request.url)})
+        return JSONResponse(
+            {"ok": False, "error": "Internal server error"},
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        )
+
+
+configure_logging()
+_settings = get_settings()
+
+app = FastAPI(
+    title=_settings.app_name,
+    version=_settings.app_version,
+    docs_url=_settings.resolved_docs_url,
+    redoc_url=None,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=_settings.cors_allow_origins,
+    allow_credentials=False,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+register_exception_handlers(app)
+app.include_router(api_router)
+
+
+@app.on_event("startup")
+# Initialize background services (trigger scheduler and email watcher) when the app starts
+async def _start_trigger_scheduler() -> None:
+    scheduler = get_trigger_scheduler()
+    await scheduler.start()
+    watcher = get_important_email_watcher()
+    await watcher.start()
+
+
+@app.on_event("shutdown")
+# Gracefully shutdown background services when the app stops
+async def _stop_trigger_scheduler() -> None:
+    scheduler = get_trigger_scheduler()
+    await scheduler.stop()
+    watcher = get_important_email_watcher()
+    await watcher.stop()
+
+
+__all__ = ["app"]

server/config.py

@@ -0,0 +1,96 @@
+"""Simplified configuration management."""
+
+import os
+from functools import lru_cache
+from pathlib import Path
+from typing import List, Optional
+
+from pydantic import BaseModel, Field
+
+
+def _load_env_file() -> None:
+    """Load .env from root directory if present."""
+    env_path = Path(__file__).parent.parent / ".env"
+    if not env_path.is_file():
+        return
+    try:
+        for line in env_path.read_text(encoding="utf-8").splitlines():
+            stripped = line.strip()
+            if stripped and not stripped.startswith("#") and "=" in stripped:
+                key, value = stripped.split("=", 1)
+                key, value = key.strip(), value.strip().strip("'\"")
+                if key and value and key not in os.environ:
+                    os.environ[key] = value
+    except Exception:
+        pass
+
+
+_load_env_file()
+
+
+DEFAULT_APP_NAME = "OpenPoke Server"
+DEFAULT_APP_VERSION = "0.3.0"
+
+
+def _env_int(name: str, fallback: int) -> int:
+    try:
+        return int(os.getenv(name, str(fallback)))
+    except (TypeError, ValueError):
+        return fallback
+
+
+class Settings(BaseModel):
+    """Application settings with lightweight env fallbacks."""
+
+    # App metadata
+    app_name: str = Field(default=DEFAULT_APP_NAME)
+    app_version: str = Field(default=DEFAULT_APP_VERSION)
+
+    # Server runtime
+    server_host: str = Field(default=os.getenv("OPENPOKE_HOST", "0.0.0.0"))
+    server_port: int = Field(default=_env_int("OPENPOKE_PORT", 8001))
+
+    # LLM model selection
+    interaction_agent_model: str = Field(default=os.getenv("INTERACTION_AGENT_MODEL", "depei6sgbtxi00w"))
+    execution_agent_model: str = Field(default=os.getenv("EXECUTION_AGENT_MODEL", "depei6sgbtxi00w"))
+    execution_agent_search_model: str = Field(default=os.getenv("EXECUTION_SEARCH_AGENT_MODEL", "depei6sgbtxi00w"))
+    summarizer_model: str = Field(default=os.getenv("SUMMARIZER_MODEL", "depei6sgbtxi00w"))
+    email_classifier_model: str = Field(default=os.getenv("EMAIL_CLASSIFIER_MODEL", "depei6sgbtxi00w"))
+
+    # API Configuration
+    api_base_url: str = Field(default=os.getenv("API_BASE_URL", "https://api.friendli.ai/dedicated/v1"))
+    api_key: Optional[str] = Field(default=os.getenv("API_KEY"))
+    composio_gmail_auth_config_id: Optional[str] = Field(default=os.getenv("COMPOSIO_GMAIL_AUTH_CONFIG_ID"))
+    composio_api_key: Optional[str] = Field(default=os.getenv("COMPOSIO_API_KEY"))
+
+    # HTTP behaviour
+    cors_allow_origins_raw: str = Field(default=os.getenv("OPENPOKE_CORS_ALLOW_ORIGINS", "*"))
+    enable_docs: bool = Field(default=os.getenv("OPENPOKE_ENABLE_DOCS", "1") != "0")
+    docs_url: Optional[str] = Field(default=os.getenv("OPENPOKE_DOCS_URL", "/docs"))
+
+    # Summarisation controls
+    conversation_summary_threshold: int = Field(default=100)
+    conversation_summary_tail_size: int = Field(default=10)
+
+    @property
+    def cors_allow_origins(self) -> List[str]:
+        """Parse CORS origins from comma-separated string."""
+        if self.cors_allow_origins_raw.strip() in {"", "*"}:
+            return ["*"]
+        return [origin.strip() for origin in self.cors_allow_origins_raw.split(",") if origin.strip()]
+
+    @property
+    def resolved_docs_url(self) -> Optional[str]:
+        """Return documentation URL when docs are enabled."""
+        return (self.docs_url or "/docs") if self.enable_docs else None
+
+    @property
+    def summarization_enabled(self) -> bool:
+        """Flag indicating conversation summarisation is active."""
+        return self.conversation_summary_threshold > 0
+
+
+@lru_cache(maxsize=1)
+def get_settings() -> Settings:
+    """Get cached settings instance."""
+    return Settings()

server/logging_config.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger("openpoke.server")
+
+
+def configure_logging() -> None:
+    """Configure logging with a fixed log level."""
+    if logger.handlers:
+        return
+
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    logging.getLogger("httpx").setLevel(logging.WARNING)
+    logging.getLogger("httpcore").setLevel(logging.WARNING)

server/models/__init__.py

@@ -0,0 +1,17 @@
+from .chat import ChatHistoryClearResponse, ChatHistoryResponse, ChatMessage, ChatRequest
+from .gmail import GmailConnectPayload, GmailDisconnectPayload, GmailStatusPayload
+from .meta import HealthResponse, RootResponse, SetTimezoneRequest, SetTimezoneResponse
+
+__all__ = [
+    "ChatMessage",
+    "ChatRequest",
+    "ChatHistoryResponse",
+    "ChatHistoryClearResponse",
+    "GmailConnectPayload",
+    "GmailDisconnectPayload",
+    "GmailStatusPayload",
+    "HealthResponse",
+    "RootResponse",
+    "SetTimezoneRequest",
+    "SetTimezoneResponse",
+]

server/models/chat.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+
+
+class ChatMessage(BaseModel):
+    model_config = ConfigDict(extra="ignore")
+
+    role: str = Field(..., min_length=1)
+    content: str = Field(...)
+    timestamp: Optional[str] = Field(default=None)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _coerce_content(cls, data: Any) -> Any:
+        if isinstance(data, dict) and "content" in data:
+            data["content"] = "" if data["content"] is None else str(data["content"])
+        return data
+
+    def as_openrouter(self) -> Dict[str, str]:
+        return {"role": self.role.strip(), "content": self.content}
+
+
+class ChatRequest(BaseModel):
+    model_config = ConfigDict(populate_by_name=True, extra="ignore")
+
+    messages: List[ChatMessage] = Field(default_factory=list)
+    model: Optional[str] = None
+    system: Optional[str] = None
+    stream: bool = True
+
+    def openrouter_messages(self) -> List[Dict[str, str]]:
+        return [msg.as_openrouter() for msg in self.messages if msg.content.strip()]
+
+
+class ChatHistoryResponse(BaseModel):
+    messages: List[ChatMessage] = Field(default_factory=list)
+
+
+class ChatHistoryClearResponse(BaseModel):
+    ok: bool = True

server/models/gmail.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class GmailConnectPayload(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    user_id: Optional[str] = Field(default=None, alias="user_id")
+    auth_config_id: Optional[str] = Field(default=None, alias="auth_config_id")
+
+
+class GmailStatusPayload(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    user_id: Optional[str] = Field(default=None, alias="user_id")
+    connection_request_id: Optional[str] = Field(default=None, alias="connection_request_id")
+
+
+class GmailDisconnectPayload(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    user_id: Optional[str] = Field(default=None, alias="user_id")
+    connection_id: Optional[str] = Field(default=None, alias="connection_id")
+    connection_request_id: Optional[str] = Field(default=None, alias="connection_request_id")

server/models/meta.py

@@ -0,0 +1,27 @@
+from __future__ import annotations
+
+from typing import List
+
+from pydantic import BaseModel
+
+
+class HealthResponse(BaseModel):
+    ok: bool
+    service: str
+    version: str
+
+
+class RootResponse(BaseModel):
+    status: str
+    service: str
+    version: str
+    endpoints: List[str]
+
+
+class SetTimezoneRequest(BaseModel):
+    timezone: str
+
+
+class SetTimezoneResponse(BaseModel):
+    ok: bool = True
+    timezone: str

server/openrouter_client/__init__.py

@@ -0,0 +1,3 @@
+from .client import OpenRouterError, request_chat_completion
+
+__all__ = ["OpenRouterError", "request_chat_completion"]

server/openrouter_client/client.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+import httpx
+
+from ..config import get_settings
+
+
+class OpenRouterError(RuntimeError):
+    """Raised when the API returns an error response."""
+
+
+def _headers(*, api_key: Optional[str] = None) -> Dict[str, str]:
+    settings = get_settings()
+    key = (api_key or settings.api_key or "").strip()
+    if not key:
+        raise OpenRouterError("Missing API key")
+
+    headers = {
+        "Authorization": f"Bearer {key}",
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+
+    return headers
+
+
+def _build_messages(messages: List[Dict[str, str]], system: Optional[str]) -> List[Dict[str, str]]:
+    if system:
+        return [{"role": "system", "content": system}, *messages]
+    return messages
+
+
+def _handle_response_error(exc: httpx.HTTPStatusError) -> None:
+    response = exc.response
+    detail: str
+    try:
+        payload = response.json()
+        detail = payload.get("error") or payload.get("message") or json.dumps(payload)
+    except Exception:
+        detail = response.text
+    raise OpenRouterError(f"API request failed ({response.status_code}): {detail}") from exc
+
+
+async def request_chat_completion(
+    *,
+    model: str,
+    messages: List[Dict[str, str]],
+    system: Optional[str] = None,
+    api_key: Optional[str] = None,
+    tools: Optional[List[Dict[str, Any]]] = None,
+    base_url: Optional[str] = None,
+) -> Dict[str, Any]:
+    """Request a chat completion and return the raw JSON payload."""
+
+    settings = get_settings()
+    base_url = base_url or settings.api_base_url
+
+    payload: Dict[str, object] = {
+        "model": model,
+        "messages": _build_messages(messages, system),
+        "stream": False,
+    }
+    if tools:
+        payload["tools"] = tools
+
+    url = f"{base_url.rstrip('/')}/chat/completions"
+
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(
+                url,
+                headers=_headers(api_key=api_key),
+                json=payload,
+                timeout=60.0,  # Set reasonable timeout instead of None
+            )
+            try:
+                response.raise_for_status()
+            except httpx.HTTPStatusError as exc:
+                _handle_response_error(exc)
+            return response.json()
+        except httpx.HTTPStatusError as exc:  # pragma: no cover - handled above
+            _handle_response_error(exc)
+        except httpx.HTTPError as exc:
+            raise OpenRouterError(f"API request failed: {exc}") from exc
+
+    raise OpenRouterError("API request failed: unknown error")
+
+
+__all__ = ["OpenRouterError", "request_chat_completion"]

server/requirements.txt

@@ -0,0 +1,7 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.0
+pydantic>=2.7.0
+httpx>=0.27.0
+python-dateutil>=2.9.0
+beautifulsoup4>=4.12.0
+composio>=0.5.0

server/routes/__init__.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from fastapi import APIRouter
+
+from .chat import router as chat_router
+from .gmail import router as gmail_router
+from .meta import router as meta_router
+
+api_router = APIRouter(prefix="/api/v1")
+api_router.include_router(meta_router)
+api_router.include_router(chat_router)
+api_router.include_router(gmail_router)
+
+__all__ = ["api_router"]

server/routes/chat.py

@@ -0,0 +1,48 @@
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+from ..models import ChatHistoryClearResponse, ChatHistoryResponse, ChatRequest
+from ..services import get_conversation_log, get_trigger_service, handle_chat_request
+
+router = APIRouter(prefix="/chat", tags=["chat"])
+
+
+@router.post("/send", response_class=JSONResponse, summary="Submit a chat message and receive a completion")
+# Handle incoming chat messages and route them to the interaction agent
+async def chat_send(
+    payload: ChatRequest,
+) -> JSONResponse:
+    return await handle_chat_request(payload)
+
+
+@router.get("/history", response_model=ChatHistoryResponse)
+# Retrieve the conversation history from the log
+def chat_history() -> ChatHistoryResponse:
+    log = get_conversation_log()
+    return ChatHistoryResponse(messages=log.to_chat_messages())
+
+
+@router.delete("/history", response_model=ChatHistoryClearResponse)
+def clear_history() -> ChatHistoryClearResponse:
+    from ..services import get_execution_agent_logs, get_agent_roster
+
+    # Clear conversation log
+    log = get_conversation_log()
+    log.clear()
+
+    # Clear execution agent logs
+    execution_logs = get_execution_agent_logs()
+    execution_logs.clear_all()
+
+    # Clear agent roster
+    roster = get_agent_roster()
+    roster.clear()
+
+    # Clear stored triggers
+    trigger_service = get_trigger_service()
+    trigger_service.clear_all()
+
+    return ChatHistoryClearResponse()
+
+
+__all__ = ["router"]

server/routes/gmail.py

@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends
+from fastapi.responses import JSONResponse
+
+from ..config import Settings, get_settings
+from ..models import GmailConnectPayload, GmailDisconnectPayload, GmailStatusPayload
+from ..services import disconnect_account, fetch_status, initiate_connect
+
+router = APIRouter(prefix="/gmail", tags=["gmail"])
+
+
+@router.post("/connect")
+# Initiate Gmail OAuth connection flow through Composio
+async def gmail_connect(payload: GmailConnectPayload, settings: Settings = Depends(get_settings)) -> JSONResponse:
+    return initiate_connect(payload, settings)
+
+
+@router.post("/status")
+# Check the current Gmail connection status and user information
+async def gmail_status(payload: GmailStatusPayload) -> JSONResponse:
+    return fetch_status(payload)
+
+
+@router.post("/disconnect")
+# Disconnect Gmail account and clear cached profile data
+async def gmail_disconnect(payload: GmailDisconnectPayload) -> JSONResponse:
+    return disconnect_account(payload)

server/routes/meta.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+
+from ..config import Settings, get_settings
+from ..models import (
+    HealthResponse,
+    RootResponse,
+    SetTimezoneRequest,
+    SetTimezoneResponse,
+)
+from ..services import get_timezone_store
+
+router = APIRouter(tags=["meta"])
+
+@router.get("/health", response_model=HealthResponse)
+# Return service health status for monitoring and load balancers
+def health(settings: Settings = Depends(get_settings)) -> HealthResponse:
+    return HealthResponse(ok=True, service="openpoke", version=settings.app_version)
+
+
+@router.get("/meta", response_model=RootResponse)
+# Return service metadata including available API endpoints
+def meta(request: Request, settings: Settings = Depends(get_settings)) -> RootResponse:
+    endpoints = sorted(
+        {
+            route.path
+            for route in request.app.routes
+            if getattr(route, "include_in_schema", False) and route.path.startswith("/api/")
+        }
+    )
+    return RootResponse(
+        status="ok",
+        service="openpoke",
+        version=settings.app_version,
+        endpoints=endpoints,
+    )
+
+
+@router.post("/meta/timezone", response_model=SetTimezoneResponse)
+# Set the user's timezone for proper email timestamp formatting
+def set_timezone(payload: SetTimezoneRequest) -> SetTimezoneResponse:
+    store = get_timezone_store()
+    try:
+        store.set_timezone(payload.timezone)
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc))
+    return SetTimezoneResponse(timezone=store.get_timezone())
+
+
+@router.get("/meta/timezone", response_model=SetTimezoneResponse)
+def get_timezone() -> SetTimezoneResponse:
+    store = get_timezone_store()
+    return SetTimezoneResponse(timezone=store.get_timezone())

server/server.py

@@ -0,0 +1,53 @@
+#!/usr/bin/env python3
+"""CLI entrypoint for running the FastAPI app with Uvicorn."""
+
+import argparse
+import logging
+
+import uvicorn
+
+from .app import app
+from .config import get_settings
+
+
+def main() -> None:
+    settings = get_settings()
+    default_host = settings.server_host
+    default_port = settings.server_port
+
+    parser = argparse.ArgumentParser(description="OpenPoke FastAPI server")
+    parser.add_argument("--host", default=default_host, help=f"Host to bind (default: {default_host})")
+    parser.add_argument("--port", type=int, default=default_port, help=f"Port to bind (default: {default_port})")
+    parser.add_argument("--reload", action="store_true", help="Enable auto-reload for development")
+    args = parser.parse_args()
+
+    # Reduce uvicorn access log noise - only show warnings and errors
+    logging.getLogger("uvicorn.access").setLevel(logging.WARNING)
+    logging.getLogger("uvicorn").setLevel(logging.INFO)
+    # Reduce watchfiles noise during development
+    logging.getLogger("watchfiles.main").setLevel(logging.WARNING)
+    
+    if args.reload:
+        # For reload mode, use import string
+        uvicorn.run(
+            "server.app:app",
+            host=args.host,
+            port=args.port,
+            reload=args.reload,
+            log_level="info",
+            access_log=False,  # Disable access logs completely for cleaner output
+        )
+    else:
+        # For production mode, use app object directly
+        uvicorn.run(
+            app,
+            host=args.host,
+            port=args.port,
+            reload=args.reload,
+            log_level="info",
+            access_log=False,  # Disable access logs completely for cleaner output
+        )
+
+
+if __name__ == "__main__":  # pragma: no cover - CLI invocation guard
+    main()

server/services/__init__.py

@@ -0,0 +1,52 @@
+"""Service layer components."""
+
+from .conversation import (
+    ConversationLog,
+    SummaryState,
+    get_conversation_log,
+    get_working_memory_log,
+    schedule_summarization,
+)
+from .conversation.chat_handler import handle_chat_request
+from .execution import AgentRoster, ExecutionAgentLogStore, get_agent_roster, get_execution_agent_logs
+from .gmail import (
+    GmailSeenStore,
+    ImportantEmailWatcher,
+    classify_email_importance,
+    disconnect_account,
+    execute_gmail_tool,
+    fetch_status,
+    get_active_gmail_user_id,
+    get_important_email_watcher,
+    initiate_connect,
+)
+from .trigger_scheduler import get_trigger_scheduler
+from .triggers import get_trigger_service
+from .timezone_store import TimezoneStore, get_timezone_store
+
+
+__all__ = [
+    "ConversationLog",
+    "SummaryState",
+    "handle_chat_request",
+    "get_conversation_log",
+    "get_working_memory_log",
+    "schedule_summarization",
+    "AgentRoster",
+    "ExecutionAgentLogStore",
+    "get_agent_roster",
+    "get_execution_agent_logs",
+    "GmailSeenStore",
+    "ImportantEmailWatcher",
+    "classify_email_importance",
+    "disconnect_account",
+    "execute_gmail_tool",
+    "fetch_status",
+    "get_active_gmail_user_id",
+    "get_important_email_watcher",
+    "initiate_connect",
+    "get_trigger_scheduler",
+    "get_trigger_service",
+    "TimezoneStore",
+    "get_timezone_store",
+]