Scenarios and Runbooks — Dynamic Model Selection Architecture

mindmap
  root((Dynamic Model<br/>Selection Failures))
    Hard-Coded Model ID
      Lambda Redeploy Required
      No Runtime Switching
      Costly Cost Spikes
    AppConfig Cache Miss
      200ms Config Latency
      Per-Request API Calls
      Throttling Risk
    Missing Schema Validation
      Bad Config Crashes Workers
      Unvalidated Model Entries
      Inference Layer Outage
    Env Var Provider Switch
      In-Flight Version Skew
      Multiple Models Serving
      Non-Deterministic Responses
    Routing Rule Conflict
      Cost vs Latency Router
      Non-Deterministic Selection
      Audit Trail Gap

flowchart TD
    A["CloudWatch Alarm:<br/>BedRock token cost<br/>anomaly detector fires"] --> B{"On-call opens<br/>Lambda console"}
    B --> C["Searches for model ID<br/>in environment variables"]
    C -->|"Not found in env vars"| D["Searches source code<br/>in Lambda function code"]
    D -->|"Model ID found<br/>in source code"| E["ROOT CAUSE:<br/>Hard-coded model ID —<br/>no runtime switching available"]
    D -->|"Not found in source"| F["Check AppConfig /<br/>Parameter Store for routing config"]
    E --> G{"Assess alternatives"}
    G -->|"No config switch available"| H["Must redeploy Lambda<br/>with updated model ID"]
    G -->|"AppConfig exists but unused"| I["Enable AppConfig-backed<br/>routing — use Runbook 1"]
    H --> J["Track deployment duration<br/>vs cost burn — escalate if > 15 min"]

"""
Runbook 1: Migrate hard-coded model ID to AppConfig-backed dynamic routing.
Provides zero-redeploy model switching for MangaAssist inference Lambda.

Prerequisites:
  - AWS AppConfig application: "manga-assist"
  - AppConfig environment: "production"
  - AppConfig configuration profile: "model-routing"
  - Initial hosted config document (JSON) deployed before Lambda update
"""

import json
import logging
import os
import time
from functools import lru_cache
from typing import Any

import boto3

logger = logging.getLogger("manga_model_router")
logger.setLevel(logging.INFO)

# ---------------------------------------------------------------------------
# AppConfig SDK cache — avoids per-request HTTP calls (see Scenario 2 fix)
# ---------------------------------------------------------------------------
_appconfig_client = boto3.client("appconfigdata", region_name=os.environ.get("AWS_REGION", "us-east-1"))

# Module-level state — persists across warm invocations
_config_session_token: str | None = None
_config_cache: dict | None = None
_cache_loaded_at: float = 0.0

# AppConfig poll interval — Lambda will not call GetLatestConfiguration
# more often than this across warm invocations (max 1 req per interval per instance)
APPCONFIG_POLL_SECONDS = int(os.environ.get("APPCONFIG_POLL_SECONDS", "60"))

APPCONFIG_APP = os.environ["APPCONFIG_APP"]           # e.g. "manga-assist"
APPCONFIG_ENV = os.environ["APPCONFIG_ENV"]           # e.g. "production"
APPCONFIG_PROFILE = os.environ["APPCONFIG_PROFILE"]   # e.g. "model-routing"


def _start_config_session() -> str:
    """Start an AppConfig data session and return the initial token."""
    response = _appconfig_client.start_configuration_session(
        ApplicationIdentifier=APPCONFIG_APP,
        EnvironmentIdentifier=APPCONFIG_ENV,
        ConfigurationProfileIdentifier=APPCONFIG_PROFILE,
        RequiredMinimumPollIntervalInSeconds=APPCONFIG_POLL_SECONDS,
    )
    return response["InitialConfigurationToken"]


def get_model_routing_config() -> dict:
    """
    Fetch model routing config from AppConfig with Lambda-level caching.

    Returns the cached config if last fetch was within APPCONFIG_POLL_SECONDS.
    Only calls GetLatestConfiguration when the poll interval has elapsed.

    Config shape:
    {
      "default_model_id": "anthropic.claude-3-haiku-20240307-v1:0",
      "intent_overrides": {
        "product_recommendation": "anthropic.claude-3-sonnet-20240229-v1:0",
        "content_moderation": "anthropic.claude-3-haiku-20240307-v1:0"
      },
      "cost_override_active": false
    }
    """
    global _config_session_token, _config_cache, _cache_loaded_at

    now = time.monotonic()

    # Return cached config if within poll interval
    if _config_cache is not None and (now - _cache_loaded_at) < APPCONFIG_POLL_SECONDS:
        return _config_cache

    # Start session on cold start or token expiry
    if _config_session_token is None:
        _config_session_token = _start_config_session()

    try:
        response = _appconfig_client.get_latest_configuration(
            ConfigurationToken=_config_session_token
        )
        # Token is rotated on every call — must always update
        _config_session_token = response["NextPollConfigurationToken"]

        raw = response["Configuration"].read()
        if raw:
            # Non-empty response means config changed (or first fetch)
            _config_cache = json.loads(raw)
            _cache_loaded_at = now
            logger.info(
                "AppConfig model routing config refreshed: default_model=%s",
                _config_cache.get("default_model_id"),
            )
        else:
            # Empty body = no change since last poll; keep existing cache
            logger.debug("AppConfig returned empty body — no config change")

    except Exception as exc:
        logger.error("AppConfig fetch failed: %s — using last known config", exc)
        if _config_cache is None:
            raise RuntimeError(
                "AppConfig unreachable and no cached config available"
            ) from exc

    return _config_cache


def resolve_model_id(intent: str) -> str:
    """
    Resolve the Bedrock model ID for a given intent using AppConfig routing rules.

    Cost override (set in AppConfig when cost alarm fires) forces all traffic
    to Haiku regardless of intent-level overrides.
    """
    config = get_model_routing_config()

    if config.get("cost_override_active", False):
        model_id = "anthropic.claude-3-haiku-20240307-v1:0"
        logger.info("Cost override active — routing all traffic to Haiku")
        return model_id

    model_id = config.get("intent_overrides", {}).get(
        intent, config["default_model_id"]
    )
    logger.info("Resolved model_id=%s for intent=%s", model_id, intent)
    return model_id


def lambda_handler(event: dict, context: Any) -> dict:
    """
    MangaAssist inference Lambda — model ID read from AppConfig, not source code.
    """
    bedrock = boto3.client("bedrock-runtime", region_name=os.environ.get("AWS_REGION", "us-east-1"))

    session_id = event["session_id"]
    user_message = event["message"]
    intent = event.get("intent", "general_qa")

    model_id = resolve_model_id(intent)

    payload = {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": user_message}],
    }

    response = bedrock.invoke_model(
        modelId=model_id,
        body=json.dumps(payload),
        contentType="application/json",
        accept="application/json",
    )

    result = json.loads(response["body"].read())
    answer = result["content"][0]["text"]

    logger.info(
        "Inference complete: session=%s intent=%s model=%s tokens_in=%d tokens_out=%d",
        session_id,
        intent,
        model_id,
        result["usage"]["input_tokens"],
        result["usage"]["output_tokens"],
    )

    return {"session_id": session_id, "answer": answer, "model_id": model_id}

flowchart TD
    A["CloudWatch Alarm:<br/>MangaAssist p99 latency > 2.5s"] --> B{"Check X-Ray<br/>service map"}
    B -->|"AppConfig segment<br/>> 180ms present on<br/>every trace"| C["AppConfig call<br/>on every invocation confirmed"]
    B -->|"Bedrock alone<br/>accounts for latency"| D["Model performance issue —<br/>Different Runbook"]
    C --> E{"Check Lambda<br/>CloudWatch Logs"}
    E -->|"'AppConfig model routing<br/>config refreshed' on<br/>EVERY invocation"| F["ROOT CAUSE:<br/>No module-level caching —<br/>fresh HTTP call per request"]
    E -->|"Throttle errors<br/>(ThrottlingException)"| G["AppConfig throttling<br/>compounding latency"]
    F --> H["Apply Runbook 2:<br/>Session-token polling loop"]
    G --> H

"""
Runbook 2: Fix AppConfig per-request fetches with session-token caching.
The fix is additive — replace the legacy GetConfiguration pattern with
the appconfigdata session pattern shown in Runbook 1, ensuring the
module-level globals (_config_session_token, _config_cache) persist
across Lambda warm invocations.

This snippet shows the broken pattern and the corrected replacement.
"""

import json
import logging
import os
import time

import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger("manga_appconfig_fix")
logger.setLevel(logging.INFO)

# ── BROKEN PATTERN (do NOT use) ─────────────────────────────────────────────
def _broken_get_config_every_call() -> dict:
    """
    Anti-pattern: calls appconfig.GetConfiguration on every Lambda invocation.
    Adds 150-250ms and risks throttling at scale.
    """
    client = boto3.client("appconfig")  # Legacy client
    response = client.get_configuration(
        Application="manga-assist",
        Environment="production",
        Configuration="model-routing",
        ClientId="manga-lambda",  # Same client ID = no cache benefit
    )
    return json.loads(response["Content"].read() or b"{}")


# ── CORRECT PATTERN ──────────────────────────────────────────────────────────
_appconfig_data_client = boto3.client(
    "appconfigdata",
    region_name=os.environ.get("AWS_REGION", "us-east-1"),
)

# These persist across warm invocations within the same Lambda execution environment
_session_token: str | None = None
_cached_config: dict | None = None
_last_poll_time: float = 0.0

POLL_INTERVAL_SECONDS = int(os.environ.get("APPCONFIG_POLL_SECONDS", "60"))


def _ensure_session_token() -> str:
    """Start a new AppConfig data session if one does not exist."""
    global _session_token
    if _session_token is None:
        resp = _appconfig_data_client.start_configuration_session(
            ApplicationIdentifier=os.environ["APPCONFIG_APP"],
            EnvironmentIdentifier=os.environ["APPCONFIG_ENV"],
            ConfigurationProfileIdentifier=os.environ["APPCONFIG_PROFILE"],
            RequiredMinimumPollIntervalInSeconds=POLL_INTERVAL_SECONDS,
        )
        _session_token = resp["InitialConfigurationToken"]
        logger.info("AppConfig session started (cold start or token expiry)")
    return _session_token


def get_model_config_cached() -> dict:
    """
    Return model routing config, fetching from AppConfig at most once per
    POLL_INTERVAL_SECONDS across warm invocations.

    Benchmark improvement:
      Before fix  — AppConfig call on every invocation: ~200ms added
      After fix   — AppConfig call every 60s (once per poll interval): ~0ms added
                    on warm invocations; session token call on cold start: ~40ms
    """
    global _session_token, _cached_config, _last_poll_time

    now = time.monotonic()
    if _cached_config is not None and (now - _last_poll_time) < POLL_INTERVAL_SECONDS:
        # Cache is fresh — return immediately, no HTTP call
        return _cached_config

    token = _ensure_session_token()

    try:
        start = time.monotonic()
        resp = _appconfig_data_client.get_latest_configuration(
            ConfigurationToken=token
        )
        elapsed_ms = (time.monotonic() - start) * 1000

        # MUST always update the token — it is rotated on every API call
        _session_token = resp["NextPollConfigurationToken"]

        body = resp["Configuration"].read()
        if body:
            _cached_config = json.loads(body)
            _last_poll_time = now
            logger.info(
                "AppConfig config updated in %.1f ms: %s",
                elapsed_ms,
                json.dumps(_cached_config),
            )
        else:
            # Empty body = no config change, keep existing cache
            _last_poll_time = now  # Reset timer so we don't poll again immediately
            logger.debug("AppConfig: no config change (%.1f ms)", elapsed_ms)

    except ClientError as exc:
        error_code = exc.response["Error"]["Code"]
        if error_code == "BadRequestException" and _cached_config is not None:
            # Token may have expired (e.g. Lambda was paused > 24h)
            # Reset session and fall back to cached data
            logger.warning(
                "AppConfig token invalid, resetting session — using cached config"
            )
            _session_token = None
        else:
            logger.error("AppConfig GetLatestConfiguration failed: %s", exc)
            if _cached_config is None:
                raise

    return _cached_config


def lambda_handler(event: dict, context) -> dict:
    """
    Demonstrates: AppConfig config retrieved without HTTP overhead on warm invocations.
    Only the FIRST call per poll interval (or cold start) hits AppConfig.
    """
    config = get_model_config_cached()  # Zero HTTP cost on warm invoke within poll window

    intent = event.get("intent", "general_qa")
    model_id = (
        config.get("intent_overrides", {}).get(intent)
        or config["default_model_id"]
    )

    logger.info("Model selected: %s for intent: %s", model_id, intent)
    return {"model_id": model_id, "intent": intent}

flowchart TD
    A["PagerDuty P1 Alert:<br/>MangaAssist error rate > 10%"] --> B{"Check Lambda<br/>CloudWatch Logs"}
    B -->|"KeyError: 'anthropic_version'<br/>in build_payload()"| C["Config schema<br/>violation detected"]
    B -->|"ValidationException<br/>from Bedrock"| D["Also config-related —<br/>check payload construction"]
    C --> E{"When did errors start?"}
    D --> E
    E -->|"~60s ago<br/>(AppConfig poll cycle)"| F["Check AppConfig<br/>deployment history"]
    F -->|"New deployment<br/>found at T-60s"| G["ROOT CAUSE:<br/>Bad config deployed —<br/>no schema validation<br/>on AppConfig document"]
    G --> H{"Is AppConfig rollback<br/>available?"}
    H -->|"Yes"| I["Roll back to previous<br/>AppConfig deployment version"]
    H -->|"No"| J["Manually push corrected<br/>config document via Runbook 3"]
    I --> K["Monitor error rate<br/>drops to baseline"]
    J --> K

"""
Runbook 3: Schema validation for AppConfig model routing documents.
Two parts:
  (a) CI validation script — runs in GitHub Actions before AppConfig deployment
  (b) Lambda-side defensive validation — rejects bad config at fetch time,
      falls back to last known good config, and fires a CloudWatch alarm
"""

import json
import logging
import os
import time

import boto3
import jsonschema
from jsonschema import ValidationError

logger = logging.getLogger("manga_config_validator")
logger.setLevel(logging.INFO)

# ── CONFIG SCHEMA ────────────────────────────────────────────────────────────
MODEL_ROUTING_SCHEMA = {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "title": "MangaAssist Model Routing Config",
    "type": "object",
    "required": ["default_model_id", "intent_overrides", "cost_override_active"],
    "additionalProperties": False,
    "properties": {
        "default_model_id": {
            "type": "string",
            "pattern": r"^anthropic\.",
            "description": "Bedrock model ID used when no intent override matches",
        },
        "intent_overrides": {
            "type": "object",
            "additionalProperties": {
                "type": "string",
                "pattern": r"^anthropic\.",
            },
            "description": "Per-intent model ID overrides",
        },
        "cost_override_active": {
            "type": "boolean",
            "description": "When true, routes all traffic to Haiku regardless of intent",
        },
        "model_registry": {
            "type": "array",
            "items": {
                "type": "object",
                "required": [
                    "model_id",
                    "anthropic_version",
                    "max_tokens_default",
                    "cost_tier",
                ],
                "additionalProperties": False,
                "properties": {
                    "model_id": {"type": "string", "pattern": r"^anthropic\."},
                    "anthropic_version": {
                        "type": "string",
                        "enum": ["bedrock-2023-05-31"],
                    },
                    "max_tokens_default": {"type": "integer", "minimum": 1, "maximum": 8192},
                    "cost_tier": {"type": "string", "enum": ["low", "medium", "high"]},
                },
            },
        },
    },
}


def validate_model_routing_config(config_doc: dict) -> None:
    """
    Validate a model routing config document against MODEL_ROUTING_SCHEMA.
    Raises jsonschema.ValidationError on failure.

    Use this function in:
      1. CI pipeline (pre-AppConfig-deploy) — raises and fails the pipeline
      2. Lambda fetch handler — catches and falls back to last known good config
    """
    jsonschema.validate(instance=config_doc, schema=MODEL_ROUTING_SCHEMA)
    logger.info("Config schema validation passed")


# ── CI SCRIPT (run as: python validate_appconfig.py <config_file.json>) ──────
def ci_validate_config_file(file_path: str) -> None:
    """Entry point for CI pipeline schema check before AppConfig deployment."""
    with open(file_path, "r", encoding="utf-8") as f:
        doc = json.load(f)
    try:
        validate_model_routing_config(doc)
        print(f"[PASS] {file_path} passes schema validation")
    except ValidationError as exc:
        print(f"[FAIL] {file_path} schema violation: {exc.message}")
        raise SystemExit(1) from exc


# ── LAMBDA-SIDE DEFENSIVE FETCH ──────────────────────────────────────────────
_appconfig_client = boto3.client(
    "appconfigdata", region_name=os.environ.get("AWS_REGION", "us-east-1")
)
_cloudwatch = boto3.client("cloudwatch", region_name=os.environ.get("AWS_REGION", "us-east-1"))

_session_token: str | None = None
_last_known_good_config: dict | None = None
_last_poll_time: float = 0.0
POLL_INTERVAL_SECONDS = int(os.environ.get("APPCONFIG_POLL_SECONDS", "60"))


def _emit_invalid_config_alarm() -> None:
    """Push a custom CloudWatch metric so the ops alarm fires within 60 seconds."""
    try:
        _cloudwatch.put_metric_data(
            Namespace="MangaAssist/ModelRouting",
            MetricData=[
                {
                    "MetricName": "InvalidConfigRejected",
                    "Value": 1.0,
                    "Unit": "Count",
                }
            ],
        )
    except Exception as cw_exc:
        logger.error("Failed to emit InvalidConfigRejected metric: %s", cw_exc)


def get_validated_model_config() -> dict:
    """
    Fetch AppConfig model routing with schema validation.
    Rejects and discards invalid documents; falls back to last known good config
    and emits a CloudWatch alarm metric.
    """
    global _session_token, _last_known_good_config, _last_poll_time

    now = time.monotonic()
    if _last_known_good_config and (now - _last_poll_time) < POLL_INTERVAL_SECONDS:
        return _last_known_good_config

    if _session_token is None:
        resp = _appconfig_client.start_configuration_session(
            ApplicationIdentifier=os.environ["APPCONFIG_APP"],
            EnvironmentIdentifier=os.environ["APPCONFIG_ENV"],
            ConfigurationProfileIdentifier=os.environ["APPCONFIG_PROFILE"],
            RequiredMinimumPollIntervalInSeconds=POLL_INTERVAL_SECONDS,
        )
        _session_token = resp["InitialConfigurationToken"]

    resp = _appconfig_client.get_latest_configuration(
        ConfigurationToken=_session_token
    )
    _session_token = resp["NextPollConfigurationToken"]

    body = resp["Configuration"].read()
    if body:
        try:
            candidate = json.loads(body)
            validate_model_routing_config(candidate)  # Raises if invalid
            _last_known_good_config = candidate
            _last_poll_time = now
            logger.info("New valid config accepted: %s", json.dumps(candidate))
        except (json.JSONDecodeError, ValidationError) as exc:
            logger.error(
                "INVALID AppConfig document rejected — keeping last known good: %s", exc
            )
            _emit_invalid_config_alarm()
            # Do NOT update _last_known_good_config — the bad doc is discarded
    else:
        _last_poll_time = now

    if _last_known_good_config is None:
        raise RuntimeError("No valid model routing config available and no fallback")

    return _last_known_good_config

flowchart TD
    A["Support tickets:<br/>'Chatbot changed personality<br/>mid-conversation'"] --> B{"Check CloudWatch Logs<br/>for session_id across invocations"}
    B -->|"Same session_id,<br/>different model_id<br/>values in logs"| C["Model skew confirmed:<br/>multiple models serving<br/>same session"]
    B -->|"Consistent model_id<br/>across session"| D["Not a model skew issue —<br/>check prompt or context"]
    C --> E{"When did mixed<br/>model_id start appearing?"}
    E -->|"~T-10 minutes:<br/>Lambda env var update"| F["ROOT CAUSE:<br/>Rolling Lambda update<br/>via env var caused<br/>in-flight version skew"]
    F --> G{"All instances<br/>updated now?"}
    G -->|"Yes — skew resolved"| H["Incident closed —<br/>implement prevention"]
    G -->|"No — skew still active"| I["Force all execution\nenvironments to update:\nDeploy dummy code change\nor increase Lambda concurrency\nto drain old environments"]
    H --> J["Migrate to AppConfig-based<br/>switching — Runbook 4"]
    I --> J

"""
Runbook 4: Replace Lambda env var model switching with AppConfig atomic flag.
Demonstrates:
  (a) Why env var switching causes skew (the broken pattern)
  (b) AppConfig-based switching that propagates atomically to all warm instances
      within one poll cycle (~60 seconds) with no rolling deployment
"""

import json
import logging
import os
import time

import boto3

logger = logging.getLogger("manga_model_switch")
logger.setLevel(logging.INFO)

# ── BROKEN PATTERN: env var read at module import time ───────────────────────
# This value is frozen at cold start for the lifetime of the execution env.
# When an env var update is deployed, old execution envs retain the old value
# until they are recycled.
_BROKEN_MODEL_ID = os.environ.get(
    "DEFAULT_MODEL_ID",
    "anthropic.claude-3-sonnet-20240229-v1:0",
)


def _broken_get_model_id_from_env() -> str:
    """Anti-pattern: returns the value captured at import time."""
    return _BROKEN_MODEL_ID  # Stale for old execution environments during rollout


# ── CORRECT PATTERN: AppConfig-backed switching ───────────────────────────────
_appconfig = boto3.client("appconfigdata", region_name=os.environ.get("AWS_REGION", "us-east-1"))
_session_token: str | None = None
_config_cache: dict | None = None
_last_poll: float = 0.0
POLL_SECONDS = int(os.environ.get("APPCONFIG_POLL_SECONDS", "60"))


def _get_appconfig_routing() -> dict:
    """Session-token-cached AppConfig fetch (same pattern as Runbooks 1–3)."""
    global _session_token, _config_cache, _last_poll
    now = time.monotonic()
    if _config_cache and (now - _last_poll) < POLL_SECONDS:
        return _config_cache
    if _session_token is None:
        resp = _appconfig.start_configuration_session(
            ApplicationIdentifier=os.environ["APPCONFIG_APP"],
            EnvironmentIdentifier=os.environ["APPCONFIG_ENV"],
            ConfigurationProfileIdentifier=os.environ["APPCONFIG_PROFILE"],
            RequiredMinimumPollIntervalInSeconds=POLL_SECONDS,
        )
        _session_token = resp["InitialConfigurationToken"]
    resp = _appconfig.get_latest_configuration(ConfigurationToken=_session_token)
    _session_token = resp["NextPollConfigurationToken"]
    body = resp["Configuration"].read()
    if body:
        _config_cache = json.loads(body)
        _last_poll = now
        logger.info("AppConfig refreshed: default_model=%s", _config_cache.get("default_model_id"))
    return _config_cache


def get_model_id_for_session(session_id: str, intent: str) -> str:
    """
    Return the model ID from AppConfig, read at request time (not import time).

    When an operator flips `default_model_id` in AppConfig, ALL warm Lambda
    instances pick up the change within POLL_SECONDS — no rolling deployment,
    no version skew, no in-flight inconsistency.
    """
    config = _get_appconfig_routing()
    if config.get("cost_override_active", False):
        model_id = "anthropic.claude-3-haiku-20240307-v1:0"
    else:
        model_id = config.get("intent_overrides", {}).get(
            intent, config["default_model_id"]
        )
    logger.info(
        "session=%s intent=%s resolved_model=%s", session_id, intent, model_id
    )
    return model_id


def lambda_handler(event: dict, context) -> dict:
    """Inference handler: model ID resolved at request time from AppConfig."""
    bedrock = boto3.client("bedrock-runtime", region_name=os.environ.get("AWS_REGION", "us-east-1"))
    session_id = event["session_id"]
    intent = event.get("intent", "general_qa")
    user_message = event["message"]

    model_id = get_model_id_for_session(session_id, intent)

    payload = json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1024,
        "messages": [{"role": "user", "content": user_message}],
    })

    response = bedrock.invoke_model(
        modelId=model_id,
        body=payload,
        contentType="application/json",
        accept="application/json",
    )
    answer = json.loads(response["body"].read())["content"][0]["text"]

    # Emit model_id to CloudWatch Logs for skew detection queries
    logger.info(
        json.dumps({
            "event": "inference_complete",
            "session_id": session_id,
            "model_id": model_id,
            "intent": intent,
        })
    )
    return {"session_id": session_id, "answer": answer, "model_id": model_id}

flowchart TD
    A["Model usage audit:<br/>product_recommendation intent<br/>split ~50/50 Haiku vs Sonnet"] --> B{"Check routing rule<br/>evaluation order in AppConfig"}
    B -->|"Both cost-router AND<br/>latency-router rules<br/>match same intent"| C["Routing rule conflict<br/>confirmed"]
    B -->|"Single rule matches"| D["Audit logging gap —<br/>check which router<br/>is executing"]
    C --> E{"Check rule<br/>priority field"}
    E -->|"Both rules have<br/>same priority value<br/>or no priority field"| F["ROOT CAUSE:<br/>Ambiguous priority —<br/>last rule wins<br/>(non-deterministic under<br/>concurrent updates)"]
    E -->|"Priority field present<br/>but both rules set<br/>to same value"| F
    F --> G["Apply Runbook 5:<br/>Priority-ordered rule<br/>evaluation with<br/>conflict detection"]
    G --> H["Audit historical logs<br/>to assess which sessions<br/>were affected — file<br/>data quality incident"]

"""
Runbook 5: Priority-ordered routing rule evaluation with conflict detection.
Replaces independent cost/latency router composition with a single unified
rule engine that:
  1. Evaluates all rules and collects all matches
  2. Detects conflicts (multiple matches for same intent)
  3. Resolves via explicit priority field
  4. Emits a CloudWatch alarm if a conflict is detected (so rules are fixed)
"""

import json
import logging
import os
import time
from dataclasses import dataclass
from typing import Any

import boto3

logger = logging.getLogger("manga_routing_engine")
logger.setLevel(logging.INFO)

_cloudwatch = boto3.client("cloudwatch", region_name=os.environ.get("AWS_REGION", "us-east-1"))


@dataclass(frozen=True)
class RoutingRule:
    """A single model routing rule with an explicit priority."""
    rule_id: str
    strategy: str            # "cost" | "latency" | "quality"
    intent_pattern: str      # Exact intent string this rule matches
    model_id: str            # Bedrock model ID to use
    priority: int            # Lower number = higher priority (1 = highest)
    enabled: bool


def parse_routing_rules(config_doc: dict) -> list[RoutingRule]:
    """
    Parse AppConfig document into RoutingRule objects.

    Expected config shape:
    {
      "default_model_id": "anthropic.claude-3-haiku-20240307-v1:0",
      "cost_override_active": false,
      "routing_rules": [
        {
          "rule_id": "cost-haiku-product-rec",
          "strategy": "cost",
          "intent_pattern": "product_recommendation",
          "model_id": "anthropic.claude-3-haiku-20240307-v1:0",
          "priority": 2,
          "enabled": true
        },
        {
          "rule_id": "latency-sonnet-product-rec",
          "strategy": "latency",
          "intent_pattern": "product_recommendation",
          "model_id": "anthropic.claude-3-sonnet-20240229-v1:0",
          "priority": 1,
          "enabled": true
        }
      ]
    }
    """
    rules = []
    for raw in config_doc.get("routing_rules", []):
        rules.append(
            RoutingRule(
                rule_id=raw["rule_id"],
                strategy=raw["strategy"],
                intent_pattern=raw["intent_pattern"],
                model_id=raw["model_id"],
                priority=raw["priority"],
                enabled=raw.get("enabled", True),
            )
        )
    return rules


def _emit_conflict_metric(intent: str, conflicting_rule_ids: list[str]) -> None:
    """Emit a CloudWatch metric when a routing conflict is detected."""
    try:
        _cloudwatch.put_metric_data(
            Namespace="MangaAssist/ModelRouting",
            MetricData=[
                {
                    "MetricName": "RoutingRuleConflict",
                    "Dimensions": [{"Name": "Intent", "Value": intent}],
                    "Value": 1.0,
                    "Unit": "Count",
                }
            ],
        )
        logger.warning(
            "Routing conflict detected for intent=%s rules=%s",
            intent,
            conflicting_rule_ids,
        )
    except Exception as exc:
        logger.error("Failed to emit RoutingRuleConflict metric: %s", exc)


def resolve_model_for_intent(
    intent: str,
    routing_rules: list[RoutingRule],
    default_model_id: str,
    cost_override_active: bool = False,
) -> tuple[str, str]:
    """
    Deterministically resolve a model ID for a given intent.

    Returns: (model_id, winning_rule_id_or_'default')

    Resolution algorithm:
    1. If cost_override_active → always return Haiku
    2. Filter to enabled rules matching intent
    3. If zero matches → return default_model_id
    4. If one match → return that rule's model_id
    5. If multiple matches → emit conflict alarm, use lowest priority number (highest priority)
    """
    if cost_override_active:
        logger.info("Cost override active — routing intent=%s to Haiku", intent)
        return "anthropic.claude-3-haiku-20240307-v1:0", "cost_override"

    matching_rules = [
        r for r in routing_rules if r.enabled and r.intent_pattern == intent
    ]

    if not matching_rules:
        logger.info(
            "No routing rule matched intent=%s — using default=%s",
            intent,
            default_model_id,
        )
        return default_model_id, "default"

    if len(matching_rules) > 1:
        # Conflict detected — emit alarm, then resolve deterministically by priority
        rule_ids = [r.rule_id for r in matching_rules]
        _emit_conflict_metric(intent, rule_ids)
        # Sort ascending by priority — lowest number wins
        matching_rules = sorted(matching_rules, key=lambda r: r.priority)
        logger.warning(
            "Conflict resolved via priority: winning_rule=%s over %s",
            matching_rules[0].rule_id,
            rule_ids,
        )

    winner = matching_rules[0]
    logger.info(
        "Routing rule applied: intent=%s rule=%s strategy=%s model=%s",
        intent,
        winner.rule_id,
        winner.strategy,
        winner.model_id,
    )
    return winner.model_id, winner.rule_id


def lambda_handler(event: dict, context: Any) -> dict:
    """
    MangaAssist inference Lambda using unified priority-ordered routing engine.
    """
    # (AppConfig fetch omitted for brevity — use get_validated_model_config from Runbook 3)
    config = json.loads(os.environ.get("_TEST_CONFIG_OVERRIDE", "{}"))  # test hook
    rules = parse_routing_rules(config)

    intent = event.get("intent", "general_qa")
    default_model = config.get("default_model_id", "anthropic.claude-3-haiku-20240307-v1:0")
    cost_override = config.get("cost_override_active", False)

    model_id, rule_id = resolve_model_for_intent(
        intent, rules, default_model, cost_override
    )

    # Structured log for downstream audit queries
    logger.info(
        json.dumps({
            "event": "model_resolved",
            "session_id": event.get("session_id"),
            "intent": intent,
            "model_id": model_id,
            "winning_rule": rule_id,
        })
    )

    return {"model_id": model_id, "winning_rule": rule_id}