AI Feature Flags & Rollout

/** AI-aware feature flag configuration. */
export interface AIFeatureFlag {
  /** Unique flag key, e.g., "chat-v2-model-upgrade" */
  key: string;

  /** Human-readable description */
  description: string;

  /** Current rollout stage */
  stage: "disabled" | "canary" | "gradual" | "full" | "killed";

  /** Percentage of traffic receiving the new variant (0-100) */
  rolloutPercentage: number;

  /** Quality gate configuration */
  qualityGate: {
    /** Minimum average quality score (1-5) to maintain rollout */
    minScore: number;
    /** Rolling window in minutes for quality evaluation */
    windowMinutes: number;
    /** Minimum number of evaluations before gate is active */
    minSamples: number;
  };

  /** Automatic promotion rules */
  promotion: {
    /** Hours at current percentage before auto-promoting */
    stabilityHours: number;
    /** Percentage increment per promotion step */
    stepSize: number;
    /** Target percentage for full rollout */
    targetPercentage: number;
  };

  /** Kill switch: instantly disable if quality drops below */
  killSwitch: {
    enabled: boolean;
    /** Quality score threshold that triggers kill */
    threshold: number;
    /** Number of consecutive low scores before killing */
    consecutiveFailures: number;
  };

  /** Model/prompt version this flag controls */
  variant: {
    modelName?: string;
    promptVersion?: string;
    configOverrides?: Record<string, unknown>;
  };
}

"""Quality-gated promotion controller for AI feature flags."""

from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone

from flags.store import FlagStore
from flags.quality import QualitySignalAggregator

logger = logging.getLogger(__name__)


class PromotionController:
    """Manages automatic promotion and kill switch for AI flags."""

    def __init__(self, store: FlagStore, quality: QualitySignalAggregator):
        self.store = store
        self.quality = quality

    async def evaluate_all_flags(self):
        """Check all active flags and promote or kill as needed."""
        flags = await self.store.get_active_flags()

        for flag in flags:
            if flag.stage in ("disabled", "full", "killed"):
                continue

            score = await self.quality.get_average_score(
                flag_key=flag.key,
                window_minutes=flag.quality_gate.window_minutes,
            )
            sample_count = await self.quality.get_sample_count(
                flag_key=flag.key,
                window_minutes=flag.quality_gate.window_minutes,
            )

            logger.info(
                "Flag %s: score=%.2f samples=%d rollout=%d%%",
                flag.key, score or 0, sample_count, flag.rollout_percentage,
            )

            # Not enough data yet — skip evaluation
            if sample_count < flag.quality_gate.min_samples:
                continue

            # Kill switch check
            if (
                flag.kill_switch.enabled
                and score is not None
                and score < flag.kill_switch.threshold
            ):
                consecutive = await self.quality.get_consecutive_failures(
                    flag.key, flag.kill_switch.threshold
                )
                if consecutive >= flag.kill_switch.consecutive_failures:
                    logger.warning("KILL SWITCH: flag %s killed", flag.key)
                    await self.store.update_flag(
                        flag.key, stage="killed", rollout_percentage=0
                    )
                    continue

            # Quality gate check for promotion
            if score is not None and score >= flag.quality_gate.min_score:
                last_change = await self.store.get_last_change_time(flag.key)
                hours_stable = (
                    datetime.now(timezone.utc) - last_change
                ).total_seconds() / 3600

                if hours_stable >= flag.promotion.stability_hours:
                    new_pct = min(
                        flag.rollout_percentage + flag.promotion.step_size,
                        flag.promotion.target_percentage,
                    )
                    new_stage = (
                        "full" if new_pct >= flag.promotion.target_percentage
                        else flag.stage
                    )
                    logger.info(
                        "Promoting flag %s: %d%% -> %d%%",
                        flag.key, flag.rollout_percentage, new_pct,
                    )
                    await self.store.update_flag(
                        flag.key, stage=new_stage, rollout_percentage=new_pct
                    )