"""Deterministic talk-identity assessment for candidate PPTX decks.

A deck becomes evidence for whichever talk the catalog says it belongs to. That
binding is made once, before persistence, and every later safeguard runs after
it — so a deck bound to the wrong talk supplies slide counts, design evidence,
OCR, and pattern observations to that wrong talk with nothing downstream able to
notice.

This module proves the binding instead of assuming it. It reads deterministic
identity facts already present on both sides — the catalog's title, conference,
and delivery date; the deck's path, document properties, and rendered title,
footer, and hashtag text — and reports which talk the deck belongs to, or
refuses to choose.

Two rules shape the taxonomy:

* Filename similarity alone never selects a talk. Reused talk families and
  nearby years produce near-identical filenames, which is precisely the signal
  that mis-assigned the live vault's decks in the first place.
* An unproven binding is a review finding, never a silent match. Refusing to
  choose costs one owner decision; choosing wrongly corrupts every derived
  analysis for that talk until someone notices.

Title and event comparison delegate to `source_identity_matching`, the same
authority the video source-identity audit uses. A deck must not be matched by a
weaker parallel rule than a recording.
"""

from __future__ import annotations

import posixpath
import re
from dataclasses import dataclass, field
from typing import Any, Iterable, Mapping, Sequence, TypeGuard

from source_identity_matching import (
    AMBIGUOUS_EVENT_ALIASES,
    EventAlias,
    YEAR_RE,
    event_alias,
    event_aliases_compatible,
    known_event_aliases,
    normalized_words,
    titles_agree,
)


# v2 binds the assessment to the deck GENERATION it was made against, not just
# to the deck's path. v1 compared `assessment["pptx_path"]` to the row's path,
# so replacing the file at that path left the `matched` verdict standing and the
# new deck's slides, OCR and pattern observations became that talk's evidence
# under a proof made about different bytes — the #176 failure surviving its own
# fix. A v1 assessment cannot be upgraded in place: nothing recorded which bytes
# it read, so it falls to `identity_assessment_schema_unsupported` and reads as
# unproven, the same position `unassessed_legacy_binding` takes for the same
# reason (a proof not witnessed cannot be manufactured).
PPTX_TALK_IDENTITY_SCHEMA_VERSION = 2

# Signal verdicts. `unknown` is the honest default: a fact the deck does not
# carry is not evidence for or against any candidate.
SIGNAL_AGREE = "agree"
SIGNAL_CONFLICT = "conflict"
SIGNAL_UNKNOWN = "unknown"
SIGNAL_VERDICTS = frozenset({SIGNAL_AGREE, SIGNAL_CONFLICT, SIGNAL_UNKNOWN})

# Signal names, in report order.
SIGNAL_TITLE = "title"
SIGNAL_VENUE = "venue"
SIGNAL_DELIVERY_YEAR = "delivery_year"
SIGNAL_HASHTAG = "hashtag"
SIGNAL_PUBLISHED_PDF = "published_pdf"
SIGNAL_FILENAME_SIMILARITY = "filename_similarity"

SIGNAL_NAMES = (
    SIGNAL_TITLE,
    SIGNAL_VENUE,
    SIGNAL_DELIVERY_YEAR,
    SIGNAL_HASHTAG,
    SIGNAL_PUBLISHED_PDF,
    SIGNAL_FILENAME_SIMILARITY,
)

# A selecting signal must be one a same-family deck from another delivery could
# not also satisfy. Two signals are deliberately excluded:
#
# * filename similarity — reused talk families produce near-identical filenames,
#   which is what mis-assigned the live vault's decks in the first place;
# * delivery year — every talk delivered that year satisfies it equally, so it
#   narrows a candidate set without identifying anything in it.
#
# Both still report, and a year MISmatch still contradicts. Vetoing and electing
# are separate powers: a wrong year is proof of the wrong delivery, while a
# right year is proof of nothing.
SELECTING_SIGNALS = frozenset(
    {
        SIGNAL_TITLE,
        SIGNAL_VENUE,
        SIGNAL_HASHTAG,
        SIGNAL_PUBLISHED_PDF,
    }
)

VERDICT_MATCHED = "matched"
VERDICT_REVIEW_REQUIRED = "review_required"
VERDICT_UNMATCHED = "unmatched"
VERDICTS = frozenset({VERDICT_MATCHED, VERDICT_REVIEW_REQUIRED, VERDICT_UNMATCHED})

REASON_NO_CANDIDATE_TALKS = "identity_no_candidate_talks"
REASON_NO_AGREEING_SIGNAL = "identity_no_agreeing_signal"
REASON_FILENAME_SIMILARITY_ONLY = "identity_filename_similarity_only"
REASON_AMBIGUOUS_CANDIDATES = "identity_ambiguous_candidates"
REASON_CONFLICTING_SIGNALS = "identity_conflicting_signals"
REASON_MATCHED = "identity_matched"
REASON_NON_DELIVERY_ARTIFACT = "identity_non_delivery_artifact"
# Stamped by migration onto a record that bound its talk before any assessment
# existed. The binding is not declared wrong — it is declared unproven, which is
# the honest reading and the one that routes it to review instead of trusting it.
REASON_UNASSESSED_LEGACY_BINDING = "identity_unassessed_legacy_binding"
# The deck could not be observed at refusal time, so the stored generation could
# not be compared to anything. Distinct from a mismatch: "we looked and it
# differs" and "we could not look" are different findings, and collapsing them
# would let an unreadable deck read as a passing comparison.
REASON_SOURCE_UNOBSERVABLE = "identity_source_unobservable"
# The assessment names a different generation of this deck than the one on disk.
REASON_GENERATION_STALE = "identity_generation_stale"

REASON_CODES = frozenset(
    {
        REASON_NO_CANDIDATE_TALKS,
        REASON_NO_AGREEING_SIGNAL,
        REASON_FILENAME_SIMILARITY_ONLY,
        REASON_AMBIGUOUS_CANDIDATES,
        REASON_CONFLICTING_SIGNALS,
        REASON_MATCHED,
        REASON_NON_DELIVERY_ARTIFACT,
        REASON_UNASSESSED_LEGACY_BINDING,
    }
)

# The only reason a `matched` verdict may claim. Every other code in the
# taxonomy explains a refusal, so pairing one with `matched` describes an
# assessment that cannot exist — most dangerously the legacy-binding code, which
# means the opposite of proven.
MATCHED_REASON_CODES = frozenset({REASON_MATCHED})

# An editable master and a published static export are legitimate artifacts for
# a delivery, but they are not the delivery deck and must not silently become
# its evidence. Roles are reported so the owner records which artifact is the
# published source and which is a later or editable variant.
ROLE_DELIVERY = "delivery"
ROLE_MASTER = "master"
ROLE_STATIC_EXPORT = "static_export"
ROLE_BACKUP = "backup"
ARTIFACT_ROLES = frozenset(
    {ROLE_DELIVERY, ROLE_MASTER, ROLE_STATIC_EXPORT, ROLE_BACKUP}
)

# Matched against whole path tokens, never substrings: `masterclass` is a talk
# topic, `master` is an artifact role.
_ROLE_TOKENS: tuple[tuple[str, frozenset[str]], ...] = (
    (
        ROLE_BACKUP,
        frozenset({"backup", "backups", "bak", "archive", "archived", "old"}),
    ),
    (ROLE_MASTER, frozenset({"master", "masters", "template", "templates", "source"})),
    (ROLE_STATIC_EXPORT, frozenset({"static", "export", "exports", "exported"})),
)

_TOKEN_RE = re.compile(r"[^\W_]+", re.UNICODE)
_HASHTAG_RE = re.compile(r"#(\w+)", re.UNICODE)
_TALK_FILENAME_DATE_RE = re.compile(
    r"\A(?P<year>\d{4})-(?P<month>\d{2})-(?P<day>\d{2})-"
)

# A deck whose filename shares this many significant words with a talk's slug is
# similar enough to report — and never, on its own, similar enough to select.
_FILENAME_OVERLAP_MINIMUM = 2


class PptxTalkIdentityError(ValueError):
    """A talk-identity assessment input violates the module's contract."""


@dataclass(frozen=True)
class DeckIdentityFacts:
    """Deterministic identity facts observed on one candidate deck.

    Every field beyond `pptx_path` is optional because a damaged or minimal
    deck still deserves an assessment from whatever it does carry. Absence
    weakens the evidence; it never invents agreement.
    """

    pptx_path: str
    document_title: str | None = None
    document_created_year: str | None = None
    rendered_title: str | None = None
    rendered_footers: tuple[str, ...] = ()
    hashtags: tuple[str, ...] = ()
    published_pdf_talk_filename: str | None = None

    @property
    def path_tokens(self) -> tuple[str, ...]:
        return tuple(_TOKEN_RE.findall(self.pptx_path.casefold()))

    @property
    def basename(self) -> str:
        return posixpath.basename(self.pptx_path.replace("\\", "/"))


@dataclass(frozen=True)
class CandidateAssessment:
    """One talk's per-signal standing against the deck under assessment."""

    talk_filename: str
    signals: Mapping[str, str]
    agreeing: tuple[str, ...]
    conflicting: tuple[str, ...]

    @property
    def selectable(self) -> bool:
        """A candidate is selectable on corroboration with no contradiction."""
        return bool(self.agreeing) and not self.conflicting

    def as_json(self) -> dict[str, Any]:
        return {
            "talk_filename": self.talk_filename,
            "signals": {name: self.signals[name] for name in SIGNAL_NAMES},
            "agreeing": list(self.agreeing),
            "conflicting": list(self.conflicting),
        }


def derive_candidate_standing(
    signals: Mapping[str, str],
) -> tuple[tuple[str, ...], tuple[str, ...]]:
    """Derive `(agreeing, conflicting)` from a candidate's per-signal map.

    One rule, used by the producer that writes a candidate and by the owner gate
    that later reads one back. A stored candidate's arrays are a summary of its
    signal map, so a gate that trusts the summary without recomputing it accepts
    any pair of values a caller cares to put there — `agreeing: ["venue"]` over a
    signal map where venue conflicts, or over no venue reading at all.

    Only selecting signals can corroborate; every signal can contradict.
    """
    agreeing = tuple(
        name
        for name in SIGNAL_NAMES
        if name in SELECTING_SIGNALS and signals.get(name) == SIGNAL_AGREE
    )
    conflicting = tuple(
        name for name in SIGNAL_NAMES if signals.get(name) == SIGNAL_CONFLICT
    )
    return agreeing, conflicting


@dataclass(frozen=True)
class TalkIdentityAssessment:
    """The full, schema-versioned identity verdict for one deck."""

    pptx_path: str
    verdict: str
    artifact_role: str
    selected_talk_filename: str | None
    reason_codes: tuple[str, ...]
    # The deck generation this verdict was reached against. `None` means the
    # deck could not be digested at assessment time; the verdict still records
    # what the facts showed, and `binding_refusal` refuses to authorize a
    # binding it cannot pin to bytes.
    source_identity: Mapping[str, Any] | None = None
    candidates: tuple[CandidateAssessment, ...] = field(default=())

    @property
    def matched(self) -> bool:
        return self.verdict == VERDICT_MATCHED

    @property
    def review_required(self) -> bool:
        return self.verdict == VERDICT_REVIEW_REQUIRED

    def as_json(self) -> dict[str, Any]:
        return {
            "schema_version": PPTX_TALK_IDENTITY_SCHEMA_VERSION,
            "pptx_path": self.pptx_path,
            "verdict": self.verdict,
            "artifact_role": self.artifact_role,
            "selected_talk_filename": self.selected_talk_filename,
            "reason_codes": list(self.reason_codes),
            "source_identity": (
                dict(self.source_identity) if self.source_identity is not None else None
            ),
            "candidates": [candidate.as_json() for candidate in self.candidates],
        }


def _require_text(value: object, label: str) -> str:
    if not isinstance(value, str) or not value.strip():
        raise PptxTalkIdentityError(f"{label} must be a non-empty string")
    return value


def _optional_text(value: object, label: str) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str):
        raise PptxTalkIdentityError(f"{label} must be a string or null")
    stripped = value.strip()
    return stripped or None


def _text_tuple(value: object, label: str) -> tuple[str, ...]:
    if value is None:
        return ()
    if isinstance(value, str) or not isinstance(value, Sequence):
        raise PptxTalkIdentityError(f"{label} must be an array of strings")
    items: list[str] = []
    for index, item in enumerate(value):
        text = _optional_text(item, f"{label}[{index}]")
        if text is not None:
            items.append(text)
    return tuple(items)


def deck_identity_facts(value: object) -> DeckIdentityFacts:
    """Validate a caller-supplied deck fact mapping into the closed shape.

    Typed `object` deliberately: this is the boundary where unvalidated caller
    input becomes a closed shape, so the isinstance guard is the contract, not
    a redundant assertion behind a narrower annotation.
    """
    if not isinstance(value, Mapping):
        raise PptxTalkIdentityError("deck facts must be a mapping")
    unknown = set(value) - {
        "pptx_path",
        "document_title",
        "document_created_year",
        "rendered_title",
        "rendered_footers",
        "hashtags",
        "published_pdf_talk_filename",
    }
    if unknown:
        raise PptxTalkIdentityError(f"deck facts carry unknown keys: {sorted(unknown)}")
    created_year = _optional_text(
        value.get("document_created_year"), "deck facts document_created_year"
    )
    if created_year is not None and not re.fullmatch(r"(?:19|20)\d{2}", created_year):
        raise PptxTalkIdentityError(
            "deck facts document_created_year must be a four-digit year"
        )
    return DeckIdentityFacts(
        pptx_path=_require_text(value.get("pptx_path"), "deck facts pptx_path"),
        document_title=_optional_text(
            value.get("document_title"), "deck facts document_title"
        ),
        document_created_year=created_year,
        rendered_title=_optional_text(
            value.get("rendered_title"), "deck facts rendered_title"
        ),
        rendered_footers=_text_tuple(
            value.get("rendered_footers"), "deck facts rendered_footers"
        ),
        hashtags=_text_tuple(value.get("hashtags"), "deck facts hashtags"),
        published_pdf_talk_filename=_optional_text(
            value.get("published_pdf_talk_filename"),
            "deck facts published_pdf_talk_filename",
        ),
    )


def classify_artifact_role(facts: DeckIdentityFacts) -> str:
    """Report whether the deck is a delivery artifact or a variant of one."""
    tokens = set(facts.path_tokens)
    for role, markers in _ROLE_TOKENS:
        if tokens & markers:
            return role
    return ROLE_DELIVERY


def _talk_year(talk: Mapping[str, Any]) -> str | None:
    """Resolve a talk's delivery year from its date, else its filename prefix.

    Both are recorded values. Neither consults the clock, so an assessment made
    today and the same assessment made next year agree.
    """
    date_value = talk.get("date")
    if isinstance(date_value, str):
        match = YEAR_RE.search(date_value)
        if match is not None:
            return match.group(0)
    filename = talk.get("filename")
    if isinstance(filename, str):
        match = _TALK_FILENAME_DATE_RE.match(filename)
        if match is not None:
            return match.group("year")
    return None


def _deck_path_years(facts: DeckIdentityFacts) -> tuple[str, ...]:
    return tuple(
        token
        for token in facts.path_tokens
        if re.fullmatch(r"(?:19|20)\d{2}", token) is not None
    )


def _deck_venue_aliases(
    facts: DeckIdentityFacts, known_aliases: frozenset[EventAlias]
) -> list[EventAlias]:
    """Read venue aliases from the deck's directory components.

    Only directories are consulted. A deck's own filename is the talk's name far
    more often than the venue's, and reading a venue out of it would manufacture
    agreement from the one signal this module refuses to trust alone.

    A component counts as a venue claim only when it names an event some talk in
    the vault actually uses. Without that gate every generic folder — `Decks/`,
    `Downloads/` — would parse as an unrecognized venue and contradict every
    candidate, turning the discriminator into a blanket refusal.
    """
    normalized = facts.pptx_path.replace("\\", "/")
    components = [part for part in posixpath.dirname(normalized).split("/") if part]
    aliases: list[EventAlias] = []
    for component in components:
        alias = event_alias(component)
        if alias is None:
            continue
        if len(alias) == 1 and alias[0] in AMBIGUOUS_EVENT_ALIASES:
            # `devops/` names a topic folder at least as often as an event.
            continue
        if not any(event_aliases_compatible(alias, known) for known in known_aliases):
            continue
        aliases.append(alias)
    return aliases


def _title_signal(
    facts: DeckIdentityFacts,
    talk: Mapping[str, Any],
    _known_aliases: frozenset[EventAlias],
) -> str:
    talk_title = talk.get("title")
    if not isinstance(talk_title, str) or not talk_title.strip():
        return SIGNAL_UNKNOWN
    observed = [
        text
        for text in (
            facts.document_title,
            facts.rendered_title,
            *facts.rendered_footers,
        )
        if text
    ]
    if not observed:
        return SIGNAL_UNKNOWN
    if any(titles_agree(talk_title, text) for text in observed):
        return SIGNAL_AGREE
    # A deck legitimately carries a punchier title than its catalog entry, so a
    # non-agreeing title is missing corroboration rather than contradiction.
    return SIGNAL_UNKNOWN


def _venue_signal(
    facts: DeckIdentityFacts,
    talk: Mapping[str, Any],
    known_aliases: frozenset[EventAlias],
) -> str:
    talk_alias = event_alias(talk.get("conference"))
    if talk_alias is None:
        return SIGNAL_UNKNOWN
    deck_aliases = _deck_venue_aliases(facts, known_aliases)
    if not deck_aliases:
        return SIGNAL_UNKNOWN
    if any(event_aliases_compatible(talk_alias, alias) for alias in deck_aliases):
        return SIGNAL_AGREE
    # The deck sits under a named event that is not this talk's event. That is
    # the discriminator between two deliveries of one reused talk.
    return SIGNAL_CONFLICT


def _delivery_year_signal(
    facts: DeckIdentityFacts,
    talk: Mapping[str, Any],
    _known_aliases: frozenset[EventAlias],
) -> str:
    talk_year = _talk_year(talk)
    if talk_year is None:
        return SIGNAL_UNKNOWN
    path_years = _deck_path_years(facts)
    if path_years:
        if talk_year in path_years:
            return SIGNAL_AGREE
        return SIGNAL_CONFLICT
    if facts.document_created_year is not None:
        if facts.document_created_year == talk_year:
            return SIGNAL_AGREE
        # A master edited or re-saved years later is normal, so a document
        # timestamp corroborates but never contradicts.
        return SIGNAL_UNKNOWN
    return SIGNAL_UNKNOWN


def _hashtag_signal(
    facts: DeckIdentityFacts,
    talk: Mapping[str, Any],
    known_aliases: frozenset[EventAlias],
) -> str:
    talk_alias = event_alias(talk.get("conference"))
    if talk_alias is None or not facts.hashtags:
        return SIGNAL_UNKNOWN
    # `#VoxxedTicino` carries no word boundary, so a compact comparison against
    # the joined alias recovers what alias-wise compatibility cannot see.
    compact_talk = "".join(talk_alias)
    for raw in facts.hashtags:
        for token in _HASHTAG_RE.findall(raw) or [raw.lstrip("#")]:
            alias = event_alias(token)
            if alias is None:
                continue
            if event_aliases_compatible(talk_alias, alias):
                return SIGNAL_AGREE
            if "".join(alias) == compact_talk:
                return SIGNAL_AGREE
    # Conference hashtags are inconsistently present and frequently abbreviated
    # beyond alias recovery, so absence of agreement proves nothing.
    return SIGNAL_UNKNOWN


def _published_pdf_signal(
    facts: DeckIdentityFacts,
    talk: Mapping[str, Any],
    _known_aliases: frozenset[EventAlias],
) -> str:
    if facts.published_pdf_talk_filename is None:
        return SIGNAL_UNKNOWN
    filename = talk.get("filename")
    if not isinstance(filename, str) or not filename:
        return SIGNAL_UNKNOWN
    if facts.published_pdf_talk_filename == filename:
        return SIGNAL_AGREE
    return SIGNAL_CONFLICT


def _filename_similarity_signal(
    facts: DeckIdentityFacts,
    talk: Mapping[str, Any],
    _known_aliases: frozenset[EventAlias],
) -> str:
    filename = talk.get("filename")
    if not isinstance(filename, str) or not filename:
        return SIGNAL_UNKNOWN
    slug = _TALK_FILENAME_DATE_RE.sub("", filename)
    slug = re.sub(r"\.md\Z", "", slug)
    talk_words = normalized_words(slug.replace("-", " "))
    deck_words = normalized_words(posixpath.splitext(facts.basename)[0])
    if not talk_words or not deck_words:
        return SIGNAL_UNKNOWN
    if len(talk_words & deck_words) >= _FILENAME_OVERLAP_MINIMUM:
        return SIGNAL_AGREE
    return SIGNAL_UNKNOWN


_SIGNAL_EVALUATORS = {
    SIGNAL_TITLE: _title_signal,
    SIGNAL_VENUE: _venue_signal,
    SIGNAL_DELIVERY_YEAR: _delivery_year_signal,
    SIGNAL_HASHTAG: _hashtag_signal,
    SIGNAL_PUBLISHED_PDF: _published_pdf_signal,
    SIGNAL_FILENAME_SIMILARITY: _filename_similarity_signal,
}


def assess_candidate(
    facts: DeckIdentityFacts,
    talk: Mapping[str, Any],
    known_aliases: frozenset[EventAlias] = frozenset(),
) -> CandidateAssessment:
    """Evaluate every signal for one candidate talk.

    `known_aliases` is the vault's vocabulary of real event names. It defaults
    to empty so a single-candidate caller still gets an assessment; the effect
    of an empty vocabulary is that no directory reads as a venue, which loses
    the signal rather than inventing one.
    """
    filename = talk.get("filename")
    if not isinstance(filename, str) or not filename:
        raise PptxTalkIdentityError("candidate talk requires a filename")
    signals = {
        name: evaluator(facts, talk, known_aliases)
        for name, evaluator in _SIGNAL_EVALUATORS.items()
    }
    agreeing, conflicting = derive_candidate_standing(signals)
    return CandidateAssessment(
        talk_filename=filename,
        signals=signals,
        agreeing=agreeing,
        conflicting=conflicting,
    )


def assess_pptx_talk_identity(
    deck: Mapping[str, Any] | DeckIdentityFacts,
    candidates: Iterable[Mapping[str, Any]],
    *,
    source_identity: Mapping[str, Any] | None = None,
) -> TalkIdentityAssessment:
    """Decide which talk a deck belongs to, or refuse to decide.

    The verdict is `matched` only when exactly one candidate is corroborated by
    a signal filename similarity cannot fake and contradicted by none. Every
    other outcome is `review_required` or `unmatched`; neither authorizes
    catalog persistence or extraction.
    """
    facts = deck if isinstance(deck, DeckIdentityFacts) else deck_identity_facts(deck)
    artifact_role = classify_artifact_role(facts)
    talks = list(candidates)
    # The candidate set is the vault's event vocabulary. A directory naming an
    # event no talk uses is a folder, not a contradicting venue.
    known_aliases = frozenset(known_event_aliases(talks))
    assessments = tuple(assess_candidate(facts, talk, known_aliases) for talk in talks)

    seen: set[str] = set()
    for assessment in assessments:
        if assessment.talk_filename in seen:
            raise PptxTalkIdentityError(
                f"candidate talk {assessment.talk_filename!r} appears twice"
            )
        seen.add(assessment.talk_filename)

    def result(
        verdict: str, selected: str | None, *reasons: str
    ) -> TalkIdentityAssessment:
        codes = list(reasons)
        # A master, backup, or static export never carries a bare `matched`
        # verdict into persistence: the owner records which artifact is the
        # published source before its contents become a talk's evidence.
        if artifact_role != ROLE_DELIVERY and verdict == VERDICT_MATCHED:
            verdict = VERDICT_REVIEW_REQUIRED
            codes.append(REASON_NON_DELIVERY_ARTIFACT)
        elif artifact_role != ROLE_DELIVERY:
            codes.append(REASON_NON_DELIVERY_ARTIFACT)
        return TalkIdentityAssessment(
            pptx_path=facts.pptx_path,
            verdict=verdict,
            artifact_role=artifact_role,
            selected_talk_filename=selected if verdict == VERDICT_MATCHED else None,
            reason_codes=tuple(dict.fromkeys(codes)),
            source_identity=source_identity,
            candidates=assessments,
        )

    if not assessments:
        return result(VERDICT_UNMATCHED, None, REASON_NO_CANDIDATE_TALKS)

    selectable = [item for item in assessments if item.selectable]
    if len(selectable) == 1:
        return result(VERDICT_MATCHED, selectable[0].talk_filename, REASON_MATCHED)
    if len(selectable) > 1:
        return result(VERDICT_REVIEW_REQUIRED, None, REASON_AMBIGUOUS_CANDIDATES)

    contradicted = [item for item in assessments if item.agreeing and item.conflicting]
    if contradicted:
        return result(VERDICT_REVIEW_REQUIRED, None, REASON_CONFLICTING_SIGNALS)

    filename_only = [
        item
        for item in assessments
        if item.signals[SIGNAL_FILENAME_SIMILARITY] == SIGNAL_AGREE
        and not item.conflicting
    ]
    if filename_only:
        return result(VERDICT_REVIEW_REQUIRED, None, REASON_FILENAME_SIMILARITY_ONLY)

    return result(VERDICT_UNMATCHED, None, REASON_NO_AGREEING_SIGNAL)


def unassessed_legacy_binding(pptx_path: str) -> dict[str, Any]:
    """The assessment for a talk binding made before assessments existed.

    Migration cannot prove a binding it did not witness, and inventing a
    `matched` verdict for one would forge exactly the evidence this module was
    written to require. It cannot call the binding wrong either — most legacy
    bindings are right. `review_required` is the only honest verdict: the record
    upgrades, the binding survives, and nothing downstream may treat it as
    proven until someone looks.
    """
    return {
        "schema_version": PPTX_TALK_IDENTITY_SCHEMA_VERSION,
        "pptx_path": pptx_path,
        "verdict": VERDICT_REVIEW_REQUIRED,
        "artifact_role": ROLE_DELIVERY,
        "selected_talk_filename": None,
        "reason_codes": [REASON_UNASSESSED_LEGACY_BINDING],
        # Null rather than absent, for the same reason the candidates list is
        # empty rather than absent: migration read no bytes, and saying so is a
        # different statement from the field never having existed.
        "source_identity": None,
        # Empty rather than absent: migration assessed no candidates, which is
        # a different statement from having assessed some and reported none.
        "candidates": [],
    }


_SOURCE_IDENTITY_FIELDS = ("algorithm", "digest", "size_bytes")
# Mirrors `tracking_database`'s PPTX source-fingerprint contract, restated
# rather than imported because that module imports THIS one
# (`unassessed_legacy_binding`), so reaching back would cycle.
# `tests/test_pptx_talk_identity.py` pins the two to each other, so a contract
# change landing in one file and not the other fails CI instead of leaving the
# assessment accepting generations the database would refuse.
_SOURCE_IDENTITY_ALGORITHMS = frozenset({"sha256"})
_SHA256_DIGEST_LENGTH = 64
_LOWER_HEX = frozenset("0123456789abcdef")


def _source_identity_comparable(value: object) -> TypeGuard[Mapping[str, Any]]:
    """Whether a source identity is a valid generation, not merely present.

    Held to the same contract the extractor's fingerprint is held to, because
    the two are compared to each other. A looser reading here would accept a
    generation the database would refuse — `{"algorithm": "x", "digest": "x",
    "size_bytes": 0}` is not a deck anything could have read — and an
    assessment carrying one would claim a binding proven against bytes that
    cannot exist.

    A partial identity is not a weak match, it is an unusable one: comparing on
    whichever fields happen to be present would let a record missing `digest`
    agree with anything that shares its size.
    """
    if not isinstance(value, Mapping):
        return False
    if set(value) != set(_SOURCE_IDENTITY_FIELDS):
        return False
    if value["algorithm"] not in _SOURCE_IDENTITY_ALGORITHMS:
        return False
    digest = value["digest"]
    if (
        not isinstance(digest, str)
        or len(digest) != _SHA256_DIGEST_LENGTH
        or not set(digest) <= _LOWER_HEX
    ):
        return False
    size = value["size_bytes"]
    # A zero-byte deck is not a deck. `bool` is an `int` in Python, so it is
    # excluded before the range check rather than sliding through as 0 or 1.
    return not isinstance(size, bool) and isinstance(size, int) and size >= 1


def _same_source_generation(
    stored: Mapping[str, Any], observed: Mapping[str, Any]
) -> bool:
    """Exact agreement on every field, never a subset.

    The algorithm is compared too: the same digest string under two algorithms
    is a coincidence of encoding, not the same bytes.
    """
    return all(stored[field] == observed[field] for field in _SOURCE_IDENTITY_FIELDS)


def binding_refusal(
    assessment: object,
    *,
    pptx_path: str,
    talk_filename: str,
    observed_source_identity: Mapping[str, Any] | None,
) -> str | None:
    """Say why an assessment fails to authorize a binding, or None if it does.

    One predicate, two callers. The owner writer raises on it so an unproven
    binding is never persisted; preflight reports it so a binding persisted
    before this contract existed cannot pass as proven. Two copies of this rule
    would drift, and the direction they drift is a reader trusting what a writer
    would have refused (`stateful-artifacts` -> Hints, Not Authority).

    Checks the whole triple, not the verdict alone: an assessment names a deck
    AND a talk, so one that proves a different pair proves nothing about this
    row. Reason codes are checked because every code but the matched one
    explains a refusal.

    Two separate things are checked about the deck generation, and they are not
    the same requirement:

    * The assessment must NAME the generation it read. Always. A path is not a
      deck: v1 pinned an assessment to `pptx_path` alone, so replacing the file
      at that path kept the `matched` verdict and handed the new deck's contents
      to the talk under a proof about different bytes.
    * It must AGREE with an independent observation, when the caller has one.

    `observed_source_identity` is REQUIRED, never defaulted, so a caller states
    which case it is in rather than falling into one. `None` says "I have no
    independent observation" — true of `mutate-tracking-database.py`, which
    takes a database and a plan and never touches the vault. That is not a pass:
    the assessment must still carry a comparable generation, so every v1
    assessment and every unwitnessed migration stamp still refuses. It is
    `preflight-vault.py` that observes the deck and makes the comparison real.

    A caller holding an observation must not pass `None` to skip the
    comparison — that is the fail-open shape the observation gate was made
    required to avoid. Preflight treats an unobservable deck as its own finding
    rather than as an absent observation.
    """
    if not isinstance(assessment, Mapping):
        return "identity_assessment_missing"
    missing = {
        "schema_version",
        "pptx_path",
        "verdict",
        "artifact_role",
        "selected_talk_filename",
        "reason_codes",
        "source_identity",
        "candidates",
    } - set(assessment)
    if missing:
        return "identity_assessment_incomplete"
    if assessment["schema_version"] != PPTX_TALK_IDENTITY_SCHEMA_VERSION:
        return "identity_assessment_schema_unsupported"
    if assessment["verdict"] != VERDICT_MATCHED:
        return "identity_verdict_not_matched"
    if assessment["pptx_path"] != pptx_path:
        return "identity_deck_mismatch"
    if assessment["selected_talk_filename"] != talk_filename:
        return "identity_talk_mismatch"
    if assessment["artifact_role"] != ROLE_DELIVERY:
        return "identity_non_delivery_artifact"
    codes = assessment["reason_codes"]
    if not isinstance(codes, list) or not all(isinstance(code, str) for code in codes):
        return "identity_reason_codes_invalid"
    if sorted(set(codes)) != sorted(MATCHED_REASON_CODES):
        return "identity_reason_codes_contradict_verdict"
    table_refusal = _candidate_table_refusal(assessment["candidates"], talk_filename)
    if table_refusal is not None:
        return table_refusal
    # Last, deliberately. Everything above asks whether the assessment is a
    # coherent proof at all; this asks whether it is a proof about the bytes
    # that are there now. Running it earlier would mask a malformed assessment
    # behind a generation complaint and change which defect the operator sees
    # first.
    stored_identity = assessment["source_identity"]
    if not _source_identity_comparable(stored_identity):
        return REASON_SOURCE_UNOBSERVABLE
    if observed_source_identity is not None and not _same_source_generation(
        stored_identity, observed_source_identity
    ):
        return REASON_GENERATION_STALE
    return None


def _candidate_table_refusal(candidates: object, talk_filename: str) -> str | None:
    """Check the evidence behind the verdict, not just the verdict.

    A `matched` verdict over an empty candidate table is a conclusion with
    nothing under it — which is exactly what a fabricated assessment looks like.
    The table must show this talk winning the way `assess_pptx_talk_identity`
    makes it win: corroborated by a selecting signal, contradicted by none, and
    with no rival equally corroborated.
    """
    if not isinstance(candidates, list) or not candidates:
        return "identity_candidate_table_missing"
    selected: tuple[tuple[str, ...], tuple[str, ...]] | None = None
    rivals = 0
    for candidate in candidates:
        if not isinstance(candidate, Mapping):
            return "identity_candidate_table_invalid"
        name = candidate.get("talk_filename")
        signals = candidate.get("signals")
        stored_agreeing = candidate.get("agreeing")
        stored_conflicting = candidate.get("conflicting")
        if (
            not isinstance(name, str)
            or not isinstance(signals, Mapping)
            or not isinstance(stored_agreeing, list)
            or not isinstance(stored_conflicting, list)
        ):
            return "identity_candidate_table_invalid"
        # The signal map is the evidence; the two arrays are its summary. Read
        # the evidence and recompute the summary, so a candidate cannot assert a
        # standing its own readings do not support.
        if set(signals) != set(SIGNAL_NAMES) or any(
            signals[key] not in SIGNAL_VERDICTS for key in signals
        ):
            return "identity_candidate_signals_invalid"
        agreeing, conflicting = derive_candidate_standing(signals)
        if list(agreeing) != stored_agreeing or list(conflicting) != stored_conflicting:
            return "identity_candidate_standing_contradicts_signals"
        if name == talk_filename:
            if selected is not None:
                return "identity_candidate_table_invalid"
            selected = (agreeing, conflicting)
        elif agreeing and not conflicting:
            rivals += 1
    if selected is None:
        return "identity_candidate_absent"
    selected_agreeing, selected_conflicting = selected
    if not selected_agreeing or selected_conflicting:
        return "identity_candidate_not_selectable"
    if rivals:
        return "identity_candidate_not_unique"
    return None