"""Compact initial final-verification prompt contract."""
from __future__ import annotations

from .agent.invocation import EXECUTION_DELIVERY_PREFIXES

import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Mapping, Sequence

from .convergence_reverify_prompt import RENDERED_BY_LINE
from .convergence_critic_verify_prompt import (
    RENDERED_BY_LINE as CRITIC_VERIFY_RENDERED_BY_LINE,
)
from .plan_items import RENDERED_BY_LINE as PLAN_VERIFY_RENDERED_BY_LINE
from .worker_prompt_body import analysis_worker_label
from .json_boundary import load_owned_object
from .worker_prompt_policy import (
    ERRORS_PATH_HEADERS,
    IMPLEMENTATION_HEADERS,
    CRITIC_VERIFY_DISPATCH_KIND,
    PromptPlan,
    is_plan_verify_dispatch_kind,
    resolve_prompt_plan_for_manifest,
)
from .worker_prompt_headers import EVIDENCE_LEDGER_HEADER


MAX_FINAL_VERIFICATION_DIRECTIVE_LINES = 40
MAX_FINAL_VERIFICATION_BODY_LINES = 96

PROMPT_DELIVERY_MODE_HEADER = "**Prompt Delivery Mode:**"
PROMPT_DELIVERY_MODES = frozenset({"eager-include", "lazy-path-reference"})
MODEL_HEADER = "**Model:**"
TASK_TYPE_HEADER = "**Task Type:**"
FORBIDDEN_ACTIONS_HEADER = "**Forbidden actions:**"

_DIRECTIVE_HEADING = "## Run-specific directive"
_WORKER_ERROR_CONTRACT_HEADER = "**Worker Error Contract Path:**"
_EVIDENCE_LEDGER_HEADER_PREFIX = "**Evidence ledger:**"
_PRIMARY_PACKET_RE = re.compile(
    r"(?im)^-\s+Primary analysis packet:\s+`[^`\n]*analysis-packet\.md`\s*$"
)
_COPIED_SECTION_PATTERNS = (
    ("Primary focus areas", re.compile(r"(?im)^(?:#{1,6}\s+|-\s+)Primary focus areas\b")),
    (
        "Required deliverable shape",
        re.compile(r"(?im)^(?:#{1,6}\s+|-\s+)Required deliverable shape\b"),
    ),
    (
        "Self-review pass",
        re.compile(r"(?im)^(?:#{1,6}\s+|-\s+)Self-review pass\b"),
    ),
)
_NON_BODY_PREFIXES = (
    "**Project Root:**",
    "**Prompt History Path:**",
    "**Result Path:**",
    "**Audit sidecar path:**",
    "Assigned worker prompt history path:",
    "**Worker Preamble Path:**",
    _EVIDENCE_LEDGER_HEADER_PREFIX,
    "**Evidence citations:**",
    *ERRORS_PATH_HEADERS,
    "**Read scope:**",
    "**File write mode:**",
    *IMPLEMENTATION_HEADERS,
    "**Verification scope:**",
    "**Verification base ref:**",
    "**Verification head ref:**",
    "**Verification target path:**",
    "**Verification target digest:**",
    PROMPT_DELIVERY_MODE_HEADER,
)
_REQUIRED_TARGET_PREFIXES = (
    "**Worktree:**",
    "**Verification scope:**",
    "**Verification base ref:**",
    "**Verification head ref:**",
    "**Verification target path:**",
    "**Verification target digest:**",
)
_WORKER_SPECIFIC_PREFIXES = (
    *EXECUTION_DELIVERY_PREFIXES,
    "**Prompt History Path:**",
    "**Result Path:**",
    "**Audit sidecar path:**",
    "Assigned worker prompt history path:",
    "**Errors sidecar path:**",
    "**Worker Result Path:**",
    # Only antigravity carries this (worker_prompt_headers.PLAIN_FILE_WRITE_HEADER)
    # because only agy has an artifact store; unstripped it reads as divergence.
    "**File write mode:**",
    "**Model:**",
    "**Pane role:**",
    "**Provider:**",
    "**Model execution value:**",
    "**Runner:**",
    "**Host runtime:**",
    "**Host model value:**",
)
_WORKER_LABEL_SUBSTITUTE = "<analysis-worker>"


@dataclass(frozen=True)
class PromptRecord:
    worker_id: str
    dispatch_kind: str
    path: Path
    expected_model: str | None = None
    expected_delivery_mode: str | None = None
    metadata_path: Path | None = None
    expected_duty_audience: str | None = None


def validate_final_verification_initial_prompt(text: str) -> list[str]:
    """Return deterministic compact-prompt contract violations."""
    errors: list[str] = []
    _reject_literal(
        text,
        "**Coding preflight pack:**",
        "Coding preflight pack is forbidden for final-verification",
        errors,
    )
    _reject_literal(
        text,
        "**Verification diff stat:**",
        "inline Verification diff stat is forbidden; use verification-target.md",
        errors,
    )
    _reject_literal(
        text,
        "## Source / fallback paths",
        "Source / fallback paths is forbidden; use analysis-packet.md",
        errors,
    )
    for label, pattern in _COPIED_SECTION_PATTERNS:
        if pattern.search(text):
            errors.append(f"copied {label} section is forbidden; use analysis-packet.md")
    _validate_compact_target_identity(text, errors)
    packet_count = len(_PRIMARY_PACKET_RE.findall(text))
    if packet_count != 1:
        errors.append(
            "exactly one Primary analysis packet path is required "
            f"(found {packet_count})"
        )
    directive_count = text.count(_DIRECTIVE_HEADING)
    if directive_count > 1:
        errors.append("at most one Run-specific directive section is allowed")
    if directive_count == 1:
        directive_lines = _directive_nonblank_lines(text)
        if directive_lines > MAX_FINAL_VERIFICATION_DIRECTIVE_LINES:
            errors.append(
                "Run-specific directive exceeds "
                f"{MAX_FINAL_VERIFICATION_DIRECTIVE_LINES} nonblank lines "
                f"(found {directive_lines})"
            )
    body_lines = _body_nonblank_lines(text)
    if body_lines > MAX_FINAL_VERIFICATION_BODY_LINES:
        errors.append(
            f"prompt body exceeds {MAX_FINAL_VERIFICATION_BODY_LINES} "
            f"nonblank lines (found {body_lines})"
        )
    return errors


def _worker_label_pattern(worker_ids: Iterable[str]) -> re.Pattern[str] | None:
    """Match the role label the body renderer titled each compared worker with.

    Built from `analysis_worker_label`, the same function that writes the label,
    so every worker id in the comparison group is covered. Restating a
    three-provider list here is what forked the roster.
    """
    labels = sorted(
        {
            analysis_worker_label(worker_id.strip())
            for worker_id in worker_ids
            if worker_id.strip()
        },
        key=lambda label: (-len(label), label),
    )
    if not labels:
        return None
    alternation = "|".join(re.escape(label) for label in labels)
    return re.compile(rf"\b(?:{alternation})\b", re.IGNORECASE)


def normalise_analysis_prompt(text: str, *, worker_ids: Iterable[str]) -> str:
    """Remove only permitted worker identity, model, role, and path deltas.

    ``worker_ids`` is every worker in the comparison group, not just this
    prompt's own: a body that names a sibling worker must normalize to the same
    bytes in all of them, or the mention itself reads as divergence.
    """
    label = _worker_label_pattern(worker_ids)
    normalized: list[str] = []
    for line in text.replace("\r\n", "\n").replace("\r", "\n").splitlines():
        stripped = line.strip()
        prefix = next(
            (candidate for candidate in _WORKER_SPECIFIC_PREFIXES if stripped.startswith(candidate)),
            "",
        )
        if prefix:
            continue
        line = line.rstrip()
        normalized.append(
            line if label is None else label.sub(_WORKER_LABEL_SUBSTITUTE, line)
        )
    return "\n".join(normalized).strip() + "\n"


def validate_analysis_prompt_set(prompts: Mapping[str, str]) -> list[str]:
    """Require byte-identical normalized bodies for initial analysis workers."""
    if len(prompts) < 2:
        return []
    normalized = {
        worker_id: normalise_analysis_prompt(text, worker_ids=prompts.keys())
        for worker_id, text in sorted(prompts.items())
    }
    baseline_worker = next(iter(normalized))
    baseline = normalized[baseline_worker]
    different = [
        worker_id
        for worker_id, body in normalized.items()
        if body != baseline
    ]
    if not different:
        return []
    workers = ", ".join([baseline_worker, *different])
    return [f"normalized initial analysis prompts differ across workers: {workers}"]


# 재검증 출력 계약의 정본은 templates/reverify-output-contract.md 이다.
# 기존 수기 입력의 공백 차이는 허용하면서 세 조항의 의미를 검사한다.
#
# 바이트 단위로 비교하지 않는다 — 손으로 복사하는 블록이라 공백·줄바꿈이 흔들리고,
# 그 흔들림으로 디스패치를 막으면 규칙이 아니라 서식을 강제하게 된다. 세 항목이
# 각각 무엇을 요구하는지 식별하는 토큰으로 본다.
_OUTPUT_CONTRACT_HEADING = "## Output Contract"
_OUTPUT_CONTRACT_CLAUSES = (
    ("result-path write", ("Result Path", "before returning")),
    ("audit sidecar write", ("Audit sidecar path", "before returning")),
    ("no inline-only response", ("inline",)),
)


def _validate_output_contract_block(text: str) -> list[str]:
    """응답 계약 블록이 프롬프트에 실려 있는지.

    빠지면 워커가 결과를 파일로 쓰지 않고 인라인으로만 답하는 경로가 열린다 —
    그 run 은 결과 파일이 없는 채로 dispatch 만 성공한 것처럼 보인다.
    """
    if _OUTPUT_CONTRACT_HEADING not in text:
        return [
            f"`{_OUTPUT_CONTRACT_HEADING}` block is required "
            "(convergence.md §\"Required reverify output contract\")"
        ]
    if text.count(_OUTPUT_CONTRACT_HEADING) != 1:
        return ["exactly one Output Contract block is required"]
    tail = text.split(_OUTPUT_CONTRACT_HEADING, 1)[1]
    missing = [
        label
        for label, tokens in _OUTPUT_CONTRACT_CLAUSES
        if not all(token.lower() in tail.lower() for token in tokens)
    ]
    if missing:
        return [
            "output contract block is missing its "
            + ", ".join(missing)
            + " clause"
        ]
    return []


def validate_reverify_prompt(
    text: str,
    *,
    task_type: str,
    forbidden_actions: str,
    expected_model: str | None = None,
    dispatch_kind: str = "reverify-r1",
) -> list[str]:
    """Require the active phase boundary in a lightweight reverify prompt.

    ``expected_model`` is the value this dispatch will actually run. A reverify
    prompt's `**Model:**` header is hand-written per round, and a header naming
    a model the runtime does not serve does not fail here — it fails as a
    provider 400 once the worker launches, where it reads as a worker fault.
    Pass the dispatch's model so the mismatch is caught before launch.

    ``dispatch_kind`` selects the renderer whose signature the instruction must
    carry: a numbered round is rendered by `okstra convergence reverify-prompt`,
    the critic gap round (`critic-verify`) by `okstra convergence
    critic-verify-prompt`. Either way a hand-written instruction is refused.
    """
    normalized = text.replace("\r\n", "\n").replace("\r", "\n")
    errors: list[str] = _validate_model_header(normalized, expected_model)
    task_values = _header_values(normalized, TASK_TYPE_HEADER)
    if task_values != [task_type]:
        errors.append(
            f"exactly one {TASK_TYPE_HEADER} {task_type} header is required"
        )
    action_blocks = _section_values(normalized, FORBIDDEN_ACTIONS_HEADER)
    if len(action_blocks) != 1:
        errors.append("exactly one **Forbidden actions:** block is required")
    elif action_blocks[0] != forbidden_actions.strip():
        errors.append(
            "Forbidden actions block must exactly match active-run-context "
            "workflow.forbiddenActions"
        )

    expected_task_header = f"{TASK_TYPE_HEADER} {task_type}"
    boundary_position = normalized.find(expected_task_header)
    read_scope_position = normalized.find("**Read scope:**")
    if boundary_position >= 0 and (
        read_scope_position < 0 or read_scope_position > boundary_position
    ):
        errors.append("phase boundary block must follow the reverify anchor headers")
    first_heading = re.compile(r"(?m)^##\s+").search(
        normalized, _task_instructions_offset(normalized)
    )
    if (
        boundary_position >= 0
        and first_heading is not None
        and boundary_position > first_heading.start()
    ):
        errors.append("phase boundary block must precede reverify instructions")
    instructions = normalized[_task_instructions_offset(normalized):]
    if is_plan_verify_dispatch_kind(dispatch_kind):
        # 계획 본문 라운드의 정본 렌더러는 convergence 가 아니라 `okstra
        # plan-items prompt` 다. 같은 서명을 요구하면 통과할 값이 하나도 없다
        # (2026-09-09 dev-10642 implementation-planning 001: 라운드 0회).
        if PLAN_VERIFY_RENDERED_BY_LINE not in instructions:
            errors.append(
                "plan-verify instruction is not the output of `okstra plan-items "
                "prompt` (missing the `**Rendered by:**` line) — render it with "
                "`okstra plan-items prompt --run-manifest <run-manifest>` and pass "
                "that output verbatim as --instruction; hand-written plan item "
                "queues are refused"
            )
    elif dispatch_kind == CRITIC_VERIFY_DISPATCH_KIND:
        if CRITIC_VERIFY_RENDERED_BY_LINE not in instructions:
            errors.append(
                "critic-verify instruction is not the output of `okstra convergence "
                "critic-verify-prompt` (missing the `**Rendered by:**` line) — render "
                "it with `okstra convergence critic-verify-prompt --run-manifest "
                "<run-manifest> --gaps <coverage-batch.json> --worker <worker-id>` and "
                "pass that output verbatim as --instruction; hand-written gap "
                "verification instructions are refused"
            )
    elif RENDERED_BY_LINE not in instructions:
        # 손으로 쓴 지시문이 한 라운드를 버렸다(2026-09-09: `- Verdict:` 형식과 축약된
        # 근거). 렌더러의 서명 줄이 없으면 그 지시문은 렌더러 출력이 아니다.
        errors.append(
            "reverify instruction is not the output of `okstra convergence "
            "reverify-prompt` (missing the `**Rendered by:**` line) — render it with "
            "`okstra convergence reverify-prompt --run-manifest <run-manifest> "
            "--plan <round-plan.json> --worker <worker-id>` and pass that output "
            "verbatim as --instruction; hand-written reverify instructions are refused"
        )
    errors.extend(_validate_output_contract_block(normalized))
    return errors


def complete_reverify_instruction(
    body: str, *, task_type: str, forbidden_actions: str,
    model: str, output_contract: str,
) -> str:
    """수기 입력의 일치하는 경계는 유지하고 빠진 실행 계약만 생성한다."""
    prefix = []
    models = _header_values(body, MODEL_HEADER)
    if not models:
        prefix.append(f"{MODEL_HEADER} {model}")
    elif _validate_model_header(body, model):
        raise ValueError(f"Model conflicts: expected {model!r}")
    tasks = _header_values(body, TASK_TYPE_HEADER)
    if not tasks:
        prefix.append(f"{TASK_TYPE_HEADER} {task_type}")
    elif tasks != [task_type]:
        raise ValueError(f"Task Type conflicts: expected {task_type!r}, got {tasks!r}")
    actions = _section_values(body, FORBIDDEN_ACTIONS_HEADER)
    if any(_header_values(body, FORBIDDEN_ACTIONS_HEADER)):
        raise ValueError(f"Forbidden actions conflicts: put the exact block after {FORBIDDEN_ACTIONS_HEADER}")
    if not actions:
        prefix.extend([FORBIDDEN_ACTIONS_HEADER, forbidden_actions.strip()])
    elif actions != [forbidden_actions.strip()]:
        raise ValueError(f"Forbidden actions conflicts: expected {forbidden_actions.strip()!r}, got {actions!r}")
    if prefix:
        # 금지 목록 뒤의 일반 문장이 목록 값에 합쳐지지 않도록 경계를 만든다.
        if not body.lstrip().startswith(("## ", "**")):
            body = "## Instructions\n\n" + body
        body = "\n".join(prefix) + "\n\n" + body
    if task_type == "implementation-planning":
        # 저장된 금지 목록은 감사 대조용으로 유지하되 폐기된 작성 의무를 정정한다.
        body = body.rstrip() + (
            "\n\n## Planning conformance ownership\n\n"
            "Planning declares conformance commands and required dependencies; "
            "implementation writes the QA scripts, manifest, and tsconfig. "
            "Do not create those files during planning. Their absence before "
            "implementation is not a planning defect by itself. This phase "
            "ownership supersedes any legacy requirement in the frozen Forbidden "
            "actions block saying this phase MUST write those artifacts. "
            "Verify that the plan assigns their creation to implementation and "
            "provides executable commands and required dependencies.\n"
        )
    if _OUTPUT_CONTRACT_HEADING not in body:
        body = body.rstrip() + "\n\n" + output_contract.strip() + "\n"
    errors = _validate_output_contract_block(body)
    if errors:
        raise ValueError("; ".join(errors))
    return body


def _task_instructions_offset(text: str) -> int:
    """Where the lead's own instruction body starts.

    The `agent-prompt` materializer composes every prompt as anchors →
    `## Duty Contract` → `## Task Instructions`, and convergence's
    materialization gate makes that the only body a reverify dispatch may send.
    The duty section's heading is therefore always the document's first `##`,
    which left the check below with no satisfiable input: the composer writes a
    heading above anything the lead can author, so a whole-document "first
    heading" test failed every materialized prompt regardless of where the lead
    put the phase boundary.

    These rules judge what the lead wrote, so they start where the lead's text
    starts — the same region `_validate_model_header` already reads. A prompt
    without the marker is judged whole.
    """
    marker = "\n\n## Task Instructions\n\n"
    index = text.find(marker)
    return 0 if index < 0 else index + len(marker)


def _section_values(text: str, header: str) -> list[str]:
    lines = text.splitlines()
    values: list[str] = []
    for index, line in enumerate(lines):
        if line.strip() != header:
            continue
        body: list[str] = []
        for candidate in lines[index + 1:]:
            stripped = candidate.strip()
            if stripped.startswith("## ") or re.match(r"^\*\*[^*]+:\*\*", stripped):
                break
            body.append(candidate)
        values.append("\n".join(body).strip())
    return values


def validate_initial_prompt_records(
    *,
    manifest: Mapping[str, Any],
    records: Sequence[PromptRecord],
    require_evidence_ledger: bool = False,
) -> list[str]:
    """Validate prompt audiences and compare their normalized equality groups.

    Newly published prompts opt into the evidence-ledger requirement. Persisted
    historical prompts still validate under the contract they were written with.
    """
    errors: list[str] = []
    equality_groups: dict[str, dict[str, str]] = {}
    for record in records:
        plan = _resolve_record_plan(manifest, record, errors)
        if plan is None or plan.audience in {"lead-only", "reverify"}:
            continue
        try:
            text = record.path.read_text(encoding="utf-8")
        except OSError as exc:
            errors.append(
                f"{record.worker_id}: cannot read prompt {record.path}: {exc}"
            )
            continue
        errors.extend(
            f"{record.worker_id}: {error}"
            for error in _validate_prompt_for_plan(text, plan, manifest)
        )
        errors.extend(
            f"{record.worker_id}: {error}"
            for error in _validate_record_metadata(text, record)
        )
        errors.extend(
            f"{record.worker_id}: {error}"
            for error in _validate_evidence_ledger_header(
                text,
                plan,
                record.dispatch_kind,
                required=require_evidence_ledger,
            )
        )
        if plan.equality_group:
            group = equality_groups.setdefault(plan.equality_group, {})
            # 최초 프롬프트의 동등성만 비교한다. 같은 역할의 교정 호출은 최초
            # assignment 를 재사용할 수 있지만 지시문은 의도적으로 달라진다.
            group.setdefault(record.worker_id, text)
    for prompts in equality_groups.values():
        errors.extend(validate_analysis_prompt_set(prompts))
    return errors


def _validate_model_header(text: str, expected_model: str | None) -> list[str]:
    """The `**Model:** <label>, <model>` header must name the requested model.

    A caller with no resolved model passes ``None`` and the header is not
    judged — there is nothing to compare it against.
    """
    if expected_model is None:
        return []
    model = _model_value(_header_values(text, MODEL_HEADER))
    if model is None:
        return ["exactly one non-empty **Model:** <label>, <model> header is required"]
    if not _model_header_matches(model, expected_model):
        return [f"prompt model does not match requested model: {expected_model}"]
    return []


def validate_prompt_model_header(text: str, expected_model: str) -> list[str]:
    """`**Model:**` 헤더가 이 디스패치가 실제로 돌릴 모델을 가리키는가.

    재료화는 호출자가 넘긴 모델로 판정한다. 그 값이 옳아도 프롬프트 본문의
    헤더가 다르면 어긋남은 `validate-run` 에서야 드러나는데, 그때는 프롬프트가
    이미 나가고 다이제스트가 기록된 뒤라 그 run 안에서 고칠 수 없다 — 프롬프트를
    고치면 다이제스트 위조가 된다. `record-dispatch` 는 워커가 돌기 전 마지막
    지점이고, 거기서는 **기록된 배정값** 을 쓸 수 있다.

    묻는 것은 "적혀 있다면 이 모델인가" 하나다. 헤더가 없는 것은 여기서 볼
    문제가 아니다 — 어떤 audience 가 그 헤더를 실어야 하는지는 프롬프트 계약이
    audience 를 알고 판정한다. 여기서 부재까지 막으면 헤더를 요구하지 않는
    디스패치가 전부 막힌다.
    """
    task_text = text.split("\n\n## Task Instructions\n\n", 1)[-1]
    if _model_value(_header_values(task_text, MODEL_HEADER)) is None:
        return []
    return _validate_model_header(text, expected_model)


def _validate_record_metadata(text: str, record: PromptRecord) -> list[str]:
    errors = [
        *_validate_delivery_mode(
            _header_values(text, PROMPT_DELIVERY_MODE_HEADER),
            record.expected_delivery_mode,
        ),
        *_validate_model_header(text, record.expected_model),
    ]
    if record.expected_duty_audience is None:
        return errors
    if record.metadata_path is None:
        return [*errors, "agent invocation metadata path is required"]
    try:
        metadata = load_owned_object(
            record.metadata_path, artifact="agent invocation metadata"
        )
    except (OSError, UnicodeError, ValueError):
        return [*errors, "agent invocation metadata is invalid"]
    duty = metadata.get("dutyContract") if isinstance(metadata, dict) else None
    if (
        not isinstance(duty, dict)
        or metadata.get("audience") != record.expected_duty_audience
        or duty.get("id") != record.expected_duty_audience
    ):
        errors.append(
            "agent invocation duty does not match expected audience: "
            + record.expected_duty_audience
        )
    return errors


def _validate_evidence_ledger_header(
    text: str,
    plan: PromptPlan,
    dispatch_kind: str,
    *,
    required: bool,
) -> list[str]:
    values = _header_values(text, _EVIDENCE_LEDGER_HEADER_PREFIX)
    if dispatch_kind != "initial":
        return []
    if plan.audience == "report-writer":
        if values:
            return ["Evidence ledger header is forbidden for report-writer"]
        return []
    if not values:
        if required:
            return [f"exactly one `{EVIDENCE_LEDGER_HEADER}` header is required"]
        return []
    if values != ["required-v1"]:
        return [f"exactly one `{EVIDENCE_LEDGER_HEADER}` header is required"]
    return []


def _validate_delivery_mode(
    values: list[str],
    expected: str | None,
) -> list[str]:
    errors = []
    if len(values) != 1 or not values[0]:
        errors.append(
            "exactly one non-empty **Prompt Delivery Mode:** header is required"
        )
    errors.extend(
        f"unsupported Prompt Delivery Mode: {value}"
        for value in values
        if value and value not in PROMPT_DELIVERY_MODES
    )
    if (
        expected is not None
        and len(values) == 1
        and values[0]
        and values[0] != expected
    ):
        errors.append(
            f"Prompt Delivery Mode does not match requested mode: {expected}"
        )
    return errors


def _header_values(text: str, prefix: str) -> list[str]:
    return [
        line.strip()[len(prefix):].strip()
        for line in text.splitlines()
        if line.strip().startswith(prefix)
    ]


def _model_header_matches(got: str, expected: str) -> bool:
    """표시 alias 와 실행 id 를 같은 요청으로 본다.

    헤더는 `opus-5` 이고 잡의 실행 값은 `claude-opus-5` 인 경우가 있다.
    콤마 형식의 두 번째 칸과 실행 값이 문자열로 다를 뿐이면 거절하지 않는다.

    `<provider>/<model>` 표기도 받는다. 위저드가 사용자에게 모델을 보여 줄 때
    쓰는 표기가 정확히 그것(`claude/sonnet`)이라, 지시문에 그대로 옮겨 적는 것이
    자연스러운 실수가 아니라 자연스러운 선택이다. provider 는 이 헤더의 판정
    대상이 아니다 — 별도의 `**Provider:**` 헤더가 담당한다. 그래서 앞의 한 칸을
    떼고 모델끼리 비교한다. 다른 모델이면 떼고 나서도 안 맞으므로 이 관용이
    불일치를 가리지는 않는다(`claude/opus-5` 대 `sonnet` 은 그대로 실패).
    """
    if got == expected:
        return True
    got_l, expected_l = got.lower(), expected.lower()
    if got_l == expected_l:
        return True
    if expected_l.endswith("-" + got_l) or got_l.endswith("-" + expected_l):
        return True
    if "/" in got_l:
        return _model_header_matches(got_l.rsplit("/", 1)[1], expected_l)
    return False


def _model_value(values: list[str]) -> str | None:
    if len(values) != 1:
        return None
    raw = values[0].strip()
    if not raw:
        return None
    parts = [part.strip() for part in raw.rsplit(",", 1)]
    if len(parts) == 2 and all(parts):
        return parts[1]
    return raw


def _resolve_record_plan(
    manifest: Mapping[str, Any],
    record: PromptRecord,
    errors: list[str],
) -> PromptPlan | None:
    try:
        return resolve_prompt_plan_for_manifest(
            manifest=manifest,
            worker_id=record.worker_id,
            dispatch_kind=record.dispatch_kind,
        )
    except ValueError as exc:
        errors.append(f"{record.worker_id}: {exc}")
        return None


def _validate_prompt_for_plan(
    text: str,
    plan: PromptPlan,
    manifest: Mapping[str, Any],
) -> list[str]:
    errors: list[str] = []
    _require_non_empty_header(text, _WORKER_ERROR_CONTRACT_HEADER, errors)
    # The compact final-verification contract below returns early, so the plan's
    # own required headers must be checked before that branch or they never are.
    for prefix in plan.required_headers:
        _require_non_empty_header(text, prefix, errors)
    if manifest.get("taskType") == "final-verification" and plan.audience == "analysis":
        # Compact target identity re-checks headers the loop above already
        # required, so drop the repeats and report each violation once.
        compact = validate_final_verification_initial_prompt(text)
        return errors + [error for error in compact if error not in errors]
    if not plan.allow_coding_preflight:
        _reject_literal(
            text,
            "**Coding preflight pack:**",
            "Coding preflight pack is forbidden for this prompt audience",
            errors,
        )
    if plan.audience == "analysis":
        packet_count = len(_PRIMARY_PACKET_RE.findall(text))
        if packet_count != 1:
            errors.append(
                "exactly one Primary analysis packet path is required "
                f"(found {packet_count})"
            )
    return errors


def _require_non_empty_header(
    text: str,
    prefix: str,
    errors: list[str],
) -> None:
    matches = [
        line.strip()
        for line in text.splitlines()
        if line.strip().startswith(prefix)
    ]
    if len(matches) != 1 or not matches[0][len(prefix):].strip():
        errors.append(f"exactly one non-empty {prefix} header is required")


def _reject_literal(
    text: str,
    literal: str,
    message: str,
    errors: list[str],
) -> None:
    if literal in text:
        errors.append(message)


def _validate_compact_target_identity(text: str, errors: list[str]) -> None:
    stripped_lines = [line.strip() for line in text.splitlines()]
    for prefix in _REQUIRED_TARGET_PREFIXES:
        matches = [line for line in stripped_lines if line.startswith(prefix)]
        if len(matches) != 1 or not matches[0][len(prefix):].strip():
            errors.append(f"exactly one non-empty {prefix} header is required")
    scope_line = next(
        (line for line in stripped_lines if line.startswith("**Verification scope:**")),
        "",
    )
    scope = scope_line.removeprefix("**Verification scope:**").strip()
    if scope and scope not in {"whole-task", "single-stage"}:
        errors.append("Verification scope must be whole-task or single-stage")
    digest_line = next(
        (
            line
            for line in stripped_lines
            if line.startswith("**Verification target digest:**")
        ),
        "",
    )
    digest = digest_line.removeprefix("**Verification target digest:**").strip()
    if digest and not re.fullmatch(r"sha256:[0-9a-f]{64}", digest):
        errors.append("Verification target digest must be sha256:<64 lowercase hex>")


def _directive_nonblank_lines(text: str) -> int:
    section = text.split(_DIRECTIVE_HEADING, 1)[1]
    section = re.split(r"(?m)^##\s+", section, maxsplit=1)[0]
    return sum(1 for line in section.splitlines() if line.strip())


def _body_nonblank_lines(text: str) -> int:
    return sum(
        1
        for line in text.splitlines()
        if line.strip()
        and not any(line.strip().startswith(prefix) for prefix in _NON_BODY_PREFIXES)
    )