LLM agent canary deployment strategy: safe rollout with cost and quality gates

import hashlib
import sqlite3
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Literal

VersionName = Literal["baseline", "canary"]

@dataclass
class SessionMetrics:
    session_id: str
    version: VersionName
    cost_usd: float = 0.0
    turns: int = 0
    loops_tripped: int = 0
    completed: bool = False
    started_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())

def assign_version(
    session_id: str,
    canary_pct: float = 0.05,  # 5% canary
) -> VersionName:
    """
    Deterministically assign a session to canary or baseline.
    Uses SHA-256 of session_id to ensure consistent assignment
    for the same session across restarts.
    """
    hash_val = int(hashlib.sha256(session_id.encode()).hexdigest(), 16)
    bucket = (hash_val % 10_000) / 10_000.0  # 0.0 – 0.9999
    return "canary" if bucket < canary_pct else "baseline"

def init_canary_db(db_path: str) -> None:
    """Initialize the canary metrics database."""
    with sqlite3.connect(db_path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS canary_sessions (
                session_id TEXT PRIMARY KEY,
                version TEXT NOT NULL,
                cost_usd REAL DEFAULT 0,
                turns INTEGER DEFAULT 0,
                loops_tripped INTEGER DEFAULT 0,
                completed INTEGER DEFAULT 0,
                started_at TEXT
            )
        """)

def record_session(db_path: str, m: SessionMetrics) -> None:
    with sqlite3.connect(db_path) as conn:
        conn.execute("""
            INSERT OR REPLACE INTO canary_sessions
            (session_id, version, cost_usd, turns, loops_tripped, completed, started_at)
            VALUES (?,?,?,?,?,?,?)
        """, (m.session_id, m.version, m.cost_usd, m.turns,
              m.loops_tripped, int(m.completed), m.started_at))

def canary_gate_check(db_path: str, min_sessions: int = 50) -> dict:
    """
    Check whether the canary meets gate criteria.
    Returns {pass: bool, reason: str, metrics: dict}.
    """
    with sqlite3.connect(db_path) as conn:
        rows = conn.execute("""
            SELECT version,
                   COUNT(*) as sessions,
                   AVG(cost_usd) as avg_cost,
                   CAST(SUM(loops_tripped) AS REAL) / COUNT(*) as loop_rate,
                   CAST(SUM(completed) AS REAL) / COUNT(*) as completion_rate,
                   AVG(turns) as avg_turns
            FROM canary_sessions
            GROUP BY version
        """).fetchall()

    versions = {r[0]: {
        "sessions": r[1], "avg_cost": r[2], "loop_rate": r[3],
        "completion_rate": r[4], "avg_turns": r[5],
    } for r in rows}

    baseline = versions.get("baseline", {})
    canary = versions.get("canary", {})

    if not baseline or not canary:
        return {"pass": None, "reason": "insufficient data", "metrics": versions}
    if canary["sessions"] < min_sessions:
        return {"pass": None, "reason": f"canary has {canary['sessions']}<{min_sessions} sessions", "metrics": versions}

    # Gate 1: cost regression
    cost_ratio = canary["avg_cost"] / baseline["avg_cost"] if baseline["avg_cost"] else 1.0
    if cost_ratio > 1.30:
        return {"pass": False, "reason": f"Cost regression: canary {cost_ratio:.2f}× baseline", "metrics": versions}

    # Gate 2: loop frequency
    loop_ratio = canary["loop_rate"] / baseline["loop_rate"] if baseline["loop_rate"] else float("inf")
    if loop_ratio > 3.0:
        return {"pass": False, "reason": f"Loop regression: canary {loop_ratio:.1f}× baseline loop rate", "metrics": versions}

    # Gate 3: completion rate
    if canary["completion_rate"] < baseline["completion_rate"] * 0.85:
        return {
            "pass": False,
            "reason": f"Completion regression: canary {canary['completion_rate']:.1%} vs baseline {baseline['completion_rate']:.1%}",
            "metrics": versions,
        }

    return {"pass": True, "reason": "all gates pass", "metrics": versions}

from runguard import LoopDetector, LoopDetectedError, BudgetExceededError

class CanaryAgentSession:
    """
    A single agent session that tracks metrics for canary analysis.
    Each session has its own LoopDetector — trips in this session
    do not affect the shared loop state.
    """
    def __init__(
        self,
        session_id: str,
        db_path: str,
        canary_pct: float = 0.05,
        budget_usd: float = 2.0,
    ):
        self.session_id = session_id
        self.db_path = db_path
        self.version = assign_version(session_id, canary_pct)
        self.detector = LoopDetector(repeats=3, max_cycle_len=3)
        self.metrics = SessionMetrics(session_id=session_id, version=self.version)
        self.budget_usd = budget_usd

    def record_tool_call(self, tool_name: str, args_key: str) -> None:
        """Check for loops before executing a tool call."""
        sig = f"{tool_name}:{args_key}"
        match = self.detector.record(sig)
        if match:
            self.metrics.loops_tripped += 1
            record_session(self.db_path, self.metrics)
            raise LoopDetectedError(
                f"[{self.version.upper()}] Loop: {tool_name} repeated "
                f"{match.repeats}× — session {self.session_id}",
                match=match,
            )

    def record_cost(self, cost_usd: float) -> None:
        self.metrics.cost_usd += cost_usd
        self.metrics.turns += 1
        if self.metrics.cost_usd > self.budget_usd:
            record_session(self.db_path, self.metrics)
            raise BudgetExceededError(
                f"[{self.version.upper()}] Budget ${self.budget_usd} exceeded "
                f"in session {self.session_id}"
            )

    def complete(self) -> None:
        self.metrics.completed = True
        record_session(self.db_path, self.metrics)

    def abort(self) -> None:
        self.metrics.completed = False
        record_session(self.db_path, self.metrics)

• Day 0: 1% canary — 50+ sessions — gate check → pass/fail
• Day 1: 5% canary — 200+ sessions — gate check → pass/fail
• Day 2: 20% canary — 500+ sessions — gate check → pass/fail
• Day 4: 50% canary — 1,000+ sessions — final gate → promote or rollback
• Day 7: 100% (promote) — decommission baseline