stonks-oracle/services/intelligence_pipeline_v3/adjudication/deployment.py

"""9B adjudicator deployment configuration for Intelligence Pipeline v3.

Manages the approved model/version pins, VRAM gating, concurrency
semaphore configuration, and alerting thresholds for the 9B adjudicator
running on RTX 4070 Ti SUPER via vLLM.
"""

from __future__ import annotations

import asyncio
from typing import Any

from pydantic import BaseModel, Field

# --- Pinned model and version constants ---

APPROVED_MODEL: str = "AxionML/Qwen3.5-9B-NVFP4"
"""Approved 9B model for adjudication. NVFP4 quantization for 4070 Ti SUPER."""

APPROVED_VLLM_VERSION: str = "0.8.5"
"""Approved vLLM version matching the cluster deployment."""

APPROVED_SERVED_NAME: str = "stonks-adjudicator-9b"
"""The served model name exposed by the vLLM deployment."""

MAX_MODEL_LEN: int = 8192
"""Maximum model context length configured for the deployment."""

MAX_NUM_SEQS: int = 8
"""Maximum concurrent sequences for the vLLM deployment."""

GPU_MEMORY_UTILIZATION: float = 0.80
"""Target GPU memory utilization fraction."""

VRAM_GATE_PERCENT: float = 5.0
"""Maximum allowed VRAM increase over baseline (percentage)."""


# --- Structured output verification ---


def verify_structured_output(target: dict[str, Any]) -> bool:
    """Verify that structured output works with the given deployment target.

    Checks that the target deployment declares json_schema support in its
    capabilities and that the required configuration fields are present.

    Args:
        target: Deployment target configuration dict containing at minimum:
            - capabilities: dict with json_schema boolean
            - served_model_name: str matching APPROVED_SERVED_NAME
            - vllm_version: str for version verification

    Returns:
        True if strict schema output is expected to work, False otherwise.
    """
    capabilities = target.get("capabilities", {})
    if not capabilities.get("json_schema", False):
        return False

    # Verify model matches approved deployment
    served_name = target.get("served_model_name", "")
    if served_name and served_name != APPROVED_SERVED_NAME:
        return False

    # Verify vLLM version compatibility
    vllm_version = target.get("vllm_version", "")
    if vllm_version and vllm_version != APPROVED_VLLM_VERSION:
        return False

    # Verify the model is the approved one
    model = target.get("model", "")
    if model and model != APPROVED_MODEL:
        return False

    return True


# --- VRAM gate ---


def check_vram_gate(peak_mb: float, baseline_mb: float) -> bool:
    """Check whether peak VRAM usage is within the +5% gate of baseline.

    The gate ensures that no deployment update exceeds the measured current
    9B deployment VRAM by more than 5 percent.

    Args:
        peak_mb: Measured peak VRAM in megabytes during test.
        baseline_mb: Baseline VRAM measurement in megabytes.

    Returns:
        True if peak is within acceptable range, False if it exceeds the gate.
    """
    if baseline_mb <= 0:
        return False
    if peak_mb <= 0:
        return False

    max_allowed_mb = baseline_mb * (1.0 + VRAM_GATE_PERCENT / 100.0)
    return peak_mb <= max_allowed_mb


# --- Concurrency semaphore ---


class ConcurrencySemaphore(BaseModel):
    """Configuration for the adjudication concurrency semaphore.

    Limits concurrent adjudication requests to protect vLLM from
    overload. Aligned with max-num-seqs and KV-cache behavior.
    """

    max_concurrent: int = Field(
        default=MAX_NUM_SEQS,
        gt=0,
        description="Maximum concurrent adjudication requests",
    )
    queue_timeout_seconds: float = Field(
        default=120.0,
        gt=0,
        description="Maximum time to wait for semaphore acquisition",
    )
    backpressure_threshold: int = Field(
        default=MAX_NUM_SEQS * 4,
        ge=0,
        description="Queue depth at which backpressure signals are emitted",
    )

    def create_semaphore(self) -> asyncio.Semaphore:
        """Create an asyncio.Semaphore with the configured max_concurrent."""
        return asyncio.Semaphore(self.max_concurrent)


# --- Alert configuration ---


class AlertConfig(BaseModel):
    """Alert thresholds for adjudicator monitoring.

    Defines queue-depth and availability thresholds that trigger alerts
    when the adjudicator is overloaded or unavailable.
    """

    queue_depth_warning: int = Field(
        default=16,
        ge=1,
        description="Queue depth that triggers a warning alert",
    )
    queue_depth_critical: int = Field(
        default=32,
        ge=1,
        description="Queue depth that triggers a critical alert",
    )
    availability_threshold_percent: float = Field(
        default=95.0,
        gt=0.0,
        le=100.0,
        description="Minimum availability percentage before alerting",
    )
    latency_p95_warning_ms: int = Field(
        default=5000,
        gt=0,
        description="p95 latency (ms) that triggers a warning",
    )
    latency_p95_critical_ms: int = Field(
        default=15000,
        gt=0,
        description="p95 latency (ms) that triggers a critical alert",
    )
    consecutive_failures_alert: int = Field(
        default=3,
        ge=1,
        description="Number of consecutive failures before alerting",
    )
    health_check_interval_seconds: float = Field(
        default=30.0,
        gt=0,
        description="Interval between health checks in seconds",
    )