"""Deterministic extraction of implementation-planning verification items."""
from __future__ import annotations

import hashlib
import json
import re
from collections.abc import Mapping, Sequence
from copy import deepcopy
from dataclasses import dataclass
from typing import Any

from .build_tools import command_invokes_build_tool
from .design_snapshot import build_design_snapshot
from .design_surfaces import (
    DesignSurfaceError,
    detect_design_surfaces,
    expected_prep_plan_item_id,
)


class PlanItemContractError(ValueError):
    """The implementation-planning body cannot produce exact plan items."""


PLAN_ITEM_SOURCES = (
    ("optionCandidates", "P-Opt", "4.5.1"),
    ("stepwiseExecution", "P-Step", "4.5.4"),
    ("dependencyMigrationRisk", "P-Dep", "4.5.5"),
    ("validationChecklist", "P-Val", "4.5.6"),
    ("rollbackStrategy", "P-Rb", "4.5.7"),
    ("requirementCoverage", "P-Req", "4.5.8"),
)

_SUBJECT_FIELDS = {
    "optionCandidates": ("subject", "name", "title", "action"),
    "stepwiseExecution": ("subject", "action", "title", "name"),
    "dependencyMigrationRisk": ("subject", "item", "title", "name", "action"),
    "validationChecklist": ("subject", "check", "title", "name", "action"),
    "rollbackStrategy": ("subject", "action", "title", "name"),
    "requirementCoverage": (
        "subject",
        "requirement",
        "originalRequirementId",
        "title",
        "name",
        "action",
    ),
}

_SELECTED_DIRECTION_CONTRACT = "selected-direction"


def _non_empty_string(value: object) -> str | None:
    if isinstance(value, str) and value.strip():
        return value
    return None


# 자연키가 있는 소스. 이 필드는 요소의 *정체*이지 내용이 아니다 — 명령이나 문구를 고쳐도
# 같은 요소로 남고, 앞의 형제가 지워져도 값이 변하지 않는다.
_NATURAL_KEY_FIELDS: dict[str, tuple[str, ...]] = {
    "optionCandidates": ("name",),
    "dependencyMigrationRisk": ("id",),
    "validationChecklist": ("id",),
    "rollbackStrategy": ("id",),
    "requirementCoverage": ("id", "originalRequirementId"),
}


def _element_key(value: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", value.strip().lower()).strip("-")


def _element_id(kind: str, key: str, fallback: str) -> str:
    """이 요소의 안정 식별자. 자연키가 없으면 위치 id 로 떨어진다.

    `P-Opt-N` / `P-Val-N` / `P-Req-N` 같은 위치 id 는 서수라, 자기수정이 요소
    하나를 지우면 뒤가 전부 밀린다 — 재작성 전의 `P-Req-16` 과 후의 `P-Req-16` 이
    같은 요소라는 보장이 없다(`prompts/lead/plan-body-verification.md` §7). 그래서
    라운드나 run 을 가로질러 판정을 이월할 수 없고, 검증 표면이 매번 전체로
    돌아간다.

    `elementId` 는 그 대조를 가능하게 한다. 위치가 아니라 요소가 스스로 지닌
    이름(`R-001`, `VC-011`, 스테이지·스텝 좌표)에서 만들어지므로 형제의 삭제에
    영향받지 않는다. 표시와 인용은 계속 `id` 를 쓴다 — 이 값은 대조용이다.

    자연키가 없는 소스는 위치 id 로 떨어지고, 그 항목은 삭제를 가로질러 안정하지
    않다. 소비자는 `elementId == id` 인 항목을 그렇게 다뤄야 한다.
    """
    normalized = _element_key(key)
    return f"{kind}:{normalized}" if normalized else fallback


# 자기 기록 블록. 계획은 실행 지시와 그 실행의 감사 기록을 한 산출물에 담는데,
# 게이트가 둘을 구분하지 않으면 기록의 부정확이 구현 착수를 막는다 — 실측 run 에서
# 남은 차단 4건이 전부 "사용자가 내린 결정을 계획이 정확히 반영했는가" 였다.
#
# 커버리지 행이 유일한 기록 항목이다. 승인 처분과 decision 참조는 그 행의 필드이고,
# 나머지 소스는 전부 무엇을 어떻게 만드는지를 말한다.
_RECORD_SOURCES = frozenset({"requirementCoverage"})


def _item_block(source: str) -> str:
    return "record" if source in _RECORD_SOURCES else "execution"


def _stage_scope(row: Mapping[str, Any] | None, *coordinates: int) -> list[int] | None:
    """이 항목이 걸리는 스테이지. 어디에도 안 걸리면 ``None``.

    게이트 범위를 다음 실행 스테이지로 좁히려면 항목마다 이 값이 있어야 한다.
    스텝과 design-prep 은 id 자체가 좌표를 지니지만, 커버리지 행과 검증 행은
    `stageRefs` 를 실어야 알 수 있다 — 그 전에는 `coveredBy` 산문을 정규식으로
    읽는 수밖에 없었고, 그것은 게이트 판정의 근거가 되지 못한다.

    ``None`` 은 "스테이지에 걸리지 않는다" 이고, 소비처는 이를 **모든 스테이지에
    적용** 으로 읽는다. 그게 안전한 쪽이고 지금 동작과 같다. `P-Opt-*` 처럼 계획
    전체의 성격을 판정하는 항목이 여기 해당하고, `stageRefs` 를 아직 안 쓰는 계획도
    같은 자리로 떨어진다.
    """
    if coordinates:
        return [int(value) for value in coordinates]
    refs = (row or {}).get("stageRefs")
    if not isinstance(refs, list):
        return None
    stages = [
        value for value in refs
        if isinstance(value, int) and not isinstance(value, bool) and value >= 1
    ]
    return stages or None


def _natural_element_id(source: str, prefix: str, row: Mapping[str, Any], item_id: str) -> str:
    for field in _NATURAL_KEY_FIELDS.get(source, ()):
        value = _non_empty_string(row.get(field))
        if value is not None:
            return _element_id(prefix.removeprefix("P-").lower(), value, item_id)
    return item_id


def _subject(row: Mapping[str, Any], source: str) -> str:
    for field in _SUBJECT_FIELDS[source]:
        value = _non_empty_string(row.get(field))
        if value is not None:
            return value
    raise PlanItemContractError(f"{source} row has no non-empty subject candidate")


def _ticket_id(row: Mapping[str, Any]) -> str:
    return _non_empty_string(row.get("ticketId")) or ""


def _row_payload(row: Mapping[str, Any], **coordinates: int) -> dict[str, Any]:
    payload = deepcopy(dict(row))
    for field, value in coordinates.items():
        existing = payload.get(field)
        if field in payload and existing != value:
            raise PlanItemContractError(
                f"row {field} coordinate {existing!r} conflicts with {value!r}"
            )
        payload[field] = value
    return payload


def _add_item(items: list[dict[str, Any]], item: dict[str, Any]) -> None:
    if any(existing["id"] == item["id"] for existing in items):
        raise PlanItemContractError(f"duplicate plan item ID: {item['id']}")
    element_id = item.get("elementId")
    if element_id and any(existing.get("elementId") == element_id for existing in items):
        # 두 요소가 같은 자연키를 쓰면 이월이 엉뚱한 항목에 붙는다. 위치 id 가
        # 갈라져 있어도 대조는 elementId 로 하므로 여기서 막는다.
        raise PlanItemContractError(f"duplicate plan element ID: {element_id}")
    items.append(item)


def _rows(planning: Mapping[str, Any], source: str) -> list[Mapping[str, Any]]:
    value = planning.get(source, [])
    if value is None:
        return []
    if not isinstance(value, list):
        raise PlanItemContractError(f"{source} must be an array")
    rows: list[Mapping[str, Any]] = []
    for row in value:
        if not isinstance(row, Mapping):
            raise PlanItemContractError(f"{source} row must be an object")
        rows.append(row)
    return rows


def _coordinate(value: object, field: str) -> int:
    if not isinstance(value, int) or isinstance(value, bool) or value < 1:
        raise PlanItemContractError(f"{field} must be a positive integer")
    return value


def _extract_standard_items(planning: Mapping[str, Any]) -> list[dict[str, Any]]:
    items: list[dict[str, Any]] = []
    sources = (
        PLAN_ITEM_SOURCES[1:]
        if planning.get("planningContract") == _SELECTED_DIRECTION_CONTRACT
        else PLAN_ITEM_SOURCES
    )
    if planning.get("planningContract") == _SELECTED_DIRECTION_CONTRACT:
        realization = planning.get("directionRealization")
        if not isinstance(realization, Mapping):
            raise PlanItemContractError("directionRealization must be an object")
        _add_item(
            items,
            {
                "id": "P-Dir-1",
                "elementId": "dir:selected-direction",
                "stageScope": None,
                "block": "execution",
                "subject": _non_empty_string(realization.get("goal"))
                or "Selected direction realization",
                "sourceSection": "Selected Direction",
                "ticketId": "",
                "payload": deepcopy(dict(realization)),
            },
        )
    for source, prefix, section in sources:
        if source == "stepwiseExecution":
            _extract_step_items(planning, items, prefix, section)
            continue
        for index, row in enumerate(_rows(planning, source), start=1):
            item_id = f"{prefix}-{index}"
            _add_item(
                items,
                {
                    "id": item_id,
                    "elementId": _natural_element_id(source, prefix, row, item_id),
                    "stageScope": _stage_scope(row),
                    "block": _item_block(source),
                    "subject": _subject(row, source),
                    "sourceSection": section,
                    "ticketId": _ticket_id(row),
                    "payload": _row_payload(row),
                },
            )
    return items


def _extract_step_items(
    planning: Mapping[str, Any],
    items: list[dict[str, Any]],
    prefix: str,
    section: str,
) -> None:
    if "stages" not in planning:
        for index, row in enumerate(_rows(planning, "stepwiseExecution"), start=1):
            _add_item(
                items,
                {
                    # 평면 배열에는 좌표가 없어 위치 id 로 떨어진다.
                    "id": f"{prefix}-{index}",
                    "elementId": f"{prefix}-{index}",
                    "stageScope": _stage_scope(row),
                    "block": "execution",                    "subject": _subject(row, "stepwiseExecution"),
                    "sourceSection": section,
                    "ticketId": _ticket_id(row),
                    "payload": _row_payload(row),
                },
            )
        return
    stages = planning["stages"]
    if not isinstance(stages, list):
        raise PlanItemContractError("stages must be an array")
    for stage in stages:
        if not isinstance(stage, Mapping):
            raise PlanItemContractError("stages row must be an object")
        stage_number = _coordinate(stage.get("stage"), "stage")
        for row in _rows(stage, "stepwiseExecution"):
            step_number = _coordinate(row.get("step"), "step")
            _add_item(
                items,
                {
                    "id": f"{prefix}-{stage_number}.{step_number}",
                    "elementId": f"step:{stage_number}.{step_number}",
                    "stageScope": _stage_scope(None, stage_number),
                    "block": "execution",
                    "subject": _subject(row, "stepwiseExecution"),
                    "sourceSection": section,
                    "ticketId": _ticket_id(row),
                    "payload": _row_payload(
                        row,
                        stage=stage_number,
                        step=step_number,
                    ),
                },
            )


def _extract_prep_items(
    planning: Mapping[str, Any], items: list[dict[str, Any]]
) -> None:
    if not planning.get("stages"):
        return
    detection_input = planning
    if planning.get("planningContract") == _SELECTED_DIRECTION_CONTRACT:
        realization = planning.get("directionRealization")
        if not isinstance(realization, Mapping):
            raise PlanItemContractError("directionRealization must be an object")
        detection_input = dict(planning)
        detection_input["optionCandidates"] = [
            {
                "name": "Selected Direction",
                "fileStructure": realization.get("fileStructure") or [],
            }
        ]
        detection_input["recommendedOption"] = {"name": "Selected Direction"}
    try:
        triggers = detect_design_surfaces(detection_input)
    except (DesignSurfaceError, TypeError, AttributeError) as exc:
        raise PlanItemContractError(f"design-surface detection failed: {exc}") from exc
    for trigger in triggers:
        item_id = expected_prep_plan_item_id(trigger.stage, trigger.kind)
        _add_item(
            items,
            {
                "id": item_id,
                "elementId": f"prep:{trigger.stage}:{_element_key(trigger.kind)}",
                "stageScope": _stage_scope(None, trigger.stage),
                "block": "execution",
                "subject": f"Stage {trigger.stage} {trigger.kind} design preparation",
                "sourceSection": "5.5.10",
                "ticketId": "",
                "payload": {
                    "stage": trigger.stage,
                    "kind": trigger.kind,
                    "evidence": [
                        {
                            "step": evidence.step,
                            "field": evidence.field,
                            "match": evidence.match,
                        }
                        for evidence in trigger.evidence
                    ],
                },
            },
        )


def _extract_variation_point_items(
    planning: Mapping[str, Any], items: list[dict[str, Any]]
) -> None:
    analysis = planning.get("variationPointAnalysis")
    if not isinstance(analysis, Mapping):
        raise PlanItemContractError("variationPointAnalysis must be an object")
    points = analysis.get("points") or []
    if not analysis.get("hasMultipleImplementations") or not points:
        _add_item(
            items,
            {
                "id": "P-Var-0",
                "elementId": "var:none",
                "stageScope": None,
                "block": "execution",                "subject": "No variation point declared",
                "sourceSection": "5.5.11",
                "ticketId": "",
                "payload": _row_payload(dict(analysis)),
            },
        )
        return
    for index, point in enumerate(points, start=1):
        if not isinstance(point, Mapping):
            raise PlanItemContractError(
                "variationPointAnalysis.points row must be an object"
            )
        _add_item(
            items,
            {
                "id": f"P-Var-{index}",
                "elementId": _element_id(
                    "var",
                    _non_empty_string(point.get("behavior")) or "",
                    f"P-Var-{index}",
                ),
                "stageScope": None,
                "block": "execution",                "subject": _non_empty_string(point.get("behavior"))
                or f"variation point {index}",
                "sourceSection": "5.5.11",
                "ticketId": "",
                "payload": _row_payload(dict(point)),
            },
        )


def extract_plan_items(implementation_planning: Mapping[str, Any]) -> list[dict[str, Any]]:
    """Return deterministic, lossless P-* items in contract order."""
    if not isinstance(implementation_planning, Mapping):
        raise PlanItemContractError("implementationPlanning must be an object")
    if (
        implementation_planning.get("planningContract")
        == _SELECTED_DIRECTION_CONTRACT
        and implementation_planning.get("outcome") == "direction-invalidated"
    ):
        return []
    items = _extract_standard_items(implementation_planning)
    _extract_prep_items(implementation_planning, items)
    _extract_variation_point_items(implementation_planning, items)
    return items


def expected_plan_item_ids(implementation_planning: Mapping[str, Any]) -> list[str]:
    """Return the exact ordered IDs produced by extract_plan_items()."""
    return [item["id"] for item in extract_plan_items(implementation_planning)]


_STARTABLE_STATUSES = frozenset({"ready", "active"})

# 계획 전체를 판정하는 항목. 스테이지 하나를 고쳐도 요청하지 않은 작업이
# 들어왔는지 다시 봐야 한다. P-Val / P-Req / P-Rb 는 여기 넣지 않는다 —
# stageRefs 가 비어 있어도 이웃 수정의 일소 대상이 되면 C 행이 매 라운드 늘어난다.
_PLAN_WIDE_ID_PREFIXES = ("P-Opt-", "P-Var-", "P-Dir-", "P-Dep-")


def _is_plan_wide(item_id: str) -> bool:
    return item_id.startswith(_PLAN_WIDE_ID_PREFIXES)


def content_hash(item: Mapping[str, Any]) -> str:
    """이 항목 본문의 지문. 문구가 같으면 라운드가 달라도 같은 판정이다.

    라운드 번호만 보면 고치지 않은 항목까지 재검증해야 한다. subject·payload·
    block·stageScope 가 같으면 그 항목이 가리키는 텍스트는 재작성 전과 같다.
    """
    payload = {
        "block": item.get("block"),
        "payload": item.get("payload"),
        "stageScope": item.get("stageScope"),
        "subject": item.get("subject"),
    }
    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()


def stage_scope_bucket(
    item: Mapping[str, Any],
    ledger: Mapping[str, str] | None,
) -> str:
    """게이트·디스패치가 공유하는 범위. ``in-scope`` / ``observed`` / ``deferred``.

    원장이 없거나 계획 전체 항목이면 in-scope 이다. 검증기
    ``_stage_scope_bucket`` 과 같은 판정이어야 한다 — 디스패치와 게이트가
    다른 통을 쓰면 워커가 본 항목과 승인을 막는 항목이 갈라진다.

    단계가 없는 ``P-Val`` / ``P-Req`` / ``P-Rb`` 는 원장에 ``done`` 이 생긴
    뒤부터는 다음 착수를 막지 않는다. 생략된 ``stageRefs`` 를 모든 스테이지로
    읽으면 이미 끝난 계획의 재실행마다 체크리스트 전부가 다시 채점된다.
    """
    if not isinstance(ledger, Mapping) or not ledger:
        return "in-scope"
    scope = item.get("stageScope")
    stages = [
        value for value in scope
        if isinstance(value, int) and not isinstance(value, bool)
    ] if isinstance(scope, list) else []
    if not stages:
        item_id = str(item.get("id") or "")
        if _is_plan_wide(item_id):
            return "in-scope"
        if "done" in {str(value) for value in ledger.values()}:
            return "deferred"
        return "in-scope"
    statuses = {str(ledger.get(str(stage)) or "") for stage in stages}
    if statuses & _STARTABLE_STATUSES:
        return "in-scope"
    return "observed" if "done" in statuses else "deferred"


def _depends_on(value: object) -> tuple[int, ...]:
    if value in (None, "", "(none)", []):
        return ()
    if isinstance(value, list):
        return tuple(
            entry for entry in value
            if isinstance(entry, int) and not isinstance(entry, bool) and entry >= 1
        )
    if isinstance(value, str):
        stripped = value.strip()
        if stripped in {"", "(none)"}:
            return ()
        numbers: list[int] = []
        for token in stripped.split(","):
            piece = token.strip()
            if piece.isdigit() and int(piece) >= 1:
                numbers.append(int(piece))
        return tuple(numbers)
    return ()


def _planning_stage_rows(planning: Mapping[str, Any]) -> list[tuple[int, tuple[int, ...]]]:
    stage_map = planning.get("stageMap")
    if isinstance(stage_map, list) and stage_map:
        rows: list[tuple[int, tuple[int, ...]]] = []
        for row in stage_map:
            if not isinstance(row, Mapping):
                continue
            number = row.get("stage")
            if not isinstance(number, int) or isinstance(number, bool) or number < 1:
                continue
            rows.append((number, _depends_on(row.get("dependsOn"))))
        return rows
    stages = planning.get("stages")
    if not isinstance(stages, list):
        return []
    rows = []
    for row in stages:
        if not isinstance(row, Mapping):
            continue
        number = row.get("stage")
        if not isinstance(number, int) or isinstance(number, bool) or number < 1:
            continue
        rows.append((number, _depends_on(row.get("dependsOn"))))
    return rows


def planning_stage_ledger(
    planning: Mapping[str, Any],
    disk_status: Mapping[str, str] | None = None,
) -> dict[str, str]:
    """지금 계획의 스테이지를 ``done`` / ``active`` / ``ready`` / ``blocked`` 로.

    디스크 원장은 이미 구현된 것만 안다. 첫 계획 run 에는 원장이 없어서
    게이트가 전 항목을 in-scope 로 읽었다. 이 함수는 현재 계획의 depends-on 으로
    그 빈칸을 채운다. 디스크에 ``done`` / ``active`` 가 있으면 그쪽이 이긴다.
    """
    recorded = {
        key: value
        for key, value in (disk_status or {}).items()
        if value in {"done", "active", "ready", "blocked"}
    }
    done = {key for key, value in recorded.items() if value == "done"}
    ledger: dict[str, str] = {}
    for number, depends in _planning_stage_rows(planning):
        key = str(number)
        status = recorded.get(key)
        if status in {"done", "active"}:
            ledger[key] = status
            continue
        ready = all(str(dep) in done for dep in depends)
        ledger[key] = "ready" if ready else "blocked"
    return ledger


def dispatch_item_ids(
    items: Sequence[Mapping[str, Any]],
    ledger: Mapping[str, str] | None,
) -> list[str]:
    """워커에게 보낼 항목. 게이트가 묻는 범위와 같다."""
    ids: list[str] = []
    for item in items:
        item_id = item.get("id")
        if not isinstance(item_id, str) or not item_id:
            continue
        if stage_scope_bucket(item, ledger) == "in-scope":
            ids.append(item_id)
    return ids


def reverify_item_ids(
    items: Sequence[Mapping[str, Any]],
    previous_hashes: Mapping[str, str],
    ledger: Mapping[str, str] | None,
) -> list[str]:
    """self-fix 뒤 다시 볼 항목. 해시가 바뀐 것과 같은 스테이지, 계획 전체.

    이전이 없으면 첫 라운드라 디스패치 큐 전부다. 해시가 같은 다른 스테이지
    항목은 보내지 않는다 — 그게 96개 일소 배치를 만들던 규칙이다.

    단계가 없는 항목 전부가 계획 전체는 아니다. `P-Opt` / `P-Var` / `P-Dir` /
    `P-Dep` 만 이웃 수정에 다시 묶인다. `P-Val` / `P-Req` / `P-Rb` 는 자기
    해시가 바뀔 때만 다시 본다.
    """
    dispatched = dispatch_item_ids(items, ledger)
    by_id = {
        item["id"]: item
        for item in items
        if isinstance(item.get("id"), str)
    }
    if not previous_hashes:
        return dispatched
    changed = [
        item_id for item_id, item in by_id.items()
        if previous_hashes.get(item_id) != content_hash(item)
    ]
    changed_stages: set[int] = set()
    for item_id in changed:
        scope = by_id[item_id].get("stageScope")
        if isinstance(scope, list):
            changed_stages.update(
                value for value in scope
                if isinstance(value, int) and not isinstance(value, bool)
            )
    queue: list[str] = []
    queued: set[str] = set()
    for item_id in dispatched:
        item = by_id[item_id]
        scope = item.get("stageScope")
        stages = [
            value for value in scope
            if isinstance(value, int) and not isinstance(value, bool)
        ] if isinstance(scope, list) else []
        if not stages:
            if item_id in changed or (changed and _is_plan_wide(item_id)):
                queue.append(item_id)
                queued.add(item_id)
            continue
        if item_id in changed or changed_stages.intersection(stages):
            queue.append(item_id)
            queued.add(item_id)
    # done 이후 디스패치에서 빠진 체크리스트라도 본문이 바뀌면 다시 본다.
    for item_id in changed:
        if item_id in queued or item_id not in by_id:
            continue
        scope = by_id[item_id].get("stageScope")
        stages = [
            value for value in scope
            if isinstance(value, int) and not isinstance(value, bool)
        ] if isinstance(scope, list) else []
        if not stages:
            queue.append(item_id)
    return queue


def advisory_plan_body_gating(
    planning: Mapping[str, Any],
    extracted: Sequence[Mapping[str, Any]] | None = None,
) -> bool:
    """검출 표면 0 + 스테이지 1이면 본문 검증은 자문만 한다.

    준비 시점에는 계획이 없어 ``gating`` 을 false 로 둘 수 없다. 작성
    초안과 탐지기 스냅샷이 생긴 뒤에만 이 판정을 쓴다. 다단계이거나
    PREP 항목이 있으면 지금 게이트 계약 그대로다.
    """
    if len(_planning_stage_rows(planning)) != 1:
        return False
    if extracted is not None and any(
        str(item.get("id") or "").startswith("P-Prep-") for item in extracted
    ):
        return False
    preparation = planning.get("designPreparation")
    if not isinstance(preparation, Mapping):
        preparation = build_design_snapshot(planning).get("designPreparation")
    if not isinstance(preparation, Mapping):
        return False
    items = preparation.get("items")
    if preparation.get("mode") != "no-design-inputs":
        return False
    return not (isinstance(items, list) and items)


CRITIC_WORKER_ID = "critic-worker"


def is_critic_worker(worker: str) -> bool:
    """본문 동수를 가르는 critic 표인지."""
    name = str(worker or "").strip().lower()
    return name == CRITIC_WORKER_ID or name.endswith("-critic-worker")


def tie_vote_item_ids(
    items: Sequence[Mapping[str, Any]],
    ledger: Mapping[str, str] | None,
    tied_ids: Sequence[str],
) -> list[str]:
    """needs-reverify 동수 항목만 critic 에 보낸다.

    첫 라운드 큐나 self-fix 재검증 큐와 섞지 않는다. 동수가 없으면 빈 목록이고
    critic 은 이 배치를 띄우지 않는다.
    """
    allowed = set(dispatch_item_ids(items, ledger))
    queue: list[str] = []
    seen: set[str] = set()
    for item_id in tied_ids:
        if item_id in allowed and item_id not in seen:
            queue.append(item_id)
            seen.add(item_id)
    return queue


_ERROR_VERDICTS = frozenset({
    "verification-error", "UNVERIFIABLE", "VERIFICATION-ERROR",
})
_COMMAND_KEYS = ("command", "commandOrObservation")

ENVIRONMENT_CORRECTION_PREAMBLE = (
    "Planning-time environment gap: this worktree has no build/test "
    "dependencies installed. A command that is declared in package.json / "
    "Makefile / the task runner but fails here on a missing module, binary, "
    "exit 127, or command-not-found is UNVERIFIABLE, not DISAGREE(b). A "
    "referenced path that does not exist is still DISAGREE(b). Whether a "
    "path exists, whether a command is declared, and whether the plan is "
    "internally consistent are all checkable without installing dependencies. "
    "A blanket \"capability constraints prevent workspace resolution\" is not "
    "a valid answer to any of them.\n"
)

CRITIC_TIE_PREAMBLE = (
    "You are the critic tie-break. Only the items below are in dispute. "
    "Each item already has one AGREE and one DISAGREE from the two plan-body "
    "verifiers. Decide the item: AGREE or DISAGREE(<kind>). Your verdict "
    "settles the split. Do not re-open items that are not listed.\n"
)


def item_cites_build_command(item: Mapping[str, Any]) -> bool:
    """이 항목이 계획 워크트리에서 실행 불가한 빌드/테스트 명령을 인용하는가."""
    payload = item.get("payload") if isinstance(item.get("payload"), Mapping) else item
    if not isinstance(payload, Mapping):
        return False
    for key in _COMMAND_KEYS:
        value = payload.get(key)
        if isinstance(value, str) and command_invokes_build_tool(value):
            return True
    return False


def _error_verdict(token: str) -> bool:
    stripped = token.strip()
    return stripped in _ERROR_VERDICTS or stripped.upper() == "UNVERIFIABLE"


def _item_votes(item: Mapping[str, Any]) -> list[tuple[str, str]]:
    rows: list[tuple[str, str]] = []
    for row in item.get("verdicts") or []:
        if not isinstance(row, Mapping):
            continue
        worker = str(row.get("worker") or "").strip()
        token = str(row.get("verdict") or "").strip()
        if worker and token:
            rows.append((worker, token))
    return rows


def _worker_vote_map(
    items: Sequence[Mapping[str, Any]],
) -> dict[str, list[tuple[str, str]]]:
    votes: dict[str, list[tuple[str, str]]] = {}
    for item in items:
        item_id = str(item.get("id") or "")
        if not item_id:
            continue
        for worker, token in _item_votes(item):
            votes.setdefault(worker, []).append((item_id, token))
    return votes


def _with_payload(
    item: Mapping[str, Any],
    payloads: Mapping[str, Mapping[str, Any]] | None,
) -> Mapping[str, Any]:
    extra = payloads.get(str(item.get("id") or "")) if payloads else None
    if extra is None or "payload" not in extra:
        return item
    merged = dict(item)
    merged["payload"] = extra["payload"]
    return merged


def _worker_is_blanket(
    votes: Sequence[tuple[str, str]],
    items_by_id: Mapping[str, Mapping[str, Any]],
    payloads: Mapping[str, Mapping[str, Any]] | None,
) -> bool:
    """전 표가 오류이고, 빌드 명령이 아닌 확인 가능 항목이 하나라도 있다."""
    if not votes or any(not _error_verdict(token) for _item_id, token in votes):
        return False
    return any(
        not item_cites_build_command(_with_payload(items_by_id[item_id], payloads))
        for item_id, _token in votes
        if item_id in items_by_id
    )


def blanket_unverifiable_workers(
    items: Sequence[Mapping[str, Any]],
    payloads: Mapping[str, Mapping[str, Any]] | None = None,
) -> list[str]:
    """전 배정 표가 UNVERIFIABLE/verification-error 인 워커. 경로 확인은 예외가 아니다."""
    by_id = {
        str(item["id"]): item
        for item in items
        if isinstance(item.get("id"), str) and item["id"]
    }
    return sorted(
        worker
        for worker, votes in _worker_vote_map(items).items()
        if _worker_is_blanket(votes, by_id, payloads)
    )


def _is_tie(item: Mapping[str, Any]) -> bool:
    tokens = [
        token
        for worker, token in _item_votes(item)
        if not _error_verdict(token) and not is_critic_worker(worker)
    ]
    if len(tokens) < 2:
        return False
    if any(is_critic_worker(worker) and not _error_verdict(token)
           for worker, token in _item_votes(item)):
        return False
    disagree = sum(1 for token in tokens if token.upper().startswith("DISAGREE"))
    agree = sum(1 for token in tokens if token.upper() in {"AGREE", "SUPPLEMENT"})
    return disagree == agree and disagree > 0


@dataclass(frozen=True)
class NextDispatch:
    """다음 워커 배치. ``none`` 은 배치를 열지 않는다."""

    kind: str
    workers: tuple[str, ...]
    item_ids: tuple[str, ...]
    reason: str

    def as_dict(self) -> dict[str, Any]:
        return {
            "kind": self.kind,
            "workers": list(self.workers),
            "itemIds": list(self.item_ids),
            "reason": self.reason,
        }


def next_dispatch(
    items: Sequence[Mapping[str, Any]],
    payloads: Mapping[str, Mapping[str, Any]] | None = None,
) -> NextDispatch:
    """환경 전용 UNVERIFIABLE 은 전체 라운드를 만들지 않는다. 일괄 오류만 그 워커."""
    assigned = tuple(
        str(item["id"]) for item in items
        if isinstance(item.get("id"), str) and item["id"]
    )
    blanket = tuple(blanket_unverifiable_workers(items, payloads))
    if blanket:
        return NextDispatch(
            kind="worker-correction",
            workers=blanket,
            item_ids=assigned,
            reason=(
                "blanket UNVERIFIABLE/verification-error; "
                "correct those workers only"
            ),
        )
    ties = tuple(
        str(item["id"]) for item in items
        if isinstance(item.get("id"), str) and _is_tie(item)
    )
    if ties:
        return NextDispatch(
            kind="critic-tie",
            workers=(CRITIC_WORKER_ID,),
            item_ids=ties,
            reason="unsettled analyser tie; critic settles",
        )
    return NextDispatch(
        kind="none", workers=(), item_ids=(),
        reason="no worker batch",
    )


def correction_prompt_text(queue_markdown: str) -> str:
    """시정 프롬프트. 환경 예외 단락이 큐보다 앞이다."""
    return f"{ENVIRONMENT_CORRECTION_PREAMBLE.rstrip()}\n\n{queue_markdown.lstrip()}"


def critic_tie_prompt_text(queue_markdown: str) -> str:
    """동수 critic 프롬프트. 가르는 지시가 큐보다 앞이다."""
    return f"{CRITIC_TIE_PREAMBLE.rstrip()}\n\n{queue_markdown.lstrip()}"