"""Deterministic browser data contract for the R&D Explorer.""" from __future__ import annotations import json from typing import Any, Mapping EXPLORER_CONTRACT_VERSION = "mountain_twin_explorer_v0_7" SUPPORTED_EXPLORER_CONTRACT_VERSIONS = frozenset( { "mountain_twin_explorer_v0_1", "mountain_twin_explorer_v0_2", "mountain_twin_explorer_v0_3", "mountain_twin_explorer_v0_4", "mountain_twin_explorer_v0_5", "mountain_twin_explorer_v0_6", "mountain_twin_explorer_v0_7", } ) SOLAR_DISPLAY_STATES = frozenset({"DIRECT", "SHADOW", "UNKNOWN", "NOT_APPLICABLE", "NOT_EVALUATED"}) RCI_COVERAGE_STATES = frozenset({"FULL", "PARTIAL", "UNAVAILABLE", "UNKNOWN", "NOT_APPLICABLE"}) def serialize_explorer_document(document: Mapping[str, Any]) -> str: """Validate and serialize an Explorer document deterministically.""" validate_explorer_document(document) return json.dumps(document, indent=2, sort_keys=True, allow_nan=False) + "\n" def validate_explorer_document(document: Mapping[str, Any]) -> None: """Reject exports that could visually imply unavailable solar coverage.""" contract_version = document.get("contract_version") if contract_version not in SUPPORTED_EXPLORER_CONTRACT_VERSIONS: raise ValueError("unsupported Explorer contract version") route = document.get("route", {}) route_points = document.get("route_points", []) if route.get("route_id") != "tmb_day_01": raise ValueError("Explorer v0.1 requires TMB Day 1") if route.get("point_count") != len(route_points): raise ValueError("route point count does not match route geometry") if not route_points or [point.get("point_index") for point in route_points] != list( range(len(route_points)) ): raise ValueError("route points must be ordered and contiguous") scenarios = document.get("scenarios", []) if {scenario.get("name") for scenario in scenarios} != {"FAST", "NOMINAL", "SLOW"}: raise ValueError("Explorer requires FAST, NOMINAL, and SLOW scenarios") for scenario in scenarios: coverage = scenario.get("coverage", {}) weather = coverage.get("weather", {}) solar = coverage.get("solar", {}) total = route["point_count"] if weather.get("total_points") != total: raise ValueError("weather coverage total does not match route") weather_mode = weather.get("mode") weather_evaluated = weather.get("evaluated_points") if weather_mode == "FULL" and weather_evaluated != total: raise ValueError("FULL weather coverage requires every route point") if weather_mode not in {"FULL", "PARTIAL", "UNAVAILABLE"}: raise ValueError("Explorer requires explicit weather coverage mode") if not isinstance(weather_evaluated, int) or not 0 <= weather_evaluated <= total: raise ValueError("weather coverage evaluated points are invalid") if weather_mode == "UNAVAILABLE" and weather_evaluated != 0: raise ValueError("UNAVAILABLE weather coverage cannot evaluate route points") if solar.get("total_points") != total: raise ValueError("solar coverage total does not match route") if contract_version == "mountain_twin_explorer_v0_1": if solar.get("evaluated_points") != 100 or solar.get("mode") != "SAMPLED": raise ValueError("Explorer v0.1 requires the 100-point sampled solar subset") elif solar.get("evaluated_points") != total or solar.get("mode") != "FULL": raise ValueError("Explorer v0.2 requires FULL solar coverage") points = scenario.get("points", []) if len(points) != total: raise ValueError("each Explorer scenario must include full route geometry") for point in points: state = point.get("solar", {}).get("status") if state not in SOLAR_DISPLAY_STATES: raise ValueError(f"invalid Explorer solar display state: {state}") if state == "NOT_EVALUATED" and point.get("solar", {}).get("evaluated") is not False: raise ValueError("NOT_EVALUATED must explicitly be unevaluated") if state != "NOT_EVALUATED" and point.get("solar", {}).get("evaluated") is not True: raise ValueError("evaluated solar states must be explicit") selected_count = sum(point["solar"]["status"] != "NOT_EVALUATED" for point in points) if selected_count != solar["evaluated_points"]: raise ValueError("displayed solar states do not match sampled coverage") if contract_version in { "mountain_twin_explorer_v0_3", "mountain_twin_explorer_v0_4", "mountain_twin_explorer_v0_5", }: _validate_rci_projection(scenario.get("rci")) if contract_version in {"mountain_twin_explorer_v0_4", "mountain_twin_explorer_v0_5"}: _validate_photographer_projection(scenario.get("photographer"), scenario.get("name")) if contract_version in { "mountain_twin_explorer_v0_5", "mountain_twin_explorer_v0_6", "mountain_twin_explorer_v0_7", }: _validate_snow_projection(scenario.get("snow"), scenario.get("name"), total) if contract_version in {"mountain_twin_explorer_v0_6", "mountain_twin_explorer_v0_7"}: _validate_history_projection(scenario.get("history"), scenario.get("name")) if contract_version == "mountain_twin_explorer_v0_7": _validate_trail_terrain_character(document.get("trail_terrain_character"), route) def _validate_trail_terrain_character(character: Any, route: Mapping[str, Any]) -> None: """Validate compact factual mapped-character data; never a route point matrix.""" if not isinstance(character, Mapping): raise ValueError("Explorer v0.7 requires Trail / Terrain Character") if character.get("contract_version") != "trail_terrain_character_explorer_v0_1": raise ValueError("Explorer Trail / Terrain Character contract is unsupported") if character.get("availability") != "AVAILABLE": raise ValueError("Explorer Trail / Terrain Character availability is invalid") if any(key in character for key in ("points", "route_points", "geometry", "confidence", "safety")): raise ValueError("Explorer Trail / Terrain Character contains prohibited payload fields") reference = character.get("route_reference") or {} if reference.get("route_id") != route.get("route_id"): raise ValueError("Explorer Trail / Terrain Character route identity is invalid") mapped = character.get("mapped_character") or {} policy = mapped.get("normalization_policy") or {} if not policy.get("policy_id") or not policy.get("version"): raise ValueError("Explorer Trail / Terrain Character normalization policy is required") if not isinstance(mapped.get("segments"), list) or not isinstance(mapped.get("characteristics"), list): raise ValueError("Explorer Trail / Terrain Character segments are required") for segment in mapped["segments"]: if segment.get("characteristic_id") not in {"route_class", "surface"}: raise ValueError("Explorer Trail / Terrain Character characteristic is invalid") if segment.get("value_state") not in {"KNOWN", "UNKNOWN", "AMBIGUOUS", "UNMATCHED"}: raise ValueError("Explorer Trail / Terrain Character value state is invalid") if not isinstance(segment.get("source_feature_ids"), list): raise ValueError("Explorer Trail / Terrain Character source references are required") def _validate_history_projection(history: Any, scenario_name: Any) -> None: """Validate a compact server-owned History evidence projection.""" if not isinstance(history, Mapping): raise ValueError("Explorer v0.6 requires a History projection") if history.get("contract_version") != "history_time_machine_explorer_v0_1": raise ValueError("Explorer History contract version is unsupported") if history.get("mode") != "RECONSTRUCTED_PAST": raise ValueError("Explorer History only supports RECONSTRUCTED_PAST") if history.get("availability") not in {"AVAILABLE", "UNAVAILABLE"}: raise ValueError("Explorer History availability is invalid") if any( key in history for key in ( "points", "route_point_matrix", "confidence", "safety", "severity", "recommendation", ) ): raise ValueError("Explorer History projection contains prohibited fields") if history["availability"] == "UNAVAILABLE": if history.get("engines") or not isinstance(history.get("reason_codes"), list): raise ValueError("unavailable Explorer History must retain explicit reasons") return window = history.get("requested_window") or {} if ( not window.get("requested_start") or not window.get("requested_end") or not window.get("timezone") ): raise ValueError("available Explorer History requires requested window") timelines = history.get("route_timeline_references") if not isinstance(timelines, list) or not timelines: raise ValueError("available Explorer History requires route/timeline references") if any(item.get("scenario_name") != scenario_name for item in timelines): raise ValueError("Explorer History scenario must match its route timeline") engines = history.get("engines") if not isinstance(engines, list) or not engines: raise ValueError("available Explorer History requires engine evidence") engine_ids = [item.get("engine_id") for item in engines] if len(engine_ids) != len(set(engine_ids)): raise ValueError("Explorer History engine IDs must be unique") allowed_classes = { "OBSERVED", "REANALYSIS", "HISTORICAL_FORECAST", "FORECAST", "DETERMINISTIC_ASTRONOMY", "TERRAIN_DERIVED", "DERIVED", "UNKNOWN", } for engine in engines: if engine.get("availability") not in RCI_COVERAGE_STATES: raise ValueError("Explorer History engine availability is invalid") if not isinstance(engine.get("result_references"), list): raise ValueError("Explorer History engine references are required") for entry in engine.get("evidence", []): if entry.get("evidence_class") not in allowed_classes: raise ValueError("Explorer History evidence class is invalid") if entry.get("temporal_coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer History evidence coverage is invalid") if not entry.get("display_label") or not entry.get("display_note"): raise ValueError("Explorer History evidence presentation is required") def _validate_rci_projection(rci: Any) -> None: """Validate the compact RCI facts used by the browser inspector. The projection deliberately excludes RCI condition samples and Unified points: retaining either would make the Explorer response duplicate the authoritative point-level analysis rather than reference it. """ if not isinstance(rci, Mapping): raise ValueError("Explorer v0.3 requires an RCI projection") if rci.get("contract_version") != "0.2": raise ValueError("Explorer RCI contract version is unsupported") if "condition_states" in rci or "points" in rci: raise ValueError("Explorer RCI projection must not duplicate point-level state") for key in ("catalog_version", "policy_version", "analysis_metadata", "state_summary"): if key not in rci: raise ValueError(f"Explorer RCI projection is missing {key}") for key in ("signals", "clusters", "events"): if not isinstance(rci.get(key), list): raise ValueError(f"Explorer RCI {key} must be serialized as a list") quality = rci.get("quality") or {} if quality.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer RCI quality coverage is invalid") for signal in rci["signals"]: if signal.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer RCI signal coverage is invalid") if not isinstance(signal.get("active_spans"), list): raise ValueError("Explorer RCI signal active spans are required") if not isinstance(signal.get("bridged_false_gaps"), list): raise ValueError("Explorer RCI bridged FALSE gaps are required") if signal.get("why") is None: raise ValueError("Explorer RCI signal WHY is required") for item in (*rci["clusters"], *rci["events"]): if item.get("why") is None: raise ValueError("Explorer RCI structured WHY is required") def _validate_photographer_projection(photographer: Any, scenario_name: Any) -> None: """Validate the compact Photographer facts consumed by Explorer v0.4. The full route point matrix stays authoritative in the Unified/Explorer route projection. Photographer contributes policy-labelled windows, sampled/bracketed terrain events, and weather ranges only. """ if not isinstance(photographer, Mapping): raise ValueError("Explorer v0.4 requires a Photographer projection") if photographer.get("contract_version") != "0.1": raise ValueError("Explorer Photographer contract version is unsupported") if "points" in photographer or "route_encounters" in photographer: raise ValueError("Explorer Photographer projection must not duplicate point-level state") timeline = photographer.get("route_timeline_reference") or {} if timeline.get("scenario_name") != scenario_name: raise ValueError("Explorer Photographer scenario must match its route timeline") policy = photographer.get("policy") or {} if not policy.get("policy_id") or not policy.get("version"): raise ValueError("Explorer Photographer policy identity is required") for convention_key, convention_name in ( ("golden_hour", "GOLDEN_HOUR"), ("blue_hour", "BLUE_HOUR"), ): result = photographer.get(convention_key) or {} if result.get("convention") != convention_name: raise ValueError("Explorer Photographer convention identity is invalid") if result.get("support_state") != "SUPPORTED": raise ValueError("Explorer Photographer convention must retain its support state") if result.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer Photographer convention coverage is invalid") if not isinstance(result.get("windows"), list): raise ValueError("Explorer Photographer convention windows are required") if result.get("threshold_provenance") is None: raise ValueError("Explorer Photographer threshold provenance is required") for window in result["windows"]: if window.get("convention") != convention_name: raise ValueError("Explorer Photographer window convention is invalid") _validate_photographer_boundary(window.get("start_boundary")) _validate_photographer_boundary(window.get("end_boundary")) terrain = photographer.get("terrain_light") or {} if terrain.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer Photographer terrain coverage is invalid") if not isinstance(terrain.get("events"), list): raise ValueError("Explorer Photographer terrain events are required") for event in terrain["events"]: _validate_photographer_boundary(event.get("boundary")) atmosphere = photographer.get("atmospheric_context") or {} if atmosphere.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer Photographer atmospheric coverage is invalid") if not isinstance(atmosphere.get("route_variables"), list): raise ValueError("Explorer Photographer route atmospheric variables are required") if not isinstance(atmosphere.get("scopes"), list): raise ValueError("Explorer Photographer atmospheric scopes are required") if "source_artifact" in (atmosphere.get("source") or {}): raise ValueError("Explorer Photographer must not expose local source artifacts") if "cache_identity" in (atmosphere.get("source") or {}): raise ValueError("Explorer Photographer must not expose local cache identities") for item in ( *atmosphere["route_variables"], *(variable for scope in atmosphere["scopes"] for variable in scope.get("variables", [])), ): if item.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer Photographer atmospheric variable coverage is invalid") if (item.get("minimum") is None) != (item.get("maximum") is None): raise ValueError("Explorer Photographer atmospheric range endpoints must agree") def _validate_snow_projection(snow: Any, scenario_name: Any, total_points: int) -> None: """Validate descriptive Snow Core facts without admitting interpretations.""" if not isinstance(snow, Mapping): raise ValueError("Explorer v0.5 requires a Snow projection") if snow.get("contract_version") != "0.1": raise ValueError("Explorer Snow contract version is unsupported") if "points" in snow or "signals" in snow: raise ValueError("Explorer Snow projection must not expose a point matrix or signals") timeline = snow.get("route_timeline_reference") or {} if timeline.get("scenario_name") != scenario_name: raise ValueError("Explorer Snow scenario must match its route timeline") quality = snow.get("quality") or {} if quality.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer Snow quality coverage is invalid") facts = snow.get("snow_facts") if not isinstance(facts, list) or len(facts) != total_points * 2: raise ValueError("Explorer Snow requires two independent facts per route point") fact_types = {"MODELLED_SNOWFALL", "MODELLED_GROUND_SNOW_DEPTH"} for fact in facts: if fact.get("semantics", {}).get("fact_type") not in fact_types: raise ValueError("Explorer Snow fact type is invalid") if fact.get("coverage") not in RCI_COVERAGE_STATES: raise ValueError("Explorer Snow fact coverage is invalid") if fact.get("semantics", {}).get("fact_type") == "MODELLED_SNOWFALL": if fact.get("semantics", {}).get("canonical_unit") != "cm": raise ValueError("Explorer snowfall unit is invalid") else: if fact.get("semantics", {}).get("canonical_unit") != "m": raise ValueError("Explorer snow-depth unit is invalid") if fact.get("projection_policy_id") != "modelled_snow_depth_nearest_context_v0_1": raise ValueError("Explorer snow-depth projection policy is invalid") def _validate_photographer_boundary(boundary: Any) -> None: if not isinstance(boundary, Mapping): raise ValueError("Explorer Photographer boundary is required") semantics = boundary.get("semantics") expected_samples = 2 if semantics in {"TRANSITION_BRACKET", "DATA_GAP_BRACKET"} else 1 if ( semantics not in { "ANALYSIS_WINDOW_START", "ANALYSIS_WINDOW_END", "TRANSITION_BRACKET", "DATA_GAP_BRACKET", "SAMPLED_POINT", } or len(boundary.get("samples", [])) != expected_samples ): raise ValueError("Explorer Photographer boundary precision is invalid") def write_explorer_document(document: Mapping[str, Any], destination) -> None: destination.parent.mkdir(parents=True, exist_ok=True) destination.write_text(serialize_explorer_document(document), encoding="utf-8")