"""CLI adapter for deterministic implementation-planning item extraction.

`extract` / `validate` own the `P-*` queue; `collect-verdicts` / `apply-verdicts`
own the round's votes on that queue. Both halves exist so the round's fidelity
does not depend on a parser the lead re-writes each time.
"""
from __future__ import annotations

import argparse
import copy
import importlib.util
import json
import re
from collections.abc import Mapping, Sequence
from datetime import datetime, timezone
from pathlib import Path
import sys
from typing import Any

from .convergence import (
    ConvergenceContractError,
    canonical_run_state_artifact,
    validated_run_authority,
)
from .convergence_store import write_json_atomic
from .incremental_carry import sync_prepared_dispatch_queue
from .json_boundary import JsonBoundaryError, load_owned_object
from .report_finalize import task_manifest_path
from .plan_derivations import extract_tokens, find_derivations
from .plan_items import (
    NextDispatch,
    PlanItemContractError,
    advisory_plan_body_gating,
    requires_plan_repair,
    content_hash,
    correction_prompt_text,
    critic_is_rostered,
    is_critic_worker,
    critic_tie_prompt_text,
    dispatch_item_ids,
    extract_plan_items,
    lead_decision_basis,
    next_dispatch,
    planning_stage_ledger,
    reverify_item_ids,
    reverify_prompt_text,
    self_fix_rounds,
    tie_vote_item_ids,
    voting_analyser_keys,
    with_rendered_by,
    with_response_format,
)
from .paths import RunRef
from .stage_ledger import build_stage_ledger
from .claim_reproduction import NOT_RUNNABLE, reproduce
from .final_report_schema import load_schema_version
from .report_narrative import parse_narrative
from .report_assembly import validate_plan_draft
from .error_log_write import record_runtime_failure
from .user_response import parse_user_response_entries
from .verdict_blocks import (
    PLAN_ITEM_VERDICTS,
    VerdictBlock,
    VerdictBlockError,
    parse_verdict_blocks,
)
from .fixed_text import line, scalar


def _load_json_object(path: Path) -> dict[str, Any]:
    try:
        return load_owned_object(path, artifact="plan-items artifact")
    except JsonBoundaryError as exc:
        raise PlanItemContractError(str(exc)) from exc


def _planning(data: Mapping[str, Any]) -> Mapping[str, Any]:
    planning = data.get("implementationPlanning")
    if not isinstance(planning, Mapping):
        raise PlanItemContractError("implementationPlanning must be an object")
    return planning


def _envelope(data: Mapping[str, Any]) -> dict[str, Any]:
    return {
        "schemaVersion": "1.0",
        "taskType": "implementation-planning",
        "items": extract_plan_items(_planning(data)),
    }


def _merged_ledger(
    planning: Mapping[str, Any],
    run_manifest: Path | None,
) -> dict[str, str]:
    return planning_stage_ledger(planning, _stage_ledger_snapshot(run_manifest))


def _previous_hashes(state_path: Path | None) -> dict[str, str]:
    if state_path is None or not state_path.is_file():
        return {}
    rows = _state_plan_body_items(_load_json_object(state_path), state_path)
    return {
        str(row["id"]): str(row["verifiedContentHash"])
        for row in rows
        if isinstance(row, Mapping)
        and isinstance(row.get("id"), str)
        and isinstance(row.get("verifiedContentHash"), str)
    }


def _queue_for(
    items: list[dict[str, Any]],
    planning: Mapping[str, Any],
    run_manifest: Path | None,
    previous_hashes: Mapping[str, str] | None = None,
    tied_ids: Sequence[str] | None = None,
) -> list[str]:
    ledger = _merged_ledger(planning, run_manifest)
    if tied_ids is not None:
        return tie_vote_item_ids(items, ledger, tied_ids)
    if previous_hashes:
        return reverify_item_ids(items, previous_hashes, ledger)
    return dispatch_item_ids(items, ledger)


def _queue_kwargs(args: argparse.Namespace) -> dict[str, Any]:
    if getattr(args, "tie_vote", False):
        state = getattr(args, "state", None)
        if state is None:
            raise PlanItemContractError("--tie-vote requires --state")
        return {"tied_ids": _tied_ids(state)}
    return {"previous_hashes": _previous_hashes(getattr(args, "state", None))}


def _tied_ids(state_path: Path) -> list[str]:
    if not state_path.is_file():
        raise PlanItemContractError("--tie-vote requires --state with recorded verdicts")
    items = _state_plan_body_items(_load_json_object(state_path), state_path)
    gate = _gate_module()
    return [
        str(item["id"])
        for item in items
        if isinstance(item, Mapping)
        and isinstance(item.get("id"), str)
        and gate._is_unsettled_tie(dict(item))
    ]


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="okstra plan-items",
        description="Extract and validate deterministic plan-body items.",
    )
    commands = parser.add_subparsers(dest="command", required=True)
    extract = commands.add_parser("extract")
    _add_plan_source(extract)
    extract.add_argument("--output", type=Path, required=True)
    validate = commands.add_parser("validate")
    _add_plan_source(validate)
    validate.add_argument("--items", type=Path, required=True)
    prepare = commands.add_parser("prepare")
    _add_plan_source(prepare)
    prepare.add_argument("--run-manifest", type=Path, required=True)
    prepare.add_argument(
        "--state", type=Path,
        help="previous plan-body state; when it carries verifiedContentHash "
             "rows the dispatch queue shrinks to the self-fix reverify set, and "
             "each queued item's recorded votes and selfFixNote are carried "
             "into the prompt as its `**Prior round dissent**` block",
    )
    prepare.add_argument(
        "--tie-vote", action="store_true",
        help="dispatch queue is the needs-reverify ties in --state for critic-worker",
    )
    prompt = commands.add_parser("prompt")
    prompt.add_argument("--run-manifest", type=Path, required=True)
    validate_prepared = commands.add_parser("validate-prepared")
    _add_plan_source(validate_prepared)
    validate_prepared.add_argument("--run-manifest", type=Path, required=True)
    validate_prepared.add_argument(
        "--state", type=Path,
        help="same previous state prepare used to shrink the dispatch queue "
             "and to build the prior-dissent carry; both are re-derived here",
    )
    validate_prepared.add_argument(
        "--tie-vote", action="store_true",
        help="same --tie-vote prepare used to shrink the dispatch queue",
    )
    collect = commands.add_parser(
        "collect-verdicts",
        help="read this round's worker responses into a verdicts envelope",
    )
    collect.add_argument("--result", action="append", default=[], required=True,
                         metavar="<worker-id>=<path>",
                         help="one worker's plan-verify result file (repeatable)")
    collect.add_argument(
        "--items", type=Path, required=True,
        metavar="<plan-items artifact>",
        help="the plan-items artifact this round dispatched (not a list of "
             "ids). Its `dispatchQueue` is the id set the results must answer, "
             "so a tie round takes the `--tie-vote` artifact, not the full "
             "round's one",
    )
    collect.add_argument("--output", type=Path, required=True)
    derivations = commands.add_parser(
        "derivations",
        help="list plan statements an answered clarification may have falsified",
    )
    derivations.add_argument("--data", type=Path, required=True)
    derivations.add_argument("--response", type=Path, required=True,
                             help="the user-responses sidecar for this run")
    derivations.add_argument("--clarification", default=None,
                             help="only this C-id (default: every answered one)")
    seed = commands.add_parser(
        "seed",
        help="create the planBodyVerification.planItems[] rows a round lands in",
    )
    _add_plan_source(seed)
    seed.add_argument("--state", type=Path)
    seed.add_argument(
        "--run-manifest", type=Path,
        help="record the stage ledger so the gate can scope itself to the "
             "stage about to start; without it the gate does not narrow",
    )
    seed.add_argument(
        "--prior-state", type=Path,
        help="the previous run's plan-body-verification-<task-type>-<seq>.json. "
             "A newly seeded item whose contentHash still equals that run's "
             "verifiedContentHash inherits its verdicts and is tagged "
             "carriedForwardFromSeq, so round 1 does not re-judge text nobody "
             "changed. Requires --state, refuses a state file belonging to "
             "another task, and never carries on a matching id alone",
    )
    apply_verdicts = commands.add_parser(
        "apply-verdicts",
        help="overwrite planBodyVerification.planItems[].verdicts in owned state",
    )
    target = apply_verdicts.add_mutually_exclusive_group(required=True)
    target.add_argument("--data", type=Path)
    target.add_argument("--state", type=Path)
    incoming = apply_verdicts.add_mutually_exclusive_group(required=True)
    incoming.add_argument(
        "--verdicts", type=Path,
        help="historical automation envelope; model-facing calls use --result",
    )
    incoming.add_argument(
        "--result", action="append", default=[],
        metavar="<worker-id>=<path>",
        help="one worker's plan-verify Markdown result (repeatable)",
    )
    apply_verdicts.add_argument(
        "--items", type=Path,
        metavar="<plan-items artifact>",
        help="the plan-items artifact this round dispatched, when it dispatched "
             "part of the queue (the `--tie-vote` artifact). Its dispatchQueue "
             "is what each --result must answer; without it a result is checked "
             "against the whole persisted queue",
    )
    apply_verdicts.add_argument(
        "--run-manifest", type=Path,
        help="resolve the project a `fact` claim's probe runs against; without "
             "it every such claim records `not-runnable` and takes the quorum "
             "route instead of blocking on one vote",
    )
    apply_verdicts.add_argument(
        "--round", type=int, required=True, dest="round_number",
        help="the verification round these verdicts were cast in; stamped on "
             "every row so a later self-fix can be told from a current judgement",
    )
    apply_verdicts.add_argument(
        "--append", action="store_true",
        help="add critic corrections while preserving analyser votes; a later "
             "critic verdict updates that critic's current row after its earlier round is complete; "
             "without it every recorded verdict row is replaced, so a round "
             "that was never closed with complete-round is refused first",
    )
    apply_verdicts.add_argument(
        "--discard-open-rounds", action="store_true",
        help="replace even the rows of a round that complete-round never "
             "closed — the recovery path when those rows are being re-applied "
             "from their result files in order; the discarded rows are listed, "
             "and dispatchQueue is restored to the items those result files "
             "answer, since the persisted queue belongs to the latest round",
    )
    complete = commands.add_parser(
        "complete-round",
        help="derive and atomically record one verified plan-body round",
    )
    complete.add_argument("--state", type=Path, required=True)
    complete.add_argument("--run-manifest", type=Path, required=True)
    complete.add_argument("--round", type=int, required=True, dest="round_number")
    complete.add_argument(
        "--items", type=Path,
        help="restore this round's dispatched queue from its prepared items artifact; "
             "defaults to this run's canonical prepared queue when it covers the round's "
             "recorded votes; earlier verdicts outside that queue remain unchanged",
    )
    complete.add_argument("--self-fix-note", action="append", default=[], metavar="<item-id>=<markdown-file>")
    complete.add_argument(
        "--self-fix-group", action="append", default=[],
        metavar="<cause-file>=<item-id>[,<item-id>...]",
        help="this round's cause groups; requires --self-fix-stop-reason",
    )
    complete.add_argument(
        "--self-fix-stop-reason",
        choices=("all-resolved", "no-progress", "max-rounds-reached"),
        help="why the self-fix loop stops. Required with --self-fix-group, and "
             "valid on its own to record a stop the round did not rewrite for; "
             "on its own it leaves selfFixRoundsApplied and selfFixGroups alone",
    )
    _add_dispatch_commands(commands)
    return parser


def _add_dispatch_commands(commands: Any) -> None:
    resolve = commands.add_parser(
        "resolve-dissent", help="record an evidence-based lead decision after the single self-fix",
    )
    resolve.add_argument("--state", type=Path, required=True)
    resolve.add_argument("--item", required=True)
    resolve.add_argument("--decision-file", type=Path, required=True)
    nxt = commands.add_parser(
        "next-dispatch",
        help="decide whether this round opens a worker batch",
    )
    nxt.add_argument("--state", type=Path, required=True)
    nxt.add_argument("--run-manifest", type=Path)
    correction = commands.add_parser(
        "correction-prompt",
        help="environment-exception preamble then the assigned queue",
    )
    correction.add_argument("--state", type=Path, required=True)
    correction.add_argument("--run-manifest", type=Path, required=True)
    correction.add_argument("--worker", required=True)


def _add_plan_source(parser: argparse.ArgumentParser) -> None:
    source = parser.add_mutually_exclusive_group(required=True)
    source.add_argument("--data", type=Path)
    source.add_argument("--narrative", type=Path)


def _plan_source(args: argparse.Namespace) -> dict[str, Any]:
    if args.data is not None:
        return _load_json_object(args.data)
    try:
        markdown = args.narrative.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        raise PlanItemContractError(
            f"cannot read report narrative {args.narrative}: {exc}"
        ) from exc
    try:
        return parse_narrative(markdown, load_schema_version("3.0"))
    except ValueError as exc:
        raise PlanItemContractError(f"invalid report narrative: {exc}") from exc


def _extract(args: argparse.Namespace) -> dict[str, Any]:
    envelope = _envelope(_plan_source(args))
    write_json_atomic(args.output, envelope)
    return {"ok": True, "operation": "extract", "path": str(args.output)}


def _validate(args: argparse.Namespace) -> dict[str, Any]:
    expected = _envelope(_plan_source(args))
    actual = _load_json_object(args.items)
    if actual != expected:
        raise PlanItemContractError("plan items envelope does not match deterministic extraction")
    return {"ok": True, "operation": "validate", "path": str(args.items)}


def _prepared_items_path(
    run_manifest: Path, *, require_regular: bool = False
) -> Path:
    try:
        authority = validated_run_authority(run_manifest)
        state = canonical_run_state_artifact(
            authority,
            manifest_field="planBodyVerificationPath",
            prefix="plan-body-verification",
            label="plan body verification path",
        )
    except ConvergenceContractError as exc:
        raise PlanItemContractError(str(exc)) from exc
    path = state.with_name(
        "plan-items-" + state.name.removeprefix("plan-body-verification-")
    )
    if path.absolute() != path.resolve(strict=False):
        raise PlanItemContractError("plan items path does not match run authority")
    if require_regular and not path.is_file():
        raise PlanItemContractError("plan items path is not a regular file")
    if path.exists() and not path.is_file():
        raise PlanItemContractError("plan items path is not a regular file")
    return path


def _sync_task_manifest_gating(run_manifest: Path, gating: bool) -> None:
    """준비 이후 매니페스트 ``gating`` 을 계획 사실로 맞춘다."""
    try:
        authority = validated_run_authority(run_manifest)
        path = task_manifest_path(authority.project_root, authority.payload)
        payload = load_owned_object(path, artifact="task manifest")
    except (ConvergenceContractError, JsonBoundaryError, OSError, ValueError):
        return
    block = payload.get("convergence")
    if not isinstance(block, dict):
        return
    pbv = block.get("planBodyVerification")
    if not isinstance(pbv, dict):
        return
    pbv["gating"] = gating
    write_json_atomic(path, payload)


def _prepare(args: argparse.Namespace) -> dict[str, Any]:
    output = _prepared_items_path(args.run_manifest)
    source = _plan_source(args)
    authority = validated_run_authority(args.run_manifest)
    if authority.payload.get("reportContractVersion") == "3.0":
        failures = validate_plan_draft(source, authority.project_root, authority.payload)
        if failures:
            raise PlanItemContractError(
                "owner=report-writer: "
                + "; ".join(failures)
                + "; correct the narrative in this run, then retry plan-items prepare"
            )
    envelope = _envelope(source)
    envelope["dispatchQueue"] = _queue_for(
        envelope["items"],
        _planning(source),
        args.run_manifest,
        **_queue_kwargs(args),
    )
    if getattr(args, "tie_vote", False):
        envelope["dispatchKind"] = "critic-tie"
        envelope["tieSplits"] = _tie_splits_from_state(
            getattr(args, "state", None), envelope["dispatchQueue"],
        )
    else:
        envelope.update(_reverify_carry(args, envelope["dispatchQueue"]))
    gating = not advisory_plan_body_gating(_planning(source), envelope["items"])
    _sync_task_manifest_gating(args.run_manifest, gating)
    write_json_atomic(output, envelope)
    return {
        "ok": True,
        "operation": "prepare",
        "path": str(output),
        "gating": gating,
    }


_PAYLOAD_FIELDS = {
    "Dir": ("goal", "coreMechanism", "architectureBoundaries", "expectedChangeAreas",
            "fileStructure", "interfaces", "blastRadius", "testSeams", "assumptions",
            "planningInvariants", "userConstraints"),
    "Opt": ("name", "ticketId", "fileStructure", "interfaces", "blastRadius"),
    "Step": ("stage", "step", "ticketId", "action", "files", "plannedPaths",
             "command", "outcome", "expectedDetail"),
    "Dep": ("id", "ticketId", "kind", "item", "impact", "mitigation"),
    "Val": ("id", "phase", "ticketId", "check", "commandOrObservation", "expectedOutcome",
            "stageRefs"),
    "Rb": ("id", "ticketId", "step", "action", "triggerSignal", "verificationMethod"),
    # 두 계약의 합집합이다. `ImplementationRequirementCoverageRow` 는 id/source/
    # requirement/coveredBy/status 와 사용자 처분 기록(approvalDisposition,
    # decisionRefs)을, `SelectedDirectionRequirementCoverage` 는 originalRequirementId
    # 와 네 개의 *Refs 를 싣는다. 어느 한쪽만 남기면 다른 계약의 계획이 렌더에서
    # 거부되므로 줄이지 말 것.
    "Req": ("id", "source", "requirement", "coveredBy", "originalRequirementId",
            "stageRefs", "stepRefs", "validationRefs", "fileRefs", "ticketId", "status",
            "approvalDisposition", "decisionRefs", "crossProjectDependencyRefs"),
    "Prep": ("stage", "kind", "evidence"),
}
_VAR_ANALYSIS_FIELDS = (
    "hasMultipleImplementations", "noVariationRationale", "points",
)
_VAR_POINT_FIELDS = (
    "behavior", "implementations", "evidence", "extractionDecision",
)
_OBJECT_LIST_FIELDS = {
    "fileStructure": ("id", "ticketId", "action", "path", "summary", "details"),
    "testSeams": ("boundary", "injectedAs", "replacedInTest"),
    "planningInvariants": ("id", "statement", "requirementIds", "evidence"),
    "evidence": ("step", "field", "match"),
}


def _label(key: str) -> str:
    return " ".join(part.capitalize() for part in key.replace("Id", " ID").split())


def _render_literal(label: str, value: str, indent: str = "") -> list[str]:
    """명령·코드의 줄바꿈을 보존하고 본문보다 긴 울타리로 감싼다."""
    fence = "`" * max(
        3, 1 + max((len(part) for part in re.findall(r"`+", value)), default=0)
    )
    body = "".join(indent + "  " + row for row in value.splitlines(keepends=True))
    ending = "" if value.endswith("\n") else "\n"
    return [
        f"{indent}- {label}:\n\n{indent}  {fence}text\n{body}{ending}{indent}  {fence}\n"
    ]


def _render_object_list(key: str, value: object) -> list[str]:
    if not isinstance(value, list):
        raise PlanItemContractError(f"plan item {key} must be an array")
    rows = [f"- {_label(key)}:\n"]
    allowed = _OBJECT_LIST_FIELDS[key]
    for index, entry in enumerate(value, 1):
        if not isinstance(entry, Mapping) or set(entry) - set(allowed):
            raise PlanItemContractError(f"plan item {key}[{index}] has unknown fields")
        rows.append(f"  - Entry {index}:\n")
        for field in allowed:
            if field in entry:
                field_path = f"{key}[{index - 1}].{field}"
                field_value = entry[field]
                if isinstance(field_value, list):
                    rows.append(f"    - `{field_path}`:\n")
                    rows.extend(
                        f"      - `{scalar(item)}`\n" for item in field_value
                    )
                elif isinstance(field_value, Mapping):
                    raise PlanItemContractError(
                        f"plan item {field_path} must be scalar"
                    )
                elif field == "details" and isinstance(field_value, str):
                    rows.extend(_render_literal(f"`{field_path}`", field_value, "    "))
                else:
                    rows.append(
                        f"    - `{field_path}`: `{scalar(field_value)}`\n"
                    )
    return rows


def _render_variation_points(value: object) -> list[str]:
    if not isinstance(value, list):
        raise PlanItemContractError("plan item points must be an array")
    rows = ["- Points:\n"]
    allowed = {"behavior", "implementations", "evidence", "extractionDecision"}
    decision_fields = ("extract", "interfaceKind", "coveredBy", "rationale")
    for index, point in enumerate(value, 1):
        if not isinstance(point, Mapping) or set(point) - allowed:
            raise PlanItemContractError(f"plan item points[{index}] has unknown fields")
        rows.append(f"  - Point {index}:\n")
        for field in ("behavior", "implementations", "evidence"):
            rows.extend(_render_scalar_or_list(field, point.get(field), "    "))
        decision = point.get("extractionDecision")
        if not isinstance(decision, Mapping) or set(decision) - set(decision_fields):
            raise PlanItemContractError("variation extractionDecision has unknown fields")
        rows.append("    - Extraction decision:\n")
        for field in decision_fields:
            rows.append("      " + line(_label(field), decision.get(field)))
    return rows


def _render_variation_point_payload(payload: Mapping[str, Any]) -> list[str]:
    rows: list[str] = []
    for field in ("behavior", "implementations", "evidence"):
        rows.extend(_render_scalar_or_list(field, payload.get(field)))
    decision = payload.get("extractionDecision")
    decision_fields = ("extract", "interfaceKind", "coveredBy", "rationale")
    if not isinstance(decision, Mapping) or set(decision) - set(decision_fields):
        raise PlanItemContractError("variation extractionDecision has unknown fields")
    rows.append("- Extraction decision:\n")
    for field in decision_fields:
        rows.append("  " + line(_label(field), decision.get(field)))
    return rows


def _render_scalar_or_list(key: str, value: object, indent: str = "") -> list[str]:
    if isinstance(value, list):
        if any(isinstance(entry, (Mapping, list)) for entry in value):
            raise PlanItemContractError(f"plan item {key} must contain scalars")
        rows = [f"{indent}- {_label(key)}:\n"]
        rows.extend(f"{indent}  - `{scalar(entry)}`\n" for entry in value)
        return rows
    if isinstance(value, Mapping):
        raise PlanItemContractError(f"plan item {key} must be scalar")
    if key in {"command", "commandOrObservation"} and isinstance(value, str):
        return _render_literal(_label(key), value, indent)
    return [indent + line(_label(key), value)]


def _render_payload(item_id: str, payload: object) -> list[str]:
    if not isinstance(payload, Mapping):
        raise PlanItemContractError(f"plan item {item_id} payload must be an object")
    kind = item_id.split("-", 2)[1] if item_id.startswith("P-") else ""
    allowed = (
        _VAR_ANALYSIS_FIELDS if item_id == "P-Var-0"
        else _VAR_POINT_FIELDS if kind == "Var"
        else _PAYLOAD_FIELDS.get(kind)
    )
    if allowed is None or set(payload) - set(allowed):
        raise PlanItemContractError(f"plan item {item_id} has unknown payload fields")
    if kind == "Var" and item_id != "P-Var-0":
        return _render_variation_point_payload(payload)
    rows: list[str] = []
    for key in allowed:
        if key not in payload:
            continue
        if key in _OBJECT_LIST_FIELDS:
            rows.extend(_render_object_list(key, payload[key]))
        elif key == "points":
            rows.extend(_render_variation_points(payload[key]))
        else:
            rows.extend(_render_scalar_or_list(key, payload[key]))
    return rows


# 검증 시점의 서사에는 조립 소유 필드가 없다는 계약을 검증자에게 직접 알린다.
# run 003 실측: critic 이 P-Prep 항목에 `designSurfaceCoverage` 처분을 요구하며
# DISAGREE(b) — writer 가 그 필드를 쓰면 계약 위반이라 self-fix 로도 풀 수 없는
# 차단이었다.
_PREP_OWNERSHIP_NOTE = (
    "\nOwnership boundary: `designPreparation` / `designSurfaceCoverage` rows "
    "are produced by report assembly after this verification, so the narrative "
    "under review carries neither — their absence is the contract, not a "
    "defect. Judge the trigger evidence above; do not DISAGREE because the "
    "writer did not produce those fields.\n"
)


def _step_coordinate(payload: object, field: str) -> int | None:
    if not isinstance(payload, Mapping):
        return None
    value = payload.get(field)
    if isinstance(value, int) and not isinstance(value, bool) and value >= 1:
        return value
    return None


def _render_prior_steps(item: Mapping[str, Any], all_items: object) -> str:
    """같은 stage 의 선행 step 요약. run 003 실측: step 1.1 이 만드는 스크립트를
    1.2 검증자가 '없다'고 판정했다 — 검증 단위는 step 인데 payload 가 자기 행뿐이라
    계획이 앞 step 에서 만드는 것을 볼 수 없었다. dispatchQueue 가 좁혀져도 선행
    맥락은 봉투의 전체 items 에서 온다."""
    stage = _step_coordinate(item.get("payload"), "stage")
    step = _step_coordinate(item.get("payload"), "step")
    if stage is None or step is None or step < 2:
        return ""
    priors: list[tuple[int, str]] = []
    for other in all_items if isinstance(all_items, list) else []:
        if not isinstance(other, Mapping):
            continue
        other_step = _step_coordinate(other.get("payload"), "step")
        if (
            other_step is None
            or other_step >= step
            or _step_coordinate(other.get("payload"), "stage") != stage
        ):
            continue
        payload = other.get("payload")
        action = str(payload.get("action") or other.get("subject") or "").strip()
        files = payload.get("files")
        if isinstance(files, str) and files.strip():
            files = [files]
        suffix = (
            " — files: " + ", ".join(f"`{scalar(entry)}`" for entry in files)
            if isinstance(files, list) and files
            else ""
        )
        priors.append((other_step, f"- step {other_step}: {action}{suffix}\n"))
    if not priors:
        return ""
    priors.sort()
    return (
        "\nPrior steps in this stage (already applied when this step runs):\n"
        + "".join(text for _, text in priors)
        + "Judge this step against the state after them — a script, file, or "
        "path a prior step creates is not missing.\n"
    )


def _render_analyser_split(verdicts: object) -> str:
    """동수 항목에 이미 찍힌 분석자 표를 critic 프롬프트에 붙인다."""
    if not isinstance(verdicts, list):
        return ""
    rows: list[str] = []
    for verdict in verdicts:
        if not isinstance(verdict, Mapping):
            continue
        worker = str(verdict.get("worker") or "").strip()
        token = str(verdict.get("verdict") or "").strip()
        if not worker or not token:
            continue
        kind = str(verdict.get("breakageKind") or "").strip()
        label = f"{token}({kind})" if kind else token
        rows.append(f"- `{worker}`: `{label}`\n")
    if not rows:
        return ""
    return "Analyser split:\n" + "".join(rows)


def _tie_splits_from_state(
    state_path: Path | None, queue: Sequence[str],
) -> dict[str, list[Any]]:
    if state_path is None or not state_path.is_file():
        return {}
    items = _state_plan_body_items(_load_json_object(state_path), state_path)
    allowed = set(queue)
    splits: dict[str, list[Any]] = {}
    for row in items:
        if not isinstance(row, Mapping):
            continue
        item_id = str(row.get("id") or "")
        verdicts = row.get("verdicts")
        if item_id in allowed and isinstance(verdicts, list):
            splits[item_id] = verdicts
    return splits


_PRIOR_DISSENT_PROMPT_ANCHOR = "**Prior round dissent**"
_NO_SELF_FIX_NOTE = (
    "no correction — this item's text shifted under a neighbouring rewrite"
)


def _prior_vote(row: Mapping[str, Any]) -> dict[str, Any]:
    """직전 라운드의 한 표를 프롬프트가 렌더할 만큼만 옮긴다.

    `basis` 는 워커가 적은 근거다. `explanation` 이 계약상 필수 줄이고 `note` 는
    반증 후보라, 설명이 비어 있을 때만 후자로 내려간다. 둘 다 없으면 근거 없는
    표이므로 빈 문자열을 남긴다 — 없는 근거를 지어내지 않는다.
    """
    vote: dict[str, Any] = {
        "worker": str(row.get("worker") or ""),
        "verdict": str(row.get("verdict") or ""),
    }
    kind = str(row.get("breakageKind") or "").strip()
    if kind:
        vote["breakageKind"] = kind
    fixability = str(row.get("fixability") or "").strip()
    if fixability:
        vote["fixability"] = fixability
    basis = (
        str(row.get("explanation") or "").strip()
        or str(row.get("note") or "").strip()
    )
    if basis:
        vote["basis"] = basis
    return vote


def _prior_rounds_from_state(
    state_path: Path | None, queue: Sequence[str],
) -> dict[str, dict[str, Any]]:
    """큐에 오른 항목별로 직전 라운드의 표와 자가수정 노트를 모은다.

    `planBodyVerification.planItems[].verdicts[]` 는 라운드마다 덮어써지므로 지금
    거기 남아 있는 것이 직전 라운드의 판정이다. 자가수정 노트는 감사용
    `planItems[]` 행의 `selfFixNote` 에 본문 그대로 들어 있다.

    라운드 번호가 없는 표는 어느 라운드의 것인지 알 수 없어 싣지 않는다 —
    `apply-verdicts --round` 가 필수라 계약을 지킨 행에는 항상 번호가 있다.
    큐 순서를 그대로 따라 담아 같은 입력이 같은 바이트를 내도록 한다.
    """
    if state_path is None or not state_path.is_file():
        return {}
    state = _load_json_object(state_path)
    audit = state.get("planItems")
    notes = {
        str(row["id"]): row["selfFixNote"].strip()
        for row in (audit if isinstance(audit, list) else [])
        if isinstance(row, Mapping)
        and isinstance(row.get("id"), str)
        and isinstance(row.get("selfFixNote"), str)
        and row["selfFixNote"].strip()
    }
    voted = {
        str(row["id"]): row
        for row in _state_plan_body_items(state, state_path)
        if isinstance(row, Mapping) and isinstance(row.get("id"), str)
    }
    prior: dict[str, dict[str, Any]] = {}
    for item_id in queue:
        row = voted.get(item_id)
        if row is None:
            continue
        verdicts = [
            vote for vote in (row.get("verdicts") or [])
            if isinstance(vote, Mapping) and isinstance(vote.get("round"), int)
        ]
        if not verdicts:
            continue
        last = max(int(vote["round"]) for vote in verdicts)
        entry: dict[str, Any] = {
            "round": last,
            "votes": [
                _prior_vote(vote) for vote in verdicts if int(vote["round"]) == last
            ],
        }
        if item_id in notes:
            entry["selfFixNote"] = notes[item_id]
        prior[item_id] = entry
    return prior


def _reverify_carry(
    args: argparse.Namespace, queue: Sequence[str],
) -> dict[str, Any]:
    """라운드 2+ 봉투에 실을 직전 라운드 맥락. 실을 것이 없으면 빈 dict.

    라운드 1 은 `--state` 없이 준비되고, `--state` 가 있어도 판정이 하나도
    기록돼 있지 않으면 옮길 반대 의견이 없다. 두 경우 모두 봉투는 라운드 1 의
    모양 그대로다.
    """
    prior = _prior_rounds_from_state(getattr(args, "state", None), queue)
    if not prior:
        return {}
    return {"dispatchKind": "reverify", "priorRounds": prior}


def _render_prior_round(entry: object) -> str:
    """직전 라운드의 반대 의견을 이 항목의 블록으로 렌더한다.

    앵커는 `**Prior round dissent**` 이고, 워커가 답에 적는 줄의 앵커
    (`**Prior dissent**`, `verdict_blocks.py` 가 읽는다) 와 다른 문자열이다.
    두 표기는 서로 다른 산출물의 서로 다른 앵커다.
    """
    if not isinstance(entry, Mapping):
        return ""
    votes = entry.get("votes")
    if not isinstance(votes, list) or not votes:
        return ""
    rows = [
        f"{_PRIOR_DISSENT_PROMPT_ANCHOR} (round {scalar(entry.get('round'))}):\n"
    ]
    for vote in votes:
        if not isinstance(vote, Mapping):
            continue
        kind = str(vote.get("breakageKind") or "")
        token = str(vote.get("verdict") or "")
        if kind:
            token = f"{token}({kind})"
        rows.append(
            f"- `{scalar(vote.get('worker'))}`: `{scalar(token)}` — "
            f"`{scalar(vote.get('basis'))}`\n"
        )
        if vote.get("fixability"):
            rows.append("  " + line("Fixability", vote.get("fixability")))
    rows.append(
        line("What the planner changed", entry.get("selfFixNote") or _NO_SELF_FIX_NOTE)
    )
    return "".join(rows)


def _prompt(args: argparse.Namespace) -> str:
    envelope = _load_json_object(
        _prepared_items_path(args.run_manifest, require_regular=True)
    )
    all_items = envelope.get("items") if isinstance(envelope.get("items"), list) else []
    items = all_items
    queue = envelope.get("dispatchQueue")
    if isinstance(queue, list):
        allowed = {item_id for item_id in queue if isinstance(item_id, str)}
        items = [
            item for item in items
            if isinstance(item, Mapping) and item.get("id") in allowed
        ]
    prior_rounds = (
        envelope.get("priorRounds")
        if isinstance(envelope.get("priorRounds"), Mapping) else {}
    )
    rows = ["# Plan verification queue\n", line("Item count", len(items))]
    for index, item in enumerate(items, 1):
        if not isinstance(item, Mapping):
            continue
        item_id = str(item.get("id") or "")
        rows.extend((f"\n## Plan item {index}\n", line("Item ID", item_id),
                     line("Subject", item.get("subject"))))
        # 직전 반대 의견이 현재 본문보다 앞이다 — 계약이 요구하는 판단 순서가
        # "반대 의견과 정정을 읽고 나서 지금 본문을 본다" 이기 때문이다.
        rows.append(_render_prior_round(prior_rounds.get(item_id)))
        rows.extend(_render_payload(item_id, item.get("payload")))
        rows.append(_render_prior_steps(item, all_items))
        if item_id.startswith("P-Prep-"):
            rows.append(_PREP_OWNERSHIP_NOTE)
        split = _render_analyser_split(
            (envelope.get("tieSplits") or {}).get(item_id) if isinstance(
                envelope.get("tieSplits"), Mapping
            ) else None
        )
        if split:
            rows.append(split)
    body = with_response_format("".join(rows))
    if envelope.get("dispatchKind") == "critic-tie":
        return with_rendered_by(critic_tie_prompt_text(body))
    if envelope.get("dispatchKind") == "reverify":
        return with_rendered_by(reverify_prompt_text(body))
    return with_rendered_by(body)


def _validate_prepared(args: argparse.Namespace) -> dict[str, Any]:
    source = _plan_source(args)
    expected = _envelope(source)
    path = _prepared_items_path(args.run_manifest, require_regular=True)
    actual = _load_json_object(path)
    if actual.get("items") != expected["items"]:
        raise PlanItemContractError(
            "prepared plan items do not match deterministic extraction"
        )
    expected_queue = _queue_for(
        expected["items"],
        _planning(source),
        args.run_manifest,
        **_queue_kwargs(args),
    )
    if actual.get("dispatchQueue") != expected_queue:
        raise PlanItemContractError("prepared dispatch queue does not match")
    if not getattr(args, "tie_vote", False):
        # 라운드 2+ 의 반대 의견 반입은 여기서만 검사된다. 봉투에서 빠지면
        # 프롬프트에도 빠지고, 반대한 워커는 자기 판정을 그대로 다시 적는다.
        expected_carry = _reverify_carry(args, expected_queue)
        if actual.get("priorRounds") != expected_carry.get("priorRounds"):
            raise PlanItemContractError(
                "prepared plan items do not carry the previous round's dissent "
                "— re-run `okstra plan-items prepare` with the same --state so "
                "each re-dispatched item carries its `**Prior round dissent**` "
                "block"
            )
    return {"ok": True, "operation": "validate-prepared", "path": str(path)}


def _split_result_arg(raw: str) -> tuple[str, Path]:
    worker, separator, path = raw.partition("=")
    if not separator or not worker.strip() or not path.strip():
        raise PlanItemContractError(
            f"--result must be <worker-id>=<path>, got: {raw}"
        )
    return worker.strip(), Path(path.strip())


def _assigned_item_ids(items_path: Path) -> list[str]:
    envelope = _load_json_object(items_path)
    items = envelope.get("items")
    if not isinstance(items, list):
        raise PlanItemContractError(f"items envelope has no `items` array: {items_path}")
    queue = envelope.get("dispatchQueue")
    if isinstance(queue, list):
        return [item_id for item_id in queue if isinstance(item_id, str) and item_id]
    ids: list[str] = []
    for item in items:
        item_id = item.get("id") if isinstance(item, Mapping) else None
        if not isinstance(item_id, str) or not item_id:
            raise PlanItemContractError(f"every item needs an `id`: {items_path}")
        ids.append(item_id)
    return ids


def _verdict_row(worker: str, block: VerdictBlock) -> dict[str, Any]:
    """One `planItems[].verdicts[]` row. Optional fields stay absent when empty
    so the recorded table shows what the worker actually said.

    The verdict crosses a vocabulary boundary here: a worker answers
    `UNVERIFIABLE`, and the schema persists that as `verification-error`
    (`PLAN_ITEM_VERDICTS`). Writing the worker's token straight through produced
    a data.json its own schema rejects, and the mapping the contract prescribes
    had to be applied by hand every round.
    """
    row: dict[str, Any] = {
        "worker": worker,
        "verdict": PLAN_ITEM_VERDICTS[block.verdict],
    }
    for key, value in (
        ("breakageKind", block.breakage_kind),
        ("fixability", block.fixability),
        ("note", block.note),
        ("explanation", block.explanation),
    ):
        if value:
            row[key] = value
    return row


def _worker_blocks(
    raw_results: list[str], assigned: set[str]
) -> list[tuple[str, dict[str, VerdictBlock]]]:
    """Each worker's parsed response, refusing any queue mismatch.

    A missing vote and an invented item are both silent in a hand-written
    parser; each is a round scored on a table that does not match the queue.
    """
    collected: list[tuple[str, dict[str, VerdictBlock]]] = []
    seen_workers: set[str] = set()
    for raw in raw_results:
        worker, path = _split_result_arg(raw)
        if worker in seen_workers:
            raise PlanItemContractError(
                f"duplicate worker result for `{worker}` — one worker may cast "
                "only one verdict per plan-item round"
            )
        seen_workers.add(worker)
        try:
            blocks = parse_verdict_blocks(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError) as exc:
            raise PlanItemContractError(f"cannot read result {path}: {exc}") from exc
        answered = set(blocks)
        missing = sorted(assigned - answered)
        if missing:
            raise PlanItemContractError(
                f"worker `{worker}` was assigned {len(assigned)} items but "
                f"returned no verdict for {missing} — an unanswered item cannot "
                f"be scored, and dropping it silently is what makes a round look "
                f"complete when it is not"
            )
        unknown = sorted(answered - assigned)
        if unknown:
            raise PlanItemContractError(
                f"worker `{worker}` returned verdicts for {unknown}, which are "
                f"not in the persisted plan-item queue"
            )
        collected.append((worker, blocks))
    return collected


def _collect_verdicts(args: argparse.Namespace) -> dict[str, Any]:
    assigned = _assigned_item_ids(args.items)
    collected = _worker_blocks(args.result, set(assigned))
    envelope = {
        "schemaVersion": "1.0",
        "taskType": "implementation-planning",
        "planItems": [
            {
                "id": item_id,
                "verdicts": [
                    _verdict_row(worker, blocks[item_id])
                    for worker, blocks in collected
                ],
            }
            for item_id in assigned
        ],
    }
    write_json_atomic(args.output, envelope)
    return {"ok": True, "operation": "collect-verdicts", "path": str(args.output)}


def _plan_body_items(data: dict[str, Any], data_path: Path) -> list[dict[str, Any]]:
    verification = _planning(data).get("planBodyVerification")
    if not isinstance(verification, Mapping):
        raise PlanItemContractError(
            f"implementationPlanning.planBodyVerification must be an object: {data_path}"
        )
    items = verification.get("planItems")
    if not isinstance(items, list):
        raise PlanItemContractError(
            f"planBodyVerification.planItems must be an array: {data_path}"
        )
    return items


def _state_plan_body_items(
    state: dict[str, Any], state_path: Path,
) -> list[dict[str, Any]]:
    if state.get("owner") != "convergence":
        raise PlanItemContractError(
            f"plan-body state owner must be convergence: {state_path}"
        )
    verification = state.get("planBodyVerification")
    if not isinstance(verification, Mapping):
        raise PlanItemContractError(
            f"planBodyVerification must be an object: {state_path}"
        )
    items = verification.get("planItems")
    if not isinstance(items, list):
        raise PlanItemContractError(
            f"planBodyVerification.planItems must be an array: {state_path}"
        )
    return items


def _new_v3_state() -> dict[str, Any]:
    return {
        "schemaVersion": "1.1",
        "owner": "convergence",
        "planItems": [],
        "roundHistory": [],
        "selfFixRoundsApplied": 0,
        "planBodyVerification": {
            "roundCount": 0,
            "gateResult": "passed",
            "gateBlockedBy": [],
            "selfFixRoundsApplied": 0,
            "selfFixStopReason": "not-attempted",
            "planItems": [],
            "dissentLog": [],
        },
    }


def _derivations(args: argparse.Namespace) -> dict[str, Any]:
    """Candidate statements each answered clarification may have falsified.

    Advisory by construction: it reports where a decision's subject is mentioned
    and never which mentions are now wrong. Both contracts require the author to
    enumerate before editing; this supplies the enumeration, which is the half
    that was being skipped, and leaves the judgement where it belongs.
    """
    try:
        sidecar = args.response.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        raise PlanItemContractError(
            f"cannot read user-response sidecar {args.response}: {exc}"
        ) from exc
    planning = _planning(_load_json_object(args.data))
    entries = [
        entry for entry in parse_user_response_entries(sidecar)
        if args.clarification is None or entry.response_id == args.clarification
    ]
    if args.clarification is not None and not entries:
        raise PlanItemContractError(
            f"{args.response} has no response block for {args.clarification}"
        )
    clarifications = []
    for entry in entries:
        tokens = extract_tokens(f"{entry.value}\n{entry.rationale or ''}")
        clarifications.append({
            "id": entry.response_id,
            "disposition": entry.disposition,
            "tokens": tokens,
            "candidates": find_derivations(planning, tokens),
        })
    return {
        "ok": True,
        "operation": "derivations",
        "advisory": True,
        "clarifications": clarifications,
    }


def _stage_ledger_snapshot(run_manifest: Path | None) -> dict[str, str] | None:
    """`{스테이지 번호: 상태}`. 원장을 못 읽으면 ``None``.

    게이트가 범위를 좁히려면 어느 스테이지가 지금 시작 가능한지 알아야 한다. 그
    사실은 Stage 원장에 있고, 상태 어휘(`done` / `active` / `ready` / `blocked`)는
    `stage_targets.StageLifecycle.status` 의 것을 그대로 쓴다 — 원장이 자기 어휘를
    따로 가지면 같은 stage 가 소비처마다 다르게 읽힌다.

    ``None`` 은 판정 근거가 없다는 뜻이고, 소비처는 좁히지 않는다. 첫 계획 run 이라
    원장이 아직 없는 경우와 `--run-manifest` 를 안 넘긴 경우가 여기 해당한다.
    """
    if run_manifest is None:
        return None
    try:
        authority = validated_run_authority(run_manifest)
        task_root = RunRef.from_run_dir(authority.run_dir).task_root
        ledger = build_stage_ledger(task_root)
    except (ConvergenceContractError, ValueError, OSError):
        return None
    if not isinstance(ledger, Mapping) or not isinstance(ledger.get("stages"), list):
        return None
    snapshot = {
        str(row["stage"]): str(row["status"])
        for row in ledger["stages"]
        if isinstance(row, Mapping)
        and isinstance(row.get("stage"), int)
        and isinstance(row.get("status"), str)
    }
    return snapshot or None


# `plan-body-verification-<task-type-segment>-<seq>.json`. seq 는 파일 이름에만
# 있다 — 상태 본문은 자기 run 의 순번을 담지 않는다.
_PRIOR_STATE_NAME_RE = re.compile(
    r"^plan-body-verification-.+-(?P<seq>\d{3,})\.json$"
)


class _PriorRunCarry:
    """직전 run 에서 실제로 이월된 것. 이월을 요청하지 않았으면 만들지 않는다."""

    def __init__(self, prev_seq: str, count: int) -> None:
        self.prev_seq = prev_seq
        self.count = count


def _state_task_root(path: Path, *, option: str) -> Path:
    """``<task_root>/runs/<task-type>/state/<file>`` 의 task_root.

    plan-body 상태 파일은 자기 안에 task 신원을 담지 않는다(`_new_v3_state` 는
    `schemaVersion` 과 `owner` 만 쓴다). 남은 신원은 경로뿐이므로, 이월 양쪽이
    같은 task 의 기록인지는 여기서만 판정할 수 있다. 정규 run 경로가 아니면
    판정할 근거가 없다는 뜻이고, 그때는 이월을 거절한다 — 판정 못 하는 신원을
    통과시키면 남의 task 판정이 이 run 의 게이트를 결정한다.
    """
    try:
        return RunRef.from_run_dir(path.parent.parent).task_root
    except ValueError as exc:
        raise PlanItemContractError(
            f"{option} must be a plan-body state under "
            f"<task-root>/runs/<task-type>/state/: {path}"
        ) from exc


def _carry_prior_run_verdicts(
    args: argparse.Namespace,
    added: list[dict[str, Any]],
    hashes: Mapping[str, str],
) -> _PriorRunCarry | None:
    """직전 run 이 같은 본문에 내린 판정을 새로 시드된 행에 옮긴다.

    실패한 계획 run 을 다시 돌면 판정이 하나도 넘어오지 않았다. 이월은 답변된
    clarification 이 있는 경로에만 있고(`incremental_scope.decide_scope` 는
    답변된 `C-NNN` 이 없으면 `full` 을 되돌려 `incremental_carry` 를 도달 불가로
    만든다), 그래서 아무도 고치지 않은 문장이 매 재실행마다 다시 채점됐다.

    행 모양은 `incremental_carry._stamp_carried` 와 같다 — `verdicts` 를 그대로
    옮기고, `carriedForwardFromSeq` 에 직전 seq 를 찍고, `verifiedContentHash` 를
    이번 추출 해시로 맞춘다. 그 해시가 시드의 `dispatchQueue` 계산에 그대로
    들어가므로(`_queue_for` → `reverify_item_ids`), 이월된 행은 라운드 1 큐에서
    빠진다. 큐를 따로 줄이지 않는 이유가 이것이다: incremental 경로가
    `incremental_carry._recompute_dispatch_queue` 에서 쓰는 것과 같은 장치다.

    id 만 같은 행은 절대 이월하지 않는다. `P-*` id 는 위치에서 나오므로 계획이
    한 줄만 밀려도 같은 번호가 다른 문장을 가리킨다. 해시가 같아야 그 문장이
    같은 문장이다.
    """
    prior = getattr(args, "prior_state", None)
    if prior is None:
        return None
    if args.state is None:
        raise PlanItemContractError("--prior-state requires --state")
    matched = _PRIOR_STATE_NAME_RE.match(prior.name)
    if matched is None:
        raise PlanItemContractError(
            "--prior-state must name a plan-body-verification-<task-type>-<seq>"
            f".json file; the seq is what a carried row is tagged with: {prior}"
        )
    if _state_task_root(prior, option="--prior-state") != _state_task_root(
        args.state, option="--state"
    ):
        raise PlanItemContractError(
            f"--prior-state belongs to another task than --state: {prior}"
        )
    prev_seq = matched.group("seq")
    prior_rows = {
        str(row["id"]): row
        for row in _state_plan_body_items(_load_json_object(prior), prior)
        if isinstance(row, Mapping) and isinstance(row.get("id"), str)
    }
    count = 0
    for row in added:
        item_id = row.get("id")
        previous = prior_rows.get(item_id)
        if previous is None:
            continue
        verdicts = previous.get("verdicts")
        if not isinstance(verdicts, list) or not verdicts:
            continue
        current_hash = hashes.get(item_id)
        if not current_hash or previous.get("verifiedContentHash") != current_hash:
            continue
        row["verdicts"] = copy.deepcopy(verdicts)
        row["carriedForwardFromSeq"] = prev_seq
        row["verifiedContentHash"] = current_hash
        count += 1
    return _PriorRunCarry(prev_seq, count)


def _seed(args: argparse.Namespace) -> dict[str, Any]:
    """Create the `planBodyVerification.planItems[]` rows a round lands in.

    `apply-verdicts` refuses a verdict whose item has no row — correctly, since
    the gate is re-derived from that table and a verdict with nowhere to land
    would score as never cast. But nothing created the rows: the report writer
    leaves `planItems: []` (§5.5.9 is a lead substep that runs after it), and
    there was no step, in code or in the contract, that filled them. Every round
    had to be hand-seeded before the CLI would accept its own output.

    Idempotent by id. An existing row keeps everything it carries — verdicts
    already applied, `carriedForwardFromSeq`, `selfFixNote` — because a re-seed
    between rounds must not erase the round before it.

    `--prior-state` extends that to the previous *run*: a newly seeded item
    whose text the previous run already judged inherits its verdicts instead of
    entering round 1 again (`_carry_prior_run_verdicts`).
    """
    source = _plan_source(args)
    extracted = _envelope(source)["items"]
    hashes = {item["id"]: content_hash(item) for item in extracted}
    if args.state is not None:
        data = _load_json_object(args.state) if args.state.is_file() else _new_v3_state()
        recorded = _state_plan_body_items(data, args.state)
        target = args.state
    else:
        if args.data is None:
            raise PlanItemContractError("--narrative requires --state")
        data = _load_json_object(args.data)
        recorded = _plan_body_items(data, args.data)
        target = args.data
    known = {
        item.get("id")
        for item in recorded
        if isinstance(item, Mapping)
    }
    added = [
        # Only the fields a `planItems[]` row may carry. The extraction also
        # yields `payload` and `ticketId` for the verifier prompt, and the row
        # schema is `additionalProperties: false` — copying the item wholesale
        # put two schema violations in every seeded row, on the exact path the
        # contract tells a lead to follow.
        {
            key: item[key]
            for key in ("id", "subject", "sourceSection")
            if key in item
        } | (
            {"stageScope": item["stageScope"]}
            if item.get("stageScope") else {}
        ) | (
            {"block": item["block"]} if item.get("block") else {}
        ) | (
            {"contentHash": hashes[item["id"]]} if item["id"] in hashes else {}
        ) | {"verdicts": []}
        for item in extracted
        if item["id"] not in known
    ]
    carried = _carry_prior_run_verdicts(args, added, hashes)
    recorded.extend(added)
    for row in recorded:
        item_id = row.get("id") if isinstance(row, Mapping) else None
        if isinstance(item_id, str) and item_id in hashes:
            row["contentHash"] = hashes[item_id]
    if args.state is not None:
        audit_rows = data.get("planItems")
        if not isinstance(audit_rows, list):
            raise PlanItemContractError("state planItems must be an array")
        audited = {
            item.get("id") for item in audit_rows if isinstance(item, Mapping)
        }
        audit_rows.extend(
            {
                "id": item["id"],
                "subject": item["subject"],
                "sourceSection": item["sourceSection"],
                "ticketId": item.get("ticketId") or "unknown",
                "rounds": [],
                "clarificationId": None,
            }
            for item in extracted
            if item["id"] not in audited
        )
    verification = (
        data["planBodyVerification"] if args.state is not None
        else _planning(data)["planBodyVerification"]
    )
    ledger = _merged_ledger(_planning(source), getattr(args, "run_manifest", None))
    if ledger:
        verification["stageLedger"] = ledger
    previous = {
        str(row["id"]): str(row["verifiedContentHash"])
        for row in recorded
        if isinstance(row, Mapping)
        and isinstance(row.get("id"), str)
        and isinstance(row.get("verifiedContentHash"), str)
    }
    verification["dispatchQueue"] = _queue_for(
        extracted, _planning(source), getattr(args, "run_manifest", None), previous,
    )
    verification["gating"] = (
        verification.get("gating") is True or requires_plan_repair(verification)
        or not advisory_plan_body_gating(_planning(source), extracted)
    )
    write_json_atomic(target, data)
    result = {
        "ok": True,
        "operation": "seed",
        "path": str(target),
        "seeded": len(added),
        "existing": len(known),
    }
    if carried is not None:
        result["carried"] = carried.count
        result["carriedForwardFromSeq"] = carried.prev_seq
        if carried.count:
            # 시드는 prepare **뒤에** 돈다(리드 계약 step 1). 이월이 상태의
            # 큐만 줄이면 `plan-items prompt` 가 읽는 형제 파일은 여전히 전량
            # 이라, 워커는 직전 run 이 이미 판정한 행을 다시 받는다.
            sync_prepared_dispatch_queue(target, data)
    return result


def _judged_row(row: Mapping[str, Any], project_root: Path | None) -> dict[str, Any]:
    """한 verdict 행에 재현 결과를 채워 돌려준다.

    `reproductionResult` 는 들어온 값을 쓰지 않고 **항상 여기서 덮어쓴다.** 그
    필드가 1표 차단의 자격을 결정하므로, 워커가 보낸 값을 그대로 실으면 검증자가
    자기 주장의 판정을 스스로 적는 것이 된다. 계약이 "okstra 가 쓴다" 고 적은 것이
    이 뜻이고, 그 문장은 이 덮어쓰기가 있어야 참이다.

    프로젝트를 못 찾으면 `not-runnable` 이다 — 못 돌렸다는 사실이 기록되고, 그
    주장은 정족수로 내려간다. 조용히 통과시키지도, 근거 없이 차단하지도 않는다.
    """
    judged = dict(row)
    if str(judged.get("claimKind") or "") != "fact":
        judged.pop("reproductionResult", None)
        return judged
    judged["reproductionResult"] = (
        reproduce(judged.get("reproduction"), project_root=project_root)
        if project_root is not None
        else NOT_RUNNABLE
    )
    return judged


def _probe_project_root(run_manifest: Path | None) -> Path | None:
    if run_manifest is None:
        return None
    try:
        return validated_run_authority(run_manifest).project_root
    except ConvergenceContractError:
        return None


def _apply_verdicts(args: argparse.Namespace) -> dict[str, Any]:
    target = args.state or args.data
    data = _load_json_object(target)
    recorded = (
        _state_plan_body_items(data, target)
        if args.state is not None
        else _plan_body_items(data, target)
    )
    known = {item.get("id") for item in recorded if isinstance(item, Mapping)}
    verification = (
        data["planBodyVerification"] if args.state is not None
        else _planning(data)["planBodyVerification"]
    )
    discard_open_rounds = bool(getattr(args, "discard_open_rounds", False))
    if discard_open_rounds and args.state is not None and args.result:
        _restore_queue_from_results(verification, args.result, recorded)
    assigned = _narrow_dispatch_queue(args, verification, known)
    rows = _incoming_verdict_rows(args, assigned)
    missing = sorted(item_id for item_id in rows if item_id not in known)
    if missing:
        raise PlanItemContractError(
            f"the report's planBodyVerification has no row for {missing} — the "
            f"gate is re-derived from that table, so a verdict with nowhere to "
            f"land would be scored as if it were never cast"
        )
    if args.round_number < 1:
        raise PlanItemContractError("--round must be 1 or greater")
    project_root = _probe_project_root(getattr(args, "run_manifest", None))
    append = bool(getattr(args, "append", False))
    replaces_critic = append and any(
        is_critic_worker(row.get("worker", ""))
        and any(v.get("worker") == row.get("worker") for v in rows.get(item.get("id"), []))
        for item in recorded for row in item.get("verdicts", [])
    )
    if not append or replaces_critic:
        _reject_uncompleted_round_loss(
            recorded, rows, data.get("roundHistory"), args.round_number, target,
            discard_open_rounds=discard_open_rounds,
        )
    writer = _append_item_verdicts if append else _replace_item_verdicts
    for item in recorded:
        if isinstance(item, Mapping) and item.get("id") in rows:
            writer(item, rows[item["id"]], args.round_number, project_root)
    if requires_plan_repair(verification):
        verification["gating"] = True
    _validate_advisory_round(verification, args.round_number)
    write_json_atomic(target, data)
    return {"ok": True, "operation": "apply-verdicts", "path": str(target)}


def _stamped_verdicts(
    incoming: list[dict[str, Any]], round_number: int, project_root: Path | None,
) -> list[dict[str, Any]]:
    return [
        {**_judged_row(row, project_root), "round": round_number}
        for row in incoming
    ]


def _remember_verified_hash(item: dict[str, Any]) -> None:
    if item.get("contentHash"):
        item["verifiedContentHash"] = item["contentHash"]


def _replace_item_verdicts(
    item: dict[str, Any],
    incoming: list[dict[str, Any]],
    round_number: int,
    project_root: Path | None,
) -> None:
    item["verdicts"] = _stamped_verdicts(incoming, round_number, project_root)
    _remember_verified_hash(item)


def _append_item_verdicts(
    item: dict[str, Any],
    incoming: list[dict[str, Any]],
    round_number: int,
    project_root: Path | None,
) -> None:
    """분석자 표를 보존하고, 비판 검토자의 후속 판정을 현재 표로 갱신한다."""
    stamped = _stamped_verdicts(incoming, round_number, project_root)
    existing = item.get("verdicts")
    current = existing if isinstance(existing, list) else []
    seen = {row.get("worker"): row.get("round", 1) for row in current if isinstance(row, Mapping)}
    refreshed = {
        row.get("worker") for row in stamped
        if is_critic_worker(row.get("worker", ""))
        and row.get("worker") in seen and round_number > seen[row["worker"]]
    }
    clash = [row.get("worker") for row in stamped if row.get("worker") in seen and row.get("worker") not in refreshed]
    if clash:
        raise PlanItemContractError(
            f"plan item {item.get('id')} already has a vote from {clash} — "
            "the extra vote must come from a worker who has not voted on it. "
            "--append accepts critic corrections; an analyser re-voting in a later "
            "round is recorded without --append, after the earlier round is "
            "closed with complete-round"
        )
    item["verdicts"] = [row for row in current if row.get("worker") not in refreshed] + stamped
    _remember_verified_hash(item)


def _restore_queue_from_results(
    verification: Any,
    raw_results: list[str],
    recorded: Sequence[Mapping[str, Any]],
) -> None:
    """재적용하는 라운드의 큐를 그 라운드의 결과 파일이 답한 항목 집합으로 되돌린다.

    `dispatchQueue` 는 값이 하나뿐이라 라운드마다 덮인다 — 라운드 1 이 26개,
    critic 동수 라운드가 8개, 라운드 3 이 19개였던 run 에서 상태에 남는 것은
    19개뿐이다. `apply-verdicts` 와 `complete-round` 는 둘 다 그 큐로 스코프하고,
    `_queue_for` 는 이미 두 번 전진한 해시로부터의 순수 계산이라 지난 큐를
    되살릴 명령이 없었다. 그래서 `--discard-open-rounds` 로 라운드 1 을 다시
    적용하면 큐 밖 id 라며 거절됐고, 복구는 상태 파일을 손으로 고쳐야만 됐다
    (실측 2026-09-09, `fontsninja-v3-site` dev-10627 planning 002).

    결과 파일의 `### P-*` 블록 집합이 곧 그 라운드의 큐다 — 워커는 배정된 항목
    전부에 답해야 통과하고(`_worker_blocks`), 큐 밖 항목에는 답할 수 없었다.
    순서는 계획 항목 순서를 따른다. 계획에 없는 id 는 큐에 넣지 않으므로 뒤의
    `_worker_blocks` 가 그 id 를 이름하며 거절한다.
    """
    if not isinstance(verification, dict):
        return
    answered: set[str] = set()
    for raw in raw_results:
        _worker, path = _split_result_arg(raw)
        try:
            answered |= set(parse_verdict_blocks(path.read_text(encoding="utf-8")))
        except (OSError, UnicodeError) as exc:
            raise PlanItemContractError(f"cannot read result {path}: {exc}") from exc
    restored = [
        str(item["id"]) for item in recorded
        if isinstance(item, Mapping) and item.get("id") in answered
    ]
    verification["dispatchQueue"] = restored
    print(
        f"apply-verdicts: dispatchQueue restored from the result files "
        f"({len(restored)} items)",
        file=sys.stderr,
    )


def _reject_uncompleted_round_loss(
    recorded: Sequence[Mapping[str, Any]],
    rows: Mapping[str, Any],
    history: object,
    round_number: int,
    target: Path,
    *,
    discard_open_rounds: bool = False,
) -> None:
    """이번 라운드가 아직 닫히지 않은 다른 라운드의 표를 지우려 하면 쓰기 전에 거절한다.

    `--append` 없는 apply-verdicts 는 항목의 verdicts 를 통째로 교체한다 — 지난
    라운드 표는 `complete-round` 가 `planItems[].rounds` 에 스냅샷으로 남길 때만
    살아남는다. 그 스냅샷 없이 교체하면 표는 복구할 수 없이 사라지고, 유일한
    가드였던 `_reject_round_gap` 은 complete-round 안에 있어 사라진 뒤에야 말했다.
    실측(2026-09-09, `fontsninja-v3-site` dev-10627 planning 002): r1·r2 를 닫지
    않고 r3 를 적용해 큐 19건의 r1 표가 없어졌다.

    이력이 없는 데이터(final-report `--data` 경로)는 대상이 아니다 — 라운드
    스냅샷을 갖는 것은 convergence 소유 상태 파일뿐이다.
    """
    if not isinstance(history, list):
        return
    closed = {
        row.get("round")
        for row in history
        if isinstance(row, Mapping) and isinstance(row.get("round"), int)
    }
    at_risk: dict[int, list[str]] = {}
    for item in recorded:
        if not isinstance(item, Mapping) or item.get("id") not in rows:
            continue
        for row in item.get("verdicts") or []:
            if not isinstance(row, Mapping):
                continue
            recorded_round = row.get("round")
            if (
                isinstance(recorded_round, int)
                and recorded_round != round_number
                and recorded_round not in closed
            ):
                at_risk.setdefault(recorded_round, []).append(str(item.get("id")))
    if not at_risk:
        return
    detail = "; ".join(
        f"round {number}: {', '.join(sorted(set(ids)))}"
        for number, ids in sorted(at_risk.items())
    )
    if discard_open_rounds:
        # 복구 경로 — 잃어버린 라운드를 결과 파일에서 순서대로 다시 적용할 때는
        # 지금 남은 뒤 라운드 표가 버려야 할 쪽이다. 무엇을 버리는지는 남긴다.
        print(
            f"apply-verdicts: discarding open-round verdicts ({detail})",
            file=sys.stderr,
        )
        return
    commands = " then ".join(
        f"`okstra plan-items complete-round --state {target} "
        f"--run-manifest <run-manifest> --round {number}`"
        for number in sorted(at_risk)
    )
    raise PlanItemContractError(
        f"round {round_number} would replace verdicts of a round that was never "
        f"completed ({detail}) — replacing these verdicts would lose their history; "
        "only complete-round keeps a round's votes in planItems[].rounds. Close "
        f"the earlier round first: run {commands}, then re-run this command. "
        "If those rows are themselves being re-applied from their result files "
        "in order, pass --discard-open-rounds. Nothing was written."
    )


def _gate_module() -> Any:
    # 부모 개수를 세면 체크아웃에서만 맞는다 — 설치본은 패키지가
    # `~/.okstra/lib/python/` 이라 validators 가 `~/.okstra/lib/` 아래다.
    from .paths import RUN_VALIDATOR_RELATIVE, find_asset_root

    root = find_asset_root(RUN_VALIDATOR_RELATIVE)
    if root is None:
        raise PlanItemContractError("cannot locate plan-body gate authority")
    path = root.joinpath(*RUN_VALIDATOR_RELATIVE)
    spec = importlib.util.spec_from_file_location("okstra_plan_gate", path)
    if spec is None or spec.loader is None:
        raise PlanItemContractError("cannot load plan-body gate authority")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _round_snapshot(item: Mapping[str, Any], round_number: int) -> dict[str, Any]:
    votes = {
        str(row["worker"]): str(row["verdict"])
        for row in item.get("verdicts", [])
        if isinstance(row, Mapping) and row.get("round") == round_number
        and isinstance(row.get("worker"), str) and isinstance(row.get("verdict"), str)
    }
    disagrees = sum(value.startswith("DISAGREE") for value in votes.values())
    if votes and disagrees == 0:
        classification = "full-consensus"
    elif disagrees * 2 > len(votes):
        classification = "majority-disagree"
    else:
        classification = "partial-consensus"
    return {"round": round_number, "votes": votes, "classification": classification}


def _plan_gate_summary(verification: Mapping[str, Any]) -> dict[str, Any]:
    module = _gate_module()
    summary = module.plan_body_gate_summary({
        "implementationPlanning": {"planBodyVerification": dict(verification)}
    })
    if not isinstance(summary, Mapping):
        raise PlanItemContractError("plan-body gate authority returned no gate")
    return dict(summary)


def _record_self_fixes(args: argparse.Namespace, audit: list[object], verification: dict[str, Any]) -> None:
    previous = self_fix_rounds(verification)
    if (args.self_fix_group or args.self_fix_note) and previous - {args.round_number}:
        raise PlanItemContractError(
            "automatic self-fix is limited to one rewrite; resolve remaining "
            "items through a lead decision or user confirmation"
        )
    notes: dict[str, str] = {}
    known = {item.get("id") for item in audit if isinstance(item, Mapping)}
    if args.self_fix_group and not args.self_fix_stop_reason:
        raise PlanItemContractError(
            "--self-fix-group requires --self-fix-stop-reason — a self-fix "
            "round has to say why the loop stops, and a defaulted "
            "`all-resolved` is the one value that later forbids promoting the "
            "items it left unresolved"
        )
    for raw in args.self_fix_note:
        item_id, separator, filename = raw.partition("=")
        if not separator or not item_id or not filename:
            raise PlanItemContractError("--self-fix-note must be <item-id>=<markdown-file>")
        if item_id not in known:
            raise PlanItemContractError(f"unknown self-fix item `{item_id}`")
        try:
            notes[item_id] = Path(filename).read_text(encoding="utf-8").strip()
        except (OSError, UnicodeError) as exc:
            raise PlanItemContractError(f"cannot read self-fix note {filename}: {exc}") from exc
    for item in audit:
        if isinstance(item, dict) and item.get("id") in notes:
            item["selfFixNote"] = notes[item["id"]]
    if notes and not args.self_fix_group and not previous:
        raise PlanItemContractError("--self-fix-note requires --self-fix-group to account for the rewrite")
    groups = []
    for raw in args.self_fix_group:
        filename, separator, item_ids = raw.partition("=")
        if not separator or not filename or not item_ids:
            raise PlanItemContractError("--self-fix-group must be <cause-file>=<item-id>[,<item-id>...]")
        try:
            cause = Path(filename).read_text(encoding="utf-8").strip()
        except (OSError, UnicodeError) as exc:
            raise PlanItemContractError(f"cannot read self-fix cause {filename}: {exc}") from exc
        ids = [item for item in item_ids.split(",") if item]
        unknown = sorted(set(ids) - known)
        if unknown:
            raise PlanItemContractError("unknown self-fix item(s): " + ", ".join(unknown))
        groups.append({"round": args.round_number, "causeSummary": cause, "itemIds": ids})
    if groups:
        verification["selfFixGroups"] = groups
        verification["selfFixRoundsApplied"] = args.round_number
    if args.self_fix_stop_reason:
        # 원인 그룹 없이 중단 사유만 기록하는 라운드는 `selfFixRoundsApplied` 를
        # 건드리지 않는다. `validate-run.py` `_validate_self_fix_grouping` 이
        # `max(selfFixGroups[].round) == selfFixRoundsApplied` 를 요구하므로,
        # 그룹 없이 올린 숫자는 통과할 수 있는 값이 없는 상태를 만든다.
        verification["selfFixStopReason"] = args.self_fix_stop_reason


def _round_inputs(args: argparse.Namespace) -> tuple[dict[str, Any], list[dict[str, Any]], list[object], list[object]]:
    data = _load_json_object(args.state)
    audit, history = data.get("planItems"), data.get("roundHistory")
    if not isinstance(audit, list) or not isinstance(history, list):
        raise PlanItemContractError("state planItems and roundHistory must be arrays")
    current = _state_plan_body_items(data, args.state)
    if args.command == "complete-round" and args.items is None:
        manifest = _load_json_object(args.run_manifest)
        if manifest.get("planBodyVerificationPath"):
            prepared = _prepared_items_path(args.run_manifest)
            if prepared.is_file():
                voted = {item["id"] for item in current
                         if _round_snapshot(item, args.round_number)["votes"]}
                # 이후 회차용 준비 파일이 이미 저장된 현재 회차의 표를 제외하면 쓰지 않는다.
                if voted <= set(_assigned_item_ids(prepared)):
                    args = copy.copy(args)
                    args.items = prepared
    _narrow_dispatch_queue(
        args, data["planBodyVerification"], {item.get("id") for item in current},
    )
    return data, current, audit, history


def _round_snapshots(current: list[dict[str, Any]], verification: Mapping[str, Any], round_number: int) -> tuple[dict[str, dict[str, Any]], dict[str, Any]]:
    queue = verification.get("dispatchQueue")
    scoped = current
    if isinstance(queue, list):
        allowed = {item_id for item_id in queue if isinstance(item_id, str)}
        scoped = [
            item for item in current
            if isinstance(item, Mapping) and item.get("id") in allowed
        ]
    snapshots = {str(item.get("id")): _round_snapshot(item, round_number) for item in scoped}
    if not snapshots:
        raise PlanItemContractError(f"round {round_number} dispatch queue has no current plan items")
    missing = [item_id for item_id, row in snapshots.items() if not row["votes"]]
    if missing:
        raise PlanItemContractError(
            f"round {round_number} has no verdict for dispatched plan items: {', '.join(missing)}. "
            "Earlier-round verdicts remain recorded; check this round's prepared queue "
            "before requesting new verdicts."
        )
    summary = _plan_gate_summary(verification)
    classes = {row["id"]: row["stateClassification"] for row in summary["items"]}
    for item_id, snapshot in snapshots.items():
        snapshot["classification"] = classes.get(item_id, snapshot["classification"])
    return snapshots, summary


def _record_audit_round(audit: list[object], snapshots: Mapping[str, dict[str, Any]], round_number: int) -> None:
    for item in audit:
        if isinstance(item, dict) and str(item.get("id")) in snapshots:
            rounds = item.get("rounds")
            if not isinstance(rounds, list):
                raise PlanItemContractError("state plan item rounds must be an array")
            item["rounds"] = [row for row in rounds if row.get("round") != round_number]
            item["rounds"].append(snapshots[str(item["id"])])


def _participant_counts(
    manifest: Mapping[str, Any],
    snapshots: Mapping[str, dict[str, Any]],
    items: Sequence[Mapping[str, Any]],
) -> tuple[set[str], int, int]:
    """이번 라운드의 투표자, run 전체의 분석자 수, 로스터 크기.

    `workers` 는 uniformVerifiers 를 뽑기 위한 이번 라운드 큐의 투표자다.
    `voting` 은 게이트 산술의 분모이므로 라운드 큐가 아니라 run 전체를 세고,
    검증기와 같은 함수(`voting_analyser_keys`)를 쓴다 — 종전에는 이쪽만 라운드
    큐를 세서, critic 이 동수만 가른 라운드에서 기록값 0 과 검증기 재계산값이
    갈렸다.
    """
    workers = {worker for row in snapshots.values() for worker in row["votes"]}
    voting = voting_analyser_keys(items)
    assignments = manifest.get("workerAssignments")
    if not isinstance(assignments, list):
        raise PlanItemContractError("run manifest workerAssignments must be an array")
    rostered = sum(1 for row in assignments if isinstance(row, Mapping)
                   and isinstance(row.get("workerId"), str) and row["workerId"] != "report-writer")
    return workers, len(voting), rostered


def _reject_round_gap(history: list[Any], round_number: int) -> None:
    """앞 라운드의 결과가 기록되기 전에 다음 라운드를 닫지 못하게 한다.

    `roundCount` 는 이 인자의 최대값이라 건너뛴 라운드까지 세지만, 이력에는
    행이 남지 않는다. 그 불일치는 phase 끝에서 활동 건수 검증이 잡아내는데
    (`recorded=3` vs `rounds=4`), 그때는 라운드를 다시 닫을 방법이 없어 실행
    전체가 막힌다. 빠진 라운드의 판정도 그대로 사라진다 — 다음 라운드는 그것을
    settle 하러 존재하므로 근거 없는 표결이 된다.

    실측(2026-08-27, `fontsninja-v3-site` `dev-10341`): 이력이 1·2·4 이고
    `roundCount` 가 4 였다.
    """
    recorded = {
        row.get("round")
        for row in history
        if isinstance(row, Mapping) and isinstance(row.get("round"), int)
    }
    missing = [
        number for number in range(1, round_number) if number not in recorded
    ]
    if missing:
        raise PlanItemContractError(
            f"round {round_number} cannot be completed while round(s) "
            f"{', '.join(str(number) for number in missing)} have no history "
            "entry — complete them in order so each round's verdict is recorded"
        )


def _validate_advisory_round(verification: Mapping[str, Any], round_number: int) -> None:
    """분석자 검증 횟수 제한은 비판 검토자의 교정에 적용하지 않는다."""
    if verification.get("gating") is not False or round_number <= 1:
        return
    if any(
        row.get("round") == round_number and not is_critic_worker(row.get("worker", ""))
        for item in verification.get("planItems", []) for row in item.get("verdicts", [])
    ):
        raise PlanItemContractError(
            "advisory plan-body gating allows one verification round for analysers; "
            "critic corrections do not consume that limit"
        )


def _complete_round(args: argparse.Namespace) -> dict[str, Any]:
    if args.round_number < 1:
        raise PlanItemContractError("--round must be 1 or greater")
    data, current, audit, history = _round_inputs(args)
    verification = data["planBodyVerification"]
    if requires_plan_repair(verification):
        verification["gating"] = True
    _validate_advisory_round(verification, args.round_number)
    if verification.get("gating") is False:
        if args.self_fix_group or args.self_fix_note or args.self_fix_stop_reason:
            raise PlanItemContractError(
                "advisory plan-body gating forbids the self-fix loop"
            )
    _record_self_fixes(args, audit, verification)
    _reject_round_gap(history, args.round_number)
    snapshots, summary = _round_snapshots(current, verification, args.round_number)
    _record_audit_round(audit, snapshots, args.round_number)
    gate = str(summary["recomputed"])
    manifest = _load_json_object(args.run_manifest)
    workers, voting, rostered = _participant_counts(manifest, snapshots, current)
    # critic-tie 는 큐 일부에만 표를 남긴다. 합집합 워커를 모든 행에
    # 강제하면 표 없는 행에서 KeyError 가 난다. 모든 행에 같은 판정이
    # 있을 때만 uniform 이다.
    uniform = []
    for worker in workers:
        values = [
            row["votes"][worker]
            for row in snapshots.values()
            if worker in row["votes"]
        ]
        if len(values) != len(snapshots) or len(set(values)) != 1:
            continue
        uniform.append(
            {"worker": worker, "verdict": values[0], "itemCount": len(values)}
        )
    verification.update({"roundCount": max(int(verification.get("roundCount", 0)), args.round_number),
                         "gateResult": gate, "gateBlockedBy": summary["blockedBy"],
                         # 무엇이 막았는지는 `gateBlockedBy` 가, 무엇이 막지 않았고
                         # 왜인지는 이 쪽이 적는다. 둘 다 같은 계산에서 나온다.
                         "setAside": summary["setAside"],
                         "participatingAnalysers": {"rostered": rostered, "voting": voting},
                         "uniformVerifiers": uniform})
    if args.self_fix_group:
        data["selfFixRoundsApplied"] = args.round_number
    history[:] = [row for row in history if row.get("round") != args.round_number]
    history.append({"round": args.round_number, "completedAt": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), "gateResult": gate, "gateBlockedBy": verification["gateBlockedBy"]})
    write_json_atomic(args.state, data)
    return {
        "ok": True,
        "operation": "complete-round",
        "path": str(args.state),
        "gateResult": gate,
        "nextDispatch": _state_next_dispatch(args).as_dict(),
    }


def _prepared_payloads(run_manifest: Path | None) -> dict[str, dict[str, Any]]:
    if run_manifest is None:
        return {}
    try:
        envelope = _load_json_object(
            _prepared_items_path(run_manifest, require_regular=True)
        )
    except PlanItemContractError:
        return {}
    items = envelope.get("items")
    if not isinstance(items, list):
        return {}
    return {
        str(item["id"]): item
        for item in items
        if isinstance(item, Mapping) and isinstance(item.get("id"), str)
    }


def _rostered_critic(run_manifest: Path | None) -> bool:
    """매니페스트 없이 부르면 종전대로 critic 이 있다고 본다.

    없다고 단정하면 critic 을 둔 run 이 가를 수 있는 동수까지 사용자에게
    넘긴다 — 그쪽이 더 나쁜 오답이다. 리드 계약은 `next-dispatch` 를
    `--run-manifest` 와 함께 부르도록 못박고 있다
    (`prompts/lead/plan-body-verification.md` §"Round protocol" step 6).
    """
    if run_manifest is None:
        return True
    return critic_is_rostered(_load_json_object(run_manifest))


def _state_next_dispatch(args: argparse.Namespace) -> NextDispatch:
    data, current, _audit, _history = _round_inputs(args)
    run_manifest = getattr(args, "run_manifest", None)
    payloads = _prepared_payloads(run_manifest)
    # 실제 목록에서 제외할 항목을 다음 비평 검증으로 요구하면 빈 큐에서 멈춘다.
    allowed = set(dispatch_item_ids(
        [payloads.get(item["id"], item) for item in current],
        data["planBodyVerification"].get("stageLedger"),
    ))
    current = [item for item in current if item["id"] in allowed]
    return next_dispatch(
        current, payloads, critic_rostered=_rostered_critic(run_manifest),
        decision_items=(
            _plan_gate_summary(data["planBodyVerification"])["items"]
            if self_fix_rounds(data["planBodyVerification"]) else ()
        ),
    )


def _resolve_dissent(args: argparse.Namespace) -> dict[str, Any]:
    data, current, _audit, history = _round_inputs(args)
    verification = data["planBodyVerification"]
    item = next((row for row in current if row.get("id") == args.item), None)
    if item is None:
        raise PlanItemContractError(f"unknown plan item `{args.item}`")
    closed = {row.get("round") for row in history if isinstance(row, Mapping)}
    if any(row.get("round") not in closed for row in item.get("verdicts", [])):
        raise PlanItemContractError("complete-round must record every verdict before a lead decision")
    summary = _plan_gate_summary(verification)
    authority = next(row for row in summary["items"] if row["id"] == args.item)
    if authority["decisionAuthority"] != "lead":
        raise PlanItemContractError(
            f"`{args.item}` is not a lead-owned judgement after self-fix; "
            "use user confirmation for unresolved facts, requirements, risks or preferences"
        )
    try:
        decision = args.decision_file.read_text(encoding="utf-8").strip()
    except (OSError, UnicodeError) as exc:
        raise PlanItemContractError(f"cannot read lead decision: {exc}") from exc
    if not decision:
        raise PlanItemContractError("lead decision must state the decision, authority and cited evidence")
    item["leadDecision"] = {"basisHash": lead_decision_basis(item), "decision": decision}
    entry = {"planItem": args.item, "workerRole": "lead", "body": decision}
    dissent = verification.setdefault("dissentLog", [])
    if entry not in dissent:
        dissent.append(entry)
    summary = _plan_gate_summary(verification)
    verification.update({
        "gateResult": summary["recomputed"], "gateBlockedBy": summary["blockedBy"],
        "setAside": summary["setAside"],
    })
    write_json_atomic(args.state, data)
    return {"ok": True, "operation": "resolve-dissent", "itemId": args.item,
            "gateResult": summary["recomputed"]}


def _next_dispatch(args: argparse.Namespace) -> dict[str, Any]:
    decision = _state_next_dispatch(args)
    return {"ok": True, "operation": "next-dispatch", **decision.as_dict()}


def _correction_prompt(args: argparse.Namespace) -> str:
    decision = _state_next_dispatch(args)
    if decision.kind != "worker-correction" or args.worker not in decision.workers:
        raise PlanItemContractError(
            f"worker `{args.worker}` is not a blanket-UNVERIFIABLE correction "
            "target — do not open a queue round"
        )
    return correction_prompt_text(_prompt(args))


def _narrow_dispatch_queue(
    args: argparse.Namespace, verification: dict[str, Any], known: set[object],
) -> set[object]:
    """이 라운드가 실제로 배정한 항목들.

    tie 라운드는 큐의 일부(7항목)만 critic 에게 보낸다. 그런데 `--result` 는
    지금까지 state 에 남은 라운드 큐(44항목) 전체를 배정으로 보고 답 없는
    37항목을 미응답으로 거절했다 — 문서가 "model-facing" 이라고 적은 형식이
    tie 라운드에서는 쓸 수 없고, 우회로가 헬프 스스로 historical 이라 적은
    `--verdicts` 뿐이었다(실측 2026-09-10, fontsninja-v3-site dev-10628-3).
    배정 범위는 판정 저장과 완료 기록이 함께 써야 한다. 지역 변수만 좁히면
    저장은 성공해도 완료가 과거 큐 전체에서 새 라운드의 표를 요구한다.
    상태의 큐만 좁히고 기존 표와 라운드 이력은 보존한다.
    """
    queue = verification.get("dispatchQueue")
    assigned = (
        {item_id for item_id in queue if isinstance(item_id, str)}
        if isinstance(queue, list) else known
    )
    items_path = getattr(args, "items", None)
    if items_path is None:
        return assigned
    narrowed = {item_id for item_id in _assigned_item_ids(items_path)}
    if not narrowed:
        raise PlanItemContractError(
            f"items artifact dispatches nothing: {items_path}"
        )
    unknown = sorted(narrowed - {i for i in assigned if isinstance(i, str)})
    if unknown:
        raise PlanItemContractError(
            f"items artifact dispatches {unknown}, which this round's persisted "
            f"queue does not contain — pass the artifact this round dispatched, "
            f"not another round's"
        )
    verification["dispatchQueue"] = [
        item_id for item_id in (queue if isinstance(queue, list) else sorted(assigned))
        if item_id in narrowed
    ]
    return narrowed


def _incoming_verdict_rows(
    args: argparse.Namespace, known: set[object],
) -> dict[str, list[dict[str, Any]]]:
    if args.verdicts is not None:
        incoming = _load_json_object(args.verdicts).get("planItems")
        if not isinstance(incoming, list):
            raise PlanItemContractError("verdicts envelope has no `planItems` array")
        return {
            item["id"]: item.get("verdicts", [])
            for item in incoming
            if isinstance(item, Mapping) and isinstance(item.get("id"), str)
        }
    assigned = {item_id for item_id in known if isinstance(item_id, str)}
    collected = _worker_blocks(args.result, assigned)
    return {
        item_id: [_verdict_row(worker, blocks[item_id]) for worker, blocks in collected]
        for item_id in assigned
    }


_HANDLERS = {
    "extract": _extract,
    "validate": _validate,
    "prepare": _prepare,
    "prompt": _prompt,
    "validate-prepared": _validate_prepared,
    "collect-verdicts": _collect_verdicts,
    "derivations": _derivations,
    "seed": _seed,
    "apply-verdicts": _apply_verdicts,
    "complete-round": _complete_round,
    "next-dispatch": _next_dispatch,
    "resolve-dissent": _resolve_dissent,
    "correction-prompt": _correction_prompt,
}


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        result = _HANDLERS[args.command](args)
    except (PlanItemContractError, VerdictBlockError, OSError, ValueError) as exc:
        print(f"plan-items: {exc}", file=sys.stderr)
        manifest = getattr(args, "run_manifest", None)
        if manifest:
            logged = record_runtime_failure(
                Path(manifest), command=f"plan-items {args.command}", exit_code=2, detail=str(exc)
            )
            if not logged["ok"]:
                print(f"error-log: {logged['reason']}", file=sys.stderr)
        return 2
    if isinstance(result, str):
        print(result, end="")
    elif args.command in {"prepare", "validate-prepared"}:
        rows = (
            "Plan items\n"
            + line("Status", "ready")
            + line("Operation", result["operation"])
        )
        if "gating" in result:
            rows += line("Gating", result["gating"])
        print(rows, end="")
    else:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())