implement the database layer for session scoring

backend/alembic/versions/20260202_1721_776dca9d9a2f_add_session_scores_table.py

@@ -0,0 +1,100 @@
+"""add_session_scores_table
+
+Revision ID: 776dca9d9a2f
+Revises: b67c135119d7
+Create Date: 2026-01-13 17:21:32.869647
+
+Adds session_scores table for alert session scoring API (EP-0028):
+- Tracks scoring attempts for alert sessions
+- Stores prompt hash for detecting stale scores
+- Async status tracking (pending -> in_progress -> completed/failed)
+- Partial unique constraint prevents duplicate concurrent scoring attempts
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy import inspect
+
+
+# revision identifiers, used by Alembic.
+revision: str = "776dca9d9a2f"
+down_revision: Union[str, Sequence[str], None] = "b67c135119d7"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Create session_scores table with indexes and constraints."""
+    conn = op.get_bind()
+    inspector = inspect(conn)
+    existing_tables = inspector.get_table_names()
+
+    if "session_scores" not in existing_tables:
+        # Create table
+        op.create_table(
+            "session_scores",
+            sa.Column("score_id", sa.String(), nullable=False),
+            sa.Column("session_id", sa.String(), nullable=False),
+            sa.Column("prompt_hash", sa.String(length=64), nullable=False),
+            sa.Column("total_score", sa.Integer(), nullable=True),
+            sa.Column("score_analysis", sa.Text(), nullable=True),
+            sa.Column("missing_tools_analysis", sa.Text(), nullable=True),
+            sa.Column("score_triggered_by", sa.String(length=255), nullable=False),
+            sa.Column("scored_at_us", sa.BIGINT(), nullable=False),
+            sa.Column("status", sa.String(length=50), nullable=False),
+            sa.Column("started_at_us", sa.BIGINT(), nullable=False),
+            sa.Column("completed_at_us", sa.BIGINT(), nullable=True),
+            sa.Column("error_message", sa.Text(), nullable=True),
+            sa.ForeignKeyConstraint(
+                ["session_id"],
+                ["alert_sessions.session_id"],
+                name="fk_session_scores_session",
+                ondelete="CASCADE",
+            ),
+            sa.PrimaryKeyConstraint("score_id"),
+        )
+
+        # Create indexes
+        op.create_index(
+            "ix_session_scores_session_id", "session_scores", ["session_id"]
+        )
+        op.create_index(
+            "ix_session_scores_prompt_hash", "session_scores", ["prompt_hash"]
+        )
+        op.create_index(
+            "ix_session_scores_total_score", "session_scores", ["total_score"]
+        )
+        op.create_index("ix_session_scores_status", "session_scores", ["status"])
+        op.create_index(
+            "ix_session_scores_session_status",
+            "session_scores",
+            ["session_id", "status"],
+        )
+        op.create_index(
+            "ix_session_scores_status_started",
+            "session_scores",
+            ["status", "started_at_us"],
+        )
+
+        # Create partial unique index (prevents duplicate active scorings)
+        op.execute("""
+            CREATE UNIQUE INDEX ix_session_scores_unique_active
+            ON session_scores(session_id)
+            WHERE status IN ('pending', 'in_progress')
+        """)
+
+
+def downgrade() -> None:
+    """Drop session_scores table."""
+    conn = op.get_bind()
+    inspector = inspect(conn)
+    existing_tables = inspector.get_table_names()
+
+    if "session_scores" in existing_tables:
+        # Drop partial unique index
+        op.execute("DROP INDEX IF EXISTS ix_session_scores_unique_active")
+
+        # Drop table (indexes drop automatically)
+        op.drop_table("session_scores")

backend/tarsy/models/api_models.py

@@ -10,71 +10,94 @@
 
 from pydantic import BaseModel, Field, field_validator
 
-from tarsy.models.constants import ChainStatus
+from tarsy.models.constants import ChainStatus, ScoringStatus
 
 # Non-history related response models
 
+
 class HealthCheckResponse(BaseModel):
     """Response for health check endpoints."""
+
     service: str = Field(description="Service name")
-    status: str = Field(description="Service status ('healthy', 'unhealthy', 'disabled')")
-    timestamp_us: int = Field(description="Health check timestamp (microseconds since epoch UTC)")
+    status: str = Field(
+        description="Service status ('healthy', 'unhealthy', 'disabled')"
+    )
+    timestamp_us: int = Field(
+        description="Health check timestamp (microseconds since epoch UTC)"
+    )
     details: Dict[str, Any] = Field(description="Additional health check details")
 
+
 class ErrorResponse(BaseModel):
     """Standard error response model."""
+
     error: str = Field(description="Error type or category")
     message: str = Field(description="Human-readable error message")
     details: Optional[Dict[str, Any]] = Field(description="Additional error details")
-    timestamp_us: int = Field(description="When the error occurred (microseconds since epoch UTC)")
+    timestamp_us: int = Field(
+        description="When the error occurred (microseconds since epoch UTC)"
+    )
+
 
 class ChainExecutionResult(BaseModel):
     """
     Result from chain execution.
-    
+
     This model represents the result of executing a chain of agent stages,
     containing both success and failure information.
     """
-    
+
     # Core execution metadata - always present
     status: ChainStatus = Field(description="Overall execution status")
-    timestamp_us: int = Field(description="Execution completion timestamp (microseconds since epoch UTC)")
-
+    timestamp_us: int = Field(
+        description="Execution completion timestamp (microseconds since epoch UTC)"
+    )
+
     # Success case fields - present when status is completed or partial
-    final_analysis: Optional[str] = Field(None, description="Final analysis result from the chain")
-
+    final_analysis: Optional[str] = Field(
+        None, description="Final analysis result from the chain"
+    )
+
     # Error case fields - present when status is failed
     error: Optional[str] = Field(None, description="Error message when execution fails")
 
 
 # ===== Chat API Models =====
 
+
 class ChatResponse(BaseModel):
     """Response for chat creation endpoint."""
+
     chat_id: str = Field(description="Unique chat identifier")
     session_id: str = Field(description="Original session ID")
-    created_at_us: int = Field(description="Chat creation timestamp (microseconds since epoch UTC)")
+    created_at_us: int = Field(
+        description="Chat creation timestamp (microseconds since epoch UTC)"
+    )
     created_by: str = Field(description="User who initiated the chat")
     message_count: int = Field(default=0, description="Number of messages in chat")
 
 
 class ChatAvailabilityResponse(BaseModel):
     """Response for chat availability check endpoint."""
+
     available: bool = Field(description="Whether chat is available for this session")
     reason: Optional[str] = Field(default=None, description="Reason if unavailable")
-    chat_id: Optional[str] = Field(default=None, description="Existing chat ID if already created")
+    chat_id: Optional[str] = Field(
+        default=None, description="Existing chat ID if already created"
+    )
 
 
 class ChatMessageRequest(BaseModel):
     """Request body for sending chat message."""
+
     content: str = Field(
-        ..., 
-        min_length=1, 
+        ...,
+        min_length=1,
         max_length=100000,
-        description="User's question or follow-up message"
+        description="User's question or follow-up message",
     )
-    
-    @field_validator('content')
+
+    @field_validator("content")
     @classmethod
     def validate_content(cls, v: str) -> str:
         """Ensure content is not just whitespace."""
@@ -85,25 +108,81 @@ def validate_content(cls, v: str) -> str:
 
 class ChatMessageResponse(BaseModel):
     """Response for message creation endpoint."""
+
     message_id: str = Field(description="Unique message identifier")
     chat_id: str = Field(description="Parent chat ID")
     content: str = Field(description="Message content")
     author: str = Field(description="Message author")
-    created_at_us: int = Field(description="Message timestamp (microseconds since epoch UTC)")
-    stage_execution_id: Optional[str] = Field(default=None, description="Stage execution ID for AI response (streams via WebSocket)")
+    created_at_us: int = Field(
+        description="Message timestamp (microseconds since epoch UTC)"
+    )
+    stage_execution_id: Optional[str] = Field(
+        default=None,
+        description="Stage execution ID for AI response (streams via WebSocket)",
+    )
 
 
 class ChatUserMessageListResponse(BaseModel):
     """Response for chat message history endpoint."""
+
     messages: List[ChatMessageResponse] = Field(description="List of user messages")
     total_count: int = Field(description="Total message count")
     chat_id: str = Field(description="Chat identifier")
 
 
+# ===== Scoring API Models =====
+
+
+class SessionScoreResponse(BaseModel):
+    """
+    Session scoring result for API responses.
+
+    Includes computed field to indicate if score uses current prompt template.
+    """
+
+    score_id: str = Field(description="Unique scoring attempt identifier")
+    session_id: str = Field(description="Alert session that was scored")
+    prompt_hash: str = Field(description="Hash of prompt template used for scoring")
+    total_score: Optional[int] = Field(
+        None, ge=0, le=100, description="Total score (0-100), None if not yet completed"
+    )
+    score_analysis: Optional[str] = Field(None, description="Detailed scoring analysis")
+    missing_tools_analysis: Optional[str] = Field(
+        None, description="Missing tools analysis"
+    )
+    score_triggered_by: str = Field(description="Who/what triggered the scoring")
+    scored_at_us: int = Field(
+        description="When scoring was initiated (microseconds since epoch)"
+    )
+
+    # Async status fields
+    status: ScoringStatus = Field(
+        description="Scoring status (pending|in_progress|completed|failed|timed_out)"
+    )
+    started_at_us: int = Field(
+        description="Processing start time (microseconds since epoch)"
+    )
+    completed_at_us: Optional[int] = Field(
+        None, description="Processing end time (microseconds since epoch)"
+    )
+    error_message: Optional[str] = Field(None, description="Error details if failed")
+
+    # Computed field (set to False in Phase 1, will be calculated in Phase 2)
+    current_prompt_used: bool = Field(
+        default=False,
+        description="Whether this score uses the current prompt template (Phase 2 feature)",
+    )
+
+
 # ===== Session Control API Models =====
 
+
 class CancelAgentResponse(BaseModel):
     """Response for individual parallel agent cancellation endpoint."""
+
     success: bool = Field(description="Whether the cancellation was successful")
     session_status: str = Field(description="Updated session status after cancellation")
-    stage_status: str = Field(description="Updated parent stage status after re-evaluation")
+    stage_status: str = Field(
+        description="Updated parent stage status after re-evaluation"
+    )
+

backend/tarsy/models/constants.py

@@ -158,7 +158,7 @@ class SystemHealthStatus(Enum):
 class IterationStrategy(str, Enum):
     """
     Available iteration strategies for agent processing.
-    
+
     Each strategy implements a different approach to alert analysis:
     - REACT: Standard ReAct pattern with Think→Action→Observation cycles for complete analysis
     - REACT_STAGE: ReAct pattern for stage-specific analysis within multi-stage chains
@@ -175,6 +175,26 @@ class IterationStrategy(str, Enum):
     SYNTHESIS_NATIVE_THINKING = "synthesis-native-thinking"  # Gemini synthesis with thinking
 
 
+class ScoringStatus(Enum):
+    """Status values for session scoring operations."""
+
+    PENDING = "pending"           # Scoring queued but not started
+    IN_PROGRESS = "in_progress"   # LLM currently scoring the session
+    COMPLETED = "completed"       # Scoring finished successfully
+    FAILED = "failed"             # Scoring failed with error
+    TIMED_OUT = "timed_out"       # Scoring timed out
+
+    @classmethod
+    def active_values(cls) -> list['ScoringStatus']:
+        """Active status enum instances (scoring in progress)."""
+        return [cls.PENDING, cls.IN_PROGRESS]
+
+    @classmethod
+    def values(cls) -> List[str]:
+        """All status values as strings."""
+        return [status.value for status in cls]
+
+
 class LLMInteractionType(str, Enum):
     """
     Types of LLM interactions for categorization and UI rendering.
@@ -237,4 +257,4 @@ class StreamingEventType(str, Enum):
 # - MCP results are often large data dumps (less critical than LLM reasoning)
 # - Higher probability of hitting limits with tools like events_list, logs
 # - Browser performance: smaller payloads = better dashboard responsiveness
-MAX_MCP_TOOL_RESULT_SIZE = 524288  # 512KB
+MAX_MCP_TOOL_RESULT_SIZE = 524288  # 512KB