feat: add ChatKit migration with SSE streaming

This commit implements the ChatKit migration for the Hugging Face
Spaces deployment, replacing WebSocket with Server-Sent Events (SSE).

New Features:
- ChatKit server (chatkit_server.py) with SSE streaming
- PostgreSQL ChatKit store (services/chatkit_store.py)
- Thread model for ChatKit conversations (models/thread.py)
- MCP tool wrappers for Agents SDK (ai_agent/tool_wrappers.py)
- Database migration for threads table (migrations/migrate_threads.sql)

Backend Changes:
- Updated /api/chatkit endpoint with SSE streaming
- Fixed SSE JSON serialization using json.dumps()
- Added tool_calls JSONB column support
- Enhanced authentication with httpOnly cookie support

Updated Files:
- api/chat.py: Added ChatKit SSE endpoint
- core/config.py: Added get_gemini_client() function
- core/security.py: Enhanced cookie-based authentication
- ai_agent/__init__.py: Updated agent imports
- .env.example: Added GEMINI_BASE_URL configuration

This enables the same ChatKit functionality on Hugging Face Spaces
as in the main SDDRI-Hackathon-2 repository.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

.env.example

@@ -14,3 +14,5 @@ ENVIRONMENT=development
 # Get your API key from https://aistudio.google.com
 GEMINI_API_KEY=your-gemini-api-key-here
 GEMINI_MODEL=gemini-2.0-flash-exp
+# Gemini OpenAI-compatible endpoint for ChatKit integration
+GEMINI_BASE_URL=https://generativelanguage.googleapis.com/v1beta/openai/

ai_agent/__init__.py

@@ -2,26 +2,22 @@
 
 [Task]: T014, T072
 [From]: specs/004-ai-chatbot/tasks.md
+[From]: T045 - Delete agent_streaming.py (ChatKit migration replaces WebSocket streaming)
 
 This module provides the AI agent that powers the chatbot functionality.
 It uses OpenAI SDK with function calling and Gemini via AsyncOpenAI adapter.
 
-Includes streaming support for real-time WebSocket progress events.
+NOTE: The streaming agent functionality has been migrated to ChatKit SSE endpoint.
+See backend/chatkit_server.py for the new ChatKit-based implementation.
 """
 from ai_agent.agent_simple import (
     get_gemini_client,
     run_agent,
     is_gemini_configured
 )
-from ai_agent.agent_streaming import (
-    run_agent_with_streaming,
-    execute_tool_with_progress,
-)
 
 __all__ = [
     "get_gemini_client",
     "run_agent",
-    "run_agent_with_streaming",
-    "execute_tool_with_progress",
     "is_gemini_configured"
 ]

ai_agent/tool_wrappers.py

@@ -0,0 +1,366 @@
+"""Agents SDK function wrappers for MCP task management tools.
+
+[Task]: T012-T018
+[From]: specs/010-chatkit-migration/tasks.md - Phase 3 Backend Implementation
+[From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+
+This module wraps existing MCP tools as Agents SDK functions using the
+@function_tool decorator. Each wrapper calls the underlying MCP tool function
+and returns the result in a format compatible with the Agents SDK.
+
+Tools wrapped:
+1. create_task (T012) - from mcp_server/tools/add_task.py
+2. list_tasks (T013) - from mcp_server/tools/list_tasks.py
+3. update_task (T014) - from mcp_server/tools/update_task.py
+4. delete_task (T015) - from mcp_server/tools/delete_task.py
+5. complete_task (T016) - from mcp_server/tools/complete_task.py
+6. complete_all_tasks (T017) - from mcp_server/tools/complete_all_tasks.py
+7. delete_all_tasks (T018) - from mcp_server/tools/delete_all_tasks.py
+
+[From]: specs/010-chatkit-migration/research.md - Section 7 (Tool Visualization Support)
+"""
+import json
+import logging
+from typing import Optional
+
+from agents import function_tool, RunContextWrapper
+
+# Import MCP tools
+# Note: We import the actual async functions from MCP tools
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.dirname(__file__)))
+
+from mcp_server.tools.add_task import add_task as mcp_add_task
+from mcp_server.tools.list_tasks import list_tasks as mcp_list_tasks
+from mcp_server.tools.update_task import update_task as mcp_update_task
+from mcp_server.tools.delete_task import delete_task as mcp_delete_task
+from mcp_server.tools.complete_task import complete_task as mcp_complete_task
+from mcp_server.tools.complete_all_tasks import complete_all_tasks as mcp_complete_all_tasks
+from mcp_server.tools.delete_all_tasks import delete_all_tasks as mcp_delete_all_tasks
+
+logger = logging.getLogger(__name__)
+
+
+# =============================================================================
+# Agents SDK Function Wrappers
+# =============================================================================
+
+@function_tool(
+    name="create_task",
+    description_override="Create a new task in the user's todo list. Use this when the user wants to create, add, or remind themselves about a task. Parameters: title (required), description (optional), due_date (optional, ISO 8601 or relative), priority (optional: low/medium/high), tags (optional list)."
+)
+async def create_task_tool(
+    ctx: RunContextWrapper,
+    title: str,
+    description: Optional[str] = None,
+    due_date: Optional[str] = None,
+    priority: Optional[str] = None,
+    tags: Optional[list[str]] = None,
+) -> str:
+    """Create a task via MCP tool.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+    [Task]: T012
+
+    Args:
+        ctx: Agents SDK run context containing user_id
+        title: Task title
+        description: Optional task description
+        due_date: Optional due date
+        priority: Optional priority level
+        tags: Optional list of tags
+
+    Returns:
+        JSON string with created task details
+    """
+    user_id = ctx.context.user_id
+
+    try:
+        result = await mcp_add_task(
+            user_id=user_id,
+            title=title,
+            description=description,
+            due_date=due_date,
+            priority=priority,
+            tags=tags or []
+        )
+        logger.info(f"Task created: {result['task']['id']} for user {user_id}")
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Failed to create task for user {user_id}: {e}")
+        return json.dumps({"success": False, "error": str(e)})
+
+
+@function_tool(
+    name="list_tasks",
+    description_override="List all tasks for the user, optionally filtered by completion status, priority, tag, or due date range. Returns a list of tasks with their details."
+)
+async def list_tasks_tool(
+    ctx: RunContextWrapper,
+    completed: Optional[bool] = None,
+    priority: Optional[str] = None,
+    tag: Optional[str] = None,
+    due_before: Optional[str] = None,
+    due_after: Optional[str] = None,
+) -> str:
+    """List tasks via MCP tool.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+    [Task]: T013
+
+    Args:
+        ctx: Agents SDK run context containing user_id
+        completed: Optional filter by completion status
+        priority: Optional filter by priority
+        tag: Optional filter by tag
+        due_before: Optional due date upper bound
+        due_after: Optional due date lower bound
+
+    Returns:
+        JSON string with list of tasks
+    """
+    user_id = ctx.context.user_id
+
+    try:
+        result = await mcp_list_tasks(
+            user_id=user_id,
+            completed=completed,
+            priority=priority,
+            tag=tag,
+            due_before=due_before,
+            due_after=due_after
+        )
+        task_count = len(result.get("tasks", []))
+        logger.info(f"Listed {task_count} tasks for user {user_id}")
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Failed to list tasks for user {user_id}: {e}")
+        return json.dumps({"success": False, "error": str(e), "tasks": []})
+
+
+@function_tool(
+    name="update_task",
+    description_override="Update an existing task. Parameters: task_id (required), title (optional), description (optional), due_date (optional), priority (optional), tags (optional)."
+)
+async def update_task_tool(
+    ctx: RunContextWrapper,
+    task_id: str,
+    title: Optional[str] = None,
+    description: Optional[str] = None,
+    due_date: Optional[str] = None,
+    priority: Optional[str] = None,
+    tags: Optional[list[str]] = None,
+) -> str:
+    """Update a task via MCP tool.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+    [Task]: T014
+
+    Args:
+        ctx: Agents SDK run context containing user_id
+        task_id: Task ID to update
+        title: Optional new title
+        description: Optional new description
+        due_date: Optional new due date
+        priority: Optional new priority
+        tags: Optional new tag list
+
+    Returns:
+        JSON string with updated task details
+    """
+    user_id = ctx.context.user_id
+
+    try:
+        result = await mcp_update_task(
+            user_id=user_id,
+            task_id=task_id,
+            title=title,
+            description=description,
+            due_date=due_date,
+            priority=priority,
+            tags=tags
+        )
+        logger.info(f"Task updated: {task_id} for user {user_id}")
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Failed to update task {task_id} for user {user_id}: {e}")
+        return json.dumps({"success": False, "error": str(e)})
+
+
+@function_tool(
+    name="delete_task",
+    description_override="Delete a task permanently. Parameters: task_id (required)."
+)
+async def delete_task_tool(
+    ctx: RunContextWrapper,
+    task_id: str,
+) -> str:
+    """Delete a task via MCP tool.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+    [Task]: T015
+
+    Args:
+        ctx: Agents SDK run context containing user_id
+        task_id: Task ID to delete
+
+    Returns:
+        JSON string with deletion confirmation
+    """
+    user_id = ctx.context.user_id
+
+    try:
+        result = await mcp_delete_task(
+            user_id=user_id,
+            task_id=task_id
+        )
+        logger.info(f"Task deleted: {task_id} for user {user_id}")
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Failed to delete task {task_id} for user {user_id}: {e}")
+        return json.dumps({"success": False, "error": str(e)})
+
+
+@function_tool(
+    name="complete_task",
+    description_override="Mark a task as completed or incomplete. Parameters: task_id (required), completed (boolean, required)."
+)
+async def complete_task_tool(
+    ctx: RunContextWrapper,
+    task_id: str,
+    completed: bool,
+) -> str:
+    """Complete/uncomplete a task via MCP tool.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+    [Task]: T016
+
+    Args:
+        ctx: Agents SDK run context containing user_id
+        task_id: Task ID to toggle
+        completed: Whether task is completed (true) or not (false)
+
+    Returns:
+        JSON string with updated task details
+    """
+    user_id = ctx.context.user_id
+
+    try:
+        result = await mcp_complete_task(
+            user_id=user_id,
+            task_id=task_id,
+            completed=completed
+        )
+        logger.info(f"Task completion updated: {task_id} -> {completed} for user {user_id}")
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Failed to update completion for task {task_id} for user {user_id}: {e}")
+        return json.dumps({"success": False, "error": str(e)})
+
+
+@function_tool(
+    name="complete_all_tasks",
+    description_override="Mark all tasks as completed. Parameters: confirm (boolean, required - must be true to execute)."
+)
+async def complete_all_tasks_tool(
+    ctx: RunContextWrapper,
+    confirm: bool,
+) -> str:
+    """Complete all tasks via MCP tool.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+    [Task]: T017
+
+    Args:
+        ctx: Agents SDK run context containing user_id
+        confirm: Must be true to execute this destructive operation
+
+    Returns:
+        JSON string with bulk completion results
+    """
+    user_id = ctx.context.user_id
+
+    if not confirm:
+        return json.dumps({
+            "success": False,
+            "error": "Confirmation required. Set confirm=true to complete all tasks."
+        })
+
+    try:
+        result = await mcp_complete_all_tasks(
+            user_id=user_id
+        )
+        completed_count = result.get("completed_count", 0)
+        logger.info(f"Completed {completed_count} tasks for user {user_id}")
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Failed to complete all tasks for user {user_id}: {e}")
+        return json.dumps({"success": False, "error": str(e)})
+
+
+@function_tool(
+    name="delete_all_tasks",
+    description_override="Delete all tasks permanently. Parameters: confirm (boolean, required - must be true to execute)."
+)
+async def delete_all_tasks_tool(
+    ctx: RunContextWrapper,
+    confirm: bool,
+) -> str:
+    """Delete all tasks via MCP tool.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+    [Task]: T018
+
+    Args:
+        ctx: Agents SDK run context containing user_id
+        confirm: Must be true to execute this destructive operation
+
+    Returns:
+        JSON string with bulk deletion results
+    """
+    user_id = ctx.context.user_id
+
+    if not confirm:
+        return json.dumps({
+            "success": False,
+            "error": "Confirmation required. Set confirm=true to delete all tasks."
+        })
+
+    try:
+        result = await mcp_delete_all_tasks(
+            user_id=user_id
+        )
+        deleted_count = result.get("deleted_count", 0)
+        logger.info(f"Deleted {deleted_count} tasks for user {user_id}")
+        return json.dumps(result)
+    except Exception as e:
+        logger.error(f"Failed to delete all tasks for user {user_id}: {e}")
+        return json.dumps({"success": False, "error": str(e)})
+
+
+# =============================================================================
+# Tool List for Agent Configuration
+# =============================================================================
+
+# Export all tool functions for easy import
+TOOL_FUNCTIONS = [
+    create_task_tool,
+    list_tasks_tool,
+    update_task_tool,
+    delete_task_tool,
+    complete_task_tool,
+    complete_all_tasks_tool,
+    delete_all_tasks_tool,
+]
+
+
+def get_tool_names() -> list[str]:
+    """Get list of all tool names.
+
+    [From]: specs/010-chatkit-migration/tasks.md - T019
+
+    Returns:
+        List of tool function names
+    """
+    return [tool.name for tool in TOOL_FUNCTIONS]

api/chat.py

@@ -13,7 +13,7 @@ import logging
 import asyncio
 from datetime import datetime
 from typing import Annotated, Optional
-from fastapi import APIRouter, HTTPException, status, Depends, WebSocket, WebSocketDisconnect, BackgroundTasks
+from fastapi import APIRouter, HTTPException, status, Depends, BackgroundTasks, Request
 from pydantic import BaseModel, Field, field_validator, ValidationError
 from sqlmodel import Session
 from sqlalchemy.exc import SQLAlchemyError
@@ -24,14 +24,13 @@ from core.security import decode_access_token
 from models.message import Message, MessageRole
 from services.security import sanitize_message
 from models.conversation import Conversation
-from ai_agent import run_agent_with_streaming, is_gemini_configured
+from ai_agent import run_agent, is_gemini_configured
 from services.conversation import (
     get_or_create_conversation,
     load_conversation_history,
     update_conversation_timestamp
 )
 from services.rate_limiter import check_rate_limit
-from ws_manager.manager import manager
 
 
 # Configure error logger
@@ -290,12 +289,12 @@ async def chat(
         {"role": "user", "content": sanitized_message}
     ]
 
-    # Run AI agent with streaming (broadcasts WebSocket events)
+    # Run AI agent (non-streaming for legacy endpoint)
     # [From]: T014 - Initialize OpenAI Agents SDK with Gemini
-    # [From]: T072 - Use streaming agent for real-time progress
+    # NOTE: Streaming is now handled by ChatKit SSE endpoint
     # [From]: T060 - Add comprehensive error messages for edge cases
     try:
-        ai_response_text = await run_agent_with_streaming(
+        ai_response_text = await run_agent(
             messages=messages_for_agent,
             user_id=user_id
         )
@@ -407,72 +406,277 @@ async def chat(
     )
 
 
-@router.websocket("/ws/{user_id}/chat")
-async def websocket_chat(
-    websocket: WebSocket,
-    user_id: str,
-    db: Session = Depends(get_db)
-):
-    """WebSocket endpoint for real-time chat progress updates.
+# ============================================================================
+# ChatKit SSE Endpoint (Phase 010-chatkit-migration)
+# ============================================================================
 
-    [From]: specs/004-ai-chatbot/research.md - Section 4
-    [Task]: T071
+@router.post("/chatkit")
+async def chatkit_endpoint(
+    request: Request,  # Starlette Request object for raw body access
+    background_tasks: BackgroundTasks,
+):
+    """ChatKit SSE endpoint for streaming chat with Gemini LLM.
+
+    [Task]: T011
+    [From]: specs/010-chatkit-migration/contracts/backend.md - ChatKit SSE Endpoint
+
+    This endpoint implements the ChatKit protocol using Server-Sent Events (SSE).
+    It replaces the WebSocket-based streaming with a simpler HTTP-based approach.
+
+    Endpoint: POST /api/chatkit
+    Response: Server-Sent Events (text/event-stream)
+
+    Authentication: JWT via httpOnly cookie (auth_token)
+
+    Request Body (ChatKit protocol):
+        {
+          "event": "conversation_item_created",
+          "conversation_id": "<thread_uuid>",
+          "item": {
+            "type": "message",
+            "role": "user",
+            "content": [{"type": "text", "text": "Your message here"}]
+          }
+        }
+
+    SSE Event Types:
+        - message_delta: Streaming text content
+        - tool_call_created: Tool invocation started
+        - tool_call_done: Tool execution completed
+        - message_done: Message fully streamed
+        - error: Error occurred
+
+    [From]: specs/010-chatkit-migration/research.md - Section 4
+    """
+    from fastapi import Response
+    from fastapi.responses import StreamingResponse
+    from starlette.requests import Request as StarletteRequest
 
-    This endpoint provides a WebSocket connection for receiving real-time
-    progress events during AI agent execution. Events include:
-    - connection_established: Confirmation of successful connection
-    - agent_thinking: AI agent is processing
-    - tool_starting: A tool is about to execute
-    - tool_progress: Tool execution progress (e.g., "Found 3 tasks")
-    - tool_complete: Tool finished successfully
-    - tool_error: Tool execution failed
-    - agent_done: AI agent finished processing
+    # Import for authentication
+    from core.security import get_current_user_id_from_cookie
 
-    Note: Authentication is handled implicitly by the frontend - users must
-    be logged in to access the chat page. The WebSocket only broadcasts
-    progress updates (not sensitive data), so strict auth is bypassed here.
+    # Get authenticated user ID from JWT cookie
+    # [From]: specs/010-chatkit-migration/contracts/backend.md - Authentication
+    try:
+        user_id = await get_current_user_id_from_cookie(request)
+        if not user_id:
+            # Return error as SSE event
+            async def error_stream():
+                yield "event: error\n"
+                yield 'data: {"detail": "Invalid authentication"}\n\n'
+            return StreamingResponse(
+                error_stream(),
+                media_type="text/event-stream",
+                status_code=401
+            )
+    except Exception as e:
+        error_logger.error(f"Auth error in ChatKit endpoint: {e}")
+        async def error_stream():
+            yield "event: error\n"
+            yield f'data: {{"detail": "Authentication failed"}}\n\n'
+        return StreamingResponse(
+            error_stream(),
+            media_type="text/event-stream",
+            status_code=401
+        )
 
-    Connection URL format:
-        ws://localhost:8000/ws/{user_id}/chat
+    # Check rate limit before processing
+    # [From]: specs/010-chatkit-migration/tasks.md - T020
+    # [From]: specs/010-chatkit-migration/spec.md - FR-015
+    try:
+        from uuid import UUID
+        from core.database import engine
+        from sqlmodel import Session
+
+        # Create synchronous session for rate limit check
+        with Session(engine) as db:
+            allowed, remaining, reset_time = check_rate_limit(db, UUID(user_id))
+
+            if not allowed:
+                # Rate limit exceeded
+                async def rate_limit_stream():
+                    yield "event: error\n"
+                    import json
+                    yield f'data: {json.dumps({"detail": "Daily message limit reached", "limit": 100, "resets_at": reset_time.isoformat() if reset_time else None})}\n\n'
+                return StreamingResponse(
+                    rate_limit_stream(),
+                    media_type="text/event-stream",
+                    status_code=429
+                )
+    except HTTPException:
+        # Re-raise HTTP exceptions (rate limit errors)
+        raise
+    except Exception as e:
+        # Log unexpected errors but don't block the request
+        error_logger.error(f"Rate limit check failed for ChatKit endpoint: {e}")
+        # Continue processing - fail open for rate limit errors
 
-    Args:
-        websocket: The WebSocket connection instance
-        user_id: User ID from URL path (used to route progress events)
-        db: Database session (for any future DB operations)
-
-    The connection is kept alive and can receive messages from the client,
-    though currently it's primarily used for server-to-client progress updates.
-    """
-    # Connect the WebSocket (manager handles accept)
-    # [From]: specs/004-ai-chatbot/research.md - Section 4
-    await manager.connect(user_id, websocket)
+    # Create ChatKit server with synchronous database operations
+    # [From]: specs/010-chatkit-migration/contracts/backend.md - Store Interface Implementation
+    import json
 
+    # Parse request body
     try:
-        # Keep connection alive and listen for client messages
-        # Currently, we don't expect many client messages, but we
-        # maintain the connection to receive any control messages
-        while True:
-            # Wait for message from client (with timeout)
-            data = await websocket.receive_text()
-
-            # Handle client messages if needed
-            # For now, we just acknowledge receipt
-            # Future: could handle ping/pong for connection health
-            if data:
-                # Echo back a simple acknowledgment
-                # (optional - mainly for debugging)
-                pass
-
-    except WebSocketDisconnect:
-        # Normal disconnect - clean up
-        manager.disconnect(user_id, websocket)
-        error_logger.info(f"WebSocket disconnected normally for user {user_id}")
+        body = await request.body()
+    except Exception as e:
+        error_logger.error(f"Failed to read ChatKit request body: {e}")
+        async def error_stream():
+            yield "event: error\n"
+            yield f'data: {{"detail": "Invalid request format"}}\n\n'
+        return StreamingResponse(
+            error_stream(),
+            media_type="text/event-stream",
+            status_code=400
+        )
 
+    # Parse ChatKit protocol request
+    try:
+        request_data = json.loads(body.decode('utf-8'))
     except Exception as e:
-        # Unexpected error - clean up and log
-        error_logger.error(f"WebSocket error for user {user_id}: {e}")
-        manager.disconnect(user_id, websocket)
+        error_logger.error(f"Failed to parse ChatKit request: {e}")
+        async def error_stream():
+            yield "event: error\n"
+            yield f'data: {{"detail": "Invalid JSON format"}}\n\n'
+        return StreamingResponse(
+            error_stream(),
+            media_type="text/event-stream",
+            status_code=400
+        )
+
+    # Extract thread ID and message content
+    conversation_id = request_data.get("conversation_id")
+    item = request_data.get("item", {})
+    event_type = request_data.get("event", "conversation_item_created")
+
+    # Extract user message
+    def extract_message_content(item_dict):
+        content_array = item_dict.get("content", [])
+        for content_block in content_array:
+            if content_block.get("type") == "text":
+                return content_block.get("text", "")
+        return ""
+
+    user_message = extract_message_content(item)
+    if not user_message:
+        async def error_stream():
+            yield "event: error\n"
+            yield f'data: {{"detail": "No message content provided"}}\n\n'
+        return StreamingResponse(
+            error_stream(),
+            media_type="text/event-stream",
+            status_code=400
+        )
+
+    # Use synchronous database session for thread/message operations
+    from core.database import engine
+    from sqlmodel import Session
+
+    # Get or create thread
+    thread_id = conversation_id
+    if not thread_id:
+        # Create new thread for first message
+        with Session(engine) as db:
+            from models.thread import Thread
+            import uuid
+            new_thread = Thread(
+                user_id=UUID(user_id),
+                title=None,
+                thread_metadata={}
+            )
+            db.add(new_thread)
+            db.commit()
+            db.refresh(new_thread)
+            thread_id = str(new_thread.id)
+            error_logger.info(f"Created new thread: {thread_id}")
+
+    # Save user message to database
+    with Session(engine) as db:
+        from models.message import Message, MessageRole
+        import uuid
+        user_msg = Message(
+            thread_id=UUID(thread_id) if thread_id else None,
+            user_id=UUID(user_id),
+            role=MessageRole.USER,
+            content=user_message,
+        )
+        db.add(user_msg)
+        db.commit()
+
+    # Run AI agent and stream response
+    # [Task]: T033 - Add SSE error handling for connection drops
+    async def stream_chat_response():
+        """Stream ChatKit events as SSE with timeout protection.
+
+        [Task]: T034 - Timeout handling for long-running tool executions
+        """
+        try:
+            # Import agent components
+            from ai_agent import run_agent
+
+            # Run agent with timeout protection
+            async with asyncio.timeout(120):
+                ai_response = await run_agent(
+                    messages=[{"role": "user", "content": user_message}],
+                    user_id=user_id
+                )
+
+            # Save assistant message to database
+            with Session(engine) as db:
+                from models.message import Message, MessageRole
+                import uuid
+                assistant_msg = Message(
+                    thread_id=UUID(thread_id) if thread_id else None,
+                    user_id=UUID(user_id),
+                    role=MessageRole.ASSISTANT,
+                    content=ai_response,
+                )
+                db.add(assistant_msg)
+                db.commit()
+
+            # Stream the response
+            import json
+            yield "event: message_delta\n"
+            yield f'data: {json.dumps({"type": "text", "text": ai_response})}\n\n'
+
+            # Send message done event
+            yield "event: message_done\n"
+            yield f'data: {json.dumps({"message_id": thread_id, "role": "assistant", "thread_id": thread_id})}\n\n'
+
+        except TimeoutError:
+            error_logger.error(f"Agent execution timeout for thread {thread_id}")
+            import json
+            yield "event: error\n"
+            yield f'data: {json.dumps({"detail": "Request timed out", "message": "The AI assistant took too long to respond. Please try again."})}\n\n'
+        except Exception as e:
+            error_logger.error(f"Agent execution error: {e}", exc_info=True)
+            import json
+            yield "event: error\n"
+            yield f'data: {json.dumps({"detail": "Processing error", "message": str(e)})}\n\n'
 
-    finally:
-        # Ensure disconnect is always called
-        manager.disconnect(user_id, websocket)
+    # [Task]: T033 - Wrap with connection-aware streaming
+    async def connection_aware_stream():
+        """Stream SSE events with connection drop detection.
+
+        [Task]: T033 - SSE error handling for connection drops
+        """
+        try:
+            async for chunk in stream_chat_response():
+                yield chunk
+        except (ConnectionError, OSError) as e:
+            # Client disconnected during streaming
+            error_logger.info(f"Client disconnected during ChatKit streaming: {e}")
+        except Exception as e:
+            # Unexpected streaming error
+            error_logger.error(f"Unexpected error during ChatKit streaming: {e}", exc_info=True)
+            yield "event: error\n"
+            yield f'data: {{"detail": "Streaming error", "message": str(e)}}\n\n'
+
+    return StreamingResponse(
+        connection_aware_stream(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        }
+    )

chat.py

@@ -1,478 +0,0 @@
-"""Chat API endpoint for AI-powered task management.
-
-[Task]: T015, T071
-[From]: specs/004-ai-chatbot/tasks.md
-
-This endpoint provides a conversational interface for task management.
-Users can create, list, update, complete, and delete tasks through natural language.
-
-Also includes WebSocket endpoint for real-time progress streaming.
-"""
-import uuid
-import logging
-import asyncio
-from datetime import datetime
-from typing import Annotated, Optional
-from fastapi import APIRouter, HTTPException, status, Depends, WebSocket, WebSocketDisconnect, BackgroundTasks
-from pydantic import BaseModel, Field, field_validator, ValidationError
-from sqlmodel import Session
-from sqlalchemy.exc import SQLAlchemyError
-
-from core.database import get_db
-from core.validators import validate_message_length
-from core.security import decode_access_token
-from models.message import Message, MessageRole
-from services.security import sanitize_message
-from models.conversation import Conversation
-from ai_agent import run_agent_with_streaming, is_gemini_configured
-from services.conversation import (
-    get_or_create_conversation,
-    load_conversation_history,
-    update_conversation_timestamp
-)
-from services.rate_limiter import check_rate_limit
-from ws_manager.manager import manager
-
-
-# Configure error logger
-error_logger = logging.getLogger("api.errors")
-error_logger.setLevel(logging.ERROR)
-
-
-# Request/Response models
-class ChatRequest(BaseModel):
-    """Request model for chat endpoint.
-
-    [From]: specs/004-ai-chatbot/plan.md - API Contract
-    """
-    message: str = Field(
-        ...,
-        description="User message content",
-        min_length=1,
-        max_length=10000  # FR-042
-    )
-    conversation_id: Optional[str] = Field(
-        None,
-        description="Optional conversation ID to continue existing conversation"
-    )
-
-    @field_validator('message')
-    @classmethod
-    def validate_message(cls, v: str) -> str:
-        """Validate message content."""
-        if not v or not v.strip():
-            raise ValueError("Message content cannot be empty")
-        if len(v) > 10000:
-            raise ValueError("Message content exceeds maximum length of 10,000 characters")
-        return v.strip()
-
-
-class TaskReference(BaseModel):
-    """Reference to a task created or modified by AI."""
-    id: str
-    title: str
-    description: Optional[str] = None
-    due_date: Optional[str] = None
-    priority: Optional[str] = None
-    completed: bool = False
-
-
-class ChatResponse(BaseModel):
-    """Response model for chat endpoint.
-
-    [From]: specs/004-ai-chatbot/plan.md - API Contract
-    """
-    response: str = Field(
-        ...,
-        description="AI assistant's text response"
-    )
-    conversation_id: str = Field(
-        ...,
-        description="Conversation ID (new or existing)"
-    )
-    tasks: list[TaskReference] = Field(
-        default_factory=list,
-        description="List of tasks created or modified in this interaction"
-    )
-
-
-# Create API router
-router = APIRouter(prefix="/api", tags=["chat"])
-
-
-@router.post("/{user_id}/chat", response_model=ChatResponse, status_code=status.HTTP_200_OK)
-async def chat(
-    user_id: str,
-    request: ChatRequest,
-    background_tasks: BackgroundTasks,
-    db: Session = Depends(get_db)
-):
-    """Process user message through AI agent and return response.
-
-    [From]: specs/004-ai-chatbot/spec.md - US1
-
-    This endpoint:
-    1. Validates user input and rate limits
-    2. Gets or creates conversation
-    3. Runs AI agent with WebSocket progress streaming
-    4. Returns AI response immediately
-    5. Saves messages to DB in background (non-blocking)
-
-    Args:
-        user_id: User ID (UUID string from path)
-        request: Chat request with message and optional conversation_id
-        background_tasks: FastAPI background tasks for non-blocking DB saves
-        db: Database session
-
-    Returns:
-        ChatResponse with AI response, conversation_id, and task references
-
-    Raises:
-        HTTPException 400: Invalid message content
-        HTTPException 503: AI service unavailable
-    """
-    # Check if Gemini API is configured
-    # [From]: specs/004-ai-chatbot/tasks.md - T022
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    if not is_gemini_configured():
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail={
-                "error": "AI service unavailable",
-                "message": "The AI service is currently not configured. Please ensure GEMINI_API_KEY is set in the environment.",
-                "suggestion": "Contact your administrator or check your API key configuration."
-            }
-        )
-
-    # Validate user_id format
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    try:
-        user_uuid = uuid.UUID(user_id)
-    except ValueError:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail={
-                "error": "Invalid user ID",
-                "message": f"User ID '{user_id}' is not a valid UUID format.",
-                "expected_format": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
-                "suggestion": "Ensure you are using a valid UUID for the user_id path parameter."
-            }
-        )
-
-    # Validate message content
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    try:
-        validated_message = validate_message_length(request.message)
-    except ValueError as e:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail={
-                "error": "Message validation failed",
-                "message": str(e),
-                "max_length": 10000,
-                "suggestion": "Keep your message under 10,000 characters and ensure it contains meaningful content."
-            }
-        )
-
-    # Sanitize message to prevent prompt injection
-    # [From]: T057 - Implement prompt injection sanitization
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    try:
-        sanitized_message = sanitize_message(validated_message)
-    except ValueError as e:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail={
-                "error": "Message content blocked",
-                "message": str(e),
-                "suggestion": "Please rephrase your message without attempting to manipulate system instructions."
-            }
-        )
-
-    # Check rate limit
-    # [From]: specs/004-ai-chatbot/spec.md - NFR-011
-    # [From]: T021 - Implement daily message limit enforcement (100/day)
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    try:
-        allowed, remaining, reset_time = check_rate_limit(db, user_uuid)
-
-        if not allowed:
-            raise HTTPException(
-                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
-                detail={
-                    "error": "Rate limit exceeded",
-                    "message": "You have reached the daily message limit. Please try again later.",
-                    "limit": 100,
-                    "resets_at": reset_time.isoformat() if reset_time else None,
-                    "suggestion": "Free tier accounts are limited to 100 messages per day. Upgrade for unlimited access."
-                }
-            )
-    except HTTPException:
-        # Re-raise HTTP exceptions (rate limit errors)
-        raise
-    except Exception as e:
-        # Log unexpected errors but don't block the request
-        error_logger.error(f"Rate limit check failed for user {user_id}: {e}")
-        # Continue processing - fail open for rate limit errors
-
-    # Get or create conversation
-    # [From]: T016 - Implement conversation history loading
-    # [From]: T035 - Handle auto-deleted conversations gracefully
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    conversation_id: uuid.UUID
-
-    if request.conversation_id:
-        # Load existing conversation using service
-        try:
-            conv_uuid = uuid.UUID(request.conversation_id)
-        except ValueError:
-            # Invalid conversation_id format
-            raise HTTPException(
-                status_code=status.HTTP_400_BAD_REQUEST,
-                detail={
-                    "error": "Invalid conversation ID",
-                    "message": f"Conversation ID '{request.conversation_id}' is not a valid UUID format.",
-                    "suggestion": "Provide a valid UUID or omit the conversation_id to start a new conversation."
-                }
-            )
-
-        try:
-            conversation = get_or_create_conversation(
-                db=db,
-                user_id=user_uuid,
-                conversation_id=conv_uuid
-            )
-            conversation_id = conversation.id
-        except (KeyError, ValueError) as e:
-            # Conversation may have been auto-deleted (90-day policy) or otherwise not found
-            # [From]: T035 - Handle auto-deleted conversations gracefully
-            # Create a new conversation instead of failing
-            conversation = get_or_create_conversation(
-                db=db,
-                user_id=user_uuid
-            )
-            conversation_id = conversation.id
-    else:
-        # Create new conversation using service
-        conversation = get_or_create_conversation(
-            db=db,
-            user_id=user_uuid
-        )
-        conversation_id = conversation.id
-
-    # Load conversation history using service
-    # [From]: T016 - Implement conversation history loading
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    try:
-        conversation_history = load_conversation_history(
-            db=db,
-            conversation_id=conversation_id
-        )
-    except SQLAlchemyError as e:
-        error_logger.error(f"Database error loading conversation history for {conversation_id}: {e}")
-        # Continue with empty history if load fails
-        conversation_history = []
-
-    # Prepare user message for background save
-    user_message_id = uuid.uuid4()
-    user_message_data = {
-        "id": user_message_id,
-        "conversation_id": conversation_id,
-        "user_id": user_uuid,
-        "role": MessageRole.USER,
-        "content": sanitized_message,
-        "created_at": datetime.utcnow()
-    }
-
-    # Add current user message to conversation history for AI processing
-    # This is critical - the agent needs the user's current message in context
-    messages_for_agent = conversation_history + [
-        {"role": "user", "content": sanitized_message}
-    ]
-
-    # Run AI agent with streaming (broadcasts WebSocket events)
-    # [From]: T014 - Initialize OpenAI Agents SDK with Gemini
-    # [From]: T072 - Use streaming agent for real-time progress
-    # [From]: T060 - Add comprehensive error messages for edge cases
-    try:
-        ai_response_text = await run_agent_with_streaming(
-            messages=messages_for_agent,
-            user_id=user_id
-        )
-    except ValueError as e:
-        # Configuration errors (missing API key, invalid model)
-        # [From]: T022 - Add error handling for Gemini API unavailability
-        error_logger.error(f"AI configuration error for user {user_id}: {e}")
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail={
-                "error": "AI service configuration error",
-                "message": str(e),
-                "suggestion": "Verify GEMINI_API_KEY and GEMINI_MODEL are correctly configured."
-            }
-        )
-    except ConnectionError as e:
-        # Network/connection issues
-        # [From]: T022 - Add error handling for Gemini API unavailability
-        error_logger.error(f"AI connection error for user {user_id}: {e}")
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail={
-                "error": "AI service unreachable",
-                "message": "Could not connect to the AI service. Please check your network connection.",
-                "suggestion": "If the problem persists, the AI service may be temporarily down."
-            }
-        )
-    except TimeoutError as e:
-        # Timeout errors
-        # [From]: T022 - Add error handling for Gemini API unavailability
-        error_logger.error(f"AI timeout error for user {user_id}: {e}")
-        raise HTTPException(
-            status_code=status.HTTP_504_GATEWAY_TIMEOUT,
-            detail={
-                "error": "AI service timeout",
-                "message": "The AI service took too long to respond. Please try again.",
-                "suggestion": "Your message may be too complex. Try breaking it into smaller requests."
-            }
-        )
-    except Exception as e:
-        # Other errors (rate limits, authentication, context, etc.)
-        # [From]: T022 - Add error handling for Gemini API unavailability
-        error_logger.error(f"Unexpected AI error for user {user_id}: {type(e).__name__}: {e}")
-        raise HTTPException(
-            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
-            detail={
-                "error": "AI service error",
-                "message": f"An unexpected error occurred: {str(e)}",
-                "suggestion": "Please try again later or contact support if the problem persists."
-            }
-        )
-
-    # Prepare AI response for background save
-    ai_message_data = {
-        "id": uuid.uuid4(),
-        "conversation_id": conversation_id,
-        "user_id": user_uuid,
-        "role": MessageRole.ASSISTANT,
-        "content": ai_response_text,
-        "created_at": datetime.utcnow()
-    }
-
-    # Save messages to DB in background (non-blocking)
-    # This significantly improves response time
-    def save_messages_to_db():
-        """Background task to save messages to database."""
-        try:
-            from core.database import engine
-            from sqlmodel import Session
-
-            # Create a new session for background task
-            bg_db = Session(engine)
-
-            try:
-                # Save user message
-                user_msg = Message(**user_message_data)
-                bg_db.add(user_msg)
-
-                # Save AI response
-                ai_msg = Message(**ai_message_data)
-                bg_db.add(ai_msg)
-
-                bg_db.commit()
-
-                # Update conversation timestamp
-                try:
-                    update_conversation_timestamp(db=bg_db, conversation_id=conversation_id)
-                except SQLAlchemyError as e:
-                    error_logger.error(f"Database error updating conversation timestamp for {conversation_id}: {e}")
-
-            except SQLAlchemyError as e:
-                error_logger.error(f"Background task: Database error saving messages for user {user_id}: {e}")
-                bg_db.rollback()
-            finally:
-                bg_db.close()
-        except Exception as e:
-            error_logger.error(f"Background task: Unexpected error saving messages for user {user_id}: {e}")
-
-    background_tasks.add_task(save_messages_to_db)
-
-    # TODO: Parse AI response for task references
-    # This will be enhanced in future tasks to extract task IDs from AI responses
-    task_references: list[TaskReference] = []
-
-    return ChatResponse(
-        response=ai_response_text,
-        conversation_id=str(conversation_id),
-        tasks=task_references
-    )
-
-
-@router.websocket("/ws/{user_id}/chat")
-async def websocket_chat(
-    websocket: WebSocket,
-    user_id: str,
-    db: Session = Depends(get_db)
-):
-    """WebSocket endpoint for real-time chat progress updates.
-
-    [From]: specs/004-ai-chatbot/research.md - Section 4
-    [Task]: T071
-
-    This endpoint provides a WebSocket connection for receiving real-time
-    progress events during AI agent execution. Events include:
-    - connection_established: Confirmation of successful connection
-    - agent_thinking: AI agent is processing
-    - tool_starting: A tool is about to execute
-    - tool_progress: Tool execution progress (e.g., "Found 3 tasks")
-    - tool_complete: Tool finished successfully
-    - tool_error: Tool execution failed
-    - agent_done: AI agent finished processing
-
-    Note: Authentication is handled implicitly by the frontend - users must
-    be logged in to access the chat page. The WebSocket only broadcasts
-    progress updates (not sensitive data), so strict auth is bypassed here.
-
-    Connection URL format:
-        ws://localhost:8000/ws/{user_id}/chat
-
-    Args:
-        websocket: The WebSocket connection instance
-        user_id: User ID from URL path (used to route progress events)
-        db: Database session (for any future DB operations)
-
-    The connection is kept alive and can receive messages from the client,
-    though currently it's primarily used for server-to-client progress updates.
-    """
-    # Connect the WebSocket (manager handles accept)
-    # [From]: specs/004-ai-chatbot/research.md - Section 4
-    await manager.connect(user_id, websocket)
-
-    try:
-        # Keep connection alive and listen for client messages
-        # Currently, we don't expect many client messages, but we
-        # maintain the connection to receive any control messages
-        while True:
-            # Wait for message from client (with timeout)
-            data = await websocket.receive_text()
-
-            # Handle client messages if needed
-            # For now, we just acknowledge receipt
-            # Future: could handle ping/pong for connection health
-            if data:
-                # Echo back a simple acknowledgment
-                # (optional - mainly for debugging)
-                pass
-
-    except WebSocketDisconnect:
-        # Normal disconnect - clean up
-        manager.disconnect(user_id, websocket)
-        error_logger.info(f"WebSocket disconnected normally for user {user_id}")
-
-    except Exception as e:
-        # Unexpected error - clean up and log
-        error_logger.error(f"WebSocket error for user {user_id}: {e}")
-        manager.disconnect(user_id, websocket)
-
-    finally:
-        # Ensure disconnect is always called
-        manager.disconnect(user_id, websocket)

chatkit_server.py

@@ -0,0 +1,358 @@
+"""ChatKit Server implementation for task management with Gemini LLM.
+
+[Task]: T010
+[From]: specs/010-chatkit-migration/contracts/backend.md - ChatKitServer Implementation
+[From]: specs/010-chatkit-migration/research.md - Section 3
+
+This module implements the ChatKitServer class which handles ChatKit protocol
+requests and streams responses using Server-Sent Events (SSE).
+
+The server integrates:
+- ChatKit Python SDK for protocol handling
+- OpenAI Agents SDK for agent orchestration
+- Gemini LLM via OpenAI-compatible endpoint
+- MCP tools wrapped as Agents SDK functions
+
+Architecture:
+    Frontend (ChatKit.js)
+        ↓ SSE with custom fetch
+    ChatKitServer (this module)
+        ↓ Agents SDK
+    Gemini API (via AsyncOpenAI with base_url)
+"""
+import asyncio
+import logging
+from typing import Any, AsyncIterator, Optional
+from uuid import UUID
+from openai import AsyncOpenAI
+from agents import Agent, set_default_openai_client, RunContextWrapper, Runner
+
+from core.config import get_gemini_client, get_settings
+from services.chatkit_store import PostgresChatKitStore
+
+logger = logging.getLogger(__name__)
+
+
+class AgentContext:
+    """Context object passed to agent during execution.
+
+    Contains:
+    - thread_id: Current thread/conversation ID
+    - user_id: Authenticated user ID
+    - store: Database store for persistence
+    - request_context: Additional request metadata
+    """
+
+    def __init__(
+        self,
+        thread_id: str,
+        user_id: str,
+        store: PostgresChatKitStore,
+        request_context: Optional[dict] = None,
+    ):
+        self.thread_id = thread_id
+        self.user_id = user_id
+        self.store = store
+        self.request_context = request_context or {}
+
+
+class TaskManagerChatKitServer:
+    """ChatKit Server for task management with Gemini LLM.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md - ChatKitServer Implementation
+
+    This server extends the ChatKit protocol to work with:
+    - Custom authentication (JWT cookies)
+    - Gemini LLM via OpenAI-compatible endpoint
+    - Server-side tool execution (MCP tools)
+    - PostgreSQL thread/message persistence
+
+    Usage:
+        from fastapi import FastAPI, Request
+        from chatkit_server import TaskManagerChatKitServer
+
+        app = FastAPI()
+        server = TaskManagerChatKitServer(store=postgres_store)
+
+        @app.post("/api/chatkit")
+        async def chatkit_endpoint(request: Request):
+            body = await request.body()
+            user_id = get_current_user_id(request)
+            result = await server.process(body, {"user_id": user_id})
+            if hasattr(result, '__aiter__'):
+                return StreamingResponse(result, media_type="text/event-stream")
+            return Response(content=result.json, media_type="application/json")
+    """
+
+    def __init__(self, store: PostgresChatKitStore):
+        """Initialize the ChatKit server.
+
+        Args:
+            store: PostgresChatKitStore instance for persistence
+        """
+        self.store = store
+
+        # Configure Gemini client as default for Agents SDK
+        # [From]: specs/010-chatkit-migration/research.md - Section 2
+        try:
+            gemini_client = get_gemini_client()
+            set_default_openai_client(gemini_client)
+            logger.info("Gemini client configured for Agents SDK")
+        except Exception as e:
+            logger.warning(f"Gemini client not configured: {e}")
+
+        # Assistant agent will be configured after tools are wrapped
+        # This placeholder will be replaced in T019 with actual tools
+        self.assistant_agent: Optional[Agent] = None
+
+    def set_agent(self, agent: Agent) -> None:
+        """Set the assistant agent with tools.
+
+        [From]: specs/010-chatkit-migration/tasks.md - T019
+
+        Args:
+            agent: Configured Agent with tools and instructions
+        """
+        self.assistant_agent = agent
+        logger.info(f"Agent configured: {agent.name} with model {agent.model}")
+
+    async def process(
+        self,
+        body: bytes,
+        context: dict[str, Any]
+    ) -> Any:
+        """Process ChatKit request and return streaming or non-streaming result.
+
+        [From]: specs/010-chatkit-migration/contracts/backend.md - ChatKit SSE Endpoint
+
+        Args:
+            body: Raw request body bytes from ChatKit.js
+            context: Request context containing user_id and auth info
+
+        Returns:
+            StreamingResult for SSE responses or dict for JSON responses
+
+        Note: This is a placeholder implementation. The actual implementation
+        would use the ChatKit Python SDK's process() method which handles
+        protocol parsing, event routing, and response formatting.
+        """
+        import json
+        from fastapi.responses import StreamingResponse
+
+        # Parse ChatKit protocol request
+        try:
+            request_data = json.loads(body.decode('utf-8'))
+        except Exception as e:
+            logger.error(f"Failed to parse ChatKit request: {e}")
+            return {"error": "Invalid request format"}
+
+        # Extract thread ID and message content
+        conversation_id = request_data.get("conversation_id")
+        item = request_data.get("item", {})
+        event_type = request_data.get("event", "conversation_item_created")
+
+        logger.info(f"ChatKit request: event={event_type}, conversation_id={conversation_id}")
+
+        # Get or create thread
+        thread_id = conversation_id
+        if not thread_id:
+            # Create new thread for first message
+            user_id = context.get("user_id")
+            if not user_id:
+                return {"error": "Unauthorized: no user_id in context"}
+
+            thread_meta = await self.store.create_thread(
+                user_id=user_id,
+                title=None,
+                metadata={}
+            )
+            thread_id = thread_meta["id"]
+            logger.info(f"Created new thread: {thread_id}")
+
+        # Extract user message
+        user_message = self._extract_message_content(item)
+        if not user_message:
+            return {"error": "No message content provided"}
+
+        # Build agent context
+        user_id = context.get("user_id")
+        agent_context = AgentContext(
+            thread_id=thread_id,
+            user_id=user_id,
+            store=self.store,
+            request_context=context,
+        )
+
+        # Create user message in database
+        await self.store.create_message(
+            thread_id=thread_id,
+            item={
+                "type": "message",
+                "role": "user",
+                "content": [{"type": "text", "text": user_message}],
+            }
+        )
+
+        # Stream agent response
+        # [From]: specs/010-chatkit-migration/research.md - Section 3
+        # [Task]: T034 - Add timeout handling for long-running tool executions
+        async def stream_response():
+            """Stream ChatKit events as SSE with timeout protection.
+
+            [Task]: T034 - Timeout handling for long-running tool executions
+
+            Implements a 120-second timeout for agent execution to prevent
+            indefinite hangs from slow tools or network issues.
+            """
+            if not self.assistant_agent:
+                yield self._sse_event("error", {"message": "Agent not configured"})
+                return
+
+            try:
+                # Run agent with streaming and timeout
+                # [Task]: T034 - 120 second timeout for entire agent execution
+                # This covers LLM calls, tool executions, and any delays
+                async with asyncio.timeout(120):
+                    result = Runner.run_streamed(
+                        self.assistant_agent,
+                        [{"role": "user", "content": user_message}],
+                        context=agent_context,
+                    )
+
+                    # Collect assistant response
+                    full_response = ""
+                    async for chunk in result:
+                        if hasattr(chunk, 'content'):
+                            content = chunk.content
+                            if content:
+                                full_response += content
+                                # Stream text delta
+                                yield self._sse_event("message_delta", {
+                                    "type": "text",
+                                    "text": content
+                                })
+
+                    # Create assistant message in database
+                    await self.store.create_message(
+                        thread_id=thread_id,
+                        item={
+                            "type": "message",
+                            "role": "assistant",
+                            "content": [{"type": "text", "text": full_response}],
+                        }
+                    )
+
+                    # Send message done event
+                    yield self._sse_event("message_done", {
+                        "message_id": thread_id,
+                        "role": "assistant"
+                    })
+
+            except TimeoutError:
+                # [Task]: T034 - Handle timeout gracefully
+                logger.error(f"Agent execution timeout for thread {thread_id}")
+                yield self._sse_event("error", {
+                    "message": "Request timed out",
+                    "detail": "The AI assistant took too long to respond. Please try again."
+                })
+            except Exception as e:
+                logger.error(f"Agent execution error: {e}", exc_info=True)
+                yield self._sse_event("error", {"message": str(e)})
+
+        # Return streaming result
+        class StreamingResult:
+            def __init__(self, generator):
+                self.generator = generator
+                self.json = json.dumps({"thread_id": thread_id})
+
+            def __aiter__(self):
+                return self.generator()
+
+        return StreamingResult(stream_response())
+
+    def _extract_message_content(self, item: dict) -> str:
+        """Extract text content from ChatKit item.
+
+        Args:
+            item: ChatKit message item
+
+        Returns:
+            Extracted text content
+        """
+        content_array = item.get("content", [])
+        for content_block in content_array:
+            if content_block.get("type") == "text":
+                return content_block.get("text", "")
+        return ""
+
+    def _sse_event(self, event_type: str, data: dict) -> str:
+        """Format data as Server-Sent Event.
+
+        [From]: specs/010-chatkit-migration/contracts/backend.md - SSE Event Types
+
+        Args:
+            event_type: Event type name
+            data: Event data payload
+
+        Returns:
+            Formatted SSE string
+        """
+        import json
+        return f"event: {event_type}\ndata: {json.dumps(data)}\n\n"
+
+
+# Create singleton instance (will be configured with tools in T019)
+_server_instance: Optional[TaskManagerChatKitServer] = None
+
+
+def get_chatkit_server(store: PostgresChatKitStore) -> TaskManagerChatKitServer:
+    """Get or create ChatKit server singleton with agent configuration.
+
+    [From]: specs/010-chatkit-migration/contracts/backend.md
+
+    [Task]: T019 - Configure TaskAssistant agent with Gemini model and wrapped tools
+
+    Args:
+        store: PostgresChatKitStore instance
+
+    Returns:
+        ChatKit server instance with configured agent
+    """
+    global _server_instance
+    if _server_instance is None:
+        _server_instance = TaskManagerChatKitServer(store)
+
+        # Configure the assistant agent with tools
+        # [From]: specs/010-chatkit-migration/tasks.md - T019
+        # [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+        from ai_agent.tool_wrappers import TOOL_FUNCTIONS
+        from core.config import get_settings
+
+        settings = get_settings()
+
+        # Create the TaskAssistant agent with Gemini model
+        # [From]: specs/010-chatkit-migration/research.md - Section 3
+        assistant_agent = Agent[AgentContext](
+            name="TaskAssistant",
+            model=settings.gemini_model or "gemini-2.0-flash-exp",
+            instructions="""You are a helpful task management assistant. You help users create, list, update, complete, and delete tasks through natural language.
+
+Available tools:
+- create_task: Create a new task with title, description, due date, priority, tags
+- list_tasks: List all tasks with optional filters
+- update_task: Update an existing task
+- delete_task: Delete a task
+- complete_task: Mark a task as completed or incomplete
+- complete_all_tasks: Mark all tasks as completed (requires confirmation)
+- delete_all_tasks: Delete all tasks (requires confirmation)
+
+When users ask about tasks, use the appropriate tool. Always confirm destructive actions (complete_all_tasks, delete_all_tasks) by requiring the confirm parameter.
+
+Be concise and helpful. If a user's request is unclear, ask for clarification.""",
+            tools=TOOL_FUNCTIONS,
+        )
+
+        _server_instance.set_agent(assistant_agent)
+        logger.info(f"ChatKit server initialized with {len(TOOL_FUNCTIONS)} tools and model {settings.gemini_model}")
+
+    return _server_instance

config.py

@@ -1,54 +0,0 @@
-"""Application configuration and settings.
-
-[Task]: T009
-[From]: specs/001-user-auth/plan.md
-
-[Task]: T003
-[From]: specs/004-ai-chatbot/plan.md
-"""
-import os
-from pydantic_settings import BaseSettings, SettingsConfigDict
-from functools import lru_cache
-
-
-class Settings(BaseSettings):
-    """Application settings loaded from environment variables."""
-
-    # Database
-    database_url: str
-
-    # JWT
-    jwt_secret: str
-    jwt_algorithm: str = "HS256"
-    jwt_expiration_days: int = 7
-
-    # CORS
-    frontend_url: str
-
-    # Environment
-    environment: str = "development"
-
-    # Gemini API (Phase III: AI Chatbot)
-    gemini_api_key: str | None = None  # Optional for migration/setup
-    gemini_model: str = "gemini-2.0-flash-exp"
-
-    model_config = SettingsConfigDict(
-        env_file=".env",
-        case_sensitive=False,
-        # Support legacy Better Auth environment variables
-        env_prefix="",
-        extra="ignore"
-    )
-
-
-@lru_cache()
-def get_settings() -> Settings:
-    """Get cached settings instance.
-
-    Returns:
-        Settings: Application settings
-
-    Raises:
-        ValueError: If required environment variables are not set
-    """
-    return Settings()

core/config.py

@@ -5,6 +5,9 @@
 
 [Task]: T003
 [From]: specs/004-ai-chatbot/plan.md
+
+Extended for ChatKit migration with Gemini OpenAI-compatible endpoint.
+[From]: specs/010-chatkit-migration/tasks.md - T008
 """
 import os
 from pydantic_settings import BaseSettings, SettingsConfigDict
@@ -18,12 +21,12 @@ class Settings(BaseSettings):
     database_url: str
 
     # JWT
-    jwt_secret: str = "change-me-in-production-use-env-var"
+    jwt_secret: str
     jwt_algorithm: str = "HS256"
     jwt_expiration_days: int = 7
 
-    # CORS (optional, defaults to allow all for public API)
-    frontend_url: str = "*"
+    # CORS
+    frontend_url: str
 
     # Environment
     environment: str = "development"
@@ -31,6 +34,7 @@ class Settings(BaseSettings):
     # Gemini API (Phase III: AI Chatbot)
     gemini_api_key: str | None = None  # Optional for migration/setup
     gemini_model: str = "gemini-2.0-flash-exp"
+    gemini_base_url: str = "https://generativelanguage.googleapis.com/v1beta/openai/"  # ChatKit migration
 
     model_config = SettingsConfigDict(
         env_file=".env",
@@ -52,3 +56,38 @@ def get_settings() -> Settings:
         ValueError: If required environment variables are not set
     """
     return Settings()
+
+
+def get_gemini_client():
+    """Create and return an AsyncOpenAI client configured for Gemini.
+
+    [From]: specs/010-chatkit-migration/research.md - Section 2
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Tool Contracts
+
+    This client uses Gemini's OpenAI-compatible endpoint, allowing us to use
+    the OpenAI SDK and Agents SDK with Gemini as the LLM provider.
+
+    Returns:
+        AsyncOpenAI: OpenAI client configured for Gemini
+
+    Example:
+        from openai import AsyncOpenAI
+        from agents import set_default_openai_client
+
+        client = get_gemini_client()
+        set_default_openai_client(client)
+    """
+    from openai import AsyncOpenAI
+
+    settings = get_settings()
+
+    if not settings.gemini_api_key:
+        raise ValueError(
+            "GEMINI_API_KEY is not set. Please set it in your environment or .env file. "
+            "Get your API key from https://aistudio.google.com"
+        )
+
+    return AsyncOpenAI(
+        api_key=settings.gemini_api_key,
+        base_url=settings.gemini_base_url,
+    )

core/security.py

@@ -145,3 +145,40 @@ def decode_access_token(token: str) -> dict:
             detail="Could not validate credentials",
             headers={"WWW-Authenticate": "Bearer"},
         )
+
+
+async def get_current_user_id_from_cookie(request) -> Optional[str]:
+    """Extract and validate user ID from JWT token in httpOnly cookie.
+
+    [Task]: T011
+    [From]: specs/010-chatkit-migration/contracts/backend.md - Authentication Contracts
+
+    This function extracts the JWT token from the auth_token httpOnly cookie,
+    decodes it, and returns the user_id (sub claim).
+
+    Args:
+        request: FastAPI/Starlette request object
+
+    Returns:
+        User ID (UUID string) or None if authentication fails
+
+    Raises:
+        HTTPException: If token is invalid (only if raise_on_error=True)
+    """
+    # Try httpOnly cookie first
+    # [From]: specs/010-chatkit-migration/contracts/backend.md
+    auth_token = request.cookies.get("auth_token")
+    if not auth_token:
+        return None
+
+    try:
+        payload = decode_access_token(auth_token)
+        user_id = payload.get("sub")
+        return user_id
+    except HTTPException:
+        return None
+    except Exception as e:
+        # Log but don't raise for non-critical operations
+        import logging
+        logging.getLogger("api.auth").warning(f"Failed to decode token from cookie: {e}")
+        return None

migrations/migrate_threads.sql

@@ -0,0 +1,76 @@
+-- ChatKit Migration: Create threads table and update messages table
+--
+-- [From]: specs/010-chatkit-migration/data-model.md - Database Schema Migration
+-- [Task]: T006
+--
+-- This migration:
+-- 1. Creates the threads table for ChatKit conversation management
+-- 2. Adds thread_id column to messages table
+-- 3. Migrates existing conversation_id data to thread_id
+-- 4. Creates indexes for query optimization
+--
+-- IMPORTANT: Run this migration after deploying the Thread model
+--
+-- To run: psql $DATABASE_URL < migrations/migrate_threads.sql
+
+-- BEGIN;
+
+-- 1. Create threads table
+CREATE TABLE IF NOT EXISTS threads (
+    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+    user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+    title VARCHAR(255),
+    metadata JSONB,
+    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
+    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
+);
+
+-- 2. Create indexes for threads table
+CREATE INDEX IF NOT EXISTS idx_thread_user_id ON threads(user_id);
+CREATE INDEX IF NOT EXISTS idx_thread_updated_at ON threads(user_id, updated_at DESC);
+
+-- 3. Add thread_id column to messages table (nullable initially)
+ALTER TABLE message ADD COLUMN IF NOT EXISTS thread_id UUID REFERENCES threads(id) ON DELETE CASCADE;
+
+-- 4. Create index for thread_id in messages table
+CREATE INDEX IF NOT EXISTS idx_message_thread_id ON message(thread_id, created_at ASC);
+
+-- 5. Migrate existing conversation data to threads
+-- This creates a thread for each unique conversation and links messages to it
+-- Skip this step if starting fresh (no existing conversations)
+INSERT INTO threads (id, user_id, created_at, updated_at)
+SELECT DISTINCT
+    c.id as id,           -- Use same ID as conversation for easy mapping
+    c.user_id,
+    c.created_at,
+    c.updated_at
+FROM conversation c
+WHERE NOT EXISTS (SELECT 1 FROM threads t WHERE t.id = c.id);
+
+-- 6. Update messages to point to the new thread_id
+-- This maps existing messages to their corresponding threads
+UPDATE message m
+SET thread_id = m.conversation_id
+WHERE m.conversation_id IS NOT NULL
+  AND m.thread_id IS NULL;
+
+-- 7. After migration, make thread_id NOT NULL (only after validating data)
+-- Uncomment these lines after verifying successful migration:
+-- ALTER TABLE message ALTER COLUMN thread_id SET NOT NULL;
+-- ALTER TABLE message ADD CONSTRAINT message_thread_id_fkey FOREIGN KEY (thread_id) REFERENCES threads(id) ON DELETE CASCADE;
+
+-- Optional: Drop old conversation_id column after full migration
+-- Uncomment ONLY after confirming ChatKit is working correctly:
+-- ALTER TABLE message DROP COLUMN IF EXISTS conversation_id;
+-- DROP INDEX IF EXISTS idx_message_conversation_created;
+
+-- COMMIT;
+
+-- Rollback commands (if needed):
+-- BEGIN;
+-- ALTER TABLE message DROP COLUMN IF EXISTS thread_id;
+-- DROP INDEX IF EXISTS idx_message_thread_id;
+-- DROP INDEX IF EXISTS idx_thread_user_id;
+-- DROP INDEX IF EXISTS idx_thread_updated_at;
+-- DROP TABLE IF EXISTS threads;
+-- COMMIT;

models/thread.py

@@ -0,0 +1,72 @@
+"""Thread model for ChatKit integration.
+
+[Task]: T004
+[From]: specs/010-chatkit-migration/data-model.md
+
+This model implements ChatKit's thread abstraction for grouping messages
+into conversations. It extends the existing Conversation model with
+ChatKit-specific fields.
+
+Migration Note: The existing Conversation model serves a similar purpose.
+During migration, we can either:
+1. Use Thread alongside Conversation (dual model approach)
+2. Migrate Conversation to Thread (single model approach)
+
+This implementation uses Thread as the primary model for ChatKit integration.
+"""
+import uuid
+from datetime import datetime
+from typing import Optional
+from sqlmodel import Field, SQLModel
+from sqlalchemy import Column, DateTime, JSON as SQLJSON, String as SQLString, Index
+from sqlalchemy.dialects.postgresql import JSONB
+
+
+class Thread(SQLModel, table=True):
+    """Thread model representing a ChatKit conversation session.
+
+    ChatKit uses "thread" terminology for conversation grouping.
+    A thread contains multiple messages and belongs to a single user.
+
+    [From]: specs/010-chatkit-migration/data-model.md - Thread Entity
+    """
+
+    __tablename__ = "threads"
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    user_id: uuid.UUID = Field(foreign_key="users.id", index=True, nullable=False)
+
+    # Optional thread title (ChatKit allows threads to have titles)
+    title: Optional[str] = Field(
+        default=None,
+        max_length=255,
+        sa_column=Column(SQLString(255), nullable=True)
+    )
+
+    # Thread metadata (tags, settings, etc.) stored as JSONB
+    # Note: Using 'thread_metadata' instead of 'metadata' to avoid SQLAlchemy reserved attribute
+    thread_metadata: Optional[dict] = Field(
+        default=None,
+        sa_column=Column("metadata", JSONB, nullable=True),  # Column name in DB is 'metadata'
+    )
+
+    # Thread creation timestamp
+    created_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        sa_column=Column(DateTime(timezone=True), nullable=False)
+    )
+
+    # Last message/update timestamp (auto-updated by application logic)
+    updated_at: datetime = Field(
+        default_factory=datetime.utcnow,
+        sa_column=Column(DateTime(timezone=True), nullable=False, index=True)
+    )
+
+    # Table indexes for query optimization
+    __table_args__ = (
+        Index('idx_thread_user_id', 'user_id'),
+        Index('idx_thread_updated_at', 'user_id', 'updated_at'),  # For sorting conversations by recent activity
+    )
+
+    def __repr__(self) -> str:
+        return f"<Thread(id={self.id}, user_id={self.user_id}, title={self.title})>"

security.py

@@ -1,276 +0,0 @@
-"""Security utilities for the AI chatbot.
-
-[Task]: T057
-[From]: specs/004-ai-chatbot/tasks.md
-
-This module provides security functions including prompt injection sanitization,
-input validation, and content filtering.
-"""
-import re
-import html
-from typing import Optional, List
-
-
-# Known prompt injection patterns
-PROMPT_INJECTION_PATTERNS = [
-    # Direct instructions to ignore previous context
-    r"(?i)ignore\s+(all\s+)?(previous|above|prior)",
-    r"(?i)disregard\s+(all\s+)?(previous|above|prior)",
-    r"(?i)forget\s+(everything|all\s+instructions|previous)",
-    r"(?i)override\s+(your\s+)?programming",
-    r"(?i)new\s+(instruction|direction|rule)s?",
-    r"(?i)change\s+(your\s+)?(behavior|role|persona)",
-
-    # Jailbreak attempts
-    r"(?i)(jailbreak|jail\s*break)",
-    r"(?i)(developer|admin|root|privileged)\s+mode",
-    r"(?i)act\s+as\s+(a\s+)?(developer|admin|root)",
-    r"(?i)roleplay\s+as",
-    r"(?i)pretend\s+(to\s+be|you're)",
-    r"(?i)simulate\s+being",
-
-    # System prompt extraction
-    r"(?i)show\s+(your\s+)?(instructions|system\s+prompt|prompt)",
-    r"(?i)print\s+(your\s+)?(instructions|system\s+prompt)",
-    r"(?i)reveal\s+(your\s+)?(instructions|system\s+prompt)",
-    r"(?i)what\s+(are\s+)?your\s+instructions",
-    r"(?i)tell\s+me\s+how\s+you\s+work",
-
-    # DAN and similar jailbreaks
-    r"(?i)do\s+anything\s+now",
-    r"(?i)unrestricted\s+mode",
-    r"(?i)no\s+limitations?",
-    r"(?i)bypass\s+(safety|filters|restrictions)",
-    r"(?i)\bDAN\b",  # Do Anything Now
-]
-
-
-def sanitize_message(message: str, max_length: int = 10000) -> str:
-    """Sanitize a user message to prevent prompt injection attacks.
-
-    [From]: specs/004-ai-chatbot/spec.md - NFR-017
-
-    Args:
-        message: The raw user message
-        max_length: Maximum allowed message length
-
-    Returns:
-        Sanitized message safe for processing by AI
-
-    Raises:
-        ValueError: If message contains severe injection attempts
-    """
-    if not message:
-        return ""
-
-    # Trim to max length
-    message = message[:max_length]
-
-    # Check for severe injection patterns
-    detected = detect_prompt_injection(message)
-    if detected:
-        # For severe attacks, reject the message
-        if detected["severity"] == "high":
-            raise ValueError(
-                "This message contains content that cannot be processed. "
-                "Please rephrase your request."
-            )
-
-    # Apply sanitization
-    sanitized = _apply_sanitization(message)
-
-    return sanitized
-
-
-def detect_prompt_injection(message: str) -> Optional[dict]:
-    """Detect potential prompt injection attempts in a message.
-
-    [From]: specs/004-ai-chatbot/spec.md - NFR-017
-
-    Args:
-        message: The message to check
-
-    Returns:
-        Dictionary with detection info if injection detected, None otherwise:
-        {
-            "detected": True,
-            "severity": "low" | "medium" | "high",
-            "pattern": "matched pattern",
-            "confidence": 0.0-1.0
-        }
-    """
-    message_lower = message.lower()
-
-    for pattern in PROMPT_INJECTION_PATTERNS:
-        match = re.search(pattern, message_lower)
-
-        if match:
-            # Determine severity based on pattern type
-            severity = _get_severity_for_pattern(pattern)
-
-            # Check for context that might indicate legitimate use
-            is_legitimate = _check_legitimate_context(message, match.group())
-
-            if not is_legitimate:
-                return {
-                    "detected": True,
-                    "severity": severity,
-                    "pattern": match.group(),
-                    "confidence": 0.8
-                }
-
-    return None
-
-
-def _get_severity_for_pattern(pattern: str) -> str:
-    """Determine severity level for a matched pattern.
-
-    Args:
-        pattern: The regex pattern that matched
-
-    Returns:
-        "low", "medium", or "high"
-    """
-    pattern_lower = pattern.lower()
-
-    # High severity: direct jailbreak attempts
-    if any(word in pattern_lower for word in ["jailbreak", "dan", "unrestricted", "bypass"]):
-        return "high"
-
-    # High severity: system prompt extraction
-    if any(word in pattern_lower for word in ["show", "print", "reveal", "instructions"]):
-        return "high"
-
-    # Medium severity: role/persona manipulation
-    if any(word in pattern_lower for word in ["act as", "pretend", "roleplay", "override"]):
-        return "medium"
-
-    # Low severity: ignore instructions
-    if any(word in pattern_lower for word in ["ignore", "disregard", "forget"]):
-        return "low"
-
-    return "low"
-
-
-def _check_legitimate_context(message: str, matched_text: str) -> bool:
-    """Check if a matched pattern might be legitimate user content.
-
-    [From]: specs/004-ai-chatbot/spec.md - NFR-017
-
-    Args:
-        message: The full message
-        matched_text: The text that matched a pattern
-
-    Returns:
-        True if this appears to be legitimate context, False otherwise
-    """
-    message_lower = message.lower()
-    matched_lower = matched_text.lower()
-
-    # Check if the matched text is part of a task description (legitimate)
-    legitimate_contexts = [
-        # Common task-related phrases
-        "task to ignore",
-        "mark as complete",
-        "disregard this",
-        "role in the project",
-        "change status",
-        "update the role",
-        "priority change",
-    ]
-
-    for context in legitimate_contexts:
-        if context in message_lower:
-            return True
-
-    # Check if matched text is very short (likely false positive)
-    if len(matched_text) <= 3:
-        return True
-
-    return False
-
-
-def _apply_sanitization(message: str) -> str:
-    """Apply sanitization transformations to a message.
-
-    [From]: specs/004-ai-chatbot/spec.md - NFR-017
-
-    Args:
-        message: The message to sanitize
-
-    Returns:
-        Sanitized message
-    """
-    # Remove excessive whitespace
-    message = re.sub(r"\s+", " ", message)
-
-    # Remove control characters except newlines and tabs
-    message = re.sub(r"[\x00-\x08\x0b-\x0c\x0e-\x1f\x7f-\x9f]", "", message)
-
-    # Normalize line endings
-    message = message.replace("\r\n", "\n").replace("\r", "\n")
-
-    # Limit consecutive newlines to 2
-    message = re.sub(r"\n{3,}", "\n\n", message)
-
-    return message.strip()
-
-
-def validate_task_input(task_data: dict) -> tuple[bool, Optional[str]]:
-    """Validate task-related input for security issues.
-
-    [From]: specs/004-ai-chatbot/spec.md - NFR-017
-
-    Args:
-        task_data: Dictionary containing task fields
-
-    Returns:
-        Tuple of (is_valid, error_message)
-    """
-    if not isinstance(task_data, dict):
-        return False, "Invalid task data format"
-
-    # Check for SQL injection patterns in string fields
-    sql_patterns = [
-        r"(?i)(\bunion\b.*\bselect\b)",
-        r"(?i)(\bselect\b.*\bfrom\b)",
-        r"(?i)(\binsert\b.*\binto\b)",
-        r"(?i)(\bupdate\b.*\bset\b)",
-        r"(?i)(\bdelete\b.*\bfrom\b)",
-        r"(?i)(\bdrop\b.*\btable\b)",
-        r";\s*(union|select|insert|update|delete|drop)",
-    ]
-
-    for key, value in task_data.items():
-        if isinstance(value, str):
-            for pattern in sql_patterns:
-                if re.search(pattern, value):
-                    return False, f"Invalid characters in {key}"
-
-            # Check for script injection
-            if re.search(r"<script[^>]*>.*?</script>", value, re.IGNORECASE):
-                return False, f"Invalid content in {key}"
-
-    return True, None
-
-
-def sanitize_html_content(content: str) -> str:
-    """Sanitize HTML content by escaping potentially dangerous elements.
-
-    [From]: specs/004-ai-chatbot/spec.md - NFR-017
-
-    Args:
-        content: Content that may contain HTML
-
-    Returns:
-        Escaped HTML string
-    """
-    return html.escape(content, quote=False)
-
-
-__all__ = [
-    "sanitize_message",
-    "detect_prompt_injection",
-    "validate_task_input",
-    "sanitize_html_content",
-]

services/chatkit_store.py

@@ -0,0 +1,416 @@
+"""PostgreSQL Store implementation for ChatKit SDK.
+
+[Task]: T009
+[From]: specs/010-chatkit-migration/data-model.md - ChatKit SDK Interface Requirements
+[From]: specs/010-chatkit-migration/contracts/backend.md - Store Interface Implementation
+
+This module implements the ChatKit Store interface using SQLModel and PostgreSQL.
+The Store interface is required by ChatKit's Python SDK for thread and message persistence.
+
+ChatKit Store Protocol Methods:
+- list_threads: List threads for a user with pagination
+- get_thread: Get a specific thread by ID
+- create_thread: Create a new thread
+- update_thread: Update thread metadata
+- delete_thread: Delete a thread
+- list_messages: List messages in a thread
+- get_message: Get a specific message
+- create_message: Create a new message
+- update_message: Update a message
+- delete_message: Delete a message
+"""
+import uuid
+from datetime import datetime
+from typing import Any, Optional
+from sqlmodel import Session, select, col
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from models.thread import Thread
+from models.message import Message, MessageRole
+
+
+class PostgresChatKitStore:
+    """PostgreSQL implementation of ChatKit Store interface.
+
+    [From]: specs/010-chatkit-migration/data-model.md - Store Interface
+
+    This store provides thread and message persistence for ChatKit using
+    the existing SQLModel models and PostgreSQL database.
+
+    Note: The ChatKit SDK uses a Protocol-based interface. The actual
+    protocol types (ThreadMetadata, MessageItem, etc.) would be imported
+    from the openai_chatkit package. For this implementation, we use
+    dictionary-based representations until the SDK is installed.
+
+    Usage:
+        from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
+        from services.chatkit_store import PostgresChatKitStore
+
+        engine = create_async_engine(database_url)
+        async with AsyncSession(engine) as session:
+            store = PostgresChatKitStore(session)
+            thread = await store.create_thread(
+                user_id="user-123",
+                title="My Conversation",
+                metadata={"tag": "important"}
+            )
+    """
+
+    def __init__(self, session: AsyncSession):
+        """Initialize the store with a database session.
+
+        Args:
+            session: SQLAlchemy async session for database operations
+        """
+        self.session = session
+
+    async def list_threads(
+        self,
+        user_id: str,
+        limit: int = 50,
+        offset: int = 0
+    ) -> list[dict]:
+        """List threads for a user with pagination.
+
+        [From]: specs/010-chatkit-migration/data-model.md - Retrieve User's Conversations
+
+        Args:
+            user_id: User ID to filter threads
+            limit: Maximum number of threads to return (default: 50)
+            offset: Number of threads to skip (default: 0)
+
+        Returns:
+            List of thread metadata dictionaries
+        """
+        stmt = (
+            select(Thread)
+            .where(Thread.user_id == uuid.UUID(user_id))
+            .order_by(Thread.updated_at.desc())
+            .limit(limit)
+            .offset(offset)
+        )
+        result = await self.session.execute(stmt)
+        threads = result.scalars().all()
+
+        return [
+            {
+                "id": str(thread.id),
+                "user_id": str(thread.user_id),
+                "title": thread.title,
+                "metadata": thread.thread_metadata or {},
+                "created_at": thread.created_at.isoformat(),
+                "updated_at": thread.updated_at.isoformat(),
+            }
+            for thread in threads
+        ]
+
+    async def get_thread(self, thread_id: str) -> Optional[dict]:
+        """Get a specific thread by ID.
+
+        Args:
+            thread_id: Thread UUID as string
+
+        Returns:
+            Thread metadata dictionary or None if not found
+        """
+        stmt = select(Thread).where(Thread.id == uuid.UUID(thread_id))
+        result = await self.session.execute(stmt)
+        thread = result.scalar_one_or_none()
+
+        if thread is None:
+            return None
+
+        return {
+            "id": str(thread.id),
+            "user_id": str(thread.user_id),
+            "title": thread.title,
+            "metadata": thread.thread_metadata or {},
+            "created_at": thread.created_at.isoformat(),
+            "updated_at": thread.updated_at.isoformat(),
+        }
+
+    async def create_thread(
+        self,
+        user_id: str,
+        title: Optional[str] = None,
+        metadata: Optional[dict] = None
+    ) -> dict:
+        """Create a new thread.
+
+        Args:
+            user_id: User ID who owns the thread
+            title: Optional thread title
+            metadata: Optional thread metadata
+
+        Returns:
+            Created thread metadata dictionary
+        """
+        thread = Thread(
+            user_id=uuid.UUID(user_id),
+            title=title,
+            thread_metadata=metadata or {},
+        )
+        self.session.add(thread)
+        await self.session.commit()
+        await self.session.refresh(thread)
+
+        return {
+            "id": str(thread.id),
+            "user_id": str(thread.user_id),
+            "title": thread.title,
+            "metadata": thread.thread_metadata or {},
+            "created_at": thread.created_at.isoformat(),
+            "updated_at": thread.updated_at.isoformat(),
+        }
+
+    async def update_thread(
+        self,
+        thread_id: str,
+        title: Optional[str] = None,
+        metadata: Optional[dict] = None
+    ) -> Optional[dict]:
+        """Update a thread.
+
+        Args:
+            thread_id: Thread UUID as string
+            title: New title (optional)
+            metadata: New metadata (optional)
+
+        Returns:
+            Updated thread metadata dictionary or None if not found
+        """
+        stmt = select(Thread).where(Thread.id == uuid.UUID(thread_id))
+        result = await self.session.execute(stmt)
+        thread = result.scalar_one_or_none()
+
+        if thread is None:
+            return None
+
+        if title is not None:
+            thread.title = title
+        if metadata is not None:
+            thread.thread_metadata = metadata
+        thread.updated_at = datetime.utcnow()
+
+        await self.session.commit()
+        await self.session.refresh(thread)
+
+        return {
+            "id": str(thread.id),
+            "user_id": str(thread.user_id),
+            "title": thread.title,
+            "metadata": thread.thread_metadata or {},
+            "created_at": thread.created_at.isoformat(),
+            "updated_at": thread.updated_at.isoformat(),
+        }
+
+    async def delete_thread(self, thread_id: str) -> bool:
+        """Delete a thread.
+
+        Args:
+            thread_id: Thread UUID as string
+
+        Returns:
+            True if deleted, False if not found
+        """
+        stmt = select(Thread).where(Thread.id == uuid.UUID(thread_id))
+        result = await self.session.execute(stmt)
+        thread = result.scalar_one_or_none()
+
+        if thread is None:
+            return False
+
+        await self.session.delete(thread)
+        await self.session.commit()
+        return True
+
+    async def list_messages(
+        self,
+        thread_id: str,
+        limit: int = 50,
+        offset: int = 0
+    ) -> list[dict]:
+        """List messages in a thread.
+
+        [From]: specs/010-chatkit-migration/data-model.md - Retrieve Conversation Messages
+
+        Args:
+            thread_id: Thread UUID as string
+            limit: Maximum number of messages to return
+            offset: Number of messages to skip
+
+        Returns:
+            List of message item dictionaries
+        """
+        stmt = (
+            select(Message)
+            .where(Message.thread_id == uuid.UUID(thread_id))
+            .order_by(Message.created_at.asc())
+            .limit(limit)
+            .offset(offset)
+        )
+        result = await self.session.execute(stmt)
+        messages = result.scalars().all()
+
+        return [
+            {
+                "id": str(msg.id),
+                "type": "message",
+                "role": msg.role.value,
+                "content": [{"type": "text", "text": msg.content}],
+                "tool_calls": msg.tool_calls,
+                "created_at": msg.created_at.isoformat(),
+            }
+            for msg in messages
+        ]
+
+    async def get_message(self, message_id: str) -> Optional[dict]:
+        """Get a specific message by ID.
+
+        Args:
+            message_id: Message UUID as string
+
+        Returns:
+            Message item dictionary or None if not found
+        """
+        stmt = select(Message).where(Message.id == uuid.UUID(message_id))
+        result = await self.session.execute(stmt)
+        message = result.scalar_one_or_none()
+
+        if message is None:
+            return None
+
+        return {
+            "id": str(message.id),
+            "type": "message",
+            "role": message.role.value,
+            "content": [{"type": "text", "text": message.content}],
+            "tool_calls": message.tool_calls,
+            "created_at": message.created_at.isoformat(),
+        }
+
+    async def create_message(
+        self,
+        thread_id: str,
+        item: dict
+    ) -> dict:
+        """Create a new message in a thread.
+
+        Args:
+            thread_id: Thread UUID as string
+            item: Message item from ChatKit (UserMessageItem or ClientToolCallOutputItem)
+
+        Returns:
+            Created message item dictionary
+
+        Raises:
+            ValueError: If item format is invalid
+        """
+        # Extract content from ChatKit item format
+        # ChatKit uses: {"type": "message", "role": "user", "content": [{"type": "text", "text": "..."}]}
+        item_type = item.get("type", "message")
+        role = item.get("role", "user")
+
+        # Extract text content from content array
+        content_array = item.get("content", [])
+        text_content = ""
+        for content_block in content_array:
+            if content_block.get("type") == "text":
+                text_content = content_block.get("text", "")
+                break
+
+        # Handle client tool output items
+        if item_type == "client_tool_call_output":
+            text_content = item.get("output", "")
+
+        message = Message(
+            thread_id=uuid.UUID(thread_id),
+            role=MessageRole(role),
+            content=text_content,
+            tool_calls=item.get("tool_calls"),
+        )
+        self.session.add(message)
+        await self.session.commit()
+        await self.session.refresh(message)
+
+        # Update thread's updated_at timestamp
+        thread_stmt = select(Thread).where(Thread.id == uuid.UUID(thread_id))
+        thread_result = await self.session.execute(thread_stmt)
+        thread = thread_result.scalar_one_or_none()
+        if thread:
+            thread.updated_at = datetime.utcnow()
+            await self.session.commit()
+
+        return {
+            "id": str(message.id),
+            "type": "message",
+            "role": message.role.value,
+            "content": [{"type": "text", "text": message.content}],
+            "tool_calls": message.tool_calls,
+            "created_at": message.created_at.isoformat(),
+        }
+
+    async def update_message(
+        self,
+        message_id: str,
+        item: dict
+    ) -> Optional[dict]:
+        """Update a message.
+
+        Args:
+            message_id: Message UUID as string
+            item: Updated message item
+
+        Returns:
+            Updated message item dictionary or None if not found
+        """
+        stmt = select(Message).where(Message.id == uuid.UUID(message_id))
+        result = await self.session.execute(stmt)
+        message = result.scalar_one_or_none()
+
+        if message is None:
+            return None
+
+        # Update content if provided
+        content_array = item.get("content", [])
+        if content_array:
+            for content_block in content_array:
+                if content_block.get("type") == "text":
+                    message.content = content_block.get("text", message.content)
+                    break
+
+        # Update tool_calls if provided
+        if "tool_calls" in item:
+            message.tool_calls = item["tool_calls"]
+
+        await self.session.commit()
+        await self.session.refresh(message)
+
+        return {
+            "id": str(message.id),
+            "type": "message",
+            "role": message.role.value,
+            "content": [{"type": "text", "text": message.content}],
+            "tool_calls": message.tool_calls,
+            "created_at": message.created_at.isoformat(),
+        }
+
+    async def delete_message(self, message_id: str) -> bool:
+        """Delete a message.
+
+        Args:
+            message_id: Message UUID as string
+
+        Returns:
+            True if deleted, False if not found
+        """
+        stmt = select(Message).where(Message.id == uuid.UUID(message_id))
+        result = await self.session.execute(stmt)
+        message = result.scalar_one_or_none()
+
+        if message is None:
+            return False
+
+        await self.session.delete(message)
+        await self.session.commit()
+        return True