Merge remote-tracking branch 'origin/hackathon/copilot' into hackathon/copilot

autogpt_platform/backend/backend/data/understanding.py

@@ -0,0 +1,429 @@
+"""Data models and access layer for user business understanding."""
+
+import logging
+from datetime import datetime
+from typing import Any, Optional, cast
+
+import pydantic
+from prisma.models import UserBusinessUnderstanding
+from prisma.types import (
+    UserBusinessUnderstandingCreateInput,
+    UserBusinessUnderstandingUpdateInput,
+)
+
+from backend.data.redis_client import get_redis_async
+from backend.util.json import SafeJson
+
+logger = logging.getLogger(__name__)
+
+# Cache configuration
+CACHE_KEY_PREFIX = "understanding"
+CACHE_TTL_SECONDS = 48 * 60 * 60  # 48 hours
+
+
+def _cache_key(user_id: str) -> str:
+    """Generate cache key for user business understanding."""
+    return f"{CACHE_KEY_PREFIX}:{user_id}"
+
+
+def _json_to_list(value: Any) -> list[str]:
+    """Convert Json field to list[str], handling None."""
+    if value is None:
+        return []
+    if isinstance(value, list):
+        return cast(list[str], value)
+    return []
+
+
+class BusinessUnderstandingInput(pydantic.BaseModel):
+    """Input model for updating business understanding - all fields optional for incremental updates."""
+
+    # User info
+    user_name: Optional[str] = pydantic.Field(None, description="The user's name")
+    job_title: Optional[str] = pydantic.Field(None, description="The user's job title")
+
+    # Business basics
+    business_name: Optional[str] = pydantic.Field(
+        None, description="Name of the user's business"
+    )
+    industry: Optional[str] = pydantic.Field(None, description="Industry or sector")
+    business_size: Optional[str] = pydantic.Field(
+        None, description="Company size (e.g., '1-10', '11-50')"
+    )
+    user_role: Optional[str] = pydantic.Field(
+        None,
+        description="User's role in the organization (e.g., 'decision maker', 'implementer')",
+    )
+
+    # Processes & activities
+    key_workflows: Optional[list[str]] = pydantic.Field(
+        None, description="Key business workflows"
+    )
+    daily_activities: Optional[list[str]] = pydantic.Field(
+        None, description="Daily activities performed"
+    )
+
+    # Pain points & goals
+    pain_points: Optional[list[str]] = pydantic.Field(
+        None, description="Current pain points"
+    )
+    bottlenecks: Optional[list[str]] = pydantic.Field(
+        None, description="Process bottlenecks"
+    )
+    manual_tasks: Optional[list[str]] = pydantic.Field(
+        None, description="Manual/repetitive tasks"
+    )
+    automation_goals: Optional[list[str]] = pydantic.Field(
+        None, description="Desired automation goals"
+    )
+
+    # Current tools
+    current_software: Optional[list[str]] = pydantic.Field(
+        None, description="Software/tools currently used"
+    )
+    existing_automation: Optional[list[str]] = pydantic.Field(
+        None, description="Existing automations"
+    )
+
+    # Additional context
+    additional_notes: Optional[str] = pydantic.Field(
+        None, description="Any additional context"
+    )
+
+
+class BusinessUnderstanding(pydantic.BaseModel):
+    """Full business understanding model returned from database."""
+
+    id: str
+    user_id: str
+    created_at: datetime
+    updated_at: datetime
+
+    # User info
+    user_name: Optional[str] = None
+    job_title: Optional[str] = None
+
+    # Business basics
+    business_name: Optional[str] = None
+    industry: Optional[str] = None
+    business_size: Optional[str] = None
+    user_role: Optional[str] = None
+
+    # Processes & activities
+    key_workflows: list[str] = pydantic.Field(default_factory=list)
+    daily_activities: list[str] = pydantic.Field(default_factory=list)
+
+    # Pain points & goals
+    pain_points: list[str] = pydantic.Field(default_factory=list)
+    bottlenecks: list[str] = pydantic.Field(default_factory=list)
+    manual_tasks: list[str] = pydantic.Field(default_factory=list)
+    automation_goals: list[str] = pydantic.Field(default_factory=list)
+
+    # Current tools
+    current_software: list[str] = pydantic.Field(default_factory=list)
+    existing_automation: list[str] = pydantic.Field(default_factory=list)
+
+    # Additional context
+    additional_notes: Optional[str] = None
+
+    @classmethod
+    def from_db(cls, db_record: UserBusinessUnderstanding) -> "BusinessUnderstanding":
+        """Convert database record to Pydantic model."""
+        return cls(
+            id=db_record.id,
+            user_id=db_record.userId,
+            created_at=db_record.createdAt,
+            updated_at=db_record.updatedAt,
+            user_name=db_record.userName,
+            job_title=db_record.jobTitle,
+            business_name=db_record.businessName,
+            industry=db_record.industry,
+            business_size=db_record.businessSize,
+            user_role=db_record.userRole,
+            key_workflows=_json_to_list(db_record.keyWorkflows),
+            daily_activities=_json_to_list(db_record.dailyActivities),
+            pain_points=_json_to_list(db_record.painPoints),
+            bottlenecks=_json_to_list(db_record.bottlenecks),
+            manual_tasks=_json_to_list(db_record.manualTasks),
+            automation_goals=_json_to_list(db_record.automationGoals),
+            current_software=_json_to_list(db_record.currentSoftware),
+            existing_automation=_json_to_list(db_record.existingAutomation),
+            additional_notes=db_record.additionalNotes,
+        )
+
+
+def _merge_lists(existing: list | None, new: list | None) -> list | None:
+    """Merge two lists, removing duplicates while preserving order."""
+    if new is None:
+        return existing
+    if existing is None:
+        return new
+    # Preserve order, add new items that don't exist
+    merged = list(existing)
+    for item in new:
+        if item not in merged:
+            merged.append(item)
+    return merged
+
+
+async def _get_from_cache(user_id: str) -> Optional[BusinessUnderstanding]:
+    """Get business understanding from Redis cache."""
+    try:
+        redis = await get_redis_async()
+        cached_data = await redis.get(_cache_key(user_id))
+        if cached_data:
+            return BusinessUnderstanding.model_validate_json(cached_data)
+    except Exception as e:
+        logger.warning(f"Failed to get understanding from cache: {e}")
+    return None
+
+
+async def _set_cache(user_id: str, understanding: BusinessUnderstanding) -> None:
+    """Set business understanding in Redis cache with TTL."""
+    try:
+        redis = await get_redis_async()
+        await redis.setex(
+            _cache_key(user_id),
+            CACHE_TTL_SECONDS,
+            understanding.model_dump_json(),
+        )
+    except Exception as e:
+        logger.warning(f"Failed to set understanding in cache: {e}")
+
+
+async def _delete_cache(user_id: str) -> None:
+    """Delete business understanding from Redis cache."""
+    try:
+        redis = await get_redis_async()
+        await redis.delete(_cache_key(user_id))
+    except Exception as e:
+        logger.warning(f"Failed to delete understanding from cache: {e}")
+
+
+async def get_business_understanding(
+    user_id: str,
+) -> Optional[BusinessUnderstanding]:
+    """Get the business understanding for a user.
+
+    Checks cache first, falls back to database if not cached.
+    Results are cached for 48 hours.
+    """
+    # Try cache first
+    cached = await _get_from_cache(user_id)
+    if cached:
+        logger.debug(f"Business understanding cache hit for user {user_id}")
+        return cached
+
+    # Cache miss - load from database
+    logger.debug(f"Business understanding cache miss for user {user_id}")
+    record = await UserBusinessUnderstanding.prisma().find_unique(
+        where={"userId": user_id}
+    )
+    if record is None:
+        return None
+
+    understanding = BusinessUnderstanding.from_db(record)
+
+    # Store in cache for next time
+    await _set_cache(user_id, understanding)
+
+    return understanding
+
+
+async def upsert_business_understanding(
+    user_id: str,
+    data: BusinessUnderstandingInput,
+) -> BusinessUnderstanding:
+    """
+    Create or update business understanding with incremental merge strategy.
+
+    - String fields: new value overwrites if provided (not None)
+    - List fields: new items are appended to existing (deduplicated)
+    """
+    # Get existing record for merge
+    existing = await UserBusinessUnderstanding.prisma().find_unique(
+        where={"userId": user_id}
+    )
+
+    # Build update data with merge strategy
+    update_data: UserBusinessUnderstandingUpdateInput = {}
+    create_data: dict[str, Any] = {"userId": user_id}
+
+    # String fields - overwrite if provided
+    if data.user_name is not None:
+        update_data["userName"] = data.user_name
+        create_data["userName"] = data.user_name
+    if data.job_title is not None:
+        update_data["jobTitle"] = data.job_title
+        create_data["jobTitle"] = data.job_title
+    if data.business_name is not None:
+        update_data["businessName"] = data.business_name
+        create_data["businessName"] = data.business_name
+    if data.industry is not None:
+        update_data["industry"] = data.industry
+        create_data["industry"] = data.industry
+    if data.business_size is not None:
+        update_data["businessSize"] = data.business_size
+        create_data["businessSize"] = data.business_size
+    if data.user_role is not None:
+        update_data["userRole"] = data.user_role
+        create_data["userRole"] = data.user_role
+    if data.additional_notes is not None:
+        update_data["additionalNotes"] = data.additional_notes
+        create_data["additionalNotes"] = data.additional_notes
+
+    # List fields - merge with existing
+    if data.key_workflows is not None:
+        existing_list = _json_to_list(existing.keyWorkflows) if existing else None
+        merged = _merge_lists(existing_list, data.key_workflows)
+        update_data["keyWorkflows"] = SafeJson(merged)
+        create_data["keyWorkflows"] = SafeJson(merged)
+
+    if data.daily_activities is not None:
+        existing_list = _json_to_list(existing.dailyActivities) if existing else None
+        merged = _merge_lists(existing_list, data.daily_activities)
+        update_data["dailyActivities"] = SafeJson(merged)
+        create_data["dailyActivities"] = SafeJson(merged)
+
+    if data.pain_points is not None:
+        existing_list = _json_to_list(existing.painPoints) if existing else None
+        merged = _merge_lists(existing_list, data.pain_points)
+        update_data["painPoints"] = SafeJson(merged)
+        create_data["painPoints"] = SafeJson(merged)
+
+    if data.bottlenecks is not None:
+        existing_list = _json_to_list(existing.bottlenecks) if existing else None
+        merged = _merge_lists(existing_list, data.bottlenecks)
+        update_data["bottlenecks"] = SafeJson(merged)
+        create_data["bottlenecks"] = SafeJson(merged)
+
+    if data.manual_tasks is not None:
+        existing_list = _json_to_list(existing.manualTasks) if existing else None
+        merged = _merge_lists(existing_list, data.manual_tasks)
+        update_data["manualTasks"] = SafeJson(merged)
+        create_data["manualTasks"] = SafeJson(merged)
+
+    if data.automation_goals is not None:
+        existing_list = _json_to_list(existing.automationGoals) if existing else None
+        merged = _merge_lists(existing_list, data.automation_goals)
+        update_data["automationGoals"] = SafeJson(merged)
+        create_data["automationGoals"] = SafeJson(merged)
+
+    if data.current_software is not None:
+        existing_list = _json_to_list(existing.currentSoftware) if existing else None
+        merged = _merge_lists(existing_list, data.current_software)
+        update_data["currentSoftware"] = SafeJson(merged)
+        create_data["currentSoftware"] = SafeJson(merged)
+
+    if data.existing_automation is not None:
+        existing_list = _json_to_list(existing.existingAutomation) if existing else None
+        merged = _merge_lists(existing_list, data.existing_automation)
+        update_data["existingAutomation"] = SafeJson(merged)
+        create_data["existingAutomation"] = SafeJson(merged)
+
+    # Upsert
+    record = await UserBusinessUnderstanding.prisma().upsert(
+        where={"userId": user_id},
+        data={
+            "create": UserBusinessUnderstandingCreateInput(**create_data),
+            "update": update_data,
+        },
+    )
+
+    understanding = BusinessUnderstanding.from_db(record)
+
+    # Update cache with new understanding
+    await _set_cache(user_id, understanding)
+
+    return understanding
+
+
+async def clear_business_understanding(user_id: str) -> bool:
+    """Clear/delete business understanding for a user from both DB and cache."""
+    # Delete from cache first
+    await _delete_cache(user_id)
+
+    try:
+        await UserBusinessUnderstanding.prisma().delete(where={"userId": user_id})
+        return True
+    except Exception:
+        # Record might not exist
+        return False
+
+
+def format_understanding_for_prompt(understanding: BusinessUnderstanding) -> str:
+    """Format business understanding as text for system prompt injection."""
+    sections = []
+
+    # User info section
+    user_info = []
+    if understanding.user_name:
+        user_info.append(f"Name: {understanding.user_name}")
+    if understanding.job_title:
+        user_info.append(f"Job Title: {understanding.job_title}")
+    if user_info:
+        sections.append("## User\n" + "\n".join(user_info))
+
+    # Business section
+    business_info = []
+    if understanding.business_name:
+        business_info.append(f"Company: {understanding.business_name}")
+    if understanding.industry:
+        business_info.append(f"Industry: {understanding.industry}")
+    if understanding.business_size:
+        business_info.append(f"Size: {understanding.business_size}")
+    if understanding.user_role:
+        business_info.append(f"Role Context: {understanding.user_role}")
+    if business_info:
+        sections.append("## Business\n" + "\n".join(business_info))
+
+    # Processes section
+    processes = []
+    if understanding.key_workflows:
+        processes.append(f"Key Workflows: {', '.join(understanding.key_workflows)}")
+    if understanding.daily_activities:
+        processes.append(
+            f"Daily Activities: {', '.join(understanding.daily_activities)}"
+        )
+    if processes:
+        sections.append("## Processes\n" + "\n".join(processes))
+
+    # Pain points section
+    pain_points = []
+    if understanding.pain_points:
+        pain_points.append(f"Pain Points: {', '.join(understanding.pain_points)}")
+    if understanding.bottlenecks:
+        pain_points.append(f"Bottlenecks: {', '.join(understanding.bottlenecks)}")
+    if understanding.manual_tasks:
+        pain_points.append(f"Manual Tasks: {', '.join(understanding.manual_tasks)}")
+    if pain_points:
+        sections.append("## Pain Points\n" + "\n".join(pain_points))
+
+    # Goals section
+    if understanding.automation_goals:
+        sections.append(
+            "## Automation Goals\n"
+            + "\n".join(f"- {goal}" for goal in understanding.automation_goals)
+        )
+
+    # Current tools section
+    tools_info = []
+    if understanding.current_software:
+        tools_info.append(
+            f"Current Software: {', '.join(understanding.current_software)}"
+        )
+    if understanding.existing_automation:
+        tools_info.append(
+            f"Existing Automation: {', '.join(understanding.existing_automation)}"
+        )
+    if tools_info:
+        sections.append("## Current Tools\n" + "\n".join(tools_info))
+
+    # Additional notes
+    if understanding.additional_notes:
+        sections.append(f"## Additional Context\n{understanding.additional_notes}")
+
+    if not sections:
+        return ""
+
+    return "# User Business Context\n\n" + "\n\n".join(sections)

autogpt_platform/backend/backend/server/v2/chat/db.py

@@ -0,0 +1,194 @@
+"""Database operations for chat sessions."""
+
+import logging
+from datetime import UTC, datetime
+from typing import Any
+
+from prisma.models import ChatMessage as PrismaChatMessage
+from prisma.models import ChatSession as PrismaChatSession
+from prisma.types import (
+    ChatMessageCreateInput,
+    ChatSessionCreateInput,
+    ChatSessionUpdateInput,
+)
+
+from backend.util.json import SafeJson
+
+logger = logging.getLogger(__name__)
+
+
+async def get_chat_session(session_id: str) -> PrismaChatSession | None:
+    """Get a chat session by ID from the database."""
+    session = await PrismaChatSession.prisma().find_unique(
+        where={"id": session_id},
+        include={"Messages": True},
+    )
+    if session and session.Messages:
+        # Sort messages by sequence in Python since Prisma doesn't support order_by in include
+        session.Messages.sort(key=lambda m: m.sequence)
+    return session
+
+
+async def create_chat_session(
+    session_id: str,
+    user_id: str | None,
+) -> PrismaChatSession:
+    """Create a new chat session in the database."""
+    data: ChatSessionCreateInput = {
+        "id": session_id,
+        "userId": user_id,
+        "credentials": SafeJson({}),
+        "successfulAgentRuns": SafeJson({}),
+        "successfulAgentSchedules": SafeJson({}),
+    }
+    return await PrismaChatSession.prisma().create(
+        data=data,
+        include={"Messages": True},
+    )
+
+
+async def update_chat_session(
+    session_id: str,
+    credentials: dict[str, Any] | None = None,
+    successful_agent_runs: dict[str, Any] | None = None,
+    successful_agent_schedules: dict[str, Any] | None = None,
+    total_prompt_tokens: int | None = None,
+    total_completion_tokens: int | None = None,
+    title: str | None = None,
+) -> PrismaChatSession | None:
+    """Update a chat session's metadata."""
+    data: ChatSessionUpdateInput = {"updatedAt": datetime.now(UTC)}
+
+    if credentials is not None:
+        data["credentials"] = SafeJson(credentials)
+    if successful_agent_runs is not None:
+        data["successfulAgentRuns"] = SafeJson(successful_agent_runs)
+    if successful_agent_schedules is not None:
+        data["successfulAgentSchedules"] = SafeJson(successful_agent_schedules)
+    if total_prompt_tokens is not None:
+        data["totalPromptTokens"] = total_prompt_tokens
+    if total_completion_tokens is not None:
+        data["totalCompletionTokens"] = total_completion_tokens
+    if title is not None:
+        data["title"] = title
+
+    session = await PrismaChatSession.prisma().update(
+        where={"id": session_id},
+        data=data,
+        include={"Messages": True},
+    )
+    if session and session.Messages:
+        session.Messages.sort(key=lambda m: m.sequence)
+    return session
+
+
+async def add_chat_message(
+    session_id: str,
+    role: str,
+    sequence: int,
+    content: str | None = None,
+    name: str | None = None,
+    tool_call_id: str | None = None,
+    refusal: str | None = None,
+    tool_calls: list[dict[str, Any]] | None = None,
+    function_call: dict[str, Any] | None = None,
+) -> PrismaChatMessage:
+    """Add a message to a chat session."""
+    data: ChatMessageCreateInput = {
+        "Session": {"connect": {"id": session_id}},
+        "role": role,
+        "sequence": sequence,
+    }
+
+    if content is not None:
+        data["content"] = content
+    if name is not None:
+        data["name"] = name
+    if tool_call_id is not None:
+        data["toolCallId"] = tool_call_id
+    if refusal is not None:
+        data["refusal"] = refusal
+    if tool_calls is not None:
+        data["toolCalls"] = SafeJson(tool_calls)
+    if function_call is not None:
+        data["functionCall"] = SafeJson(function_call)
+
+    # Update session's updatedAt timestamp
+    await PrismaChatSession.prisma().update(
+        where={"id": session_id},
+        data={"updatedAt": datetime.now(UTC)},
+    )
+
+    return await PrismaChatMessage.prisma().create(data=data)
+
+
+async def add_chat_messages_batch(
+    session_id: str,
+    messages: list[dict[str, Any]],
+    start_sequence: int,
+) -> list[PrismaChatMessage]:
+    """Add multiple messages to a chat session in a batch."""
+    if not messages:
+        return []
+
+    created_messages = []
+    for i, msg in enumerate(messages):
+        data: ChatMessageCreateInput = {
+            "Session": {"connect": {"id": session_id}},
+            "role": msg["role"],
+            "sequence": start_sequence + i,
+        }
+
+        if msg.get("content") is not None:
+            data["content"] = msg["content"]
+        if msg.get("name") is not None:
+            data["name"] = msg["name"]
+        if msg.get("tool_call_id") is not None:
+            data["toolCallId"] = msg["tool_call_id"]
+        if msg.get("refusal") is not None:
+            data["refusal"] = msg["refusal"]
+        if msg.get("tool_calls") is not None:
+            data["toolCalls"] = SafeJson(msg["tool_calls"])
+        if msg.get("function_call") is not None:
+            data["functionCall"] = SafeJson(msg["function_call"])
+
+        created = await PrismaChatMessage.prisma().create(data=data)
+        created_messages.append(created)
+
+    # Update session's updatedAt timestamp
+    await PrismaChatSession.prisma().update(
+        where={"id": session_id},
+        data={"updatedAt": datetime.now(UTC)},
+    )
+
+    return created_messages
+
+
+async def get_user_chat_sessions(
+    user_id: str,
+    limit: int = 50,
+    offset: int = 0,
+) -> list[PrismaChatSession]:
+    """Get chat sessions for a user, ordered by most recent."""
+    return await PrismaChatSession.prisma().find_many(
+        where={"userId": user_id},
+        order={"updatedAt": "desc"},
+        take=limit,
+        skip=offset,
+    )
+
+
+async def delete_chat_session(session_id: str) -> bool:
+    """Delete a chat session and all its messages."""
+    try:
+        await PrismaChatSession.prisma().delete(where={"id": session_id})
+        return True
+    except Exception as e:
+        logger.error(f"Failed to delete chat session {session_id}: {e}")
+        return False
+
+
+async def get_chat_session_message_count(session_id: str) -> int:
+    """Get the number of messages in a chat session."""
+    count = await PrismaChatMessage.prisma().count(where={"sessionId": session_id})
+    return count

autogpt_platform/backend/backend/server/v2/chat/model.py

@@ -16,10 +16,14 @@ from openai.types.chat.chat_completion_message_tool_call_param import (
     ChatCompletionMessageToolCallParam,
     Function,
 )
+from prisma.models import ChatMessage as PrismaChatMessage
+from prisma.models import ChatSession as PrismaChatSession
 from pydantic import BaseModel
 
 from backend.data.redis_client import get_redis_async
+from backend.server.v2.chat import db as chat_db
 from backend.server.v2.chat.config import ChatConfig
+from backend.util import json
 from backend.util.exceptions import RedisError
 
 logger = logging.getLogger(__name__)
@@ -65,6 +69,84 @@ class ChatSession(BaseModel):
             updated_at=datetime.now(UTC),
         )
 
+    @staticmethod
+    def from_prisma(
+        prisma_session: PrismaChatSession,
+        prisma_messages: list[PrismaChatMessage] | None = None,
+    ) -> "ChatSession":
+        """Convert Prisma models to Pydantic ChatSession."""
+        messages = []
+        if prisma_messages:
+            for msg in prisma_messages:
+                tool_calls = None
+                if msg.toolCalls:
+                    tool_calls = (
+                        json.loads(msg.toolCalls)
+                        if isinstance(msg.toolCalls, str)
+                        else msg.toolCalls
+                    )
+
+                function_call = None
+                if msg.functionCall:
+                    function_call = (
+                        json.loads(msg.functionCall)
+                        if isinstance(msg.functionCall, str)
+                        else msg.functionCall
+                    )
+
+                messages.append(
+                    ChatMessage(
+                        role=msg.role,
+                        content=msg.content,
+                        name=msg.name,
+                        tool_call_id=msg.toolCallId,
+                        refusal=msg.refusal,
+                        tool_calls=tool_calls,
+                        function_call=function_call,
+                    )
+                )
+
+        # Parse JSON fields from Prisma
+        credentials = (
+            json.loads(prisma_session.credentials)
+            if isinstance(prisma_session.credentials, str)
+            else prisma_session.credentials or {}
+        )
+        successful_agent_runs = (
+            json.loads(prisma_session.successfulAgentRuns)
+            if isinstance(prisma_session.successfulAgentRuns, str)
+            else prisma_session.successfulAgentRuns or {}
+        )
+        successful_agent_schedules = (
+            json.loads(prisma_session.successfulAgentSchedules)
+            if isinstance(prisma_session.successfulAgentSchedules, str)
+            else prisma_session.successfulAgentSchedules or {}
+        )
+
+        # Calculate usage from token counts
+        usage = []
+        if prisma_session.totalPromptTokens or prisma_session.totalCompletionTokens:
+            usage.append(
+                Usage(
+                    prompt_tokens=prisma_session.totalPromptTokens or 0,
+                    completion_tokens=prisma_session.totalCompletionTokens or 0,
+                    total_tokens=(prisma_session.totalPromptTokens or 0)
+                    + (prisma_session.totalCompletionTokens or 0),
+                )
+            )
+
+        return ChatSession(
+            session_id=prisma_session.id,
+            user_id=prisma_session.userId,
+            messages=messages,
+            usage=usage,
+            credentials=credentials,
+            started_at=prisma_session.createdAt,
+            updated_at=prisma_session.updatedAt,
+            successful_agent_runs=successful_agent_runs,
+            successful_agent_schedules=successful_agent_schedules,
+        )
+
     def to_openai_messages(self) -> list[ChatCompletionMessageParam]:
         messages = []
         for message in self.messages:
@@ -154,50 +236,215 @@ class ChatSession(BaseModel):
         return messages
 
 
-async def get_chat_session(
-    session_id: str,
-    user_id: str | None,
-) -> ChatSession | None:
-    """Get a chat session by ID."""
+async def _get_session_from_cache(session_id: str) -> ChatSession | None:
+    """Get a chat session from Redis cache."""
     redis_key = f"chat:session:{session_id}"
     async_redis = await get_redis_async()
-
     raw_session: bytes | None = await async_redis.get(redis_key)
 
     if raw_session is None:
-        logger.warning(f"Session {session_id} not found in Redis")
         return None
 
     try:
-        session = ChatSession.model_validate_json(raw_session)
+        return ChatSession.model_validate_json(raw_session)
     except Exception as e:
         logger.error(f"Failed to deserialize session {session_id}: {e}", exc_info=True)
         raise RedisError(f"Corrupted session data for {session_id}") from e
 
+
+async def _cache_session(session: ChatSession) -> None:
+    """Cache a chat session in Redis."""
+    redis_key = f"chat:session:{session.session_id}"
+    async_redis = await get_redis_async()
+    await async_redis.setex(redis_key, config.session_ttl, session.model_dump_json())
+
+
+async def _get_session_from_db(session_id: str) -> ChatSession | None:
+    """Get a chat session from the database."""
+    prisma_session = await chat_db.get_chat_session(session_id)
+    if not prisma_session:
+        return None
+
+    return ChatSession.from_prisma(prisma_session, prisma_session.Messages)
+
+
+async def _save_session_to_db(
+    session: ChatSession, existing_message_count: int
+) -> None:
+    """Save or update a chat session in the database."""
+    # Check if session exists in DB
+    existing = await chat_db.get_chat_session(session.session_id)
+
+    if not existing:
+        # Create new session
+        await chat_db.create_chat_session(
+            session_id=session.session_id,
+            user_id=session.user_id,
+        )
+        existing_message_count = 0
+
+    # Calculate total tokens from usage
+    total_prompt = sum(u.prompt_tokens for u in session.usage)
+    total_completion = sum(u.completion_tokens for u in session.usage)
+
+    # Update session metadata
+    await chat_db.update_chat_session(
+        session_id=session.session_id,
+        credentials=session.credentials,
+        successful_agent_runs=session.successful_agent_runs,
+        successful_agent_schedules=session.successful_agent_schedules,
+        total_prompt_tokens=total_prompt,
+        total_completion_tokens=total_completion,
+    )
+
+    # Add new messages (only those after existing count)
+    new_messages = session.messages[existing_message_count:]
+    if new_messages:
+        messages_data = []
+        for msg in new_messages:
+            messages_data.append(
+                {
+                    "role": msg.role,
+                    "content": msg.content,
+                    "name": msg.name,
+                    "tool_call_id": msg.tool_call_id,
+                    "refusal": msg.refusal,
+                    "tool_calls": msg.tool_calls,
+                    "function_call": msg.function_call,
+                }
+            )
+        await chat_db.add_chat_messages_batch(
+            session_id=session.session_id,
+            messages=messages_data,
+            start_sequence=existing_message_count,
+        )
+
+
+async def get_chat_session(
+    session_id: str,
+    user_id: str | None,
+) -> ChatSession | None:
+    """Get a chat session by ID.
+
+    Checks Redis cache first, falls back to database if not found.
+    Caches database results back to Redis.
+    """
+    # Try cache first
+    try:
+        session = await _get_session_from_cache(session_id)
+        if session:
+            # Verify user ownership
+            if session.user_id is not None and session.user_id != user_id:
+                logger.warning(
+                    f"Session {session_id} user id mismatch: {session.user_id} != {user_id}"
+                )
+                return None
+            return session
+    except RedisError:
+        logger.warning(f"Cache error for session {session_id}, trying database")
+    except Exception as e:
+        logger.warning(f"Unexpected cache error for session {session_id}: {e}")
+
+    # Fall back to database
+    logger.info(f"Session {session_id} not in cache, checking database")
+    session = await _get_session_from_db(session_id)
+
+    if session is None:
+        logger.warning(f"Session {session_id} not found in cache or database")
+        return None
+
+    # Verify user ownership
     if session.user_id is not None and session.user_id != user_id:
         logger.warning(
             f"Session {session_id} user id mismatch: {session.user_id} != {user_id}"
         )
         return None
 
+    # Cache the session from DB
+    try:
+        await _cache_session(session)
+        logger.info(f"Cached session {session_id} from database")
+    except Exception as e:
+        logger.warning(f"Failed to cache session {session_id}: {e}")
+
     return session
 
 
 async def upsert_chat_session(
     session: ChatSession,
 ) -> ChatSession:
-    """Update a chat session with the given messages."""
-
-    redis_key = f"chat:session:{session.session_id}"
-
-    async_redis = await get_redis_async()
-    resp = await async_redis.setex(
-        redis_key, config.session_ttl, session.model_dump_json()
+    """Update a chat session in both cache and database."""
+    # Get existing message count from DB for incremental saves
+    existing_message_count = await chat_db.get_chat_session_message_count(
+        session.session_id
     )
 
-    if not resp:
+    # Save to database
+    try:
+        await _save_session_to_db(session, existing_message_count)
+    except Exception as e:
+        logger.error(f"Failed to save session {session.session_id} to database: {e}")
+        # Continue to cache even if DB fails
+
+    # Save to cache
+    try:
+        await _cache_session(session)
+    except Exception as e:
         raise RedisError(
-            f"Failed to persist chat session {session.session_id} to Redis: {resp}"
-        )
+            f"Failed to persist chat session {session.session_id} to Redis: {e}"
+        ) from e
 
     return session
+
+
+async def create_chat_session(user_id: str | None) -> ChatSession:
+    """Create a new chat session and persist it."""
+    session = ChatSession.new(user_id)
+
+    # Create in database first
+    try:
+        await chat_db.create_chat_session(
+            session_id=session.session_id,
+            user_id=user_id,
+        )
+    except Exception as e:
+        logger.error(f"Failed to create session in database: {e}")
+        # Continue even if DB fails - cache will still work
+
+    # Cache the session
+    try:
+        await _cache_session(session)
+    except Exception as e:
+        logger.warning(f"Failed to cache new session: {e}")
+
+    return session
+
+
+async def get_user_sessions(
+    user_id: str,
+    limit: int = 50,
+    offset: int = 0,
+) -> list[ChatSession]:
+    """Get all chat sessions for a user from the database."""
+    prisma_sessions = await chat_db.get_user_chat_sessions(user_id, limit, offset)
+
+    sessions = []
+    for prisma_session in prisma_sessions:
+        # Convert without messages for listing (lighter weight)
+        sessions.append(ChatSession.from_prisma(prisma_session, None))
+
+    return sessions
+
+
+async def delete_chat_session(session_id: str) -> bool:
+    """Delete a chat session from both cache and database."""
+    # Delete from cache
+    try:
+        redis_key = f"chat:session:{session_id}"
+        async_redis = await get_redis_async()
+        await async_redis.delete(redis_key)
+    except Exception as e:
+        logger.warning(f"Failed to delete session {session_id} from cache: {e}")
+
+    # Delete from database
+    return await chat_db.delete_chat_session(session_id)

autogpt_platform/backend/backend/server/v2/chat/prompts/chat_system.md

@@ -3,10 +3,38 @@ You are Otto, an AI Co-Pilot and Forward Deployed Engineer for AutoGPT, an AI Bu
 Here are the functions available to you:
 
 <functions>
-1. **find_agent** - Search for agents that solve the user's problem
-2. **run_agent** - Run or schedule an agent (automatically handles setup)
+1. **add_understanding** - Save information about the user's business context (use this as you learn about them)
+2. **find_agent** - Search for agents that solve the user's problem
+3. **run_agent** - Run or schedule an agent (automatically handles setup)
 </functions>
 
+## BUILDING USER UNDERSTANDING
+
+**If no User Business Context is provided below**, you should gather information about the user to better help them. Do this naturally during conversation - don't interrogate them.
+
+**Key information to gather:**
+- Their name and job title
+- Their business/company and industry
+- Key workflows and daily activities
+- Pain points and manual tasks they want to automate
+- Tools they currently use
+
+**How to gather this information:**
+- Ask naturally as part of helping them (e.g., "What's your role?" or "What industry are you in?")
+- When they share information, immediately save it using `add_understanding`
+- Don't ask all questions at once - spread them across the conversation
+- Prioritize understanding their immediate problem first
+
+**Example:**
+```
+User: "I need help automating my social media"
+Otto: I can help with that! To find the best solution, what's your role - are you a social media manager or business owner?
+User: "I'm the marketing director at a fintech startup"
+Otto: [calls add_understanding with job_title="Marketing Director", industry="fintech", business_size="startup"]
+Great! Let me find social media automation agents that work well for fintech marketing.
+[calls find_agent with query="social media automation marketing"]
+```
+
 ## HOW run_agent WORKS
 
 The `run_agent` tool automatically handles the entire setup flow:
@@ -23,16 +51,18 @@ Parameters:
 
 ## WORKFLOW
 
-1. **find_agent** - Search for agents that solve the user's problem
-2. **run_agent** (first call, no inputs) - Get available inputs for the agent
-3. **Ask user** what values they want to use OR if they want to use defaults
-4. **run_agent** (second call) - Either with `inputs={...}` or `use_defaults=true`
+1. **Understand context** - If you don't have user context, ask 1 quick question while helping
+2. **find_agent** - Search for agents that solve the user's problem
+3. **run_agent** (first call, no inputs) - Get available inputs for the agent
+4. **Ask user** what values they want to use OR if they want to use defaults
+5. **run_agent** (second call) - Either with `inputs={...}` or `use_defaults=true`
 
 ## YOUR APPROACH
 
 **Step 1: Understand the Problem**
 - Ask maximum 1-2 targeted questions
 - Focus on: What business problem are they solving?
+- If you lack context about them, weave in a quick question about their role/business
 - Move quickly to searching for solutions
 
 **Step 2: Find Agents**
@@ -54,16 +84,18 @@ Parameters:
 - Add `schedule_name` and `cron` parameters
 - Example: `run_agent(username_agent_slug="...", inputs={...}, schedule_name="Daily Report", cron="0 9 * * *")`
 
-## FUNCTION CALL FORMAT
+## USING add_understanding
 
-To call a function, use this exact format:
-`<function_call>function_name(parameter="value")</function_call>`
+Call `add_understanding` whenever you learn something about the user. You can call it with any subset of fields:
 
-Examples:
-- `<function_call>find_agent(query="social media automation")</function_call>`
-- `<function_call>run_agent(username_agent_slug="creator/agent-name")</function_call>` (get inputs)
-- `<function_call>run_agent(username_agent_slug="creator/agent-name", inputs={"topic": "AI news"})</function_call>`
-- `<function_call>run_agent(username_agent_slug="creator/agent-name", use_defaults=true)</function_call>`
+**User info:** `user_name`, `job_title`
+**Business:** `business_name`, `industry`, `business_size` (1-10, 11-50, 51-200, 201-1000, 1000+), `user_role` (decision maker, implementer, end user)
+**Processes:** `key_workflows` (array), `daily_activities` (array)
+**Pain points:** `pain_points` (array), `bottlenecks` (array), `manual_tasks` (array), `automation_goals` (array)
+**Tools:** `current_software` (array), `existing_automation` (array)
+**Other:** `additional_notes`
+
+Example: `add_understanding(job_title="Marketing Director", industry="fintech", pain_points=["manual reporting", "social media scheduling"])`
 
 ## KEY RULES
 
@@ -73,8 +105,10 @@ Examples:
 - Don't run agents without first showing available inputs to the user
 - Don't use `use_defaults=true` without user explicitly confirming
 - Don't write responses longer than 3 sentences
+- Don't interrogate users with many questions - gather info naturally
 
 **What You DO:**
+- Save user information with `add_understanding` as you learn it
 - Always call run_agent first without inputs to see what's available
 - Ask user what values they want OR if they want to use defaults
 - Keep all responses to maximum 3 sentences
@@ -87,6 +121,7 @@ Examples:
 ## RESPONSE STRUCTURE
 
 Before responding, wrap your analysis in <thinking> tags to systematically plan your approach:
+- Check if you have user context - if not, plan to gather some naturally
 - Extract the key business problem or request from the user's message
 - Determine what function call (if any) you need to make next
 - Plan your response to stay under the 3-sentence maximum

autogpt_platform/backend/backend/server/v2/chat/routes.py

@@ -5,7 +5,7 @@ from collections.abc import AsyncGenerator
 from typing import Annotated
 
 from autogpt_libs import auth
-from fastapi import APIRouter, Depends, Query, Request, Security
+from fastapi import APIRouter, Depends, Query, Security
 from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
@@ -27,6 +27,7 @@ router = APIRouter(
 
 class StreamChatRequest(BaseModel):
     """Request model for streaming chat with optional context."""
+
     message: str
     is_user_message: bool = True
     context: dict[str, str] | None = None  # {url: str, content: str}
@@ -50,9 +51,64 @@ class SessionDetailResponse(BaseModel):
     messages: list[dict]
 
 
+class SessionSummaryResponse(BaseModel):
+    """Response model for a session summary (without messages)."""
+
+    id: str
+    created_at: str
+    updated_at: str
+    title: str | None = None
+
+
+class ListSessionsResponse(BaseModel):
+    """Response model for listing chat sessions."""
+
+    sessions: list[SessionSummaryResponse]
+    total: int
+
+
 # ========== Routes ==========
 
 
+@router.get(
+    "/sessions",
+    dependencies=[Security(auth.requires_user)],
+)
+async def list_sessions(
+    user_id: Annotated[str, Security(auth.get_user_id)],
+    limit: int = Query(default=50, ge=1, le=100),
+    offset: int = Query(default=0, ge=0),
+) -> ListSessionsResponse:
+    """
+    List chat sessions for the authenticated user.
+
+    Returns a paginated list of chat sessions belonging to the current user,
+    ordered by most recently updated.
+
+    Args:
+        user_id: The authenticated user's ID.
+        limit: Maximum number of sessions to return (1-100).
+        offset: Number of sessions to skip for pagination.
+
+    Returns:
+        ListSessionsResponse: List of session summaries and total count.
+    """
+    sessions = await chat_service.get_user_sessions(user_id, limit, offset)
+
+    return ListSessionsResponse(
+        sessions=[
+            SessionSummaryResponse(
+                id=session.session_id,
+                created_at=session.started_at.isoformat(),
+                updated_at=session.updated_at.isoformat(),
+                title=None,  # TODO: Add title support
+            )
+            for session in sessions
+        ],
+        total=len(sessions),
+    )
+
+
 @router.post(
     "/sessions",
 )

autogpt_platform/backend/backend/server/v2/chat/service.py

@@ -8,13 +8,15 @@ from openai import AsyncOpenAI
 from openai.types.chat import ChatCompletionChunk, ChatCompletionToolParam
 
 import backend.server.v2.chat.config
-from backend.server.v2.chat.model import (
-    ChatMessage,
-    ChatSession,
-    Usage,
-    get_chat_session,
-    upsert_chat_session,
+from backend.data.understanding import (
+    format_understanding_for_prompt,
+    get_business_understanding,
 )
+from backend.server.v2.chat.model import ChatMessage, ChatSession, Usage
+from backend.server.v2.chat.model import (
+    create_chat_session as model_create_chat_session,
+)
+from backend.server.v2.chat.model import get_chat_session, upsert_chat_session
 from backend.server.v2.chat.response_model import (
     StreamBaseResponse,
     StreamEnd,
@@ -35,15 +37,38 @@ config = backend.server.v2.chat.config.ChatConfig()
 client = AsyncOpenAI(api_key=config.api_key, base_url=config.base_url)
 
 
+async def _build_system_prompt(user_id: str | None) -> str:
+    """Build the full system prompt including business understanding if available."""
+    # Start with the base system prompt
+    base_prompt = config.get_system_prompt()
+
+    # If user is authenticated, try to fetch their business understanding
+    if user_id:
+        try:
+            understanding = await get_business_understanding(user_id)
+            if understanding:
+                context = format_understanding_for_prompt(understanding)
+                if context:
+                    return (
+                        f"{base_prompt}\n\n---\n\n"
+                        f"{context}\n\n"
+                        "Use this context to provide more personalized recommendations "
+                        "and to better understand the user's business needs when "
+                        "suggesting agents and automations."
+                    )
+        except Exception as e:
+            logger.warning(f"Failed to fetch business understanding: {e}")
+
+    return base_prompt
+
+
 async def create_chat_session(
     user_id: str | None = None,
 ) -> ChatSession:
     """
     Create a new chat session and persist it to the database.
     """
-    session = ChatSession.new(user_id)
-    # Persist the session immediately so it can be used for streaming
-    return await upsert_chat_session(session)
+    return await model_create_chat_session(user_id)
 
 
 async def get_session(
@@ -56,6 +81,21 @@ async def get_session(
     return await get_chat_session(session_id, user_id)
 
 
+async def get_user_sessions(
+    user_id: str,
+    limit: int = 50,
+    offset: int = 0,
+) -> list[ChatSession]:
+    """
+    Get all chat sessions for a user.
+    """
+    from backend.server.v2.chat.model import (
+        get_user_sessions as model_get_user_sessions,
+    )
+
+    return await model_get_user_sessions(user_id, limit, offset)
+
+
 async def assign_user_to_session(
     session_id: str,
     user_id: str,
@@ -150,6 +190,9 @@ async def stream_chat_completion(
     session = await upsert_chat_session(session)
     assert session, "Session not found"
 
+    # Build system prompt with business understanding
+    system_prompt = await _build_system_prompt(user_id)
+
     assistant_response = ChatMessage(
         role="assistant",
         content="",
@@ -168,6 +211,7 @@ async def stream_chat_completion(
         async for chunk in _stream_chat_chunks(
             session=session,
             tools=tools,
+            system_prompt=system_prompt,
         ):
 
             if isinstance(chunk, StreamTextChunk):
@@ -340,6 +384,7 @@ async def stream_chat_completion(
 async def _stream_chat_chunks(
     session: ChatSession,
     tools: list[ChatCompletionToolParam],
+    system_prompt: str | None = None,
 ) -> AsyncGenerator[StreamBaseResponse, None]:
     """
     Pure streaming function for OpenAI chat completions with tool calling.
@@ -347,9 +392,9 @@ async def _stream_chat_chunks(
     This function is database-agnostic and focuses only on streaming logic.
 
     Args:
-        messages: Conversation context as ChatCompletionMessageParam list
-        session_id: Session ID
-        user_id: User ID for tool execution
+        session: Chat session with conversation history
+        tools: Available tools for the model
+        system_prompt: System prompt to prepend to messages
 
     Yields:
         SSE formatted JSON response objects
@@ -359,6 +404,17 @@ async def _stream_chat_chunks(
 
     logger.info("Starting pure chat stream")
 
+    # Build messages with system prompt prepended
+    messages = session.to_openai_messages()
+    if system_prompt:
+        from openai.types.chat import ChatCompletionSystemMessageParam
+
+        system_message = ChatCompletionSystemMessageParam(
+            role="system",
+            content=system_prompt,
+        )
+        messages = [system_message] + messages
+
     # Loop to handle tool calls and continue conversation
     while True:
         try:
@@ -367,7 +423,7 @@ async def _stream_chat_chunks(
             # Create the stream with proper types
             stream = await client.chat.completions.create(
                 model=model,
-                messages=session.to_openai_messages(),
+                messages=messages,
                 tools=tools,
                 tool_choice="auto",
                 stream=True,

autogpt_platform/backend/backend/server/v2/chat/tools/__init__.py

@@ -4,28 +4,37 @@ from openai.types.chat import ChatCompletionToolParam
 
 from backend.server.v2.chat.model import ChatSession
 
+from .add_understanding import AddUnderstandingTool
 from .agent_output import AgentOutputTool
 from .base import BaseTool
 from .find_agent import FindAgentTool
+from .find_block import FindBlockTool
 from .find_library_agent import FindLibraryAgentTool
 from .run_agent import RunAgentTool
+from .run_block import RunBlockTool
 from .search_docs import SearchDocsTool
 
 if TYPE_CHECKING:
     from backend.server.v2.chat.response_model import StreamToolExecutionResult
 
 # Initialize tool instances
+add_understanding_tool = AddUnderstandingTool()
 find_agent_tool = FindAgentTool()
+find_block_tool = FindBlockTool()
 find_library_agent_tool = FindLibraryAgentTool()
 run_agent_tool = RunAgentTool()
+run_block_tool = RunBlockTool()
 search_docs_tool = SearchDocsTool()
 agent_output_tool = AgentOutputTool()
 
 # Export tools as OpenAI format
 tools: list[ChatCompletionToolParam] = [
+    add_understanding_tool.as_openai_tool(),
     find_agent_tool.as_openai_tool(),
+    find_block_tool.as_openai_tool(),
     find_library_agent_tool.as_openai_tool(),
     run_agent_tool.as_openai_tool(),
+    run_block_tool.as_openai_tool(),
     search_docs_tool.as_openai_tool(),
     agent_output_tool.as_openai_tool(),
 ]
@@ -40,9 +49,12 @@ async def execute_tool(
 ) -> "StreamToolExecutionResult":
 
     tool_map: dict[str, BaseTool] = {
+        "add_understanding": add_understanding_tool,
         "find_agent": find_agent_tool,
+        "find_block": find_block_tool,
         "find_library_agent": find_library_agent_tool,
         "run_agent": run_agent_tool,
+        "run_block": run_block_tool,
         "search_platform_docs": search_docs_tool,
         "agent_output": agent_output_tool,
     }

autogpt_platform/backend/backend/server/v2/chat/tools/add_understanding.py

@@ -0,0 +1,205 @@
+"""Tool for capturing user business understanding incrementally."""
+
+import logging
+from typing import Any
+
+from backend.data.understanding import (
+    BusinessUnderstandingInput,
+    upsert_business_understanding,
+)
+from backend.server.v2.chat.model import ChatSession
+from backend.server.v2.chat.tools.base import BaseTool
+from backend.server.v2.chat.tools.models import (
+    ErrorResponse,
+    ToolResponseBase,
+    UnderstandingUpdatedResponse,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class AddUnderstandingTool(BaseTool):
+    """Tool for capturing user's business understanding incrementally."""
+
+    @property
+    def name(self) -> str:
+        return "add_understanding"
+
+    @property
+    def description(self) -> str:
+        return """Capture and store information about the user's business context,
+workflows, pain points, and automation goals. Call this tool whenever the user
+shares information about their business. Each call incrementally adds to the
+existing understanding - you don't need to provide all fields at once.
+
+Use this to build a comprehensive profile that helps recommend better agents
+and automations for the user's specific needs."""
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "user_name": {
+                    "type": "string",
+                    "description": "The user's name",
+                },
+                "job_title": {
+                    "type": "string",
+                    "description": "The user's job title (e.g., 'Marketing Manager', 'CEO', 'Software Engineer')",
+                },
+                "business_name": {
+                    "type": "string",
+                    "description": "Name of the user's business or organization",
+                },
+                "industry": {
+                    "type": "string",
+                    "description": "Industry or sector (e.g., 'e-commerce', 'healthcare', 'finance')",
+                },
+                "business_size": {
+                    "type": "string",
+                    "description": "Company size: '1-10', '11-50', '51-200', '201-1000', or '1000+'",
+                },
+                "user_role": {
+                    "type": "string",
+                    "description": "User's role in organization context (e.g., 'decision maker', 'implementer', 'end user')",
+                },
+                "key_workflows": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Key business workflows (e.g., 'lead qualification', 'content publishing')",
+                },
+                "daily_activities": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Regular daily activities the user performs",
+                },
+                "pain_points": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Current pain points or challenges",
+                },
+                "bottlenecks": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Process bottlenecks slowing things down",
+                },
+                "manual_tasks": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Manual or repetitive tasks that could be automated",
+                },
+                "automation_goals": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Desired automation outcomes or goals",
+                },
+                "current_software": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Software and tools currently in use",
+                },
+                "existing_automation": {
+                    "type": "array",
+                    "items": {"type": "string"},
+                    "description": "Any existing automations or integrations",
+                },
+                "additional_notes": {
+                    "type": "string",
+                    "description": "Any other relevant context or notes",
+                },
+            },
+            "required": [],
+        }
+
+    @property
+    def requires_auth(self) -> bool:
+        """Requires authentication to store user-specific data."""
+        return True
+
+    async def _execute(
+        self,
+        user_id: str | None,
+        session: ChatSession,
+        **kwargs,
+    ) -> ToolResponseBase:
+        """
+        Capture and store business understanding incrementally.
+
+        Each call merges new data with existing understanding:
+        - String fields are overwritten if provided
+        - List fields are appended (with deduplication)
+        """
+        session_id = session.session_id
+
+        if not user_id:
+            return ErrorResponse(
+                message="Authentication required to save business understanding.",
+                session_id=session_id,
+            )
+
+        # Check if any data was provided
+        if not any(v is not None for v in kwargs.values()):
+            return ErrorResponse(
+                message="Please provide at least one field to update.",
+                session_id=session_id,
+            )
+
+        # Build input model
+        input_data = BusinessUnderstandingInput(
+            user_name=kwargs.get("user_name"),
+            job_title=kwargs.get("job_title"),
+            business_name=kwargs.get("business_name"),
+            industry=kwargs.get("industry"),
+            business_size=kwargs.get("business_size"),
+            user_role=kwargs.get("user_role"),
+            key_workflows=kwargs.get("key_workflows"),
+            daily_activities=kwargs.get("daily_activities"),
+            pain_points=kwargs.get("pain_points"),
+            bottlenecks=kwargs.get("bottlenecks"),
+            manual_tasks=kwargs.get("manual_tasks"),
+            automation_goals=kwargs.get("automation_goals"),
+            current_software=kwargs.get("current_software"),
+            existing_automation=kwargs.get("existing_automation"),
+            additional_notes=kwargs.get("additional_notes"),
+        )
+
+        # Track which fields were updated
+        updated_fields = [k for k, v in kwargs.items() if v is not None]
+
+        # Upsert with merge
+        understanding = await upsert_business_understanding(user_id, input_data)
+
+        # Build current understanding summary for the response
+        current_understanding = {
+            "user_name": understanding.user_name,
+            "job_title": understanding.job_title,
+            "business_name": understanding.business_name,
+            "industry": understanding.industry,
+            "business_size": understanding.business_size,
+            "user_role": understanding.user_role,
+            "key_workflows": understanding.key_workflows,
+            "daily_activities": understanding.daily_activities,
+            "pain_points": understanding.pain_points,
+            "bottlenecks": understanding.bottlenecks,
+            "manual_tasks": understanding.manual_tasks,
+            "automation_goals": understanding.automation_goals,
+            "current_software": understanding.current_software,
+            "existing_automation": understanding.existing_automation,
+            "additional_notes": understanding.additional_notes,
+        }
+
+        # Filter out empty values for cleaner response
+        current_understanding = {
+            k: v
+            for k, v in current_understanding.items()
+            if v is not None and v != [] and v != ""
+        }
+
+        return UnderstandingUpdatedResponse(
+            message=f"Updated understanding with: {', '.join(updated_fields)}. "
+            "I now have a better picture of your business context.",
+            session_id=session_id,
+            updated_fields=updated_fields,
+            current_understanding=current_understanding,
+        )

autogpt_platform/backend/backend/server/v2/chat/tools/find_block.py

@@ -0,0 +1,177 @@
+"""Tool for searching available blocks."""
+
+import logging
+from typing import Any
+
+from backend.blocks import load_all_blocks
+from backend.server.v2.chat.model import ChatSession
+from backend.server.v2.chat.tools.base import BaseTool
+from backend.server.v2.chat.tools.models import (
+    BlockInfoSummary,
+    BlockListResponse,
+    ErrorResponse,
+    NoResultsResponse,
+    ToolResponseBase,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class FindBlockTool(BaseTool):
+    """Tool for searching available blocks."""
+
+    @property
+    def name(self) -> str:
+        return "find_block"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Search for available blocks by name or description. "
+            "Blocks are reusable components that perform specific tasks like "
+            "sending emails, making API calls, processing text, etc. "
+            "Use this to find blocks that can be executed directly."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "query": {
+                    "type": "string",
+                    "description": (
+                        "Search query to find blocks by name or description. "
+                        "Use keywords like 'email', 'http', 'text', 'ai', etc."
+                    ),
+                },
+            },
+            "required": ["query"],
+        }
+
+    @property
+    def requires_auth(self) -> bool:
+        return True
+
+    def _matches_query(self, block, query: str) -> tuple[int, bool]:
+        """
+        Check if a block matches the query and return a priority score.
+
+        Returns (priority, matches) where:
+        - priority 0: exact name match
+        - priority 1: name contains query
+        - priority 2: description contains query
+        - priority 3: category contains query
+        """
+        query_lower = query.lower()
+        name_lower = block.name.lower()
+        desc_lower = block.description.lower()
+
+        # Exact name match
+        if query_lower == name_lower:
+            return 0, True
+
+        # Name contains query
+        if query_lower in name_lower:
+            return 1, True
+
+        # Description contains query
+        if query_lower in desc_lower:
+            return 2, True
+
+        # Category contains query
+        for category in block.categories:
+            if query_lower in category.name.lower():
+                return 3, True
+
+        return 4, False
+
+    async def _execute(
+        self,
+        user_id: str | None,
+        session: ChatSession,
+        **kwargs,
+    ) -> ToolResponseBase:
+        """Search for blocks matching the query.
+
+        Args:
+            user_id: User ID (required)
+            session: Chat session
+            query: Search query
+
+        Returns:
+            BlockListResponse: List of matching blocks
+            NoResultsResponse: No blocks found
+            ErrorResponse: Error message
+        """
+        query = kwargs.get("query", "").strip()
+        session_id = session.session_id
+
+        if not query:
+            return ErrorResponse(
+                message="Please provide a search query",
+                session_id=session_id,
+            )
+
+        try:
+            all_blocks = load_all_blocks()
+            logger.info(f"Searching {len(all_blocks)} blocks for: {query}")
+
+            # Find matching blocks with priority scores
+            matches: list[tuple[int, Any]] = []
+            for block_id, block_cls in all_blocks.items():
+                block = block_cls()
+                priority, is_match = self._matches_query(block, query)
+                if is_match:
+                    matches.append((priority, block))
+
+            # Sort by priority (lower is better)
+            matches.sort(key=lambda x: x[0])
+
+            # Take top 10 results
+            top_matches = [block for _, block in matches[:10]]
+
+            if not top_matches:
+                return NoResultsResponse(
+                    message=f"No blocks found matching '{query}'",
+                    session_id=session_id,
+                    suggestions=[
+                        "Try more general terms",
+                        "Search by category: ai, text, social, search, etc.",
+                        "Check block names like 'SendEmail', 'HttpRequest', etc.",
+                    ],
+                )
+
+            # Build response
+            blocks = []
+            for block in top_matches:
+                blocks.append(
+                    BlockInfoSummary(
+                        id=block.id,
+                        name=block.name,
+                        description=block.description,
+                        categories=[cat.name for cat in block.categories],
+                        input_schema=block.input_schema.jsonschema(),
+                        output_schema=block.output_schema.jsonschema(),
+                    )
+                )
+
+            return BlockListResponse(
+                message=(
+                    f"Found {len(blocks)} block{'s' if len(blocks) != 1 else ''} "
+                    f"matching '{query}'. Use run_block to execute a block with "
+                    "the required inputs."
+                ),
+                blocks=blocks,
+                count=len(blocks),
+                query=query,
+                session_id=session_id,
+            )
+
+        except Exception as e:
+            logger.error(f"Error searching blocks: {e}", exc_info=True)
+            return ErrorResponse(
+                message="Failed to search blocks. Please try again.",
+                error=str(e),
+                session_id=session_id,
+            )

autogpt_platform/backend/backend/server/v2/chat/tools/models.py

@@ -22,6 +22,9 @@ class ResponseType(str, Enum):
     SUCCESS = "success"
     DOC_SEARCH_RESULTS = "doc_search_results"
     AGENT_OUTPUT = "agent_output"
+    BLOCK_LIST = "block_list"
+    BLOCK_OUTPUT = "block_output"
+    UNDERSTANDING_UPDATED = "understanding_updated"
 
 
 # Base response model
@@ -223,3 +226,43 @@ class AgentOutputResponse(ToolResponseBase):
     execution: ExecutionOutputInfo | None = None
     available_executions: list[dict[str, Any]] | None = None
     total_executions: int = 0
+
+
+# Block models
+class BlockInfoSummary(BaseModel):
+    """Summary of a block for search results."""
+
+    id: str
+    name: str
+    description: str
+    categories: list[str]
+    input_schema: dict[str, Any]
+    output_schema: dict[str, Any]
+
+
+class BlockListResponse(ToolResponseBase):
+    """Response for find_block tool."""
+
+    type: ResponseType = ResponseType.BLOCK_LIST
+    blocks: list[BlockInfoSummary]
+    count: int
+    query: str
+
+
+class BlockOutputResponse(ToolResponseBase):
+    """Response for run_block tool."""
+
+    type: ResponseType = ResponseType.BLOCK_OUTPUT
+    block_id: str
+    block_name: str
+    outputs: dict[str, list[Any]]
+    success: bool = True
+
+
+# Business understanding models
+class UnderstandingUpdatedResponse(ToolResponseBase):
+    """Response for add_understanding tool."""
+
+    type: ResponseType = ResponseType.UNDERSTANDING_UPDATED
+    updated_fields: list[str] = Field(default_factory=list)
+    current_understanding: dict[str, Any] = Field(default_factory=dict)

autogpt_platform/backend/backend/server/v2/chat/tools/run_block.py

@@ -0,0 +1,270 @@
+"""Tool for executing blocks directly."""
+
+import logging
+from collections import defaultdict
+from typing import Any
+
+from backend.data.block import get_block
+from backend.data.model import CredentialsMetaInput
+from backend.integrations.creds_manager import IntegrationCredentialsManager
+from backend.server.v2.chat.model import ChatSession
+from backend.server.v2.chat.tools.base import BaseTool
+from backend.server.v2.chat.tools.models import (
+    BlockOutputResponse,
+    ErrorResponse,
+    SetupInfo,
+    SetupRequirementsResponse,
+    ToolResponseBase,
+    UserReadiness,
+)
+from backend.util.exceptions import BlockError
+
+logger = logging.getLogger(__name__)
+
+
+class RunBlockTool(BaseTool):
+    """Tool for executing a block and returning its outputs."""
+
+    @property
+    def name(self) -> str:
+        return "run_block"
+
+    @property
+    def description(self) -> str:
+        return (
+            "Execute a specific block with the provided input data. "
+            "Use find_block to discover available blocks and their input schemas. "
+            "The block will run and return its outputs once complete."
+        )
+
+    @property
+    def parameters(self) -> dict[str, Any]:
+        return {
+            "type": "object",
+            "properties": {
+                "block_id": {
+                    "type": "string",
+                    "description": "The UUID of the block to execute",
+                },
+                "input_data": {
+                    "type": "object",
+                    "description": (
+                        "Input values for the block. Must match the block's input schema. "
+                        "Check the block's input_schema from find_block for required fields."
+                    ),
+                },
+            },
+            "required": ["block_id", "input_data"],
+        }
+
+    @property
+    def requires_auth(self) -> bool:
+        return True
+
+    async def _check_block_credentials(
+        self,
+        user_id: str,
+        block: Any,
+    ) -> tuple[dict[str, CredentialsMetaInput], list[CredentialsMetaInput]]:
+        """
+        Check if user has required credentials for a block.
+
+        Returns:
+            tuple[matched_credentials, missing_credentials]
+        """
+        matched_credentials: dict[str, CredentialsMetaInput] = {}
+        missing_credentials: list[CredentialsMetaInput] = []
+
+        # Get credential field info from block's input schema
+        credentials_fields_info = block.input_schema.get_credentials_fields_info()
+
+        if not credentials_fields_info:
+            return matched_credentials, missing_credentials
+
+        # Get user's available credentials
+        creds_manager = IntegrationCredentialsManager()
+        available_creds = await creds_manager.store.get_all_creds(user_id)
+
+        for field_name, field_info in credentials_fields_info.items():
+            # field_info.provider is a frozenset of acceptable providers
+            # field_info.supported_types is a frozenset of acceptable types
+            matching_cred = next(
+                (
+                    cred
+                    for cred in available_creds
+                    if cred.provider in field_info.provider
+                    and cred.type in field_info.supported_types
+                ),
+                None,
+            )
+
+            if matching_cred:
+                matched_credentials[field_name] = CredentialsMetaInput(
+                    id=matching_cred.id,
+                    provider=matching_cred.provider,  # type: ignore
+                    type=matching_cred.type,
+                    title=matching_cred.title,
+                )
+            else:
+                # Create a placeholder for the missing credential
+                provider = next(iter(field_info.provider), "unknown")
+                cred_type = next(iter(field_info.supported_types), "api_key")
+                missing_credentials.append(
+                    CredentialsMetaInput(
+                        id=field_name,
+                        provider=provider,  # type: ignore
+                        type=cred_type,  # type: ignore
+                        title=field_name.replace("_", " ").title(),
+                    )
+                )
+
+        return matched_credentials, missing_credentials
+
+    async def _execute(
+        self,
+        user_id: str | None,
+        session: ChatSession,
+        **kwargs,
+    ) -> ToolResponseBase:
+        """Execute a block with the given input data.
+
+        Args:
+            user_id: User ID (required)
+            session: Chat session
+            block_id: Block UUID to execute
+            input_data: Input values for the block
+
+        Returns:
+            BlockOutputResponse: Block execution outputs
+            SetupRequirementsResponse: Missing credentials
+            ErrorResponse: Error message
+        """
+        block_id = kwargs.get("block_id", "").strip()
+        input_data = kwargs.get("input_data", {})
+        session_id = session.session_id
+
+        if not block_id:
+            return ErrorResponse(
+                message="Please provide a block_id",
+                session_id=session_id,
+            )
+
+        if not isinstance(input_data, dict):
+            return ErrorResponse(
+                message="input_data must be an object",
+                session_id=session_id,
+            )
+
+        if not user_id:
+            return ErrorResponse(
+                message="Authentication required",
+                session_id=session_id,
+            )
+
+        # Get the block
+        block = get_block(block_id)
+        if not block:
+            return ErrorResponse(
+                message=f"Block '{block_id}' not found",
+                session_id=session_id,
+            )
+
+        logger.info(f"Executing block {block.name} ({block_id}) for user {user_id}")
+
+        # Check credentials
+        matched_credentials, missing_credentials = await self._check_block_credentials(
+            user_id, block
+        )
+
+        if missing_credentials:
+            # Return setup requirements response with missing credentials
+            missing_creds_dict = {c.id: c.model_dump() for c in missing_credentials}
+
+            return SetupRequirementsResponse(
+                message=(
+                    f"Block '{block.name}' requires credentials that are not configured. "
+                    "Please set up the required credentials before running this block."
+                ),
+                session_id=session_id,
+                setup_info=SetupInfo(
+                    agent_id=block_id,
+                    agent_name=block.name,
+                    user_readiness=UserReadiness(
+                        has_all_credentials=False,
+                        missing_credentials=missing_creds_dict,
+                        ready_to_run=False,
+                    ),
+                    requirements={
+                        "credentials": [c.model_dump() for c in missing_credentials],
+                        "inputs": self._get_inputs_list(block),
+                        "execution_modes": ["immediate"],
+                    },
+                ),
+                graph_id=None,
+                graph_version=None,
+            )
+
+        try:
+            # Inject matched credentials into input_data
+            for field_name, cred_meta in matched_credentials.items():
+                if field_name not in input_data:
+                    input_data[field_name] = cred_meta.model_dump()
+
+            # Execute the block and collect outputs
+            outputs: dict[str, list[Any]] = defaultdict(list)
+            async for output_name, output_data in block.execute(
+                input_data,
+                user_id=user_id,
+            ):
+                outputs[output_name].append(output_data)
+
+            return BlockOutputResponse(
+                message=f"Block '{block.name}' executed successfully",
+                block_id=block_id,
+                block_name=block.name,
+                outputs=dict(outputs),
+                success=True,
+                session_id=session_id,
+            )
+
+        except BlockError as e:
+            logger.warning(f"Block execution failed: {e}")
+            return ErrorResponse(
+                message=f"Block execution failed: {e}",
+                error=str(e),
+                session_id=session_id,
+            )
+        except Exception as e:
+            logger.error(f"Unexpected error executing block: {e}", exc_info=True)
+            return ErrorResponse(
+                message=f"Failed to execute block: {str(e)}",
+                error=str(e),
+                session_id=session_id,
+            )
+
+    def _get_inputs_list(self, block: Any) -> list[dict[str, Any]]:
+        """Extract non-credential inputs from block schema."""
+        inputs_list = []
+        schema = block.input_schema.jsonschema()
+        properties = schema.get("properties", {})
+        required_fields = set(schema.get("required", []))
+
+        # Get credential field names to exclude
+        credentials_fields = set(block.input_schema.get_credentials_fields().keys())
+
+        for field_name, field_schema in properties.items():
+            # Skip credential fields
+            if field_name in credentials_fields:
+                continue
+
+            inputs_list.append(
+                {
+                    "name": field_name,
+                    "title": field_schema.get("title", field_name),
+                    "type": field_schema.get("type", "string"),
+                    "description": field_schema.get("description", ""),
+                    "required": field_name in required_fields,
+                }
+            )
+
+        return inputs_list

autogpt_platform/backend/migrations/20251216153446_add_chat_tables/migration.sql

@@ -0,0 +1,78 @@
+-- CreateTable
+CREATE TABLE "UserBusinessUnderstanding" (
+    "id" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "userId" TEXT NOT NULL,
+    "userName" TEXT,
+    "jobTitle" TEXT,
+    "businessName" TEXT,
+    "industry" TEXT,
+    "businessSize" TEXT,
+    "userRole" TEXT,
+    "keyWorkflows" JSONB,
+    "dailyActivities" JSONB,
+    "painPoints" JSONB,
+    "bottlenecks" JSONB,
+    "manualTasks" JSONB,
+    "automationGoals" JSONB,
+    "currentSoftware" JSONB,
+    "existingAutomation" JSONB,
+    "additionalNotes" TEXT,
+
+    CONSTRAINT "UserBusinessUnderstanding_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "ChatSession" (
+    "id" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "userId" TEXT,
+    "title" TEXT,
+    "credentials" JSONB NOT NULL DEFAULT '{}',
+    "successfulAgentRuns" JSONB NOT NULL DEFAULT '{}',
+    "successfulAgentSchedules" JSONB NOT NULL DEFAULT '{}',
+    "totalPromptTokens" INTEGER NOT NULL DEFAULT 0,
+    "totalCompletionTokens" INTEGER NOT NULL DEFAULT 0,
+
+    CONSTRAINT "ChatSession_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateTable
+CREATE TABLE "ChatMessage" (
+    "id" TEXT NOT NULL,
+    "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "sessionId" TEXT NOT NULL,
+    "role" TEXT NOT NULL,
+    "content" TEXT,
+    "name" TEXT,
+    "toolCallId" TEXT,
+    "refusal" TEXT,
+    "toolCalls" JSONB,
+    "functionCall" JSONB,
+    "sequence" INTEGER NOT NULL,
+
+    CONSTRAINT "ChatMessage_pkey" PRIMARY KEY ("id")
+);
+
+-- CreateIndex
+CREATE UNIQUE INDEX "UserBusinessUnderstanding_userId_key" ON "UserBusinessUnderstanding"("userId");
+
+-- CreateIndex
+CREATE INDEX "UserBusinessUnderstanding_userId_idx" ON "UserBusinessUnderstanding"("userId");
+
+-- CreateIndex
+CREATE INDEX "ChatSession_userId_updatedAt_idx" ON "ChatSession"("userId", "updatedAt");
+
+-- CreateIndex
+CREATE INDEX "ChatMessage_sessionId_sequence_idx" ON "ChatMessage"("sessionId", "sequence");
+
+-- CreateIndex
+CREATE UNIQUE INDEX "ChatMessage_sessionId_sequence_key" ON "ChatMessage"("sessionId", "sequence");
+
+-- AddForeignKey
+ALTER TABLE "UserBusinessUnderstanding" ADD CONSTRAINT "UserBusinessUnderstanding_userId_fkey" FOREIGN KEY ("userId") REFERENCES "User"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- AddForeignKey
+ALTER TABLE "ChatMessage" ADD CONSTRAINT "ChatMessage_sessionId_fkey" FOREIGN KEY ("sessionId") REFERENCES "ChatSession"("id") ON DELETE CASCADE ON UPDATE CASCADE;

autogpt_platform/backend/schema.prisma

@@ -53,6 +53,7 @@ model User {
 
   Profile               Profile[]
   UserOnboarding        UserOnboarding?
+  BusinessUnderstanding UserBusinessUnderstanding?
   BuilderSearchHistory  BuilderSearchHistory[]
   StoreListings         StoreListing[]
   StoreListingReviews   StoreListingReview[]
@@ -115,6 +116,43 @@ model UserOnboarding {
   User   User   @relation(fields: [userId], references: [id], onDelete: Cascade)
 }
 
+model UserBusinessUnderstanding {
+  id        String   @id @default(uuid())
+  createdAt DateTime @default(now())
+  updatedAt DateTime @default(now()) @updatedAt
+
+  userId String @unique
+  User   User   @relation(fields: [userId], references: [id], onDelete: Cascade)
+
+  // User info
+  userName String?
+  jobTitle String?
+
+  // Business basics (string columns)
+  businessName String?
+  industry     String?
+  businessSize String? // "1-10", "11-50", "51-200", "201-1000", "1000+"
+  userRole     String? // Role in organization context (e.g., "decision maker", "implementer")
+
+  // Processes & activities (JSON arrays)
+  keyWorkflows    Json?
+  dailyActivities Json?
+
+  // Pain points & goals (JSON arrays)
+  painPoints      Json?
+  bottlenecks     Json?
+  manualTasks     Json?
+  automationGoals Json?
+
+  // Current tools (JSON arrays)
+  currentSoftware    Json?
+  existingAutomation Json?
+
+  additionalNotes String?
+
+  @@index([userId])
+}
+
 model BuilderSearchHistory {
   id        String   @id @default(uuid())
   createdAt DateTime @default(now())
@@ -128,6 +166,59 @@ model BuilderSearchHistory {
   User   User   @relation(fields: [userId], references: [id], onDelete: Cascade)
 }
 
+////////////////////////////////////////////////////////////
+////////////////////////////////////////////////////////////
+////////////////   CHAT SESSION TABLES   ///////////////////
+////////////////////////////////////////////////////////////
+////////////////////////////////////////////////////////////
+
+model ChatSession {
+  id        String   @id @default(uuid())
+  createdAt DateTime @default(now())
+  updatedAt DateTime @default(now()) @updatedAt
+
+  userId String?
+
+  // Session metadata
+  title       String?
+  credentials Json    @default("{}") // Map of provider -> credential metadata
+
+  // Rate limiting counters (stored as JSON maps)
+  successfulAgentRuns      Json @default("{}") // Map of graph_id -> count
+  successfulAgentSchedules Json @default("{}") // Map of graph_id -> count
+
+  // Usage tracking
+  totalPromptTokens     Int @default(0)
+  totalCompletionTokens Int @default(0)
+
+  Messages ChatMessage[]
+
+  @@index([userId, updatedAt])
+}
+
+model ChatMessage {
+  id        String   @id @default(uuid())
+  createdAt DateTime @default(now())
+
+  sessionId String
+  Session   ChatSession @relation(fields: [sessionId], references: [id], onDelete: Cascade)
+
+  // Message content
+  role        String // "user", "assistant", "system", "tool", "function"
+  content     String?
+  name        String?
+  toolCallId  String?
+  refusal     String?
+  toolCalls   Json? // List of tool calls for assistant messages
+  functionCall Json? // Deprecated but kept for compatibility
+
+  // Ordering within session
+  sequence Int
+
+  @@unique([sessionId, sequence])
+  @@index([sessionId, sequence])
+}
+
 // This model describes the Agent Graph/Flow (Multi Agent System).
 model AgentGraph {
   id        String    @default(uuid())