#!/usr/bin/env python3
"""Canonical schema, validator, and generator for Ralph `prd.json` backlogs.

`prd.json` is the single authoritative source for a Ralph-managed backlog.
This module defines the field shape every story must have, validates
structural and cross-story invariants (unique IDs/titles, resolvable
`dependsOn`, acyclic dependency graph, `blocks` derived correctly), and
renders the same fields into the generated Markdown/Gitea-issue view used
by `.scratch/<feature>/issues/*.md`.

Generation is one-directional: prd.json -> Markdown / Gitea payload. There
is no Markdown -> prd.json parser, so a generated artifact can never feed
`passes` (or any other field) back into the authoritative source. Every
generated Markdown document must start with the authority disclaimer
(`authority_disclaimer()`, sourced from prd.json's own `metadataSchema` when
present) so a stray edit of the generated file cannot be mistaken for a
source-of-truth change. The shared "quality gates" boilerplate is likewise
sourced from prd.json's `qualityGates`, not duplicated as a Python literal.

Example:
    python scripts/ralph_prd_schema.py validate .scratch/distributed-gguf-runtime/prd.json
    python scripts/ralph_prd_schema.py render .scratch/distributed-gguf-runtime/prd.json DGR-018
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from pathlib import Path
from typing import Any

REPO_ROOT = Path(__file__).resolve().parents[1]

AUTHORITY_DISCLAIMER = (
    "<!-- GENERATED FROM prd.json — DO NOT EDIT AS AN INDEPENDENT SOURCE. "
    "prd.json IS AUTHORITATIVE. -->"
)

EXECUTION_MODES = {"AFK", "HITL"}
EVIDENCE_CLASSES = {"model-free", "fixture", "real-model", "real-hardware", "release"}
HARDWARE_FLAGS = {"none", "optional", "required"}
UPSTREAM_FLAGS = {"yes", "no", "conditional"}
TRIAGE_VALUES = {"ready-for-agent", "ready-for-human"}

# Canonical field registry: name -> (required, expected python type(s)).
# This is the single definition of "what a story is" referenced by both
# validation and documentation; keep it in sync with any schema change.
STORY_FIELDS: dict[str, tuple[bool, tuple[type, ...]]] = {
    "id": (True, (str,)),
    "title": (True, (str,)),
    "priority": (True, (int,)),
    "milestone": (True, (str,)),
    "executionMode": (True, (str,)),
    "labels": (True, (list,)),
    "evidenceClass": (True, (str,)),
    "evidencePath": (True, (str,)),
    "hardware": (True, (str,)),
    "model": (True, (str,)),
    "upstream": (True, (str,)),
    "dependsOn": (True, (list,)),
    "triage": (True, (str,)),
    "description": (True, (str,)),
    "acceptanceCriteria": (True, (list,)),
    "passes": (True, (bool,)),
    "notes": (True, (str,)),
    "blocks": (True, (list,)),
    "completionNotes": (False, (str,)),
}

_ID_RE = re.compile(r"^[A-Z][A-Z0-9]*-\d+$")


class PrdValidationError(ValueError):
    """Raised with all accumulated validation errors, one per line."""

    def __init__(self, errors: list[str]):
        super().__init__("\n".join(errors))
        self.errors = errors


def load_prd(path: Path) -> dict[str, Any]:
    try:
        text = path.read_text(encoding="utf-8")
    except FileNotFoundError as exc:
        raise PrdValidationError([f"PRD not found: {path}"]) from exc
    try:
        data = json.loads(text)
    except json.JSONDecodeError as exc:
        raise PrdValidationError([f"Invalid JSON in {path}: {exc}"]) from exc
    if not isinstance(data, dict):
        raise PrdValidationError([f"{path}: top-level PRD document must be an object"])
    return data


def stories_of(data: dict[str, Any]) -> list[dict[str, Any]]:
    stories = data.get("userStories")
    if not isinstance(stories, list):
        return []
    return [s for s in stories if isinstance(s, dict)]


def derive_type(labels: list[str]) -> str | None:
    """Derive the story's `type` from its `type:` label, if present.

    Release-gate stories (labelled `gate:hitl`) intentionally carry no
    `type:` label; they derive the synthetic type `release-gate` instead.
    """
    types = [label.split(":", 1)[1] for label in labels if label.startswith("type:")]
    if types:
        return types[0]
    if any(label.startswith("gate:") for label in labels):
        return "release-gate"
    return None


def _label_prefix_counts(labels: list[str], prefix: str) -> int:
    return sum(1 for label in labels if label.startswith(prefix))


def validate_schema(data: dict[str, Any]) -> list[str]:
    """Structural validation: required top-level keys and per-story field shape."""
    errors: list[str] = []
    for key in ("name", "description", "branchName", "userStories"):
        if key not in data:
            errors.append(f"prd: missing top-level field '{key}'")
    stories = data.get("userStories")
    if not isinstance(stories, list):
        errors.append("prd: 'userStories' must be a list")
        return errors

    for index, story in enumerate(stories):
        label = f"userStories[{index}]"
        if not isinstance(story, dict):
            errors.append(f"{label}: story must be an object")
            continue
        sid = story.get("id", label)

        for field, (required, types) in STORY_FIELDS.items():
            if field not in story:
                if required:
                    errors.append(f"{sid}: missing required field '{field}'")
                continue
            value = story[field]
            if types == (bool,) and isinstance(value, bool):
                continue
            if not isinstance(value, types):
                errors.append(
                    f"{sid}: field '{field}' must be {types}, got {type(value).__name__}"
                )

        if isinstance(story.get("id"), str) and not _ID_RE.match(story["id"]):
            errors.append(f"{sid}: id '{story['id']}' does not match PREFIX-NUMBER convention")
        if isinstance(story.get("executionMode"), str) and story["executionMode"] not in EXECUTION_MODES:
            errors.append(f"{sid}: executionMode '{story['executionMode']}' not in {sorted(EXECUTION_MODES)}")
        if isinstance(story.get("evidenceClass"), str) and story["evidenceClass"] not in EVIDENCE_CLASSES:
            errors.append(f"{sid}: evidenceClass '{story['evidenceClass']}' not in {sorted(EVIDENCE_CLASSES)}")
        if isinstance(story.get("hardware"), str) and story["hardware"] not in HARDWARE_FLAGS:
            errors.append(f"{sid}: hardware '{story['hardware']}' not in {sorted(HARDWARE_FLAGS)}")
        if isinstance(story.get("upstream"), str) and story["upstream"] not in UPSTREAM_FLAGS:
            errors.append(f"{sid}: upstream '{story['upstream']}' not in {sorted(UPSTREAM_FLAGS)}")
        if isinstance(story.get("triage"), str) and story["triage"] not in TRIAGE_VALUES:
            errors.append(f"{sid}: triage '{story['triage']}' not in {sorted(TRIAGE_VALUES)}")

        labels = story.get("labels")
        if isinstance(labels, list):
            if not all(isinstance(label, str) for label in labels):
                errors.append(f"{sid}: labels must all be strings")
            elif _label_prefix_counts(labels, "type:") > 1:
                errors.append(f"{sid}: at most one 'type:' label allowed")
            elif _label_prefix_counts(labels, "priority:") > 1:
                errors.append(f"{sid}: at most one 'priority:' label allowed")

        depends_on = story.get("dependsOn")
        if isinstance(depends_on, list) and not all(isinstance(dep, str) for dep in depends_on):
            errors.append(f"{sid}: dependsOn entries must all be strings")

        acceptance = story.get("acceptanceCriteria")
        if isinstance(acceptance, list):
            if not acceptance:
                errors.append(f"{sid}: acceptanceCriteria must be non-empty")
            elif not all(isinstance(item, str) and item.strip() for item in acceptance):
                errors.append(f"{sid}: acceptanceCriteria entries must be non-empty strings")

    return errors


def _find_cycle(story_ids: set[str], depends_on: dict[str, list[str]]) -> list[str] | None:
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {sid: WHITE for sid in story_ids}
    path: list[str] = []

    def dfs(node: str) -> list[str] | None:
        color[node] = GRAY
        path.append(node)
        for dep in depends_on.get(node, []):
            if dep not in color:
                continue
            if color[dep] == GRAY:
                cycle_start = path.index(dep)
                return path[cycle_start:] + [dep]
            if color[dep] == WHITE:
                found = dfs(dep)
                if found:
                    return found
        path.pop()
        color[node] = BLACK
        return None

    for sid in sorted(story_ids):
        if color[sid] == WHITE:
            cycle = dfs(sid)
            if cycle:
                return cycle
    return None


def validate_semantics(data: dict[str, Any]) -> list[str]:
    """Cross-story validation: uniqueness, dependency graph, derived `blocks`."""
    errors: list[str] = []
    stories = stories_of(data)

    ids = [s.get("id") for s in stories if isinstance(s.get("id"), str)]
    seen_ids: set[str] = set()
    for sid in ids:
        if sid in seen_ids:
            errors.append(f"duplicate story id: '{sid}'")
        seen_ids.add(sid)

    titles = [s.get("title") for s in stories if isinstance(s.get("title"), str)]
    seen_titles: set[str] = set()
    for title in titles:
        if title in seen_titles:
            errors.append(f"duplicate story title: '{title}'")
        seen_titles.add(title)

    by_id = {s["id"]: s for s in stories if isinstance(s.get("id"), str)}
    depends_on: dict[str, list[str]] = {}
    for story in stories:
        sid = story.get("id")
        if not isinstance(sid, str):
            continue
        deps = story.get("dependsOn") if isinstance(story.get("dependsOn"), list) else []
        deps = [d for d in deps if isinstance(d, str)]
        depends_on[sid] = deps
        if sid in deps:
            errors.append(f"{sid}: cannot depend on itself")
        for dep in deps:
            if dep not in by_id:
                errors.append(f"{sid}: depends on unknown story '{dep}'")

    cycle = _find_cycle(set(by_id), depends_on)
    if cycle:
        errors.append("dependency cycle detected: " + " -> ".join(cycle))

    derived_blocks: dict[str, list[str]] = {sid: [] for sid in by_id}
    for sid, deps in depends_on.items():
        for dep in deps:
            if dep in derived_blocks:
                derived_blocks[dep].append(sid)
    for sid in derived_blocks:
        derived_blocks[sid] = sorted(derived_blocks[sid])

    for story in stories:
        sid = story.get("id")
        if not isinstance(sid, str):
            continue
        given = story.get("blocks") if isinstance(story.get("blocks"), list) else []
        given_sorted = sorted(d for d in given if isinstance(d, str))
        expected = derived_blocks.get(sid, [])
        if given_sorted != expected:
            errors.append(
                f"{sid}: 'blocks' does not match derived dependents "
                f"(got {given_sorted}, expected {expected})"
            )

        evidence_path = story.get("evidencePath")
        if isinstance(evidence_path, str):
            expected_path = f".scratch/distributed-gguf-runtime/evidence/{sid}/README.md"
            if evidence_path != expected_path:
                errors.append(
                    f"{sid}: evidencePath '{evidence_path}' does not match convention "
                    f"'{expected_path}'"
                )

    return errors


def validate_fresh_backlog(data: dict[str, Any]) -> list[str]:
    """Extra invariant for a freshly generated (not-yet-started) backlog.

    A fresh backlog must not claim any completion credit: every story
    starts `passes: false`. This is distinct from validating an in-flight
    backlog (use validate_schema + validate_semantics for that), because a
    real backlog legitimately accumulates `passes: true` stories over time.
    """
    errors = validate_backlog(data)
    for story in stories_of(data):
        if story.get("passes") is not False:
            errors.append(
                f"{story.get('id')}: fresh backlog requires passes=false, "
                f"got {story.get('passes')!r}"
            )
    return errors


def validate_backlog(data: dict[str, Any]) -> list[str]:
    """Full validation for an in-progress backlog: structure, semantics, and
    (when present) prd.json's self-documented metadataSchema/qualityGates."""
    return (
        validate_schema(data)
        + validate_semantics(data)
        + validate_metadata_schema_consistency(data)
    )


def validate_metadata_schema_consistency(data: dict[str, Any]) -> list[str]:
    """If prd.json documents its own `metadataSchema`/`qualityGates`, verify
    that self-description hasn't drifted from what this validator actually
    enforces. Absent entirely, this is a no-op (a minimal fixture PRD is not
    required to carry self-documentation to be structurally valid).
    """
    errors: list[str] = []
    ms = data.get("metadataSchema")
    if ms is None:
        return errors

    enum_checks = {
        "triageValues": TRIAGE_VALUES,
        "executionModeValues": EXECUTION_MODES,
        "evidenceClassValues": EVIDENCE_CLASSES,
        "hardwareValues": HARDWARE_FLAGS,
        "upstreamValues": UPSTREAM_FLAGS,
    }
    for key, expected in enum_checks.items():
        declared = ms.get(key)
        if declared is None:
            errors.append(f"metadataSchema: missing '{key}'")
            continue
        if set(declared) != expected:
            errors.append(
                f"metadataSchema.{key} {sorted(declared)} does not match "
                f"enforced values {sorted(expected)}"
            )

    required_in_code = {field for field, (required, _) in STORY_FIELDS.items() if required}
    declared_required = set(ms.get("requiredStoryFields", []))
    if declared_required != required_in_code:
        errors.append(
            f"metadataSchema.requiredStoryFields {sorted(declared_required)} does not "
            f"match code-enforced required fields {sorted(required_in_code)}"
        )

    optional_in_code = {field for field, (required, _) in STORY_FIELDS.items() if not required}
    declared_optional = set(ms.get("optionalStoryFields", []))
    if declared_optional != optional_in_code:
        errors.append(
            f"metadataSchema.optionalStoryFields {sorted(declared_optional)} does not "
            f"match code-enforced optional fields {sorted(optional_in_code)}"
        )

    if not ms.get("generatedArtifactDisclaimer"):
        errors.append("metadataSchema: missing 'generatedArtifactDisclaimer'")

    if "qualityGates" not in data:
        errors.append("prd: metadataSchema is documented but top-level 'qualityGates' is missing")
    elif not quality_gate_bullets(data):
        errors.append("qualityGates: no bullets declared under universal/native/realModelHardware/scope")

    return errors


def quality_gate_bullets(data: dict[str, Any]) -> list[str]:
    """Flatten prd.json's `qualityGates` into the bullet order used by the
    generated "Shared quality gates" Markdown section."""
    gates = data.get("qualityGates", {})
    bullets: list[str] = []
    for key in ("universal", "native", "realModelHardware", "scope"):
        bullets.extend(gates.get(key, []))
    return bullets


def authority_disclaimer(data: dict[str, Any]) -> str:
    """The exact disclaimer generated Markdown/Gitea bodies must start with.

    Sourced from prd.json's `metadataSchema.generatedArtifactDisclaimer` when
    present, falling back to the module default for minimal fixtures.
    """
    return data.get("metadataSchema", {}).get("generatedArtifactDisclaimer", AUTHORITY_DISCLAIMER)


def check_generated_markdown_authority(text: str, disclaimer: str = AUTHORITY_DISCLAIMER) -> list[str]:
    """Reject generated Markdown that fails to defer authority to prd.json."""
    errors: list[str] = []
    first_line = text.splitlines()[0] if text else ""
    if first_line != disclaimer:
        errors.append(
            "generated Markdown must start with the authority disclaimer: "
            f"expected {disclaimer!r}, got {first_line!r}"
        )
    lowered = text.lower()
    for claim in ("this file is authoritative", "this document is authoritative", "markdown is authoritative"):
        if claim in lowered:
            errors.append(f"generated Markdown claims authority over prd.json: found '{claim}'")
    return errors


def slugify(title: str) -> str:
    slug = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-")
    return re.sub(r"-{2,}", "-", slug)


def issue_number(story_id: str) -> str:
    match = re.search(r"(\d+)$", story_id)
    if not match:
        raise ValueError(f"story id '{story_id}' has no trailing number")
    return match.group(1)


def issue_filename(story: dict[str, Any]) -> str:
    return f"{issue_number(story['id'])}-{slugify(story['title'])}.md"


# Fallback bullets used only when a minimal fixture PRD has no qualityGates
# block of its own; the real backlog always sources these from prd.json via
# quality_gate_bullets() so this list is a default, not a duplicate of truth.
_DEFAULT_QUALITY_GATE_BULLETS = [
    "Targeted deterministic tests pass; Python changes also pass `python -m compileall packages tests`.",
    "`git diff --check` passes.",
    "Default tests are model-download-free, API-credit-free, and GPU-free.",
    "Evidence README records exact changed files, commands/results, limitations, and dependency handoff; no fabricated evidence or inherited completion credit.",
]


def render_issue_markdown(story: dict[str, Any], data: dict[str, Any] | None = None) -> str:
    """Render the canonical generated Markdown view of a story.

    `data` supplies the shared quality-gates bullets and authority disclaimer
    from prd.json's own `qualityGates`/`metadataSchema`; omit it only for
    fixtures that don't model those fields. This is byte-for-byte identical
    to the convention already used by `.scratch/distributed-gguf-runtime/
    issues/*.md`; see the round-trip tests in tests/test_ralph_prd_schema.py.
    """
    data = data or {}
    sid = story["id"]
    passes = story["passes"]
    deps = story.get("dependsOn", [])
    blocks = sorted(story.get("blocks", []))

    deps_text = ", ".join(f"`{d}`" for d in deps) if deps else "None"
    blocks_text = ", ".join(f"`{b}`" for b in blocks) if blocks else "None"
    labels_text = ", ".join(f"`{label}`" for label in story.get("labels", []))

    if passes:
        status_line = "completed; `passes: true`"
    else:
        status_line = f"specification only; `{story['triage']}`; `passes: false`"

    checkbox = "x" if passes else " "
    acceptance_lines = "\n".join(
        f"- [{checkbox}] {criterion}" for criterion in story["acceptanceCriteria"]
    )

    if passes:
        evidence_handoff = (
            f"Verified evidence: `{story['evidencePath']}`. Legacy evidence remains "
            "provenance only and grants no implementation completion credit."
        )
    else:
        evidence_handoff = (
            f"Write and verify `{story['evidencePath']}`. Until every criterion and "
            f"applicable gate has real evidence, this story remains `passes: false`. "
            "Legacy evidence is provenance only, not completion credit."
        )

    description = story["description"]

    gate_bullets = quality_gate_bullets(data) if "qualityGates" in data else _DEFAULT_QUALITY_GATE_BULLETS
    gates_section = "## Shared quality gates\n\n" + "\n".join(f"- {bullet}" for bullet in gate_bullets)

    lines = [
        authority_disclaimer(data),
        f"# {sid}: {story['title']}",
        "",
        f"- **Status / triage:** {status_line}",
        f"- **Execution mode:** `{story['executionMode']}`",
        f"- **Milestone:** `{story['milestone']}`",
        f"- **Dependencies:** {deps_text}",
        f"- **Blocks (derived):** {blocks_text}",
        f"- **Labels:** {labels_text}",
        f"- **Evidence class:** `{story['evidenceClass']}`",
        f"- **Hardware:** `{story['hardware']}`",
        f"- **Model:** `{story['model']}`",
        f"- **Upstream:** `{story['upstream']}`",
        "",
        "## Objective / description",
        "",
        description,
        "",
        "## Acceptance criteria",
        "",
        acceptance_lines,
        "",
        gates_section,
        "",
        "## Evidence handoff",
        "",
        evidence_handoff,
        "",
    ]
    return "\n".join(lines[:-1]) + "\n"


def to_gitea_issue_payload(story: dict[str, Any], data: dict[str, Any] | None = None) -> dict[str, Any]:
    """Structured payload matching the Gitea create-issue API shape.

    `body` is exactly `render_issue_markdown`'s output, so the payload
    carries no information that isn't losslessly derivable from prd.json.
    """
    return {
        "title": f"{story['id']}: {story['title']}",
        "body": render_issue_markdown(story, data),
        "labels": list(story.get("labels", [])),
        "milestone": story["milestone"],
    }


def _cmd_validate(args: argparse.Namespace) -> int:
    data = load_prd(args.prd)
    errors = validate_fresh_backlog(data) if args.fresh else validate_backlog(data)
    if errors:
        for error in errors:
            print(f"ERROR: {error}", file=sys.stderr)
        print(f"{len(errors)} validation error(s).", file=sys.stderr)
        return 1
    print(f"OK: {len(stories_of(data))} stories validated.")
    return 0


def _cmd_render(args: argparse.Namespace) -> int:
    data = load_prd(args.prd)
    by_id = {s["id"]: s for s in stories_of(data)}
    story = by_id.get(args.story_id)
    if story is None:
        print(f"Unknown story id: {args.story_id}", file=sys.stderr)
        return 1
    print(render_issue_markdown(story, data), end="")
    return 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    subparsers = parser.add_subparsers(dest="command", required=True)

    validate = subparsers.add_parser("validate", help="Validate a prd.json against the canonical schema")
    validate.add_argument("prd", type=Path)
    validate.add_argument("--fresh", action="store_true", help="Also require passes=false on every story")
    validate.set_defaults(func=_cmd_validate)

    render = subparsers.add_parser("render", help="Render the generated Markdown view of one story")
    render.add_argument("prd", type=Path)
    render.add_argument("story_id")
    render.set_defaults(func=_cmd_render)

    return parser


def main(argv: list[str]) -> int:
    parser = build_parser()
    args = parser.parse_args(argv[1:])
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))