"""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)