feat: add userId storage (#92)

Author: enitrat

API_DOCUMENTATION.md

@@ -23,6 +23,19 @@ No authentication is enforced by the server. Add your own gateway (API key, OAut
 - `Content-Type: application/json` — required for JSON POSTs.
 - `Accept: application/json` or `text/event-stream` depending on `stream` usage.
 - `x-mcp-mode: true` or `mcp: true` — optional. When present (any value), the request runs in **MCP mode**, returning raw documentation snippets instead of synthesized answers. See [MCP Mode](#mcp-mode).
+- `x-conversation-id` — optional. Links multiple requests to the same conversation for analytics.
+- `x-user-id` — optional. Anonymous user identifier for tracking usage across sessions. The server hashes this value before storage for privacy.
+- `x-api-key` — optional. When present, the server uses a hash of the API key as the user identifier (takes precedence over `x-user-id`).
+
+#### User Identification
+
+The server derives an anonymized `user_id` for each request using the following priority:
+
+1. If `x-api-key` header is present → hash the API key
+2. Else if `x-user-id` header is present → hash the user ID
+3. Else → no user identification
+
+This allows tracking user behavior across sessions without storing sensitive identifiers.
 
 ## Health Check
 
@@ -366,6 +379,8 @@ Fetch paginated user queries. If no date range is provided, returns the most rec
 - `query_text` _(optional)_ — filter by text contained in the query (case-insensitive).
 - `limit` _(default `100`)_ — maximum rows returned.
 - `offset` _(default `0`)_ — pagination offset.
+- `conversation_id` _(optional)_ — filter by conversation id.
+- `user_id` _(optional)_ — filter by hashed user id. Useful for tracking individual user journeys and retention.
 
 **Response** `200 OK`
 
@@ -381,7 +396,9 @@ Fetch paginated user queries. If no date range is provided, returns the most rec
         { "role": "user", "content": "What is Cairo?" },
         { "role": "assistant", "content": "Cairo is a programming language..." }
       ],
-      "output": "To declare a storage variable in Cairo 1, you use the #[storage] attribute..."
+      "output": "To declare a storage variable in Cairo 1, you use the #[storage] attribute...",
+      "conversation_id": "123e4567-e89b-12d3-a456-426614174000",
+      "user_id": "a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6"
     }
   ],
   "total": 1,
@@ -390,6 +407,8 @@ Fetch paginated user queries. If no date range is provided, returns the most rec
 }
 ```
 
+The `user_id` field contains the hashed identifier derived from either the `x-api-key` or `x-user-id` header at request time. This enables user retention and usage analytics without storing raw identifiers.
+
 ## Versioning & Compatibility
 
 - Current API version: `1.0.0` (see FastAPI metadata).

python/src/cairo_coder/db/models.py

@@ -21,6 +21,7 @@ class UserInteraction(BaseModel):
     agent_id: str
     mcp_mode: bool = False
     conversation_id: Optional[str] = None
+    user_id: Optional[str] = None
     chat_history: Optional[list[dict[str, Any]]] = None
     query: str
     generated_answer: Optional[str] = None

python/src/cairo_coder/db/repository.py

@@ -84,18 +84,20 @@ async def create_user_interaction(interaction: UserInteraction) -> None:
                     agent_id,
                     mcp_mode,
                     conversation_id,
+                    user_id,
                     chat_history,
                     query,
                     generated_answer,
                     retrieved_sources,
                     llm_usage
                 )
-                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
+                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
                 """,
                 interaction.id,
                 interaction.agent_id,
                 interaction.mcp_mode,
                 interaction.conversation_id,
+                interaction.user_id,
                 _serialize_json_field(interaction.chat_history),
                 interaction.query,
                 interaction.generated_answer,
@@ -115,6 +117,7 @@ async def get_interactions(
     offset: int,
     query_text: str | None = None,
     conversation_id: str | None = None,
+    user_id: str | None = None,
 ) -> tuple[list[dict[str, Any]], int]:
     """Fetch paginated interactions matching the supplied filters.
 
@@ -146,6 +149,10 @@ async def get_interactions(
             params.append(conversation_id)
             filters.append(f"conversation_id = ${len(params)}")
 
+        if user_id:
+            params.append(user_id)
+            filters.append(f"user_id = ${len(params)}")
+
         where_clause = "WHERE " + " AND ".join(filters) if filters else ""
 
         count_query = f"""
@@ -159,7 +166,7 @@ async def get_interactions(
         limit_placeholder = len(params) - 1
         offset_placeholder = len(params)
         data_query = f"""
-            SELECT id, created_at, agent_id, query, chat_history, generated_answer, conversation_id
+            SELECT id, created_at, agent_id, query, chat_history, generated_answer, conversation_id, user_id
             FROM user_interactions
             {where_clause}
             ORDER BY created_at DESC
@@ -200,18 +207,20 @@ async def migrate_user_interaction(interaction: UserInteraction) -> tuple[bool,
                     agent_id,
                     mcp_mode,
                     conversation_id,
+                    user_id,
                     chat_history,
                     query,
                     generated_answer,
                     retrieved_sources,
                     llm_usage
                 )
-                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+                VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
                 ON CONFLICT (id) DO UPDATE SET
                     created_at = EXCLUDED.created_at,
                     agent_id = EXCLUDED.agent_id,
                     mcp_mode = EXCLUDED.mcp_mode,
                     conversation_id = EXCLUDED.conversation_id,
+                    user_id = EXCLUDED.user_id,
                     chat_history = EXCLUDED.chat_history,
                     query = EXCLUDED.query,
                     generated_answer = EXCLUDED.generated_answer,
@@ -224,6 +233,7 @@ async def migrate_user_interaction(interaction: UserInteraction) -> tuple[bool,
                 interaction.agent_id,
                 interaction.mcp_mode,
                 interaction.conversation_id,
+                interaction.user_id,
                 _serialize_json_field(interaction.chat_history),
                 interaction.query,
                 interaction.generated_answer,

python/src/cairo_coder/db/session.py

@@ -72,6 +72,7 @@ async def execute_schema_scripts() -> None:
                 agent_id VARCHAR(50) NOT NULL,
                 mcp_mode BOOLEAN NOT NULL DEFAULT FALSE,
                 conversation_id VARCHAR(100),
+                user_id VARCHAR(100),
                 chat_history JSONB,
                 query TEXT NOT NULL,
                 generated_answer TEXT,
@@ -87,6 +88,13 @@ async def execute_schema_scripts() -> None:
             ADD COLUMN IF NOT EXISTS conversation_id VARCHAR(100);
             """
         )
+        # Migration: add user_id column if it doesn't exist (for existing tables)
+        await connection.execute(
+            """
+            ALTER TABLE user_interactions
+            ADD COLUMN IF NOT EXISTS user_id VARCHAR(100);
+            """
+        )
         await connection.execute(
             """
             CREATE INDEX IF NOT EXISTS idx_interactions_created_at
@@ -95,6 +103,8 @@ async def execute_schema_scripts() -> None:
                 ON user_interactions(agent_id);
             CREATE INDEX IF NOT EXISTS idx_interactions_conversation_id
                 ON user_interactions(conversation_id);
+            CREATE INDEX IF NOT EXISTS idx_interactions_user_id
+                ON user_interactions(user_id);
             """
         )
     logger.info("Database schema initialized.")

python/src/cairo_coder/server/app.py

@@ -6,6 +6,7 @@
 """
 
 import argparse
+import hashlib
 import json
 import os
 import time
@@ -40,6 +41,24 @@
 setup_logging(os.environ.get("LOG_LEVEL", "INFO"), os.environ.get("LOG_FORMAT", "console"))
 logger = structlog.get_logger(__name__)
 
+
+def hash_user_id(user_id: str | None) -> str | None:
+    """
+    Hash a user ID for anonymization.
+
+    Uses SHA-256 to create a consistent, anonymized identifier from the raw user ID.
+    This ensures privacy while maintaining the ability to track user behavior.
+
+    Args:
+        user_id: Raw user ID from header (UUID or API key)
+
+    Returns:
+        Hashed user ID (first 32 chars of SHA-256 hex digest) or None if no input
+    """
+    if user_id is None:
+        return None
+    return hashlib.sha256(user_id.encode()).hexdigest()[:32]
+
 # Global vector DB instance managed by FastAPI lifecycle
 _vector_db: SourceFilteredPgVectorRM | None = None
 _agent_factory: AgentFactory | None = None
@@ -152,6 +171,7 @@ async def log_interaction_task(
     response: ChatCompletionResponse,
     agent: RagPipeline,
     conversation_id: str | None = None,
+    user_id: str | None = None,
 ) -> None:
     """Background task that persists a user interaction."""
     sources_data = [
@@ -169,6 +189,7 @@ async def log_interaction_task(
         agent_id=agent_id,
         mcp_mode=mcp_mode,
         conversation_id=conversation_id,
+        user_id=user_id,
         chat_history=chat_history_dicts,
         query=query,
         generated_answer=response.choices[0].message.content if response.choices else None,
@@ -186,6 +207,7 @@ async def log_interaction_raw(
     generated_answer: str | None,
     agent: RagPipeline,
     conversation_id: str | None = None,
+    user_id: str | None = None,
 ) -> None:
     """Persist a user interaction without constructing a full response object."""
     sources_data = [
@@ -202,6 +224,7 @@ async def log_interaction_raw(
         agent_id=agent_id,
         mcp_mode=mcp_mode,
         conversation_id=conversation_id,
+        user_id=user_id,
         chat_history=chat_history_dicts,
         query=query,
         generated_answer=generated_answer,
@@ -429,6 +452,11 @@ async def _handle_chat_completion(
         """Handle chat completion request."""
         # Extract conversation ID from header
         conversation_id = req.headers.get("x-conversation-id")
+        # Extract user identifier for anonymized tracking
+        # Priority: 1) x-api-key (for authenticated users), 2) x-user-id (for frontend users)
+        api_key = req.headers.get("x-api-key")
+        raw_user_id = req.headers.get("x-user-id")
+        user_id = hash_user_id(api_key) if api_key else hash_user_id(raw_user_id)
 
         # Convert messages to internal format
         messages = []
@@ -451,7 +479,7 @@ async def _handle_chat_completion(
         if request.stream:
             return StreamingResponse(
                 self._stream_chat_completion(
-                    agent, query, messages[:-1], mcp_mode, effective_agent_id, conversation_id
+                    agent, query, messages[:-1], mcp_mode, effective_agent_id, conversation_id, user_id
                 ),
                 media_type="text/event-stream",
                 headers={
@@ -472,6 +500,7 @@ async def _handle_chat_completion(
             response=response,
             agent=agent,
             conversation_id=conversation_id,
+            user_id=user_id,
         )
 
         return response
@@ -484,6 +513,7 @@ async def _stream_chat_completion(
         mcp_mode: bool,
         agent_id: str,
         conversation_id: str | None = None,
+        user_id: str | None = None,
     ) -> AsyncGenerator[str, None]:
         """Stream chat completion response - replicates TypeScript streaming."""
         response_id = str(uuid.uuid4())
@@ -597,6 +627,7 @@ async def _stream_chat_completion(
                     generated_answer=final_response,
                     agent=agent,
                     conversation_id=conversation_id,
+                    user_id=user_id,
                 )
             except Exception as log_error:
                 logger.error("Failed to log streaming interaction", error=str(log_error), exc_info=True)

python/src/cairo_coder/server/insights_api.py

@@ -29,6 +29,7 @@ class QueryResponse(BaseModel):
     chat_history: list[dict[str, Any]]
     output: str | None
     conversation_id: str | None = None
+    user_id: str | None = None
 
 
 class PaginatedQueryResponse(BaseModel):
@@ -47,6 +48,7 @@ async def get_raw_queries(
     agent_id: str | None = None,
     query_text: str | None = None,
     conversation_id: str | None = None,
+    user_id: str | None = None,
     limit: int = 100,
     offset: int = 0,
 ) -> PaginatedQueryResponse:
@@ -56,9 +58,10 @@ async def get_raw_queries(
     ordered by creation time (where N is the limit parameter).
 
     Use conversation_id to filter queries belonging to a specific conversation.
+    Use user_id to filter queries belonging to a specific user.
     """
     items, total = await get_interactions(
-        start_date, end_date, agent_id, limit, offset, query_text, conversation_id
+        start_date, end_date, agent_id, limit, offset, query_text, conversation_id, user_id
     )
     # Map generated_answer to output for API response
     responses = [
@@ -70,6 +73,7 @@ async def get_raw_queries(
             chat_history=item["chat_history"] or [],
             output=item.get("generated_answer"),
             conversation_id=item.get("conversation_id"),
+            user_id=item.get("user_id"),
         )
         for item in items
     ]

python/tests/integration/conftest.py

@@ -165,9 +165,16 @@ async def test_db_pool(postgres_container):
     """Asyncpg pool connected to the ephemeral Postgres.
 
     Creates schema directly to avoid cross-loop pool reuse with the app.
+    Clears global pools dict before creating a new pool to prevent connection exhaustion.
     """
     import asyncpg  # local import to avoid import at collection when skipped
 
+    from cairo_coder.db import session as db_session
+
+    # Clear any lingering pools from the global dict to prevent connection exhaustion
+    # during long test runs
+    await db_session.close_pool()
+
     raw_dsn = postgres_container.get_connection_url()
     # Convert SQLAlchemy-style DSN to asyncpg-compatible DSN
     dsn = raw_dsn.replace("postgresql+psycopg2", "postgresql")
@@ -183,6 +190,7 @@ async def test_db_pool(postgres_container):
                 agent_id VARCHAR(50) NOT NULL,
                 mcp_mode BOOLEAN NOT NULL DEFAULT FALSE,
                 conversation_id VARCHAR(100),
+                user_id VARCHAR(100),
                 chat_history JSONB,
                 query TEXT NOT NULL,
                 generated_answer TEXT,
@@ -199,6 +207,8 @@ async def test_db_pool(postgres_container):
                 ON user_interactions(agent_id);
             CREATE INDEX IF NOT EXISTS idx_interactions_conversation_id
                 ON user_interactions(conversation_id);
+            CREATE INDEX IF NOT EXISTS idx_interactions_user_id
+                ON user_interactions(user_id);
             """
         )