"""Build a task-type-scoped excerpt of a versioned final-report schema.

The full schema carries the
deliverable property blocks for ALL task-types (``errorAnalysis``, the three
read-only analysis blocks, ``implementationPlanning``, ``releaseHandoff``,
``implementation``, and ``finalVerification``) plus a
``$defs`` library (~38% of the file) shared across them. A single run only
authors ONE task-type's narrative, so the report-writer worker only needs
the common structure + its own task-type's block + the ``$defs`` those
reach.

This module produces that scoped excerpt, written into the run's
instruction-set at prep time so the worker reads a smaller, path-local
file (`instruction-set/final-report-schema.json`) instead of the full
repo/installed schema (whose `schemas/...` path is not even resolvable
from inside a consumer project's task bundle).

The excerpt is ADVISORY — a reading aid for the author. Validation always
runs against the matching FULL schema, so
the excerpt never gates correctness; it only trims what the worker reads.
"""
from __future__ import annotations

import json
import re
from collections.abc import Sequence
from dataclasses import dataclass
from pathlib import Path

from .json_boundary import JsonBoundaryError, load_owned_object

from .final_report_schema import load_schema_version
from .report_contract import TASK_TYPE_DATA_PROPERTY


_ALL_PER_TYPE_PROPERTIES = frozenset(TASK_TYPE_DATA_PROPERTY.values())
_ANALYSIS_COMMON_TASK_TYPES = frozenset(
    {"project-analysis", "feature-analysis", "change-impact-analysis"}
)

_REF_RE = re.compile(r'"\$ref"\s*:\s*"#/\$defs/([^"]+)"')


def _refs_in(obj) -> set[str]:
    """Every ``#/$defs/<name>`` referenced anywhere inside ``obj``.

    ``ensure_ascii=False`` keeps non-ASCII ``$defs`` names in their literal
    form so the captured ref matches the actual ``$defs`` dict key; with the
    default escaping a non-ASCII name would dump as ``\\uXXXX`` and the closure
    would prune the def as unreachable, leaving a dangling ``$ref``.
    """
    return set(_REF_RE.findall(json.dumps(obj, ensure_ascii=False)))


def _conditional_applies(entry: dict, task_type: str) -> bool:
    """True when an ``allOf`` if/then entry is relevant to *task_type*.

    Universal entries (no ``header.taskType`` constraint) always apply;
    ``const`` entries apply on exact match; ``enum`` entries apply when
    *task_type* is a member.
    """
    constraint = (
        entry.get("if", {})
        .get("properties", {})
        .get("header", {})
        .get("properties", {})
        .get("taskType", {})
    )
    const = constraint.get("const")
    enum = constraint.get("enum")
    if const is None and enum is None:
        return True
    if const is not None:
        return const == task_type
    return task_type in (enum or [])


EXCERPT_VERSION_KEY = "x-okstraCutFromVersion"


def excerpt_cut_from_version(excerpt: dict) -> str:
    """The okstra version *excerpt* was cut from; empty when unstamped."""
    value = excerpt.get(EXCERPT_VERSION_KEY)
    return value if isinstance(value, str) else ""


@dataclass(frozen=True)
class ExcerptSkew:
    """A bundle excerpt whose contract text differs from the installed one.

    ``cut_from`` is the older okstra version stamped into the bundle;
    ``changed`` names the excerpt members that differ, as ``properties.<name>``
    / ``$defs.<name>`` / a bare top-level key.
    """

    cut_from: str
    changed: tuple[str, ...]


def _changed_members(section: str, old: object, fresh: object) -> list[str]:
    old_map = old if isinstance(old, dict) else {}
    fresh_map = fresh if isinstance(fresh, dict) else {}
    return [
        f"{section}.{name}"
        for name in sorted(set(old_map) | set(fresh_map))
        if old_map.get(name) != fresh_map.get(name)
    ]


def _changed_excerpt_members(old: dict, fresh: dict) -> tuple[str, ...]:
    """Every member of *old* that *fresh* states differently.

    The stamp is excluded by definition — it is the thing that differs whenever
    the two were cut by different runtimes, and it says nothing about what the
    author is being told to write.
    """
    changed = []
    for section in ("properties", "$defs"):
        changed.extend(_changed_members(section, old.get(section), fresh.get(section)))
    changed.extend(
        key
        for key in sorted(set(old) | set(fresh))
        if key not in ("properties", "$defs", EXCERPT_VERSION_KEY)
        and old.get(key) != fresh.get(key)
    )
    return tuple(changed)


def describe_changed(changed: Sequence[str], limit: int = 5) -> str:
    """A one-clause rendering of :attr:`ExcerptSkew.changed` for an error message."""
    if not changed:
        return "no contract member differs"
    head = ", ".join(f"`{name}`" for name in changed[:limit])
    rest = len(changed) - limit
    return f"{head} and {rest} more" if rest > 0 else head


def excerpt_contract_skew(
    excerpt_path: Path, task_type: str, installed: str
) -> ExcerptSkew | None:
    """What this task-type's contract states differently since the bundle was cut.

    ``None`` means the bundle excerpt still states the installed contract, and a
    caller that blocks on skew must let the run through. Four ways to get it:
    the file is absent or unreadable, it carries no stamp, the stamp matches, or
    — the case a version comparison gets wrong — the stamp is older but every
    contract member is identical because the release changed nothing this
    task-type authors against. A patch release that touches only dispatch or
    wizard code lands in that fourth case, and blocking an in-flight run on it
    costs a full bundle re-prep to rewrite one stamp line.

    The installed schema is loaded here rather than passed in, so a caller only
    needs the excerpt path and the manifest's task-type. An install too old to
    carry `schemas/` cannot be compared at all: that returns ``None`` too, since
    an unprovable drift is not grounds to stop a dispatch.

    The comparison used to live only in the renderer's error decorator, so it ran
    in Phase 6 — after a worker had already authored a whole report against a
    stale excerpt. The same inputs are available much earlier, and the fix
    (re-prepare the bundle) is the same either way.
    """
    if not installed:
        return None
    try:
        excerpt = load_owned_object(excerpt_path, artifact="schema excerpt")
    except JsonBoundaryError:
        return None
    if not isinstance(excerpt, dict):
        return None
    cut_from = excerpt_cut_from_version(excerpt)
    if not cut_from or cut_from == installed:
        return None
    schema_version = (
        excerpt.get("properties", {})
        .get("schemaVersion", {})
        .get("const", "2.0")
    )
    try:
        fresh = build_schema_excerpt(
            load_schema_version(str(schema_version)), task_type, installed
        )
    except Exception:  # noqa: BLE001 — an unloadable schema proves no drift
        return None
    changed = _changed_excerpt_members(excerpt, fresh)
    return ExcerptSkew(cut_from, changed) if changed else None


def bundle_excerpt_path(start: Path) -> Path | None:
    """The task bundle's schema excerpt, found by walking up from *start*."""
    for ancestor in Path(start).resolve().parents:
        candidate = ancestor / "instruction-set" / "final-report-schema.json"
        if candidate.is_file():
            return candidate
    return None


def build_schema_excerpt(schema: dict, task_type: str, cut_from_version: str = "") -> dict:
    """Return a task-type-scoped copy of *schema*.

    Drops the per-type property blocks that do not belong to *task_type*,
    the ``allOf`` conditionals that cannot fire for it, and the ``$defs``
    that become unreachable as a result (transitive closure preserves any
    def reachable from a kept property or conditional). Top-level metadata
    and ``required`` are preserved (``required`` never lists per-type
    blocks, but it is filtered defensively).

    *cut_from_version* stamps the excerpt with the runtime that produced it.
    Bundle prep and report render can be hours apart and the runtime upgrades
    itself in between, so a mismatch is what turns an opaque "additional
    property … not allowed" into "the bundle excerpt is from an older okstra".
    """
    keep_per_type = TASK_TYPE_DATA_PROPERTY.get(task_type)
    drop_props = _ALL_PER_TYPE_PROPERTIES - ({keep_per_type} if keep_per_type else set())
    if task_type not in _ANALYSIS_COMMON_TASK_TYPES:
        drop_props = drop_props | {"analysisCommon"}

    props = {
        k: v for k, v in schema.get("properties", {}).items() if k not in drop_props
    }
    all_of = [e for e in schema.get("allOf", []) if _conditional_applies(e, task_type)]

    # Reachable $defs = transitive closure of $ref from kept props + allOf.
    defs = schema.get("$defs", {})
    reachable: set[str] = set()
    work = list(_refs_in(props) | _refs_in(all_of))
    while work:
        name = work.pop()
        if name in reachable or name not in defs:
            continue
        reachable.add(name)
        work.extend(_refs_in(defs[name]))

    excerpt = {
        k: v
        for k, v in schema.items()
        if k not in ("properties", "allOf", "$defs", "required", "description")
    }
    schema_id = schema.get("$id", "the full final-report schema")
    excerpt["description"] = (
        f"Per-task-type excerpt of the okstra final-report schema, scoped to "
        f"`{task_type}`. Reading aid for the report-writer worker — validation "
        f"runs against the full schema ({schema_id})."
    )
    excerpt["properties"] = props
    excerpt["required"] = [
        r for r in schema.get("required", []) if r not in drop_props
    ]
    if all_of:
        excerpt["allOf"] = all_of
    if reachable:
        excerpt["$defs"] = {k: v for k, v in defs.items() if k in reachable}
    if cut_from_version:
        excerpt[EXCERPT_VERSION_KEY] = cut_from_version
    return excerpt