src/ad_creative_env/studio/electronics.py

"""Offline-first Electronics vertical slice for the Ad Studio showcase runtime."""

from __future__ import annotations

import base64
import copy
import io
import json
import threading
import uuid
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterator

from PIL import Image, ImageDraw, ImageFont, ImageOps

from ad_creative_env.cached_models import CACHED_MODEL_BY_SLUG, ordered_slugs
from ad_creative_env.config import FONT_PATH_BOLD
from ad_creative_env.studio.electronics_agent import (
    ElectronicsAgentDecisionError,
    build_electronics_agent_prompt,
    parse_electronics_decision,
)
from ad_creative_env.studio.electronics_judge import (
    ElectronicsJudgeError,
    build_electronics_judge_prompt,
    parse_electronics_judgement,
)


class StudioDataError(ValueError):
    """A scenario pack is incomplete, inconsistent, or unsafe to load."""


class StudioJudgeError(RuntimeError):
    """The live judge failed. Deliberately NOT a StudioDataError: the autonomous loop
    retries agent mistakes, and a judge/provider failure must escape that loop instead
    of being retried as if the agent had decided badly."""


class StudioScenarioNotFound(KeyError):
    """A scenario is not present in this showcase service."""


class StudioCacheMiss(LookupError):
    """No recorded run exists for the requested scenario/model pair."""


_REGULAR_FONT = str(Path(FONT_PATH_BOLD).with_name("DejaVuSans.ttf"))
_REQUIRED_PACK_FILES = ("manifest.json", "catalog.json", "evidence.json", "evaluation.json")
_REQUIRED_MANIFEST_FIELDS = {
    "scenario_id",
    "version",
    "domain",
    "title",
    "provenance",
    "marketer_request",
    "customer_context",
    "constraints",
    "allowed_actions",
    "allowed_tools",
    "required_branch_fixtures",
    "final_creative",
    "excluded_public_fields",
}


def _read_json(path: Path) -> dict[str, Any]:
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise StudioDataError(f"cannot read scenario pack file: {path.name}") from exc
    if not isinstance(value, dict):
        raise StudioDataError(f"scenario pack file must contain an object: {path.name}")
    return value


def _require_string(value: Any, field: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise StudioDataError(f"{field} must be a non-empty string")
    return " ".join(value.split())


def _require_string_list(value: Any, field: str) -> list[str]:
    if not isinstance(value, list) or not value:
        raise StudioDataError(f"{field} must be a non-empty list")
    return [_require_string(item, field) for item in value]


def _wrap_text(
    draw: ImageDraw.ImageDraw,
    text: str,
    font: ImageFont.FreeTypeFont,
    max_width: int,
    max_lines: int,
) -> tuple[str, ...]:
    lines: list[str] = []
    current = ""
    for word in text.split():
        candidate = f"{current} {word}".strip()
        if draw.textlength(candidate, font=font) <= max_width:
            current = candidate
            continue
        if not current or len(lines) >= max_lines - 1:
            raise StudioDataError("creative text does not fit the Electronics card")
        lines.append(current)
        current = word
    if current:
        lines.append(current)
    if len(lines) > max_lines:
        raise StudioDataError("creative text does not fit the Electronics card")
    return tuple(lines)


@dataclass(frozen=True, slots=True)
class _Pack:
    manifest: dict[str, Any]
    candidates: tuple[dict[str, Any], ...]
    evidence: tuple[dict[str, Any], ...]
    evaluation: dict[str, Any]


def _load_pack(pack_dir: Path) -> _Pack:
    pack_dir = Path(pack_dir)
    missing_files = [name for name in _REQUIRED_PACK_FILES if not (pack_dir / name).is_file()]
    if missing_files:
        raise StudioDataError(f"scenario pack is missing: {', '.join(missing_files)}")
    manifest = _read_json(pack_dir / "manifest.json")
    missing_fields = _REQUIRED_MANIFEST_FIELDS - set(manifest)
    if missing_fields:
        raise StudioDataError(f"manifest is missing: {', '.join(sorted(missing_fields))}")
    for field in ("scenario_id", "version", "domain", "title", "provenance", "marketer_request"):
        _require_string(manifest[field], field)
    if manifest["domain"] != "electronics":
        raise StudioDataError("electronics service requires an electronics manifest")
    allowed_actions = _require_string_list(manifest["allowed_actions"], "allowed_actions")
    allowed_tools = _require_string_list(manifest["allowed_tools"], "allowed_tools")
    if len(set(allowed_actions)) != len(allowed_actions) or len(set(allowed_tools)) != len(
        allowed_tools
    ):
        raise StudioDataError("actions and tools must be unique")
    branch_fixtures = set(
        _require_string_list(manifest["required_branch_fixtures"], "required_branch_fixtures")
    )

    catalog = _read_json(pack_dir / "catalog.json")
    raw_candidates = catalog.get("candidates")
    if not isinstance(raw_candidates, list) or len(raw_candidates) < 2:
        raise StudioDataError("catalog must contain at least two candidates")
    candidates: list[dict[str, Any]] = []
    candidate_refs: set[str] = set()
    present_tags: set[str] = set()
    for candidate in raw_candidates:
        if not isinstance(candidate, dict):
            raise StudioDataError("catalog candidates must be objects")
        candidate_ref = _require_string(candidate.get("candidate_ref"), "candidate_ref")
        if candidate_ref in candidate_refs:
            raise StudioDataError(f"duplicate candidate: {candidate_ref}")
        candidate_refs.add(candidate_ref)
        for field in ("name", "colour", "graphics_profile"):
            _require_string(candidate.get(field), field)
        for field in (
            "price_usd",
            "memory_gb",
            "external_4k_displays",
            "battery_hours",
            "weight_kg",
        ):
            if not isinstance(candidate.get(field), (int, float)):
                raise StudioDataError(f"candidate {candidate_ref} has invalid {field}")
        candidate["evidence_refs"] = _require_string_list(
            candidate.get("evidence_refs"), "evidence_refs"
        )
        tags = candidate.get("fixture_tags")
        if not isinstance(tags, list):
            raise StudioDataError(f"candidate {candidate_ref} has invalid fixture_tags")
        present_tags.update(_require_string(tag, "fixture_tags") for tag in tags)
        candidates.append(candidate)
    if not branch_fixtures <= present_tags:
        raise StudioDataError("catalog does not exercise every required branch fixture")

    evidence_document = _read_json(pack_dir / "evidence.json")
    raw_evidence = evidence_document.get("records")
    if evidence_document.get("mode") != "deterministic_test" or not isinstance(raw_evidence, list):
        raise StudioDataError("offline evidence must use deterministic_test mode")
    evidence: list[dict[str, Any]] = []
    evidence_refs: set[str] = set()
    for record in raw_evidence:
        if not isinstance(record, dict) or not isinstance(record.get("claims"), dict):
            raise StudioDataError("evidence records must contain structured claims")
        evidence_ref = _require_string(record.get("evidence_ref"), "evidence_ref")
        if evidence_ref in evidence_refs:
            raise StudioDataError(f"duplicate evidence: {evidence_ref}")
        evidence_refs.add(evidence_ref)
        _require_string(record.get("source_title"), "source_title")
        _require_string(record.get("source_uri"), "source_uri")
        evidence.append(record)
    referenced_evidence = {
        evidence_ref for candidate in candidates for evidence_ref in candidate["evidence_refs"]
    }
    if not referenced_evidence <= evidence_refs:
        raise StudioDataError("candidate references missing evidence")

    evaluation = _read_json(pack_dir / "evaluation.json")
    expected_ref = evaluation.get("hidden_expected_candidate_ref")
    if expected_ref not in candidate_refs:
        raise StudioDataError("evaluation expects an unknown candidate")
    dimensions = evaluation.get("dimensions")
    if not isinstance(dimensions, dict) or abs(sum(dimensions.values()) - 1.0) > 1e-9:
        raise StudioDataError("evaluation dimensions must sum to one")
    return _Pack(manifest, tuple(candidates), tuple(evidence), evaluation)


def _contains_live_execution_mode(value: Any) -> bool:
    if isinstance(value, dict):
        for key, item in value.items():
            if key in {"execution_mode", "judge_execution_mode"} and item == "live":
                return True
            if _contains_live_execution_mode(item):
                return True
        return False
    if isinstance(value, list):
        return any(_contains_live_execution_mode(item) for item in value)
    return False


def _load_cached_runs(cache_dir: Path, scenario_id: str) -> dict[str, dict[str, Any]]:
    """Load recorded per-model replays, refusing anything mislabelled or unknown."""
    if not cache_dir.is_dir():
        return {}
    cached_runs: dict[str, dict[str, Any]] = {}
    for path in sorted(cache_dir.glob("*.json")):
        slug = path.stem
        if slug not in CACHED_MODEL_BY_SLUG:
            raise StudioDataError(f"cached run references an unknown model: {slug}")
        document = _read_json(path)
        if document.get("kind") != "electronics-cached-run" or document.get("version") != 1:
            raise StudioDataError(f"cached run has an unsupported format: {path.name}")
        if document.get("model") != slug or document.get("scenario_id") != scenario_id:
            raise StudioDataError(f"cached run does not match its scenario or model: {path.name}")
        messages = document.get("messages")
        if not isinstance(messages, list) or not messages:
            raise StudioDataError(f"cached run contains no messages: {path.name}")
        for message in messages:
            if not isinstance(message, dict) or message.get("type") not in {
                "session_event",
                "session_result",
            }:
                raise StudioDataError(f"cached run contains an invalid message: {path.name}")
        if messages[-1].get("type") != "session_result" or not isinstance(
            messages[-1].get("result"), dict
        ):
            raise StudioDataError(f"cached run does not end with a result: {path.name}")
        if _contains_live_execution_mode(messages):
            raise StudioDataError(
                f"cached run is still labelled live; re-record it as recorded_replay: {path.name}"
            )
        cached_runs[slug] = document
    return cached_runs


class _EventBuilder:
    def __init__(self) -> None:
        self.sequence = 0

    def event(
        self,
        *,
        role: str,
        kind: str,
        status: str,
        title: str,
        summary: str,
        progress: int,
        tool_name: str | None = None,
        call_id: str | None = None,
        public_data: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        self.sequence += 1
        return {
            "sequence": self.sequence,
            "role": role,
            "kind": kind,
            "status": status,
            "title": title,
            "summary": summary,
            "progress": progress,
            "tool_name": tool_name,
            "call_id": call_id,
            "public_data": public_data or {},
        }


class ElectronicsStudioService:
    """Load and execute the reviewed deterministic Electronics showcase scenario."""

    def __init__(
        self,
        pack: _Pack,
        *,
        live_copy_generator: Callable[[str], tuple[dict[str, Any], dict[str, str]]] | None = None,
        live_image_generator: Callable[[str], tuple[bytes, dict[str, Any]]] | None = None,
        live_decision_generator: Callable[[str], tuple[dict[str, Any] | str, dict[str, str]]]
        | None = None,
        live_judge_generator: Callable[[str], tuple[dict[str, Any] | str, dict[str, str]]]
        | None = None,
        live_episode_reset: Callable[[], None] | None = None,
        autonomous_max_turns: int = 20,
        autonomous_max_errors: int = 3,
        cached_runs: dict[str, dict[str, Any]] | None = None,
    ) -> None:
        self._pack = pack
        self._scenario_id = pack.manifest["scenario_id"]
        self._cached_runs = cached_runs or {}
        self._candidate_by_ref = {
            candidate["candidate_ref"]: candidate for candidate in pack.candidates
        }
        self._evidence_by_ref = {record["evidence_ref"]: record for record in pack.evidence}
        self._live_copy_generator = live_copy_generator
        self._live_image_generator = live_image_generator
        self._live_decision_generator = live_decision_generator
        self._live_judge_generator = live_judge_generator
        self._live_episode_reset = live_episode_reset
        if autonomous_max_turns < 1 or autonomous_max_errors < 1:
            raise StudioDataError("autonomous agent limits must be positive")
        self._autonomous_max_turns = autonomous_max_turns
        self._autonomous_max_errors = autonomous_max_errors
        self._asset_lock = threading.Lock()
        self._asset_store: dict[str, tuple[bytes, dict[str, Any]]] = {}
        self._tool_registry = {
            "get_customer_profile": self._get_customer_profile,
            "search_catalog": self._search_catalog,
            "search_web": self._search_web,
            "fetch_page_evidence": self._fetch_page_evidence,
            "inspect_specs": self._inspect_specs,
            "compare_candidates": self._compare_candidates,
            "check_compatibility": self._check_compatibility,
            "generate_image": self._generate_image,
            "compose_ad": self._compose_ad,
        }
        missing_tools = set(pack.manifest["allowed_tools"]) - set(self._tool_registry)
        if missing_tools:
            raise StudioDataError(
                f"scenario requires unregistered tools: {', '.join(sorted(missing_tools))}"
            )

    @classmethod
    def from_pack(
        cls,
        pack_dir: Path,
        *,
        live_copy_generator: Callable[[str], tuple[dict[str, Any], dict[str, str]]] | None = None,
        live_image_generator: Callable[[str], tuple[bytes, dict[str, Any]]] | None = None,
        live_decision_generator: Callable[[str], tuple[dict[str, Any] | str, dict[str, str]]]
        | None = None,
        live_judge_generator: Callable[[str], tuple[dict[str, Any] | str, dict[str, str]]]
        | None = None,
        live_episode_reset: Callable[[], None] | None = None,
        autonomous_max_turns: int = 20,
        autonomous_max_errors: int = 3,
    ) -> ElectronicsStudioService:
        pack = _load_pack(pack_dir)
        return cls(
            pack,
            live_copy_generator=live_copy_generator,
            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=autonomous_max_turns,
            autonomous_max_errors=autonomous_max_errors,
            cached_runs=_load_cached_runs(
                Path(pack_dir) / "cache", pack.manifest["scenario_id"]
            ),
        )

    def cached_model_options(self) -> list[dict[str, str]]:
        """Recorded replay models for this scenario, in registry (newest-first) order."""
        return [
            {
                "id": slug,
                "label": CACHED_MODEL_BY_SLUG[slug].label,
            }
            for slug in ordered_slugs(set(self._cached_runs))
        ]

    def list_scenarios(self) -> list[dict[str, Any]]:
        manifest = self._pack.manifest
        return [
            {
                "scenario_id": self._scenario_id,
                "query": manifest["marketer_request"],
                "title": manifest["title"],
                "domain": manifest["domain"],
                "product_name": "Agent-selected creator laptop",
                "product_type": "Laptop comparison",
                "data_provenance": manifest["provenance"],
                "runtime": "studio",
                "cached_models": self.cached_model_options(),
            }
        ]

    def scenario_detail(self, scenario_id: str) -> dict[str, Any]:
        self._require_scenario(scenario_id)
        manifest = self._pack.manifest
        return {
            **self.list_scenarios()[0],
            "observation": {
                "customer_context": manifest["customer_context"],
                "query": manifest["marketer_request"],
                "task": {
                    "domain": manifest["domain"],
                    "constraints": manifest["constraints"],
                    "product_selection": "agent_selected",
                },
            },
            "allowed_actions": list(manifest["allowed_actions"]),
            "allowed_tools": list(manifest["allowed_tools"]),
            "branch_fixtures": list(manifest["required_branch_fixtures"]),
            "execution_mode": "deterministic_test",
            "trajectory_steps": 14,
        }

    def _require_scenario(self, scenario_id: str) -> None:
        if scenario_id != self._scenario_id:
            raise StudioScenarioNotFound(f"scenario not found: {scenario_id}")

    def _comparison(self) -> dict[str, Any]:
        constraints = self._pack.manifest["constraints"]
        ranked: list[dict[str, Any]] = []
        rejected: list[dict[str, Any]] = []
        for candidate in self._pack.candidates:
            reasons: list[str] = []
            if candidate["price_usd"] > constraints["maximum_price_usd"]:
                reasons.append("over budget")
            if candidate["memory_gb"] < constraints["minimum_memory_gb"]:
                reasons.append("insufficient memory")
            display_values = {
                record["claims"].get("external_4k_displays")
                for evidence_ref in candidate["evidence_refs"]
                for record in (self._evidence_by_ref[evidence_ref],)
                if "external_4k_displays" in record["claims"]
            }
            if len(display_values) > 1:
                reasons.append("conflicting display evidence")
            elif (
                not display_values
                or min(display_values) < constraints["minimum_external_4k_displays"]
            ):
                reasons.append("insufficient external display support")
            summary = {
                "candidate_ref": candidate["candidate_ref"],
                "name": candidate["name"],
                "price_usd": candidate["price_usd"],
                "memory_gb": candidate["memory_gb"],
                "external_4k_displays": candidate["external_4k_displays"],
                "battery_hours": candidate["battery_hours"],
            }
            if reasons:
                rejected.append({**summary, "reasons": reasons})
            else:
                ranked.append(summary)
        ranked.sort(key=lambda item: (-item["battery_hours"], item["price_usd"]))
        return {
            "ranked": ranked,
            "rejected": rejected,
            "ranking_preference": "battery life, then lower price",
            "display_items": [
                {
                    "title": f"#{index} {item['name']}",
                    "status": "Meets all required constraints",
                    "details": (
                        f"${item['price_usd']} · {item['memory_gb']} GB · "
                        f"{item['external_4k_displays']} external 4K displays · "
                        f"{item['battery_hours']}h battery"
                    ),
                }
                for index, item in enumerate(ranked, start=1)
            ]
            + [
                {
                    "title": item["name"],
                    "status": "Rejected",
                    "details": ", ".join(item["reasons"]),
                }
                for item in rejected
            ],
        }

    @staticmethod
    def _validate_arguments(
        tool_name: str,
        arguments: dict[str, Any],
        *,
        required: set[str] | None = None,
    ) -> None:
        if not isinstance(arguments, dict):
            raise StudioDataError(f"{tool_name} arguments must be an object")
        expected = required or set()
        if set(arguments) != expected:
            raise StudioDataError(
                f"{tool_name} requires exactly: {', '.join(sorted(expected)) or 'no arguments'}"
            )

    def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
        if tool_name not in self._pack.manifest["allowed_tools"]:
            raise StudioDataError(f"tool is not allowed for this scenario: {tool_name}")
        try:
            tool = self._tool_registry[tool_name]
        except KeyError as exc:
            raise StudioDataError(f"tool is not registered: {tool_name}") from exc
        return tool(arguments)

    def _get_customer_profile(self, arguments: dict[str, Any]) -> dict[str, Any]:
        self._validate_arguments("get_customer_profile", arguments)
        return {"customer_context": self._pack.manifest["customer_context"]}

    def _search_catalog(self, arguments: dict[str, Any]) -> dict[str, Any]:
        self._validate_arguments("search_catalog", arguments, required={"constraints"})
        if arguments["constraints"] != self._pack.manifest["constraints"]:
            raise StudioDataError("search_catalog constraints do not match the working intent")
        return {
            "candidate_count": len(self._pack.candidates),
            "candidate_refs": [candidate["candidate_ref"] for candidate in self._pack.candidates],
            "display_items": [
                {
                    "title": candidate["name"],
                    "status": f"${candidate['price_usd']} · {candidate['memory_gb']} GB",
                    "details": (
                        f"{candidate['external_4k_displays']} external 4K displays · "
                        f"{candidate['battery_hours']}h battery"
                    ),
                }
                for candidate in self._pack.candidates
            ],
        }

    def _search_web(self, arguments: dict[str, Any]) -> dict[str, Any]:
        self._validate_arguments("search_web", arguments, required={"query"})
        _require_string(arguments["query"], "search_web.query")
        return {
            "mode": "deterministic_test",
            "provider": "authored_snapshot_index",
            "result_count": len(self._pack.evidence),
            "evidence_refs": [record["evidence_ref"] for record in self._pack.evidence],
            "display_items": [
                {
                    "title": record["source_title"],
                    "status": "Offline evidence snapshot",
                    "details": record["source_uri"],
                }
                for record in self._pack.evidence
            ],
        }

    def _fetch_page_evidence(self, arguments: dict[str, Any]) -> dict[str, Any]:
        self._validate_arguments("fetch_page_evidence", arguments, required={"evidence_refs"})
        evidence_refs = _require_string_list(arguments["evidence_refs"], "evidence_refs")
        try:
            records = [self._evidence_by_ref[ref] for ref in evidence_refs]
        except KeyError as exc:
            raise StudioDataError("fetch_page_evidence received an unknown reference") from exc
        return {
            "records": records,
            "display_items": [
                {
                    "title": record["source_title"],
                    "status": "Claims extracted",
                    "details": ", ".join(
                        f"{key.replace('_', ' ')}: {value}"
                        for key, value in record["claims"].items()
                    ),
                }
                for record in records
            ],
        }

    def _candidate_list(self, tool_name: str, arguments: dict[str, Any]) -> list[dict[str, Any]]:
        self._validate_arguments(tool_name, arguments, required={"candidate_refs"})
        candidate_refs = _require_string_list(arguments["candidate_refs"], "candidate_refs")
        try:
            return [self._candidate_by_ref[ref] for ref in candidate_refs]
        except KeyError as exc:
            raise StudioDataError(f"{tool_name} received an unknown candidate") from exc

    def _inspect_specs(self, arguments: dict[str, Any]) -> dict[str, Any]:
        candidates = self._candidate_list("inspect_specs", arguments)
        return {
            "inspected": [
                {
                    "candidate_ref": candidate["candidate_ref"],
                    "memory_gb": candidate["memory_gb"],
                    "external_4k_displays": candidate["external_4k_displays"],
                    "battery_hours": candidate["battery_hours"],
                }
                for candidate in candidates
            ],
            "conflict_detected": "laptop-nomad-13",
        }

    def _compare_candidates(self, arguments: dict[str, Any]) -> dict[str, Any]:
        candidates = self._candidate_list("compare_candidates", arguments)
        if {item["candidate_ref"] for item in candidates} != set(self._candidate_by_ref):
            raise StudioDataError("compare_candidates must evaluate the complete returned set")
        return self._comparison()

    def _check_compatibility(self, arguments: dict[str, Any]) -> dict[str, Any]:
        self._validate_arguments("check_compatibility", arguments, required={"candidate_ref"})
        candidate_ref = _require_string(arguments["candidate_ref"], "candidate_ref")
        try:
            candidate = self._candidate_by_ref[candidate_ref]
        except KeyError as exc:
            raise StudioDataError("check_compatibility received an unknown candidate") from exc
        comparison = self._comparison()
        compatible_refs = {item["candidate_ref"] for item in comparison["ranked"]}
        return {
            "candidate_ref": candidate_ref,
            "compatible": candidate_ref in compatible_refs,
            "required_external_4k_displays": 2,
            "supported_external_4k_displays": candidate["external_4k_displays"],
            "evidence_refs": candidate["evidence_refs"],
        }

    def _fixture_product_image(self, candidate: dict[str, Any]) -> bytes:
        canvas = Image.new("RGB", (1024, 1024), "#e8edf3")
        draw = ImageDraw.Draw(canvas)
        draw.ellipse((112, 90, 912, 890), fill="#d7deea")
        draw.rounded_rectangle((220, 220, 804, 610), radius=26, fill="#20242b")
        draw.rectangle((246, 249, 778, 581), fill="#9162d9")
        draw.polygon(((155, 650), (869, 650), (946, 770), (78, 770)), fill="#555d69")
        draw.rounded_rectangle((350, 690, 674, 728), radius=14, fill="#303640")
        output = io.BytesIO()
        canvas.save(output, format="PNG", compress_level=9)
        return output.getvalue()

    def _generate_image(self, arguments: dict[str, Any]) -> dict[str, Any]:
        self._validate_arguments(
            "generate_image", arguments, required={"candidate_ref", "execution_mode"}
        )
        candidate_ref = _require_string(arguments["candidate_ref"], "candidate_ref")
        try:
            candidate = self._candidate_by_ref[candidate_ref]
        except KeyError as exc:
            raise StudioDataError("generate_image received an unknown candidate") from exc
        execution_mode = arguments["execution_mode"]
        prompt = (
            "Photorealistic premium graphite 14-inch creator laptop on a clean studio desk, "
            "screen showing an abstract purple creative workspace, slim portable design, "
            "soft professional lighting, three-quarter product photography, no text, no logo, "
            "no watermark, square composition."
        )
        if execution_mode == "live":
            if self._live_image_generator is None:
                raise StudioDataError("live image generation is not configured")
            try:
                image_bytes, provider_metadata = self._live_image_generator(prompt)
            except Exception as exc:
                raise StudioDataError("live image generation failed") from exc
            provenance = {
                "execution_mode": "live",
                "source": "live_image_generator",
                **provider_metadata,
            }
        elif execution_mode == "deterministic_test":
            image_bytes = self._fixture_product_image(candidate)
            provenance = {
                "execution_mode": "deterministic_test",
                "source": "local_fixture_renderer",
                "provider": "local",
                "model": "electronics-fixture-v1",
            }
        else:
            raise StudioDataError("unsupported image execution mode")
        try:
            with Image.open(io.BytesIO(image_bytes)) as generated:
                generated.verify()
        except Exception as exc:
            raise StudioDataError("image generator returned invalid image bytes") from exc
        asset_ref = f"electronics-image-{uuid.uuid4().hex}"
        with self._asset_lock:
            self._asset_store[asset_ref] = (image_bytes, provenance)
        return {
            "asset_ref": asset_ref,
            "candidate_ref": candidate_ref,
            "execution_mode": execution_mode,
            "provenance": provenance,
            "prompt": prompt,
        }

    def _compose_ad(self, arguments: dict[str, Any]) -> dict[str, Any]:
        self._validate_arguments(
            "compose_ad",
            arguments,
            required={"candidate_ref", "asset_ref", "action"},
        )
        candidate_ref = _require_string(arguments["candidate_ref"], "candidate_ref")
        asset_ref = _require_string(arguments["asset_ref"], "asset_ref")
        with self._asset_lock:
            asset = self._asset_store.get(asset_ref)
        if asset is None:
            raise StudioDataError("compose_ad received an unknown asset")
        image_bytes, image_provenance = asset
        action = arguments["action"]
        if not isinstance(action, dict) or set(action) != {"headline", "body", "cta"}:
            raise StudioDataError("compose_ad received invalid creative fields")
        try:
            candidate = self._candidate_by_ref[candidate_ref]
        except KeyError as exc:
            raise StudioDataError("compose_ad received an unknown candidate") from exc
        return {
            "artifact_ref": f"electronics-{candidate_ref}-ad-v1",
            "card_artifact": self._render_card(candidate, action, image_bytes),
            "width": 1200,
            "height": 628,
            "image_provenance": image_provenance,
        }

    def _render_card(
        self,
        candidate: dict[str, Any],
        action: dict[str, str],
        image_bytes: bytes,
    ) -> str:
        canvas = Image.new("RGB", (1200, 628), "#f3f0eb")
        draw = ImageDraw.Draw(canvas)
        bold_18 = ImageFont.truetype(FONT_PATH_BOLD, 18)
        bold_26 = ImageFont.truetype(FONT_PATH_BOLD, 26)
        bold_38 = ImageFont.truetype(FONT_PATH_BOLD, 38)
        regular_23 = ImageFont.truetype(_REGULAR_FONT, 23)
        try:
            with Image.open(io.BytesIO(image_bytes)) as source:
                product_image = ImageOps.fit(
                    source.convert("RGB"), (600, 628), method=Image.Resampling.LANCZOS
                )
        except Exception as exc:
            raise StudioDataError("generated product image cannot be composed") from exc
        canvas.paste(product_image, (0, 0))
        draw.rectangle((600, 0, 1200, 628), fill="#17151c")
        left = 656
        draw.text((left, 54), "CREATOR WORKSTATION", font=bold_18, fill="#d6b7ff")
        draw.text((left, 90), candidate["name"].upper(), font=bold_26, fill="#ffffff")
        headline_lines = _wrap_text(draw, action["headline"], bold_38, 488, 2)
        y = 144
        for line in headline_lines:
            draw.text((left, y), line, font=bold_38, fill="#ffffff")
            y += 48
        body_lines = _wrap_text(draw, action["body"], regular_23, 488, 4)
        y += 28
        for line in body_lines:
            draw.text((left, y), line, font=regular_23, fill="#d9d5df")
            y += 38
        draw.rounded_rectangle((left, 492, left + 190, 550), radius=8, fill="#e9d7ff")
        draw.text((left + 18, 508), action["cta"], font=bold_18, fill="#22182d")
        output = io.BytesIO()
        canvas.save(output, format="PNG", compress_level=9)
        return "data:image/png;base64," + base64.b64encode(output.getvalue()).decode("ascii")

    @staticmethod
    def _validate_creative_fit(action: dict[str, str]) -> None:
        probe = Image.new("RGB", (600, 628))
        draw = ImageDraw.Draw(probe)
        headline_font = ImageFont.truetype(FONT_PATH_BOLD, 38)
        body_font = ImageFont.truetype(_REGULAR_FONT, 23)
        _wrap_text(draw, action["headline"], headline_font, 488, 2)
        _wrap_text(draw, action["body"], body_font, 488, 4)

    def _write_creative(
        self, candidate: dict[str, Any], execution_mode: str
    ) -> tuple[dict[str, str], dict[str, Any]]:
        if execution_mode == "deterministic_test":
            return (
                {
                    "headline": "Create anywhere. Connect everything.",
                    "body": (
                        "32 GB memory, two external 4K displays, and up to 14 hours of battery "
                        "life for portable creative work — $1,649."
                    ),
                    "cta": "View details",
                },
                {
                    "source": "deterministic_test",
                    "execution_mode": "deterministic_test",
                    "identity": {
                        "provider": "local",
                        "model": "electronics-validation-client",
                        "config_version": "electronics-client-v1",
                    },
                },
            )
        if execution_mode != "live" or self._live_copy_generator is None:
            raise StudioDataError("live creative generation is not configured")
        prompt = (
            "Write JSON with exactly headline, body, and cta for a display ad. "
            "Headline: maximum 60 characters. Body: maximum 180 characters and no more than "
            "three short lines when wrapped. CTA must be exactly 'View details'. Use only these "
            f"facts: product {candidate['name']}; price ${candidate['price_usd']}; "
            f"memory {candidate['memory_gb']} GB; two external 4K displays; battery up to "
            f"{candidate['battery_hours']} hours; weight {candidate['weight_kg']} kg. "
            "Do not invent claims, discounts, awards, or performance results."
        )
        action, metadata = self._live_copy_generator(prompt)
        if not isinstance(action, dict) or set(action) != {"headline", "body", "cta"}:
            raise StudioDataError("live creative model returned an invalid action")
        headline = _require_string(action["headline"], "headline")
        body = _require_string(action["body"], "body")
        cta = _require_string(action["cta"], "cta")
        if len(headline) > 60 or len(body) > 180 or cta != "View details":
            raise StudioDataError("live creative model returned out-of-contract text")
        return (
            {"headline": headline, "body": body, "cta": cta},
            {
                "source": "live_generator",
                "execution_mode": "live",
                "identity": metadata,
            },
        )

    def _autonomous_actions(self, state: dict[str, Any]) -> list[dict[str, Any]]:
        manifest = self._pack.manifest
        if state["intent"] is None:
            return [
                {
                    "action": "update_working_intent",
                    "required_constraints": manifest["constraints"],
                }
            ]
        actions: list[dict[str, Any]] = []
        if state["profile"] is None:
            actions.append(
                {"action": "call_tool", "tool_name": "get_customer_profile", "arguments": {}}
            )
        if state["catalog"] is None:
            actions.append(
                {
                    "action": "call_tool",
                    "tool_name": "search_catalog",
                    "arguments": {"constraints": manifest["constraints"]},
                }
            )
        if state["search"] is None:
            actions.append(
                {
                    "action": "call_tool",
                    "tool_name": "search_web",
                    "arguments": {"query": "creator laptop 32 GB two external 4K displays"},
                }
            )
        if state["catalog"] is not None and state["specs"] is None:
            actions.append(
                {
                    "action": "call_tool",
                    "tool_name": "inspect_specs",
                    "arguments": {"candidate_refs": state["catalog"]["candidate_refs"]},
                }
            )
        if state["search"] is not None and state["evidence"] is None:
            actions.append(
                {
                    "action": "call_tool",
                    "tool_name": "fetch_page_evidence",
                    "arguments": {"evidence_refs": state["search"]["evidence_refs"]},
                }
            )
        if (
            state["specs"] is not None
            and state["evidence"] is not None
            and state["comparison"] is None
        ):
            actions.append(
                {
                    "action": "call_tool",
                    "tool_name": "compare_candidates",
                    "arguments": {"candidate_refs": state["catalog"]["candidate_refs"]},
                }
            )
        if state["comparison"] is not None and state["selected"] is None:
            actions.append(
                {
                    "action": "select_candidate",
                    "candidate_options": state["comparison"]["ranked"],
                }
            )
        if state["selected"] is not None and state["compatibility"] is None:
            actions.append(
                {
                    "action": "call_tool",
                    "tool_name": "check_compatibility",
                    "arguments": {"candidate_ref": state["selected"]["candidate_ref"]},
                }
            )
        if state["compatibility"] is not None and state["compatibility"]["compatible"]:
            if state["image"] is None:
                actions.append(
                    {
                        "action": "call_tool",
                        "tool_name": "generate_image",
                        "arguments": {
                            "candidate_ref": state["selected"]["candidate_ref"],
                            "execution_mode": "live",
                        },
                    }
                )
            if state["creative"] is None:
                candidate = state["selected"]
                actions.append(
                    {
                        "action": "write_creative",
                        "required_fields": ["headline", "body", "cta"],
                        "cta": "View details",
                        "headline_max_chars": 60,
                        "body_max_chars": 180,
                        "allowed_facts": candidate,
                    }
                )
        if state["image"] is not None and state["creative"] is not None and state["card"] is None:
            actions.append(
                {
                    "action": "call_tool",
                    "tool_name": "compose_ad",
                    "arguments": {
                        "candidate_ref": state["selected"]["candidate_ref"],
                        "asset_ref": state["image"]["asset_ref"],
                        "action": state["creative"],
                    },
                }
            )
        if state["card"] is not None and state["profile"] is not None:
            actions.append(
                {
                    "action": "submit",
                    "artifact_ref": state["card"]["artifact_ref"],
                    "request_version": 1,
                }
            )
        return actions

    @staticmethod
    def _autonomous_progress(state: dict[str, Any]) -> int:
        keys = (
            "intent",
            "profile",
            "catalog",
            "search",
            "evidence",
            "specs",
            "comparison",
            "compatibility",
            "image",
            "selected",
            "creative",
            "card",
        )
        return sum(state[key] is not None for key in keys)

    @staticmethod
    def _require_decision_available(
        decision: dict[str, Any], available: list[dict[str, Any]]
    ) -> None:
        # Rejections name the currently available choices: a bare "not available" makes
        # weaker decision models repeat the same invalid action until the error limit.
        choices = ", ".join(
            item.get("tool_name") or item["action"] for item in available
        )
        action = decision["action"]
        if action == "call_tool":
            matches = [
                item
                for item in available
                if item["action"] == action and item.get("tool_name") == decision["tool_name"]
            ]
            if not matches:
                raise ElectronicsAgentDecisionError(
                    "action_unavailable",
                    f"The selected tool is not currently available. "
                    f"Choose one of the currently available actions: {choices}.",
                )
            expected = matches[0]["arguments"]
            if decision["arguments"] != expected:
                raise ElectronicsAgentDecisionError(
                    "invalid_arguments",
                    "Use the exact arguments currently shown for this tool: "
                    + json.dumps(expected, sort_keys=True),
                )
            return
        if action == "select_candidate":
            if any(
                decision["candidate_ref"]
                in {candidate["candidate_ref"] for candidate in item["candidate_options"]}
                for item in available
                if item["action"] == action
            ):
                return
            raise ElectronicsAgentDecisionError(
                "action_unavailable",
                f"select_candidate is not currently available for that candidate. "
                f"Choose one of the currently available actions: {choices}.",
            )
        if not any(item["action"] == action for item in available):
            raise ElectronicsAgentDecisionError(
                "action_unavailable",
                f"The selected action is not currently available. "
                f"Choose one of the currently available actions: {choices}.",
            )

    def _evaluate_autonomous(
        self,
        *,
        scenario_id: str,
        marketer_request: str,
        state: dict[str, Any],
        action_provenance: dict[str, Any],
        tool_names: list[str],
        recovered_errors: int,
    ) -> dict[str, Any]:
        manifest = self._pack.manifest
        selected = state["selected"]
        comparison = state["comparison"]
        rejected_refs = {item["candidate_ref"] for item in comparison["rejected"]}
        checks = [
            {
                "check_id": "electronics.budget",
                "passed": selected["price_usd"] <= manifest["constraints"]["maximum_price_usd"],
                "explanation": (
                    f"${selected['price_usd']:,} is within the "
                    f"${manifest['constraints']['maximum_price_usd']:,} budget."
                ),
            },
            {
                "check_id": "electronics.memory",
                "passed": selected["memory_gb"] >= manifest["constraints"]["minimum_memory_gb"],
                "explanation": "The selected laptop has the required memory.",
            },
            {
                "check_id": "electronics.display_support",
                "passed": bool(state["compatibility"]["compatible"]),
                "explanation": "The compatibility tool confirmed two external 4K displays.",
            },
            {
                "check_id": "electronics.evidence_consistency",
                "passed": selected["candidate_ref"] not in rejected_refs,
                "explanation": "The selected candidate has no unresolved evidence conflict.",
            },
            {
                "check_id": "electronics.fresh_selection",
                "passed": state["card"]["artifact_ref"]
                == f"electronics-{selected['candidate_ref']}-ad-v1",
                "explanation": "The submitted creative uses the current selected candidate.",
            },
        ]
        constraint_score = 1.0 if all(item["passed"] for item in checks[:3]) else 0.0
        evidence_score = 1.0 if checks[3]["passed"] else 0.0
        expected_tools = set(self._pack.manifest["allowed_tools"])
        trajectory_score = max(
            0.0,
            min(1.0, len(set(tool_names)) / len(expected_tools) - 0.05 * recovered_errors),
        )
        creative_score = 1.0
        scores = {
            "constraint_match": constraint_score,
            "evidence_grounding": evidence_score,
            "trajectory_quality": trajectory_score,
            "creative_quality": creative_score,
        }
        explanations = {
            "constraint_match": "Computed from the selected candidate and compatibility result.",
            "evidence_grounding": "The selected candidate survived the evidence-conflict checks.",
            "trajectory_quality": (
                f"The live agent chose {len(tool_names)} tool calls across "
                f"{len(set(tool_names))} distinct tools and recovered from "
                f"{recovered_errors} rejected decisions."
            ),
            "creative_quality": (
                "The deterministic rubric confirmed the structured copy/card contract; it is not "
                "a live model judgement."
            ),
        }
        judge_execution_mode = "deterministic_test"
        judge_identity = {
            "provider": "local",
            "model": "electronics-rubric",
            "config_version": "electronics-eval-v3-hybrid-draft",
        }
        if self._live_judge_generator is not None:
            # Hybrid reward: constraint_match and trajectory_quality stay deterministic —
            # rules score structured data better than a model — while the two dimensions a
            # rule cannot score well come from the LLM judge.
            evidence_records = (state.get("evidence") or {}).get("records") or []
            judge_prompt = build_electronics_judge_prompt(
                marketer_request=marketer_request,
                constraints=manifest["constraints"],
                selected_candidate=selected,
                evidence=evidence_records,
                action=state["creative"],
            )
            try:
                raw_judgement, judge_metadata = self._live_judge_generator(judge_prompt)
            except Exception as exc:
                raise StudioJudgeError(
                    "The live judge request failed; the episode cannot be scored."
                ) from exc
            try:
                judgement = parse_electronics_judgement(raw_judgement)
            except ElectronicsJudgeError as exc:
                raise StudioJudgeError(
                    f"The live judge response was invalid ({exc.code}); "
                    "the episode cannot be scored."
                ) from exc
            scores.update(judgement["scores"])
            explanations.update(judgement["explanations"])
            judge_execution_mode = "live"
            judge_identity = {key: str(value) for key, value in judge_metadata.items()}
        dimensions = self._pack.evaluation["dimensions"]
        weighted = {name: scores[name] * weight for name, weight in dimensions.items()}
        reward = sum(weighted.values()) if all(item["passed"] for item in checks) else 0.0
        return {
            "scenario_id": scenario_id,
            "data_provenance": manifest["provenance"],
            "selected_candidate": selected,
            "action": state["creative"],
            "action_provenance": action_provenance,
            "image_provenance": state["card"]["image_provenance"],
            "card_artifact": state["card"]["card_artifact"],
            "checks": checks,
            "judge_scores": scores,
            "judge_explanations": explanations,
            "judge_execution_mode": judge_execution_mode,
            "judge_identity": judge_identity,
            "reward_policy_version": "electronics-eval-v3-hybrid-draft",
            "weighted_components": weighted,
            "base_score": sum(weighted.values()),
            "failed_checks": [item["check_id"] for item in checks if not item["passed"]],
            "applied_cap": None,
            "reward": reward,
            "review_status": "accepted" if reward > 0 else "rejected",
            "trajectory_step_count": 14,
            "autonomous_tool_calls": len(tool_names),
            "autonomous_recovered_errors": recovered_errors,
        }

    def _run_autonomous(self, scenario_id: str, marketer_request: str) -> Iterator[dict[str, Any]]:
        if self._live_decision_generator is None or self._live_image_generator is None:
            raise StudioDataError("live autonomous Electronics generation is not configured")
        if self._live_episode_reset is not None:
            # Provider request/image counters bound ONE episode; without this reset the
            # second Generate ad click inherits an exhausted budget and aborts.
            self._live_episode_reset()
        events = _EventBuilder()

        def emit(event: dict[str, Any]) -> dict[str, Any]:
            return {"type": "session_event", "event": event}

        state = {
            key: None
            for key in (
                "intent",
                "profile",
                "catalog",
                "search",
                "evidence",
                "specs",
                "comparison",
                "selected",
                "compatibility",
                "image",
                "creative",
                "card",
            )
        }
        yield emit(
            events.event(
                role="system",
                kind="session",
                status="completed",
                title="Electronics task ready",
                summary="Started the bounded autonomous Electronics agent.",
                progress=0,
                public_data={"scenario_id": scenario_id, "execution_mode": "live"},
            )
        )
        last_error = None
        consecutive_errors = 0
        recovered_errors = 0
        tool_names: list[str] = []
        action_provenance: dict[str, Any] | None = None
        for turn_number in range(1, self._autonomous_max_turns + 1):
            available = self._autonomous_actions(state)
            public_state = {key: value for key, value in state.items() if key != "card"}
            if state["card"] is not None:
                public_state["card"] = {"artifact_ref": state["card"]["artifact_ref"]}
            prompt = build_electronics_agent_prompt(
                request=marketer_request,
                available_actions=available,
                public_state=public_state,
                last_error=last_error,
            )
            try:
                try:
                    raw_decision, metadata = self._live_decision_generator(prompt)
                except Exception as exc:
                    raise ElectronicsAgentDecisionError(
                        getattr(exc, "code", "provider_error"),
                        "The live decision model request failed.",
                    ) from exc
                if not isinstance(metadata, dict):
                    raise ElectronicsAgentDecisionError(
                        "invalid_identity", "The decision model metadata is invalid."
                    )
                decision = parse_electronics_decision(raw_decision)
                self._require_decision_available(decision, available)
                progress = self._autonomous_progress(state)
                identity = {key: str(value) for key, value in metadata.items()}
                action = decision["action"]
                if action == "update_working_intent":
                    state["intent"] = _require_string(decision["intent"], "intent")
                    yield emit(
                        events.event(
                            role="agent",
                            kind="intent",
                            status="completed",
                            title="Working intent updated",
                            summary=state["intent"],
                            progress=self._autonomous_progress(state),
                            public_data={"decision": decision, "identity": identity},
                        )
                    )
                elif action == "call_tool":
                    tool_name = decision["tool_name"]
                    arguments = decision["arguments"]
                    yield emit(
                        events.event(
                            role="agent",
                            kind="tool_decision",
                            status="completed",
                            title=f"Use {tool_name}",
                            summary=f"The live agent chose {tool_name} from the currently available actions.",
                            progress=progress,
                            tool_name=tool_name,
                            public_data={"decision": decision, "identity": identity},
                        )
                    )
                    call_id = f"electronics-live-{turn_number}"
                    yield emit(
                        events.event(
                            role="tool",
                            kind="tool",
                            status="running",
                            title=tool_name.replace("_", " ").title(),
                            summary=f"Calling {tool_name} with validated agent arguments.",
                            progress=progress,
                            tool_name=tool_name,
                            call_id=call_id,
                            public_data={"arguments": arguments},
                        )
                    )
                    result = self.call_tool(tool_name, arguments)
                    state_key = {
                        "get_customer_profile": "profile",
                        "search_catalog": "catalog",
                        "search_web": "search",
                        "fetch_page_evidence": "evidence",
                        "inspect_specs": "specs",
                        "compare_candidates": "comparison",
                        "check_compatibility": "compatibility",
                        "generate_image": "image",
                        "compose_ad": "card",
                    }[tool_name]
                    state[state_key] = result
                    tool_names.append(tool_name)
                    yield emit(
                        events.event(
                            role="tool",
                            kind="tool",
                            status="completed",
                            title=tool_name.replace("_", " ").title(),
                            summary=f"{tool_name} completed and updated the agent state.",
                            progress=self._autonomous_progress(state),
                            tool_name=tool_name,
                            call_id=call_id,
                            public_data={"result": result},
                        )
                    )
                elif action == "select_candidate":
                    state["selected"] = self._candidate_by_ref[decision["candidate_ref"]]
                    yield emit(
                        events.event(
                            role="agent",
                            kind="selection",
                            status="completed",
                            title="Best candidate selected",
                            summary=f"The live agent selected {state['selected']['name']} after comparison.",
                            progress=self._autonomous_progress(state),
                            public_data={"decision": decision, "identity": identity},
                        )
                    )
                elif action == "write_creative":
                    headline = _require_string(decision["headline"], "headline")
                    body = _require_string(decision["body"], "body")
                    cta = _require_string(decision["cta"], "cta")
                    creative_problems = []
                    if len(headline) > 60:
                        creative_problems.append("headline must be at most 60 characters")
                    if len(body) > 180:
                        creative_problems.append("body must be at most 180 characters")
                    if cta != "View details":
                        creative_problems.append("cta must be exactly 'View details'")
                    if creative_problems:
                        raise ElectronicsAgentDecisionError(
                            "invalid_creative",
                            "The creative does not satisfy its contract: "
                            + "; ".join(creative_problems)
                            + ".",
                        )
                    creative = {"headline": headline, "body": body, "cta": cta}
                    self._validate_creative_fit(creative)
                    state["creative"] = creative
                    action_provenance = {
                        "source": "external_agent",
                        "execution_mode": "live",
                        "identity": identity,
                    }
                    yield emit(
                        events.event(
                            role="agent",
                            kind="copy",
                            status="completed",
                            title="Ad text created",
                            summary="The live agent wrote evidence-grounded ad text as a direct action.",
                            progress=self._autonomous_progress(state),
                            public_data={
                                "action": state["creative"],
                                "decision": decision,
                                "identity": identity,
                            },
                        )
                    )
                elif action == "submit":
                    if (
                        decision["artifact_ref"] != state["card"]["artifact_ref"]
                        or decision["request_version"] != 1
                    ):
                        raise ElectronicsAgentDecisionError(
                            "invalid_submission", "The submission reference is stale or invalid."
                        )
                    yield emit(
                        events.event(
                            role="agent",
                            kind="submission",
                            status="completed",
                            title="Creative submitted",
                            summary="The live agent chose to submit the completed creative.",
                            progress=13,
                            public_data={"decision": decision, "identity": identity},
                        )
                    )
                    result = self._evaluate_autonomous(
                        scenario_id=scenario_id,
                        marketer_request=marketer_request,
                        state=state,
                        action_provenance=action_provenance or {},
                        tool_names=tool_names,
                        recovered_errors=recovered_errors,
                    )
                    yield emit(
                        events.event(
                            role="system",
                            kind="verification",
                            status="completed",
                            title="Requirements verified",
                            summary="Computed five Electronics checks from the autonomous run state.",
                            progress=13,
                            public_data={"checks": result["checks"]},
                        )
                    )
                    yield emit(
                        events.event(
                            role="system",
                            kind="judgement",
                            status="completed",
                            title="Scenario quality assessed",
                            summary="Applied the provisional scenario-specific deterministic rubric.",
                            progress=13,
                            public_data={
                                "judge_scores": result["judge_scores"],
                                "judge_explanations": result["judge_explanations"],
                                "judge_execution_mode": result["judge_execution_mode"],
                                "judge_identity": result["judge_identity"],
                            },
                        )
                    )
                    yield emit(
                        events.event(
                            role="system",
                            kind="evaluation",
                            status="completed",
                            title="Evaluation complete",
                            summary="The autonomous trajectory and final creative were evaluated.",
                            progress=14,
                            public_data={"result": result},
                        )
                    )
                    yield {"type": "session_result", "result": result}
                    return
                last_error = None
                if consecutive_errors:
                    recovered_errors += consecutive_errors
                consecutive_errors = 0
            except (ElectronicsAgentDecisionError, StudioDataError) as exc:
                code = getattr(exc, "code", "invalid_action")
                last_error = {"code": code, "message": str(exc)}
                if (
                    isinstance(exc, StudioDataError)
                    and "creative text does not fit" in str(exc)
                    and state["card"] is None
                ):
                    # Composition is the first place exact font/layout fit is known. Reopen the
                    # upstream creative action so the model can shorten its text on the next turn.
                    state["creative"] = None
                    action_provenance = None
                    last_error = {
                        "code": "creative_does_not_fit",
                        "message": (
                            "Rewrite the creative with a headline of at most 32 characters and a "
                            "body of at most 100 characters. Use short words and sentences."
                        ),
                    }
                consecutive_errors += 1
                # The attempt count keeps every retry prompt distinct — at temperature 0
                # an unchanged prompt deterministically repeats the same invalid decision.
                last_error["rejected_attempts"] = consecutive_errors
                yield emit(
                    events.event(
                        role="agent",
                        kind="error",
                        status="failed",
                        title="Decision rejected",
                        summary=str(exc),
                        progress=self._autonomous_progress(state),
                        public_data={"error": last_error, "turn": turn_number},
                    )
                )
                if consecutive_errors >= self._autonomous_max_errors:
                    raise StudioDataError(
                        "The autonomous agent could not recover from repeated invalid decisions."
                    ) from exc
        raise StudioDataError("The autonomous agent did not finish within the turn limit.")

    def run_cached(
        self,
        scenario_id: str,
        request: str | None = None,
        *,
        model: str | None = None,
    ) -> Iterator[dict[str, Any]]:
        """Replay one recorded model run verbatim; every label stays recorded_replay."""
        self._require_scenario(scenario_id)
        manifest = self._pack.manifest
        marketer_request = request or manifest["marketer_request"]
        if " ".join(marketer_request.split()) != manifest["marketer_request"]:
            raise StudioDataError("cached Electronics runs use their recorded marketer request")
        available = ordered_slugs(set(self._cached_runs))
        if not available:
            raise StudioCacheMiss("no cached Electronics run has been recorded yet")
        slug = model or available[0]
        if slug not in self._cached_runs:
            raise StudioCacheMiss(f"no cached Electronics run exists for this model: {slug}")
        for message in self._cached_runs[slug]["messages"]:
            yield copy.deepcopy(message)

    def run(
        self,
        scenario_id: str,
        request: str | None = None,
        *,
        execution_mode: str = "deterministic_test",
    ) -> Iterator[dict[str, Any]]:
        self._require_scenario(scenario_id)
        manifest = self._pack.manifest
        marketer_request = request or manifest["marketer_request"]
        if " ".join(marketer_request.split()) != manifest["marketer_request"]:
            raise StudioDataError("the Electronics checkpoint uses its reviewed marketer request")
        if execution_mode not in {"deterministic_test", "live"}:
            raise StudioDataError("unsupported Electronics execution mode")
        if execution_mode == "live":
            yield from self._run_autonomous(scenario_id, marketer_request)
            return
        events = _EventBuilder()

        def emit(event: dict[str, Any]) -> dict[str, Any]:
            return {"type": "session_event", "event": event}

        yield emit(
            events.event(
                role="system",
                kind="session",
                status="completed",
                title="Electronics task ready",
                summary="Loaded the reviewed offline comparison scenario and its evidence pack.",
                progress=0,
                public_data={"scenario_id": scenario_id, "execution_mode": execution_mode},
            )
        )
        yield emit(
            events.event(
                role="agent",
                kind="intent",
                status="completed",
                title="Working intent updated",
                summary="Find a portable creator laptop under $1,700 with 32 GB memory and two external 4K displays; prefer battery life.",
                progress=1,
                public_data={
                    "constraints": manifest["constraints"],
                    "action": "update_working_intent",
                },
            )
        )

        fetched_refs = [
            "evidence-aero-specs",
            "evidence-aero-displays",
            "evidence-nomad-catalog",
            "evidence-nomad-official",
        ]
        candidate_refs = [candidate["candidate_ref"] for candidate in self._pack.candidates]
        comparison = self._comparison()
        selected = self._candidate_by_ref[comparison["ranked"][0]["candidate_ref"]]
        tool_calls: list[tuple[str, dict[str, Any], str, str]] = [
            (
                "get_customer_profile",
                {},
                "Loaded approved creator-workflow preferences.",
                "Customer profile",
            ),
            (
                "search_catalog",
                {"constraints": manifest["constraints"]},
                "Found six authored catalog candidates for comparison.",
                "Catalog search",
            ),
            (
                "search_web",
                {"query": "creator laptop 32 GB two external 4K displays"},
                "Searched the reviewed offline source index; no live web request was made.",
                "Evidence search",
            ),
            (
                "fetch_page_evidence",
                {"evidence_refs": fetched_refs},
                "Loaded detailed claims for the strongest candidate and the conflicting candidate.",
                "Evidence retrieval",
            ),
            (
                "inspect_specs",
                {"candidate_refs": candidate_refs},
                "Checked memory, display support, price, and battery evidence for every candidate.",
                "Specification inspection",
            ),
            (
                "compare_candidates",
                {"candidate_refs": candidate_refs},
                "Ranked two valid candidates and rejected four with clear reasons.",
                "Candidate comparison",
            ),
            (
                "check_compatibility",
                {"candidate_ref": selected["candidate_ref"]},
                "Confirmed the selected laptop supports the required two external 4K displays.",
                "Display compatibility",
            ),
            (
                "generate_image",
                {
                    "candidate_ref": selected["candidate_ref"],
                    "execution_mode": execution_mode,
                },
                (
                    "Generated a live product image for the selected laptop."
                    if execution_mode == "live"
                    else "Generated the deterministic test image for the selected laptop."
                ),
                "Product image generation",
            ),
        ]

        tool_progress = {
            "get_customer_profile": 2,
            "search_catalog": 3,
            "search_web": 4,
            "fetch_page_evidence": 5,
            "inspect_specs": 6,
            "compare_candidates": 7,
            "check_compatibility": 8,
            "generate_image": 9,
        }
        executed_tools: list[tuple[str, dict[str, Any], str, str, dict[str, Any]]] = []
        for index, (tool_name, arguments, summary, title) in enumerate(tool_calls, start=1):
            call_id = f"electronics-call-{index}"
            progress = tool_progress[tool_name]
            yield emit(
                events.event(
                    role="agent",
                    kind="tool_decision",
                    status="completed",
                    title=f"Use {tool_name}",
                    summary=(
                        "The scenario controller selected this allowed tool for the current state; "
                        "the live model owns creative text only in this checkpoint."
                        if execution_mode == "live"
                        else "The deterministic validation client selected this allowed tool for the current scenario state."
                    ),
                    progress=progress,
                    tool_name=tool_name,
                    public_data={
                        "action": "call_tool",
                        "tool_name": tool_name,
                        "arguments": arguments,
                    },
                )
            )
            yield emit(
                events.event(
                    role="tool",
                    kind="tool",
                    status="running",
                    title=title,
                    summary=f"Calling {tool_name} with validated offline inputs.",
                    progress=progress,
                    tool_name=tool_name,
                    call_id=call_id,
                    public_data={"arguments": arguments},
                )
            )
            result = self.call_tool(tool_name, arguments)
            executed_tools.append((tool_name, arguments, summary, title, result))
            yield emit(
                events.event(
                    role="tool",
                    kind="tool",
                    status="completed",
                    title=title,
                    summary=summary,
                    progress=progress,
                    tool_name=tool_name,
                    call_id=call_id,
                    public_data={"result": result},
                )
            )

        yield emit(
            events.event(
                role="agent",
                kind="selection",
                status="completed",
                title="Best candidate selected",
                summary="Selected AeroBook Creator 14 because it meets every requirement and has the best supported battery life among valid candidates.",
                progress=10,
                public_data={
                    "action": "select_candidate",
                    "candidate_ref": selected["candidate_ref"],
                },
            )
        )
        action, action_provenance = self._write_creative(selected, execution_mode)
        yield emit(
            events.event(
                role="agent",
                kind="copy",
                status="completed",
                title="Ad text created",
                summary=(
                    "The live creative model wrote evidence-backed ad text."
                    if execution_mode == "live"
                    else "The deterministic validation client wrote the test ad text."
                ),
                progress=11,
                public_data={
                    "action": action,
                    "candidate_ref": selected["candidate_ref"],
                    "action_provenance": action_provenance,
                },
            )
        )
        image_result = next(
            result
            for tool_name, _arguments, _summary, _title, result in executed_tools
            if tool_name == "generate_image"
        )
        compose_arguments = {
            "candidate_ref": selected["candidate_ref"],
            "asset_ref": image_result["asset_ref"],
            "action": action,
        }
        compose_call_id = "electronics-call-9"
        yield emit(
            events.event(
                role="agent",
                kind="tool_decision",
                status="completed",
                title="Use compose_ad",
                summary=(
                    "The scenario controller selected composition after candidate, evidence, asset, "
                    "and copy requirements were ready."
                    if execution_mode == "live"
                    else "The deterministic validation client selected composition after candidate, "
                    "evidence, asset, and copy requirements were ready."
                ),
                progress=12,
                tool_name="compose_ad",
                public_data={
                    "action": "call_tool",
                    "tool_name": "compose_ad",
                    "arguments": compose_arguments,
                },
            )
        )
        yield emit(
            events.event(
                role="tool",
                kind="tool",
                status="running",
                title="Compose display ad",
                summary="Calling compose_ad with the selected asset and verified copy.",
                progress=12,
                tool_name="compose_ad",
                call_id=compose_call_id,
                public_data={"arguments": compose_arguments},
            )
        )
        composition = self.call_tool("compose_ad", compose_arguments)
        card_artifact = composition["card_artifact"]
        yield emit(
            events.event(
                role="tool",
                kind="tool",
                status="completed",
                title="Compose display ad",
                summary="Composed a 1200 × 628 display card from the selected laptop and verified copy.",
                progress=12,
                tool_name="compose_ad",
                call_id=compose_call_id,
                public_data={"result": composition},
            )
        )
        yield emit(
            events.event(
                role="agent",
                kind="submission",
                status="completed",
                title="Creative submitted",
                summary="Submitted the current non-stale Electronics creative for evaluation.",
                progress=13,
                public_data={"artifact_ref": "electronics-aero-14-ad-v1", "request_version": 1},
            )
        )

        checks = [
            {
                "check_id": "electronics.budget",
                "passed": True,
                "explanation": "$1,649 is within the $1,700 budget.",
            },
            {
                "check_id": "electronics.memory",
                "passed": True,
                "explanation": "The selected laptop has the required 32 GB memory.",
            },
            {
                "check_id": "electronics.display_support",
                "passed": True,
                "explanation": "Two external 4K displays are supported by cited evidence.",
            },
            {
                "check_id": "electronics.evidence_consistency",
                "passed": True,
                "explanation": "No conflicting evidence remains for the selected candidate.",
            },
            {
                "check_id": "electronics.fresh_selection",
                "passed": True,
                "explanation": "The submitted creative uses the selected current candidate.",
            },
        ]
        yield emit(
            events.event(
                role="system",
                kind="verification",
                status="completed",
                title="Requirements verified",
                summary="All hard Electronics requirements passed.",
                progress=13,
                public_data={"checks": checks},
            )
        )
        scores = {
            "constraint_match": 1.0,
            "evidence_grounding": 1.0,
            "trajectory_quality": 1.0,
            "creative_quality": 1.0,
        }
        explanations = {
            "constraint_match": "The selected laptop satisfies budget, memory, display, and battery preference requirements.",
            "evidence_grounding": "Every factual claim maps to a reviewed evidence record.",
            "trajectory_quality": (
                "The scenario controller inspected conflicts and rejected unsupported candidates "
                "before selection."
                if execution_mode == "live"
                else "The deterministic client inspected conflicts and rejected unsupported "
                "candidates before selection."
            ),
            "creative_quality": "The offline deterministic rubric confirms the copy and card contract; it is not a live model judgement.",
        }
        yield emit(
            events.event(
                role="system",
                kind="judgement",
                status="completed",
                title="Scenario quality assessed",
                summary="Applied the scenario-specific deterministic Electronics rubric.",
                progress=13,
                public_data={
                    "judge_scores": scores,
                    "judge_explanations": explanations,
                    "judge_execution_mode": "deterministic_test",
                    "judge_identity": {
                        "provider": "local",
                        "model": "electronics-rubric",
                        "config_version": "electronics-eval-v1-provisional",
                    },
                },
            )
        )
        dimensions = self._pack.evaluation["dimensions"]
        weighted = {name: scores[name] * weight for name, weight in dimensions.items()}
        result = {
            "scenario_id": scenario_id,
            "data_provenance": manifest["provenance"],
            "selected_candidate": {
                key: selected[key]
                for key in (
                    "candidate_ref",
                    "name",
                    "price_usd",
                    "memory_gb",
                    "external_4k_displays",
                    "battery_hours",
                )
            },
            "action": action,
            "action_provenance": action_provenance,
            "image_provenance": composition["image_provenance"],
            "card_artifact": card_artifact,
            "checks": checks,
            "judge_scores": scores,
            "judge_explanations": explanations,
            "judge_execution_mode": "deterministic_test",
            "judge_identity": {
                "provider": "local",
                "model": "electronics-rubric",
                "config_version": "electronics-eval-v1-provisional",
            },
            "reward_policy_version": "electronics-eval-v1-provisional",
            "weighted_components": weighted,
            "base_score": sum(weighted.values()),
            "failed_checks": [],
            "applied_cap": None,
            "reward": 1.0,
            "review_status": "accepted",
            "trajectory_step_count": 14,
        }
        yield emit(
            events.event(
                role="system",
                kind="evaluation",
                status="completed",
                title="Evaluation complete",
                summary="The selected candidate, trajectory, evidence, and creative passed the scenario contract.",
                progress=14,
                public_data={"result": result},
            )
        )
        yield {"type": "session_result", "result": result}