"""검색 후보 라운드 journal — 순번 예약과 상태 전이의 단일 권위 (A4).

재실행이 후보 파일을 덮어쓰던 것보다 큰 문제는 **계보 소급 변조**였다.
form_ref 자산이 그룹당 한 row 였고 그 `file_path` 를 계속 갱신했으므로,
재검색 한 번이 과거 씨드의 입력 간선을 다른 이미지로 바꾼다. 그래서 라운드
디렉터리와 선택 자산 UUID 를 **불변**으로 두고, 그 예약·전이를 이 모듈이
단독으로 소유한다.

★step CP 는 journal 이 될 수 없다. force 는 `_execute` **전에**
`clear_checkpoint()` 로 CP 를 지우고(`step_runner.py:1194`),
`save_checkpoint` 는 `_execute` 가 **끝난 뒤**에 불린다
(`_execute_and_finalize` 7단계 중 (6)). 그래서 실행 중에 durable 한 저장소가
따로 필요하다. CP 는 최종 소비 projection 일 뿐이다.

저장 구조 — **상태를 둘이 중복 소유하지 않는다**:

    <group>/rounds/index.json          순번 예약 + manifest 경로만
    <group>/rounds/rNNN/manifest.json  ★그 라운드 상태의 유일한 SOT

index 와 manifest 가 어긋나거나, index 가 깨졌거나, registry 에 없는 라운드
디렉터리가 있으면 **디렉터리 스캔으로 복원하지 않고 fail-closed** 한다.
부분 실패의 잔재를 정답으로 승격시키지 않기 위해서다.

동시성 — 현재 StepRunner 의 단일 claim 이 allocator 직렬화 전제다. 한 그룹에
두 실행이 동시에 들어오는 구조가 생기면 이 모듈에 락이 필요하다.

계획 = docs/superpowers/plans/2026-08-02-a4-round-immutable-candidate-store.md
"""
from __future__ import annotations

import json
import os
import re
import uuid
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List

# 라운드 저장 계약의 버전. 경로 규약·manifest shape·전이 규칙이 바뀌면 올린다.
# ★`_config_hash` 와 `_group_fingerprint` 양쪽에 싣는다. 반대로 `round_id` 는
#  싣지 않는다 — 입력이 아니라 산출 identity 라 넣으면 재개가 항상 miss 된다.
#   v1 (2026-08-02): 라운드 디렉터리 · 상태 머신 · 예약 UUID.
#   v2 (2026-08-02, Codex 6차): manifest header 에 **provenance 필수**.
#     선택 필드로 두면 header 를 지우는 것만으로 결속이 통째로 무력화된다
#     (실측: provenance 삭제 + 산출·CP 동시 변조가 paid=0·completed=1 로 통과).
#
# ★**구 계약은 지원하지 않는다.** index·manifest 둘 다 이 값과 exact 일치해야
#  하고, 어긋나면 유료 호출 전에 typed failure 로 선다. 근거 = A4 는 유료
#  실행 전이고 실제 라운드 저장소가 **0개**다(전수: rounds/ 디렉터리 0 ·
#  manifest 0). 이행 경로를 지어내는 대신 거부를 계약으로 확정한다 —
#  "read-only 로 읽을 수 있다"고 써 두면 그 경로가 실제로는 index 검사에
#  막혀 실행되지 않는데도 실행되는 것처럼 읽힌다(Codex 7차 지적).
ROUND_STORAGE_CONTRACT_VERSION = 2

# 라운드 계약으로 저장된 산출임을 가리키는 표기.
ROUND_PATH_KIND = "round"

# 구 flat 저장(`<group>/cand_NN.png`)을 가리키는 표기. `r000` 은 새 계약의
# 정상 순번처럼 보이므로 쓰지 않는다.
LEGACY_ROUND_ID = "legacy"
LEGACY_PATH_KIND = "legacy_flat"
LEGACY_CONTRACT_VERSION = 0

_ROUNDS_DIRNAME = "rounds"
_INDEX_NAME = "index.json"
_MANIFEST_NAME = "manifest.json"
_ROUND_ID_RE = re.compile(r"^r(\d{3,})$")
_CAND_RE = re.compile(r"^cand_(\d+)\.png$")
_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")

#: 키 부재와 "값이 None" 을 구분하기 위한 표식. `dict.get()` 은 둘을 같은
#: 것으로 만든다 — 그러면 명시적 null 이 "구 스키마"로 오인된다(실측).
MISSING = object()

#: 라운드를 열 때 header 에 고정하는 역사 기록 키.
PROVENANCE_KEYS = ("ref_pack_version", "target_policy_version")

# CP 에 투영되는 **불변** 필드 — journal 의 최종 산출과 exact 대조 대상.
# `reused` · `round_replayed` 는 실행 시 동적 값이라 제외한다.
# ★journal 이 CP 의 단일 권위이므로 이 shape 검증도 journal 이 소유한다.
ROUND_CP_PROJECTION_KEYS = (
    "status",
    "form_ref_asset_id",
    "form_ref_path",
    "form_ref_sha256",
    "round_id",
    "path_kind",
    "round_contract_version",
    "ref_pack_version",
    "target_policy_version",
    "group_fingerprint",
)


def typed_equal(a: Any, b: Any) -> bool:
    """값과 **타입 계약**을 함께 본다.

    ★`True == 1` 이라 순수 값 비교는 bool 을 정수로 받아들인다 — 결속과
    exact 대조가 둘 다 뚫린다(실측).
    """
    if isinstance(a, bool) != isinstance(b, bool):
        return False
    if isinstance(a, float) != isinstance(b, float):
        return False
    return a == b


class RoundState(str, Enum):
    OPENED = "opened"
    CANDIDATES = "candidates"
    SELECTED = "selected"
    ASSET_BOUND = "asset_bound"
    FINALIZED = "finalized"
    ABANDONED = "abandoned"


# 더 쓸 수 없는 상태 — FINALIZED 뿐 아니라 ABANDONED 도 불변 경계다.
TERMINAL_STATES = frozenset({RoundState.FINALIZED, RoundState.ABANDONED})
# 아직 살아 있는(이어갈 수 있는) 상태
OPEN_STATES = frozenset({
    RoundState.OPENED, RoundState.CANDIDATES,
    RoundState.SELECTED, RoundState.ASSET_BOUND,
})
# 자산이 DB 에 durable 해진 뒤의 상태 — 여기부터는 유료 없이 이어갈 수 있다.
BOUND_STATES = frozenset({RoundState.SELECTED, RoundState.ASSET_BOUND})
# ★선택 산출이 **저장돼 있어야만** 가능한 상태들. 비어 있으면 그 상태 자체가
#  계약 위반이다 — 재개가 유료 재검색으로 하강하는 것을 막는다.
RESULT_REQUIRED_STATES = frozenset({
    RoundState.SELECTED, RoundState.ASSET_BOUND, RoundState.FINALIZED,
})

# ★합법 전이 그래프. 없으면 nonterminal 에서 **아무 상태로나** 갈 수 있어
#  ASSET_BOUND 를 건너뛴 FINALIZED, CANDIDATES 로의 역전이가 전부 통과한다.
#  같은 상태로의 no-op 도 허용하지 않는다 — 호출측이 상태를 보고 건너뛴다.
_ALLOWED_TRANSITIONS: Dict[RoundState, frozenset] = {
    RoundState.OPENED: frozenset({
        RoundState.CANDIDATES, RoundState.SELECTED, RoundState.ABANDONED}),
    RoundState.CANDIDATES: frozenset({
        RoundState.SELECTED, RoundState.ABANDONED}),
    RoundState.SELECTED: frozenset({
        RoundState.ASSET_BOUND, RoundState.ABANDONED}),
    RoundState.ASSET_BOUND: frozenset({
        RoundState.FINALIZED, RoundState.ABANDONED}),
}


def _fail(code: str, message: str) -> None:
    from app.core.errors import AppError

    raise AppError(code=code, message=message, status_code=422)


@dataclass(frozen=True)
class Round:
    round_id: str
    state: RoundState
    input_fp: str
    contract_version: int
    preallocated_asset_id: str
    #: 라운드를 **열 때** 고정한 역사 기록(팩·정책 등). 이후 불변이며,
    #: 산출의 같은 이름 필드는 이 값과 결속된다.
    #: ★"현재 값과 같은가"로 검사하면 안 된다 — 무효화 판단은 **지문**의
    #: 몫이고(정책은 지문에서 의도적으로 빠져 있다) 이 값은 **기록**의 몫이다.
    #: 둘을 섞으면 정상적인 역사 provenance 까지 거부하게 된다.
    provenance: Dict[str, str] = field(default_factory=dict)
    # FINALIZED 에서 확정되는 최종 소비 projection. step CP 가 날아가도
    # 이것이 남아 있으면 **유료 0콜**로 그 라운드를 다시 투영할 수 있다.
    result: Dict[str, Any] = field(default_factory=dict)

    def is_resumable(self, *, input_fp: str) -> bool:
        """이어갈 수 있는가 — 지문과 계약이 **exact 일치**할 때만.

        어긋난 라운드를 이어가면 다른 입력의 산출이 한 라운드에 섞인다.
        """
        return (
            self.state in OPEN_STATES
            and self.input_fp == input_fp
            and self.contract_version == ROUND_STORAGE_CONTRACT_VERSION
        )

    def is_reusable_finalized(self, *, input_fp: str) -> bool:
        """유료 호출 없이 CP 로 다시 투영할 수 있는 완료 라운드인가."""
        return (
            self.state == RoundState.FINALIZED
            and self.input_fp == input_fp
            and self.contract_version == ROUND_STORAGE_CONTRACT_VERSION
            and bool(self.result)
        )


@dataclass(frozen=True)
class RoundResolution:
    """이번 실행이 어느 라운드에서 이어가는가."""

    round: Round
    #: True 면 검색·판정·다운로드·asset INSERT 를 **하지 않는다**.
    reused_finalized: bool


def _rounds_dir(group_dir: Path) -> Path:
    return Path(group_dir) / _ROUNDS_DIRNAME


def _index_path(group_dir: Path) -> Path:
    return _rounds_dir(group_dir) / _INDEX_NAME


def _round_dir(group_dir: Path, round_id: str) -> Path:
    return _rounds_dir(group_dir) / round_id


def _manifest_path(group_dir: Path, round_id: str) -> Path:
    return _round_dir(group_dir, round_id) / _MANIFEST_NAME


def _atomic_write_json(path: Path, payload: Dict[str, Any]) -> None:
    """temp→replace. 중간에 끊겨도 반쪽 파일이 남지 않는다."""
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    text = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True)
    with open(tmp, "w", encoding="utf-8") as fh:
        fh.write(text)
        fh.flush()
        os.fsync(fh.fileno())
    tmp.replace(path)


def _exact_int(value: Any, *, path: Path, field: str,
               code: str = "round_manifest_corrupt") -> int:
    """정확한 정수만 받는다 — `bool` 은 `int` 의 서브클래스라 새어 들어온다.

    실측: `contract_version=true` 가 `int()` 를 통과해 1 이 되고, 그 라운드가
    `is_resumable=True` 로 이어졌다. ★`code` 를 고정하면 index 손상이
    manifest 손상으로 잘못 분류된다.
    """
    if isinstance(value, bool) or not isinstance(value, int):
        _fail(code, f"{path}: {field} 가 정수가 아니다: {value!r}")
    return int(value)


def _read_json(path: Path, *, code: str) -> Dict[str, Any]:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        _fail(code, f"{path} 를 읽을 수 없다: {exc}")
        raise  # pragma: no cover - _fail 이 항상 raise


def _nonblank_str(value: Any) -> bool:
    return isinstance(value, str) and bool(value.strip())


def _validate_result_shape(result: Dict[str, Any], *, state: RoundState,
                           path: Path) -> None:
    """저장된 산출이 그 상태에서 **쓸 수 있는 모양**인가.

    ★journal 이 CP 의 단일 권위라면, CP 없이 재투영할 때 이 산출이 **그대로
    성공 CP** 가 된다. shape 를 안 보면 손상된 journal 이 완료로 승격된다 —
    실측: `status='failed'` 인 산출이 `completed_count=1` 로 올라갔고,
    `path_kind='legacy_flat'` · `ref_pack_version='tampered-pack'` 도 그대로
    새 CP 가 됐다.
    """
    def bad(msg: str) -> None:
        _fail("round_manifest_projection_invalid", f"{path}: {msg}")

    # 어느 상태든 선택 산출은 성공이어야 한다 — 실패는 라운드를 열어 둔다.
    if result.get("status") != "ok":
        bad(f"status={result.get('status')!r} (저장된 산출은 'ok' 여야 한다)")
    if not _nonblank_str(result.get("form_ref_path")):
        bad(f"form_ref_path={result.get('form_ref_path')!r}")
    sha = result.get("form_ref_sha256")
    if not isinstance(sha, str) or not _SHA256_RE.match(sha):
        bad(f"form_ref_sha256={sha!r} (64자리 소문자 hex 여야 한다)")
    if state != RoundState.FINALIZED:
        return
    # 완료 산출만이 CP projection 으로 그대로 나간다 — 전 필드를 본다.
    if result.get("path_kind") != ROUND_PATH_KIND:
        bad(f"path_kind={result.get('path_kind')!r} "
            f"(완료 산출은 {ROUND_PATH_KIND!r} 여야 한다)")
    for key in ("ref_pack_version", "target_policy_version",
                "group_fingerprint", "round_id"):
        if not _nonblank_str(result.get(key)):
            bad(f"{key}={result.get(key)!r} (비어 있지 않은 문자열이어야 한다)")
    _exact_int(result.get("round_contract_version"), path=path,
               field="result.round_contract_version",
               code="round_manifest_projection_invalid")
    missing = [k for k in ROUND_CP_PROJECTION_KEYS if k not in result]
    if missing:
        bad(f"완료 산출에 투영 필드가 없다: {missing}")


def _assert_no_stray_round_dirs(group_dir: Path, *, known: set,
                                idx_path: Path) -> None:
    """registry 에 없는 라운드 디렉터리를 주워 담지 않는다.

    ★index 가 아예 없을 때도 검사한다 — manifest 를 쓰고 index 를 쓰기 전에
    끊긴 상태가 정확히 그 모양이고, 그때 "라운드 없음"으로 답하면 다음 실행이
    같은 번호를 다시 열어 예약 UUID 를 갈아치운다.
    """
    rounds_dir = _rounds_dir(group_dir)
    if not rounds_dir.exists():
        return
    for child in sorted(rounds_dir.iterdir()):
        if not child.is_dir() or not _ROUND_ID_RE.match(child.name):
            continue
        if child.name not in known:
            _fail(
                "round_registry_mismatch",
                f"registry({idx_path}) 에 없는 라운드 디렉터리: {child} — "
                f"스캔으로 복원하지 않는다")


def read_index(group_dir: Path) -> Dict[str, Any]:
    """registry 를 읽고 디스크와의 정합을 확인한다.

    ★깨졌거나 어긋나면 **디렉터리 스캔으로 복원하지 않는다.** 부분 실패
    잔재를 세면 이미 예약된 번호를 다시 내주게 된다.
    """
    idx_path = _index_path(group_dir)
    if not idx_path.exists():
        # ★index 가 없다고 "라운드가 하나도 없다"는 뜻은 아니다. manifest 를
        #  쓰고 index 를 쓰기 전에 끊기면 r001 디렉터리만 남는데, 그때 빈
        #  registry 를 돌려주면 다음 실행이 **같은 번호를 다시 열어 기존
        #  manifest·예약 UUID 를 덮어쓴다**(실측: UUID 보존 False).
        _assert_no_stray_round_dirs(group_dir, known=set(), idx_path=idx_path)
        return {"contract_version": ROUND_STORAGE_CONTRACT_VERSION,
                "rounds": []}
    index = _read_json(idx_path, code="round_index_corrupt")
    if not isinstance(index, dict):
        _fail("round_index_corrupt", f"{idx_path}: root 가 dict 가 아니다")
    idx_contract = _exact_int(index.get("contract_version"), path=idx_path,
                              field="contract_version",
                              code="round_index_corrupt")
    if idx_contract != ROUND_STORAGE_CONTRACT_VERSION:
        # ★타입만 보고 값을 안 보면 999 도 통과한다 — "계약 exact 일치"를
        #  내세운 상태 머신과 모순이다.
        _fail("round_index_corrupt",
              f"{idx_path}: 지원하지 않는 저장 계약 {idx_contract} "
              f"(현재 {ROUND_STORAGE_CONTRACT_VERSION})")
    entries = index.get("rounds")
    if not isinstance(entries, list):
        _fail("round_index_corrupt", f"{idx_path}: rounds 가 목록이 아니다")
    known = set()
    for entry in entries:
        if not isinstance(entry, dict):
            # ★dict 가 아니면 아래 `.get` 이 AttributeError 로 새어 계약 위반이
            #  "스텝 고장"으로 둔갑한다(422 대신 500).
            _fail("round_index_corrupt",
                  f"{idx_path}: rounds 원소가 dict 가 아니다: {entry!r}")
        rid = (entry or {}).get("round_id")
        if not isinstance(rid, str) or not _ROUND_ID_RE.match(rid):
            _fail("round_index_corrupt", f"{idx_path}: 잘못된 round_id {rid!r}")
        if rid in known:
            # 중복이 있으면 순번 예약이 이미 깨진 것이다 — 어느 쪽이 그
            # 번호의 주인인지 알 수 없다.
            _fail("round_index_corrupt", f"{idx_path}: round_id 중복 {rid!r}")
        # 경로도 **exact** 로 본다. 임의 경로를 허용하면 registry 가 다른
        # 라운드의 manifest 를 가리켜도 통과한다.
        want_path = f"{_ROUNDS_DIRNAME}/{rid}/{_MANIFEST_NAME}"
        got_path = (entry or {}).get("manifest_path")
        if got_path != want_path:
            _fail("round_index_corrupt",
                  f"{idx_path}: {rid} 의 경로가 {got_path!r} 이다 "
                  f"(기대 {want_path!r})")
        known.add(rid)
    _assert_no_stray_round_dirs(group_dir, known=known, idx_path=idx_path)
    return index


def _round_seq(round_obj: "Round") -> int:
    """라운드 번호의 **수치** 순번. 문자열 비교는 r1000 이후 뒤집힌다."""
    m = _ROUND_ID_RE.match(round_obj.round_id)
    return int(m.group(1)) if m else -1


def _next_round_id(index: Dict[str, Any]) -> str:
    """★열린 partial 라운드도 이미 예약된 번호이므로 반드시 카운트한다."""
    used = []
    for entry in index.get("rounds") or []:
        m = _ROUND_ID_RE.match(entry.get("round_id") or "")
        if m:
            used.append(int(m.group(1)))
    return f"r{(max(used) + 1) if used else 1:03d}"


def load_round(group_dir: Path, round_id: str) -> Round:
    """manifest 를 읽고 **shape 과 identity 까지** 확인한다.

    ★identity 검증이 없으면 manifest 의 `round_id` 만 바꿔도 그 값이 그대로
    권위가 된다(실측: r001 의 manifest 에 r999 를 적자 `resolve_round` 가
    디스크에 없는 r999 를 정상 반환했고, 그 상태는 **유료 `_run_group` 을 탄
    뒤에야** manifest missing 으로 죽었다).
    """
    path = _manifest_path(group_dir, round_id)
    if not path.exists():
        _fail("round_manifest_missing", f"{path} 가 없다")
    data = _read_json(path, code="round_manifest_corrupt")
    return parse_manifest(data, round_id=round_id, path=path)


def parse_manifest(data: Any, *, round_id: str, path: Path) -> Round:
    """manifest dict 를 검증하고 `Round` 로 만든다.

    ★**읽기와 쓰기가 같은 validator 를 쓴다.** 쓰고 나서 읽으며 검증하면,
    잘못된 산출이 이미 durable 해진 뒤에 오류가 난다 — 실측: bad sha 로 전이를
    요청하자 오류는 났지만 `state='selected'` 와 그 sha 가 영속됐고, 이어지는
    재개도 같은 오류로 막혀 **그 라운드가 영구 재개 불능**이 됐다. 그러면
    "실패면 라운드를 열어 두고 이어간다"는 계약과 정반대가 된다.
    """
    if not isinstance(data, dict):
        _fail("round_manifest_corrupt",
              f"{path}: manifest root 가 dict 가 아니다")
    got_id = data.get("round_id")
    if got_id != round_id:
        _fail("round_manifest_identity_mismatch",
              f"{path}: manifest 가 {got_id!r} 이라고 말한다 (기대 "
              f"{round_id!r}) — 그 값을 권위로 삼지 않는다")
    try:
        state = RoundState(data["state"])
    except (KeyError, ValueError):
        _fail("round_manifest_corrupt", f"{path}: 알 수 없는 state")
        raise  # pragma: no cover
    input_fp = data.get("input_fp")
    if not isinstance(input_fp, str) or not input_fp:
        _fail("round_manifest_corrupt",
              f"{path}: input_fp 가 비어 있지 않은 문자열이 아니다")
    raw_uuid = data.get("preallocated_asset_id")
    try:
        # canonical 형태까지 본다 — 자산 identity 라 임의 문자열이면 안 된다.
        asset_uuid = str(uuid.UUID(str(raw_uuid)))
    except (ValueError, AttributeError, TypeError):
        _fail("round_manifest_corrupt",
              f"{path}: preallocated_asset_id 가 UUID 가 아니다: {raw_uuid!r}")
        raise  # pragma: no cover
    if str(raw_uuid) != asset_uuid:
        # ★canonical 변환만 하고 원문을 안 보면 하이픈 없는·대문자 표현이
        #  통과한다. 자산 id 는 DB 조회 키라 표현이 흔들리면 안 된다.
        _fail("round_manifest_corrupt",
              f"{path}: preallocated_asset_id 가 canonical UUID 가 아니다: "
              f"{raw_uuid!r}")
    result = data.get("result")
    if result is not None and not isinstance(result, dict):
        _fail("round_manifest_corrupt", f"{path}: result 가 dict 가 아니다")
    contract_version = _exact_int(data.get("contract_version"), path=path,
                                  field="contract_version")
    if contract_version != ROUND_STORAGE_CONTRACT_VERSION:
        # ★구 계약은 지원하지 않는다(위 상수 주석). 읽어서 순번만 세는 절충을
        #  두면 그 경로가 실제로는 index 검사에 막혀 실행되지 않는다.
        _fail("round_manifest_unsupported_contract",
              f"{path}: 저장 계약 {contract_version} 은 지원하지 않는다 "
              f"(현재 {ROUND_STORAGE_CONTRACT_VERSION})")
    # ★provenance 는 **필수**다. 선택으로 두면 header 를 지우는 것만으로 아래
    #  결속이 통째로 무력화된다(실측: provenance 삭제 + 산출·CP 동시 변조가
    #  paid=0 · completed=1 로 통과했다). 부재와 explicit null 도 가른다.
    raw_prov = data.get("provenance", MISSING)
    if raw_prov is MISSING:
        _fail("round_manifest_corrupt", f"{path}: provenance header 가 없다")
    if not isinstance(raw_prov, dict):
        _fail("round_manifest_corrupt",
              f"{path}: provenance 가 dict 가 아니다: {raw_prov!r}")
    if set(raw_prov) != set(PROVENANCE_KEYS):
        _fail("round_manifest_corrupt",
              f"{path}: provenance 키가 {sorted(raw_prov)} 다 "
              f"(정확히 {sorted(PROVENANCE_KEYS)} 여야 한다)")
    for key in PROVENANCE_KEYS:
        if not _nonblank_str(raw_prov[key]):
            _fail("round_manifest_corrupt",
                  f"{path}: provenance.{key}={raw_prov[key]!r} "
                  f"(비어 있지 않은 문자열이어야 한다)")
    provenance = dict(raw_prov)
    if state in RESULT_REQUIRED_STATES:
        if not result:
            # ★SELECTED/ASSET_BOUND 는 **선택 산출이 저장돼 있어야만** 가능한
            #  상태다. 비어 있으면 재개가 유료 재검색으로 내려간다(실측).
            _fail("round_manifest_corrupt",
                  f"{path}: {state.value} 인데 산출이 없다 — 그 상태는 산출 "
                  f"없이 존재할 수 없다")
        # ★header 와 산출을 **서로 묶는다.** 각 필드의 형식만 보면 header 가
        #  산출과 다른 라운드를 말해도 통과한다. 실측: ASSET_BOUND 에서
        #  `preallocated_asset_id` 만 새 UUID 로 바꾸고 재개하자 자산을 **또
        #  INSERT** 해 row 가 2개가 되고 새 UUID 로 완료됐다 — "라운드당 예약
        #  UUID 불변" 계약이 깨진다.
        bindings = [
            ("round_id", result.get("round_id"), round_id),
            ("group_fingerprint", result.get("group_fingerprint"), input_fp),
            ("round_contract_version",
             result.get("round_contract_version"), contract_version),
        ]
        if state in (RoundState.ASSET_BOUND, RoundState.FINALIZED):
            bindings.append(
                ("form_ref_asset_id", result.get("form_ref_asset_id"),
                 asset_uuid))
        # ★팩·정책은 라운드를 열 때 header 에 고정된 **역사값**이다. 산출이
        #  그것과 다르면 부분 변조다 — "현재 값과 같은가"가 아니라 "그때의
        #  값과 같은가"로 본다.
        # 위에서 key set 을 exact 로 세웠으므로 전 키가 있다.
        for key in PROVENANCE_KEYS:
            bindings.append((key, result.get(key), provenance[key]))
        for field, got, want in bindings:
            # ★타입까지 본다 — `True == 1` 이라 순수 값 비교는 bool 을 정수로
            #  받아들여 결속이 뚫린다(실측).
            if not typed_equal(got, want):
                _fail("round_manifest_header_mismatch",
                      f"{path}: 산출의 {field}={got!r} 이 header 의 {want!r} 과 "
                      f"다르다 — 같은 라운드의 것이 아니다")
        _validate_result_shape(result, state=state, path=path)
    return Round(
        round_id=round_id,
        state=state,
        input_fp=input_fp,
        contract_version=contract_version,
        preallocated_asset_id=asset_uuid,
        provenance=provenance,
        result=dict(result or {}),
    )


def load_registered_round(group_dir: Path, round_id: str) -> Round:
    """**registry 에 등록된** 라운드만 읽는다.

    ★`load_round` 를 직접 부르면 `index.json` 을 우회한다 — registry 가
    깨졌거나 그 라운드가 등록돼 있지 않아도 manifest 만으로 통과한다(실측:
    index.json 을 깨뜨린 뒤에도 CP 재사용 검증이 통과했다). journal 이 단일
    권위라면 registry 정합이 먼저다.
    """
    index = read_index(group_dir)
    known = {(e or {}).get("round_id") for e in (index.get("rounds") or [])}
    if round_id not in known:
        _fail("round_not_registered",
              f"{round_id} 가 registry 에 등록돼 있지 않다")
    return load_round(group_dir, round_id)


def candidate_dir(group_dir: Path, round_id: str) -> Path:
    """그 라운드의 후보 사진이 사는 곳. **라운드 밖으로 새지 않는다.**"""
    return _round_dir(group_dir, round_id)


def next_candidate_index(group_dir: Path, round_id: str) -> int:
    """그 라운드에서 다음 후보 파일이 쓸 번호.

    ★라운드 **번호** 예약과 달리 이쪽은 디렉터리를 본다 — 층이 다르다.
    라운드 번호는 journal 이 소유하는 전역 순번이라 스캔으로 복원하면 이미
    예약된 번호를 다시 내주지만, 후보 번호는 **그 라운드 디렉터리 안에서만**
    의미가 있고 거기 있는 파일이 곧 사실이다.

    이 값이 없으면 실패로 열린 채 남은 라운드를 다음 재개가 이어갈 때 1번부터
    다시 써서 **그 라운드의 이전 후보를 덮어쓴다.**
    """
    d = candidate_dir(group_dir, round_id)
    if not d.is_dir():
        return 1
    used = []
    for child in d.iterdir():
        m = _CAND_RE.match(child.name)
        if m:
            used.append(int(m.group(1)))
    return (max(used) + 1) if used else 1


def list_rounds(group_dir: Path) -> List[Round]:
    index = read_index(group_dir)
    return [load_round(group_dir, e["round_id"])
            for e in (index.get("rounds") or [])]


def find_open_round(group_dir: Path) -> Round | None:
    """살아 있는 라운드 — 있으면 하나뿐이어야 한다."""
    live = [r for r in list_rounds(group_dir) if r.state in OPEN_STATES]
    if len(live) > 1:
        _fail(
            "round_multiple_open",
            f"열린 라운드가 {len(live)}개다 — 어느 쪽이 권위인지 알 수 없다")
    return live[0] if live else None


def open_round(group_dir: Path, *, input_fp: str,
               provenance: Dict[str, str]) -> Round:
    """새 라운드를 열고 **자산 UUID까지 미리 채번**해 durable 하게 남긴다.

    ★UUID 를 여기서 박아 두지 않으면, 같은 라운드를 이어가는 재개가 자산을
    또 INSERT 한다. 미리 정해 두면 어느 저장소가 먼저 끊겨도 재개가 같은
    UUID 로 수렴하고, 불일치는 검증에서 드러난다.
    """
    index = read_index(group_dir)
    live = find_open_round(group_dir)
    if live is not None:
        _fail(
            "round_already_open",
            f"이미 열린 라운드가 있다: {live.round_id} ({live.state.value}) — "
            f"닫고 나서 새로 연다")
    round_id = _next_round_id(index)
    payload = {
        "round_id": round_id,
        "state": RoundState.OPENED.value,
        "input_fp": input_fp,
        "contract_version": ROUND_STORAGE_CONTRACT_VERSION,
        "preallocated_asset_id": str(uuid.uuid4()),
        # ★역사 기록을 **여는 순간** 고정한다. 뒤에 현재 값을 다시 읽으면
        #  재개 시점의 값이 섞여 "그때 무엇으로 만들었는가"가 흔들린다.
        "provenance": dict(provenance),
    }
    # 쓰기 전에 검증한다 — 읽기와 같은 validator 를 쓴다.
    parse_manifest(payload, round_id=round_id,
                   path=_manifest_path(group_dir, round_id))
    # 순서: manifest(상태 SOT) 먼저, 그 다음 index(예약). 반대로 하면 예약만
    # 있고 상태가 없는 창이 생겨 registry 정합 검사에 걸린다.
    _atomic_write_json(_manifest_path(group_dir, round_id), payload)
    entries = list(index.get("rounds") or [])
    entries.append({
        "round_id": round_id,
        # ★index 는 경로만 소유한다. 상태는 manifest 단독 소유.
        "manifest_path": f"{_ROUNDS_DIRNAME}/{round_id}/{_MANIFEST_NAME}",
    })
    _atomic_write_json(_index_path(group_dir), {
        "contract_version": ROUND_STORAGE_CONTRACT_VERSION,
        "rounds": entries,
    })
    return load_round(group_dir, round_id)


def transition_round(
    group_dir: Path, round_id: str, new_state: RoundState,
    *, result: Dict[str, Any] | None = None,
) -> Round:
    """상태 전이 — 종료된 라운드는 더 쓸 수 없다.

    ``result`` 를 주면 상태와 **한 번의 원자 쓰기**로 함께 확정한다. 상태와
    산출을 따로 쓰면 "FINALIZED 인데 result 가 없는" 창이 생기고, 그 창에서
    끊기면 재개가 완료 라운드를 재사용할 수 없다.
    """
    current = load_round(group_dir, round_id)
    if current.state in TERMINAL_STATES:
        _fail(
            "round_immutable",
            f"{round_id} 는 {current.state.value} 로 닫혔다 — 이후 쓰기 금지")
    allowed = _ALLOWED_TRANSITIONS.get(current.state, frozenset())
    if new_state not in allowed:
        # ★없으면 ASSET_BOUND 를 건너뛴 FINALIZED, SELECTED→CANDIDATES 역전이가
        #  전부 통과한다. 상태가 계약을 지키지 않으면 재개 지점도 못 믿는다.
        _fail(
            "round_illegal_transition",
            f"{round_id}: {current.state.value} → {new_state.value} 는 "
            f"허용되지 않는다 (가능: "
            f"{sorted(s.value for s in allowed)})")
    # 안 주면 기존 것을 보존한다 — 전이마다 지우면 안 된다.
    effective = dict(result) if result is not None else current.result
    if new_state in RESULT_REQUIRED_STATES and not effective:
        # ★산출 없이 그 상태로 갈 수 있으면, 재개가 "선택은 끝났다"는 라운드
        #  에서 산출을 못 찾아 **유료 재검색으로 하강**한다.
        _fail("round_result_missing",
              f"{round_id}: {new_state.value} 는 저장된 산출을 요구한다")
    payload = {
        "round_id": current.round_id,
        "state": new_state.value,
        "input_fp": current.input_fp,
        "contract_version": current.contract_version,
        "preallocated_asset_id": current.preallocated_asset_id,
        "provenance": dict(current.provenance),
        "result": effective,
    }
    # ★**쓰기 전에** 검증한다. 쓰고 나서 읽으며 검증하면 잘못된 산출이 이미
    #  durable 해진 뒤에 오류가 나고, 그 라운드는 영구 재개 불능이 된다
    #  (실측: bad sha 로 전이를 요청하자 오류는 났지만 state='selected' 와
    #  그 sha 가 영속됐고 이어지는 재개도 같은 오류로 막혔다).
    mpath = _manifest_path(group_dir, round_id)
    parse_manifest(payload, round_id=round_id, path=mpath)
    _atomic_write_json(mpath, payload)
    return load_round(group_dir, round_id)


def finalize_round(
    group_dir: Path, round_id: str, *, result: Dict[str, Any],
) -> Round:
    """라운드를 닫는다. 이후 bytes·manifest·asset 전부 불변."""
    if not isinstance(result, dict) or not result:
        _fail("round_result_missing",
              f"{round_id}: 최종 산출 없이 닫을 수 없다")
    return transition_round(group_dir, round_id, RoundState.FINALIZED,
                            result=result)


def abandon_round(group_dir: Path, round_id: str) -> Round:
    """라운드를 버린다. 번호는 소비된 채 남는다(재사용 금지)."""
    return transition_round(group_dir, round_id, RoundState.ABANDONED)


def resolve_round(
    group_dir: Path, *, input_fp: str, force: bool,
    provenance: Dict[str, str],
) -> RoundResolution:
    """이번 실행이 쓸 라운드를 정한다 (계획 §2.3 행렬의 단일 구현).

    | 상황 | 동작 |
    |---|---|
    | explicit force | 열린 라운드를 ABANDONED 로 닫고 **반드시 새 라운드** |
    | resume · 지문 exact 일치 | 그 라운드를 이어간다 |
    | resume · drift | ABANDONED → 새 라운드 (번호는 소비된 채 남는다) |
    | FINALIZED 인데 최종 CP 만 없음 | **유료 0콜**로 그 라운드를 재투영 |

    ★마지막 행이 없으면 `_execute` 종료 후 `save_checkpoint` 직전 crash 가
    유료 전체 재실행으로 바뀐다.
    """
    rounds = list_rounds(group_dir)
    live = [r for r in rounds if r.state in OPEN_STATES]
    if len(live) > 1:
        _fail("round_multiple_open",
              f"열린 라운드가 {len(live)}개다 — 어느 쪽이 권위인지 알 수 없다")

    if force:
        # 명시 force 는 재사용을 우회한다 — 완료 라운드가 있어도 새로 연다.
        for r in live:
            abandon_round(group_dir, r.round_id)
        return RoundResolution(
            open_round(group_dir, input_fp=input_fp,
                       provenance=provenance), False)

    if live:
        current = live[0]
        if current.is_resumable(input_fp=input_fp):
            return RoundResolution(current, False)
        abandon_round(group_dir, current.round_id)
        return RoundResolution(
            open_round(group_dir, input_fp=input_fp,
                       provenance=provenance), False)

    finalized = [r for r in rounds if r.is_reusable_finalized(input_fp=input_fp)]
    if finalized:
        # 가장 최근 라운드가 CP 에 못 닿은 것이므로 그것을 투영한다.
        # ★문자열로 비교하면 r1000 이후 뒤집힌다("r999" > "r1000").
        latest = max(finalized, key=_round_seq)
        return RoundResolution(latest, True)
    return RoundResolution(
        open_round(group_dir, input_fp=input_fp, provenance=provenance), False)