src/ad_creative_env/web/app.py

"""Thin FastAPI host for the H&M episode service and static frontend."""

from __future__ import annotations

import asyncio
import base64
import copy
import json
import logging
import os
import queue
import threading
from dataclasses import asdict
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Literal

from fastapi import FastAPI, HTTPException, Request
from fastapi.exceptions import RequestValidationError
from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel, ConfigDict, Field

from ad_creative_env.cached_models import (
    BASELINE_MODEL_SLUG,
    CACHED_MODEL_BY_SLUG,
    ordered_slugs,
)
from ad_creative_env.config import GEN_PARAMS, OPENROUTER_IMAGE_URL, PINNED_IMAGE_MODEL
from ad_creative_env.hm.agent import (
    AGENT_DECISION_RESPONSE_SCHEMA,
    AGENT_PROMPT_VERSION,
    AgentDecisionError,
    AutonomousAgentRunner,
    DeterministicDecisionModel,
    LiveDecisionModel,
)
from ad_creative_env.hm.generation import (
    AD_COPY_RESPONSE_SCHEMA,
    COPY_PROMPT_VERSION,
    AdapterIdentity,
    DeterministicCopyGenerator,
    GeneratedCopy,
    GenerationError,
    LiveCopyGenerator,
)
from ad_creative_env.hm.judging import (
    JUDGE_PROMPT_VERSION,
    JUDGE_RESPONSE_SCHEMA,
    DeterministicJudge,
    JudgeError,
    Judgement,
    JudgeScores,
    LiveJudge,
)
from ad_creative_env.hm.models import (
    ActionProvenance,
    ActionSource,
    AdCopy,
    DataProvenance,
    ExecutionMode,
    ValidationError,
)
from ad_creative_env.hm.scenarios import (
    HMLocalScenarioRepository,
    ScenarioNotFoundError,
    SyntheticDemoScenarioRepository,
)
from ad_creative_env.hm.service import EpisodeError, EpisodeService
from ad_creative_env.hm.sessions import MultiStepSessionService, SessionError
from ad_creative_env.hm.text_provider import TextModelTransport, TextProviderError
from ad_creative_env.renderer.provider import ProviderError, generate_image, reset_spend
from ad_creative_env.studio.electronics import (
    ElectronicsStudioService,
    StudioCacheMiss,
    StudioDataError,
    StudioJudgeError,
    StudioScenarioNotFound,
)
from ad_creative_env.studio.electronics_agent import (
    ELECTRONICS_AGENT_PROMPT_VERSION,
    ELECTRONICS_AGENT_RESPONSE_SCHEMA,
)
from ad_creative_env.studio.electronics_judge import (
    ELECTRONICS_JUDGE_PROMPT_VERSION,
    ELECTRONICS_JUDGE_RESPONSE_SCHEMA,
)

_APP_DIR = Path(__file__).resolve().parents[3]
_STATIC_DIR = _APP_DIR / "ui" / "static"
_IDENTITY_PATTERN = r"^[A-Za-z0-9][A-Za-z0-9._:-]*$"
_OPENROUTER_ENDPOINT = "https://openrouter.ai/api/v1/chat/completions"
_server_logger = logging.getLogger("uvicorn.error")
_CACHED_STAGE_IDS = (
    ("scenario_loaded", "Scenario loaded"),
    ("copy_generated", "Ad text generated"),
    ("output_validated", "Output validated"),
    ("image_loaded", "Image loaded"),
    ("card_composed", "Card composed"),
    ("deterministic_checks", "Deterministic checks run"),
    ("judge_run", "Judge run"),
    ("reward_calculated", "Reward calculated"),
)


class CachedRunStore:
    """Read-only playback of previously completed, public-safe episode results."""

    def __init__(self, report_path: Path) -> None:
        self._report_path = report_path.resolve()
        try:
            document = json.loads(self._report_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError, TypeError) as exc:
            raise RuntimeError("cached run report cannot be read") from exc
        episodes = document.get("episodes") if isinstance(document, dict) else None
        if not isinstance(episodes, list) or not episodes:
            raise RuntimeError("cached run report contains no episodes")
        self._episodes: dict[str, dict[str, Any]] = {}
        for episode in episodes:
            if not isinstance(episode, dict) or not isinstance(episode.get("scenario_id"), str):
                raise RuntimeError("cached run report contains an invalid episode")
            scenario_id = episode["scenario_id"]
            if scenario_id in self._episodes:
                raise RuntimeError("cached run report contains duplicate scenarios")
            self._episodes[scenario_id] = episode

    @property
    def scenario_ids(self) -> frozenset[str]:
        return frozenset(self._episodes)

    def playback(
        self, scenario_id: str, query: str | None
    ) -> tuple[list[dict[str, Any]], dict[str, Any]]:
        source = self._episodes.get(scenario_id)
        if source is None:
            raise EpisodeError("cache_miss", "No cached run exists for this scenario.", ())
        if query is not None and " ".join(query.split()) != " ".join(
            str(source.get("query", "")).split()
        ):
            raise EpisodeError(
                "cache_query_mismatch",
                "Cached runs use the original scenario query. Choose Live models to edit it.",
                (),
            )
        card_path_value = source.get("card_path")
        if not isinstance(card_path_value, str):
            raise EpisodeError("cache_invalid", "The cached card is unavailable.", ())
        card_path = (self._report_path.parent / card_path_value).resolve()
        if not card_path.is_relative_to(self._report_path.parent) or not card_path.is_file():
            raise EpisodeError("cache_invalid", "The cached card is unavailable.", ())
        card_artifact = "data:image/png;base64," + base64.b64encode(card_path.read_bytes()).decode(
            "ascii"
        )

        result = copy.deepcopy(source)
        result.pop("card_path", None)
        result.pop("episode_reported_cost_usd", None)
        result.pop("product_name", None)
        result.pop("query", None)
        action_provenance = result.get("action_provenance", {})
        action_provenance["source"] = "recorded_replay"
        action_provenance["execution_mode"] = "recorded_replay"
        result["action_provenance"] = action_provenance
        result["judge_execution_mode"] = "recorded_replay"
        result["card_artifact"] = card_artifact

        public_by_stage = {
            "copy_generated": {"action": result.get("action")},
            "output_validated": {"action_provenance": action_provenance},
            "card_composed": {"card_artifact": card_artifact},
            "deterministic_checks": {"checks": result.get("checks", [])},
            "judge_run": {
                "judge_scores": result.get("judge_scores", {}),
                "judge_explanations": result.get("judge_explanations", {}),
                "judge_execution_mode": "recorded_replay",
                "judge_identity": result.get("judge_identity", {}),
            },
            "reward_calculated": {
                key: result.get(key)
                for key in (
                    "reward_policy_version",
                    "reward",
                    "review_status",
                    "weighted_components",
                    "base_score",
                    "failed_checks",
                    "applied_cap",
                )
            },
        }
        events = [
            {
                "stage_id": stage_id,
                "label": label,
                "public_data": public_by_stage.get(stage_id),
            }
            for stage_id, label in _CACHED_STAGE_IDS
        ]
        return events, result


class _CachedCopyGenerator:
    """Return one recorded action while the session executes real local tools."""

    def __init__(self, result: dict[str, Any]) -> None:
        identity = result["action_provenance"]["identity"]
        self.identity = AdapterIdentity(
            identity["provider"], identity["model"], identity["config_version"]
        )
        self._action = AdCopy(**result["action"])
        self._metadata = {
            key: str(value)
            for key, value in identity.items()
            if key not in {"provider", "model", "config_version"}
        }

    def generate(self, scenario_id, observation, provenance) -> GeneratedCopy:
        del scenario_id, observation, provenance
        return GeneratedCopy(
            action=self._action,
            execution_mode=ExecutionMode.RECORDED_REPLAY,
            identity=self.identity,
            metadata=self._metadata,
        )


class _CachedJudge:
    """Return one recorded judgement after the real local submission boundary."""

    def __init__(self, result: dict[str, Any]) -> None:
        identity = result["judge_identity"]
        self._identity = AdapterIdentity(
            identity["provider"], identity["model"], identity["config_version"]
        )
        self._metadata = {
            key: str(value)
            for key, value in identity.items()
            if key not in {"provider", "model", "config_version"}
        }
        self._scores = JudgeScores(**result["judge_scores"])
        self._explanations = dict(result["judge_explanations"])

    def judge(self, scenario_id, observation, action, target, provenance) -> Judgement:
        del scenario_id, observation, action, target, provenance
        return Judgement(
            scores=self._scores,
            explanations=self._explanations,
            execution_mode=ExecutionMode.RECORDED_REPLAY,
            identity=self._identity,
            metadata=self._metadata,
        )


def _positive_env(name: str, default: int) -> int:
    raw = os.environ.get(name, str(default))
    try:
        value = int(raw)
    except ValueError as exc:
        raise RuntimeError(f"{name} must be a positive integer") from exc
    if value < 1:
        raise RuntimeError(f"{name} must be a positive integer")
    return value


def _public_budget_limit() -> Decimal:
    raw = os.environ.get("AD_PROVIDER_BUDGET_USD", "")
    try:
        value = Decimal(raw)
    except InvalidOperation as exc:
        raise RuntimeError("AD_PROVIDER_BUDGET_USD must match the provider key limit") from exc
    if value <= 0 or value > Decimal("20"):
        raise RuntimeError("public provider key limit must be between $0 and $20")
    return value


class EvaluationAction(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    headline: str = Field(min_length=1, max_length=60)
    body: str = Field(min_length=1, max_length=180)
    cta: str | None = Field(default=None, max_length=32)


class EvaluationAgentIdentity(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    provider: str = Field(min_length=1, max_length=64, pattern=_IDENTITY_PATTERN)
    model: str = Field(min_length=1, max_length=120, pattern=_IDENTITY_PATTERN)
    config_version: str = Field(min_length=1, max_length=64, pattern=_IDENTITY_PATTERN)
    execution_mode: Literal["live", "recorded_replay", "deterministic_test"] = "deterministic_test"


class EvaluationRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    scenario_id: str = Field(min_length=1, max_length=128, pattern=_IDENTITY_PATTERN)
    action: EvaluationAction
    agent_identity: EvaluationAgentIdentity


class SessionCreateRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    scenario_id: str = Field(min_length=1, max_length=128, pattern=_IDENTITY_PATTERN)
    request: str | None = Field(default=None, min_length=1, max_length=500)


class StudioRunRequest(SessionCreateRequest):
    workflow: Literal["test", "live", "cached"] = "test"
    model: str | None = Field(
        default=None, min_length=1, max_length=64, pattern=_IDENTITY_PATTERN
    )


class CachedRunRequest(SessionCreateRequest):
    model: str | None = Field(
        default=None, min_length=1, max_length=64, pattern=_IDENTITY_PATTERN
    )


class SessionToolRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    arguments: dict[str, Any]


class SessionCopyRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    action: EvaluationAction
    agent_identity: EvaluationAgentIdentity


class SessionSubmitRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    artifact_ref: str = Field(min_length=1, max_length=128, pattern=_IDENTITY_PATTERN)
    request_version: int = Field(ge=1)


class SessionRevisionRequest(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)

    request: str = Field(min_length=1, max_length=500)


def _default_service() -> EpisodeService:
    asset_dir = Path(os.environ.get("AD_SYNTHETIC_ASSET_DIR", "/tmp/ad-creative-env/assets"))
    deployment_profile = os.environ.get("AD_DEPLOYMENT_PROFILE", "local")
    scenario_source = os.environ.get("AD_SCENARIO_SOURCE", "synthetic_demo")
    requested_mode = os.environ.get("AD_EXECUTION_MODE", "deterministic_test")
    requested_judge_mode = os.environ.get("AD_JUDGE_MODE", "deterministic_test")
    if deployment_profile not in {"local", "public_anonymous", "public_budgeted"}:
        raise RuntimeError(f"unsupported AD_DEPLOYMENT_PROFILE: {deployment_profile}")
    if scenario_source not in {"synthetic_demo", "hm_local", "hm_pilot"}:
        raise RuntimeError(f"unsupported AD_SCENARIO_SOURCE: {scenario_source}")
    if scenario_source == "hm_local":
        if deployment_profile != "local":
            raise RuntimeError("hm_local scenarios are restricted to the local deployment profile")
        repository = HMLocalScenarioRepository(
            Path(os.environ.get("AD_HM_CSV_PATH", _APP_DIR.parent / "data/hm_flat_sample.csv")),
            Path(os.environ.get("AD_HM_IMAGE_ROOT", _APP_DIR.parent / "data/hm_images")),
        )
        allowed_live_provenance = frozenset({DataProvenance.HM_LOCAL})
    elif scenario_source == "hm_pilot":
        if deployment_profile not in {"local", "public_budgeted"}:
            raise RuntimeError("hm_pilot scenarios require a cleared deployment profile")
        pilot_dir = _APP_DIR / "assets/hm_pilot"
        repository = HMLocalScenarioRepository(
            pilot_dir / "scenarios.csv",
            pilot_dir / "images",
            provenance=DataProvenance.HM_PILOT,
        )
        allowed_live_provenance = frozenset({DataProvenance.HM_PILOT})
    else:
        repository = SyntheticDemoScenarioRepository(asset_dir)
        allowed_live_provenance = frozenset({DataProvenance.SYNTHETIC_DEMO})
    if requested_mode not in {"deterministic_test", "live", "recorded_replay"}:
        raise RuntimeError(f"unsupported AD_EXECUTION_MODE: {requested_mode}")
    if requested_judge_mode not in {"deterministic_test", "live", "recorded_replay"}:
        raise RuntimeError(f"unsupported AD_JUDGE_MODE: {requested_judge_mode}")

    public_live = deployment_profile == "public_budgeted"
    if (
        public_live
        and (requested_mode == "live" or requested_judge_mode == "live")
        and os.environ.get("OPENROUTER_API_KEY")
    ):
        _public_budget_limit()

    if requested_mode == "deterministic_test":
        generator = DeterministicCopyGenerator()
    elif (
        requested_mode == "live"
        and deployment_profile in {"local", "public_budgeted"}
        and os.environ.get("OPENROUTER_API_KEY")
    ):
        copy_identity = AdapterIdentity(
            "openrouter",
            os.environ.get("AD_COPY_MODEL", "qwen/qwen3-32b"),
            COPY_PROMPT_VERSION,
        )
        copy_transport = TextModelTransport(
            identity=copy_identity,
            endpoint=_OPENROUTER_ENDPOINT,
            api_key=os.environ.get("OPENROUTER_API_KEY"),
            max_requests=_positive_env("AD_COPY_MAX_REQUESTS", 25),
            max_attempts=_positive_env("AD_TEXT_MAX_ATTEMPTS", 2),
            request_style="openrouter_chat",
            max_output_tokens=250,
            temperature=0.2,
            response_schema=AD_COPY_RESPONSE_SCHEMA,
            reasoning_enabled=False,
        )
        generator = LiveCopyGenerator(
            copy_identity,
            copy_transport.invoke,
            allowed_provenance=allowed_live_provenance,
            reset_live_counters=copy_transport.reset_request_count,
        )
    else:
        code = (
            "public_live_disabled"
            if deployment_profile == "public_anonymous" and requested_mode == "live"
            else f"{requested_mode}_unconfigured"
        )

        class UnconfiguredGenerator:
            def generate(self, scenario_id, observation, provenance):
                del scenario_id, observation, provenance
                raise GenerationError(
                    code,
                    f"{requested_mode} execution requires approved provider/replay configuration.",
                )

        generator = UnconfiguredGenerator()

    if requested_judge_mode == "deterministic_test":
        judge = DeterministicJudge()
    elif (
        requested_judge_mode == "live"
        and deployment_profile in {"local", "public_budgeted"}
        and os.environ.get("OPENROUTER_API_KEY")
    ):
        judge_identity = AdapterIdentity(
            "openrouter",
            os.environ.get("AD_JUDGE_MODEL", "google/gemini-2.5-flash-lite"),
            JUDGE_PROMPT_VERSION,
        )
        judge_transport = TextModelTransport(
            identity=judge_identity,
            endpoint=_OPENROUTER_ENDPOINT,
            api_key=os.environ.get("OPENROUTER_API_KEY"),
            max_requests=_positive_env("AD_JUDGE_MAX_REQUESTS", 25),
            max_attempts=_positive_env("AD_TEXT_MAX_ATTEMPTS", 2),
            request_style="openrouter_chat",
            max_output_tokens=600,
            temperature=0.0,
            response_schema=JUDGE_RESPONSE_SCHEMA,
            reasoning_enabled=False,
        )
        judge = LiveJudge(
            judge_identity,
            judge_transport.invoke,
            allowed_provenance=allowed_live_provenance,
            reset_live_counters=judge_transport.reset_request_count,
        )
    else:
        judge_code = (
            "public_live_disabled"
            if deployment_profile == "public_anonymous" and requested_judge_mode == "live"
            else f"judge_{requested_judge_mode}_unconfigured"
        )

        class UnconfiguredJudge:
            def judge(self, scenario_id, observation, action, target, provenance):
                del scenario_id, observation, action, target, provenance
                raise JudgeError(
                    judge_code,
                    f"{requested_judge_mode} judge requires approved provider/replay configuration.",
                )

        judge = UnconfiguredJudge()
    return EpisodeService(
        repository,
        generator,
        judge,
    )


def _default_decision_model():
    requested_mode = os.environ.get("AD_EXECUTION_MODE", "deterministic_test")
    deployment_profile = os.environ.get("AD_DEPLOYMENT_PROFILE", "local")
    if requested_mode == "deterministic_test":
        return DeterministicDecisionModel()
    if (
        requested_mode == "live"
        and deployment_profile in {"local", "public_budgeted"}
        and os.environ.get("OPENROUTER_API_KEY")
    ):
        identity = AdapterIdentity(
            "openrouter",
            os.environ.get("AD_AGENT_MODEL", os.environ.get("AD_COPY_MODEL", "qwen/qwen3-32b")),
            AGENT_PROMPT_VERSION,
        )
        transport = TextModelTransport(
            identity=identity,
            endpoint=_OPENROUTER_ENDPOINT,
            api_key=os.environ.get("OPENROUTER_API_KEY"),
            max_requests=_positive_env("AD_AGENT_MAX_REQUESTS", 25),
            max_attempts=_positive_env("AD_TEXT_MAX_ATTEMPTS", 2),
            request_style="openrouter_chat",
            max_output_tokens=500,
            temperature=0.0,
            response_schema=AGENT_DECISION_RESPONSE_SCHEMA,
            reasoning_enabled=False,
        )
        return LiveDecisionModel(
            identity, transport.invoke, reset_live_counters=transport.reset_request_count
        )

    class UnconfiguredDecisionModel:
        def decide(self, context):
            del context
            raise AgentDecisionError(
                f"agent_{requested_mode}_unconfigured",
                f"{requested_mode} agent decisions require approved provider configuration.",
            )

    return UnconfiguredDecisionModel()


def _default_studio_service() -> ElectronicsStudioService:
    """Build the optional showcase with bounded live adapters when explicitly configured."""

    live_decision_generator = None
    live_image_generator = None
    live_judge_generator = None
    live_ready = (
        os.environ.get("AD_EXECUTION_MODE") == "live"
        and os.environ.get("AD_DEPLOYMENT_PROFILE", "local") in {"local", "public_budgeted"}
        and bool(os.environ.get("OPENROUTER_API_KEY"))
    )
    if live_ready:
        agent_identity = AdapterIdentity(
            "openrouter",
            os.environ.get("AD_AGENT_MODEL", os.environ.get("AD_COPY_MODEL", "qwen/qwen3-32b")),
            ELECTRONICS_AGENT_PROMPT_VERSION,
        )
        agent_transport = TextModelTransport(
            identity=agent_identity,
            endpoint=_OPENROUTER_ENDPOINT,
            api_key=os.environ.get("OPENROUTER_API_KEY"),
            max_requests=_positive_env("AD_STUDIO_AGENT_MAX_REQUESTS", 20),
            max_attempts=1,
            request_style="openrouter_chat",
            max_output_tokens=500,
            temperature=0.0,
            response_schema=ELECTRONICS_AGENT_RESPONSE_SCHEMA,
            reasoning_enabled=False,
        )

        def generate_live_decision(
            prompt: str,
        ) -> tuple[dict[str, Any] | str, dict[str, str]]:
            response = agent_transport.invoke(prompt)
            output = response.output
            return output, {
                **agent_identity.as_dict(),
                **response.metadata,
            }

        image_model = os.environ.get("AD_IMAGE_MODEL", PINNED_IMAGE_MODEL)
        # Per-episode ceiling, not a dollar guard — mid-task revisions may regenerate the
        # image, and the provider key's $20 hard limit is the only cumulative cap.
        max_images_per_episode = _positive_env("AD_STUDIO_MAX_IMAGES_PER_EPISODE", 6)

        def generate_live_image(prompt: str) -> tuple[bytes, dict[str, Any]]:
            image_bytes, metadata = generate_image(
                prompt,
                model=image_model,
                api_key=os.environ.get("OPENROUTER_API_KEY"),
                params=GEN_PARAMS,
                endpoint=OPENROUTER_IMAGE_URL,
                max_images=max_images_per_episode,
            )
            return image_bytes, {
                "provider": "openrouter",
                "model": image_model,
                "config_version": "electronics-image-live-v1",
                "request_count": "1",
                **metadata,
            }

        studio_judge_identity = AdapterIdentity(
            "openrouter",
            os.environ.get("AD_JUDGE_MODEL", "google/gemini-2.5-flash-lite"),
            ELECTRONICS_JUDGE_PROMPT_VERSION,
        )
        studio_judge_transport = TextModelTransport(
            identity=studio_judge_identity,
            endpoint=_OPENROUTER_ENDPOINT,
            api_key=os.environ.get("OPENROUTER_API_KEY"),
            max_requests=_positive_env("AD_STUDIO_JUDGE_MAX_REQUESTS", 5),
            max_attempts=2,
            request_style="openrouter_chat",
            max_output_tokens=600,
            temperature=0.0,
            response_schema=ELECTRONICS_JUDGE_RESPONSE_SCHEMA,
            reasoning_enabled=False,
        )

        def generate_live_judgement(
            prompt: str,
        ) -> tuple[dict[str, Any] | str, dict[str, str]]:
            response = studio_judge_transport.invoke(prompt)
            return response.output, {
                **studio_judge_identity.as_dict(),
                **response.metadata,
            }

        def reset_live_episode_counters() -> None:
            agent_transport.reset_request_count()
            studio_judge_transport.reset_request_count()
            reset_spend()

        live_decision_generator = generate_live_decision
        live_image_generator = generate_live_image
        live_judge_generator = generate_live_judgement
        live_episode_reset = reset_live_episode_counters
    else:
        live_episode_reset = None

    return ElectronicsStudioService.from_pack(
        _APP_DIR / "assets" / "studio_demo" / "electronics",
        live_image_generator=live_image_generator,
        live_decision_generator=live_decision_generator,
        live_judge_generator=live_judge_generator,
        live_episode_reset=live_episode_reset,
        autonomous_max_turns=_positive_env("AD_STUDIO_AGENT_MAX_TURNS", 20),
        autonomous_max_errors=_positive_env("AD_STUDIO_AGENT_MAX_ERRORS", 3),
    )


def _public_stream_error(exc: Exception) -> dict[str, str]:
    """Map a mid-stream failure to an honest, public-safe NDJSON error payload.

    Provider error bodies may echo private text, so only our own messages and
    classified codes ever reach the browser."""
    if isinstance(exc, StudioDataError):
        return {"code": "studio_run_failed", "message": str(exc)}
    if isinstance(exc, StudioJudgeError):
        return {"code": "judge_failed", "message": str(exc)}
    if isinstance(exc, TextProviderError):
        return {"code": exc.code, "message": str(exc)}
    if isinstance(exc, ProviderError):
        return {
            "code": f"image_provider_{exc.kind}",
            "message": "The live image provider call failed, so the run stopped safely.",
        }
    return {
        "code": "internal_error",
        "message": "The run stopped because of an unexpected internal error.",
    }


def _episode_http_status(code: str) -> int:
    if code in {
        "malformed_json",
        "invalid_shape",
        "missing_fields",
        "extra_fields",
        "invalid_copy",
        "text_overflow",
    }:
        return 422
    if code in {
        "live_unconfigured",
        "recorded_replay_unconfigured",
        "judge_live_unconfigured",
        "judge_recorded_replay_unconfigured",
        "public_live_disabled",
    }:
        return 503
    if code in {"timeout", "provider_error", "rate_limit", "request_cap_exceeded"}:
        return 502
    if code == "budget_exhausted":
        return 503
    return 500


def _scenario_payload(scenario, *, detail: bool) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "scenario_id": scenario.scenario_id,
        "query": scenario.query,
        "product_name": scenario.product.name,
        "product_type": scenario.product.product_type,
        "data_provenance": scenario.provenance.value,
    }
    if detail:
        payload["observation"] = asdict(scenario.to_observation())
    return payload


def _episode_input(payload: dict[str, Any]) -> tuple[str, str | None]:
    if set(payload) - {"scenario_id", "query"}:
        raise HTTPException(
            422, detail={"code": "invalid_request", "message": "Episode request is invalid."}
        )
    scenario_id = payload.get("scenario_id")
    if not isinstance(scenario_id, str) or not scenario_id.strip():
        raise HTTPException(
            422, detail={"code": "invalid_request", "message": "scenario_id is required."}
        )
    query = payload.get("query")
    if query is not None and (not isinstance(query, str) or not query.strip() or len(query) > 500):
        raise HTTPException(
            422,
            detail={
                "code": "invalid_query",
                "message": "query must contain 1 to 500 characters.",
            },
        )
    return scenario_id.strip(), query


def create_app(
    service: EpisodeService | None = None,
    session_service: MultiStepSessionService | None = None,
    decision_model=None,
    static_dir: Path | None = None,
    cached_report_path: Path | None = None,
    studio_service: ElectronicsStudioService | None = None,
) -> FastAPI:
    episode_service = service or _default_service()
    multi_step_service = session_service or MultiStepSessionService(
        episode_service.repository,
        episode_service.generator,
        episode_service.judge,
        compositor=episode_service.compositor,
        reward_policy=episode_service.reward_policy,
    )
    active_decision_model = decision_model or _default_decision_model()
    autonomous_runner = AutonomousAgentRunner(
        active_decision_model,
        max_turns=_positive_env("AD_AGENT_MAX_TURNS", 10),
        max_errors=_positive_env("AD_AGENT_MAX_ERRORS", 3),
    )

    def _reset_live_episode_counters() -> None:
        # Live request caps bound ONE episode; the provider key's hard dollar limit is
        # the only cumulative guard. Without this, a second run inherits an exhausted
        # process-lifetime counter and fails before reaching the network.
        for boundary in (episode_service.generator, episode_service.judge, active_decision_model):
            reset = getattr(boundary, "reset_live_counters", None)
            if reset is not None:
                reset()
    if studio_service is None and os.environ.get("AD_STUDIO_SHOWCASE") == "electronics":
        studio_service = _default_studio_service()
    assets = Path(static_dir) if static_dir is not None else _STATIC_DIR
    configured_report = cached_report_path
    if configured_report is None:
        default_report = (
            _APP_DIR / "assets/hm_pilot/cache/report.json"
            if os.environ.get("AD_SCENARIO_SOURCE") == "hm_pilot"
            else _APP_DIR / "outputs/hm_live_validation_2026-07-14/report.json"
        )
        configured_report = Path(
            os.environ.get(
                "AD_CACHED_RUN_REPORT",
                default_report,
            )
        )
    cached_runs = CachedRunStore(configured_report) if configured_report.is_file() else None
    service_scenario_ids = {scenario.scenario_id for scenario in episode_service.list_scenarios()}
    cached_scenario_ids = (
        service_scenario_ids & cached_runs.scenario_ids if cached_runs is not None else set()
    )
    cached_model_stores: dict[str, CachedRunStore] = {}
    if cached_runs is not None:
        cached_model_stores[BASELINE_MODEL_SLUG] = cached_runs
        models_dir = configured_report.parent / "models"
        if models_dir.is_dir():
            for child in sorted(models_dir.iterdir()):
                if not child.is_dir() or not (child / "report.json").is_file():
                    continue
                if child.name not in CACHED_MODEL_BY_SLUG:
                    raise RuntimeError(
                        f"cached model directory is not in the approved registry: {child.name}"
                    )
                cached_model_stores[child.name] = CachedRunStore(child / "report.json")

    def hm_cached_models(scenario_id: str) -> list[dict[str, str]]:
        available = {
            slug
            for slug, store in cached_model_stores.items()
            if scenario_id in store.scenario_ids
        }
        return [
            {"id": slug, "label": CACHED_MODEL_BY_SLUG[slug].label}
            for slug in ordered_slugs(available)
        ]
    application = FastAPI(title="Ad Studio Agent", version="0.2.0")

    @application.exception_handler(RequestValidationError)
    async def evaluation_validation_error(
        request: Request, exc: RequestValidationError
    ) -> JSONResponse:
        del request, exc
        return JSONResponse(
            status_code=422,
            content={
                "detail": {
                    "code": "invalid_evaluation_request",
                    "message": "Evaluation request is invalid.",
                }
            },
        )

    @application.get("/health")
    def health() -> dict[str, str]:
        return {"status": "ok"}

    @application.get("/api/config")
    def public_config() -> dict[str, Any]:
        execution_mode = os.environ.get("AD_EXECUTION_MODE", "deterministic_test")
        judge_mode = os.environ.get("AD_JUDGE_MODE", "deterministic_test")
        available_run_modes: list[str] = []
        if cached_scenario_ids == service_scenario_ids and service_scenario_ids:
            available_run_modes.append("cached")
        if execution_mode == "live":
            available_run_modes.append("live")
        if not available_run_modes:
            available_run_modes.append("cached")
        available_workflows: list[dict[str, str | bool]] = []
        if cached_scenario_ids == service_scenario_ids and service_scenario_ids:
            available_workflows.append(
                {
                    "id": "cached",
                    "label": "Cached run",
                    "description": "Replay a previously completed run with no model calls.",
                    "supports_refinement": False,
                }
            )
        live_ready = (
            execution_mode == "live"
            and judge_mode == "live"
            and os.environ.get("AD_DEPLOYMENT_PROFILE", "local") in {"local", "public_budgeted"}
            and bool(os.environ.get("OPENROUTER_API_KEY"))
        )
        if live_ready:
            available_workflows.append(
                {
                    "id": "live",
                    "label": "Live generation",
                    "description": (
                        "Let the live agent choose Electronics tools and actions, then generate "
                        "the ad image."
                        if studio_service is not None
                        else "Let the configured agent model choose tools and create the ad text."
                    ),
                    "supports_refinement": studio_service is None,
                }
            )
        elif execution_mode == "deterministic_test" and judge_mode == "deterministic_test":
            available_workflows.append(
                {
                    "id": "test",
                    "label": "Test client",
                    "description": "Exercise the autonomous loop with deterministic local decisions.",
                    "supports_refinement": False,
                }
            )
        return {
            "data_provenance": os.environ.get("AD_SCENARIO_SOURCE", "synthetic_demo"),
            "deployment_profile": os.environ.get("AD_DEPLOYMENT_PROFILE", "local"),
            "execution_mode": execution_mode,
            "judge_mode": judge_mode,
            "available_run_modes": available_run_modes,
            "available_workflows": available_workflows,
            "default_workflow": (available_workflows[0]["id"] if available_workflows else None),
            "cached_mode_label": (
                "Cached run" if cached_scenario_ids == service_scenario_ids else "Baseline run"
            ),
            "agent_model_label": (
                os.environ.get("AD_AGENT_MODEL", os.environ.get("AD_COPY_MODEL", "qwen/qwen3-32b"))
                if execution_mode == "live"
                else "Deterministic baseline"
            ),
            "judge_model_label": (
                os.environ.get("AD_JUDGE_MODEL", "google/gemini-2.5-flash-lite")
                if judge_mode == "live"
                else "Deterministic rubric"
            ),
            "studio_showcase": studio_service is not None,
        }

    @application.get("/api/scenarios")
    def scenarios() -> list[dict[str, Any]]:
        baseline = [
            {
                **_scenario_payload(scenario, detail=False),
                "runtime": "hm",
                "cached_models": hm_cached_models(scenario.scenario_id),
            }
            for scenario in episode_service.list_scenarios()
        ]
        # Scenarios carrying latest-model recorded runs list first; the stable sort
        # keeps the original ordering inside each group.
        baseline.sort(
            key=lambda payload: 0
            if any(model["id"] != BASELINE_MODEL_SLUG for model in payload["cached_models"])
            else 1
        )
        if studio_service is not None:
            return [*studio_service.list_scenarios(), *baseline]
        return baseline

    @application.get("/api/scenarios/{scenario_id}")
    def scenario_detail(scenario_id: str) -> dict[str, Any]:
        try:
            scenario = episode_service.get_scenario(scenario_id)
        except ScenarioNotFoundError as exc:
            if studio_service is not None:
                try:
                    return studio_service.scenario_detail(scenario_id)
                except StudioScenarioNotFound:
                    pass
            raise HTTPException(
                404, detail={"code": "scenario_not_found", "message": "Scenario not found."}
            ) from exc
        return {
            **_scenario_payload(scenario, detail=True),
            "runtime": "hm",
            "cached_models": hm_cached_models(scenario_id),
        }

    @application.post("/api/studio-session-runs/stream")
    def stream_studio_session_run(payload: StudioRunRequest) -> StreamingResponse:
        """Run one explicitly enabled offline showcase through real local tool adapters."""
        if studio_service is None:
            raise HTTPException(
                404,
                detail={"code": "studio_showcase_disabled", "message": "Showcase is disabled."},
            )
        try:
            studio_service.scenario_detail(payload.scenario_id)
            expected_request = studio_service.list_scenarios()[0]["query"]
            if (
                payload.request is not None
                and " ".join(payload.request.split()) != expected_request
            ):
                raise StudioDataError(
                    "The offline Electronics checkpoint uses its reviewed marketer request."
                )
        except StudioScenarioNotFound as exc:
            raise HTTPException(
                404,
                detail={"code": "scenario_not_found", "message": "Scenario not found."},
            ) from exc
        except StudioDataError as exc:
            raise HTTPException(
                422,
                detail={"code": "invalid_showcase_request", "message": str(exc)},
            ) from exc

        if payload.workflow == "cached":
            try:
                # Materialize before streaming so a cache miss becomes a real 404.
                messages = list(
                    studio_service.run_cached(
                        payload.scenario_id, payload.request, model=payload.model
                    )
                )
            except StudioCacheMiss as exc:
                raise HTTPException(
                    404, detail={"code": "cache_miss", "message": str(exc)}
                ) from exc
            except StudioDataError as exc:
                raise HTTPException(
                    422, detail={"code": "invalid_showcase_request", "message": str(exc)}
                ) from exc

            def cached_body():
                for message in messages:
                    yield json.dumps(message) + "\n"

            return StreamingResponse(cached_body(), media_type="application/x-ndjson")

        def body():
            execution_mode = "live" if payload.workflow == "live" else "deterministic_test"
            try:
                for message in studio_service.run(
                    payload.scenario_id,
                    payload.request,
                    execution_mode=execution_mode,
                ):
                    yield json.dumps(message) + "\n"
            except Exception as exc:
                # A raised exception here would drop the HTTP stream mid-response and
                # surface in the browser as a fake "network error"; label it instead.
                _server_logger.exception("studio stream run failed")
                yield json.dumps({"type": "error", "error": _public_stream_error(exc)}) + "\n"

        return StreamingResponse(body(), media_type="application/x-ndjson")

    def session_http_error(exc: SessionError) -> HTTPException:
        status = (
            404
            if exc.code in {"session_not_found", "tool_not_found", "artifact_not_found"}
            else 422
        )
        return HTTPException(
            status,
            detail={"code": exc.code, "message": str(exc)},
        )

    @application.post("/api/sessions")
    def create_session(payload: SessionCreateRequest) -> dict[str, Any]:
        """Create state for an external multi-step agent without exposing hidden scenario data."""
        try:
            session = multi_step_service.create_session(payload.scenario_id, payload.request)
            return multi_step_service.snapshot(session.session_id)
        except ScenarioNotFoundError as exc:
            raise HTTPException(
                404,
                detail={"code": "scenario_not_found", "message": "Scenario not found."},
            ) from exc
        except SessionError as exc:
            raise session_http_error(exc) from exc

    @application.get("/api/sessions/{session_id}")
    def session_detail(session_id: str) -> dict[str, Any]:
        try:
            return multi_step_service.snapshot(session_id)
        except SessionError as exc:
            raise session_http_error(exc) from exc

    @application.post("/api/sessions/{session_id}/tools/{tool_name}")
    def call_session_tool(
        session_id: str, tool_name: str, payload: SessionToolRequest
    ) -> dict[str, Any]:
        """Execute one validated environment tool for an external agent."""
        try:
            result = multi_step_service.call_tool(session_id, tool_name, payload.arguments)
            return {"result": result, "session": multi_step_service.snapshot(session_id)}
        except SessionError as exc:
            raise session_http_error(exc) from exc

    @application.post("/api/sessions/{session_id}/actions/copy")
    def submit_session_copy(session_id: str, payload: SessionCopyRequest) -> dict[str, Any]:
        """Record direct structured ad text from an external agent."""
        try:
            identity = payload.agent_identity.model_dump(mode="json")
            execution_mode = ExecutionMode(identity.pop("execution_mode"))
            provenance = ActionProvenance(
                source=ActionSource.EXTERNAL_AGENT,
                execution_mode=execution_mode,
                identity=identity,
            )
            result = multi_step_service.submit_copy(
                session_id, AdCopy(**payload.action.model_dump()), provenance
            )
            return {"result": result, "session": multi_step_service.snapshot(session_id)}
        except (SessionError, ValidationError, ValueError, TypeError) as exc:
            if isinstance(exc, SessionError):
                raise session_http_error(exc) from exc
            raise HTTPException(
                422,
                detail={"code": "invalid_copy", "message": "Structured ad text is invalid."},
            ) from exc

    @application.post("/api/sessions/{session_id}/submit")
    def submit_session_final(session_id: str, payload: SessionSubmitRequest) -> dict[str, Any]:
        """Finalize the current artifact and trigger environment-owned evaluation."""
        try:
            result = multi_step_service.submit_final(
                session_id, payload.artifact_ref, payload.request_version
            )
            return {"result": asdict(result), "session": multi_step_service.snapshot(session_id)}
        except SessionError as exc:
            raise session_http_error(exc) from exc

    @application.get("/api/sessions/{session_id}/artifacts/{artifact_ref}")
    def session_artifact(session_id: str, artifact_ref: str) -> FileResponse:
        try:
            return FileResponse(multi_step_service.artifact_path(session_id, artifact_ref))
        except SessionError as exc:
            raise session_http_error(exc) from exc

    @application.post("/api/session-runs/stream")
    def stream_session_run(payload: SessionCreateRequest, request: Request) -> StreamingResponse:
        """Run an autonomous decision loop over real session/tool boundaries."""
        _reset_live_episode_counters()
        event_queue: queue.Queue[dict[str, Any] | object] = queue.Queue()
        sentinel = object()
        disconnected = threading.Event()
        try:
            session = multi_step_service.create_session(payload.scenario_id, payload.request)
        except ScenarioNotFoundError as exc:
            raise HTTPException(
                404,
                detail={"code": "scenario_not_found", "message": "Scenario not found."},
            ) from exc
        initial_event = asdict(session.events[0])

        def worker() -> None:
            def emit(event) -> None:
                if not disconnected.is_set():
                    event_queue.put({"type": "session_event", "event": asdict(event)})

            try:
                result = autonomous_runner.run(
                    multi_step_service, session.session_id, event_sink=emit
                )
                event_queue.put(
                    {
                        "type": "session_result",
                        "session_id": session.session_id,
                        "result": asdict(result),
                    }
                )
            except SessionError as exc:
                event_queue.put(
                    {
                        "type": "error",
                        "error": {"code": exc.code, "message": str(exc)},
                    }
                )
            except Exception as exc:
                _server_logger.exception("autonomous session run failed")
                event_queue.put({"type": "error", "error": _public_stream_error(exc)})
            finally:
                event_queue.put(sentinel)

        async def body():
            yield (
                json.dumps(
                    {
                        "type": "session_started",
                        "session_id": session.session_id,
                        "event": initial_event,
                    }
                )
                + "\n"
            )
            thread = threading.Thread(target=worker, daemon=True)
            thread.start()
            while True:
                if await request.is_disconnected():
                    disconnected.set()
                    break
                item = await asyncio.to_thread(event_queue.get)
                if item is sentinel:
                    break
                yield json.dumps(item) + "\n"

        return StreamingResponse(body(), media_type="application/x-ndjson")

    @application.post("/api/sessions/{session_id}/revisions/stream")
    def stream_session_revision(
        session_id: str, payload: SessionRevisionRequest, request: Request
    ) -> StreamingResponse:
        """Apply a user refinement and rerun only request-dependent agent work."""
        _reset_live_episode_counters()
        event_queue: queue.Queue[dict[str, Any] | object] = queue.Queue()
        sentinel = object()
        disconnected = threading.Event()
        try:
            before = multi_step_service.snapshot(session_id)["events"][-1]["sequence"]
            multi_step_service.refine_request(session_id, payload.request)
            changed = [
                event
                for event in multi_step_service.snapshot(session_id)["events"]
                if event["sequence"] > before
            ]
        except SessionError as exc:
            raise session_http_error(exc) from exc

        def worker() -> None:
            def emit(event) -> None:
                if not disconnected.is_set():
                    event_queue.put({"type": "session_event", "event": asdict(event)})

            try:
                result = multi_step_service.run_test_agent_revision(session_id, event_sink=emit)
                event_queue.put(
                    {
                        "type": "session_result",
                        "session_id": session_id,
                        "result": asdict(result),
                    }
                )
            except SessionError as exc:
                event_queue.put({"type": "error", "error": {"code": exc.code, "message": str(exc)}})
            except Exception as exc:
                _server_logger.exception("session stream worker failed")
                event_queue.put({"type": "error", "error": _public_stream_error(exc)})
            finally:
                event_queue.put(sentinel)

        async def body():
            for event in changed:
                yield json.dumps({"type": "session_event", "event": event}) + "\n"
            thread = threading.Thread(target=worker, daemon=True)
            thread.start()
            while True:
                if await request.is_disconnected():
                    disconnected.set()
                    break
                item = await asyncio.to_thread(event_queue.get)
                if item is sentinel:
                    break
                yield json.dumps(item) + "\n"

        return StreamingResponse(body(), media_type="application/x-ndjson")

    @application.post("/api/cached-session-runs/stream")
    def stream_cached_session_run(
        payload: CachedRunRequest, request: Request
    ) -> StreamingResponse:
        """Replay cached model outputs through the real stateful tool workflow."""
        selected_store = cached_model_stores.get(payload.model or BASELINE_MODEL_SLUG)
        if (
            selected_store is None
            or payload.scenario_id not in service_scenario_ids
            or payload.scenario_id not in selected_store.scenario_ids
        ):
            raise HTTPException(
                404,
                detail={
                    "code": "cache_miss",
                    "message": "No cached run exists for this scenario and model.",
                },
            )
        try:
            _, cached_result = selected_store.playback(payload.scenario_id, payload.request)
            cached_service = MultiStepSessionService(
                episode_service.repository,
                _CachedCopyGenerator(cached_result),
                _CachedJudge(cached_result),
                compositor=episode_service.compositor,
                reward_policy=episode_service.reward_policy,
            )
            session = cached_service.create_session(payload.scenario_id, payload.request)
        except (EpisodeError, KeyError, TypeError, ValueError, ValidationError) as exc:
            if isinstance(exc, EpisodeError):
                code, message = exc.code, str(exc)
            else:
                code, message = "cache_invalid", "The cached run is invalid."
            raise HTTPException(422, detail={"code": code, "message": message}) from exc

        event_queue: queue.Queue[dict[str, Any] | object] = queue.Queue()
        sentinel = object()
        disconnected = threading.Event()
        initial_event = asdict(session.events[0])

        def worker() -> None:
            def emit(event) -> None:
                if not disconnected.is_set():
                    event_queue.put({"type": "session_event", "event": asdict(event)})

            try:
                result = cached_service.run_test_agent(session.session_id, event_sink=emit)
                event_queue.put(
                    {
                        "type": "session_result",
                        "session_id": session.session_id,
                        "result": asdict(result),
                    }
                )
            except SessionError as exc:
                event_queue.put({"type": "error", "error": {"code": exc.code, "message": str(exc)}})
            except Exception as exc:
                _server_logger.exception("session stream worker failed")
                event_queue.put({"type": "error", "error": _public_stream_error(exc)})
            finally:
                event_queue.put(sentinel)

        async def body():
            yield (
                json.dumps(
                    {
                        "type": "session_started",
                        "session_id": session.session_id,
                        "event": initial_event,
                    }
                )
                + "\n"
            )
            thread = threading.Thread(target=worker, daemon=True)
            thread.start()
            while True:
                if await request.is_disconnected():
                    disconnected.set()
                    break
                item = await asyncio.to_thread(event_queue.get)
                if item is sentinel:
                    break
                yield json.dumps(item) + "\n"

        return StreamingResponse(body(), media_type="application/x-ndjson")

    @application.post("/api/episodes")
    def create_episode(payload: dict[str, Any]) -> dict[str, Any]:
        """Run the demo path, which generates an action inside the environment host."""
        _reset_live_episode_counters()
        scenario_id, query = _episode_input(payload)
        try:
            return asdict(episode_service.run(scenario_id, query_override=query))
        except ScenarioNotFoundError as exc:
            raise HTTPException(
                404, detail={"code": "scenario_not_found", "message": "Scenario not found."}
            ) from exc
        except EpisodeError as exc:
            raise HTTPException(
                _episode_http_status(exc.code),
                detail={
                    "code": exc.code,
                    "message": str(exc),
                    "safe_stages": list(exc.safe_stages),
                },
            ) from exc

    @application.post("/api/evaluations")
    def create_evaluation(payload: EvaluationRequest) -> dict[str, Any]:
        """Score an exact submitted action without invoking the demo ad-text generator."""
        try:
            action = AdCopy(**payload.action.model_dump())
            identity = payload.agent_identity.model_dump(mode="json")
            execution_mode = ExecutionMode(identity.pop("execution_mode"))
            provenance = ActionProvenance(
                source=ActionSource.EXTERNAL_AGENT,
                execution_mode=execution_mode,
                identity=identity,
            )
        except (ValidationError, ValueError, TypeError) as exc:
            raise HTTPException(
                422,
                detail={
                    "code": "invalid_evaluation_request",
                    "message": "Evaluation request is invalid.",
                },
            ) from exc
        try:
            return asdict(episode_service.evaluate(payload.scenario_id, action, provenance))
        except ScenarioNotFoundError as exc:
            raise HTTPException(
                404,
                detail={"code": "scenario_not_found", "message": "Scenario not found."},
            ) from exc
        except EpisodeError as exc:
            raise HTTPException(
                _episode_http_status(exc.code),
                detail={
                    "code": exc.code,
                    "message": str(exc),
                    "safe_stages": list(exc.safe_stages),
                },
            ) from exc

    @application.post("/api/episodes/stream")
    def stream_episode(payload: dict[str, Any], request: Request) -> StreamingResponse:
        """Run the demo path and stream only public, actual-boundary progress as NDJSON."""
        _reset_live_episode_counters()
        event_queue: queue.Queue[dict[str, Any] | object] = queue.Queue()
        sentinel = object()
        disconnected = threading.Event()

        def worker() -> None:
            last_sequence = 0

            def emit(stage_event) -> None:
                nonlocal last_sequence
                last_sequence = stage_event.sequence
                if disconnected.is_set():
                    return
                event_queue.put(
                    {
                        "type": f"stage_{stage_event.state.value}",
                        **asdict(stage_event),
                    }
                )

            try:
                try:
                    run_mode = payload.get("run_mode")
                    if run_mode is None:
                        run_mode = (
                            "live" if os.environ.get("AD_EXECUTION_MODE") == "live" else "cached"
                        )
                    episode_payload = {
                        key: value for key, value in payload.items() if key != "run_mode"
                    }
                    if run_mode not in {"cached", "live"}:
                        raise HTTPException(
                            422,
                            detail={
                                "code": "invalid_run_mode",
                                "message": "Choose Cached run or Live models.",
                            },
                        )
                    parsed_scenario_id, parsed_query = _episode_input(episode_payload)
                except HTTPException as exc:
                    detail = exc.detail if isinstance(exc.detail, dict) else {}
                    raise EpisodeError(
                        str(detail.get("code", "invalid_request")),
                        str(detail.get("message", "Episode request is invalid.")),
                        (),
                    ) from exc
                if run_mode == "cached" and parsed_scenario_id in cached_scenario_ids:
                    cached_events, cached_result = cached_runs.playback(
                        parsed_scenario_id, parsed_query
                    )
                    cached_result["data_provenance"] = episode_service.get_scenario(
                        parsed_scenario_id
                    ).provenance.value
                    for cached_event in cached_events:
                        last_sequence += 1
                        event_queue.put(
                            {
                                "type": "stage_started",
                                "sequence": last_sequence,
                                "stage_id": cached_event["stage_id"],
                                "label": cached_event["label"],
                                "state": "started",
                                "public_data": None,
                            }
                        )
                        last_sequence += 1
                        event_queue.put(
                            {
                                "type": "stage_completed",
                                "sequence": last_sequence,
                                "stage_id": cached_event["stage_id"],
                                "label": cached_event["label"],
                                "state": "completed",
                                "public_data": cached_event["public_data"],
                            }
                        )
                    result_payload = cached_result
                else:
                    if (
                        run_mode == "cached"
                        and os.environ.get("AD_EXECUTION_MODE", "deterministic_test")
                        != "deterministic_test"
                    ):
                        raise EpisodeError(
                            "cache_miss",
                            "No cached run exists for this scenario.",
                            (),
                        )
                    if (
                        run_mode == "live"
                        and os.environ.get("AD_EXECUTION_MODE", "deterministic_test") != "live"
                    ):
                        raise EpisodeError(
                            "live_unavailable",
                            "Live models are not enabled for this environment.",
                            (),
                        )
                    result = episode_service.run(
                        parsed_scenario_id,
                        event_sink=emit,
                        query_override=parsed_query,
                    )
                    result_payload = asdict(result)
                event_queue.put(
                    {
                        "type": "result",
                        "sequence": last_sequence + 1,
                        "result": result_payload,
                    }
                )
            except ScenarioNotFoundError:
                _server_logger.warning("episode stream failed: scenario_not_found")
                event_queue.put(
                    {
                        "type": "error",
                        "sequence": last_sequence + 1,
                        "error": {
                            "code": "scenario_not_found",
                            "message": "Scenario not found.",
                            "safe_stages": [],
                        },
                    }
                )
            except EpisodeError as exc:
                _server_logger.warning(
                    "episode stream failed",
                    extra={
                        "episode_error_code": exc.code,
                        "safe_stage_count": len(exc.safe_stages),
                    },
                )
                event_queue.put(
                    {
                        "type": "error",
                        "sequence": last_sequence + 1,
                        "error": {
                            "code": exc.code,
                            "message": str(exc),
                            "safe_stages": list(exc.safe_stages),
                        },
                    }
                )
            except Exception as exc:
                _server_logger.exception("episode stream worker failed")
                event_queue.put(
                    {
                        "type": "error",
                        "sequence": last_sequence + 1,
                        "error": {**_public_stream_error(exc), "safe_stages": []},
                    }
                )
            finally:
                event_queue.put(sentinel)

        async def body():
            yield json.dumps({"type": "started", "sequence": 0}) + "\n"
            thread = threading.Thread(target=worker, daemon=True)
            thread.start()
            while True:
                if await request.is_disconnected():
                    disconnected.set()
                    break
                item = await asyncio.to_thread(event_queue.get)
                if item is sentinel:
                    break
                yield json.dumps(item) + "\n"

        return StreamingResponse(body(), media_type="application/x-ndjson")

    if assets.is_dir():
        application.mount("/static", StaticFiles(directory=assets), name="static")

        @application.get("/", include_in_schema=False)
        def index() -> FileResponse:
            return FileResponse(assets / "index.html")

    return application


app = create_app()