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

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

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,
    PromptPlan,
    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 = (
    "**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}"]


def validate_reverify_prompt(
    text: str,
    *,
    task_type: str,
    forbidden_actions: str,
    expected_model: str | None = None,
) -> 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.
    """
    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")
    return errors


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, {})
            group[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 []
    task_text = text.split("\n\n## Task Instructions\n\n", 1)[-1]
    model = _model_value(_header_values(task_text, MODEL_HEADER))
    if model is None:
        return ["exactly one non-empty **Model:** <label>, <model> header is required"]
    if model != expected_model:
        return [f"prompt model does not match requested model: {expected_model}"]
    return []


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_value(values: list[str]) -> str | None:
    if len(values) != 1:
        return None
    parts = [part.strip() for part in values[0].rsplit(",", 1)]
    if len(parts) != 2 or not all(parts):
        return None
    return parts[1]


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)
    )