Claude-skill-registry conversation-state-management
install
source · Clone the upstream repo
git clone https://github.com/majiayu000/claude-skill-registry
Claude Code · Install into ~/.claude/skills/
T=$(mktemp -d) && git clone --depth=1 https://github.com/majiayu000/claude-skill-registry "$T" && mkdir -p ~/.claude/skills && cp -r "$T/skills/data/conversation-state-management" ~/.claude/skills/majiayu000-claude-skill-registry-conversation-state-management && rm -rf "$T"
manifest:
skills/data/conversation-state-management/SKILL.mdsource content
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:
| Source | Gather |
|---|---|
| Codebase | Existing models, database.py, CRUD patterns, router structure |
| Conversation | User's specific table naming, additional fields needed |
| Skill References | Data models, API patterns, role ordering rules |
| User Guidelines | Project conventions, existing auth patterns |
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
| Table | Purpose | Key Fields |
|---|---|---|
| users | Identity (from auth) | id, email |
| conversations | Session metadata | id, user_id, title, system_prompt, created_at |
| messages | Individual messages | id, conversation_id, role, content, created_at |
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?)
- with ownership checkget_conversation(conversation_id, user_id)list_conversations(user_id, skip, limit)
- with role validationadd_message(conversation_id, user_id, role, content)
- ordered by created_atget_messages(conversation_id, user_id)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
| File | When to Read |
|---|---|
| Complete SQLModel/Pydantic definitions |
| CRUD operations and router implementations |
| Role validation logic and edge cases |
| Checkpoint and continuation strategies |