"""W21B space_set_bg step — frame/addon/life 3층 공간 세트 BG (2026-06-10 사용자 합의 방식).

장소 그룹(master_plan group = loc 묶음)당:
  ① analyze (frame/addon/life_baseline + construction + access, 씬·전문 무삭제)
  ② place_desc whole/indoor/outdoor 분리 (+visible adjacent masses)
  ③ view briefs (wide establishing 기준)
  ④ 실내: frame 2D(t2i) → addon i2i = ★최종 2D FP★ → frame_check(VLM 진단)
     → VLM 공간 좌표 → ★공간당 단일 번호 마킹★(PIL) → 마킹 사본 base 로 BG i2i
  ⑤ 옥외(open_air): FP 미경유 — 공간당 ★단 한 장★ T2I 기준 BG
  ⑥ threshold = connector(plate 생략) / same_space = canonical 사진 참조 i2i
산출: checkpoint manifest + assets/<gid>/ 아래 png + space_plate_map.

★3D FP 단계 없음(실험 비교로 제거 확정). ★generic — 시나리오/품목/장소 하드코딩 0.
opt-in: settings.space_set_bg_enabled (default False) → OFF 시 not_applicable 만
기록하고 기존 FP/BG 체인 영향 0. 하류(scene_image) 배선은 Phase 2 (이 step 은
plate 산출/계약까지 — dwelling_zone_map Phase 접근과 동일).
"""
from __future__ import annotations

import hashlib
import json
import logging
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional

from app.core.step_runner import StepRunner
from app.modules.pipeline import space_set_bg as core

logger = logging.getLogger(__name__)

SCHEMA_VERSION = 1
PROMPT_VERSION = "4.202606111309"  # 4: Phase 3 plate_action derive — SHOT_ASSIGN enum + 파생 plate i2i


class SpaceSetBgStep(StepRunner):
    """frame/addon/life 3층 공간 세트 BG 파이프라인 step."""

    # 테스트 전용 주입 슬롯 (production 은 건드리지 않는다)
    _text_override: Optional[Callable[..., str]] = None
    _vision_override: Optional[Callable[..., str]] = None
    _t2i_override: Optional[Callable[..., None]] = None
    _i2i_override: Optional[Callable[..., None]] = None

    def set_overrides_for_testing(
        self, *, text=None, vision=None, t2i=None, i2i=None
    ) -> None:
        self._text_override = text
        self._vision_override = vision
        self._t2i_override = t2i
        self._i2i_override = i2i

    # ──────────────────────── plumbing ────────────────────────
    def check_applicability(self) -> bool:
        """flag OFF 면 not_applicable 로 마킹되게 (completed empty cp 방지 — Codex 리뷰 #2).
        ON 이면 기본 applicability(if_background_mode) 평가로 위임."""
        from app.core.config import settings
        if not bool(getattr(settings, "space_set_bg_enabled", False)):
            return False
        return super().check_applicability()

    def _load_prev_checkpoint(self, step_id: str) -> Optional[Dict[str, Any]]:
        from app.core.config import settings
        cp = (
            Path(settings.projects_dir) / self.project_id / "checkpoints"
            / "episodes" / self.episode_id / step_id / "manifest.json"
        )
        if cp.exists():
            try:
                return json.loads(cp.read_text(encoding="utf-8"))
            except Exception as exc:
                logger.warning("space_set_bg: %s parse failed: %s", step_id, exc)
        return None

    def _config_hash(self) -> str:
        from app.core.config import settings
        enabled = bool(getattr(settings, "space_set_bg_enabled", False))
        payload = {
            "enabled": enabled,
            "text_model": getattr(settings, "space_set_bg_text_model", ""),
            "vision_model": getattr(settings, "space_set_bg_vision_model", ""),
            "render_model": getattr(settings, "space_set_bg_render_model", ""),
            "schema_version": SCHEMA_VERSION,
            "prompt_version": PROMPT_VERSION,
            "core_schema_version": core.SCHEMA_VERSION,
        }
        # 프롬프트가 팩으로 나간 뒤로는 ★파일이 실행 입력★ 이다 — 같은 코드로도
        # 다른 프롬프트가 나갈 수 있으니 stem 별 (판, 내용 해시)를 접는다.
        # 꺼져 있으면 팩을 아예 읽지 않는다: 지연 로드 계약을 지키고, hash 도
        # 팩 이전 전과 바이트 동일하게 남는다.
        if enabled:
            payload["prompt_pack"] = core.pack_identity()
        return hashlib.sha256(
            json.dumps(payload, sort_keys=True).encode("utf-8")
        ).hexdigest()[:16]

    def _not_applicable(self) -> Dict[str, Any]:
        return {
            "applicable_count": 0,
            "completed_count": 0,
            "failed_count": 0,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {},
        }

    # ──────────────────────── provider 디스패치 (override 우선) ────────────────────────
    def _text(self, *, system: str, user: str, max_tokens: int) -> str:
        if self._text_override is not None:
            return self._text_override(system=system, user=user, max_tokens=max_tokens)
        from app.core.config import settings
        from app.modules.pipeline import space_set_bg_provider as prov
        return prov.text_completion(
            system=system, user=user, max_tokens=max_tokens,
            model=getattr(settings, "space_set_bg_text_model", prov.TEXT_MODEL_DEFAULT),
        )

    def _vision(self, *, system: str, user: str, image_paths: List[str], max_tokens: int) -> str:
        if self._vision_override is not None:
            return self._vision_override(
                system=system, user=user, image_paths=image_paths, max_tokens=max_tokens)
        from app.core.config import settings
        from app.modules.pipeline import space_set_bg_provider as prov
        return prov.vision_completion(
            system=system, user=user, image_paths=image_paths, max_tokens=max_tokens,
            model=getattr(settings, "space_set_bg_vision_model", prov.VISION_MODEL_DEFAULT),
        )

    def _t2i(self, *, prompt: str, out_path: str) -> None:
        if self._t2i_override is not None:
            self._t2i_override(prompt=prompt, out_path=out_path)
            return
        from app.core.config import settings
        from app.modules.pipeline import space_set_bg_provider as prov
        prov.image_generate(
            prompt=prompt, out_path=out_path,
            model=getattr(settings, "space_set_bg_render_model", prov.IMAGE_MODEL_DEFAULT),
        )

    def _i2i(self, *, base_image_path: str, prompt: str, out_path: str) -> None:
        if self._i2i_override is not None:
            self._i2i_override(base_image_path=base_image_path, prompt=prompt, out_path=out_path)
            return
        from app.core.config import settings
        from app.modules.pipeline import space_set_bg_provider as prov
        prov.image_edit(
            base_image_path=base_image_path, prompt=prompt, out_path=out_path,
            model=getattr(settings, "space_set_bg_render_model", prov.IMAGE_MODEL_DEFAULT),
        )

    def _parse_json(self, text: str, *, what: str) -> Any:
        from app.modules.pipeline import space_set_bg_provider as prov
        return prov.parse_json(text, what=what)

    # ──────────────────────── 입력 로딩 ────────────────────────
    def _load_fulltext(self) -> str:
        from sqlalchemy.orm import undefer
        from app.models.project import Episode
        ep = (
            self.db.query(Episode)
            .options(undefer(Episode.fulltext))
            .filter(Episode.id == self.episode_id)
            .first()
        )
        return (ep.fulltext or "") if ep else ""

    @staticmethod
    def _world_guide_subset(world_guide_cp: Optional[Dict[str, Any]]) -> Dict[str, Any]:
        data = ((world_guide_cp or {}).get("data") or {})
        guide = data.get("guide") or data.get("guide_json") or data
        if not isinstance(guide, dict):
            return {}
        keys = ("world_setting_summary", "location_guardrails", "style_rules", "image_generation_notes")
        subset = {k: guide.get(k) for k in keys if guide.get(k) is not None}
        return subset or guide

    @staticmethod
    def _scenes_to_shot_rows(scenes: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """_select_scenes_for_group 의 scenes → core.shot_blocks 입력 행으로.
        scene 원문(text)은 summary 슬롯에 그대로(무삭제).
        shot 인덱스 포함 — Phase 2 shot→space 배정의 deterministic 조인 키."""
        rows: List[Dict[str, Any]] = []
        for sc in scenes:
            si = sc.get("scene_index")
            for sh in sc.get("shots") or []:
                rows.append({
                    "scene": si,
                    "shot": sh.get("shot_index"),
                    "heading": sc.get("heading") or "",
                    "summary": sc.get("text") or "",
                    "shot_desc": sh.get("description") or "",
                })
        return rows

    # ──────────────────────── 그룹 1개 실행 ────────────────────────
    def _run_group(
        self,
        gid: str,
        scenes: List[Dict[str, Any]],
        guide: Dict[str, Any],
        fulltext: str,
        out_dir: Path,
    ) -> Dict[str, Any]:
        out_dir.mkdir(parents=True, exist_ok=True)
        shot_rows = self._scenes_to_shot_rows(scenes)
        scene_nums = sorted({r["scene"] for r in shot_rows if r.get("scene") is not None})
        blocks = core.shot_blocks(shot_rows)

        # ① analyze (출력 한도 16000 — life_baseline 포함 JSON 이 8000 초과)
        analysis = self._parse_json(
            self._text(system=core.ANALYZE_FA_SYS,
                       user=core.build_analyze_user(blocks, fulltext, scene_nums),
                       max_tokens=16000),
            what=f"{gid} analyze")
        analysis = core.validate_fa(analysis)
        (out_dir / "space_analysis.json").write_text(
            json.dumps(analysis, ensure_ascii=False, indent=2), encoding="utf-8")

        groups = core.enclosure_groups(analysis)
        multi = len(groups) > 1
        all_spaces = analysis.get("spaces") or []

        # ② place_desc — whole / indoor / outdoor
        descs: Dict[str, str] = {}
        for gkey, gspaces in [("whole", all_spaces)] + list(groups.items()):
            descs[gkey] = self._text(
                system=core.PLACE_GROUP_SYS,
                user=core.build_place_group_user(analysis, guide, gkey, gspaces),
                max_tokens=500)
        (out_dir / "place_desc.json").write_text(
            json.dumps(descs, ensure_ascii=False, indent=2), encoding="utf-8")

        # ③ view briefs
        briefs = self._parse_json(
            self._text(system=core.VIEW_BRIEF_SYS,
                       user=core.build_view_briefs_user(analysis), max_tokens=1900),
            what=f"{gid} view_briefs")

        # ⑤ 공간별 BG 계획 — FP 필요 여부(④)가 plans 의 kind 에 의존하므로 먼저 계산.
        # 단일 marked_indoor 그룹은 plan_space_views 가 indoor_t2i 로 강등 → 그 그룹 FP 생략.
        plans, connectors = core.plan_space_views(analysis, briefs)

        # ④ 실내 그룹: frame 2D → addon i2i = 최종 2D FP (+frame_check 진단)
        # ★marked_indoor plan 이 있는 그룹만 — 옥외(open_air)·단일 실내 공간 그룹은 FP 미경유
        # (단일 공간은 마킹할 '다른 방' 이 없어 FP 가치 0 + 직역 위험만 — 사용자 2026-06-10 피드백)
        fp_by_suffix: Dict[str, Path] = {}
        frame_checks: Dict[str, Any] = {}
        marked_suffixes = {p["group_suffix"] for p in plans if p["kind"] == "marked_indoor"}
        for gk, gspaces in groups.items():
            sfx = f"_{gk}" if multi else ""
            if sfx not in marked_suffixes:
                continue  # FP 미경유 그룹 (옥외 단일 T2I / 단일 실내 공간 T2I)
            fprompt = self._text(
                system=core.FRAME_FP_SYS,
                user=core.build_fp_user(analysis, gspaces, gk, frame_only=True),
                max_tokens=900)
            (out_dir / f"frame_prompt{sfx}.txt").write_text(fprompt, encoding="utf-8")
            frame_png = out_dir / f"fp_frame_2d{sfx}.png"
            self._t2i(prompt=fprompt, out_path=str(frame_png))
            addon_items = [
                a.get("implementation") for s in gspaces
                for a in (s.get("addon") or []) if a.get("implementation")
            ]
            comp_png = out_dir / f"fp_final_2d{sfx}.png"
            self._i2i(base_image_path=str(frame_png),
                      prompt=core.ADDON_I2I.format(addons=", ".join(addon_items)),
                      out_path=str(comp_png))
            fp_by_suffix[sfx] = comp_png
            try:
                frame_checks[gk] = self._parse_json(
                    self._vision(
                        system=core.FRAME_CHECK_SYS,
                        user="첫째 이미지 = FRAME 평면도(구조 셸). 둘째 이미지 = addon i2i 결과. "
                             "frame(벽/문/창/통로/access path) 보존 여부를 schema 로 판단하라.",
                        image_paths=[str(frame_png), str(comp_png)], max_tokens=1000),
                    what=f"{gid} frame_check {gk}")
            except Exception as exc:  # 진단 실패는 비치명 (게이트 아님)
                frame_checks[gk] = {"frame_preserved": None, "error": str(exc)}

        # ⑤(실행) 공간별 BG 생성 — plans 는 ④ 앞에서 계산됨
        positions_cache: Dict[str, Dict[str, Any]] = {}
        space_by_name = {s.get("name"): s for s in all_spaces}
        plates: Dict[str, Dict[str, Any]] = {}
        for p in plans:
            rn, key, kind, sfx = p["room"], p["key"], p["kind"], p["group_suffix"]
            outp = out_dir / f"bg_{key}.png"
            position_fallback = False
            if kind == "samespace_ref":
                canon = out_dir / f"bg_{p['ref_key']}.png"
                if not canon.exists():
                    plates[rn] = {"status": "error", "error": f"canonical missing: {p['ref_room']}"}
                    continue
                self._i2i(base_image_path=str(canon),
                          prompt=core.SAMESPACE_VIEW.format(view=p["view"]),
                          out_path=str(outp))
                ref_label = f"canonical:{p['ref_room']}"
            elif kind == "outdoor_t2i":
                place = descs.get("outdoor") or descs.get("whole") or ""
                sp = space_by_name.get(rn) or {}
                t2i_prompt = self._text(
                    system=core.OUTDOOR_BG_T2I_SYS,
                    user=core.build_outdoor_bg_user(analysis, sp, place, p["view"]),
                    max_tokens=900)
                (out_dir / f"outdoor_bg_prompt_{key}.txt").write_text(t2i_prompt, encoding="utf-8")
                self._t2i(prompt=t2i_prompt, out_path=str(outp))
                ref_label = "single_t2i (FP 미경유)"
            elif kind == "indoor_t2i":
                # 단일 marked_indoor 그룹 — FP 미경유 직접 T2I (구조·life 는 텍스트로)
                place = descs.get("indoor") or descs.get("whole") or ""
                sp = space_by_name.get(rn) or {}
                t2i_prompt = self._text(
                    system=core.INDOOR_BG_T2I_SYS,
                    user=core.build_indoor_bg_user(analysis, sp, place, p["view"]),
                    max_tokens=900)
                (out_dir / f"indoor_bg_prompt_{key}.txt").write_text(t2i_prompt, encoding="utf-8")
                self._t2i(prompt=t2i_prompt, out_path=str(outp))
                ref_label = "single_t2i (FP 미경유 — 단일 실내 공간)"
            else:  # marked_indoor — 최종 2D FP 에 ★이 공간 위치 1개만★ 마킹 후 i2i
                comp = fp_by_suffix.get(sfx)
                if comp is None or not comp.exists():
                    plates[rn] = {"status": "error", "error": f"final 2D FP missing (suffix={sfx})"}
                    continue
                if sfx not in positions_cache:
                    names = [
                        q["room"] for q in plans
                        if q["kind"] == "marked_indoor" and q["group_suffix"] == sfx
                    ]
                    pos_res = self._parse_json(
                        self._vision(system=core.SPACE_POS_SYS,
                                     user=core.build_space_positions_user(names),
                                     image_paths=[str(comp)], max_tokens=1200),
                        what=f"{gid} space_positions{sfx}")
                    positions_cache[sfx] = {
                        q.get("name"): (float(q.get("cx", 0.5)), float(q.get("cy", 0.5)))
                        for q in (pos_res.get("positions") or [])
                    }
                pos = positions_cache[sfx].get(rn)
                if pos is None:
                    position_fallback = True
                    logger.warning("space_set_bg %s: VLM 위치 누락 %s → 중앙 fallback", gid, rn)
                    pos = (0.5, 0.5)
                marked = out_dir / f"fp_marked_{key}.png"
                # persist-all Wave2 (B5): 마킹된 FP 는 i2i 입력용 diagnostic 중간물.
                # ★scope 는 mark_fp 한 줄만 좁게 감싼다 — 바로 아래 self._i2i(line 339)
                # 출력은 최종 plate 라 같은 scope 에 넣으면 중복 insert("최종 등록 output 이
                # 나오지 않는 scope" 가드, Codex 합의).
                from app.services.image_capture.context import generation_context
                with generation_context(
                    self.project_id, self.episode_id, stage="space_set_marked_fp",
                ):
                    core.mark_fp(str(comp), str(marked), pos[0], pos[1], p["num"])
                sp = space_by_name.get(rn) or {}
                life = [x.get("item") for x in (sp.get("life_baseline") or []) if x.get("item")]
                place = descs.get("indoor") or descs.get("whole") or ""
                self._i2i(base_image_path=str(marked),
                          prompt=core.build_marked_view_prompt(p["num"], place, p["view"], life),
                          out_path=str(outp))
                ref_label = f"marked_fp:#{p['num']}"
            entry: Dict[str, Any] = {
                "status": "ok", "kind": kind, "png": outp.name, "key": key,
                "num": p["num"], "reference": ref_label, "enclosure": p["enclosure"],
            }
            if kind == "marked_indoor" and position_fallback:
                # 진단 surface (게이트 아님) — 육안 추적용 (Codex 리뷰 MINOR)
                entry["position_missing_fallback"] = True
            plates[rn] = entry
        # ⑥ Phase 2: shot→space 배정 (LLM 1회) + 순수 조인 → shot_plate_map.
        # 실패는 비치명 — 해당 그룹 shot 들은 기존 background/prev_shot 경로 유지 (thick gate 금지).
        # max_tokens 12000: Phase 3 derive_instruction(영어 1~3문장/derive 샷)이 출력에 추가됨.
        shot_plate_map: Dict[str, Any] = {}
        assign_diags: List[Dict[str, Any]] = []
        try:
            assigns = self._parse_json(
                self._text(system=core.SHOT_ASSIGN_SYS,
                           user=core.build_shot_assign_user(analysis, shot_rows),
                           max_tokens=12000),
                what=f"{gid} shot_assign")
            shot_plate_map, assign_diags = core.build_shot_plate_map(
                (assigns or {}).get("assignments") or [], plates,
                connector_names={c.get("room") for c in connectors if c.get("room")})
        except Exception as exc:
            logger.warning("space_set_bg %s: shot 배정 실패 (비치명) — %s", gid, exc)
            assign_diags = [{"reason": "assign_call_failed",
                             "error": f"{type(exc).__name__}: {exc}"}]
        # ⑦ Phase 3: derive_from_base 샷만 배경 전용 i2i — canonical plate 를 identity
        # 참조로 카메라만 이동한 파생 plate 생성 (bg-zone-design-sot '매우 다른 각도=derive').
        # canonical plates 불변(additive). 실패 = non-blocking — spm 은 canonical 그대로
        # (조인이 이미 canonical png 를 넣어둠) + derive_failed diagnostic.
        derived_plates: Dict[str, Dict[str, Any]] = {}
        for key, entry in shot_plate_map.items():
            if entry.get("plate_action") != "derive_from_base":
                continue
            sci, shi = (int(x) for x in key.split("_", 1))
            src = out_dir / (entry.get("canonical_plate_png") or "")
            dst_name = f"bg_derived_{key}.png"
            try:
                if not src.is_file():
                    raise FileNotFoundError(f"canonical plate png missing: {src.name}")
                self._i2i(
                    base_image_path=str(src),
                    prompt=core.build_derive_plate_prompt(entry.get("derive_instruction") or ""),
                    out_path=str(out_dir / dst_name))
                entry["plate_png"] = dst_name
                derived_plates[key] = {
                    "status": "ok", "png": dst_name,
                    "canonical_plate_key": entry.get("canonical_plate_key"),
                    "space": entry.get("space"),
                    "derive_instruction": entry.get("derive_instruction"),
                }
            except Exception as exc:
                logger.warning(
                    "space_set_bg %s: derive 실패 (비치명, canonical fallback) — "
                    "shot %s space %s: %s", gid, key, entry.get("space"), exc)
                entry["derive_failed"] = True
                derived_plates[key] = {
                    "status": "error",
                    "error": f"{type(exc).__name__}: {exc}",
                    "canonical_plate_key": entry.get("canonical_plate_key"),
                    "space": entry.get("space"),
                }
                assign_diags.append({
                    "reason": "derive_failed", "scene": sci, "shot": shi,
                    "space": entry.get("space"),
                    "error": f"{type(exc).__name__}: {exc}",
                })
        return {
            "status": "ok",
            "spaces": [s.get("name") for s in all_spaces],
            "validation": analysis.get("_validation") or {},
            "frame_checks": frame_checks,
            "plates": plates,
            "connectors": connectors,
            "descs": descs,
            "assets_dir": str(out_dir),
            "shot_plate_map": shot_plate_map,
            "shot_assign_diagnostics": assign_diags,
            "derived_plates": derived_plates,
        }

    # ──────────────────────── 실행 ────────────────────────
    def _execute(self, mode: str = "resume") -> Dict[str, Any]:
        from app.core.config import settings
        if not bool(getattr(settings, "space_set_bg_enabled", False)):
            return self._not_applicable()

        plans_cp = self._load_prev_checkpoint("background_master_plan")
        scene_save_cp = self._load_prev_checkpoint("scene_save")
        shot_validator_cp = self._load_prev_checkpoint("shot_validator")
        shot_selection_cp = self._load_prev_checkpoint("shot_selection")
        world_guide_cp = self._load_prev_checkpoint("world_guide")
        scene_director_cp = self._load_prev_checkpoint("scene_director")
        plans_map = ((plans_cp or {}).get("data", {}) or {}).get("plans", {}) or {}
        if not plans_map:
            return self._not_applicable()

        from app.core.steps.background_master_plan_step import _select_scenes_for_group
        scene_segments = (
            ((scene_save_cp or {}).get("data", {}) or {}).get("segments", []) or []
        )
        # scene → primary_location fallback (shot 에 location_id 없을 때 — master_plan Phase 5 패턴.
        # 이게 없으면 모든 그룹이 scenes=0 으로 헛돈다 — canary 실측 결함 fix)
        scene_primary: Dict[int, str] = {}
        for sc in ((scene_director_cp or {}).get("data", {}) or {}).get("scenes", []) or []:
            si = sc.get("scene_index")
            primary = sc.get("primary_location", "") or ""
            if si is not None and primary:
                scene_primary[int(si)] = primary
        guide = self._world_guide_subset(world_guide_cp)
        fulltext = self._load_fulltext()
        # ★등장 빈도 필터 (사용자 2026-06-10): 1번 정도 나타나는 배경은 기준 BG 생략
        # (그 배경 샷은 샷 생성 시 즉석 — 공동 사용 가치가 없다). 여러 번 등장 = 기준 BG 생성해 공동 사용.
        min_scenes = int(getattr(settings, "space_set_bg_min_scenes", 2))

        results: Dict[str, Any] = {}
        applicable = completed = failed = 0
        for gid, entry in plans_map.items():
            if (entry or {}).get("status") != "ok":
                continue
            plan = (entry or {}).get("plan") or {}
            member_locs = {bg.get("loc_id") for bg in (plan.get("backgrounds") or []) if bg.get("loc_id")}
            if not member_locs:
                continue
            synthetic_group = {
                "group_id": gid,
                "members": [{"loc_id": loc, "label": ""} for loc in sorted(member_locs)],
            }
            scenes, _shot_ids = _select_scenes_for_group(
                synthetic_group, scene_segments, shot_validator_cp, shot_selection_cp,
                scene_primary=scene_primary)
            if len(scenes) < min_scenes:
                if scenes:
                    results[gid] = {"status": "skipped_low_frequency",
                                    "scene_count": len(scenes), "min_scenes": min_scenes}
                continue
            applicable += 1
            out_dir = self._cp_dir / "assets" / gid
            try:
                results[gid] = self._run_group(gid, scenes, guide, fulltext, out_dir)
                completed += 1
            except Exception as exc:
                logger.exception("space_set_bg group %s failed", gid)
                results[gid] = {"status": "error", "error": f"{type(exc).__name__}: {exc}"}
                failed += 1

        return {
            "applicable_count": applicable,
            "completed_count": completed,
            "failed_count": failed,
            "schema_version": SCHEMA_VERSION,
            "config_hash": self._config_hash(),
            "data": {
                "groups": results,
                "prompt_version": PROMPT_VERSION,
                # 어느 판을 소비했는지 기록이 실행을 대변하게 — config_hash 는
                # 달라진 것만 알려주고 무엇이었는지는 못 알려준다.
                "prompt_pack": {
                    "module": core.PROMPT_PACK_MODULE,
                    "stems": core.pack_identity(),
                },
            },
        }