Function bodies 1,000 total

def download_attachment(
    file_path: str = Query(
        ...,
        description="Path to the attachment file",
    ),
    reader: ChatDBReader = Depends(get_imessage_reader),
) -> FileResponse:
    """Download an attachment file.

    Returns the original attachment file for download.

    **Security:** Only files within the iMessage attachments directory can be accessed.
    """
    # Expand tilde in path
    if file_path.startswith("~"):
        file_path = str(Path(file_path).expanduser())

    # Security check: ensure the path is within the expected attachments directory
    attachments_base = Path.home() / "Library" / "Messages" / "Attachments"
    try:
        resolved_path = Path(file_path).resolve()
        resolved_base = attachments_base.resolve()
        # Use relative_to to ensure path is within base directory
        resolved_path.relative_to(resolved_base)
    except ValueError as e:
        raise HTTPException(
            status_code=403,
            detail="Acces

    def validate_output_dir(cls, v: str | None) -> str | None:
        """Validate output_dir is within user home directory."""
        if v is None:
            return v
        from pathlib import Path

        resolved = Path(v).resolve()
        home = Path.home()
        try:
            resolved.relative_to(home)
        except ValueError:
            raise ValueError(f"output_dir must be within user home directory ({home})")
        return str(resolved)

def _get_all_chat_ids(limit: int = 50) -> list[str]:
    """Get all chat IDs from iMessage database.

    Args:
        limit: Maximum number of conversations to return.

    Returns:
        List of chat IDs.
    """
    from integrations.imessage import ChatDBReader

    with ChatDBReader() as reader:
        conversations = reader.get_conversations(limit=limit)
        return [c.chat_id for c in conversations]

def batch_export(request: BatchExportRequest) -> BatchResponse:
    """Export multiple conversations as a background task.

    Creates a task to export the specified conversations. The task runs
    in the background and can be monitored via the tasks API.

    Args:
        request: Batch export request with chat IDs and format.

    Returns:
        Created task for tracking progress.
    """
    _ensure_worker_running()

    queue = get_task_queue()
    task = queue.enqueue(
        task_type=TaskType.BATCH_EXPORT,
        params={
            "chat_ids": request.chat_ids,
            "format": request.format,
            "output_dir": request.output_dir,
        },
    )

    return BatchResponse(
        task=_task_to_response(task),
        message=f"Batch export started for {len(request.chat_ids)} conversations",
    )

def batch_export_all(request: BatchExportAllRequest) -> BatchResponse:
    """Export all conversations as a background task.

    Fetches all conversation IDs and creates a task to export them.

    Args:
        request: Export options including format and limit.

    Returns:
        Created task for tracking progress.
    """
    try:
        chat_ids = _get_all_chat_ids(limit=request.limit)
    except Exception as e:
        raise iMessageQueryError(
            "Failed to fetch conversations for batch operation",
            cause=e,
        )

    if not chat_ids:
        raise HTTPException(
            status_code=404,
            detail="No conversations found",
        )

    _ensure_worker_running()

    queue = get_task_queue()
    task = queue.enqueue(
        task_type=TaskType.BATCH_EXPORT,
        params={
            "chat_ids": chat_ids,
            "format": request.format,
            "output_dir": request.output_dir,
        },
    )

    return BatchResponse(
    

def batch_summarize(request: BatchSummarizeRequest) -> BatchResponse:
    """Summarize multiple conversations as a background task.

    Creates a task to generate summaries for the specified conversations.

    Args:
        request: Batch summarize request with chat IDs.

    Returns:
        Created task for tracking progress.
    """
    _ensure_worker_running()

    queue = get_task_queue()
    task = queue.enqueue(
        task_type=TaskType.BATCH_SUMMARIZE,
        params={
            "chat_ids": request.chat_ids,
            "num_messages": request.num_messages,
        },
    )

    return BatchResponse(
        task=_task_to_response(task),
        message=f"Batch summarization started for {len(request.chat_ids)} conversations",
    )

def batch_summarize_recent(request: BatchSummarizeRecentRequest) -> BatchResponse:
    """Summarize recent conversations as a background task.

    Fetches the most recent conversation IDs and creates a task to summarize them.

    Args:
        request: Options including limit and message count.

    Returns:
        Created task for tracking progress.
    """
    try:
        chat_ids = _get_all_chat_ids(limit=request.limit)
    except Exception as e:
        raise iMessageQueryError(
            "Failed to fetch conversations for batch operation",
            cause=e,
        )

    if not chat_ids:
        raise HTTPException(
            status_code=404,
            detail="No conversations found",
        )

    _ensure_worker_running()

    queue = get_task_queue()
    task = queue.enqueue(
        task_type=TaskType.BATCH_SUMMARIZE,
        params={
            "chat_ids": chat_ids,
            "num_messages": request.num_messages,
        },
    )

    return BatchResponse(
  

def batch_generate_replies(request: BatchGenerateRepliesRequest) -> BatchResponse:
    """Generate reply suggestions for multiple conversations as a background task.

    Creates a task to generate reply suggestions for each specified conversation.

    Args:
        request: Batch reply request with chat IDs and options.

    Returns:
        Created task for tracking progress.
    """
    _ensure_worker_running()

    queue = get_task_queue()
    task = queue.enqueue(
        task_type=TaskType.BATCH_GENERATE_REPLIES,
        params={
            "chat_ids": request.chat_ids,
            "instruction": request.instruction,
            "num_suggestions": request.num_suggestions,
        },
    )

    return BatchResponse(
        task=_task_to_response(task),
        message=f"Batch reply generation started for {len(request.chat_ids)} conversations",
    )

def get_calendar_reader_dep() -> CalendarReaderImpl:
    """Get calendar reader with access check."""
    reader = get_calendar_reader()
    if not reader.check_access():
        from jarvis.core.exceptions import calendar_permission_denied

        raise calendar_permission_denied()
    return reader

def get_calendar_writer_dep() -> CalendarWriterImpl:
    """Get calendar writer with access check."""
    writer = get_calendar_writer()
    if not writer.check_access():
        from jarvis.core.exceptions import calendar_permission_denied

        raise calendar_permission_denied()
    return writer

async def list_calendars(
    reader: CalendarReaderImpl = Depends(get_calendar_reader_dep),
) -> list[CalendarResponse]:
    """List all available calendars from macOS Calendar.

    Returns calendars that the user has access to, including both
    local calendars and synced calendars (iCloud, Google, etc.).
    """
    # Run blocking calendar read in threadpool to avoid blocking the event loop
    calendars = await run_in_threadpool(reader.get_calendars)
    return [CalendarResponse.model_validate(cal) for cal in calendars]

def get_events(
    calendar_id: str | None = Query(
        default=None,
        description="Filter by calendar ID (all calendars if not specified)",
    ),
    days: int = Query(
        default=30,
        ge=1,
        le=365,
        description="Number of days to look ahead",
    ),
    limit: int = Query(
        default=50,
        ge=1,
        le=500,
        description="Maximum number of events to return",
    ),
    reader: CalendarReaderImpl = Depends(get_calendar_reader_dep),
) -> list[CalendarEventResponse]:
    """Get upcoming calendar events.

    Returns events from the specified date range, sorted by start time.
    """
    start = datetime.now(tz=UTC)
    end = start + timedelta(days=days)

    events = reader.get_events(
        calendar_id=calendar_id,
        start=start,
        end=end,
        limit=limit,
    )

    return [CalendarEventResponse.model_validate(evt) for evt in events]

def search_events(
    query: str = Query(..., min_length=1, description="Search query"),
    calendar_id: str | None = Query(
        default=None,
        description="Filter by calendar ID",
    ),
    limit: int = Query(
        default=50,
        ge=1,
        le=200,
        description="Maximum number of results",
    ),
    reader: CalendarReaderImpl = Depends(get_calendar_reader_dep),
) -> list[CalendarEventResponse]:
    """Search for calendar events by title, location, or notes.

    Searches upcoming events (up to 1 year) for the query text.
    """
    events = reader.search_events(
        query=query,
        calendar_id=calendar_id,
        limit=limit,
    )

    return [CalendarEventResponse.model_validate(evt) for evt in events]

def create_event(
    request: CreateEventRequest,
    writer: CalendarWriterImpl = Depends(get_calendar_writer_dep),
) -> CreateEventResponse:
    """Create a new calendar event.

    Creates an event in the specified calendar with the provided details.
    """
    result = writer.create_event(
        calendar_id=request.calendar_id,
        title=request.title,
        start=request.start,
        end=request.end,
        all_day=request.all_day,
        location=request.location,
        notes=request.notes,
        url=request.url,
    )

    return CreateEventResponse(
        success=result.success,
        event_id=result.event_id,
        error=result.error,
    )

def detect_events_in_text(
    request: DetectEventsRequest,
) -> list[DetectedEventResponse]:
    """Detect potential calendar events in text using NLP.

    Analyzes text for date/time mentions and event indicators,
    returning detected events with confidence scores.
    """
    detector = get_event_detector()
    events = detector.detect_events(
        text=request.text,
        message_id=request.message_id,
    )

    return [
        DetectedEventResponse(
            title=evt.title,
            start=evt.start,
            end=evt.end,
            location=evt.location,
            description=evt.description,
            all_day=evt.all_day,
            confidence=evt.confidence,
            source_text=evt.source_text,
            message_id=evt.message_id,
        )
        for evt in events
    ]

def detect_events_in_messages(
    request: DetectEventsFromMessagesRequest,
    imessage_reader: ChatDBReader = Depends(get_imessage_reader),
) -> list[DetectedEventResponse]:
    """Detect potential calendar events in conversation messages.

    Analyzes recent messages from a conversation for date/time mentions
    and event indicators, returning detected events with confidence scores.
    """
    # Get recent messages
    messages = imessage_reader.get_messages(
        chat_id=request.chat_id,
        limit=request.limit,
    )

    if not messages:
        return []

    # Detect events in each message
    detector = get_event_detector()
    all_events: list[DetectedEventResponse] = []

    for msg in messages:
        if not msg.text:
            continue

        events = detector.detect_events(
            text=msg.text,
            reference_date=msg.date,
            message_id=msg.id,
        )

        for evt in events:
            all_events.append(
                Detec

def create_event_from_detected(
    request: CreateEventFromDetectedRequest,
    writer: CalendarWriterImpl = Depends(get_calendar_writer_dep),
) -> CreateEventResponse:
    """Create a calendar event from a detected event.

    Takes a detected event and adds it to the specified calendar.
    This is the action triggered by "Add to Calendar" buttons.
    """
    detected = request.detected_event

    # Convert to DetectedEvent dataclass
    event = DetectedEvent(
        title=detected.title,
        start=detected.start,
        end=detected.end,
        location=detected.location,
        description=detected.description,
        all_day=detected.all_day,
        confidence=detected.confidence,
        source_text=detected.source_text,
        message_id=detected.message_id,
    )

    result = writer.create_event_from_detected(
        calendar_id=request.calendar_id,
        event=event,
    )

    return CreateEventResponse(
        success=result.success,
        event_id=result

    def __init__(
        self, ttl_seconds: float = AVATAR_CACHE_TTL_SECONDS, maxsize: int = AVATAR_CACHE_MAX_SIZE
    ) -> None:
        """Initialize the avatar cache.

        Args:
            ttl_seconds: Time-to-live for cached items in seconds
            maxsize: Maximum number of items to cache
        """
        # Cache structure: key -> (timestamp, data)
        # data is either bytes (image) or ContactAvatarData (for generating defaults)
        self._cache: dict[str, tuple[float, Any]] = {}
        self._ttl = ttl_seconds
        self._maxsize = maxsize
        self._lock = threading.Lock()
        self._hits = 0
        self._misses = 0

    def get(self, key: str) -> tuple[bool, Any]:
        """Get a value from the cache.

        Args:
            key: Cache key (normalized identifier)

        Returns:
            Tuple of (found, value). found is False if key doesn't exist or is expired.
        """
        with self._lock:
            if key in self._cache:
                timestamp, value = self._cache[key]
                if time.time() - timestamp < self._ttl:
                    self._hits += 1
                    return True, value
                # Expired - remove it
                del self._cache[key]
            self._misses += 1
            return False, None

    def set(self, key: str, value: Any) -> None:
        """Store a value in the cache.

        Args:
            key: Cache key (normalized identifier)
            value: Value to store
        """
        with self._lock:
            # Evict oldest entries if at capacity
            while len(self._cache) >= self._maxsize:
                oldest_key = min(self._cache.keys(), key=lambda k: self._cache[k][0])
                del self._cache[oldest_key]

            self._cache[key] = (time.time(), value)

    def invalidate(self, key: str | None = None) -> None:
        """Invalidate cache entries.

        Args:
            key: Specific key to invalidate (None for all)
        """
        with self._lock:
            if key is None:
                self._cache.clear()
            elif key in self._cache:
                del self._cache[key]

    def stats(self) -> dict[str, Any]:
        """Get cache statistics.

        Returns:
            Dictionary with cache stats
        """
        with self._lock:
            total = self._hits + self._misses
            return {
                "size": len(self._cache),
                "maxsize": self._maxsize,
                "ttl_seconds": self._ttl,
                "hits": self._hits,
                "misses": self._misses,
                "hit_rate": self._hits / total if total > 0 else 0.0,
            }

def get_avatar_cache() -> AvatarCache:
    """Get the global avatar cache instance.

    Returns:
        Shared AvatarCache instance
    """
    global _avatar_cache
    with _cache_lock:
        if _avatar_cache is None:
            _avatar_cache = AvatarCache()
        return _avatar_cache

def _get_color_for_identifier(identifier: str) -> str:
    """Get a consistent color for an identifier.

    Uses hash of identifier to always return the same color for
    the same contact.

    Args:
        identifier: Phone number or email

    Returns:
        Hex color string (e.g., "#FF6B6B")
    """
    hash_val = int(hashlib.md5(identifier.encode(), usedforsecurity=False).hexdigest(), 16)
    return AVATAR_COLORS[hash_val % len(AVATAR_COLORS)]

def _generate_initials(identifier: str, display_name: str | None = None) -> str:
    """Generate initials for an identifier.

    Args:
        identifier: Phone number or email
        display_name: Optional display name

    Returns:
        1-2 character initials string
    """
    if display_name:
        parts = display_name.split()
        if len(parts) >= 2:
            return f"{parts[0][0]}{parts[-1][0]}".upper()
        elif parts and parts[0]:
            return parts[0][0].upper()

    # For phone numbers, use last 2 digits
    if identifier.startswith("+") or identifier.isdigit():
        digits = "".join(c for c in identifier if c.isdigit())
        if len(digits) >= 2:
            return digits[-2:]
        elif digits:
            return digits[-1]

    # For emails, use first letter of local part
    if "@" in identifier:
        local_part = identifier.split("@")[0]
        if local_part:
            return local_part[0].upper()

    return "?"

def generate_svg_avatar(
    initials: str,
    background_color: str,
    size: int = 88,
) -> bytes:
    """Generate an SVG avatar with initials on a colored background.

    Args:
        initials: 1-2 character string to display
        background_color: Hex color for background (e.g., "#FF6B6B")
        size: Avatar size in pixels (default 88 for retina displays)

    Returns:
        SVG image as bytes
    """
    # Calculate font size based on number of characters
    font_size = size * 0.4 if len(initials) <= 2 else size * 0.35

    svg = (
        f'<svg xmlns="http://www.w3.org/2000/svg" width="{size}" height="{size}" '
        f'viewBox="0 0 {size} {size}">'
        f'<circle cx="{size // 2}" cy="{size // 2}" r="{size // 2}" '
        f'fill="{background_color}"/>'
        f'<text x="50%" y="50%" dominant-baseline="central" text-anchor="middle" '
        f"font-family=\"-apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif\" "
        f'font-size="{font_size}" fo

def generate_png_avatar(
    initials: str,
    background_color: str,
    size: int = 88,
) -> bytes:
    """Generate a PNG avatar with initials on a colored background.

    Falls back to SVG if PIL is not available.

    Args:
        initials: 1-2 character string to display
        background_color: Hex color for background (e.g., "#FF6B6B")
        size: Avatar size in pixels

    Returns:
        PNG image as bytes, or SVG if PIL unavailable
    """
    try:
        from PIL import Image, ImageDraw, ImageFont
    except ImportError:
        # Fall back to SVG if PIL not available
        logger.debug("PIL not available, returning SVG avatar")
        return generate_svg_avatar(initials, background_color, size)

    # Create circular avatar
    img = Image.new("RGBA", (size, size), (0, 0, 0, 0))
    draw = ImageDraw.Draw(img)

    # Parse hex color
    bg_color = tuple(int(background_color.lstrip("#")[i : i + 2], 16) for i in (0, 2, 4))

    # Draw filled circle
    draw.ellipse(

def _normalize_identifier(identifier: str) -> str:
    """Normalize a phone number or email identifier.

    Args:
        identifier: Raw phone number or email

    Returns:
        Normalized identifier string

    Raises:
        HTTPException 400: If phone number format is invalid
    """
    if "@" in identifier:
        return identifier.lower().strip()
    normalized_phone = normalize_phone_number(identifier)
    if normalized_phone is None:
        raise HTTPException(
            status_code=400,
            detail="Invalid phone number format",
        )
    return normalized_phone

def _fetch_contact_avatar(normalized: str) -> ContactAvatarData | None:
    """Fetch avatar data from the macOS Contacts database with timeout.

    Args:
        normalized: Normalized phone number or email

    Returns:
        ContactAvatarData or None if fetch failed/timed out
    """
    try:
        with ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(get_contact_avatar, normalized)
            return future.result(timeout=5.0)
    except TimeoutError:
        logger.warning(f"Avatar fetch timed out after 5s for {normalized}")
    except (OSError, RuntimeError, ValueError) as e:
        logger.warning(f"Error fetching contact avatar for {normalized}: {e}")
    return None

def _process_contact_image(image_bytes: bytes, size: int) -> tuple[bytes, str]:
    """Detect image format and resize if needed.

    Args:
        image_bytes: Raw image data from contacts
        size: Target size in pixels

    Returns:
        Tuple of (processed_bytes, content_type)
    """
    if image_bytes[:4] == b"\x89PNG":
        return image_bytes, "image/png"
    if image_bytes[:2] == b"\xff\xd8":
        return image_bytes, "image/jpeg"
    # Unknown format - try to process with PIL
    try:
        from PIL import Image

        img = Image.open(io.BytesIO(image_bytes))
        if img.size[0] != size or img.size[1] != size:
            img = img.resize((size, size), Image.Resampling.LANCZOS)  # type: ignore[assignment]
        output = io.BytesIO()
        img.save(output, format="PNG")
        return output.getvalue(), "image/png"
    except ImportError:
        return image_bytes, "image/jpeg"

def _generate_default_avatar(
    normalized: str,
    avatar_data: ContactAvatarData | None,
    size: int,
    format: str,
) -> tuple[bytes, str]:
    """Generate a default avatar with initials.

    Args:
        normalized: Normalized identifier for color selection
        avatar_data: Optional contact data for initials
        size: Avatar size in pixels
        format: Desired format ("png" or "svg")

    Returns:
        Tuple of (image_bytes, content_type)
    """
    initials = avatar_data.initials if avatar_data else _generate_initials(normalized, None)
    background_color = _get_color_for_identifier(normalized)

    if format == "svg":
        return generate_svg_avatar(initials, background_color, size), "image/svg+xml"

    image_bytes = generate_png_avatar(initials, background_color, size)
    # Check if we got SVG back (PIL not available)
    if image_bytes.startswith(b"<svg"):
        return image_bytes, "image/svg+xml"
    return image_bytes, "image/png"

def _avatar_response(image_bytes: bytes, content_type: str, source: str) -> Response:
    """Build an avatar HTTP response with caching headers.

    Args:
        image_bytes: Image data
        content_type: MIME type
        source: Avatar source label for X-Avatar-Source header

    Returns:
        FastAPI Response
    """
    return Response(
        content=image_bytes,
        media_type=content_type,
        headers={
            "Cache-Control": "public, max-age=3600",
            "X-Avatar-Source": source,
        },
    )

def get_avatar(
    identifier: str,
    size: int = Query(
        default=88,
        ge=16,
        le=512,
        description="Avatar size in pixels",
    ),
    format: str = Query(
        default="png",
        pattern="^(png|svg)$",
        description="Image format (png or svg)",
    ),
) -> Response:
    """Get avatar image for a contact.

    Returns the contact's photo from the macOS Contacts database if available.
    If no photo exists, generates a default avatar with the contact's initials
    on a colored background.

    The identifier can be:
    - Phone number (e.g., "+15551234567" or "5551234567")
    - Email address (e.g., "[email protected]")

    **Caching:**
    Avatar lookups are cached for 1 hour to minimize database queries.

    **Default Avatars:**
    When no contact photo exists, a default avatar is generated with:
    - Initials derived from contact name or identifier
    - Consistent background color based on identifier hash

    Args:
        identifie

def get_contact_info(identifier: str) -> dict[str, Any]:
    """Get basic contact information.

    Returns contact name and avatar metadata without the actual image data.
    Useful for efficiently loading contact info for multiple contacts.

    Args:
        identifier: Phone number or email address

    Returns:
        JSON object with contact information
    """
    # Normalize the identifier
    is_email = "@" in identifier
    normalized: str
    if is_email:
        normalized = identifier.lower().strip()
    else:
        normalized_phone = normalize_phone_number(identifier)
        if normalized_phone is None:
            raise HTTPException(
                status_code=400,
                detail="Invalid phone number format",
            )
        normalized = normalized_phone

    # Try to get contact info
    avatar_data: ContactAvatarData | None = None
    try:
        avatar_data = get_contact_avatar(normalized)
    except (OSError, RuntimeError, ValueError) as e:
    

async def list_conversations(
    request: Request,
    limit: int = Query(
        default=50,
        ge=1,
        le=500,
        description="Maximum number of conversations to return",
        examples=[50, 100],
    ),
    since: datetime | None = Query(
        default=None,
        description="Only return conversations with messages after this date (ISO 8601 format)",
        examples=["2024-01-01T00:00:00Z"],
    ),
    before: datetime | None = Query(
        default=None,
        description="Pagination cursor - only return conversations before this date",
        examples=["2024-01-15T10:30:00Z"],
    ),
    reader: ChatDBReader = Depends(get_imessage_reader),
) -> ConversationsListResponse:
    """List recent iMessage conversations.

    Uses TTL cache (30s) for repeated requests with same parameters.
    Returns conversations sorted by last message date (newest first).
    Supports pagination using the `before` parameter with the last conversation's
    `last_message_da

async def get_messages(
    request: Request,
    chat_id: str = FastPath(
        ...,
        min_length=1,
        max_length=255,
        description="The unique conversation identifier",
        examples=["chat123456789"],
    ),
    limit: int = Query(
        default=100,
        ge=1,
        le=1000,
        description="Maximum number of messages to return",
        examples=[100, 500],
    ),
    before: datetime | None = Query(
        default=None,
        description="Only return messages before this date (for pagination)",
        examples=["2024-01-15T10:30:00Z"],
    ),
    reader: ChatDBReader = Depends(get_imessage_reader),
) -> MessagesListResponse:
    """Get messages for a specific conversation.

    Returns messages sorted by date (newest first). Includes attachments,
    reactions (tapbacks), and threading information (reply_to_id).

    **Rate Limiting:**
    This endpoint is rate limited to 60 requests per minute.

    **Pagination:**
    Use the `before` para

async def search_messages(
    request: Request,
    q: str = Query(
        ...,
        min_length=1,
        max_length=1000,
        description="Search query - matches against message text (max 1000 characters)",
        examples=["dinner", "meeting tomorrow"],
    ),
    limit: int = Query(
        default=50,
        ge=1,
        le=500,
        description="Maximum number of results to return",
        examples=[50, 100],
    ),
    sender: str | None = Query(
        default=None,
        description="Filter by sender phone number or email",
        examples=["+15551234567", "[email protected]"],
    ),
    after: datetime | None = Query(
        default=None,
        description="Only messages after this date",
        examples=["2024-01-01T00:00:00Z"],
    ),
    before: datetime | None = Query(
        default=None,
        description="Only messages before this date",
        examples=["2024-01-15T23:59:59Z"],
    ),
    chat_id: str | None = Query(
        default=None,
 

async def send_message(
    request: Request,
    message_request: SendMessageRequest,
    chat_id: str = FastPath(
        ...,
        min_length=1,
        max_length=255,
        description="The unique conversation identifier",
        examples=["chat123456789"],
    ),
) -> SendMessageResponse:
    """Send a text message to a conversation.

    Sends an iMessage to either an individual contact or a group chat.
    Requires Automation permission for the Messages app, which will be
    prompted on first use.

    **Rate Limiting:**
    This endpoint is rate limited to 30 requests per minute.

    **For Individual Chats:**
    - Set `is_group` to `false` (default)
    - Provide the `recipient` phone number or email
    - The `chat_id` is used for reference only

    **For Group Chats:**
    - Set `is_group` to `true`
    - The `chat_id` is used to target the group
    - `recipient` is not required

    **Example Request (Individual):**
    ```json
    {
        "text": "Hey, are you

async def send_attachment(
    request: Request,
    attachment_request: SendAttachmentRequest,
    chat_id: str = FastPath(
        ...,
        min_length=1,
        max_length=255,
        description="The unique conversation identifier",
        examples=["chat123456789"],
    ),
) -> SendMessageResponse:
    """Send a file attachment to a conversation.

    Sends a file (image, video, document, etc.) via iMessage to either
    an individual contact or a group chat. Requires Automation permission
    for the Messages app.

    **Rate Limiting:**
    This endpoint is rate limited to 30 requests per minute.

    **Supported File Types:**
    - Images: jpg, png, gif, heic
    - Videos: mp4, mov
    - Documents: pdf, doc, docx
    - And more (any file the Messages app can handle)

    **For Individual Chats:**
    - Set `is_group` to `false` (default)
    - Provide the `recipient` phone number or email

    **For Group Chats:**
    - Set `is_group` to `true`
    - The `chat_id` is used

def _template_to_response(template: CustomTemplate) -> CustomTemplateResponse:
    """Convert a CustomTemplate to CustomTemplateResponse."""
    return CustomTemplateResponse(
        id=template.id,
        name=template.name,
        template_text=template.template_text,
        trigger_phrases=template.trigger_phrases,
        category=template.category,
        tags=template.tags,
        min_group_size=template.min_group_size,
        max_group_size=template.max_group_size,
        enabled=template.enabled,
        created_at=template.created_at,
        updated_at=template.updated_at,
        usage_count=template.usage_count,
    )

def list_templates(
    category: str | None = None,
    tag: str | None = None,
    enabled_only: bool = False,
) -> CustomTemplateListResponse:
    """List all custom templates with optional filtering.

    Query Parameters:
    - **category**: Filter by category
    - **tag**: Filter by tag
    - **enabled_only**: Only return enabled templates

    Returns:
        CustomTemplateListResponse with templates and metadata
    """
    store = get_custom_template_store()

    # Get templates with filters
    if category:
        templates = store.list_by_category(category)
    elif tag:
        templates = store.list_by_tag(tag)
    elif enabled_only:
        templates = store.list_enabled()
    else:
        templates = store.list_all()

    return CustomTemplateListResponse(
        templates=[_template_to_response(t) for t in templates],
        total=len(templates),
        categories=store.get_categories(),
        tags=store.get_tags(),
    )

def create_template(request: CustomTemplateCreateRequest) -> CustomTemplateResponse:
    """Create a new custom template.

    **Required Fields:**
    - name: Human-readable template name
    - template_text: The response to return when matched
    - trigger_phrases: At least one phrase to trigger this template

    **Optional Fields:**
    - category: Category for organization (default: "general")
    - tags: Additional tags for filtering
    - min_group_size/max_group_size: Group size constraints
    - enabled: Whether the template is active (default: True)

    Returns:
        The created CustomTemplateResponse with ID assigned
    """
    store = get_custom_template_store()

    # Validate trigger phrases
    if not request.trigger_phrases:
        raise HTTPException(
            status_code=400,
            detail="At least one trigger phrase is required",
        )

    # Validate group size constraints
    if (
        request.min_group_size is not None
        and request.ma

def get_template(template_id: str) -> CustomTemplateResponse:
    """Get a specific custom template by ID.

    Args:
        template_id: The template UUID

    Returns:
        The CustomTemplateResponse

    Raises:
        HTTPException 404: Template not found
    """
    store = get_custom_template_store()
    template = store.get(template_id)

    if template is None:
        raise HTTPException(
            status_code=404,
            detail=f"Template not found: {template_id}",
        )

    return _template_to_response(template)

def update_template(
    template_id: str, request: CustomTemplateUpdateRequest
) -> CustomTemplateResponse:
    """Update an existing custom template.

    Performs a partial update - only provided fields are changed.

    Args:
        template_id: The template UUID to update
        request: Fields to update

    Returns:
        The updated CustomTemplateResponse

    Raises:
        HTTPException 404: Template not found
        HTTPException 400: Invalid update data
    """
    store = get_custom_template_store()

    # Check if template exists
    existing = store.get(template_id)
    if existing is None:
        raise HTTPException(
            status_code=404,
            detail=f"Template not found: {template_id}",
        )

    # Build updates dict from provided fields
    updates: dict[str, str | list[str] | int | bool | None] = {}
    if request.name is not None:
        updates["name"] = request.name
    if request.template_text is not None:
        updates["template_text

def delete_template(template_id: str) -> dict[str, str]:
    """Delete a custom template.

    Args:
        template_id: The template UUID to delete

    Returns:
        Confirmation with deleted template ID

    Raises:
        HTTPException 404: Template not found
    """
    store = get_custom_template_store()

    if not store.delete(template_id):
        raise HTTPException(
            status_code=404,
            detail=f"Template not found: {template_id}",
        )

    logger.info("Deleted custom template: %s", template_id)
    return {"status": "deleted", "template_id": template_id}

def get_usage_stats() -> CustomTemplateUsageStats:
    """Get usage statistics for all custom templates.

    Returns:
        CustomTemplateUsageStats with counts and top templates
    """
    store = get_custom_template_store()
    stats = store.get_usage_stats()

    return CustomTemplateUsageStats(
        total_templates=stats["total_templates"],
        enabled_templates=stats["enabled_templates"],
        total_usage=stats["total_usage"],
        usage_by_category=stats["usage_by_category"],
        top_templates=stats["top_templates"],
    )

def test_template(request: CustomTemplateTestRequest) -> CustomTemplateTestResponse:
    """Test template trigger phrases against sample inputs.

    This endpoint helps validate that trigger phrases will match
    the intended inputs before saving a template.

    Args:
        request: Trigger phrases and test inputs

    Returns:
        CustomTemplateTestResponse with match results for each input
    """
    if not request.trigger_phrases:
        raise HTTPException(
            status_code=400,
            detail="At least one trigger phrase is required",
        )

    if not request.test_inputs:
        raise HTTPException(
            status_code=400,
            detail="At least one test input is required",
        )

    try:
        # Create a temporary template matcher with just the trigger phrases
        from models.templates import ResponseTemplate

        temp_template = ResponseTemplate(
            name="test_template",
            patterns=request.trigger_phrases,

def export_templates(
    request: CustomTemplateExportRequest | None = None,
) -> CustomTemplateExportResponse:
    """Export templates for sharing as a template pack.

    Args:
        request: Optional list of specific template IDs to export.
                 If null/empty, exports all templates.

    Returns:
        CustomTemplateExportResponse with template data for import
    """
    store = get_custom_template_store()

    template_ids = request.template_ids if request else None
    export_data = store.export_templates(template_ids)

    logger.info("Exported %d templates", export_data["template_count"])

    return CustomTemplateExportResponse(
        version=export_data["version"],
        export_date=export_data["export_date"],
        template_count=export_data["template_count"],
        templates=export_data["templates"],
    )

def import_templates(request: CustomTemplateImportRequest) -> CustomTemplateImportResponse:
    """Import templates from exported template pack.

    Args:
        request: Export data and import options

    Returns:
        CustomTemplateImportResponse with import results
    """
    if not request.data:
        raise HTTPException(
            status_code=400,
            detail="Import data is required",
        )

    if "templates" not in request.data:
        raise HTTPException(
            status_code=400,
            detail="Invalid import data: missing 'templates' field",
        )

    store = get_custom_template_store()
    result = store.import_templates(request.data, overwrite=request.overwrite)

    logger.info(
        "Imported %d templates (%d errors)",
        result["imported"],
        result["errors"],
    )

    return CustomTemplateImportResponse(
        imported=result["imported"],
        skipped=result["skipped"],
        errors=result["errors"],
        to

async def generate_draft_reply(
    draft_request: DraftReplyRequest,
    request: Request,
    reader: ChatDBReader = Depends(get_imessage_reader),
) -> DraftReplyResponse:
    """Generate AI-powered reply suggestions for a conversation.

    Uses the local MLX language model with conversation context to generate
    contextually appropriate reply suggestions. All processing is done locally
    without sending data to external services.

    **How It Works:**
    1. Retrieves recent messages from the conversation for context
    2. Formats the conversation as input for the language model
    3. Generates multiple reply suggestions with varying tones
    4. Returns suggestions ranked by confidence

    **Rate Limiting:**
    This endpoint is rate limited to 10 requests per minute to prevent
    resource exhaustion from CPU-intensive model generation.

    **Customizing Replies:**
    Use the `instruction` parameter to guide the tone or content:
    - "accept enthusiastically" - positiv