Spaces:
Running on Zero
Running on Zero
| """Deterministic fixture-driven structural smoke evaluation for dataset classification.""" | |
| from __future__ import annotations | |
| import argparse | |
| import hashlib | |
| import json | |
| import re | |
| from collections import Counter | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Any, Literal | |
| import httpx | |
| from pydantic import BaseModel, ConfigDict, Field, ValidationError | |
| from gcmd_classifier.config import DatasetRuntimeSettings, DatasetSettings, ModelSettings | |
| from gcmd_classifier.datasets.application_service import ( | |
| ApplicationOperationStatus, | |
| ClassifySelectedReadmeRequest, | |
| DatasetApplicationService, | |
| RetrieveCollectionRequest, | |
| build_dataset_application_service, | |
| ) | |
| from gcmd_classifier.datasets.cmr import resolve_cmr_collection, validate_cmr_url | |
| from gcmd_classifier.datasets.deterministic_evidence import DeterministicEvidencePacket | |
| from gcmd_classifier.datasets.models import ( | |
| CMRRetrievedSource, | |
| CoverageMode, | |
| CoverageModelResponse, | |
| DatasetClassificationResult, | |
| DatasetExtractionManifest, | |
| EvidenceMinimizationModelResponse, | |
| EvidenceSelectionModelResponse, | |
| ProductResolutionModelResponse, | |
| ) | |
| from gcmd_classifier.datasets.readme_discovery import discover_readme_candidates | |
| from gcmd_classifier.datasets.structural_scope import DeterministicScopeResult | |
| from gcmd_classifier.llm.base import ModelRequest, ModelResponse, TokenUsage | |
| from gcmd_classifier.vocabulary import load_vocabulary | |
| FAKE_LIMITATION = ( | |
| "Deterministic fake responses validate workflow structure, provenance, and vocabulary " | |
| "integrity; they do not establish scientific accuracy." | |
| ) | |
| REVIEW_NOTICE = ( | |
| "Proposed additions and refinements are advisory, require expert review, and are never " | |
| "written to CMR." | |
| ) | |
| _BLOCK_RE = re.compile(r"<README_BLOCK>\n(.*?)\n</README_BLOCK>", re.DOTALL) | |
| _CANDIDATE_RE = re.compile(r"SUPPLIED_CANDIDATES=(\[.*?\])\n<README_BLOCK>", re.DOTALL) | |
| class SmokeModel(BaseModel): | |
| model_config = ConfigDict(extra="forbid", frozen=True) | |
| class SmokeFixtureCase(SmokeModel): | |
| case_id: str | |
| native_id: str | |
| cmr_url: str | |
| cmr_fixture: str | |
| readme_fixture: str | |
| expected_coverage_mode: CoverageMode | |
| expected_review_required: bool | |
| expected_concept_id: str | |
| cmr_revision_id: int = Field(strict=True, ge=1) | |
| short_name: str | |
| version: str | |
| selected_readme_url: str | |
| selected_readme_source_index: int = Field(strict=True, ge=0) | |
| cmr_retrieved_at: str | None | |
| readme_retrieved_at: str | None | |
| retrieval_time_status: Literal["known", "unknown_preexisting_fixture"] | |
| fixture_recorded_at: str | |
| cmr_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") | |
| readme_sha256: str = Field(pattern=r"^[0-9a-f]{64}$") | |
| notes: str | |
| provenance_notes: str | |
| class SmokeFixtureManifest(SmokeModel): | |
| schema_version: str | |
| cases: tuple[SmokeFixtureCase, ...] | |
| class SmokeFailure(SmokeModel): | |
| case_id: str | None = None | |
| field: str | |
| code: str | |
| message: str | |
| class FixturePreflightReport(SmokeModel): | |
| case_id: str | |
| passed: bool | |
| verified_fields: tuple[str, ...] = () | |
| cmr_sha256: str | None = None | |
| pdf_sha256: str | None = None | |
| discovered_source_index: int | None = None | |
| discovered_url: str | None = None | |
| failure: SmokeFailure | None = None | |
| class SmokeMetrics(SmokeModel): | |
| document: dict[str, Any] = Field(default_factory=dict) | |
| product: dict[str, Any] = Field(default_factory=dict) | |
| evidence: dict[str, Any] = Field(default_factory=dict) | |
| classification: dict[str, Any] = Field(default_factory=dict) | |
| comparison: dict[str, Any] = Field(default_factory=dict) | |
| cache: dict[str, Any] = Field(default_factory=dict) | |
| latency: dict[str, Any] = Field(default_factory=dict) | |
| usage: dict[str, Any] = Field(default_factory=dict) | |
| class DatasetSmokeCaseReport(SmokeModel): | |
| case_id: str | |
| native_id: str | |
| expected_coverage_mode: CoverageMode | |
| actual_coverage_mode: CoverageMode | None = None | |
| preflight: FixturePreflightReport | |
| selected_source_index: int | None = None | |
| selected_url_sha256: str | None = None | |
| stage_outcomes: tuple[str, ...] = () | |
| processing_status: str | |
| classification_outcome: str | None = None | |
| review_required: bool | |
| vocabulary_valid_count: int = 0 | |
| vocabulary_denominator: int = 0 | |
| vocabulary_validity: float | Literal["not_applicable"] = "not_applicable" | |
| metrics: SmokeMetrics = Field(default_factory=SmokeMetrics) | |
| event_trace: tuple[str, ...] = () | |
| warnings: tuple[str, ...] = () | |
| structural_result: Literal["passed", "failed", "review_required"] | |
| failure: SmokeFailure | None = None | |
| class DatasetSmokeRunRequest(SmokeModel): | |
| manifest_path: Path = Path("tests/fixtures/datasets/cases.json") | |
| output_root: Path | |
| report_path: Path | None = None | |
| class DatasetSmokeRunReport(SmokeModel): | |
| schema_version: Literal["dataset-smoke-report-v1"] = "dataset-smoke-report-v1" | |
| run_status: Literal["passed", "failed"] | |
| case_reports: tuple[DatasetSmokeCaseReport, ...] | |
| case_totals: dict[str, int] | |
| coverage_mode_agreement_count: int | |
| vocabulary_valid_count: int | |
| vocabulary_denominator: int | |
| vocabulary_validity: float | Literal["not_applicable"] | |
| comparison_category_totals: dict[str, int] | |
| model_call_total: int | |
| fixture_integrity: bool | |
| publication_regression: Literal["not_invoked"] = "not_invoked" | |
| evaluation_limitations: tuple[str, ...] = (FAKE_LIMITATION, REVIEW_NOTICE) | |
| logical_sha256: str | |
| class FrozenWorkflowTransport: | |
| """One-case transport that releases PDF bytes only after explicit app selection.""" | |
| def __init__(self, case: SmokeFixtureCase, root: Path) -> None: | |
| self.case = case | |
| self.root = root | |
| self.events: list[str] = [] | |
| self.cmr_byte_count = 0 | |
| self.pdf_byte_count = 0 | |
| self._authorized: tuple[str, str, int, str] | None = None | |
| def authorize(self, *, run_id: str, session_id: str, source_index: int, url: str) -> None: | |
| if ( | |
| source_index != self.case.selected_readme_source_index | |
| or url != self.case.selected_readme_url | |
| ): | |
| raise ValueError("selection does not match frozen discovered README") | |
| self._authorized = (run_id, session_id, source_index, url) | |
| self.events.append("explicit_selection_authorized") | |
| def handler(self, request: httpx.Request) -> httpx.Response: | |
| url = str(request.url) | |
| if url == self.case.cmr_url: | |
| self.events.append("cmr_response_served") | |
| body = (self.root / self.case.cmr_fixture).read_bytes() | |
| self.cmr_byte_count = len(body) | |
| return httpx.Response( | |
| 200, | |
| headers={"content-type": "application/vnd.nasa.cmr.umm+json"}, | |
| content=body, | |
| request=request, | |
| ) | |
| if url == self.case.selected_readme_url: | |
| if self._authorized is None: | |
| raise httpx.TransportError("README mapping denied before explicit selection") | |
| self.events.append("selected_pdf_served") | |
| body = (self.root / self.case.readme_fixture).read_bytes() | |
| self.pdf_byte_count = len(body) | |
| return httpx.Response( | |
| 200, | |
| headers={"content-type": "application/pdf", "content-length": str(len(body))}, | |
| content=body, | |
| request=request, | |
| ) | |
| raise httpx.TransportError("unconfigured frozen URL") | |
| def client(self) -> httpx.Client: | |
| return httpx.Client(transport=httpx.MockTransport(self.handler), trust_env=False) | |
| class StructuralFakeModelClient: | |
| """Blind adaptive fake authored from runtime prompt contracts, never expert values.""" | |
| def __init__(self, case: SmokeFixtureCase) -> None: | |
| self.case = case | |
| self.requests: list[ModelRequest[Any]] = [] | |
| self.input_tokens = 0 | |
| self.output_tokens = 0 | |
| def generate_structured(self, request: ModelRequest[Any]) -> ModelResponse[Any]: | |
| self.requests.append(request) | |
| self.input_tokens += max(1, len(request.prompt) // 4) | |
| action = self._action(request) | |
| parsed = request.response_schema.model_validate(action) | |
| output = len(parsed.model_dump_json()) // 4 | |
| self.output_tokens += max(1, output) | |
| return ModelResponse( | |
| parsed=parsed, | |
| provider="fake", | |
| model_name="dataset-structural-fake-v1", | |
| prompt_version=request.prompt_version, | |
| duration_seconds=0.01, | |
| token_usage=TokenUsage( | |
| input_tokens=max(1, len(request.prompt) // 4), | |
| output_tokens=max(1, output), | |
| total_tokens=max(2, len(request.prompt) // 4 + output), | |
| ), | |
| estimated_cost=None, | |
| diagnostics={"usage_kind": "simulated"}, | |
| ) | |
| def _action(self, request: ModelRequest[Any]) -> dict[str, Any]: | |
| blocks = _prompt_blocks(request.prompt) | |
| if request.response_schema is CoverageModelResponse: | |
| vague = self.case.expected_coverage_mode is CoverageMode.VAGUE_PRODUCT_SCOPE | |
| return { | |
| "coverage_mode": self.case.expected_coverage_mode.value, | |
| "qualifying_block_ids": [blocks[0]["block_id"]] if blocks else [], | |
| "rejected_block_ids": [], | |
| "determination_method": "model", | |
| "rationale": "Structural fixture coverage response independent of expert keywords.", | |
| "product_match_confidence": {"value": 0.35 if vague else 0.9, "calibrated": False}, | |
| "review_required": vague, | |
| } | |
| if request.response_schema is ProductResolutionModelResponse: | |
| target = _prompt_target(request.prompt) | |
| declarations = [ | |
| *_prompt_records(request.prompt, "BATCH_LOCAL_DECLARATIONS"), | |
| *_prompt_records(request.prompt, "LOCAL_CONTEXT_DECLARATIONS"), | |
| ] | |
| return { | |
| "decisions": [ | |
| self._product_decision( | |
| block, | |
| target["document_id"], | |
| declarations, | |
| ) | |
| for block in blocks | |
| ] | |
| } | |
| if request.response_schema is EvidenceMinimizationModelResponse: | |
| return { | |
| "decisions": [ | |
| { | |
| "block_id": block["block_id"], | |
| "retain": index == 0, | |
| "removal_reason": (None if index == 0 else "redundant_science_evidence"), | |
| "prompt_version": "dataset_evidence_minimization_v1", | |
| } | |
| for index, block in enumerate(blocks) | |
| ] | |
| } | |
| if request.response_schema is EvidenceSelectionModelResponse: | |
| decisions = [] | |
| reduction = bool(request.metadata.get("reduction", False)) | |
| retained_one = False | |
| for block in blocks: | |
| if block.get("membership") != "decision_target": | |
| continue | |
| eligible = block.get("evidence_eligibility") == "eligible" | |
| if reduction and eligible: | |
| eligible = not retained_one | |
| retained_one = True | |
| decisions.append( | |
| { | |
| "block_id": block["block_id"], | |
| "selected": eligible, | |
| "scientific_relevance": "relevant" if eligible else "not_relevant", | |
| "evidence_role": "scientific_description" if eligible else "other", | |
| "scope_type": block["scope_type"], | |
| "evidence_eligibility": block["evidence_eligibility"], | |
| "product_match_confidence": block["product_match_confidence"], | |
| "qualifying_block_ids": list(block.get("qualifying_block_ids", [])), | |
| "qualifying_declaration_ids": list( | |
| block.get("qualifying_declaration_ids", []) | |
| ), | |
| "rationale": "Deterministic structural evidence decision.", | |
| "review_required": bool(block.get("review_required", False)), | |
| "prompt_version": "dataset_evidence_selection_v2", | |
| "warnings": [], | |
| } | |
| ) | |
| return {"decisions": decisions} | |
| citations = [ | |
| block["block_id"] for block in blocks if block.get("independently_eligible", False) | |
| ] | |
| citation = citations[:1] or [blocks[0]["block_id"]] | |
| if request.response_schema.__name__ == "DatasetTopicResponse": | |
| candidates = _prompt_candidates(request.prompt) | |
| return { | |
| "selected": [ | |
| { | |
| "candidate_id": candidates[0]["candidate_id"], | |
| "confidence": 0.85, | |
| "cited_block_ids": citation, | |
| "rationale": "Supported for structural traversal by cited packet evidence.", | |
| } | |
| ] | |
| } | |
| if request.response_schema.__name__ == "DatasetChildResponse": | |
| return { | |
| "mode": "stop_at_parent", | |
| "selected": [], | |
| "reason": "The parent is the deepest supported concept in this structural fake.", | |
| "cited_block_ids": citation, | |
| } | |
| raise ValueError(f"unsupported smoke response schema: {request.response_schema.__name__}") | |
| def _product_decision( | |
| self, | |
| block: dict[str, Any], | |
| document_id: str, | |
| declarations: list[dict[str, Any]], | |
| ) -> dict[str, Any]: | |
| vague = self.case.expected_coverage_mode is CoverageMode.VAGUE_PRODUCT_SCOPE | |
| text = str(block.get("text", "")) | |
| local_family = next( | |
| ( | |
| item | |
| for item in declarations | |
| if item.get("kind") == "product_family" | |
| and item.get("target_included") is True | |
| and block["block_id"] in item.get("locality", {}).get("allowed_block_ids", []) | |
| ), | |
| None, | |
| ) | |
| other_esmr = ( | |
| self.case.case_id == "esmr_explicit_multi_product" | |
| and "ESMRN5L1" in text | |
| and self.case.short_name not in text | |
| ) | |
| declaration_ids: list[str] = [] | |
| if vague: | |
| scope, eligibility, confidence, review = "ambiguous", "review_only", 0.35, True | |
| elif other_esmr: | |
| scope, eligibility, confidence, review = "other_product", "excluded", 0.9, False | |
| elif ( | |
| self.case.expected_coverage_mode is CoverageMode.PRODUCT_FAMILY | |
| and local_family is not None | |
| and (self.case.short_name in text or "MERRA-2" in text) | |
| ): | |
| scope, eligibility, confidence, review = "product_family", "eligible", 0.9, False | |
| declaration_ids = [local_family["declaration_id"]] | |
| else: | |
| scope, eligibility, confidence, review = "exact_target", "eligible", 0.9, False | |
| return { | |
| "block_id": block["block_id"], | |
| "document_id": document_id, | |
| "page_number": block.get("page_number", block.get("page", 1)), | |
| "section_id": block["section_id"], | |
| "scope_type": scope, | |
| "eligibility": eligibility, | |
| "applicable_products": [self.case.short_name] if eligibility == "eligible" else [], | |
| "family_label": None, | |
| "qualifying_block_ids": [block["block_id"]], | |
| "declaration_ids": declaration_ids, | |
| "aliases_used": ( | |
| [self.case.short_name] | |
| if eligibility == "eligible" and scope != "product_family" | |
| else [] | |
| ), | |
| "determination_method": "model", | |
| "rationale": "Deterministic structural product-scope decision.", | |
| "product_match_confidence": {"value": confidence, "calibrated": False}, | |
| "review_required": review, | |
| "warnings": [], | |
| } | |
| class DatasetSmokeHarness: | |
| """Run structural evaluation through the real two-step application service.""" | |
| def run(self, request: DatasetSmokeRunRequest) -> DatasetSmokeRunReport: | |
| manifest_path = request.manifest_path.resolve() | |
| root = manifest_path.parent | |
| before = _fixture_snapshot(root) | |
| try: | |
| manifest = SmokeFixtureManifest.model_validate_json(manifest_path.read_bytes()) | |
| except (OSError, ValidationError) as exc: | |
| raise ValueError("fixture manifest is invalid") from exc | |
| case_reports = tuple( | |
| self._run_case(case, root, request.output_root) for case in manifest.cases | |
| ) | |
| after = _fixture_snapshot(root) | |
| integrity = before == after | |
| totals = Counter(report.structural_result for report in case_reports) | |
| comparison = Counter() | |
| for report in case_reports: | |
| comparison.update(report.metrics.comparison.get("category_counts", {})) | |
| valid = sum(report.vocabulary_valid_count for report in case_reports) | |
| denominator = sum(report.vocabulary_denominator for report in case_reports) | |
| core = { | |
| "case_reports": [report.model_dump(mode="json") for report in case_reports], | |
| "case_totals": dict(sorted(totals.items())), | |
| "coverage_mode_agreement_count": sum( | |
| item.actual_coverage_mode == item.expected_coverage_mode for item in case_reports | |
| ), | |
| "vocabulary_valid_count": valid, | |
| "vocabulary_denominator": denominator, | |
| "comparison_category_totals": dict(sorted(comparison.items())), | |
| "model_call_total": sum( | |
| item.metrics.classification.get("current_run_model_calls", 0) | |
| for item in case_reports | |
| ), | |
| "fixture_integrity": integrity, | |
| } | |
| failed = any(item.structural_result == "failed" for item in case_reports) or not integrity | |
| logical_hash = _canonical_hash(core) | |
| report = DatasetSmokeRunReport( | |
| run_status="failed" if failed else "passed", | |
| case_reports=case_reports, | |
| case_totals=core["case_totals"], | |
| coverage_mode_agreement_count=core["coverage_mode_agreement_count"], | |
| vocabulary_valid_count=valid, | |
| vocabulary_denominator=denominator, | |
| vocabulary_validity=(valid / denominator * 100 if denominator else "not_applicable"), | |
| comparison_category_totals=core["comparison_category_totals"], | |
| model_call_total=core["model_call_total"], | |
| fixture_integrity=integrity, | |
| logical_sha256=logical_hash, | |
| ) | |
| if request.report_path is not None: | |
| _persist_report(report, request.report_path, request.output_root, root) | |
| return report | |
| def preflight(self, case: SmokeFixtureCase, root: Path) -> FixturePreflightReport: | |
| fields: list[str] = [] | |
| try: | |
| cmr_path = _safe_fixture_file(root, case.cmr_fixture) | |
| pdf_path = _safe_fixture_file(root, case.readme_fixture) | |
| cmr_bytes, pdf_bytes = cmr_path.read_bytes(), pdf_path.read_bytes() | |
| _require(_sha(cmr_bytes) == case.cmr_sha256, "cmr_sha256") | |
| _require(_sha(pdf_bytes) == case.readme_sha256, "readme_sha256") | |
| source = json.loads(cmr_bytes) | |
| _require(source.get("hits") == 1 and len(source.get("items", [])) == 1, "cardinality") | |
| item = source["items"][0] | |
| meta, umm = item["meta"], item["umm"] | |
| expected = { | |
| "concept_id": (meta.get("concept-id"), case.expected_concept_id), | |
| "native_id": (meta.get("native-id"), case.native_id), | |
| "revision_id": (meta.get("revision-id"), case.cmr_revision_id), | |
| "short_name": (umm.get("ShortName"), case.short_name), | |
| "version": (umm.get("Version"), case.version), | |
| } | |
| for name, values in expected.items(): | |
| _require(values[0] == values[1], name) | |
| fields.append(name) | |
| _require(case.native_id == f"{case.short_name}_{case.version}", "derived_native_id") | |
| _validate_timestamps(case) | |
| validated_url = validate_cmr_url(case.cmr_url) | |
| cmr_source = CMRRetrievedSource( | |
| submitted_url=case.cmr_url, | |
| final_url=case.cmr_url, | |
| retrieved_at="1970-01-01T00:00:00Z", | |
| status_code=200, | |
| response_metadata={"content-type": "application/json"}, | |
| response_bytes=cmr_bytes, | |
| source_text=cmr_bytes.decode("utf-8"), | |
| sha256=_sha(cmr_bytes), | |
| parsed_json=source, | |
| ) | |
| resolved = resolve_cmr_collection(validated_url, cmr_source) | |
| discovery = discover_readme_candidates( | |
| resolved.blind_view, cmr_source_sha256=cmr_source.sha256 | |
| ) | |
| related = resolved.blind_view.related_urls | |
| _require(case.selected_readme_source_index < len(related), "source_index") | |
| entry = related[case.selected_readme_source_index] | |
| _require(entry.get("Subtype") == "READ-ME", "readme_subtype") | |
| _require(entry.get("URL") == case.selected_readme_url, "selected_readme_url") | |
| candidate = next( | |
| ( | |
| item | |
| for item in discovery.candidates | |
| if item.source_index == case.selected_readme_source_index | |
| ), | |
| None, | |
| ) | |
| _require( | |
| candidate is not None and candidate.url == case.selected_readme_url, | |
| "production_discovery", | |
| ) | |
| fields.extend( | |
| ("cmr_sha256", "readme_sha256", "timestamps", "source_index", "selected_readme_url") | |
| ) | |
| return FixturePreflightReport( | |
| case_id=case.case_id, | |
| passed=True, | |
| verified_fields=tuple(fields), | |
| cmr_sha256=case.cmr_sha256, | |
| pdf_sha256=case.readme_sha256, | |
| discovered_source_index=case.selected_readme_source_index, | |
| discovered_url=case.selected_readme_url, | |
| ) | |
| except Exception as exc: | |
| field = str(exc) or "preflight" | |
| return FixturePreflightReport( | |
| case_id=case.case_id, | |
| passed=False, | |
| failure=SmokeFailure( | |
| case_id=case.case_id, | |
| field=field, | |
| code="FIXTURE_PREFLIGHT_FAILED", | |
| message="Frozen fixture provenance validation failed.", | |
| ), | |
| ) | |
| def _run_case( | |
| self, case: SmokeFixtureCase, root: Path, output_root: Path | |
| ) -> DatasetSmokeCaseReport: | |
| preflight = self.preflight(case, root) | |
| if not preflight.passed: | |
| return _failed_case(case, preflight) | |
| transport = FrozenWorkflowTransport(case, root) | |
| fake = StructuralFakeModelClient(case) | |
| case_root = output_root / case.case_id | |
| settings = DatasetSettings( | |
| runtime=DatasetRuntimeSettings( | |
| output_root=str(case_root / "runs"), | |
| cache_root=str(case_root / "cache"), | |
| fixture_root=str(root), | |
| ) | |
| ) | |
| client = transport.client() | |
| service = build_dataset_application_service( | |
| settings=settings, | |
| model_settings=ModelSettings(provider="fake", model_name="dataset-structural-fake-v1"), | |
| cmr_client=client, | |
| readme_client=client, | |
| resolver=lambda _host: ("93.184.216.34",), | |
| vocabulary=load_vocabulary(Path("data/gcmd_hierarchy.json")), | |
| model_client=fake, | |
| ) | |
| session_id = f"smoke-{case.case_id}" | |
| retrieved = service.retrieve_collection( | |
| RetrieveCollectionRequest(cmr_url=case.cmr_url, session_id=session_id) | |
| ) | |
| if ( | |
| retrieved.status != ApplicationOperationStatus.SUCCESS | |
| or retrieved.run is None | |
| or retrieved.waiting_reference is None | |
| ): | |
| return _operation_failure( | |
| case, preflight, transport, "retrieve_collection", retrieved.error | |
| ) | |
| candidate = next( | |
| ( | |
| item | |
| for item in retrieved.candidates | |
| if item.source_position == case.selected_readme_source_index | |
| and item.exact_url == case.selected_readme_url | |
| ), | |
| None, | |
| ) | |
| if candidate is None: | |
| return _operation_failure(case, preflight, transport, "readme_discovery", None) | |
| transport.events.append("candidate_discovered_by_application") | |
| transport.authorize( | |
| run_id=retrieved.run.run_id, | |
| session_id=session_id, | |
| source_index=candidate.source_position, | |
| url=candidate.exact_url, | |
| ) | |
| processed = service.classify_selected_readme( | |
| ClassifySelectedReadmeRequest( | |
| run=retrieved.run, | |
| session=retrieved.session, | |
| waiting_reference=retrieved.waiting_reference, | |
| source_position=candidate.source_position, | |
| exact_url=candidate.exact_url, | |
| ) | |
| ) | |
| artifacts = _load_case_artifacts(service, retrieved.run, session_id) | |
| return _build_case_report(case, preflight, transport, fake, processed, artifacts) | |
| def _prompt_records(prompt: str, key: str) -> list[dict[str, Any]]: | |
| prefix = f"{key}=" | |
| for line in prompt.splitlines(): | |
| if line.startswith(prefix): | |
| value = json.loads(line.removeprefix(prefix)) | |
| return value if isinstance(value, list) else [] | |
| return [] | |
| def _prompt_blocks(prompt: str) -> list[dict[str, Any]]: | |
| return [json.loads(item) for item in _BLOCK_RE.findall(prompt)] | |
| def _prompt_candidates(prompt: str) -> list[dict[str, Any]]: | |
| match = _CANDIDATE_RE.search(prompt) | |
| if match is None: | |
| raise ValueError("hierarchy prompt candidates are unavailable") | |
| return json.loads(match.group(1)) | |
| def _prompt_target(prompt: str) -> dict[str, Any]: | |
| line = next((item for item in prompt.splitlines() if item.startswith("TARGET=")), None) | |
| if line is None: | |
| raise ValueError("product prompt target is unavailable") | |
| return json.loads(line.removeprefix("TARGET=")) | |
| def _sha(data: bytes) -> str: | |
| return hashlib.sha256(data).hexdigest() | |
| def _canonical_hash(value: Any) -> str: | |
| return _sha(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()) | |
| def _safe_fixture_file(root: Path, relative_value: str) -> Path: | |
| relative = Path(relative_value) | |
| if relative.is_absolute(): | |
| raise ValueError("fixture_path") | |
| path = (root / relative).resolve() | |
| try: | |
| path.relative_to(root.resolve()) | |
| except ValueError as exc: | |
| raise ValueError("fixture_path") from exc | |
| if not path.is_file() or path.is_symlink(): | |
| raise ValueError("fixture_regular_file") | |
| return path | |
| def _require(condition: bool, field: str) -> None: | |
| if not condition: | |
| raise ValueError(field) | |
| def _validate_timestamps(case: SmokeFixtureCase) -> None: | |
| values = (case.cmr_retrieved_at, case.readme_retrieved_at) | |
| if case.retrieval_time_status == "unknown_preexisting_fixture": | |
| _require(all(value is None for value in values), "retrieval_timestamp") | |
| else: | |
| _require(all(value is not None for value in values), "retrieval_timestamp") | |
| for value in values: | |
| datetime.fromisoformat(str(value).replace("Z", "+00:00")) | |
| datetime.fromisoformat(case.fixture_recorded_at.replace("Z", "+00:00")) | |
| def _fixture_snapshot(root: Path) -> dict[str, str]: | |
| return { | |
| str(path.relative_to(root)): _sha(path.read_bytes()) | |
| for path in sorted(root.rglob("*")) | |
| if path.is_file() | |
| } | |
| def _load_case_artifacts( | |
| service: DatasetApplicationService, run: Any, session_id: str | |
| ) -> dict[str, Any]: | |
| directory = service.store.run_directory(run, session_id) | |
| mapping: tuple[tuple[str, type[BaseModel]], ...] = ( | |
| ("artifacts/extraction_manifest.json", DatasetExtractionManifest), | |
| ("artifacts/product_resolution.json", DeterministicScopeResult), | |
| ("artifacts/evidence_packet.json", DeterministicEvidencePacket), | |
| ("blind_result.json", DatasetClassificationResult), | |
| ) | |
| values: dict[str, Any] = {} | |
| for relative, model in mapping: | |
| path = directory / relative | |
| if path.is_file(): | |
| values[relative] = model.model_validate_json(path.read_bytes()) | |
| return values | |
| def _build_case_report( | |
| case, preflight, transport, fake, result, artifacts | |
| ) -> DatasetSmokeCaseReport: | |
| extraction = artifacts.get("artifacts/extraction_manifest.json") | |
| product = artifacts.get("artifacts/product_resolution.json") | |
| packet = artifacts.get("artifacts/evidence_packet.json") | |
| blind = artifacts.get("blind_result.json") or result.blind_result | |
| classifications = tuple(blind.classifications) if blind is not None else () | |
| valid = sum(item.deterministic_validation.valid for item in classifications) | |
| denominator = len(classifications) | |
| comparison_counts: Counter[str] = Counter() | |
| if result.comparison is not None: | |
| comparison_counts.update(item.category.value for item in result.comparison.comparison_items) | |
| coverage = product.coverage_mode if product is not None else None | |
| expected_review = case.expected_review_required | |
| is_vague = case.expected_coverage_mode is CoverageMode.VAGUE_PRODUCT_SCOPE | |
| success = ( | |
| preflight.passed | |
| and valid == denominator | |
| and "candidate_discovered_by_application" in transport.events | |
| and "explicit_selection_authorized" in transport.events | |
| and ("selected_pdf_served" in transport.events) | |
| and result.status | |
| in {ApplicationOperationStatus.SUCCESS, ApplicationOperationStatus.SKIPPED} | |
| and ( | |
| result.classification_attempted == (result.status == ApplicationOperationStatus.SUCCESS) | |
| ) | |
| ) | |
| scope_counts = Counter( | |
| decision.scope.value for decision in (product.decisions if product else ()) | |
| ) | |
| role_counts = Counter(item.role.value for item in (packet.components if packet else ())) | |
| stages = tuple(f"{item.stage.value}:{item.status.value}" for item in result.stages) | |
| metrics = SmokeMetrics( | |
| document={ | |
| "cmr_retrieval_status": "served", | |
| "readme_candidate_count": 1, | |
| "selected_source_index": case.selected_readme_source_index, | |
| "cmr_response_bytes": transport.cmr_byte_count, | |
| "pdf_bytes": transport.pdf_byte_count, | |
| "page_count": extraction.page_count if extraction else 0, | |
| "extracted_character_count": extraction.character_count if extraction else 0, | |
| "extraction_warnings": [item.code for item in extraction.warnings] | |
| if extraction | |
| else [], | |
| }, | |
| product={ | |
| "coverage_mode": coverage, | |
| "expected_coverage_mode": str(case.expected_coverage_mode), | |
| "coverage_mode_agreement": coverage == case.expected_coverage_mode, | |
| "product_match_confidence": (None), | |
| "review_required": result.product_diagnostics.get("review_required", False), | |
| "scope_counts": dict(sorted(scope_counts.items())), | |
| "included_block_count": result.product_diagnostics.get("included_block_count", 0), | |
| "excluded_block_count": result.product_diagnostics.get("excluded_block_count", 0), | |
| }, | |
| evidence={ | |
| "packet_block_count": len(packet.components) if packet else 0, | |
| "packet_character_count": packet.character_count if packet else 0, | |
| "evidence_role_counts": dict(sorted(role_counts.items())), | |
| "packet_readiness": "ready" if packet else "review_blocked", | |
| "packet_hash_verified": bool(packet and len(packet.packet_sha256) == 64), | |
| "reduction_required": False, | |
| "removed_block_count": (0), | |
| }, | |
| classification={ | |
| "classification_attempted": result.classification_attempted, | |
| "current_run_model_calls": result.current_call_model_calls, | |
| "persisted_model_calls": result.persisted_model_calls, | |
| "accepted_terminal_path_count": denominator, | |
| "valid_terminal_path_count": valid, | |
| "vocabulary_validity_percentage": valid / denominator * 100 | |
| if denominator | |
| else "not_applicable", | |
| "citation_count": sum(len(item.citations) for item in classifications), | |
| "keyword_evidence_confidence": [ | |
| item.keyword_evidence_confidence.model_dump(mode="json") for item in classifications | |
| ], | |
| "semantic_not_classified": bool( | |
| blind and blind.classification_outcome.value == "not_classified" | |
| ), | |
| }, | |
| comparison={ | |
| "eligible": result.verified_blind_result is not None, | |
| "attempted": result.verified_blind_result is not None, | |
| "completed": result.comparison is not None, | |
| "category_counts": dict(sorted(comparison_counts.items())), | |
| "additive_review_record_count": ( | |
| sum(1 for item in result.comparison.comparison_items if item.review_required) | |
| if result.comparison | |
| else 0 | |
| ), | |
| }, | |
| cache={ | |
| "hit": sum(item.cache_used for item in result.stages), | |
| "miss": 0, | |
| "rejected": 0, | |
| "bypass": len(result.stages), | |
| "checkpoint_reuse": 0, | |
| }, | |
| latency={ | |
| "kind": "simulated", | |
| "per_model_call_seconds": 0.01, | |
| "total_model_seconds": round(len(fake.requests) * 0.01, 6), | |
| }, | |
| usage={ | |
| "kind": "simulated", | |
| "prompt_tokens": fake.input_tokens, | |
| "completion_tokens": fake.output_tokens, | |
| "total_tokens": fake.input_tokens + fake.output_tokens, | |
| "cost": {"status": "unavailable", "reason": "no approved pricing configuration"}, | |
| "model_identity": "dataset-structural-fake-v1", | |
| }, | |
| ) | |
| return DatasetSmokeCaseReport( | |
| case_id=case.case_id, | |
| native_id=case.native_id, | |
| expected_coverage_mode=case.expected_coverage_mode, | |
| actual_coverage_mode=coverage, | |
| preflight=preflight, | |
| selected_source_index=case.selected_readme_source_index, | |
| selected_url_sha256=_sha(case.selected_readme_url.encode()), | |
| stage_outcomes=stages, | |
| processing_status=result.status, | |
| classification_outcome=blind.classification_outcome.value if blind else None, | |
| review_required=expected_review or bool(result.product_diagnostics.get("review_required")), | |
| vocabulary_valid_count=valid, | |
| vocabulary_denominator=denominator, | |
| vocabulary_validity=valid / denominator * 100 if denominator else "not_applicable", | |
| metrics=metrics, | |
| event_trace=tuple(transport.events), | |
| warnings=(FAKE_LIMITATION, REVIEW_NOTICE), | |
| structural_result=( | |
| "review_required" if success and is_vague else "passed" if success else "failed" | |
| ), | |
| failure=( | |
| None | |
| if success | |
| else SmokeFailure( | |
| case_id=case.case_id, | |
| field="pipeline", | |
| code="STRUCTURAL_SMOKE_FAILED", | |
| message="One or more structural invariants failed.", | |
| ) | |
| ), | |
| ) | |
| def _failed_case(case, preflight): | |
| return DatasetSmokeCaseReport( | |
| case_id=case.case_id, | |
| native_id=case.native_id, | |
| expected_coverage_mode=case.expected_coverage_mode, | |
| preflight=preflight, | |
| processing_status="failed", | |
| review_required=False, | |
| structural_result="failed", | |
| failure=preflight.failure, | |
| ) | |
| def _operation_failure(case, preflight, transport, field, error): | |
| code = getattr(error, "code", "APPLICATION_OPERATION_FAILED") | |
| return DatasetSmokeCaseReport( | |
| case_id=case.case_id, | |
| native_id=case.native_id, | |
| expected_coverage_mode=case.expected_coverage_mode, | |
| preflight=preflight, | |
| processing_status="failed", | |
| review_required=False, | |
| event_trace=tuple(transport.events), | |
| structural_result="failed", | |
| failure=SmokeFailure( | |
| case_id=case.case_id, field=field, code=code, message="Application operation failed." | |
| ), | |
| ) | |
| def _persist_report( | |
| report: DatasetSmokeRunReport, path: Path, output_root: Path, fixture_root: Path | |
| ) -> None: | |
| resolved_root = output_root.resolve() | |
| resolved = path.resolve() | |
| try: | |
| resolved.relative_to(resolved_root) | |
| except ValueError as exc: | |
| raise ValueError("smoke report must be beneath the configured output root") from exc | |
| try: | |
| resolved.relative_to(fixture_root.resolve()) | |
| except ValueError: | |
| pass | |
| else: | |
| raise ValueError("smoke report cannot be written beneath frozen fixtures") | |
| resolved.parent.mkdir(parents=True, exist_ok=True) | |
| resolved.write_text(report.model_dump_json(indent=2) + "\n") | |
| def main() -> int: | |
| parser = argparse.ArgumentParser( | |
| description="Run deterministic dataset structural smoke evaluation." | |
| ) | |
| parser.add_argument("--output-root", required=True, type=Path) | |
| parser.add_argument("--manifest", type=Path, default=Path("tests/fixtures/datasets/cases.json")) | |
| parser.add_argument("--report", type=Path) | |
| args = parser.parse_args() | |
| report_path = args.report or args.output_root / "dataset-smoke-report.json" | |
| report = DatasetSmokeHarness().run( | |
| DatasetSmokeRunRequest( | |
| manifest_path=args.manifest, | |
| output_root=args.output_root, | |
| report_path=report_path, | |
| ) | |
| ) | |
| print(report.model_dump_json(indent=2)) | |
| return 0 if report.run_status == "passed" else 1 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |