integration.py

from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Tuple, Union, cast

from fastapi import APIRouter, Header, HTTPException, Query
from fastapi import Request
from fastapi.responses import JSONResponse

import database.apps as apps_db
import database.conversations as conversations_db
import utils.apps as apps_utils
from utils.apps import verify_api_key
import database.redis_db as redis_db
from database._client import db as firestore_db
from utils.memory.memory_service import MemoryService, truncate_locked_memory_preview
from database.redis_db import get_enabled_apps, r as redis_client
import database.action_items as action_items_db
import models.integrations as integration_models
import models.conversation as conversation_models
from models.shared import EmptyResponse
from models.conversation import SearchRequest
from models.app import App
from models.geolocation import Geolocation
from utils.app_integrations import (
    send_app_notification,
    trigger_external_integrations,
)
from utils.conversations.location import get_google_maps_location
from utils.conversations.render import redact_conversation_for_integration
from utils.conversations.memories import process_external_integration_memory
from utils.conversations.process_conversation import process_conversation
from utils.conversations.search import search_conversations
from utils.other.endpoints import check_rate_limit_inline
from utils.executors import run_blocking, db_executor, postprocess_executor, critical_executor
import logging

logger = logging.getLogger(__name__)

# Rate limit settings - more conservative limits to prevent notification fatigue
RATE_LIMIT_PERIOD = 3600  # 1 hour in seconds
MAX_NOTIFICATIONS_PER_HOUR = 10  # Maximum notifications per hour per app per user

# Firestore 'in' filters accept at most 30 values (see database/apps.py, database/chat.py); keep
# well under that so a caller-supplied statuses list can never blow the query up into a 500.
MAX_STATUSES_FILTER_VALUES = 20

router = APIRouter()


def check_rate_limit(app_id: str, user_id: str) -> Tuple[bool, int, int, int]:
    """
    Check if the app has exceeded its rate limit for a specific user
    Returns: (allowed, remaining, reset_time, retry_after)
    """
    now = datetime.now(timezone.utc)
    hour_key = f"notification_rate_limit:{app_id}:{user_id}:{now.strftime('%Y-%m-%d-%H')}"

    # Check hourly limit
    hour_count = redis_client.get(hour_key)
    if hour_count is None:
        # Seed to 0, not 1: the unconditional incr below is the single source of truth for the count.
        # Seeding to 1 AND incrementing made the first request consume two tokens, so only 9 of
        # MAX_NOTIFICATIONS_PER_HOUR were ever allowed and the remaining header was off by one.
        redis_client.setex(hour_key, RATE_LIMIT_PERIOD, 0)
        hour_count = 0
    else:
        hour_count = int(hour_count)

    # Calculate reset time
    hour_reset = RATE_LIMIT_PERIOD - (int(now.timestamp()) % RATE_LIMIT_PERIOD)
    reset_time = hour_reset

    # Check if hourly limit is exceeded
    if hour_count >= MAX_NOTIFICATIONS_PER_HOUR:
        return False, MAX_NOTIFICATIONS_PER_HOUR - hour_count, hour_reset, hour_reset

    # Increment counter
    redis_client.incr(hour_key)

    remaining = MAX_NOTIFICATIONS_PER_HOUR - hour_count - 1

    return True, remaining, reset_time, 0


async def _resolve_geolocation(geolocation: Optional[Geolocation]) -> Optional[Geolocation]:
    """Enrich a raw geolocation with Google Places, keeping the original coordinates when the lookup
    misses (returns None) so a geocode miss does not drop the location. Only enriches a geolocation that
    has coordinates but no google_place_id yet."""
    if geolocation and not geolocation.google_place_id:
        enriched = await run_blocking(
            db_executor, get_google_maps_location, geolocation.latitude, geolocation.longitude
        )
        if enriched:
            return enriched.model_copy(
                update={
                    'latitude': geolocation.latitude,
                    'longitude': geolocation.longitude,
                    'captured_at': geolocation.captured_at,
                    'capture_source': geolocation.capture_source,
                    'accuracy': geolocation.accuracy,
                    'altitude': geolocation.altitude,
                }
            )
    return geolocation


@router.post(
    '/v2/integrations/{app_id}/user/conversations',
    response_model=EmptyResponse,
    tags=['integration', 'conversations'],
)
async def create_conversation_via_integration(
    request: Request,
    app_id: str,
    create_conversation: conversation_models.ExternalIntegrationCreateConversation,
    uid: str,
    authorization: Optional[str] = Header(None),
) -> Dict[str, Any]:
    # Verify API key from Authorization header
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header. Must be 'Bearer API_KEY'")

    api_key = authorization.replace('Bearer ', '')
    if not await run_blocking(critical_executor, verify_api_key, app_id, api_key):
        raise HTTPException(status_code=403, detail="Invalid integration API key")

    # Rate limit per app+user
    await run_blocking(critical_executor, check_rate_limit_inline, f"{app_id}:{uid}", "integration:conversations")

    # Verify if the app exists
    app = await run_blocking(db_executor, apps_db.get_app_by_id_db, app_id)
    if not app:
        raise HTTPException(status_code=404, detail="App not found")

    # Verify if the uid has enabled the app
    enabled_plugins = await run_blocking(db_executor, redis_db.get_enabled_apps, uid)
    if app_id not in enabled_plugins:
        raise HTTPException(status_code=403, detail="App is not enabled for this user")

    # Check if the app has the capability external_integration > action > create_conversation
    if not apps_utils.app_can_create_conversation(app):
        raise HTTPException(status_code=403, detail="App does not have the capability to create conversations")

    # Time
    started_at = (
        create_conversation.started_at if create_conversation.started_at is not None else datetime.now(timezone.utc)
    )
    finished_at = (
        create_conversation.finished_at
        if create_conversation.finished_at is not None
        else started_at + timedelta(seconds=300)
    )  # 5 minutes
    create_conversation.started_at = started_at
    create_conversation.finished_at = finished_at

    # Geo: enrich raw coordinates with a Google Places lookup. Previously the enriched result was
    # computed and then unconditionally overwritten with the original value, discarding it on every call.
    create_conversation.geolocation = await _resolve_geolocation(create_conversation.geolocation)

    # Language
    language_code = create_conversation.language
    if not language_code:
        language_code = 'en'  # Default to English
        create_conversation.language = language_code

    # Set source to external_integration
    create_conversation.source = conversation_models.ConversationSource.external_integration

    # Set app_id
    create_conversation.app_id = app_id

    # Process
    conversation = await run_blocking(
        postprocess_executor, process_conversation, uid, language_code, create_conversation
    )

    # Always trigger integration
    await trigger_external_integrations(uid, conversation)

    # TODO: Empty for now, replace with ConversationCreateResponse once we don't have to wait for process_conversation
    # to finish for the conversation id
    return {}


@router.post(
    '/v2/integrations/{app_id}/user/memories',
    response_model=EmptyResponse,
    tags=['integration', 'memories'],
)
def create_memories_via_integration(
    request: Request,
    app_id: str,
    fact_data: integration_models.ExternalIntegrationCreateMemory,
    uid: str,
    authorization: Optional[str] = Header(None),
) -> Dict[str, Any]:
    # Verify API key from Authorization header
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header. Must be 'Bearer API_KEY'")

    api_key = authorization.replace('Bearer ', '')
    if not verify_api_key(app_id, api_key):
        raise HTTPException(status_code=403, detail="Invalid integrationAPI key")

    # Rate limit per app+user
    check_rate_limit_inline(f"{app_id}:{uid}", "integration:memories")

    # Verify if the app exists
    app = apps_db.get_app_by_id_db(app_id)
    if not app:
        raise HTTPException(status_code=404, detail="App not found")

    # Verify if the uid has enabled the app
    enabled_plugins = redis_db.get_enabled_apps(uid)
    if app_id not in enabled_plugins:
        raise HTTPException(status_code=403, detail="App is not enabled for this user")

    # Check if the app has the capability external_integration > action > create_memories / create_facts
    if not apps_utils.app_can_create_memories(app):
        raise HTTPException(status_code=403, detail="App does not have the capability to create memories")

    # Validate that text is provided or explicit facts are provided
    if (not fact_data.text or len(fact_data.text.strip()) == 0) and (
        not fact_data.memories or len(fact_data.memories) == 0
    ):
        raise HTTPException(
            status_code=422, detail="Either text or explicit memories(facts) are required and cannot be empty"
        )

    # Process and save the memory using the utility function
    process_external_integration_memory(uid, fact_data, app_id)

    # Empty response
    return {}


@router.get(
    '/v2/integrations/{app_id}/memories',
    response_model=integration_models.MemoriesResponse,
    response_model_exclude_none=True,
    tags=['integration', 'memories'],
)
def get_memories_via_integration(
    request: Request,
    app_id: str,
    uid: str,
    limit: int = Query(100, ge=1, le=1000),
    offset: int = Query(0, ge=0),
    authorization: Optional[str] = Header(None),
) -> Dict[str, Any]:
    """
    Get all memories (facts) for a user via integration API.
    Authentication is required via API key in the Authorization header.
    """
    # Verify API key from Authorization header
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header. Must be 'Bearer API_KEY'")

    api_key = authorization.replace('Bearer ', '')
    if not verify_api_key(app_id, api_key):
        raise HTTPException(status_code=403, detail="Invalid integrationAPI key")

    # Verify if the app exists
    app = apps_db.get_app_by_id_db(app_id)
    if not app:
        raise HTTPException(status_code=404, detail="App not found")

    # Verify if the uid has enabled the app
    enabled_plugins = redis_db.get_enabled_apps(uid)
    if app_id not in enabled_plugins:
        raise HTTPException(status_code=403, detail="App is not enabled for this user")

    # Check if the app has the capability to read memories
    if not apps_utils.app_can_read_memories(app):
        raise HTTPException(status_code=403, detail="App does not have the capability to read memories")

    memories = MemoryService(db_client=firestore_db).read(uid, limit=limit, offset=offset)
    memory_items: List[integration_models.MemoryItem] = []
    for memory in memories:
        try:
            # Keep the released integration privacy contract: a locked memory
            # may be listed, but only with its bounded preview.  MemoryService
            # is the authority for both physical origins, so apply the same
            # exposure rule after the universal read rather than trusting the
            # route's former legacy-only branch.
            exposed = truncate_locked_memory_preview(memory)
            memory_items.append(integration_models.MemoryItem(**exposed.model_dump(mode='json')))
        except Exception as e:  # noqa: BLE001 - intentional broad catch: skip any malformed record
            logger.error(f"Error parsing memory {getattr(memory, 'id', None)}: {str(e)}")
            continue

    return {"memories": memory_items}


@router.get(
    '/v2/integrations/{app_id}/conversations',
    response_model=integration_models.ConversationsResponse,
    response_model_exclude_none=True,
    tags=['integration', 'conversations'],
)
def get_conversations_via_integration(
    request: Request,
    app_id: str,
    uid: str,
    limit: int = Query(100, ge=1, le=1000),
    offset: int = Query(0, ge=0),
    include_discarded: bool = Query(False),
    statuses: List[str] = Query([]),
    start_date: Optional[Union[datetime, str]] = Query(
        None, description="Filter conversations after this date (ISO format)"
    ),
    end_date: Optional[Union[datetime, str]] = Query(
        None, description="Filter conversations before this date (ISO format)"
    ),
    max_transcript_segments: int = Query(
        100,
        ge=-1,
        le=1000,
        description="Maximum number of transcript segments to include per conversation. Use -1 for no limit.",
    ),
    authorization: Optional[str] = Header(None),
) -> Dict[str, Any]:
    """
    Get all conversations for a user via integration API.
    Authentication is required via API key in the Authorization header.

    Optional date range filtering:
    - start_date: Filter conversations after this date (ISO format)
    - end_date: Filter conversations before this date (ISO format)
    """
    # Verify API key from Authorization header
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header. Must be 'Bearer API_KEY'")

    api_key = authorization.replace('Bearer ', '')
    if not verify_api_key(app_id, api_key):
        raise HTTPException(status_code=403, detail="Invalid API key")

    # Verify if the app exists
    app = apps_db.get_app_by_id_db(app_id)
    if not app:
        raise HTTPException(status_code=404, detail="App not found")

    # Verify if the uid has enabled the app
    enabled_plugins = redis_db.get_enabled_apps(uid)
    if app_id not in enabled_plugins:
        raise HTTPException(status_code=403, detail="App is not enabled for this user")

    # Check if the app has the capability to read conversations
    if not apps_utils.app_can_read_conversations(app):
        raise HTTPException(status_code=403, detail="App does not have the capability to read conversations")

    if len(statuses) > MAX_STATUSES_FILTER_VALUES:
        raise HTTPException(status_code=400, detail=f"statuses accepts at most {MAX_STATUSES_FILTER_VALUES} values")

    # Convert string dates to datetime objects if needed
    if isinstance(start_date, str) and start_date:
        try:
            if len(start_date) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(start_date, '%Y-%m-%d')
                start_date = dt.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc)
            else:
                start_date = datetime.fromisoformat(start_date.replace('Z', '+00:00'))
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid start_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )

    if isinstance(end_date, str) and end_date:
        try:
            if len(end_date) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(end_date, '%Y-%m-%d')
                end_date = dt.replace(hour=23, minute=59, second=59, microsecond=999999, tzinfo=timezone.utc)
            else:
                end_date = datetime.fromisoformat(end_date.replace('Z', '+00:00'))
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid end_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )
    conversations_data = conversations_db.get_conversations(
        uid,
        limit=limit,
        offset=offset,
        include_discarded=include_discarded,
        statuses=statuses,
        start_date=cast(Optional[datetime], start_date),
        end_date=cast(Optional[datetime], end_date),
    )

    # Convert database conversations
    conversation_items: List[integration_models.ConversationItem] = []
    for conv in conversations_data:
        try:
            redact_conversation_for_integration(conv)
            item = integration_models.ConversationItem.model_validate(conv)

            # Limit transcript segments
            if (
                max_transcript_segments != -1
                and item.transcript_segments
                and len(item.transcript_segments) > max_transcript_segments
            ):
                item.transcript_segments = item.transcript_segments[:max_transcript_segments]

            # Convert to dict with exclude_none=True to remove null values
            conversation_items.append(item)
        except Exception as e:
            logger.error(f"Error parsing conversation {conv.get('id')}: {str(e)}")
            continue

    # Create response with exclude_none=True
    response = integration_models.ConversationsResponse(conversations=conversation_items)
    return response.model_dump(exclude_none=True)


@router.post(
    '/v2/integrations/{app_id}/search/conversations',
    response_model=integration_models.SearchConversationsResponse,
    response_model_exclude_none=True,
    tags=['integration', 'conversations'],
)
def search_conversations_via_integration(
    request: Request,
    app_id: str,
    uid: str,
    search_request: SearchRequest,
    max_transcript_segments: int = Query(
        100,
        ge=-1,
        le=1000,
        description="Maximum number of transcript segments to include per conversation. Use -1 for no limit.",
    ),
    authorization: Optional[str] = Header(None),
) -> Dict[str, Any]:
    """
    Search conversations for a user via integration API.
    Authentication is required via API key in the Authorization header.
    """
    # Verify API key from Authorization header
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header. Must be 'Bearer API_KEY'")

    api_key = authorization.replace('Bearer ', '')
    if not verify_api_key(app_id, api_key):
        raise HTTPException(status_code=403, detail="Invalid API key")

    # Verify if the app exists
    app = apps_db.get_app_by_id_db(app_id)
    if not app:
        raise HTTPException(status_code=404, detail="App not found")

    # Verify if the uid has enabled the app
    enabled_plugins = redis_db.get_enabled_apps(uid)
    if app_id not in enabled_plugins:
        raise HTTPException(status_code=403, detail="App is not enabled for this user")

    # Check if the app has the capability to read conversations
    if not apps_utils.app_can_read_conversations(app):
        raise HTTPException(status_code=403, detail="App does not have the capability to read conversations")

    # Convert ISO datetime strings to Unix timestamps if provided
    start_timestamp = None
    end_timestamp = None

    if search_request.start_date:
        try:
            start_date_str = search_request.start_date
            if len(start_date_str) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(start_date_str, '%Y-%m-%d')
                start_dt = dt.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc)
            else:
                start_dt = datetime.fromisoformat(start_date_str.replace('Z', '+00:00'))
            start_timestamp = int(start_dt.timestamp())
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid start_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )

    if search_request.end_date:
        try:
            end_date_str = search_request.end_date
            if len(end_date_str) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(end_date_str, '%Y-%m-%d')
                end_dt = dt.replace(hour=23, minute=59, second=59, microsecond=999999, tzinfo=timezone.utc)
            else:
                end_dt = datetime.fromisoformat(end_date_str.replace('Z', '+00:00'))
            end_timestamp = int(end_dt.timestamp())
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid end_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )

    # Search conversations
    search_results = search_conversations(
        query=search_request.query,
        page=cast(int, search_request.page),
        per_page=cast(int, search_request.per_page),
        uid=uid,
        include_discarded=cast(bool, search_request.include_discarded),
        start_date=cast(int, start_timestamp),
        end_date=cast(int, end_timestamp),
    )

    # Extract conversation IDs from search results
    conversation_ids = [conv.get('id') for conv in search_results['items']]

    # Get full conversation data using the IDs
    full_conversations = []
    if conversation_ids:
        # Hydration must honour the same include_discarded the search above ran with, or a
        # discarded match is dropped here after the search window already moved past it.
        full_conversations = conversations_db.get_conversations_by_id(
            uid,
            conversation_ids,
            include_discarded=cast(bool, search_request.include_discarded),
        )

    # Convert database conversations to integration model
    conversation_items: List[integration_models.ConversationItem] = []
    for conv in full_conversations:
        try:
            redact_conversation_for_integration(conv)
            item = integration_models.ConversationItem.model_validate(conv)

            # Limit transcript segments
            if (
                max_transcript_segments != -1
                and item.transcript_segments
                and len(item.transcript_segments) > max_transcript_segments
            ):
                item.transcript_segments = item.transcript_segments[:max_transcript_segments]

            conversation_items.append(item)
        except Exception as e:
            logger.error(f"Error parsing conversation {conv.get('id')}: {str(e)}")
            continue

    # Create response with pagination info
    response = integration_models.SearchConversationsResponse(
        conversations=conversation_items,
        total_pages=search_results['total_pages'],
        current_page=search_results['current_page'],
        per_page=search_results['per_page'],
    )

    return response.model_dump(exclude_none=True)


@router.post(
    '/v2/integrations/{app_id}/notification',
    response_model=integration_models.IntegrationNotificationResponse,
    tags=['integration', 'notifications'],
)
def send_notification_via_integration(
    request: Request, app_id: str, message: str, uid: str, authorization: Optional[str] = Header(None)
) -> JSONResponse:
    # Verify API key from Authorization header
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header. Must be 'Bearer API_KEY'")

    api_key = authorization.replace('Bearer ', '')
    if not verify_api_key(app_id, api_key):
        raise HTTPException(status_code=403, detail="Invalid API key")

    # Verify if the app exists
    app_data = cast(Optional[Dict[str, Any]], apps_utils.get_available_app_by_id(app_id, uid))  # type: ignore[reportUnknownMemberType]  # utils.apps.get_available_app_by_id returns bare dict
    if not app_data:
        raise HTTPException(status_code=404, detail='App not found')

    app = App(**app_data)

    # Check if user has app installed
    user_enabled = set(get_enabled_apps(uid))
    if app_id not in user_enabled:
        raise HTTPException(status_code=403, detail='User does not have this app installed')

    # Check rate limit
    allowed, remaining, reset_time, retry_after = check_rate_limit(app.id, uid)

    # Add rate limit headers to response
    headers = {
        'X-RateLimit-Limit': str(MAX_NOTIFICATIONS_PER_HOUR),
        'X-RateLimit-Remaining': str(remaining),
        'X-RateLimit-Reset': str(reset_time),
    }

    if not allowed:
        headers['Retry-After'] = str(retry_after)
        return JSONResponse(
            status_code=429,
            headers=headers,
            content={'detail': f'Rate limit exceeded. Maximum {MAX_NOTIFICATIONS_PER_HOUR} notifications per hour.'},
        )

    send_app_notification(uid, app.name, app.id, message)
    return JSONResponse(status_code=200, headers=headers, content={'status': 'Ok'})


@router.get(
    '/v2/integrations/{app_id}/tasks',
    response_model=integration_models.TasksResponse,
    response_model_exclude_none=True,
    tags=['integration', 'tasks'],
)
def get_tasks_via_integration(
    request: Request,
    app_id: str,
    uid: str,
    limit: int = Query(100, ge=1, le=1000),
    offset: int = Query(0, ge=0),
    completed: Optional[bool] = Query(None, description="Filter by completion status"),
    conversation_id: Optional[str] = Query(None, description="Filter by conversation ID"),
    start_date: Optional[Union[datetime, str]] = Query(
        None, description="Filter by creation start date (ISO format or YYYY-MM-DD)"
    ),
    end_date: Optional[Union[datetime, str]] = Query(
        None, description="Filter by creation end date (ISO format or YYYY-MM-DD)"
    ),
    due_start_date: Optional[Union[datetime, str]] = Query(
        None, description="Filter by due start date (ISO format or YYYY-MM-DD)"
    ),
    due_end_date: Optional[Union[datetime, str]] = Query(
        None, description="Filter by due end date (ISO format or YYYY-MM-DD)"
    ),
    authorization: Optional[str] = Header(None),
) -> Dict[str, Any]:
    """
    Get all tasks (action items) for a user via integration API.
    Authentication is required via API key in the Authorization header.

    Optional filters:
    - **completed**: Filter by completion status (true/false/null for all)
    - **conversation_id**: Filter by conversation ID
    - **start_date**: Filter by creation start date (ISO format or YYYY-MM-DD)
    - **end_date**: Filter by creation end date (ISO format or YYYY-MM-DD)
    - **due_start_date**:  Filter by due start date (ISO format or YYYY-MM-DD)
    - **due_end_date**: Filter by due end date (ISO format or YYYY-MM-DD)
    """
    if not authorization or not authorization.startswith('Bearer '):
        raise HTTPException(status_code=401, detail="Missing or invalid Authorization header. Must be 'Bearer API_KEY'")

    api_key = authorization.replace('Bearer ', '')
    if not verify_api_key(app_id, api_key):
        raise HTTPException(status_code=403, detail="Invalid API key")

    app = apps_db.get_app_by_id_db(app_id)
    if not app:
        raise HTTPException(status_code=404, detail="App not found")

    enabled_plugins = redis_db.get_enabled_apps(uid)
    if app_id not in enabled_plugins:
        raise HTTPException(status_code=403, detail="App is not enabled for this user")

    if not apps_utils.app_can_read_tasks(app):
        raise HTTPException(status_code=403, detail="App does not have the capability to read tasks")

    if isinstance(start_date, str) and start_date:
        try:
            if len(start_date) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(start_date, '%Y-%m-%d')
                start_date = dt.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc)
            else:
                start_date = datetime.fromisoformat(start_date.replace('Z', '+00:00'))
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid start_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )

    if isinstance(end_date, str) and end_date:
        try:
            if len(end_date) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(end_date, '%Y-%m-%d')
                end_date = dt.replace(hour=23, minute=59, second=59, microsecond=999999, tzinfo=timezone.utc)
            else:
                end_date = datetime.fromisoformat(end_date.replace('Z', '+00:00'))
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid end_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )

    if isinstance(due_start_date, str) and due_start_date:
        try:
            if len(due_start_date) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(due_start_date, '%Y-%m-%d')
                due_start_date = dt.replace(hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc)
            else:
                due_start_date = datetime.fromisoformat(due_start_date.replace('Z', '+00:00'))
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid due_start_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )

    if isinstance(due_end_date, str) and due_end_date:
        try:
            if len(due_end_date) == 10:  # YYYY-MM-DD
                dt = datetime.strptime(due_end_date, '%Y-%m-%d')
                due_end_date = dt.replace(hour=23, minute=59, second=59, microsecond=999999, tzinfo=timezone.utc)
            else:
                due_end_date = datetime.fromisoformat(due_end_date.replace('Z', '+00:00'))
        except ValueError:
            raise HTTPException(
                status_code=400,
                detail="Invalid due_end_date format. Use ISO format (YYYY-MM-DDTHH:MM:SS.sssZ) or YYYY-MM-DD",
            )
    tasks = action_items_db.get_action_items(
        uid=uid,
        conversation_id=conversation_id,
        completed=completed,
        start_date=cast(Optional[datetime], start_date),
        end_date=cast(Optional[datetime], end_date),
        due_start_date=cast(Optional[datetime], due_start_date),
        due_end_date=cast(Optional[datetime], due_end_date),
        limit=limit,
        offset=offset,
    )

    task_items: List[integration_models.TaskItem] = []
    for task in tasks:
        task_data = task.copy()
        if task_data.get('is_locked', False):
            description = task_data.get('description', '')
            task_data['description'] = (description[:70] + '...') if len(description) > 70 else description
        try:
            task_items.append(integration_models.TaskItem(**task_data))
        except Exception as e:  # noqa: BLE001 - intentional broad catch: skip any malformed record
            # One malformed/legacy record must not 500 the whole page; skip it (mirrors the
            # conversation conversion guard in get_conversations_via_integration).
            logger.error(f"Error parsing task {task_data.get('id')}: {str(e)}")
            continue

    response = integration_models.TasksResponse(tasks=task_items)
    return response.model_dump(exclude_none=True)