src/gcmd_classifier/datasets/evidence.py

"""Blind structured evidence selection and immutable packet construction."""

from __future__ import annotations

import hashlib
import json
import re
from collections.abc import Callable
from typing import TypeVar

from pydantic import BaseModel

from gcmd_classifier.config import DatasetEvidenceSettings, ModelSettings
from gcmd_classifier.datasets.errors import EvidenceSelectionError
from gcmd_classifier.datasets.evidence_prompts import (
    build_dataset_evidence_minimization_prompt,
    build_dataset_evidence_selection_prompt,
)
from gcmd_classifier.datasets.models import (
    DatasetClassifierInput,
    DatasetErrorRecord,
    DatasetEvidencePacket,
    DatasetExtractionManifest,
    DatasetReviewStatus,
    EvidenceBlockGroup,
    EvidenceEligibility,
    EvidenceExclusionReason,
    EvidenceExclusionRecord,
    EvidenceGroupMember,
    EvidenceMinimizationModelResponse,
    EvidenceNoUsableScienceOutcome,
    EvidencePacketBlock,
    EvidencePacketResult,
    EvidenceRole,
    EvidenceSelectionAudit,
    EvidenceSelectionDecision,
    EvidenceSelectionFailure,
    EvidenceSelectionInput,
    EvidenceSelectionModelResponse,
    GroupMembership,
    PacketComponentHashes,
    PacketLimitRecord,
    PacketReadiness,
    ProductScopeDecision,
    ProductScopeType,
    ScientificRelevance,
)
from gcmd_classifier.llm.base import ModelClient, ModelRequest, ModelStage

_T = TypeVar("_T", bound=BaseModel)


def canonical_json(value: object) -> bytes:
    """Return canonical UTF-8 JSON used by every Milestone 6 logical hash."""
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode(
        "utf-8"
    )


def canonical_sha256(value: object) -> str:
    """Hash one canonically serialized logical component."""
    return hashlib.sha256(canonical_json(value)).hexdigest()


def extraction_manifest_sha256(manifest: DatasetExtractionManifest) -> str:
    """Hash a manifest using the same canonical rule as product resolution."""
    return canonical_sha256(manifest.model_dump(mode="json"))


def block_sha256(text: str) -> str:
    """Hash the exact authoritative extracted block text."""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


_INDEPENDENT_ROLES = {
    EvidenceRole.SCIENTIFIC_DESCRIPTION,
    EvidenceRole.MEASURED_VARIABLE,
    EvidenceRole.GEOPHYSICAL_PARAMETER,
    EvidenceRole.METHOD_OR_ALGORITHM,
}


_TECHNICAL_HEADING = re.compile(
    r"table of contents|revision|change log|contact|references|bibliography|"
    r"download|data access|file format|quality flag|fill value|ancillary input",
    re.IGNORECASE,
)
_TECHNICAL_TEXT = re.compile(
    r"^(table of contents|revision history|contact information|references\b)|"
    r"\b(fill value|quality flag|error code|byte order|file format|data format|"
    r"dimension name|fortran source|download URL|FTP server)\b",
    re.IGNORECASE,
)
_SCIENCE_OUTPUT = re.compile(
    r"\b(measur|retriev|deriv|estimat|observ|scientific|geophysical|parameter|"
    r"variable|concentration|temperature|precipitation|radiation|aerosol|ozone|"
    r"wind|humidity|pressure|flux)\w*",
    re.IGNORECASE,
)


def _deterministic_non_science(block) -> bool:
    heading = block.section_heading or ""
    if _TECHNICAL_HEADING.search(heading):
        return not _SCIENCE_OUTPUT.search(block.text)
    return bool(_TECHNICAL_TEXT.search(block.text) and not _SCIENCE_OUTPUT.search(block.text))


def _independent_candidate_allowed(scope, target_short_name, settings):
    return (
        scope.eligibility is EvidenceEligibility.ELIGIBLE
        and scope.scope_type is not ProductScopeType.MIXED_PRODUCT_TRANSITION
        and not scope.review_required
        and scope.product_match_confidence.value >= settings.product_match_review_threshold
        and target_short_name in scope.applicable_products
    )


def _candidate_exclusions(source, settings):
    records = []
    target = source.product_resolution.target.short_name
    blocks = {item.block_id: item for item in source.extraction_manifest.blocks}
    for item in source.product_resolution.block_decisions:
        if item.eligibility is not EvidenceEligibility.ELIGIBLE:
            continue
        if _deterministic_non_science(blocks[item.block_id]):
            reason = EvidenceExclusionReason.DETERMINISTIC_NON_SCIENCE
        elif item.review_required:
            reason = EvidenceExclusionReason.PRODUCT_REVIEW_REQUIRED
        elif item.product_match_confidence.value < settings.product_match_review_threshold:
            reason = EvidenceExclusionReason.CONFIDENCE_BELOW_ACCEPTANCE
        elif target not in item.applicable_products:
            reason = EvidenceExclusionReason.APPLICABILITY_UNTRACEABLE
        else:
            continue
        records.append(
            EvidenceExclusionRecord(block_id=item.block_id, reason=reason, phase="candidate")
        )
    return tuple(records)


def build_evidence_groups(
    source: EvidenceSelectionInput,
    settings: DatasetEvidenceSettings | None = None,
    *,
    only_block_ids: tuple[str, ...] | None = None,
) -> tuple[EvidenceBlockGroup, ...]:
    """Build reproducible bounded contiguous groups from eligible blind blocks."""
    policy = settings or DatasetEvidenceSettings()
    manifest = source.extraction_manifest
    resolution = source.product_resolution
    expected_manifest_hash = extraction_manifest_sha256(manifest)
    if resolution.extraction_manifest_sha256 != expected_manifest_hash:
        raise EvidenceSelectionError(
            "STALE_EXTRACTION_MANIFEST",
            "Product resolution does not identify the supplied extraction manifest.",
        )
    decisions = {item.block_id: item for item in resolution.block_decisions}
    allowed = set(only_block_ids) if only_block_ids is not None else None
    target_ids = [
        block.block_id
        for block in manifest.blocks
        if _independent_candidate_allowed(
            decisions[block.block_id], resolution.target.short_name, policy
        )
        and not _deterministic_non_science(block)
        and (allowed is None or block.block_id in allowed)
    ]
    target_set = set(target_ids)
    if not target_ids:
        return ()

    groups: list[EvidenceBlockGroup] = []
    current: list[EvidenceGroupMember] = []
    characters = 0
    blocks = {item.block_id: item for item in manifest.blocks}
    for block_id in target_ids:
        block = blocks[block_id]
        if current and (
            len(current) >= policy.max_blocks_per_group
            or characters + len(block.text) > policy.max_characters_per_group
        ):
            groups.append(
                _make_group(
                    len(groups) + 1,
                    current,
                    characters,
                    expected_manifest_hash,
                    policy,
                )
            )
            current = []
            characters = 0
        current.append(
            EvidenceGroupMember(
                block_id=block_id,
                membership=GroupMembership.DECISION_TARGET,
                linked_target_block_ids=(),
            )
        )
        characters += len(block.text)

    if current:
        groups.append(
            _make_group(
                len(groups) + 1,
                current,
                characters,
                expected_manifest_hash,
                policy,
            )
        )

    # Add only explicitly qualifying document context when it fits. Context never
    # creates a required or final scientific decision.
    context_ids = set(resolution.document_context_block_ids)
    for index, group in enumerate(groups):
        target_members = [m.block_id for m in group.members]
        linked: list[EvidenceGroupMember] = []
        used = group.source_character_count
        for target_id in target_members:
            scope = decisions[target_id]
            for linked_id in scope.qualifying_block_ids:
                if linked_id not in context_ids or linked_id in target_set:
                    continue
                candidate = blocks[linked_id]
                if (
                    len(group.members) + len(linked) >= policy.max_blocks_per_group
                    or used + len(candidate.text) > policy.max_characters_per_group
                ):
                    continue
                if any(item.block_id == linked_id for item in linked):
                    continue
                linked.append(
                    EvidenceGroupMember(
                        block_id=linked_id,
                        membership=GroupMembership.CONTEXT,
                        linked_target_block_ids=(target_id,),
                    )
                )
                used += len(candidate.text)
        if linked:
            ordered = tuple(
                sorted(
                    (*group.members, *linked), key=lambda item: blocks[item.block_id].source_order
                )
            )
            groups[index] = group.model_copy(
                update={
                    "members": ordered,
                    "source_character_count": used,
                    "context_link_method": "explicit_product_qualifying_links_v1",
                }
            )
    return tuple(groups)


def _make_group(
    number: int,
    members: list[EvidenceGroupMember],
    characters: int,
    manifest_hash: str,
    settings: DatasetEvidenceSettings,
) -> EvidenceBlockGroup:
    return EvidenceBlockGroup(
        group_id=f"group-{number:04}",
        extraction_manifest_sha256=manifest_hash,
        grouping_policy_version=settings.grouping_policy_version,
        members=tuple(members),
        source_character_count=characters,
        max_blocks=settings.max_blocks_per_group,
        max_characters=settings.max_characters_per_group,
        context_link_method="none",
    )


def _request(
    prompt: str,
    settings: DatasetEvidenceSettings,
    model: ModelSettings,
    *,
    reduction: bool,
) -> ModelRequest[EvidenceSelectionModelResponse]:
    return ModelRequest(
        stage=ModelStage.DATASET_EVIDENCE_SELECTION,
        prompt=prompt,
        response_schema=EvidenceSelectionModelResponse,
        provider=model.provider,
        model_name=model.model_name,
        prompt_version=settings.selection_prompt_version,
        temperature=model.temperature,
        timeout_seconds=model.timeout_seconds,
        metadata={"blind_stage": True, "reduction": reduction},
    )


def _validate_response(
    response: EvidenceSelectionModelResponse,
    *,
    group: EvidenceBlockGroup,
    source: EvidenceSelectionInput,
    reduction: bool,
    settings: DatasetEvidenceSettings | None = None,
) -> None:
    required = [
        member.block_id
        for member in group.members
        if member.membership is GroupMembership.DECISION_TARGET
    ]
    actual = [item.block_id for item in response.decisions]
    if len(actual) != len(set(actual)):
        raise EvidenceSelectionError(
            "DUPLICATE_BLOCK_DECISION", "A block decision appeared more than once."
        )
    if set(actual) != set(required):
        raise EvidenceSelectionError(
            "EVIDENCE_BLOCK_PARTITION_INVALID",
            "Decisions must exactly partition supplied decision-target blocks.",
        )
    manifest_blocks = {block.block_id: block for block in source.extraction_manifest.blocks}
    scopes = {item.block_id: item for item in source.product_resolution.block_decisions}
    declarations = {item.declaration_id for item in source.product_resolution.declarations}
    group_ids = {member.block_id for member in group.members}
    for decision in response.decisions:
        if decision.block_id not in manifest_blocks or decision.block_id not in scopes:
            raise EvidenceSelectionError(
                "UNKNOWN_BLOCK_ID", "Evidence response referenced an unknown source block."
            )
        scope = scopes[decision.block_id]
        if (
            decision.scope_type is not scope.scope_type
            or decision.evidence_eligibility is not scope.eligibility
            or decision.product_match_confidence != scope.product_match_confidence
        ):
            raise EvidenceSelectionError(
                "PRODUCT_SCOPE_OVERRIDE",
                "Evidence response attempted to change authoritative product scope.",
            )
        member_by_id = {member.block_id: member for member in group.members}
        linked_blocks = set(decision.qualifying_block_ids)
        authoritative_links = set(scope.qualifying_block_ids)
        if (
            not linked_blocks.issubset(group_ids)
            or not linked_blocks.issubset(authoritative_links)
            or any(
                member_by_id[item].membership is GroupMembership.CONTEXT
                and decision.block_id not in member_by_id[item].linked_target_block_ids
                for item in linked_blocks
            )
        ):
            raise EvidenceSelectionError(
                "INVALID_QUALIFYING_LINK",
                "Evidence response used a context link not explicitly authorized for this block.",
            )
        linked_declarations = set(decision.qualifying_declaration_ids)
        if not linked_declarations.issubset(declarations) or not linked_declarations.issubset(
            scope.declaration_ids
        ):
            raise EvidenceSelectionError(
                "INVALID_QUALIFYING_LINK",
                "Evidence response used a declaration not explicitly linked to this block.",
            )
        if decision.selected and scope.eligibility is not EvidenceEligibility.ELIGIBLE:
            raise EvidenceSelectionError(
                "INELIGIBLE_BLOCK_SELECTED",
                "An ineligible or review-only source block cannot be selected.",
            )
        if decision.selected and decision.scientific_relevance is not ScientificRelevance.RELEVANT:
            raise EvidenceSelectionError(
                "SELECTION_RELEVANCE_CONFLICT",
                "Selected scientific evidence must be marked relevant.",
            )
        if decision.selected and decision.evidence_role not in _INDEPENDENT_ROLES:
            raise EvidenceSelectionError(
                "NON_INDEPENDENT_EVIDENCE_SELECTED",
                "Only a substantive science role may be independently selected.",
            )
        if decision.selected and (decision.review_required or scope.review_required):
            raise EvidenceSelectionError(
                "REVIEW_REQUIRED_EVIDENCE_SELECTED",
                "Review-required evidence cannot be independently selected.",
            )
        if decision.selected and not _independent_candidate_allowed(
            scope,
            source.product_resolution.target.short_name,
            settings or DatasetEvidenceSettings(),
        ):
            raise EvidenceSelectionError(
                "UNTRACEABLE_EVIDENCE_SELECTED",
                "Selected evidence must be confidently traceable to the target.",
            )


def _call_bounded(
    client: ModelClient,
    factory: Callable[[], ModelRequest[EvidenceSelectionModelResponse]],
    validator: Callable[[EvidenceSelectionModelResponse], None],
    max_retries: int,
    calls: list[int],
) -> EvidenceSelectionModelResponse:
    last: Exception | None = None
    for _ in range(max_retries + 1):
        try:
            calls[0] += 1
            parsed = client.generate_structured(factory()).parsed
            validator(parsed)
            return parsed
        except Exception as exc:
            last = exc
    raise EvidenceSelectionError(
        "EVIDENCE_SELECTION_RETRIES_EXHAUSTED",
        "Evidence structured response failed after bounded retries.",
    ) from last


def _consolidate(
    decisions: list[EvidenceSelectionDecision],
    manifest: DatasetExtractionManifest,
) -> tuple[EvidenceSelectionDecision, ...]:
    by_id: dict[str, EvidenceSelectionDecision] = {}
    for item in decisions:
        previous = by_id.get(item.block_id)
        if previous is not None and previous != item:
            raise EvidenceSelectionError(
                "CONFLICTING_BLOCK_DECISIONS",
                "Repeated block evaluations returned conflicting decisions.",
            )
        by_id[item.block_id] = item
    order = {block.block_id: block.source_order for block in manifest.blocks}
    return tuple(sorted(by_id.values(), key=lambda item: order[item.block_id]))


def _selected_packet_ids(
    decisions: tuple[EvidenceSelectionDecision, ...], source: EvidenceSelectionInput
) -> tuple[str, ...]:
    """Expand selected science targets with only their explicitly linked context."""
    scope_by_id = {item.block_id: item for item in source.product_resolution.block_decisions}
    selected = {item.block_id for item in decisions if item.selected}
    for item in decisions:
        if not item.selected:
            continue
        selected.update(
            linked
            for linked in item.qualifying_block_ids
            if linked in scope_by_id
            and scope_by_id[linked].eligibility is EvidenceEligibility.DOCUMENT_CONTEXT
        )
    order = {block.block_id: block.source_order for block in source.extraction_manifest.blocks}
    return tuple(sorted(selected, key=order.__getitem__))


def _measure(ids: tuple[str, ...], manifest: DatasetExtractionManifest) -> tuple[int, int, int]:
    by_id = {block.block_id: block for block in manifest.blocks}
    characters = sum(len(by_id[item].text) for item in ids)
    return len(ids), characters, (characters + 3) // 4


def _hash_components(
    source: EvidenceSelectionInput,
    blocks: tuple[EvidencePacketBlock, ...],
    decisions: tuple[EvidenceSelectionDecision, ...],
    limit: PacketLimitRecord,
    settings: DatasetEvidenceSettings,
) -> PacketComponentHashes:
    manifest = source.extraction_manifest
    resolution = source.product_resolution
    return PacketComponentHashes(
        target_sha256=canonical_sha256(resolution.target.model_dump(mode="json")),
        source_sha256=canonical_sha256(
            {
                "cmr_source_sha256": manifest.lineage.cmr_source_sha256,
                "readme_sha256": manifest.document.sha256,
                "extraction_manifest_sha256": resolution.extraction_manifest_sha256,
                "document_id": manifest.document_id,
            }
        ),
        ordered_block_identity_sha256=canonical_sha256(
            [
                {
                    "block_id": item.block.block_id,
                    "block_sha256": item.block_sha256,
                    "source_order": item.block.source_order,
                    "page": item.block.page_start,
                    "section_id": item.block.section_id,
                }
                for item in blocks
            ]
        ),
        ordered_source_text_sha256=canonical_sha256(
            [{"block_id": item.block.block_id, "text": item.block.text} for item in blocks]
        ),
        product_scope_sha256=canonical_sha256(
            [
                {
                    "block_id": item.block.block_id,
                    "product_scope": item.product_scope.model_dump(mode="json"),
                    "eligibility": item.evidence_eligibility.value,
                }
                for item in blocks
            ]
        ),
        selection_decisions_sha256=canonical_sha256(
            [item.model_dump(mode="json") for item in decisions]
        ),
        policy_versions_sha256=canonical_sha256(
            {
                "selection": settings.selection_prompt_version,
                "grouping": settings.grouping_policy_version,
                "roles": settings.evidence_role_policy_version,
                "minimality": settings.minimality_policy_version,
                "builder": settings.packet_builder_version,
                "limits": settings.packet_limit_policy_version,
                "response_schema": settings.response_schema_version_evidence,
            }
        ),
        limit_lineage_sha256=canonical_sha256(limit.model_dump(mode="json")),
    )


def _packet_from(
    source: EvidenceSelectionInput,
    decisions: tuple[EvidenceSelectionDecision, ...],
    limit: PacketLimitRecord,
    settings: DatasetEvidenceSettings,
) -> DatasetEvidencePacket:
    manifest = source.extraction_manifest
    resolution = source.product_resolution
    block_by_id = {block.block_id: block for block in manifest.blocks}
    scope_by_id = {item.block_id: item for item in resolution.block_decisions}
    final_ids = set(limit.final_block_ids)
    decision_by_id = {item.block_id: item for item in decisions}
    packet_blocks: list[EvidencePacketBlock] = []
    for block_id in sorted(final_ids, key=lambda item: block_by_id[item].source_order):
        block = block_by_id[block_id]
        scope = scope_by_id[block_id]
        decision = decision_by_id.get(block_id)
        is_context = scope.eligibility is EvidenceEligibility.DOCUMENT_CONTEXT
        if scope.scope_type is ProductScopeType.MIXED_PRODUCT_TRANSITION or (
            scope.eligibility is EvidenceEligibility.REVIEW_ONLY
        ):
            raise EvidenceSelectionError(
                "PACKET_INELIGIBLE_PRODUCT_SCOPE",
                "Mixed or review-only product scope cannot enter an evidence packet.",
            )
        if decision is None and not is_context:
            raise EvidenceSelectionError(
                "PACKET_DECISION_MISSING", "Packet block lacks a validated selection decision."
            )
        linked_targets = tuple(
            item.block_id
            for item in decisions
            if item.selected and block_id in item.qualifying_block_ids
        )
        packet_blocks.append(
            EvidencePacketBlock(
                block=block,
                block_sha256=block_sha256(block.text),
                evidence_role=(
                    EvidenceRole.QUALIFICATION if is_context else decision.evidence_role
                ),
                scientific_relevance=(
                    ScientificRelevance.CONTEXT_ONLY
                    if is_context
                    else decision.scientific_relevance
                ),
                product_scope=ProductScopeDecision(
                    scope_type=scope.scope_type,
                    applicable_short_names=scope.applicable_products,
                    family_label=scope.family_label,
                    product_match_confidence=scope.product_match_confidence,
                    rationale=scope.rationale,
                    qualifying_block_ids=scope.qualifying_block_ids,
                    review_required=scope.review_required,
                ),
                evidence_eligibility=scope.eligibility,
                selection_rationale=(
                    "Explicitly linked context required to interpret selected scientific evidence."
                    if is_context
                    else decision.rationale
                ),
                qualifying_block_ids=(
                    scope.qualifying_block_ids if is_context else decision.qualifying_block_ids
                ),
                qualifying_declaration_ids=(
                    scope.declaration_ids if is_context else decision.qualifying_declaration_ids
                ),
                context_for_block_ids=linked_targets,
                review_required=(
                    scope.review_required
                    if is_context
                    else decision.review_required or scope.review_required
                ),
            )
        )
    if not packet_blocks:
        raise EvidenceSelectionError(
            "NO_USABLE_EVIDENCE", "No eligible scientific source block was selected."
        )
    components = _hash_components(source, tuple(packet_blocks), decisions, limit, settings)
    review = resolution.review_required or any(
        item.review_required and item.scientific_relevance is ScientificRelevance.RELEVANT
        for item in packet_blocks
    )
    readiness = PacketReadiness.REVIEW_REQUIRED if review else PacketReadiness.CLASSIFICATION_READY
    logical = {
        "schema_version": "2.0",
        "packet_version": settings.packet_builder_version,
        "target": resolution.target.model_dump(mode="json"),
        "cmr_source_sha256": manifest.lineage.cmr_source_sha256,
        "readme_sha256": manifest.document.sha256,
        "extraction_manifest_sha256": resolution.extraction_manifest_sha256,
        "coverage_mode": resolution.coverage_mode.value,
        "coverage_confidence": resolution.product_match_confidence.model_dump(mode="json"),
        "blocks": [item.model_dump(mode="json") for item in packet_blocks],
        "excluded_block_ids": resolution.excluded_other_product_block_ids,
        "ambiguous_block_ids": resolution.ambiguous_review_block_ids,
        "context_only_block_ids": resolution.document_context_block_ids,
        "packet_readiness": readiness.value,
        "review_status": ("pending" if review else "not_required"),
        "limit_record": limit.model_dump(mode="json"),
        "component_hashes": components.model_dump(mode="json"),
    }
    packet_hash = canonical_sha256(logical)
    return DatasetEvidencePacket(
        schema_version="2.0",
        packet_id=f"sha256:{packet_hash}",
        packet_version=settings.packet_builder_version,
        target=resolution.target,
        cmr_source_sha256=manifest.lineage.cmr_source_sha256,
        readme_sha256=manifest.document.sha256,
        extraction_manifest_sha256=resolution.extraction_manifest_sha256,
        coverage_mode=resolution.coverage_mode,
        coverage_confidence=resolution.product_match_confidence,
        selection_version=settings.selection_prompt_version,
        grouping_policy_version=settings.grouping_policy_version,
        evidence_role_policy_version=settings.evidence_role_policy_version,
        minimality_policy_version=settings.minimality_policy_version,
        response_schema_version=settings.response_schema_version_evidence,
        blocks=tuple(packet_blocks),
        excluded_block_ids=resolution.excluded_other_product_block_ids,
        ambiguous_block_ids=resolution.ambiguous_review_block_ids,
        context_only_block_ids=resolution.document_context_block_ids,
        character_count=limit.final_character_count,
        estimated_tokens=limit.final_estimated_tokens,
        packet_readiness=(readiness),
        review_status=(DatasetReviewStatus.PENDING if review else DatasetReviewStatus.NOT_REQUIRED),
        limit_record=limit,
        component_hashes=components,
        warnings=resolution.warnings,
        packet_sha256=packet_hash,
    )


def validate_packet_identity(packet: DatasetEvidencePacket) -> None:
    """Revalidate stable packet identity and recomputable components at a consumer boundary."""
    components = packet.component_hashes
    limit = packet.limit_record
    if components is None or limit is None:
        raise EvidenceSelectionError(
            "PACKET_IDENTITY_INCOMPLETE", "Packet lacks component hashes or limit lineage."
        )
    for item in packet.blocks:
        if item.block_sha256 != block_sha256(item.block.text):
            raise EvidenceSelectionError(
                "PACKET_BLOCK_HASH_INVALID", "Packet block text does not match its hash."
            )
        if item.block.document_id != f"sha256:{packet.readme_sha256}":
            raise EvidenceSelectionError(
                "PACKET_DOCUMENT_ID_INVALID", "Packet block has a cross-document identity."
            )
    expected = {
        "target_sha256": canonical_sha256(packet.target.model_dump(mode="json")),
        "source_sha256": canonical_sha256(
            {
                "cmr_source_sha256": packet.cmr_source_sha256,
                "readme_sha256": packet.readme_sha256,
                "extraction_manifest_sha256": packet.extraction_manifest_sha256,
                "document_id": f"sha256:{packet.readme_sha256}",
            }
        ),
        "ordered_block_identity_sha256": canonical_sha256(
            [
                {
                    "block_id": item.block.block_id,
                    "block_sha256": item.block_sha256,
                    "source_order": item.block.source_order,
                    "page": item.block.page_start,
                    "section_id": item.block.section_id,
                }
                for item in packet.blocks
            ]
        ),
        "ordered_source_text_sha256": canonical_sha256(
            [{"block_id": item.block.block_id, "text": item.block.text} for item in packet.blocks]
        ),
        "product_scope_sha256": canonical_sha256(
            [
                {
                    "block_id": item.block.block_id,
                    "product_scope": item.product_scope.model_dump(mode="json"),
                    "eligibility": item.evidence_eligibility.value,
                }
                for item in packet.blocks
            ]
        ),
        "policy_versions_sha256": canonical_sha256(
            {
                "selection": packet.selection_version,
                "grouping": packet.grouping_policy_version,
                "roles": packet.evidence_role_policy_version,
                "minimality": packet.minimality_policy_version,
                "builder": packet.packet_version,
                "limits": limit.policy_version,
                "response_schema": packet.response_schema_version,
            }
        ),
        "limit_lineage_sha256": canonical_sha256(limit.model_dump(mode="json")),
    }
    serialized = components.model_dump(mode="json")
    if any(serialized[name] != value for name, value in expected.items()):
        raise EvidenceSelectionError(
            "PACKET_COMPONENT_HASH_INVALID", "Packet component hash validation failed."
        )
    logical = {
        "schema_version": packet.schema_version,
        "packet_version": packet.packet_version,
        "target": packet.target.model_dump(mode="json"),
        "cmr_source_sha256": packet.cmr_source_sha256,
        "readme_sha256": packet.readme_sha256,
        "extraction_manifest_sha256": packet.extraction_manifest_sha256,
        "coverage_mode": packet.coverage_mode.value,
        "coverage_confidence": packet.coverage_confidence.model_dump(mode="json"),
        "blocks": [item.model_dump(mode="json") for item in packet.blocks],
        "excluded_block_ids": packet.excluded_block_ids,
        "ambiguous_block_ids": packet.ambiguous_block_ids,
        "context_only_block_ids": packet.context_only_block_ids,
        "packet_readiness": packet.packet_readiness.value,
        "review_status": packet.review_status.value,
        "limit_record": limit.model_dump(mode="json"),
        "component_hashes": components.model_dump(mode="json"),
    }
    digest = canonical_sha256(logical)
    if packet.packet_sha256 != digest or packet.packet_id != f"sha256:{digest}":
        raise EvidenceSelectionError(
            "PACKET_HASH_INVALID", "Packet logical identity validation failed."
        )


def validate_packet_against_manifest(
    packet: DatasetEvidencePacket, manifest: DatasetExtractionManifest
) -> None:
    """Prove packet source content and provenance equal authoritative records."""
    authoritative = {block.block_id: block for block in manifest.blocks}
    for item in packet.blocks:
        source = authoritative.get(item.block.block_id)
        if (
            source is None
            or item.block != source
            or item.block_sha256 != block_sha256(source.text)
            or item.block.document_id != manifest.document_id
            or item.block.pdf_sha256 != manifest.document.sha256
        ):
            raise EvidenceSelectionError(
                "PACKET_SOURCE_MISMATCH",
                "Packet source content or provenance differs from the extraction manifest.",
            )


def _minimality_request(
    source: EvidenceSelectionInput,
    ids: tuple[str, ...],
    policy: DatasetEvidenceSettings,
    model: ModelSettings,
) -> ModelRequest[EvidenceMinimizationModelResponse]:
    return ModelRequest(
        stage=ModelStage.DATASET_EVIDENCE_SELECTION,
        prompt=build_dataset_evidence_minimization_prompt(
            manifest=source.extraction_manifest,
            selected_block_ids=ids,
            preferred_blocks=policy.preferred_packet_blocks,
            preferred_characters=policy.preferred_packet_characters,
        ),
        response_schema=EvidenceMinimizationModelResponse,
        provider=model.provider,
        model_name=model.model_name,
        prompt_version=policy.minimality_prompt_version,
        temperature=model.temperature,
        timeout_seconds=model.timeout_seconds,
        metadata={"blind_stage": True, "minimality": True},
    )


def _validate_minimality(response: EvidenceMinimizationModelResponse, ids: tuple[str, ...]) -> None:
    actual = [item.block_id for item in response.decisions]
    if len(actual) != len(set(actual)) or set(actual) != set(ids):
        raise EvidenceSelectionError(
            "EVIDENCE_MINIMALITY_PARTITION_INVALID",
            "Minimality decisions must exactly partition supplied selected block IDs.",
        )
    if not any(item.retain for item in response.decisions):
        raise EvidenceSelectionError(
            "EVIDENCE_MINIMALITY_EMPTY", "Minimality cannot remove all independent science."
        )


def _audit(
    source: EvidenceSelectionInput,
    policy: DatasetEvidenceSettings,
    groups: tuple[EvidenceBlockGroup, ...],
    decisions: tuple[EvidenceSelectionDecision, ...],
    *,
    calls: int,
    minimality_calls: int,
    initial_ids: tuple[str, ...],
    final_independent_ids: tuple[str, ...],
    final_packet_ids: tuple[str, ...],
    exclusion_records: tuple[EvidenceExclusionRecord, ...],
) -> EvidenceSelectionAudit:
    context_ids = set(source.product_resolution.document_context_block_ids)
    return EvidenceSelectionAudit(
        grouping_policy_version=policy.grouping_policy_version,
        prompt_version=policy.selection_prompt_version,
        response_schema_version=policy.response_schema_version_evidence,
        groups=groups,
        decisions=decisions,
        excluded_decisions=tuple(
            item
            for item in source.product_resolution.block_decisions
            if item.eligibility is EvidenceEligibility.EXCLUDED
        ),
        ambiguous_decisions=tuple(
            item
            for item in source.product_resolution.block_decisions
            if item.eligibility is EvidenceEligibility.REVIEW_ONLY
        ),
        model_calls=calls,
        selection_model_calls=calls - minimality_calls,
        minimality_model_calls=minimality_calls,
        minimality_policy_version=policy.minimality_policy_version,
        minimality_ran=minimality_calls > 0,
        candidate_block_count=sum(
            len([m for m in group.members if m.membership is GroupMembership.DECISION_TARGET])
            for group in groups
        ),
        initially_selected_block_count=len(initial_ids),
        final_independent_block_count=len(final_independent_ids),
        final_context_block_count=sum(item in context_ids for item in final_packet_ids),
        retained_block_ids=final_packet_ids,
        removed_block_ids=tuple(item for item in initial_ids if item not in final_independent_ids),
        exclusion_records=exclusion_records,
        blind_cache_identity_sha256=canonical_sha256(
            {
                "target": source.product_resolution.target.model_dump(mode="json"),
                "manifest": source.product_resolution.extraction_manifest_sha256,
                "product_resolution": source.product_resolution.blind_cache_identity_sha256,
                "selection": policy.selection_prompt_version,
                "selection_schema": policy.response_schema_version_evidence,
                "grouping": policy.grouping_policy_version,
                "roles": policy.evidence_role_policy_version,
                "minimality_prompt": policy.minimality_prompt_version,
                "minimality_policy": policy.minimality_policy_version,
                "minimality_schema": policy.minimality_response_schema_version,
                "preferred_blocks": policy.preferred_packet_blocks,
                "preferred_characters": policy.preferred_packet_characters,
                "hard_blocks": policy.max_packet_blocks,
                "hard_characters": policy.max_packet_characters,
            }
        ),
    )


def select_evidence_packet(
    source: EvidenceSelectionInput,
    *,
    model_client: ModelClient,
    model_settings: ModelSettings | None = None,
    evidence_settings: DatasetEvidenceSettings | None = None,
) -> EvidencePacketResult | EvidenceNoUsableScienceOutcome | EvidenceSelectionFailure:
    """Select independent science, minimize once when useful, and build a faithful packet."""
    model = model_settings or ModelSettings()
    policy = evidence_settings or DatasetEvidenceSettings()
    calls = [0]
    try:
        if not source.product_resolution.downstream_classification_eligible:
            raise EvidenceSelectionError(
                "REVIEW_REQUIRED_PRODUCT_SCOPE",
                "Product scope is not eligible for automatic evidence selection.",
            )
        groups = build_evidence_groups(source, policy)
        exclusions = list(_candidate_exclusions(source, policy))
        decisions: tuple[EvidenceSelectionDecision, ...] = ()
        if groups:
            collected: list[EvidenceSelectionDecision] = []
            for group in groups:
                response = _call_bounded(
                    model_client,
                    lambda group=group: _request(
                        build_dataset_evidence_selection_prompt(
                            manifest=source.extraction_manifest,
                            resolution=source.product_resolution,
                            group=group,
                        ),
                        policy,
                        model,
                        reduction=False,
                    ),
                    lambda response, group=group: _validate_response(
                        response,
                        group=group,
                        source=source,
                        reduction=False,
                        settings=policy,
                    ),
                    model.max_retries,
                    calls,
                )
                collected.extend(response.decisions)
            decisions = _consolidate(collected, source.extraction_manifest)

        initial_independent = tuple(item.block_id for item in decisions if item.selected)
        for item in decisions:
            if not item.selected:
                exclusions.append(
                    EvidenceExclusionRecord(
                        block_id=item.block_id,
                        reason=EvidenceExclusionReason.MODEL_NOT_SELECTED,
                        phase="selection",
                    )
                )

        # Exact source duplicates are deterministically removed before the one model pass.
        block_by_id = {item.block_id: item for item in source.extraction_manifest.blocks}
        seen_hashes: set[str] = set()
        deduplicated: list[str] = []
        for block_id in initial_independent:
            digest = block_sha256(block_by_id[block_id].text)
            if digest in seen_hashes:
                exclusions.append(
                    EvidenceExclusionRecord(
                        block_id=block_id,
                        reason=EvidenceExclusionReason.DUPLICATE_SOURCE,
                        phase="consolidation",
                    )
                )
            else:
                seen_hashes.add(digest)
                deduplicated.append(block_id)
        final_independent = tuple(deduplicated)
        initial_count, initial_chars, _ = _measure(final_independent, source.extraction_manifest)
        minimality_calls = 0
        soft_exceeded = (
            initial_count > policy.preferred_packet_blocks
            or initial_chars > policy.preferred_packet_characters
            or initial_count > policy.max_packet_blocks
            or initial_chars > policy.max_packet_characters
        )
        removal_rationales: dict[str, str] = {}
        if final_independent and soft_exceeded:
            before = calls[0]
            response = _call_bounded(
                model_client,
                lambda: _minimality_request(source, final_independent, policy, model),
                lambda response: _validate_minimality(response, final_independent),
                model.max_retries,
                calls,
            )
            minimality_calls = calls[0] - before
            retained = {item.block_id for item in response.decisions if item.retain}
            for item in response.decisions:
                if not item.retain:
                    exclusions.append(
                        EvidenceExclusionRecord(
                            block_id=item.block_id,
                            reason=EvidenceExclusionReason.MINIMALITY_REMOVED,
                            phase="minimality",
                        )
                    )
                    removal_rationales[item.block_id] = (
                        item.removal_reason or "non_independent_science"
                    )
            final_independent = tuple(item for item in final_independent if item in retained)

        selected_set = set(final_independent)
        decisions = tuple(
            item.model_copy(update={"selected": item.block_id in selected_set})
            for item in decisions
        )
        final_packet_ids = _selected_packet_ids(decisions, source)
        if not final_independent:
            audit = _audit(
                source,
                policy,
                groups,
                decisions,
                calls=calls[0],
                minimality_calls=minimality_calls,
                initial_ids=initial_independent,
                final_independent_ids=(),
                final_packet_ids=(),
                exclusion_records=tuple(exclusions),
            )
            return EvidenceNoUsableScienceOutcome(selection_audit=audit)

        initial_packet_ids = _selected_packet_ids(
            tuple(
                item.model_copy(update={"selected": item.block_id in set(initial_independent)})
                for item in decisions
            ),
            source,
        )
        initial_count, initial_chars, initial_tokens = _measure(
            initial_packet_ids, source.extraction_manifest
        )
        final_count, final_chars, final_tokens = _measure(
            final_packet_ids, source.extraction_manifest
        )
        within = (
            final_count <= policy.max_packet_blocks and final_chars <= policy.max_packet_characters
        )
        limit = PacketLimitRecord(
            policy_version=policy.packet_limit_policy_version,
            measurement_version=policy.token_estimate_version,
            max_blocks=policy.max_packet_blocks,
            max_characters=policy.max_packet_characters,
            preferred_blocks=policy.preferred_packet_blocks,
            preferred_characters=policy.preferred_packet_characters,
            initial_block_ids=initial_packet_ids,
            final_block_ids=final_packet_ids,
            removed_block_ids=tuple(
                item for item in initial_independent if item not in final_independent
            ),
            removal_rationales=removal_rationales,
            initial_block_count=initial_count,
            final_block_count=final_count,
            initial_character_count=initial_chars,
            final_character_count=final_chars,
            initial_estimated_tokens=initial_tokens,
            final_estimated_tokens=final_tokens,
            reduction_required=soft_exceeded,
            reduction_passes=1 if soft_exceeded else 0,
            minimality_policy_version=policy.minimality_policy_version,
            minimality_model_calls=minimality_calls,
            within_limits=within,
        )
        if not within:
            raise EvidenceSelectionError(
                "EVIDENCE_PACKET_LIMIT_EXCEEDED",
                "Evidence packet remains above a configured hard limit.",
            )
        packet = _packet_from(source, decisions, limit, policy)
        validate_packet_against_manifest(packet, source.extraction_manifest)
        audit = _audit(
            source,
            policy,
            groups,
            decisions,
            calls=calls[0],
            minimality_calls=minimality_calls,
            initial_ids=initial_independent,
            final_independent_ids=final_independent,
            final_packet_ids=final_packet_ids,
            exclusion_records=tuple(exclusions),
        )
        return EvidencePacketResult(packet=packet, selection_audit=audit)
    except Exception as exc:
        code = exc.code if isinstance(exc, EvidenceSelectionError) else "EVIDENCE_SELECTION_FAILED"
        return EvidenceSelectionFailure(
            error=DatasetErrorRecord(
                code=code,
                message="Evidence selection or packet construction failed.",
                stage="evidence_selection",
            ),
            model_calls=calls[0],
        )


def classifier_input(packet: DatasetEvidencePacket) -> DatasetClassifierInput:
    """Build the only permitted future dataset-classifier README-text boundary."""
    return DatasetClassifierInput(target=packet.target, evidence_packet=packet)