"""Validate a review partition: one owner per changed file, no file unowned.

`skills/herdr-foreman/references/team-operation.md` Review Before PR describes one reviewer per
round. A reviewer roaming an unbounded surface that reports no findings has not
established that the surface is clean -- only that this pass happened not to
reach a defect, which is why a twenty-round delivery ran three, three, two,
one, one, one, two, two, one, one blocking findings and never converged (#409).

A partition supplies the missing termination condition: a slice is saturated
when its reviewer reports clean at the current tip, and a change is reviewed
when every slice is saturated at one tip. That only holds while the partition
is DISJOINT and EXHAUSTIVE over the change. A gap is indistinguishable from a
clean slice in the result, and an overlap makes two verdicts answer for one
file while neither owns it.

This module decides that, and nothing else: it never chooses slices, never
assigns workers, and never reads a report. The foreman writes the partition; the
planner seats it; this says whether it can carry a verdict.

Contract:

* `load_partition(path)` reads the partition document.
* `validate(changed, partition)` is pure over a changed-path set and that
  document.
* `run_command(args, runner=...)` collects the changed set through `runner`,
  a callable taking a git argument list and returning its stdout, and returns
  the `(payload, failure)` pair every command returns.
* The payload lists each slice with the paths it owns. A partition that cannot
  carry a verdict raises `UsageError` naming every unowned path and every
  overlap with the slices that claim it, so the round refuses before a worker
  is spent.
"""

import fnmatch
import hashlib
import json
import re
import shlex
from pathlib import Path

from . import runnable
from .errors import UsageError
from .renderable import renderable
from .tiers import SEAT_SEPARATOR, SEATABLE_ROLES, SLICE_NAME
from .triggers import git_runner, parse_name_status

#: The partition document's own version, so a later shape change is auditable
#: (`rules/stateful-artifacts.md` Migration Policy).
PARTITION_SCHEMA_VERSION = 1
#: A `validate-partition` RESULT's own version. Version 2 adds `proof`, the
#: repo, base and head the slices were proven over (#460). A version-1 result
#: carries no proof and is refused at `plan`: re-validate.
RESULT_SCHEMA_VERSION = 2
#: A full commit id: SHA-1, or SHA-256 in a repository using that object
#: format, the shape the task ledger accepts (`recovery.SHA_RE`).
FULL_SHA = re.compile(r"[0-9a-f]{40}(?:[0-9a-f]{24})?")

COMMANDS = frozenset({"validate-partition"})

def unsafe_glob(glob):
    """True when a seat's rendered brief cannot carry `glob` safely.

    The glob renders inside a code span, so a backtick is refused alongside
    the characters `renderable.UNRENDERABLE_CATEGORIES` names.
    """
    return not renderable(glob, code_span=True)

#: The responsibilities a partition seats, which are the seatable roles
#: `tiers.SEATABLE_ROLES` names.
PARTITION_ROLES = SEATABLE_ROLES




def load_partition(path):
    """Read the partition document the foreman wrote for this round."""
    if not path:
        raise UsageError("Pass --partition naming the round's review partition; a multi-seat review round has no partition to seat without one.", {})
    try:
        document = json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise UsageError("Cannot read the partition at {}: {}. Write a JSON object with schema_version and slices.".format(path, exc), {"path": str(path)}) from None
    return validate_document(document, str(path))


def validate_document(document, path):
    """The document's own contract, shared by the raw and validated loaders."""
    if not isinstance(document, dict) or document.get("schema_version") != PARTITION_SCHEMA_VERSION:
        raise UsageError("The partition must be a JSON object at schema_version {}.".format(PARTITION_SCHEMA_VERSION), {"path": str(path)})
    partition_role(document)
    unknown = set(document) - {"schema_version", "role", "slices"}
    if unknown:
        raise UsageError("The partition carries unknown field(s) {}; it holds schema_version, an optional role, and slices.".format(", ".join(sorted(unknown))), {"path": str(path)})
    slices = document.get("slices")
    if not isinstance(slices, list) or len(slices) < 2:
        raise UsageError("A partition names at least two slices; a single-seat round needs none.", {"path": str(path)})
    seen = set()
    for entry in slices:
        if (not isinstance(entry, dict) or set(entry) != {"name", "paths"}
                or not isinstance(entry["name"], str) or not entry["name"].strip()):
            raise UsageError("Each slice is an object with a non-empty name and a paths array.", {"path": str(path)})
        if not SLICE_NAME.fullmatch(entry["name"]):
            raise UsageError(
                "Slice name {!r} cannot address its seat: name it with letters, digits, underscores, dots or hyphens, starting with a letter or digit. A seat is a CLI key, and a name carrying {!r}, a separator or whitespace does not read back.".format(
                    entry["name"], SEAT_SEPARATOR),
                {"path": str(path)})
        if entry["name"] in seen:
            raise UsageError("Slice name {!r} appears twice; each seat owns one named slice.".format(entry["name"]), {"path": str(path)})
        seen.add(entry["name"])
        patterns = entry["paths"]
        if (not isinstance(patterns, list) or not patterns
                or any(not isinstance(item, str) or not item.strip() for item in patterns)):
            raise UsageError("Slice {!r} needs a non-empty array of path globs.".format(entry["name"]), {"path": str(path)})
        # A glob is rendered verbatim into the seat's brief, where a backtick
        # or a control character could close the code span and append
        # instructions of its own. Refused here so an accepted partition always
        # composes, rather than failing a round later (#434).
        if any(unsafe_glob(item) for item in patterns):
            raise UsageError(
                "Slice {!r} has a path glob carrying a backtick or a control character; a glob needs "
                "neither, and each one is rendered into its seat's brief.".format(entry["name"]),
                {"path": str(path)})
    return document


def seat_name(role, slice_name):
    """The planner's name for the seat that owns `slice_name`."""
    return role + SEAT_SEPARATOR + slice_name


def seats_for(partition, role):
    """`{seat_name: role}` for every slice, in declaration order.

    The planner is role-keyed throughout, so several seats of one role reach it
    as distinct names mapped back to the responsibility they fill (#409).
    """
    return {seat_name(role, entry["name"]): role for entry in partition["slices"]}


def seat_digest(seat, paths, proof=None):
    """A short digest over ONE seat and the paths it owns.

    Per seat, not per round: a round-level digest is identical in every seat's
    brief, so swapping two seats' briefs passes a check that only asks whether
    the digest appears. Binding the seat's own name and globs makes each brief
    answerable for its own boundary (#453).
    """
    # With a proof, the digest also binds the diff the boundary was proven over,
    # so a brief dispatched for one tip cannot pass a gate at another (#460).
    bound = [seat, list(paths)] if proof is None else [seat, list(paths), proof]
    canonical = json.dumps(bound, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:12]


def slice_scope(seat, paths, digest):
    """The canonical scope block a seated brief must carry, verbatim.

    One sentence, not three facts a brief may scatter: a hand-written brief
    that mentions the slice name, a path and the digest while directing a
    whole-repository pass satisfies three substring checks and still dispatches
    a full-surface verdict as a slice one. Requiring this block requires the
    restrictions with it (#453).

    `compose-briefs.sh` renders the same string for the brief; its
    `slice_scope()` and this function are pinned together by
    `tests/test_slice_scope_parity.py`.
    """
    listed = ", ".join("`{}`".format(glob) for glob in paths)
    return (
        "Your slice this round is **{}**, and it owns {}. That slice is your "
        "whole surface: a full pass covers all of it and nothing beyond it. An "
        "observation outside your slice goes in a separate section of your "
        "report and forms no part of your verdict. (Partition {}.)".format(
            seat.split(SEAT_SEPARATOR, 1)[1], listed, digest))


def slice_digest(seat_paths, proof=None):
    """A short digest over the accepted `{seat: [glob, ...]}` map.

    Carried from the plan into each seat's brief and checked at dispatch, so a
    boundary edited between `validate-partition` and the send is refused rather
    than dispatched (#453). Twelve hex characters: enough to catch an edit,
    short enough to sit in a brief a worker reads.
    """
    boundary = {seat: list(paths) for seat, paths in sorted(seat_paths.items())}
    canonical = json.dumps(boundary if proof is None else {"slices": boundary, "proof": proof},
                           sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:12]


def load_validated(path):
    """Read a `validate-partition` RESULT, not the document it was built from.

    The result is the proof: it carries the slices with their RESOLVED paths
    and the `changed` set they were checked against, so a plan seated from it
    is seated from a partition proven disjoint and exhaustive over this round's
    change. `load_partition` alone reads shape and cannot check ownership —
    `plan` has no repo, base or head to check against (#453).

    Returns `(partition, proof)` from one read, so the slices and the proof a
    plan binds cannot come from two versions of the file (#460).
    """
    if not path:
        raise UsageError(
            "Pass --partition naming the output of `{}`; a "
            "multi-seat round seats from the checked result, never from the document.".format(
                runnable.command("validate-partition")),
            {})
    try:
        document = json.loads(Path(path).read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise UsageError(
            "Cannot read the validated partition at {}: {}. Pass the JSON "
            "`{}` wrote.".format(path, exc, runnable.command("validate-partition")), {"path": str(path)}) from None
    if not isinstance(document, dict) or "changed" not in document:
        raise UsageError(
            "The partition at {} carries no `changed` set, so it is the document rather "
            "than a checked result: run `{}` and pass its output.".format(
                path, runnable.command("validate-partition --repo <path> --base <sha> [--head <sha>] --partition {}".format(
                    shlex.quote(str(path))))),
            {"path": str(path)})
    changed = document["changed"]
    if not isinstance(changed, list) or not changed or any(
            not isinstance(item, str) or not item.strip() for item in changed):
        raise UsageError(
            "The validated partition's `changed` must list the round's changed paths; "
            "re-run `{}` rather than editing its result.".format(runnable.command("validate-partition")),
            {"path": str(path)})
    # The document's own contract, then its OWNERSHIP re-derived against the
    # `changed` set it carries. Reading the result's shape and trusting its
    # verdict would accept an edited result: overlapping slices and an
    # uncovered file pass a shape check, and the round seats against them.
    # Re-deriving costs nothing and re-proves the property rather than taking
    # the artifact's word for it (#453).
    if document.get("schema_version") != RESULT_SCHEMA_VERSION:
        raise UsageError(
            "The validated partition at {} is result schema {}; this build plans from schema {}, which records "
            "the proof. Re-run `{}` and plan from its output.".format(
                path, document.get("schema_version"), RESULT_SCHEMA_VERSION, runnable.command("validate-partition")), {"path": str(path)})
    proof = check_proof(document.get("proof"), "The validated partition at {}".format(path))
    inner = {key: value for key, value in document.items() if key not in ("changed", "proof")}
    inner["schema_version"] = PARTITION_SCHEMA_VERSION
    accepted = validate_document(inner, str(path))
    validate_resolved(set(changed), accepted, str(path))
    return accepted, proof


def validate_resolved(changed, partition, source):
    """Re-prove a VALIDATED result disjoint and exhaustive over `changed`.

    Set membership, not `fnmatch`. A result's `slices[].paths` carries the
    resolved changed files `validate()` assigned, not the globs the document
    it read carried, and a resolved name is a literal: re-matching it as a
    pattern reads `src/api/[x].py` as a character class and reports the file
    it names unowned, rejecting a valid result (#453).
    """
    owners_of = {}
    for entry in partition["slices"]:
        for resolved in entry["paths"]:
            owners_of.setdefault(resolved, []).append(entry["name"])
    unowned = sorted(changed - set(owners_of))
    overlaps = sorted(path for path, names in owners_of.items() if len(names) > 1)
    stray = sorted(set(owners_of) - changed)
    empty = sorted(entry["name"] for entry in partition["slices"] if not entry["paths"])
    if unowned or overlaps or stray or empty:
        parts = []
        if unowned:
            parts.append("leaves {} changed path(s) unowned, starting with {}".format(
                len(unowned), unowned[0]))
        if overlaps:
            parts.append("gives {} path(s) more than one owner, starting with {}".format(
                len(overlaps), overlaps[0]))
        if stray:
            parts.append("assigns {} path(s) the `changed` set does not carry, starting "
                         "with {}".format(len(stray), stray[0]))
        if empty:
            parts.append("leaves slice(s) {} owning nothing".format(", ".join(empty)))
        raise UsageError(
            "The validated partition at {} does not cover its own `changed` set: it {}. "
            "Re-run `{}` rather than editing its result.".format(
                source, "; and it ".join(parts), runnable.command("validate-partition")),
            {"unowned": unowned, "overlaps": overlaps, "stray": stray, "empty": empty})
    return partition


def seat_paths(partition, role):
    """`{seat_name: [path, ...]}` — the paths each seat's slice owns.

    Resolved literals, not patterns: they come from a `validate-partition`
    RESULT, whose `slices[].paths` carry the changed files `validate()`
    assigned rather than the globs the document it read carried. The composer
    requires a seat's paths and reads no partition document, so the plan
    carries them out of the validated partition rather than leaving the foreman
    to copy the boundary by hand (#434).
    """
    return {seat_name(role, entry["name"]): list(entry["paths"]) for entry in partition["slices"]}


def slice_of(seat):
    """The slice a seat owns, or None for a plain role name."""
    if not isinstance(seat, str) or SEAT_SEPARATOR not in seat:
        return None
    return seat.split(SEAT_SEPARATOR, 1)[1]


def partition_role(partition):
    """The role the partition seats; `reviewer` unless the document says."""
    role = partition.get("role", "reviewer")
    # A JSON document can name an unhashable role. The membership test would
    # raise TypeError past every caller expecting this module's UsageError.
    if not isinstance(role, str) or role not in PARTITION_ROLES:
        raise UsageError(
            "A partition seats {}; every other responsibility carries per-task gates one seat owns.".format(
                " or ".join(sorted(PARTITION_ROLES))),
            {"role": role})
    return role


def owners(path, slices):
    """The slice names whose globs match `path`, in declaration order."""
    matched = []
    for entry in slices:
        if any(fnmatch.fnmatchcase(path, pattern) for pattern in entry["paths"]):
            matched.append(entry["name"])
    return matched


def validate(changed, partition):
    """Decide whether `partition` can carry a verdict over `changed`.

    Returns the per-slice ownership payload. Raises UsageError naming every
    unowned path and every overlap: a gap reads as a clean slice, and an
    overlap leaves a file two verdicts and no owner.
    """
    slices = partition["slices"]
    assignment = {entry["name"]: [] for entry in slices}
    unowned = []
    overlaps = []
    for path in sorted(changed):
        matched = owners(path, slices)
        if not matched:
            unowned.append(path)
        elif len(matched) > 1:
            overlaps.append({"path": path, "slices": matched})
        else:
            assignment[matched[0]].append(path)
    # A slice party to an overlap owns nothing yet, but its emptiness is that
    # overlap's doing and naming it again would send the reader after the wrong
    # fix.
    contested = {name for row in overlaps for name in row["slices"]}
    empty = sorted(name for name, paths in assignment.items()
                   if not paths and name not in contested)
    # Every problem in ONE run. Raising on the first class would hide an
    # overlap behind a gap and cost a round per class to find them all.
    if unowned or overlaps or empty:
        parts = []
        if unowned:
            parts.append("leaves {} changed path(s) unowned, starting with {} (a gap is indistinguishable from a clean slice in the result)".format(
                len(unowned), ", ".join(unowned[:5])))
        if overlaps:
            first = overlaps[0]
            parts.append("gives {} changed path(s) more than one owner, starting with {} claimed by {} (two verdicts over one file leave it owned by neither)".format(
                len(overlaps), first["path"], ", ".join(first["slices"])))
        if empty:
            parts.append("has slice(s) {} owning no changed path (a seat with nothing to review is a worker spent for no verdict)".format(
                ", ".join(empty)))
        raise UsageError(
            "The partition cannot carry a verdict: it {}. Extend, narrow or drop the slices named in the details and re-run.".format(
                "; and it ".join(parts)),
            {"unowned": unowned, "overlaps": overlaps, "empty": empty},
        )
    return {"schema_version": PARTITION_SCHEMA_VERSION,
            "role": partition_role(partition),
            "slices": [{"name": entry["name"], "paths": assignment[entry["name"]]} for entry in slices],
            "changed": sorted(changed)}


def register_commands(sub, common):
    parser = sub.add_parser(
        "validate-partition", parents=[common],
        help="Check that a review partition is disjoint and exhaustive over the round's changed paths.",
    )
    parser.add_argument("--repo", required=True, metavar="PATH", help="The repository whose diff the partition covers.")
    parser.add_argument("--base", required=True, metavar="REV", help="The revision the round started from.")
    parser.add_argument("--head", metavar="REV", help="The pushed head; omit to read the working tree.")
    parser.add_argument("--partition", required=True, metavar="FILE", help="The round's partition document.")
    check = sub.add_parser(
        "verify-partition", parents=[common],
        help="At the review gate, confirm a partitioned plan still covers exactly the diff at the tip under review.",
    )
    check.add_argument("--plan", required=True, metavar="FILE", help="The plan `plan --partition` wrote.")
    check.add_argument("--repo", required=True, metavar="PATH", help="The repository the partition covers.")
    check.add_argument("--head", required=True, metavar="REV", help="The tip whose review is being accepted.")
    check.add_argument("--task", required=True, help="The task whose recorded base and dispatched seats the plan covers.")


def _revision(run, rev):
    """The full commit a revision names in the repository, or a refusal naming it."""
    # `--revs-only` prints nothing for a name the repository does not hold and
    # exits 0, so the refusal below is reached; `--verify` exits 128 and the
    # runner would raise git's bare "Needed a single revision" first. A broken
    # repository still fails in the runner with git's own diagnostic.
    resolved = run(["rev-parse", "--revs-only", "--end-of-options", rev + "^{commit}"]).strip()
    if not FULL_SHA.fullmatch(resolved):
        raise UsageError("{!r} does not name a commit in the repository; pass a revision it holds.".format(rev), {"rev": rev})
    return resolved


def run_command(args, runner=None):
    """Validate this round's partition against the paths its diff changed, and stamp what it was proven against."""
    partition = load_partition(args.partition)
    run = runner if runner is not None else git_runner(args.repo)
    head = getattr(args, "head", None)
    # Resolved first, so an unknown revision is named rather than failing the diff.
    base_sha = _revision(run, args.base)
    head_sha = _revision(run, head) if head else None
    # Two-dot: the diff between the recorded base and the head themselves,
    # never from their merge base, so a base that is not an ancestor of the
    # head still counts every path that differs (#534).
    span = [base_sha + ".." + head_sha] if head_sha else [base_sha]
    changes = parse_name_status(run(["diff", "--no-renames", *span, "--name-status", "-z"]))
    result = validate(set(changes), partition)
    result["schema_version"] = RESULT_SCHEMA_VERSION
    # The proof names what the partition was checked against, so the review
    # gate can confirm the plan still covers the diff at the tip it accepts (#460).
    result["proof"] = {"repo": str(Path(args.repo).expanduser().resolve()), "base": base_sha, "head": head_sha}
    return result, None


def check_proof(proof, where):
    """A proof's exact shape: an absolute repo, a full base commit, and a full head commit or null."""
    valid = (isinstance(proof, dict) and set(proof) == {"repo", "base", "head"}
             and isinstance(proof["repo"], str) and Path(proof["repo"]).is_absolute()
             and isinstance(proof["base"], str) and bool(FULL_SHA.fullmatch(proof["base"]))
             and (proof["head"] is None or isinstance(proof["head"], str) and bool(FULL_SHA.fullmatch(proof["head"]))))
    if not valid:
        raise UsageError("{} carries no usable proof of what the partition was checked against; re-run "
                         "`{}` with this build and plan from its output.".format(where, runnable.command("validate-partition")),
                         {"where": where})
    return proof



def check_slice_paths(slice_paths, where):
    """`{seat: [path, ...]}` with string seats and non-empty lists of non-empty strings."""
    if (not isinstance(slice_paths, dict) or not slice_paths
            or any(not isinstance(seat, str) or not isinstance(paths, list) or not paths
                   or any(not isinstance(path, str) or not path.strip() for path in paths)
                   for seat, paths in slice_paths.items())):
        raise UsageError("{} has no usable slice_paths; plan the round with `{}` "
                         "rather than editing the plan.".format(where, runnable.command("plan --partition <validate-partition output>")),
                         {"where": where})
    return slice_paths


def verify(plan, repo, head, task_base, runner=None):
    """Refuse a plan's partition unless it covers exactly the task's diff at the tip under review.

    `plan` is the loaded plan document; the caller has already checked each
    seat's dispatched brief against its seat digest. Every field is checked
    before use, and the proof is bound to this repo and the task's recorded
    base, so an edited plan is refused rather than verified.
    """
    where = "The plan"
    slice_paths = check_slice_paths(plan.get("slice_paths") if isinstance(plan, dict) else None, where)
    proof = check_proof(plan.get("partition_proof"), where)
    if plan.get("slice_digest") != slice_digest(slice_paths, proof) or plan.get("seat_digests") != {
            seat: seat_digest(seat, paths, proof) for seat, paths in slice_paths.items()}:
        raise UsageError("The plan's slice_paths or partition_proof no longer match its slice_digest and seat_digests, "
                         "so its boundary was edited after planning. Replan from the `{}` result.".format(
                             runnable.command("validate-partition")), {})
    if proof["head"] is None:
        raise UsageError("The partition was validated against the working tree, not a pushed head, so no tip can be "
                         "checked against it. Re-run `{}` at the pushed tip, replan, and "
                         "re-dispatch the slices.".format(runnable.command("validate-partition --head <sha>")), {})
    here = str(Path(repo).expanduser().resolve())
    if proof["repo"] != here:
        raise UsageError("The partition was proven in {}, not {}; verify it against the repository it covers.".format(
            proof["repo"], here), {"proven": proof["repo"], "repo": here})
    if proof["base"] != task_base:
        raise UsageError("The partition was proven from base {}, but the task's recorded base is {}; a later base "
                         "hides part of the task's change. Re-validate from the recorded base and replan.".format(
                             proof["base"], task_base), {"proven": proof["base"], "task_base": task_base})
    run = runner if runner is not None else git_runner(repo)
    tip = _revision(run, head)
    if tip != proof["head"]:
        raise UsageError("The partition was proven at {}, but the tip under review is {}. A new push can change the diff "
                         "the slices cover: re-run `{}` at the tip, replan, and review the slices again."
                         .format(proof["head"], tip, runnable.command("validate-partition")), {"proven": proof["head"], "tip": tip})
    changed = set(parse_name_status(run(["diff", "--no-renames", proof["base"] + ".." + tip, "--name-status", "-z"])))
    owners_of = {}
    for seat, paths in slice_paths.items():
        for path in paths:
            owners_of.setdefault(path, []).append(seat)
    unowned = sorted(changed - set(owners_of))
    stale = sorted(set(owners_of) - changed)
    shared = sorted(path for path, seats in owners_of.items() if len(seats) > 1)
    if unowned or stale or shared:
        raise UsageError("The plan's slices do not cover exactly the diff {}..{}: {} unowned, {} no longer changed, {} "
                         "owned twice. Re-run `{}` at the tip and replan.".format(
                             proof["base"][:12], tip[:12], len(unowned), len(stale), len(shared),
                             runnable.command("validate-partition")),
                         {"unowned": unowned, "stale": stale, "shared": shared})
    return {"schema_version": RESULT_SCHEMA_VERSION, "verified": True, "base": proof["base"], "head": tip,
            "seats": sorted(slice_paths), "changed": len(changed)}