"""Round-0 grouping provenance: does every cited source item actually exist?

The lead mints its own finding IDs, but each ``sourceItems[].itemId`` and
``discoveredBy.<worker>.itemId`` is copied verbatim from a worker's result. A
grouping citing ``codex-worker:F-007`` when codex never wrote F-007 is a
fabricated provenance link, and the shapes it arrives in are all
transcription damage: a descriptive suffix (``F-105 FontLookupPort row``), two
ids comma-joined into one field, a section title, or a reverify verdict cited
as if it were a discovery.

This module is the single implementation. ``okstra convergence seed`` runs it
when the grouping is handed over — while the lead can still fix it — and
``validators/validate-run.py`` runs it again over the finished run, so a
grouping written out of band is still caught.
"""
from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Mapping

from okstra_ctl.convergence_engine import grouped_input_digest
from okstra_ctl.json_boundary import (
    load_owned_object,
)

GROUPS_BASENAME_RE = re.compile(
    r"^convergence-groups-(?P<suffix>[a-z][a-z-]*?-\d{3})\.json$"
)


def _nonempty_string(value: object) -> bool:
    return isinstance(value, str) and bool(value.strip())


def group_claims(document: object) -> list[tuple[str, str, str]] | None:
    """Extract `(findingId, worker, itemId)` provenance claims from a grouping.

    Returns ``None`` when the document does not match the convergence-groups
    schema shape closely enough to read its provenance fields safely — the
    caller refuses to judge such a file. Otherwise returns the deduplicated
    claims drawn from every group's ``sourceItems[]`` and ``discoveredBy`` map.
    """
    if not isinstance(document, dict) or not isinstance(document.get("groups"), list):
        return None
    claims: list[tuple[str, str, str]] = []
    seen: set[tuple[str, str, str]] = set()
    for group in document["groups"]:
        if not isinstance(group, dict):
            return None
        finding_id = group.get("findingId")
        source_items = group.get("sourceItems")
        discovered_by = group.get("discoveredBy")
        if (
            not _nonempty_string(finding_id)
            or not isinstance(source_items, list)
            or not isinstance(discovered_by, dict)
        ):
            return None
        pairs: list[tuple[object, object]] = []
        for item in source_items:
            if not isinstance(item, dict):
                return None
            pairs.append((item.get("worker"), item.get("itemId")))
        for worker, discovery in discovered_by.items():
            if not isinstance(discovery, dict):
                return None
            pairs.append((worker, discovery.get("itemId")))
        for worker, item_id in pairs:
            if not _nonempty_string(worker) or not _nonempty_string(item_id):
                return None
            claim = (finding_id, worker, item_id)
            if claim not in seen:
                seen.add(claim)
                claims.append(claim)
    return claims


def groups_digest_ok(state_dir: Path, suffix: str, document: dict) -> bool:
    """False only when a recorded groupsDigest exists and no longer matches.

    ``okstra convergence seed`` pins ``grouped_input_digest`` of the grouping
    into the working state. When that record is present and disagrees with the
    on-disk grouping, the file was tampered with or written out of band — not
    evidence of fabrication — so the caller refuses to judge it. A missing
    working state or missing digest means judge the file as parsed.
    """
    work_path = state_dir / f"convergence-work-{suffix}.json"
    try:
        work = load_owned_object(work_path, artifact="convergence work state")
    except (OSError, ValueError):
        return True
    recorded = work.get("groupsDigest") if isinstance(work, dict) else None
    if not _nonempty_string(recorded):
        return True
    return grouped_input_digest(document) == recorded


def read_canonical_worker_result(
    worker_results_dir: Path, worker: str, suffix: str
) -> str | None:
    """Return the `<worker>-worker-<suffix>.md` text, or ``None`` when it is not
    a resolvable file directly inside ``worker-results/`` (missing, unreadable,
    or a worker slug that is not a single path component).

    The literal `-worker` marker comes from the writer,
    `worker_artifacts.artifacts_from_context`, which names results
    `f"{worker_id}-worker{result_suffix}.md"` from a bare worker id. Dropping it
    here made every lookup miss once the roster moved to bare ids, and a missed
    lookup is silent by design (see `provenance_errors`), so the guard stopped
    firing without any signal.
    """
    candidate = worker_results_dir / f"{worker}-worker-{suffix}.md"
    try:
        if candidate.resolve().parent != worker_results_dir.resolve():
            return None
        return candidate.read_text(encoding="utf-8")
    except (OSError, ValueError):
        return None


def id_occurs_wordbounded(text: str, item_id: str) -> bool:
    """True when `item_id` occurs as a literal not flanked by another
    identifier char, so `F-7` matches neither `F-70` nor `xF-7`."""
    pattern = r"(?<![0-9A-Za-z_-])" + re.escape(item_id) + r"(?![0-9A-Za-z_-])"
    return re.search(pattern, text) is not None


def provenance_errors(
    document: object,
    *,
    worker_results_dir: Path,
    suffix: str,
    groups_label: str,
) -> list[str]:
    """Every claim in ``document`` whose cited ID is absent from its worker's
    result file. An unjudgeable document yields no errors — a caller that
    blocks on these must not block on a shape it could not read."""
    claims = group_claims(document)
    if claims is None:
        return []
    contents: dict[str, str | None] = {}
    errors: list[str] = []
    for finding_id, worker, item_id in claims:
        if worker not in contents:
            contents[worker] = read_canonical_worker_result(
                worker_results_dir, worker, suffix
            )
        text = contents[worker]
        if text is None or id_occurs_wordbounded(text, item_id):
            continue
        errors.append(
            f"convergence groups `{groups_label}` group {finding_id} "
            f"cites source item `{worker}:{item_id}`, but that ID does not "
            f"occur in the worker's result file `{worker}-worker-{suffix}.md` — a "
            "grouping may not invent a provenance link to an item the "
            "worker never reported."
        )
    return errors


def worker_result_suffix(run_dir: Path, document: object) -> str | None:
    """`<task-type>-<seq>` naming the worker results a grouping cites.

    A grouping's own filename carries the **state** sequence, but worker results
    are named with the **workerResults** sequence — separate counters, both
    written by `render.render_run_manifest`. They happened to agree on early
    runs, which is why reusing the groups suffix went unnoticed.

    The link is the grouping's `runManifestPath`, required by
    `convergence-groups-v2.0.schema.json`. Only its basename is used: the
    grouping and its run manifest live in the same run directory, so this does
    not need a project root.
    """
    if not isinstance(document, Mapping):
        return None
    recorded = document.get("runManifestPath")
    if not _nonempty_string(recorded):
        return None
    manifest_path = Path(run_dir) / "manifests" / Path(str(recorded)).name
    try:
        manifest = load_owned_object(manifest_path, artifact="run manifest")
    except (OSError, ValueError):
        return None
    if not isinstance(manifest, Mapping):
        return None
    task_type = manifest.get("taskType")
    sequences = manifest.get("runSequencesByCategory")
    seq = sequences.get("workerResults") if isinstance(sequences, Mapping) else None
    if not _nonempty_string(task_type) or not _nonempty_string(seq):
        return None
    return f"{task_type}-{seq}"


def run_dir_provenance_errors(run_dir: Path) -> list[str]:
    """Provenance errors across every grouping persisted under ``run_dir``.

    Unjudgeable input is skipped silently: no groups file, unreadable/malformed
    JSON, a groups file that fails the schema shape, a missing worker result
    file (provider-unavailable substitution is legitimate), or a recorded
    groupsDigest that no longer matches.
    """
    state_dir = Path(run_dir) / "state"
    worker_results_dir = Path(run_dir) / "worker-results"
    if not state_dir.is_dir() or not worker_results_dir.is_dir():
        return []
    errors: list[str] = []
    for groups_path in sorted(state_dir.glob("convergence-groups-*.json")):
        match = GROUPS_BASENAME_RE.match(groups_path.name)
        if match is None:
            continue
        suffix = match.group("suffix")
        try:
            document = load_owned_object(
                groups_path, artifact="convergence groups input"
            )
        except (OSError, ValueError):
            continue
        if not isinstance(document, dict):
            continue
        if not groups_digest_ok(state_dir, suffix, document):
            continue
        result_suffix = worker_result_suffix(Path(run_dir), document)
        if result_suffix is None:
            continue
        errors.extend(
            provenance_errors(
                document,
                worker_results_dir=worker_results_dir,
                suffix=result_suffix,
                groups_label=groups_path.name,
            )
        )
    return errors


# 변이 감사가 결과를 폐기하고 재시도까지 막는 terminal 상태. 이 상태의 결과 파일은
# 디스크에 남아 있어 파일 존재만 보는 provenance 검사는 통과시킨다.
DISCARDED_ATTEMPT_STATUSES = frozenset(
    {"contract-failed-unattributed", "mutation-present-unresolved"}
)


def discarded_workers(document: object, execution_manifest: object) -> dict[str, str]:
    """``{workerId: "<invocationRef>#<attempt> <status>"}`` — 초기 attempt 가 전부
    폐기돼 채택할 결과가 없는 분석 워커.

    한 워커의 초기 invocation 이 둘 이상이면(교정 디스패치는 새 invocation 이다)
    그중 하나라도 마지막 attempt 가 ``ok`` 면 채택 가능한 결과가 있는 것이므로
    폐기로 보지 않는다. legacy 매니페스트나 v1 roster 는 판정하지 않는다.
    """
    if not isinstance(document, dict) or not isinstance(document.get("workers"), list):
        return {}
    if execution_manifest is None or getattr(execution_manifest, "legacy", False):
        return {}
    initial_refs: dict[str, set[str]] = {}
    for invocation in getattr(execution_manifest, "invocations", ()):
        if invocation.dispatch_kind == "initial":
            initial_refs.setdefault(invocation.role_execution_ref, set()).add(
                invocation.invocation_ref
            )
    latest: dict[str, object] = {}
    for attempt in getattr(execution_manifest, "attempts", ()):
        current = latest.get(attempt.invocation_ref)
        if current is None or attempt.attempt > current.attempt:
            latest[attempt.invocation_ref] = attempt
    result: dict[str, str] = {}
    for row in document["workers"]:
        if not isinstance(row, dict):
            continue
        worker = row.get("workerId")
        role_ref = row.get("sourceRoleExecutionRef")
        if not _nonempty_string(worker) or not _nonempty_string(role_ref):
            continue
        rows = [latest[ref] for ref in sorted(initial_refs.get(role_ref, ())) if ref in latest]
        if not rows or any(item.status == "ok" for item in rows):
            continue
        discarded = [item for item in rows if item.status in DISCARDED_ATTEMPT_STATUSES]
        if discarded:
            item = discarded[-1]
            result[worker] = f"{item.invocation_ref}#{item.attempt} {item.status}"
    return result


def discarded_worker_errors(document: object, execution_manifest: object) -> list[str]:
    """Groups that cite a worker whose only initial attempts were discarded.

    The mutation audit discards a result by setting the attempt terminal status
    and clearing ``resultPath``; the file itself stays in ``worker-results/``, so
    ``provenance_errors`` (which reads the file) still passes. Observed 2026-09-06
    (`fontsninja-v3-site` final-verification 001): the claude verifier's initial
    attempt was ``contract-failed-unattributed`` yet 9 of 10 groups cited it and
    3 named it ``originWorker`` — the ledger said "discarded" while convergence
    said "adopted".
    """
    discarded = discarded_workers(document, execution_manifest)
    if not discarded or not isinstance(document, dict):
        return []
    errors: list[str] = []
    for group in document.get("groups") or []:
        if not isinstance(group, dict):
            continue
        cited: set[str] = set()
        origin = group.get("originWorker")
        if isinstance(origin, str) and origin in discarded:
            cited.add(origin)
        discovered_by = group.get("discoveredBy")
        if isinstance(discovered_by, dict):
            cited.update(worker for worker in discovered_by if worker in discarded)
        for worker in sorted(cited):
            errors.append(
                f"group {group.get('findingId')} cites `{worker}`, whose initial "
                f"attempt {discarded[worker]} was discarded by the mutation audit — "
                "a discarded result may not seed convergence; re-dispatch the worker "
                "as a new invocation and cite that result"
            )
    return errors