"""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 importlib.util
import json
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 .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,
    content_hash,
    correction_prompt_text,
    critic_tie_prompt_text,
    dispatch_item_ids,
    extract_plan_items,
    next_dispatch,
    planning_stage_ledger,
    reverify_item_ids,
    tie_vote_item_ids,
)
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 .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",
    )
    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",
    )
    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)
    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",
    )
    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(
        "--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 these votes to existing rows instead of replacing the 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("--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>...]")
    complete.add_argument("--self-fix-stop-reason", choices=("all-resolved", "no-progress", "max-rounds-reached"))
    _add_dispatch_commands(commands)
    return parser


def _add_dispatch_commands(commands: Any) -> None:
    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)
    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"],
        )
    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"),
    "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_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"
                    )
                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")
    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


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


def _prompt(args: argparse.Namespace) -> str:
    envelope = _load_json_object(
        _prepared_items_path(args.run_manifest, require_regular=True)
    )
    items = envelope.get("items") if isinstance(envelope.get("items"), list) else []
    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
        ]
    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.extend(_render_payload(item_id, item.get("payload")))
        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 = "".join(rows)
    if envelope.get("dispatchKind") == "critic-tie":
        return critic_tie_prompt_text(body)
    return 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")
    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


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.
    """
    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
    ]
    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"] = not advisory_plan_body_gating(
        _planning(source), extracted,
    )
    write_json_atomic(target, data)
    return {
        "ok": True,
        "operation": "seed",
        "path": str(target),
        "seeded": len(added),
        "existing": len(known),
    }


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"]
    )
    queue = verification.get("dispatchQueue") if isinstance(verification, Mapping) else None
    assigned = (
        {item_id for item_id in queue if isinstance(item_id, str)}
        if isinstance(queue, list) else 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))
    writer = (
        _append_item_verdicts if getattr(args, "append", False)
        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)
    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:
    """동수 항목의 critic 표. 이미 투표한 워커는 거부한다."""
    stamped = _stamped_verdicts(incoming, round_number, project_root)
    existing = item.get("verdicts")
    current = existing if isinstance(existing, list) else []
    seen = {
        row.get("worker") for row in current if isinstance(row, Mapping)
    }
    clash = [row.get("worker") for row in stamped if row.get("worker") in seen]
    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"
        )
    item["verdicts"] = [*current, *stamped]
    _remember_verified_hash(item)


def _gate_module() -> Any:
    path = Path(__file__).parents[2] / "validators" / "validate-run.py"
    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]:
    path = Path(__file__).parents[2] / "validators" / "validate-run.py"
    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)
    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:
    notes: dict[str, str] = {}
    known = {item.get("id") for item in audit if isinstance(item, Mapping)}
    if args.self_fix_stop_reason and not args.self_fix_group:
        raise PlanItemContractError("--self-fix-stop-reason requires --self-fix-group")
    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"]]
    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
        verification["selfFixStopReason"] = args.self_fix_stop_reason or "all-resolved"


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")
    return data, _state_plan_body_items(data, args.state), 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 or any(not row["votes"] for row in snapshots.values()):
        raise PlanItemContractError("every current plan item needs a verdict before completion")
    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(args: argparse.Namespace, snapshots: Mapping[str, dict[str, Any]]) -> tuple[set[str], int, int]:
    workers = {worker for row in snapshots.values() for worker in row["votes"]}
    voting = {
        worker for row in snapshots.values()
        for worker, verdict in row["votes"].items()
        if verdict != "verification-error"
    }
    assignments = _load_json_object(args.run_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 _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 verification.get("gating") is False:
        if args.round_number > 1:
            raise PlanItemContractError(
                "advisory plan-body gating allows one verification round"
            )
        if args.self_fix_group or args.self_fix_note:
            raise PlanItemContractError(
                "advisory plan-body gating forbids the self-fix loop"
            )
    snapshots, summary = _round_snapshots(current, verification, args.round_number)
    _record_audit_round(audit, snapshots, args.round_number)
    gate = str(summary["recomputed"])
    workers, voting, rostered = _participant_counts(args, snapshots)
    uniform = [
        {"worker": worker, "verdict": values[0], "itemCount": len(values)}
        for worker in workers
        for values in [[row["votes"][worker] for row in snapshots.values()]]
        if len(set(values)) == 1
    ]
    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})
    _record_self_fixes(args, audit, verification)
    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)
    payloads = _prepared_payloads(args.run_manifest)
    return {
        "ok": True,
        "operation": "complete-round",
        "path": str(args.state),
        "gateResult": gate,
        "nextDispatch": next_dispatch(current, payloads).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 _state_next_dispatch(args: argparse.Namespace) -> NextDispatch:
    _data, current, _audit, _history = _round_inputs(args)
    payloads = _prepared_payloads(getattr(args, "run_manifest", None))
    return next_dispatch(current, payloads)


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 _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,
    "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)
        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())