"""Dataset-only typed contracts for the Dataset README Classifier.""" from __future__ import annotations from enum import Enum from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator from gcmd_classifier.models import HierarchyLevel class DatasetProcessingStatus(str, Enum): # noqa: UP042 """Execution state for one dataset workflow.""" COMPLETED = "completed" PARTIAL = "partial" FAILED = "failed" SKIPPED = "skipped" class DatasetClassificationOutcome(str, Enum): # noqa: UP042 """Semantic dataset outcome after classification is attempted.""" CLASSIFIED = "classified" PENDING_REVIEW = "pending_review" NOT_CLASSIFIED = "not_classified" class DatasetReviewStatus(str, Enum): # noqa: UP042 """Human-review state for a dataset result.""" NOT_REQUIRED = "not_required" PENDING = "pending" COMPLETED = "completed" class DatasetClassificationFinalStatus(str, Enum): # noqa: UP042 """Automated status for one dataset classification.""" ACCEPTED = "accepted" REDUCED_TO_ANCESTOR = "reduced_to_ancestor" REVIEW_REQUIRED = "review_required" REJECTED = "rejected" class DatasetStage(str, Enum): # noqa: UP042 """Dataset workflow stages used by diagnostics.""" SOURCE = "source" README_DISCOVERY = "readme_discovery" README_RETRIEVAL = "readme_retrieval" EXTRACTION = "extraction" PRODUCT_RESOLUTION = "product_resolution" EVIDENCE_SELECTION = "evidence_selection" CLASSIFICATION = "classification" VALIDATION = "validation" BLIND_PERSISTENCE = "blind_persistence" COMPARISON = "comparison" class CoverageMode(str, Enum): # noqa: UP042 """Supported README document coverage modes.""" SINGLE_PRODUCT = "single_product" EXPLICIT_MULTI_PRODUCT = "explicit_multi_product" PRODUCT_FAMILY = "product_family" VAGUE_PRODUCT_SCOPE = "vague_product_scope" UNDETERMINED = "undetermined" class ProductScopeType(str, Enum): # noqa: UP042 """How an evidence block applies to products.""" EXACT_TARGET = "exact_target" ENUMERATED_TARGET = "enumerated_target" PRODUCT_FAMILY = "product_family" AMBIGUOUS = "ambiguous" OTHER_PRODUCT = "other_product" DOCUMENT_CONTEXT = "document_context" MIXED_PRODUCT_TRANSITION = "mixed_product_transition" class EvidenceEligibility(str, Enum): # noqa: UP042 """Product applicability eligibility before evidence selection.""" ELIGIBLE = "eligible" EXCLUDED = "excluded" REVIEW_ONLY = "review_only" DOCUMENT_CONTEXT = "document_context" class EvidenceRole(str, Enum): # noqa: UP042 """Scientific or contextual role of one source evidence block.""" PRODUCT_IDENTITY = "product_identity" SCIENTIFIC_DESCRIPTION = "scientific_description" MEASURED_VARIABLE = "measured_variable" GEOPHYSICAL_PARAMETER = "geophysical_parameter" METHOD_OR_ALGORITHM = "method_or_algorithm" SPATIAL_TEMPORAL_CONTEXT = "spatial_temporal_context" PROCESSING_OR_FORMAT = "processing_or_format" QUALIFICATION = "qualification" OTHER = "other" class ComparisonCategory(str, Enum): # noqa: UP042 """Post-inference relationship between blind and expert assignments.""" ALREADY_ASSIGNED = "already_assigned" PROPOSED_ADDITION = "proposed_addition" PROPOSED_REFINEMENT = "proposed_refinement" REDUNDANT_RESULT = "redundant_result" REVIEW_REQUIRED = "review_required" class PageExtractionStatus(str, Enum): # noqa: UP042 """Deterministic page extraction outcome.""" EXTRACTED = "extracted" BLANK = "blank" IMAGE_ONLY = "image_only" class HeadingKind(str, Enum): # noqa: UP042 """Literal or deterministic fallback heading kind.""" NUMBERED = "numbered" ALL_CAPS = "all_caps" COLON_LABEL = "colon_label" DERIVED_DOCUMENT = "derived_document" class SectionLabelSource(str, Enum): # noqa: UP042 """Whether section metadata is literal source text or derived.""" LITERAL = "literal" DERIVED = "derived" NONE = "none" class DatasetModel(BaseModel): """Strict immutable base for serialized dataset contracts.""" model_config = ConfigDict(extra="forbid", frozen=True) class DatasetWarning(DatasetModel): """Non-fatal dataset workflow diagnostic.""" code: str = Field(min_length=1) message: str = Field(min_length=1) stage: DatasetStage | None = None details: dict[str, Any] | None = None class DatasetErrorRecord(DatasetModel): """Structured dataset workflow failure.""" code: str = Field(min_length=1) message: str = Field(min_length=1) stage: DatasetStage retry_count: int | None = Field(default=None, ge=0) details: dict[str, Any] | None = None class DatasetHierarchyAttemptDiagnostic(DatasetModel): """Sanitized facts for one failed dataset hierarchy model attempt.""" pipeline_stage: Literal["classification"] = "classification" model_stage: Literal["dataset_topic", "dataset_term", "dataset_variable"] attempt_number: int = Field(ge=1) retryable: bool failure_category: Literal[ "model_non_retryable", "model_retryable", "structured_response", "response_schema", "deterministic_validation", "internal", ] public_error_code: str = Field(min_length=1) provider: str = Field(min_length=1) model_name: str = Field(min_length=1) provider_http_status: int | None = Field(default=None, ge=100, le=599) provider_error_type: str | None = Field(default=None, max_length=128) provider_error_code: str | None = Field(default=None, max_length=128) provider_error_parameter: str | None = Field(default=None, max_length=128) provider_message: str | None = Field(default=None, max_length=256) provider_request_id: str | None = Field(default=None, max_length=256) packet_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") parent_uuid: str | None = None parent_name: str | None = None parent_level: str | None = None parent_canonical_path: str | None = None branch_id: str = Field(min_length=1) candidate_count: int = Field(ge=0) candidate_ids: tuple[str, ...] = Field(default_factory=tuple) candidate_ids_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") response_schema_name: str = Field(min_length=1) prompt_version: str = Field(min_length=1) prompt_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") prompt_character_count: int = Field(ge=0) estimated_prompt_tokens: int = Field(ge=0) elapsed_seconds: float = Field(ge=0.0) parsed_response_received: bool structured_validation_code: str | None = Field(default=None, max_length=128) class SourceArtifactReference(DatasetModel): """Immutable reference to exact source bytes.""" artifact_id: str = Field(min_length=1) sha256: str = Field(pattern=r"^[0-9a-f]{64}$") byte_size: int = Field(ge=0) media_type: str | None = None storage_reference: str | None = None class SubmittedCMRRequest(DatasetModel): """Metadata for the submitted CMR collection request.""" submitted_url: str = Field(min_length=1) submitted_native_id: str = Field(min_length=1) class DatasetIdentity(DatasetModel): """Authoritative CMR collection identity.""" concept_id: str = Field(min_length=1) native_id: str = Field(min_length=1) short_name: str = Field(min_length=1) version: str = Field(min_length=1) cmr_revision_id: int = Field(ge=1) legacy_entry_id: str | None = Field(default=None, exclude=True) @model_validator(mode="after") def native_id_matches_source_fields(self) -> DatasetIdentity: """Require the representation-specific CMR Native ID derivation.""" expected = f"{self.short_name}_{self.version}" if self.native_id != expected: raise ValueError("native_id must equal short_name + '_' + version") return self class CMRSourceRecord(DatasetModel): """Preserved CMR source artifact and selected collection identity.""" request: SubmittedCMRRequest source_artifact: SourceArtifactReference identity: DatasetIdentity retrieved_at: str | None = None final_url: str | None = None class SealedExpertKeywordInput(DatasetModel): """Opaque reference to expert keywords excluded from blind inference.""" sealed_artifact_id: str = Field(min_length=1) sealed_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") keyword_count: int = Field(ge=0) class BlindCollectionView(DatasetModel): """Allow-listed dataset metadata safe for blind stages.""" identity: DatasetIdentity derived_native_id: str = Field(min_length=1) entry_title: str | None = None summary: str | None = None related_urls: tuple[dict[str, Any], ...] = Field(default_factory=tuple) blind_view_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") expert_keywords_excluded: Literal[True] = True @model_validator(mode="after") def derived_identity_is_consistent(self) -> BlindCollectionView: if self.derived_native_id != self.identity.native_id: raise ValueError("derived_native_id must equal the validated Native ID") return self class READMECandidateState(str, Enum): # noqa: UP042 """Cardinality of usable exact READ-ME candidates.""" ZERO = "zero" ONE = "one" MULTIPLE = "multiple" class READMECandidate(DatasetModel): """One exact READ-ME entry discovered from source metadata.""" candidate_id: str = Field(min_length=1) source_index: int = Field(ge=0) url: str = Field(min_length=1) description: str | None = None type: str | None = None subtype: Literal["READ-ME"] = "READ-ME" source_entry: dict[str, Any] class READMEDiscoveryResult(DatasetModel): """Exact candidates discovered from one validated blind collection.""" identity: DatasetIdentity cmr_source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") candidate_state: READMECandidateState candidates: tuple[READMECandidate, ...] = Field(default_factory=tuple) warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) @model_validator(mode="after") def cardinality_matches_state(self) -> READMEDiscoveryResult: expected = ( READMECandidateState.ZERO if not self.candidates else READMECandidateState.ONE if len(self.candidates) == 1 else READMECandidateState.MULTIPLE ) if self.candidate_state is not expected: raise ValueError("candidate_state must match candidate cardinality") return self class READMESelection(DatasetModel): """Explicit user selection of one discovered README candidate.""" candidate_id: str = Field(min_length=1) selected_url: str = Field(min_length=1) selected_at: str | None = None class ValidatedREADMESelection(DatasetModel): """Explicit selection bound to exact CMR discovery provenance.""" concept_id: str = Field(min_length=1) native_id: str = Field(min_length=1) short_name: str = Field(min_length=1) version: str = Field(min_length=1) cmr_revision_id: int = Field(ge=1) cmr_source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") candidate_id: str = Field(min_length=1) source_index: int = Field(ge=0) selected_url: str = Field(min_length=1) source_entry: dict[str, Any] selected_at: str selection_binding_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") explicitly_confirmed: Literal[True] = True class RedirectRecord(DatasetModel): """One redirect retained as retrieval provenance.""" status_code: int = Field(ge=300, le=399) source_url: str = Field(min_length=1) target_url: str = Field(min_length=1) class READMERetrievalRecord(DatasetModel): """Sanitized provenance for a safely retrieved public README PDF.""" identity: DatasetIdentity cmr_source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") selected_candidate_id: str = Field(min_length=1) selected_source_index: int = Field(ge=0) selected_source_entry: dict[str, Any] submitted_url: str = Field(min_length=1) final_url: str = Field(min_length=1) redirects: tuple[RedirectRecord, ...] = Field(default_factory=tuple) artifact: SourceArtifactReference retrieved_at: str status_code: int = Field(ge=200, le=299) response_metadata: dict[str, str] = Field(default_factory=dict) server_filename: str | None = None detected_media_type: Literal["application/pdf"] = "application/pdf" public_access_validated: Literal[True] = True public_address_validation: str = Field(min_length=1) document_validated: Literal[True] = True document_validation_method: str = Field(min_length=1) class ExtractedSeparator(DatasetModel): """Explicit separator omitted between adjacent blocks, if any.""" page_number: int = Field(ge=1) character_start: int = Field(ge=0) character_end: int = Field(ge=0) value: str separator_type: str = Field(min_length=1) policy_version: str = Field(min_length=1) class ExtractionLineage(DatasetModel): """Blind-safe lineage from a validated PDF to its CMR selection.""" identity: DatasetIdentity cmr_source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") selected_candidate_id: str = Field(min_length=1) selected_source_index: int = Field(ge=0) selected_source_entry: dict[str, Any] selected_readme_url: str = Field(min_length=1) final_readme_url: str = Field(min_length=1) retrieval_timestamp: str = Field(min_length=1) retrieval_status_code: int = Field(ge=200, le=299) public_address_validation: str = Field(min_length=1) document_validation_method: str = Field(min_length=1) class ExtractedSection(DatasetModel): """Literal or derived section metadata with ordered membership.""" section_id: str = Field(pattern=r"^s[0-9]{4}$") source_order: int = Field(ge=0) heading: str | None = None normalized_heading: str | None = None heading_kind: HeadingKind label_source: SectionLabelSource heading_page_number: int | None = Field(default=None, ge=1) heading_character_start: int | None = Field(default=None, ge=0) heading_character_end: int | None = Field(default=None, ge=0) page_start: int = Field(ge=1) page_end: int = Field(ge=1) block_ids: tuple[str, ...] = Field(default_factory=tuple) @model_validator(mode="after") def validate_section_provenance(self) -> ExtractedSection: if self.page_end < self.page_start: raise ValueError("page_end must be greater than or equal to page_start") spans = (self.heading_page_number, self.heading_character_start, self.heading_character_end) if self.label_source is SectionLabelSource.LITERAL: if self.heading is None or any(value is None for value in spans): raise ValueError("literal headings require exact page-text provenance") elif any(value is not None for value in spans[1:]): raise ValueError("derived headings cannot claim literal character spans") return self class ExtractedBlock(DatasetModel): """Exact page-text slice with complete source and policy provenance.""" block_id: str = Field(pattern=r"^p[0-9]{4}-b[0-9]{4}$") document_id: str = Field(min_length=1) pdf_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") page_id: str = Field(min_length=1) page_start: int = Field(ge=1) page_end: int = Field(ge=1) section_id: str = Field(pattern=r"^s[0-9]{4}$") section_heading: str | None = None section_label_source: SectionLabelSource text: str character_start: int = Field(ge=0) character_end: int = Field(ge=0) source_order: int = Field(ge=0) extraction_version: str = Field(min_length=1) splitting_policy_version: str = Field(min_length=1) extraction_warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) @model_validator(mode="after") def offsets_and_pages_are_ordered(self) -> ExtractedBlock: if self.page_end < self.page_start: raise ValueError("page_end must be greater than or equal to page_start") if self.character_end < self.character_start: raise ValueError("character_end must be greater than or equal to character_start") if len(self.text) != self.character_end - self.character_start: raise ValueError("block text length must match its page-text character span") return self class ExtractedPage(DatasetModel): """One-based deterministic page extraction result.""" page_id: str = Field(min_length=1) page_number: int = Field(ge=1) source_order: int = Field(ge=0) text: str text_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") character_count: int = Field(ge=0) extraction_status: PageExtractionStatus width_points: float | None = Field(default=None, gt=0) height_points: float | None = Field(default=None, gt=0) coordinate_system: str | None = None block_ids: tuple[str, ...] = Field(default_factory=tuple) warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) @model_validator(mode="after") def validate_page_text_metrics(self) -> ExtractedPage: if self.character_count != len(self.text): raise ValueError("page character_count must equal stored text length") return self class DatasetExtractionManifest(DatasetModel): """Formal deterministic PDF extraction and provenance output.""" schema_version: str = Field(min_length=1) document: SourceArtifactReference lineage: ExtractionLineage document_id: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") extraction_method: str = Field(min_length=1) extraction_version: str = Field(min_length=1) page_text_policy_version: str = Field(min_length=1) heading_policy_version: str = Field(min_length=1) section_policy_version: str = Field(min_length=1) segmentation_version: str = Field(min_length=1) identifier_policy_version: str = Field(min_length=1) separator_policy_version: str = Field(min_length=1) page_numbering: Literal["one_based"] = "one_based" usable_for_classification: Literal[True] = True page_count: int = Field(ge=1) character_count: int = Field(ge=1) block_count: int = Field(ge=1) configured_limits: dict[str, int] pages: tuple[ExtractedPage, ...] sections: tuple[ExtractedSection, ...] blocks: tuple[ExtractedBlock, ...] excluded_separators: tuple[ExtractedSeparator, ...] = Field(default_factory=tuple) warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) @model_validator(mode="after") def validate_manifest_invariants(self) -> DatasetExtractionManifest: if self.document_id != f"sha256:{self.document.sha256}": raise ValueError("document_id must derive from the exact PDF SHA-256") if self.page_count != len(self.pages) or self.block_count != len(self.blocks): raise ValueError("manifest counts must match contained records") if self.character_count != sum(page.character_count for page in self.pages): raise ValueError("manifest character_count must match pages") if [page.page_number for page in self.pages] != list(range(1, len(self.pages) + 1)): raise ValueError("pages must be consecutive and one-based") if [page.source_order for page in self.pages] != list(range(len(self.pages))): raise ValueError("page source_order values must be consecutive") if [block.source_order for block in self.blocks] != list(range(len(self.blocks))): raise ValueError("block source_order values must be consecutive") page_by_number = {page.page_number: page for page in self.pages} for block in self.blocks: page = page_by_number[block.page_start] if block.page_start != block.page_end or page.page_id != block.page_id: raise ValueError("blocks must be page-bounded and reference the correct page") if page.text[block.character_start : block.character_end] != block.text: raise ValueError("block text must be an exact page-text slice") for page in self.pages: page_blocks = [block for block in self.blocks if block.page_start == page.page_number] if "".join(block.text for block in page_blocks) != page.text: raise ValueError("ordered blocks must exactly reconstruct approved page text") return self class ExtractionPreclassificationFailure(DatasetModel): """Fatal extraction result with an explicit zero-inference boundary.""" error: DatasetErrorRecord classification_attempted: Literal[False] = False model_calls: Literal[0] = 0 usable_for_classification: Literal[False] = False manifest: Literal[None] = None comparison: Literal[None] = None class ConfidenceSignal(DatasetModel): """Explicitly uncalibrated confidence signal.""" value: float = Field(ge=0.0, le=1.0) calibrated: Literal[False] = False class ProductScopeDecision(DatasetModel): """Product applicability decision for one source block.""" scope_type: ProductScopeType applicable_short_names: tuple[str, ...] = Field(default_factory=tuple) family_label: str | None = None product_match_confidence: ConfidenceSignal rationale: str = Field(min_length=1) qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) review_required: bool = False class ScientificRelevance(str, Enum): # noqa: UP042 """Scientific relevance assigned without making GCMD recommendations.""" RELEVANT = "relevant" NOT_RELEVANT = "not_relevant" CONTEXT_ONLY = "context_only" class GroupMembership(str, Enum): # noqa: UP042 """Whether a grouped block requires a decision or supplies context only.""" DECISION_TARGET = "decision_target" CONTEXT = "context" class PacketReadiness(str, Enum): # noqa: UP042 """Whether a packet may cross the later classifier boundary.""" CLASSIFICATION_READY = "classification_ready" REVIEW_REQUIRED = "review_required" class EvidenceSelectionDecisionV1(DatasetModel): """Historical v1 response retained for persisted-artifact recovery only.""" block_id: str = Field(min_length=1) selected: bool scientific_relevance: ScientificRelevance evidence_role: EvidenceRole scope_type: ProductScopeType evidence_eligibility: EvidenceEligibility product_match_confidence: ConfidenceSignal qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) qualifying_declaration_ids: tuple[str, ...] = Field(default_factory=tuple) rationale: str = Field(min_length=1) review_required: bool prompt_version: Literal["dataset_evidence_selection_v1"] = "dataset_evidence_selection_v1" warnings: tuple[str, ...] = Field(default_factory=tuple) class EvidenceSelectionDecision(DatasetModel): """Strict v2 model decision for one supplied decision-target block.""" block_id: str = Field(min_length=1) selected: bool scientific_relevance: ScientificRelevance evidence_role: EvidenceRole scope_type: ProductScopeType evidence_eligibility: EvidenceEligibility product_match_confidence: ConfidenceSignal qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) qualifying_declaration_ids: tuple[str, ...] = Field(default_factory=tuple) rationale: str = Field(min_length=1) review_required: bool prompt_version: Literal["dataset_evidence_selection_v2"] = "dataset_evidence_selection_v2" warnings: tuple[str, ...] = Field(default_factory=tuple) class EvidenceSelectionModelResponse(DatasetModel): """Strict response for one bounded evidence-selection group.""" decisions: tuple[EvidenceSelectionDecision, ...] class EvidenceMinimizationDecision(DatasetModel): """One source-ID-only decision in the bounded packet-minimality pass.""" block_id: str = Field(min_length=1) retain: bool removal_reason: ( Literal["exact_duplicate", "redundant_science_evidence", "non_independent_science"] | None ) = None prompt_version: Literal["dataset_evidence_minimization_v1"] = "dataset_evidence_minimization_v1" class EvidenceMinimizationModelResponse(DatasetModel): """Strict response that can only retain or remove supplied selected IDs.""" decisions: tuple[EvidenceMinimizationDecision, ...] class EvidenceExclusionReason(str, Enum): # noqa: UP042 """Stable, content-free audit reasons for candidate or packet exclusion.""" PRODUCT_REVIEW_REQUIRED = "product_review_required" CONFIDENCE_BELOW_ACCEPTANCE = "confidence_below_acceptance" APPLICABILITY_UNTRACEABLE = "applicability_untraceable" DETERMINISTIC_NON_SCIENCE = "deterministic_non_science" MODEL_NOT_SELECTED = "model_not_selected" NON_INDEPENDENT_ROLE = "non_independent_role" DUPLICATE_SOURCE = "duplicate_source" MINIMALITY_REMOVED = "minimality_removed" class EvidenceExclusionRecord(DatasetModel): """Safe audit record containing identity and policy reason, never source text.""" block_id: str = Field(min_length=1) reason: EvidenceExclusionReason phase: Literal["candidate", "selection", "consolidation", "minimality"] class EvidencePacketBlock(DatasetModel): """Authoritative source block plus clearly separate derived annotations.""" block: ExtractedBlock block_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") evidence_role: EvidenceRole scientific_relevance: ScientificRelevance = ScientificRelevance.RELEVANT product_scope: ProductScopeDecision evidence_eligibility: EvidenceEligibility = EvidenceEligibility.ELIGIBLE selection_rationale: str = Field(default="Selected scientific source evidence.", min_length=1) qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) qualifying_declaration_ids: tuple[str, ...] = Field(default_factory=tuple) context_for_block_ids: tuple[str, ...] = Field(default_factory=tuple) review_required: bool = False class PacketComponentHashes(DatasetModel): """Stable component identities for the logical evidence packet.""" target_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") ordered_block_identity_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") ordered_source_text_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") product_scope_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") selection_decisions_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") policy_versions_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") limit_lineage_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") class PacketLimitRecord(DatasetModel): """Versioned packet measurement and bounded reduction lineage.""" policy_version: str = Field(min_length=1) measurement_version: str = Field(min_length=1) max_blocks: int = Field(gt=0) max_characters: int = Field(gt=0) preferred_blocks: int | None = Field(default=None, gt=0) preferred_characters: int | None = Field(default=None, gt=0) initial_block_ids: tuple[str, ...] final_block_ids: tuple[str, ...] removed_block_ids: tuple[str, ...] = Field(default_factory=tuple) removal_rationales: dict[str, str] = Field(default_factory=dict) initial_block_count: int = Field(ge=0) final_block_count: int = Field(ge=0) initial_character_count: int = Field(ge=0) final_character_count: int = Field(ge=0) initial_estimated_tokens: int = Field(ge=0) final_estimated_tokens: int = Field(ge=0) reduction_required: bool reduction_passes: int = Field(ge=0, le=1) minimality_policy_version: str | None = None minimality_model_calls: int = Field(default=0, ge=0) within_limits: bool class DatasetEvidencePacket(DatasetModel): """Formal immutable and source-faithful later-classifier input.""" schema_version: str = Field(min_length=1) packet_id: str = Field(min_length=1) packet_version: str = Field(min_length=1) target: DatasetIdentity cmr_source_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") readme_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") extraction_manifest_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") coverage_mode: CoverageMode coverage_confidence: ConfidenceSignal selection_version: str = Field(min_length=1) grouping_policy_version: str = Field(default="dataset_evidence_grouping_v2", min_length=1) evidence_role_policy_version: str = Field(default="dataset_evidence_roles_v2", min_length=1) minimality_policy_version: str = Field(default="dataset_evidence_minimization_v1", min_length=1) response_schema_version: str = Field( default="dataset_evidence_selection_schema_v2", min_length=1 ) blocks: tuple[EvidencePacketBlock, ...] excluded_block_ids: tuple[str, ...] = Field(default_factory=tuple) ambiguous_block_ids: tuple[str, ...] = Field(default_factory=tuple) context_only_block_ids: tuple[str, ...] = Field(default_factory=tuple) character_count: int = Field(ge=0) estimated_tokens: int | None = Field(default=None, ge=0) packet_readiness: PacketReadiness = PacketReadiness.CLASSIFICATION_READY review_status: DatasetReviewStatus = DatasetReviewStatus.NOT_REQUIRED limit_record: PacketLimitRecord | None = None component_hashes: PacketComponentHashes | None = None warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) packet_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") @model_validator(mode="after") def validate_packet_shape(self) -> DatasetEvidencePacket: block_ids = [item.block.block_id for item in self.blocks] if len(block_ids) != len(set(block_ids)): raise ValueError("evidence packet block IDs must be unique") if block_ids and [item.block.source_order for item in self.blocks] != sorted( item.block.source_order for item in self.blocks ): raise ValueError("evidence packet blocks must preserve source order") if self.character_count != sum(len(item.block.text) for item in self.blocks): raise ValueError("packet character_count must match exact block text") if self.blocks and not any( item.scientific_relevance is ScientificRelevance.RELEVANT for item in self.blocks ): raise ValueError("a packet cannot contain only context blocks") return self class DatasetDeterministicValidation(DatasetModel): """Vocabulary-integrity result for one dataset classification.""" valid: bool errors: tuple[DatasetErrorRecord, ...] = Field(default_factory=tuple) warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) class KeywordEvidenceConfidence(DatasetModel): """Uncalibrated hierarchy-stage confidence signals.""" topic: float | None = Field(default=None, ge=0.0, le=1.0) term: float | None = Field(default=None, ge=0.0, le=1.0) variable_level_1: float | None = Field(default=None, ge=0.0, le=1.0) variable_level_2: float | None = Field(default=None, ge=0.0, le=1.0) variable_level_3: float | None = Field(default=None, ge=0.0, le=1.0) final: float | None = Field(default=None, ge=0.0, le=1.0) calibrated: Literal[False] = False class DatasetSemanticValidatorState(str, Enum): # noqa: UP042 """Independent semantic-validator execution state for dataset classification.""" NOT_RUN_BY_POLICY = "not_run_by_policy" NOT_APPLICABLE = "not_applicable" class DatasetEvidenceCitation(DatasetModel): """Validated reference back to one immutable evidence-packet block.""" block_id: str = Field(min_length=1) page_start: int = Field(ge=1) page_end: int = Field(ge=1) section_id: str = Field(min_length=1) section_heading: str | None = None evidence_role: EvidenceRole product_scope_type: ProductScopeType block_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") independently_eligible: bool class DatasetClassification(DatasetModel): """One blind, authoritative GCMD classification for a dataset.""" UUID: str = Field(min_length=1) name: str = Field(min_length=1) level: HierarchyLevel canonical_path: str = Field(min_length=1) path_components: tuple[str, ...] = Field(min_length=1) topic: str = Field(min_length=1) term: str | None = None cited_block_ids: tuple[str, ...] = Field(min_length=1) citations: tuple[DatasetEvidenceCitation, ...] = Field(min_length=1) deepest_supported_uuid: str = Field(min_length=1) decision_prompt_versions: tuple[str, ...] = Field(default_factory=tuple) product_match_confidence: ConfidenceSignal keyword_evidence_confidence: KeywordEvidenceConfidence reason_for_stopping: str = Field(min_length=1) deterministic_validation: DatasetDeterministicValidation final_status: DatasetClassificationFinalStatus review_required: bool = False @model_validator(mode="after") def accepted_results_are_valid(self) -> DatasetClassification: if ( self.final_status in { DatasetClassificationFinalStatus.ACCEPTED, DatasetClassificationFinalStatus.REDUCED_TO_ANCESTOR, } and not self.deterministic_validation.valid ): raise ValueError("accepted or reduced classifications require valid vocabulary data") return self class BlindResultSeal(DatasetModel): """Hash and persistence metadata for a completed blind result.""" blind_result_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") evidence_packet_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") persisted_at: str = Field(min_length=1) persistence_reference: str = Field(min_length=1) expert_keywords_excluded: Literal[True] = True class DatasetProcessingMetadata(DatasetModel): """Reproducibility metadata for dataset processing.""" run_id: str | None = None started_at: str | None = None completed_at: str | None = None application_version: str | None = None vocabulary_hash: str | None = None configuration_hash: str | None = None prompt_versions: dict[str, str] = Field(default_factory=dict) model_provider: str | None = None model_name: str | None = None model_calls: int = Field(default=0, ge=0) input_tokens: int | None = Field(default=None, ge=0) output_tokens: int | None = Field(default=None, ge=0) estimated_cost: float | None = Field(default=None, ge=0.0) cache_used: bool = False class DatasetClassificationResult(DatasetModel): """Top-level dataset classification result contract.""" schema_version: str = Field(min_length=1) dataset_key: str = Field(min_length=1) identity: DatasetIdentity cmr_source: CMRSourceRecord | None = None sealed_expert_keywords: SealedExpertKeywordInput | None = None readme_candidates: tuple[READMECandidate, ...] = Field(default_factory=tuple) selected_readme: READMESelection | None = None readme_artifact: SourceArtifactReference | None = None extraction_manifest_reference: str | None = None product_resolution_reference: str | None = None evidence_packet_reference: str | None = None evidence_packet_sha256: str | None = Field(default=None, pattern=r"^[0-9a-f]{64}$") coverage_mode: CoverageMode | None = None semantic_validator_state: DatasetSemanticValidatorState = ( DatasetSemanticValidatorState.NOT_APPLICABLE ) no_classification_cited_block_ids: tuple[str, ...] = Field(default_factory=tuple) classification_attempted: bool processing_status: DatasetProcessingStatus classification_outcome: DatasetClassificationOutcome | None = None classifications: tuple[DatasetClassification, ...] = Field(default_factory=tuple) no_classification_reason: str | None = None blind_result: BlindResultSeal | None = None comparison_reference: str | None = None review_status: DatasetReviewStatus = DatasetReviewStatus.NOT_REQUIRED warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) errors: tuple[DatasetErrorRecord, ...] = Field(default_factory=tuple) hierarchy_attempt_diagnostics: tuple[DatasetHierarchyAttemptDiagnostic, ...] = Field( default_factory=tuple ) processing_metadata: DatasetProcessingMetadata = Field( default_factory=DatasetProcessingMetadata ) @model_validator(mode="after") def validate_result_scope(self) -> DatasetClassificationResult: if self.dataset_key != self.identity.concept_id: raise ValueError("dataset_key must equal identity.concept_id") if self.cmr_source is not None and self.cmr_source.identity != self.identity: raise ValueError("cmr_source identity must match result identity") if not self.classification_attempted: if self.classification_outcome is not None: raise ValueError("unattempted classification cannot have an outcome") if self.classifications: raise ValueError("unattempted classification cannot contain classifications") if self.blind_result is not None: raise ValueError("unattempted classification cannot have a blind result") if self.comparison_reference is not None: raise ValueError("unattempted classification cannot have comparison output") if self.processing_metadata.model_calls != 0: raise ValueError("unattempted classification requires zero model calls") else: if self.coverage_mode is None: raise ValueError("attempted dataset processing requires coverage identity") semantic_no_science = ( self.evidence_packet_sha256 is None and self.classification_outcome is DatasetClassificationOutcome.NOT_CLASSIFIED and not self.classifications and bool(self.no_classification_reason) ) if semantic_no_science: if ( self.semantic_validator_state is not DatasetSemanticValidatorState.NOT_APPLICABLE ): raise ValueError( "pre-classifier semantic stops require validator not applicable" ) elif ( self.evidence_packet_sha256 is None or self.semantic_validator_state is not DatasetSemanticValidatorState.NOT_RUN_BY_POLICY ): raise ValueError("GCMD classification requires a packet and validator policy state") if ( self.classification_attempted and self.processing_status is DatasetProcessingStatus.COMPLETED ): if self.classification_outcome is None: raise ValueError("completed attempted classification requires an outcome") if ( self.classification_outcome is DatasetClassificationOutcome.CLASSIFIED and not self.classifications ): raise ValueError("classified dataset results require classifications") if self.classification_outcome is DatasetClassificationOutcome.NOT_CLASSIFIED and ( self.classifications or not self.no_classification_reason ): raise ValueError("not_classified requires no classifications and a reason") if self.comparison_reference is not None and self.blind_result is None: raise ValueError("comparison output requires a persisted blind result") return self class ResolvedExpertKeyword(DatasetModel): """Expert keyword deterministically resolved to local vocabulary.""" source_index: int = Field(ge=0) UUID: str = Field(min_length=1) name: str = Field(min_length=1) canonical_path: str = Field(min_length=1) class UnresolvedExpertKeyword(DatasetModel): """Expert source entry that could not be resolved unambiguously.""" source_index: int = Field(ge=0) source_entry: dict[str, Any] reason: str = Field(min_length=1) class DatasetComparisonRecord(DatasetModel): """One additive post-inference comparison decision.""" blind_classification_uuid: str = Field(min_length=1) expert_keyword_uuids: tuple[str, ...] = Field(default_factory=tuple) hierarchy_relationship: str = Field(min_length=1) category: ComparisonCategory rationale: str = Field(min_length=1) review_required: bool class DatasetComparison(DatasetModel): """Formal post-inference comparison contract.""" schema_version: str = Field(min_length=1) dataset_key: str = Field(min_length=1) blind_result_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") sealed_expert_keyword_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") vocabulary_hash: str = Field(pattern=r"^[0-9a-f]{64}$") resolved_expert_keywords: tuple[ResolvedExpertKeyword, ...] = Field(default_factory=tuple) unresolved_expert_keywords: tuple[UnresolvedExpertKeyword, ...] = Field(default_factory=tuple) comparisons: tuple[DatasetComparisonRecord, ...] = Field(default_factory=tuple) review_status: DatasetReviewStatus warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) errors: tuple[DatasetErrorRecord, ...] = Field(default_factory=tuple) class CMRRedirectProvenance(DatasetModel): """One manually validated CMR redirect.""" status_code: int = Field(ge=300, le=399) source_url: str = Field(min_length=1) target_url: str = Field(min_length=1) class CMRRetrievedSource(DatasetModel): """Exact CMR response bytes plus derived parsed representation.""" submitted_url: str = Field(min_length=1) final_url: str = Field(min_length=1) redirect_history: tuple[CMRRedirectProvenance, ...] = Field(default_factory=tuple) retrieved_at: str = Field(min_length=1) status_code: int = Field(ge=100, le=599) response_metadata: dict[str, str] = Field(default_factory=dict) response_bytes: bytes source_text: str sha256: str = Field(pattern=r"^[0-9a-f]{64}$") parsed_json: dict[str, Any] class SealedExpertKeywordPayload(DatasetModel): """Original expert assignments held outside all blind-stage interfaces.""" dataset_key: str = Field(min_length=1) source_path: Literal["umm.ScienceKeywords"] = "umm.ScienceKeywords" science_keywords: tuple[dict[str, Any], ...] = Field(default_factory=tuple) payload_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") class ResolvedCMRCollection(DatasetModel): """Validated source, authoritative identity, blind view, and sealed payload.""" dataset_key: str = Field(min_length=1) submitted_native_id: str = Field(min_length=1) returned_native_id: str = Field(min_length=1) derived_native_id: str = Field(min_length=1) identity: DatasetIdentity source: CMRRetrievedSource source_item: dict[str, Any] blind_view: BlindCollectionView sealed_expert_keywords: SealedExpertKeywordPayload @model_validator(mode="after") def identifiers_are_consistent(self) -> ResolvedCMRCollection: values = { self.submitted_native_id, self.returned_native_id, self.derived_native_id, self.identity.native_id, } if len(values) != 1: raise ValueError("submitted, returned, derived, and typed Native IDs must match") if self.dataset_key != self.identity.concept_id: raise ValueError("dataset_key must equal the CMR Concept ID") return self class CMRPreclassificationFailure(DatasetModel): """Explicit CMR-stage failure that cannot be mistaken for classification.""" error: DatasetErrorRecord classification_attempted: Literal[False] = False model_calls: Literal[0] = 0 classifications: tuple[()] = () classification_outcome: Literal[None] = None comparison: Literal[None] = None class READMEPreclassificationOutcome(DatasetModel): """Discovery/retrieval outcome that explicitly performed no inference.""" reason_code: str = Field(min_length=1) processing_status: DatasetProcessingStatus classification_attempted: Literal[False] = False model_calls: Literal[0] = 0 classifications: tuple[()] = () classification_outcome: Literal[None] = None extraction: Literal[None] = None comparison: Literal[None] = None class ProductSignalType(str, Enum): # noqa: UP042 """Versioned deterministic product-applicability signal kinds.""" EXACT_SHORT_NAME = "exact_short_name" EXACT_NATIVE_ID = "exact_native_id" EXACT_VERSION = "exact_version" EXPLICIT_ALIAS_DECLARATION = "explicit_alias_declaration" ENUMERATED_PRODUCT_LIST = "enumerated_product_list" TARGET_IN_ENUMERATION = "target_in_enumeration" OTHER_PRODUCT_IN_ENUMERATION = "other_product_in_enumeration" FAMILY_DECLARATION = "family_declaration" TARGET_FAMILY_MEMBERSHIP = "target_family_membership" UNPROVEN_FAMILY_MEMBERSHIP = "unproven_family_membership" TEMPLATE_HEADING = "template_heading" DOCUMENT_WIDE_CONTEXT = "document_wide_context" SHARED_CONTEXT = "shared_context" OTHER_PRODUCT_RESTRICTION = "other_product_restriction" UNRESOLVED_SCOPE = "unresolved_scope" NORMALIZED_TARGET_ALIAS = "normalized_target_alias" OTHER_PRODUCT_HEADING = "other_product_heading" class DeclarationKind(str, Enum): # noqa: UP042 """Deterministically recognized declaration structure.""" IDENTITY = "identity" ENUMERATED_PRODUCTS = "enumerated_products" PRODUCT_FAMILY = "product_family" DOCUMENT_WIDE = "document_wide" OTHER_PRODUCT_ONLY = "other_product_only" class AliasSource(str, Enum): # noqa: UP042 """Approved provenance for a product identity alias.""" VALIDATED_CMR_IDENTITY = "validated_cmr_identity" DETERMINISTIC_DERIVATION = "deterministic_derivation" DETERMINISTIC_NORMALIZATION = "deterministic_normalization" README_DECLARATION = "readme_declaration" class ProductReviewReason(str, Enum): # noqa: UP042 """Versioned reasons for block- or dataset-level product review.""" MODEL_REVIEW_ADVICE = "model_review_advice" TARGET_BLOCK_CONFIDENCE_BELOW_THRESHOLD = "target_block_confidence_below_threshold" AMBIGUOUS_PRODUCT_SCOPE = "ambiguous_product_scope" MIXED_PRODUCT_TRANSITION = "mixed_product_transition" DOCUMENT_TARGET_CONFIDENCE_BELOW_THRESHOLD = "document_target_confidence_below_threshold" VAGUE_OR_UNDETERMINED_COVERAGE = "vague_or_undetermined_coverage" TARGET_INCLUSION_UNPROVEN = "target_inclusion_unproven" FAMILY_MEMBERSHIP_UNPROVEN = "family_membership_unproven" DOCUMENT_CONTRADICTION = "document_contradiction" NO_DEFENSIBLE_TARGET_PASSAGE = "no_defensible_target_passage" class DeclarationLocality(DatasetModel): """Deterministic forward-only applicability range for one declaration.""" policy_version: str = Field(min_length=1) anchor_block_id: str = Field(min_length=1) section_id: str = Field(min_length=1) source_order_start: int = Field(ge=0) source_order_end: int = Field(ge=0) allowed_block_ids: tuple[str, ...] = Field(min_length=1) explicit_backward_scope: bool = False @model_validator(mode="after") def validate_locality(self) -> DeclarationLocality: if self.source_order_end < self.source_order_start: raise ValueError("declaration locality source order must be ordered") if len(self.allowed_block_ids) != len(set(self.allowed_block_ids)): raise ValueError("declaration locality block IDs must be unique") if self.anchor_block_id not in self.allowed_block_ids: raise ValueError("declaration locality must contain its anchor block") return self class ResolutionDeterminationMethod(str, Enum): # noqa: UP042 """Whether a result came from rules, model analysis, or both.""" DETERMINISTIC = "deterministic" MODEL = "model" HYBRID = "hybrid" class DeterministicProductSignal(DatasetModel): """Exact source span and context for a deterministic product signal.""" signal_id: str = Field(min_length=1) signal_type: ProductSignalType block_id: str = Field(min_length=1) page_number: int = Field(ge=1) section_id: str = Field(min_length=1) literal_value: str normalized_value: str | None = None character_start: int = Field(ge=0) character_end: int = Field(ge=0) method_version: str = Field(min_length=1) @model_validator(mode="after") def validate_signal_span(self) -> DeterministicProductSignal: if self.character_end < self.character_start: raise ValueError("signal character span must be ordered") if len(self.literal_value) != self.character_end - self.character_start: raise ValueError("literal signal length must match its block-text span") return self class ProductDeclaration(DatasetModel): """Auditable product or family declaration assembled from source signals.""" declaration_id: str = Field(min_length=1) kind: DeclarationKind block_id: str = Field(min_length=1) signal_ids: tuple[str, ...] = Field(min_length=1) product_values: tuple[str, ...] = Field(default_factory=tuple) family_label: str | None = None target_included: bool contradictory: bool = False method_version: str = Field(min_length=1) locality: DeclarationLocality | None = None class ProductAliasProvenance(DatasetModel): """Allowed product identity with explicit non-model provenance.""" alias: str = Field(min_length=1) source: AliasSource derivation_method: str = Field(min_length=1) method_version: str = Field(min_length=1) source_block_id: str | None = None character_start: int | None = Field(default=None, ge=0) character_end: int | None = Field(default=None, ge=0) source_literal: str | None = None normalization_rule: str | None = None class CoverageAnalysisInput(DatasetModel): """Minimum blind-stage input for coverage and passage resolution.""" target: DatasetIdentity derived_native_id: str = Field(min_length=1) entry_title: str | None = None summary: str | None = None extraction_manifest: DatasetExtractionManifest deterministic_analysis_version: Literal["dataset_product_signals_v3"] = ( "dataset_product_signals_v3" ) declaration_locality_policy_version: Literal["dataset_declaration_locality_v3"] = ( "dataset_declaration_locality_v3" ) downstream_eligibility_policy_version: Literal["dataset_downstream_eligibility_v5"] = ( "dataset_downstream_eligibility_v5" ) review_aggregation_policy_version: Literal["dataset_product_review_v5"] = ( "dataset_product_review_v5" ) mixed_product_transition_policy_version: Literal["dataset_mixed_product_transition_v1"] = ( "dataset_mixed_product_transition_v1" ) coverage_prompt_version: Literal["dataset_coverage_v3"] = "dataset_coverage_v3" product_resolution_prompt_version: Literal["dataset_product_resolution_v3"] = ( "dataset_product_resolution_v3" ) @model_validator(mode="after") def validate_blind_lineage(self) -> CoverageAnalysisInput: if self.target != self.extraction_manifest.lineage.identity: raise ValueError("coverage target must match extraction lineage identity") if self.derived_native_id != self.target.native_id: raise ValueError("derived Native ID must match validated target identity") return self class CoverageModelResponseV2(DatasetModel): """Legacy v2 coverage response retained only for persisted-artifact recovery.""" coverage_mode: CoverageMode qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) rejected_block_ids: tuple[str, ...] = Field(default_factory=tuple) determination_method: ResolutionDeterminationMethod rationale: str = Field(min_length=1) product_match_confidence: ConfidenceSignal review_required: bool downstream_classification_eligible: bool class CoverageModelResponseV3(DatasetModel): """Strict v3 coverage advice without pipeline-continuation authority.""" coverage_mode: CoverageMode qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) rejected_block_ids: tuple[str, ...] = Field(default_factory=tuple) determination_method: ResolutionDeterminationMethod rationale: str = Field(min_length=1) product_match_confidence: ConfidenceSignal review_required: bool CoverageModelResponse = CoverageModelResponseV3 class ProductBlockModelDecision(DatasetModel): """Strict model decision for one supplied extraction block.""" block_id: str = Field(min_length=1) document_id: str = Field(min_length=1) page_number: int = Field(ge=1) section_id: str = Field(min_length=1) scope_type: ProductScopeType eligibility: EvidenceEligibility applicable_products: tuple[str, ...] = Field(default_factory=tuple) family_label: str | None = None qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) declaration_ids: tuple[str, ...] = Field(default_factory=tuple) aliases_used: tuple[str, ...] = Field(default_factory=tuple) determination_method: ResolutionDeterminationMethod rationale: str = Field(min_length=1) product_match_confidence: ConfidenceSignal review_required: bool warnings: tuple[str, ...] = Field(default_factory=tuple) class ProductResolutionModelResponse(DatasetModel): """Strict versioned dataset product-resolution batch response.""" decisions: tuple[ProductBlockModelDecision, ...] class ProductPolicyViolationDiagnostic(DatasetModel): """Sanitized facts identifying one deterministic product-policy violation.""" rule_id: Literal[ "EXACT_TARGET_REQUIRES_ELIGIBLE", "EXACT_TARGET_REQUIRES_TARGET_SHORT_NAME", "ENUMERATED_TARGET_REQUIRES_ELIGIBLE", "ENUMERATED_TARGET_REQUIRES_TARGET_SHORT_NAME", "OTHER_PRODUCT_REQUIRES_EXCLUDED", "AMBIGUOUS_REQUIRES_REVIEW_ONLY", "MIXED_PRODUCT_REQUIRES_REVIEW_ONLY", "ELIGIBLE_FAMILY_REQUIRES_FAMILY_DECLARATION", "ELIGIBLE_FAMILY_REQUIRES_TARGET_MEMBERSHIP", "DOCUMENT_CONTEXT_ELIGIBILITY_INVALID", ] block_id: str = Field(min_length=1) scope_type: ProductScopeType eligibility: EvidenceEligibility eligibility_valid: bool target_short_name_present: bool | None = None family_declaration_present: bool | None = None target_membership_present: bool | None = None class ProductResolutionAttemptDiagnostic(DatasetModel): """Sanitized metadata for one rejected coverage or product-resolution attempt.""" phase: Literal["coverage", "product_resolution"] batch_index: int | None = Field(default=None, ge=1) total_batch_count: int | None = Field(default=None, ge=1) block_count: int = Field(ge=0) total_block_characters: int = Field(ge=0) signal_count: int = Field(ge=0) declaration_count: int = Field(ge=0) attempt_number: int = Field(ge=1) maximum_attempts: int = Field(ge=1) failure_code: str = Field(min_length=1) failure_category: Literal["validation", "schema", "provider"] policy_violation: ProductPolicyViolationDiagnostic | None = None @model_validator(mode="after") def validate_attempt_context(self) -> ProductResolutionAttemptDiagnostic: if self.attempt_number > self.maximum_attempts: raise ValueError("attempt_number cannot exceed maximum_attempts") if self.phase == "coverage": if self.batch_index is not None or self.total_batch_count is not None: raise ValueError("coverage diagnostics cannot contain batch indices") elif ( self.batch_index is None or self.total_batch_count is None or self.batch_index > self.total_batch_count ): raise ValueError("product-resolution diagnostics require a valid batch index") if self.failure_code == "PRODUCT_POLICY_INVALID": if self.policy_violation is None: raise ValueError("product-policy failures require a sanitized subrule diagnostic") elif self.policy_violation is not None: raise ValueError("policy_violation is only valid for product-policy failures") return self class ProductBlockDecision(DatasetModel): """Validated authoritative linkage for one block-level scope decision.""" block_id: str = Field(min_length=1) document_id: str = Field(min_length=1) page_number: int = Field(ge=1) section_id: str = Field(min_length=1) source_order: int = Field(ge=0) scope_type: ProductScopeType eligibility: EvidenceEligibility applicable_products: tuple[str, ...] = Field(default_factory=tuple) family_label: str | None = None qualifying_block_ids: tuple[str, ...] = Field(default_factory=tuple) declaration_ids: tuple[str, ...] = Field(default_factory=tuple) determination_method: ResolutionDeterminationMethod rule_version: str = Field(min_length=1) prompt_version: str = Field(min_length=1) rationale: str = Field(min_length=1) product_match_confidence: ConfidenceSignal review_required: bool review_reasons: tuple[ProductReviewReason, ...] = Field(default_factory=tuple) warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) class DatasetCoverageResolution(DatasetModel): """Complete blind coverage and product-scope output before evidence selection.""" target: DatasetIdentity extraction_manifest_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") selected_candidate_id: str = Field(min_length=1) selected_readme_url: str = Field(min_length=1) pdf_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") deterministic_analysis_version: str = Field(min_length=1) coverage_prompt_version: Literal[ "dataset_coverage_v1", "dataset_coverage_v2", "dataset_coverage_v3" ] product_resolution_prompt_version: Literal[ "dataset_product_resolution_v1", "dataset_product_resolution_v2", "dataset_product_resolution_v3", ] response_schema_version: Literal[ "dataset_product_resolution_schema_v1", "dataset_product_resolution_schema_v2", "dataset_product_resolution_schema_v3", ] declaration_locality_policy_version: str | None = None downstream_eligibility_policy_version: str | None = None review_aggregation_policy_version: str | None = None mixed_product_transition_policy_version: str | None = None signals: tuple[DeterministicProductSignal, ...] declarations: tuple[ProductDeclaration, ...] aliases: tuple[ProductAliasProvenance, ...] coverage_mode: CoverageMode qualifying_block_ids: tuple[str, ...] rejected_block_ids: tuple[str, ...] determination_method: ResolutionDeterminationMethod rationale: str = Field(min_length=1) product_match_confidence: ConfidenceSignal review_required: bool review_reasons: tuple[ProductReviewReason, ...] = Field(default_factory=tuple) uncertain_block_ids: tuple[str, ...] = Field(default_factory=tuple) downstream_classification_eligible: bool block_decisions: tuple[ProductBlockDecision, ...] target_eligible_block_ids: tuple[str, ...] excluded_other_product_block_ids: tuple[str, ...] ambiguous_review_block_ids: tuple[str, ...] document_context_block_ids: tuple[str, ...] warnings: tuple[DatasetWarning, ...] = Field(default_factory=tuple) blind_cache_identity_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") model_calls: int = Field(ge=0) @model_validator(mode="after") def validate_block_partition(self) -> DatasetCoverageResolution: decisions = list(self.block_decisions) if [decision.source_order for decision in decisions] != list(range(len(decisions))): raise ValueError("block decisions must preserve consecutive source order") ids = [decision.block_id for decision in decisions] if len(ids) != len(set(ids)): raise ValueError("block decisions must be unique") partitions = ( self.target_eligible_block_ids, self.excluded_other_product_block_ids, self.ambiguous_review_block_ids, self.document_context_block_ids, ) flattened = [item for group in partitions for item in group] if len(flattened) != len(set(flattened)) or set(flattened) != set(ids): raise ValueError("eligibility collections must exactly partition block decisions") if not set(self.uncertain_block_ids).issubset(ids): raise ValueError("uncertain block IDs must reference block decisions") if self.response_schema_version == "dataset_product_resolution_schema_v3": policies = ( self.declaration_locality_policy_version, self.downstream_eligibility_policy_version, self.review_aggregation_policy_version, self.mixed_product_transition_policy_version, ) if any(value is None for value in policies): raise ValueError("v3 product resolution requires all policy versions") if any(item.locality is None for item in self.declarations): raise ValueError("v3 declarations require deterministic locality") return self class CoverageResolutionFailure(DatasetModel): """Typed pre-classification coverage failure after bounded retries.""" error: DatasetErrorRecord classification_attempted: Literal[False] = False gcmd_classifier_calls: Literal[0] = 0 downstream_classification_eligible: Literal[False] = False model_calls: int = Field(ge=0) attempt_diagnostics: tuple[ProductResolutionAttemptDiagnostic, ...] = Field( default_factory=tuple ) coverage_result: Literal[None] = None comparison: Literal[None] = None class EvidenceGroupMember(DatasetModel): """Traceable membership of one authoritative block in one group.""" block_id: str = Field(min_length=1) membership: GroupMembership linked_target_block_ids: tuple[str, ...] = Field(default_factory=tuple) class EvidenceBlockGroup(DatasetModel): """Deterministic bounded source-ordered evidence-selection group.""" group_id: str = Field(pattern=r"^group-[0-9]{4}$") extraction_manifest_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") grouping_policy_version: str = Field(min_length=1) members: tuple[EvidenceGroupMember, ...] source_character_count: int = Field(ge=0) max_blocks: int = Field(gt=0) max_characters: int = Field(gt=0) context_link_method: str = Field(min_length=1) class EvidenceSelectionInput(DatasetModel): """Minimum blind typed input accepted by the evidence stage.""" extraction_manifest: DatasetExtractionManifest product_resolution: DatasetCoverageResolution @model_validator(mode="after") def validate_lineage(self) -> EvidenceSelectionInput: if self.extraction_manifest.lineage.identity != self.product_resolution.target: raise ValueError("evidence target must match extraction and product resolution") if self.extraction_manifest.document.sha256 != self.product_resolution.pdf_sha256: raise ValueError("evidence PDF identity must match product resolution") return self class EvidenceSelectionAudit(DatasetModel): """Derived selection metadata retained separately from packet source text.""" grouping_policy_version: str = Field(min_length=1) prompt_version: str = Field(min_length=1) response_schema_version: str = Field(min_length=1) groups: tuple[EvidenceBlockGroup, ...] decisions: tuple[EvidenceSelectionDecision | EvidenceSelectionDecisionV1, ...] excluded_decisions: tuple[ProductBlockDecision, ...] ambiguous_decisions: tuple[ProductBlockDecision, ...] model_calls: int = Field(ge=0) selection_model_calls: int = Field(default=0, ge=0) minimality_model_calls: int = Field(default=0, ge=0) minimality_policy_version: str = Field(default="dataset_evidence_minimization_v1") minimality_ran: bool = False candidate_block_count: int = Field(default=0, ge=0) initially_selected_block_count: int = Field(default=0, ge=0) final_independent_block_count: int = Field(default=0, ge=0) final_context_block_count: int = Field(default=0, ge=0) retained_block_ids: tuple[str, ...] = Field(default_factory=tuple) removed_block_ids: tuple[str, ...] = Field(default_factory=tuple) exclusion_records: tuple[EvidenceExclusionRecord, ...] = Field(default_factory=tuple) blind_cache_identity_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") class EvidencePacketResult(DatasetModel): """Successful Milestone 6 result without any GCMD classification.""" packet: DatasetEvidencePacket selection_audit: EvidenceSelectionAudit classification_attempted: Literal[False] = False gcmd_classifier_calls: Literal[0] = 0 comparison: Literal[None] = None class EvidenceNoUsableScienceOutcome(DatasetModel): """Successful semantic stop before the GCMD classifier boundary.""" reason_code: Literal["NO_USABLE_SCIENCE_EVIDENCE"] = "NO_USABLE_SCIENCE_EVIDENCE" processing_status: Literal[DatasetProcessingStatus.COMPLETED] = ( DatasetProcessingStatus.COMPLETED ) classification_outcome: Literal[DatasetClassificationOutcome.NOT_CLASSIFIED] = ( DatasetClassificationOutcome.NOT_CLASSIFIED ) no_classification_reason: str = "No independently usable target-applicable science evidence." selection_audit: EvidenceSelectionAudit classification_attempted: Literal[False] = False gcmd_classifier_calls: Literal[0] = 0 packet: Literal[None] = None comparison: Literal[None] = None class EvidenceSelectionFailure(DatasetModel): """Typed evidence-stage stop after validation or bounded retry failure.""" error: DatasetErrorRecord review_status: DatasetReviewStatus = DatasetReviewStatus.PENDING classification_attempted: Literal[False] = False gcmd_classifier_calls: Literal[0] = 0 model_calls: int = Field(ge=0) packet: Literal[None] = None comparison: Literal[None] = None class DatasetClassifierInput(DatasetModel): """Minimal future boundary: a validated packet is the only README text input.""" target: DatasetIdentity evidence_packet: DatasetEvidencePacket @model_validator(mode="after") def validate_packet_target(self) -> DatasetClassifierInput: if self.target != self.evidence_packet.target: raise ValueError("classifier target must match the validated evidence packet") if self.evidence_packet.packet_readiness is not PacketReadiness.CLASSIFICATION_READY: raise ValueError("review-required packets cannot cross the classifier boundary") return self