name: conversation-state-management description: | Implement DB-persisted conversation state for stateless AI agent servers. This skill should be used when users need to create/load conversations, append messages with role ordering, fetch history for agents, or resume conversations after server restart.

Conversation State Management

Guide for implementing stateless server conversation persistence with Neon PostgreSQL.

What This Skill Does

• Design conversation and message data models
• Implement create/load conversation operations
• Enforce message role ordering (system → user ↔ assistant)
• Fetch full history for agent context
• Resume conversations after server restart
• User-scoped conversation isolation

What This Skill Does NOT Do

• Implement LLM/AI provider integration
• Handle authentication (assumes auth exists)
• Manage token counting or context window truncation
• Deploy database infrastructure

Before Implementation

Gather context to ensure successful implementation:

Core Architecture

Stateless Server Principle

Request → Load full conversation from DB → Process → Persist → Response
           ↑                                           ↓
           └──────── No in-memory state ───────────────┘

Each request is independent. Server instances are interchangeable.

Three-Table Architecture

Implementation Workflow

1. Create Data Models

See references/data-models.md for complete SQLModel definitions.

# models/conversation.py
class Conversation(SQLModel, table=True):
    id: str = Field(default_factory=lambda: str(uuid4()), primary_key=True)
    user_id: str = Field(foreign_key="user.id", index=True)
    title: str | None = None
    system_prompt: str | None = None
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

class Message(SQLModel, table=True):
    id: str = Field(default_factory=lambda: str(uuid4()), primary_key=True)
    conversation_id: str = Field(foreign_key="conversation.id", index=True)
    role: str  # 'system', 'user', 'assistant'
    content: str
    created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))

2. Create CRUD Operations

See references/api-patterns.md for complete implementations.

Essential operations:

• create_conversation(user_id, title?, system_prompt?)
• get_conversation(conversation_id, user_id) - with ownership check
• list_conversations(user_id, skip, limit)
• add_message(conversation_id, user_id, role, content) - with role validation
• get_messages(conversation_id, user_id) - ordered by created_at
• delete_conversation(conversation_id, user_id)

3. Enforce Role Ordering

See references/role-ordering.md for validation logic.

Rules:

• System message: only at start, max 1
• After system (if any): strict user ↔ assistant alternation
• No consecutive same-role messages

def validate_role_sequence(messages: list[Message], new_role: str) -> bool:
    if not messages:
        return new_role in ('system', 'user')

    last_role = messages[-1].role

    if new_role == 'system':
        return False  # System only at start

    if last_role == 'system':
        return new_role == 'user'

    # Must alternate user ↔ assistant
    return (last_role == 'user' and new_role == 'assistant') or \
           (last_role == 'assistant' and new_role == 'user')

4. Implement API Endpoints

# routers/conversations.py
@router.post("/conversations")
async def create_conversation(data: ConversationCreate, user: AuthUser):
    return crud.create_conversation(user_id=user.id, **data.model_dump())

@router.get("/conversations/{id}")
async def get_conversation(id: str, user: AuthUser):
    conv = crud.get_conversation(id, user_id=user.id)
    if not conv:
        raise HTTPException(404, "Conversation not found")
    return conv

@router.post("/conversations/{id}/messages")
async def add_message(id: str, data: MessageCreate, user: AuthUser):
    # Validates role ordering internally
    return crud.add_message(id, user_id=user.id, **data.model_dump())

@router.get("/conversations/{id}/messages")
async def get_messages(id: str, user: AuthUser):
    return crud.get_messages(id, user_id=user.id)

5. Resume Pattern

See references/resume-patterns.md for checkpoint strategies.

def get_conversation_context(conversation_id: str, user_id: str) -> list[dict]:
    """Fetch full history formatted for LLM."""
    messages = crud.get_messages(conversation_id, user_id)
    return [{"role": m.role, "content": m.content} for m in messages]

# On resume, simply load and continue
@router.post("/conversations/{id}/continue")
async def continue_conversation(id: str, data: MessageCreate, user: AuthUser):
    # Load existing context
    context = get_conversation_context(id, user.id)

    # Add new user message
    crud.add_message(id, user.id, role="user", content=data.content)
    context.append({"role": "user", "content": data.content})

    # Call LLM with full context
    response = await llm.generate(messages=context)

    # Persist assistant response
    crud.add_message(id, user.id, role="assistant", content=response)

    return {"response": response}

Stateless Request Cycle

1. Authenticate request (JWT/session)
2. Load conversation from DB (includes all messages)
3. Validate operation (ownership, role ordering)
4. Execute operation (add message, call LLM)
5. Persist changes to DB
6. Return response
7. Connection closed (no state retained)

Key principle: Any server instance can handle any request.

User Isolation

Always filter by user_id at the data layer:

def get_conversation(conversation_id: str, user_id: str) -> Conversation | None:
    statement = select(Conversation).where(
        Conversation.id == conversation_id,
        Conversation.user_id == user_id  # Ownership check
    )
    return session.exec(statement).first()

Return 404 (not 403) when conversation exists but belongs to another user.

Output Checklist

Before delivering implementation:

• Conversation model with user_id foreign key
• Message model with conversation_id foreign key and role field
• Role enum/validation (system, user, assistant)
• Role ordering validation before insert
• User ownership check on all operations
• Messages ordered by created_at
• CRUD: create, get, list, add_message, get_messages, delete
• API endpoints with auth dependency
• No in-memory conversation state

Reference Files