from __future__ import annotations from collections import Counter from datetime import datetime, timezone import hashlib import json import re from pathlib import Path from typing import Any import mapbox_vector_tile from .production import ProductionLedger, load_plan, sha256_file from .style_contract import audit_pbf_style_and_sprite RELEASE_SCHEMA = "navsea.cm95.release-manifest/v1" def _builder_contract_at_least(value: Any, target: tuple[int, int, int]) -> bool: match = re.match(r"^cm95pbf-builder/(\d+)\.(\d+)\.(\d+)", str(value or "")) if not match: return False return tuple(int(part) for part in match.groups()) >= target def _utc_now(): return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") def _external_audit_contract( path: Path | None, *, required: bool, expected_schema: str, expected_plan_id: str | None = None, expected_pbf_dir: Path | None = None, ) -> dict[str, Any]: result: dict[str, Any] = { "required": required, "status": "NOT_AUDITED", "path": None, "sha256": None, "schema": None, "audit_status": None, "failures": [], } if path is None: if required: result["status"] = "FAIL" result["failures"].append("required_audit_missing") return result path = Path(path) result["path"] = str(path) if not path.is_file(): result["status"] = "FAIL" result["failures"].append("audit_file_missing") return result result["sha256"] = sha256_file(path) try: obj = json.loads(path.read_text(encoding="utf-8")) except Exception as exc: result["status"] = "FAIL" result["failures"].append( f"audit_parse_error:{type(exc).__name__}" ) return result result["schema"] = obj.get("schema") result["audit_status"] = obj.get("status") if obj.get("schema") != expected_schema: result["failures"].append("audit_schema_mismatch") if obj.get("status") != "PASS": result["failures"].append("audit_not_pass") if expected_plan_id is not None and obj.get("plan_id") != expected_plan_id: result["failures"].append("audit_plan_id_mismatch") if expected_pbf_dir is not None: raw_pbf_dir = obj.get("pbf_dir") if not raw_pbf_dir: result["failures"].append("audit_pbf_dir_missing") else: try: actual = Path(raw_pbf_dir).resolve() expected = Path(expected_pbf_dir).resolve() except Exception: result["failures"].append("audit_pbf_dir_invalid") else: if actual != expected: result["failures"].append("audit_pbf_dir_mismatch") for key in ( "expected_multi_light_groups", "actual_multi_light_groups", "expected_presentation_records", "actual_presentation_records", "signature_mismatch_count", "occurrence_drift_count", "occurrence_scope_mismatch_count", "required_runtime_kinds", "missing_runtime_kinds", "noncompliant_runtime_kinds", ): if key in obj: result[key] = obj.get(key) result["status"] = "PASS" if not result["failures"] else "FAIL" return result def _collect_nav_name_conflict_expectations( report_root: Path, conflict_reports: list[tuple[Path, int]], ) -> dict[str, Any]: """Collect ambiguous navigation-name outputs that must remain unlabeled. This is intentionally fail-closed: - a report-side conflict must have exactly matching trace rows; - each trace row must explicitly say nav_name_source=conflict; - there must be at least two distinct direct-name candidates; - the corresponding PBF feature is validated later and must omit chart_label_text rather than guessing one candidate. """ expected_refs: dict[tuple[str, str, int], dict[str, Any]] = {} unique_outputs: dict[tuple[str, int], dict[str, Any]] = {} issues: list[dict[str, Any]] = [] trace_occurrences = 0 for report_path, report_count in conflict_reports: rel_report = report_path.relative_to(report_root) trace_path = report_path.with_name(report_path.stem + ".trace.jsonl") pbf_rel = rel_report.with_suffix(".pbf").as_posix() if not trace_path.exists(): issues.append({ "type": "missing_trace_file", "report": rel_report.as_posix(), "expected_conflicts": int(report_count), }) continue tile_rows: list[dict[str, Any]] = [] with trace_path.open("r", encoding="utf-8") as handle: for lineno, raw in enumerate(handle, start=1): raw = raw.strip() if not raw: continue try: row = json.loads(raw) except json.JSONDecodeError as exc: issues.append({ "type": "invalid_trace_json", "trace": str(trace_path.relative_to(report_root)), "line": lineno, "error": str(exc), }) continue if isinstance(row, dict) and row.get("nav_name_conflict") is True: tile_rows.append(row) trace_occurrences += len(tile_rows) if len(tile_rows) != int(report_count): issues.append({ "type": "report_trace_conflict_count_mismatch", "report": rel_report.as_posix(), "report_count": int(report_count), "trace_count": len(tile_rows), }) for row in tile_rows: output_id = row.get("output_id") layer = str(row.get("output_layer") or "") name_source = str(row.get("nav_name_source") or "") raw_candidates = row.get("nav_name_candidates") or [] candidates: list[str] = [] for value in raw_candidates: text = str(value).strip() if text and text not in candidates: candidates.append(text) if output_id is None or not layer: issues.append({ "type": "invalid_conflict_trace_identity", "report": rel_report.as_posix(), "row": row, }) continue try: fid = int(output_id) except (TypeError, ValueError): issues.append({ "type": "invalid_conflict_output_id", "report": rel_report.as_posix(), "output_id": output_id, }) continue if name_source != "conflict": issues.append({ "type": "conflict_trace_wrong_name_source", "report": rel_report.as_posix(), "output_id": fid, "nav_name_source": name_source, }) if len(candidates) < 2: issues.append({ "type": "conflict_trace_insufficient_candidates", "report": rel_report.as_posix(), "output_id": fid, "candidates": candidates, }) ref_key = (pbf_rel, layer, fid) if ref_key in expected_refs: issues.append({ "type": "duplicate_conflict_trace_reference", "pbf": pbf_rel, "layer": layer, "output_id": fid, }) expected_refs[ref_key] = { "pbf": pbf_rel, "layer": layer, "output_id": fid, "source_id": row.get("source_id"), "relation20": row.get("relation20"), "native_scale": row.get("native_scale"), "candidates": candidates, } unique_key = (layer, fid) unique = unique_outputs.setdefault(unique_key, { "output_id": fid, "output_layer": layer, "source_id": row.get("source_id"), "relation20": row.get("relation20"), "native_scale": row.get("native_scale"), "nav_name_candidates": list(candidates), "occurrences": 0, "tiles": [], }) unique["occurrences"] += 1 if pbf_rel not in unique["tiles"]: unique["tiles"].append(pbf_rel) for candidate in candidates: if candidate not in unique["nav_name_candidates"]: unique["nav_name_candidates"].append(candidate) return { "report_occurrences": sum(int(count) for _path, count in conflict_reports), "trace_occurrences": trace_occurrences, "expected_refs": expected_refs, "unique_outputs": unique_outputs, "issues": issues, "seen_refs": set(), "labeled_conflicts": [], } def _observe_nav_name_conflict_feature( state: dict[str, Any], pbf_rel: str, layer: str, fid: Any, properties: dict[str, Any], ) -> None: if fid is None: return try: key = (pbf_rel, str(layer), int(fid)) except (TypeError, ValueError): return expected = state["expected_refs"].get(key) if expected is None: return state["seen_refs"].add(key) label = properties.get("chart_label_text") if label is not None and str(label).strip(): state["labeled_conflicts"].append({ **expected, "chart_label_text": str(label), }) def _finalize_nav_name_conflict_audit(state: dict[str, Any]) -> dict[str, Any]: expected_keys = set(state["expected_refs"]) seen_keys = set(state["seen_refs"]) missing = sorted(expected_keys - seen_keys) issues = list(state["issues"]) if state["report_occurrences"] != state["trace_occurrences"]: issues.append({ "type": "global_report_trace_conflict_count_mismatch", "report_occurrences": state["report_occurrences"], "trace_occurrences": state["trace_occurrences"], }) for pbf_rel, layer, fid in missing: issues.append({ "type": "conflict_output_missing_from_pbf", "pbf": pbf_rel, "layer": layer, "output_id": fid, }) for item in state["labeled_conflicts"]: issues.append({ "type": "ambiguous_conflict_was_labeled", **item, }) unique_details = [] for row in state["unique_outputs"].values(): item = dict(row) item["tiles"] = sorted(item.get("tiles") or []) unique_details.append(item) unique_details.sort( key=lambda row: ( int(row.get("native_scale") or 0), str(row.get("source_id") or ""), int(row.get("output_id") or 0), ) ) affected_tiles = sorted({ key[0] for key in state["expected_refs"] }) safe_omissions = len(seen_keys) - len(state["labeled_conflicts"]) return { "overall": "PASS" if not issues else "FAIL", "policy": "AMBIGUOUS_DIRECT_NAMES_OMIT_LABEL_FAIL_CLOSED", "report_occurrences": int(state["report_occurrences"]), "trace_occurrences": int(state["trace_occurrences"]), "safe_omission_occurrences": int(safe_omissions), "unique_conflict_outputs": len(unique_details), "affected_tiles": len(affected_tiles), "labeled_conflict_occurrences": len(state["labeled_conflicts"]), "missing_pbf_occurrences": len(missing), "issues": issues, "unique_conflicts": unique_details, } def finalize_release( plan_path: Path, release_root: Path, style_path: Path | None = None, sprite_json_path: Path | None = None, gap_disposition_path: Path | None = None, presentation_conservation_audit_path: Path | None = None, runtime_capability_audit_path: Path | None = None, ) -> dict[str, Any]: plan = load_plan(plan_path) release_dir = Path(release_root) / str(plan["release_id"]) ledger = ProductionLedger(release_dir / "production_run_v1.sqlite") status = ledger.summary(str(plan["profile"])) if status["states"].get("FAILED", 0) or status["states"].get("PENDING", 0) or status["states"].get("RUNNING", 0): raise RuntimeError(f"release has unfinished jobs: {status['states']}") if status["states"].get("PASS", 0) != int(plan["job_count"]): raise RuntimeError(f"PASS job count mismatch: {status['states']} plan={plan['job_count']}") pbf_root = release_dir / "pbf" / str(plan["profile"]) report_root = release_dir / "reports" / str(plan["profile"]) pbf_files = sorted(pbf_root.rglob("*.pbf")) if pbf_root.exists() else [] report_files = sorted( p for p in report_root.rglob("*.json") if not p.name.endswith(".trace.json") ) if report_root.exists() else [] hard_gate = [] encoder_dropped = [] name_conflicts = 0 nav_name_conflict_reports: list[tuple[Path, int]] = [] non_renderable_omissions = 0 non_renderable_omission_layers = Counter() quantization_fallbacks = 0 quantization_omissions = 0 quantization_omission_layers = Counter() encoder_reconciled_candidates = 0 encoder_reconciled_tiles = 0 report_multi_light_groups = 0 report_light_presentation_records = 0 planned = 0 output = 0 report_layer_counts = Counter() for path in report_files: obj = json.loads(path.read_text(encoding="utf-8")) hard_gate += [{"report": str(path.relative_to(release_dir)), **x} for x in obj.get("hard_gate_unmapped") or []] encoder_dropped += [{"report": str(path.relative_to(release_dir)), **x} for x in obj.get("encoder_dropped_features") or []] nav_summary = obj.get("navigation_composites") or {} tile_name_conflicts = int(nav_summary.get("name_conflict_groups") or 0) report_multi_light_groups += int(nav_summary.get("multi_light_groups") or 0) report_light_presentation_records += int(nav_summary.get("light_presentation_records") or 0) name_conflicts += tile_name_conflicts if tile_name_conflicts: nav_name_conflict_reports.append((path, tile_name_conflicts)) label_omissions = obj.get("non_renderable_omissions") or [] non_renderable_omissions += len(label_omissions) for item in label_omissions: if isinstance(item, dict): non_renderable_omission_layers[str(item.get("layer") or "UNKNOWN")] += 1 quantization_fallbacks += len(obj.get("quantization_fallbacks") or []) omissions = obj.get("quantization_omissions") or [] quantization_omissions += len(omissions) reconcile_candidates = obj.get("encoder_drop_candidates") or [] if reconcile_candidates: encoder_reconciled_tiles += 1 for candidate in reconcile_candidates: if isinstance(candidate, dict): encoder_reconciled_candidates += len(candidate.get("dropped") or []) for item in omissions: if isinstance(item, dict): quantization_omission_layers[str(item.get("layer") or "UNKNOWN")] += 1 planned += int(obj.get("planned_output_feature_count") or 0) output += int(obj.get("output_feature_count") or 0) report_layer_counts.update(obj.get("output_layer_counts") or {}) nav_name_conflict_state = _collect_nav_name_conflict_expectations( report_root, nav_name_conflict_reports, ) inventory = [] decoded_features = 0 decoded_layers = Counter() duplicate_feature_ids = [] s52_plan = dict(plan.get("s52") or {}) s52_runtime_enabled = bool(s52_plan.get("enabled")) and s52_plan.get("mariner_mode") == "runtime" s52_runtime_failures: list[str] = [] s52_runtime_csp_counts = Counter() s52_runtime_layer_counts = Counter() s52_runtime_dependent_count = 0 s52_runtime_leaks = Counter() s52_runtime_fact_counts = Counter() s52_runtime_unknown_fact_exceptions = Counter() light_presentation_required = str(plan.get("builder_contract_version") or "").startswith( "cm95pbf-builder/0.4.11-s52-multi-light-presentations" ) pbf_multi_light_groups = 0 pbf_light_presentation_records = 0 light_presentation_failures: list[str] = [] s52_dynamic_final_fields = { "s52_is_safety_contour", "s52_sounding_prefix", "s52_sounding_is_shallow", "s52_isolated_danger", "s52_display_base_promoted", } for path in pbf_files: raw = path.read_bytes() decoded = mapbox_vector_tile.decode(raw) seen = set() tile_features = 0 for layer, payload in decoded.items(): feats = payload.get("features", []) decoded_layers[layer] += len(feats) tile_features += len(feats) pbf_rel = str(path.relative_to(pbf_root)).replace("\\", "/") for feat in feats: fid = feat.get("id") props = feat.get("properties") or {} raw_light_presentations = props.get("s52_light_presentations") if raw_light_presentations not in (None, ""): try: light_presentations = ( json.loads(raw_light_presentations) if isinstance(raw_light_presentations, str) else raw_light_presentations ) except Exception as exc: light_presentation_failures.append( f"{pbf_rel}:{layer}:{fid} invalid s52_light_presentations JSON: {type(exc).__name__}" ) light_presentations = [] if not isinstance(light_presentations, list) or len(light_presentations) < 2: light_presentation_failures.append( f"{pbf_rel}:{layer}:{fid} multi-light presentation array must contain at least 2 items" ) else: pbf_multi_light_groups += 1 pbf_light_presentation_records += len(light_presentations) for index, presentation in enumerate(light_presentations): if not isinstance(presentation, dict): light_presentation_failures.append( f"{pbf_rel}:{layer}:{fid} light presentation {index} is not an object" ) continue if presentation.get("s52_csp_resolved") != "LIGHTS05": light_presentation_failures.append( f"{pbf_rel}:{layer}:{fid} light presentation {index} is not resolved LIGHTS05" ) if "source_record_index" not in presentation: light_presentation_failures.append( f"{pbf_rel}:{layer}:{fid} light presentation {index} missing source_record_index" ) if s52_runtime_enabled: if props.get("s52_mariner_dependent") == 1: s52_runtime_dependent_count += 1 s52_runtime_layer_counts[layer] += 1 csp = str(props.get("s52_runtime_csp") or "") if not csp: s52_runtime_failures.append(f"{pbf_rel}:{layer}:{fid} missing s52_runtime_csp") else: s52_runtime_csp_counts[csp] += 1 for key in s52_dynamic_final_fields: if key in props: s52_runtime_leaks[key] += 1 if csp in {"DEPARE01", "DEPARE02"}: if "depth_min_m" not in props: if props.get("canonical_object_type") == "dredged_area": s52_runtime_unknown_fact_exceptions[ "DEPARE:dredged_area_missing_depth_min" ] += 1 else: s52_runtime_failures.append( f"{pbf_rel}:{layer}:{fid} {csp} missing depth_min_m" ) if "s52_area_color" in props: s52_runtime_leaks["DEPARE:s52_area_color"] += 1 elif csp == "DEPCNT02" and "contour_depth_m" not in props: s52_runtime_failures.append(f"{pbf_rel}:{layer}:{fid} DEPCNT02 missing contour_depth_m") elif csp == "SOUNDG02" and "s52_sounding_depth_m" not in props: s52_runtime_failures.append(f"{pbf_rel}:{layer}:{fid} SOUNDG02 missing s52_sounding_depth_m") elif csp in {"OBSTRN04", "WRECKS02"} and "s52_hazard_depth_m" not in props: if csp == "OBSTRN04" and props.get("s52_expsou_code") == 1: s52_runtime_unknown_fact_exceptions[ "OBSTRN04:EXPSOU1_unknown_hazard_depth" ] += 1 else: s52_runtime_failures.append( f"{pbf_rel}:{layer}:{fid} {csp} missing s52_hazard_depth_m" ) elif "s52_runtime_csp" in props: s52_runtime_failures.append(f"{pbf_rel}:{layer}:{fid} runtime CSP without mariner-dependent marker") for key in ( "depth_min_m", "depth_max_m", "contour_depth_m", "s52_sounding_depth_m", "s52_hazard_depth_m", "s52_surrounding_max_drval1_m", "s52_surrounding_min_drval2_m", ): if key in props: s52_runtime_fact_counts[key] += 1 _observe_nav_name_conflict_feature( nav_name_conflict_state, pbf_rel, layer, fid, props, ) if fid is None: continue key = (layer, int(fid)) if key in seen: duplicate_feature_ids.append({"file": str(path.relative_to(release_dir)), "layer": layer, "id": int(fid)}) seen.add(key) decoded_features += tile_features inventory.append({ "path": str(path.relative_to(release_dir)).replace("\\", "/"), "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest(), "features": tile_features, "layers": len(decoded), }) style_sha = None sprite_sha = None style_contract = None if style_path is not None: style_path = Path(style_path) style_sha = sha256_file(style_path) if sprite_json_path is not None: sprite_json_path = Path(sprite_json_path) sprite_sha = sha256_file(sprite_json_path) style_contract = audit_pbf_style_and_sprite( pbf_root, style_path, sprite_json_path, ) gap_disposition_sha = None semantic_backlog = None semantic_completeness = "NOT_AUDITED" if gap_disposition_path is not None: import yaml gap_disposition_path = Path(gap_disposition_path) if gap_disposition_path.exists(): gap_disposition_sha = sha256_file(gap_disposition_path) gap_obj = yaml.safe_load(gap_disposition_path.read_text(encoding="utf-8")) or {} semantic_backlog = dict(gap_obj.get("summary") or {}) pending = int((semantic_backlog or {}).get("pending_semantic_review") or 0) semantic_completeness = "COMPLETE" if pending == 0 else "PENDING_REVIEW" nav_name_conflict_audit = _finalize_nav_name_conflict_audit( nav_name_conflict_state ) if s52_runtime_leaks: s52_runtime_failures.append(f"dynamic final state leaked into runtime PBF: {dict(s52_runtime_leaks)}") if light_presentation_required: if report_multi_light_groups != pbf_multi_light_groups: light_presentation_failures.append( "report/PBF multi-light group count mismatch: " f"report={report_multi_light_groups} pbf={pbf_multi_light_groups}" ) if report_light_presentation_records != pbf_light_presentation_records: light_presentation_failures.append( "report/PBF light presentation record count mismatch: " f"report={report_light_presentation_records} pbf={pbf_light_presentation_records}" ) s52_light_presentation_contract = { "status": ( "PASS" if not light_presentation_failures else "FAIL" ) if light_presentation_required else "NOT_APPLICABLE", "required": light_presentation_required, "report_multi_light_groups": report_multi_light_groups, "pbf_multi_light_groups": pbf_multi_light_groups, "report_light_presentation_records": report_light_presentation_records, "pbf_light_presentation_records": pbf_light_presentation_records, "failure_count": len(light_presentation_failures), "failure_examples": light_presentation_failures[:100], } presentation_conservation_required = _builder_contract_at_least( plan.get("builder_contract_version"), (0, 4, 11) ) presentation_conservation_contract = _external_audit_contract( presentation_conservation_audit_path, required=presentation_conservation_required, expected_schema="navsea.cm95.s52-presentation-conservation-audit/v1", expected_plan_id=str(plan.get("plan_id") or ""), expected_pbf_dir=pbf_root, ) runtime_capability_contract = _external_audit_contract( runtime_capability_audit_path, required=False, expected_schema="navsea.cm95.s52-runtime-capability-audit/v1", expected_plan_id=str(plan.get("plan_id") or ""), expected_pbf_dir=None, ) s52_runtime_contract = { "status": ("PASS" if not s52_runtime_failures else "FAIL") if s52_runtime_enabled else "NOT_APPLICABLE", "enabled": s52_runtime_enabled, "mariner_dependent_count": s52_runtime_dependent_count, "runtime_csp_counts": dict(sorted(s52_runtime_csp_counts.items())), "runtime_layer_counts": dict(sorted(s52_runtime_layer_counts.items())), "fact_counts": dict(sorted(s52_runtime_fact_counts.items())), "unknown_fact_exceptions": dict(sorted(s52_runtime_unknown_fact_exceptions.items())), "dynamic_final_leaks": dict(sorted(s52_runtime_leaks.items())), "failure_count": len(s52_runtime_failures), "failure_examples": s52_runtime_failures[:100], } tile_enumeration_mode = str(plan.get("tile_enumeration_mode") or "coverage") pbf_rel_paths = { path.relative_to(pbf_root).as_posix() for path in pbf_files } explicit_empty_tiles: list[str] = [] missing_nonempty_pbf_tiles: list[str] = [] report_tile_paths: set[str] = set() for path in report_files: obj = json.loads(path.read_text(encoding="utf-8")) rel_pbf = path.relative_to(report_root).with_suffix(".pbf").as_posix() report_tile_paths.add(rel_pbf) if rel_pbf in pbf_rel_paths: continue planned_count = int(obj.get("planned_output_feature_count") or 0) output_count = int(obj.get("output_feature_count") or 0) hard_gate_count = len(obj.get("hard_gate_unmapped") or []) encoder_drop_count = len(obj.get("encoder_dropped_features") or []) if ( planned_count == 0 and output_count == 0 and hard_gate_count == 0 and encoder_drop_count == 0 ): explicit_empty_tiles.append(rel_pbf) else: missing_nonempty_pbf_tiles.append(rel_pbf) unexpected_pbf_tiles = sorted(pbf_rel_paths - report_tile_paths) tile_footprint_contract = { "status": "NOT_APPLICABLE", "mode": tile_enumeration_mode, "planned_tile_jobs": int(plan["job_count"]), "pbf_file_count": len(pbf_files), "report_file_count": len(report_files), "explicit_empty_tile_count": len(explicit_empty_tiles), "materialized_or_empty_tile_count": len(pbf_files) + len(explicit_empty_tiles), "missing_nonempty_pbf_tiles": missing_nonempty_pbf_tiles, "unexpected_pbf_tiles": unexpected_pbf_tiles, } if tile_enumeration_mode == "aoi-complete": tile_footprint_contract["status"] = ( "PASS" if ( len(report_files) == int(plan["job_count"]) and len(pbf_files) + len(explicit_empty_tiles) == int(plan["job_count"]) and not missing_nonempty_pbf_tiles and not unexpected_pbf_tiles ) else "FAIL" ) s52_resource_contract = { "status": "NOT_APPLICABLE", "path": plan.get("s52_resource_contract_path"), "sha256": None, "expected_sha256": (plan.get("s52") or {}).get("resource_contract_sha256"), "defined_resource_counts": None, "active_resource_counts": None, "failures": [], } expected_resource_sha = str( (plan.get("s52") or {}).get("resource_contract_sha256") or "" ) if expected_resource_sha: resource_path_raw = plan.get("s52_resource_contract_path") resource_failures: list[str] = [] resource_obj: dict[str, Any] = {} if not resource_path_raw: resource_failures.append("missing_resource_contract_path") else: resource_path = Path(resource_path_raw) if not resource_path.exists(): resource_failures.append("resource_contract_file_missing") else: actual_sha = sha256_file(resource_path) s52_resource_contract["sha256"] = actual_sha if actual_sha != expected_resource_sha: resource_failures.append("resource_contract_sha256_mismatch") try: resource_obj = json.loads( resource_path.read_text(encoding="utf-8") ) except Exception as exc: resource_failures.append( f"resource_contract_parse_error:{type(exc).__name__}" ) else: if resource_obj.get("status") != "PASS": resource_failures.append("resource_contract_not_pass") s52_resource_contract["defined_resource_counts"] = ( resource_obj.get("defined_resource_counts") ) s52_resource_contract["active_resource_counts"] = ( resource_obj.get("active_resource_counts") ) s52_resource_contract["failures"] = resource_failures s52_resource_contract["status"] = ( "PASS" if not resource_failures else "FAIL" ) failures = [] if expected_resource_sha and s52_resource_contract["status"] != "PASS": failures.append("s52_resource_contract") if tile_enumeration_mode == "aoi-complete": if len(report_files) != int(plan["job_count"]): failures.append("aoi_complete_report_count_mismatch") if tile_footprint_contract["status"] != "PASS": failures.append("aoi_complete_tile_footprint_mismatch") if hard_gate: failures.append("hard_gate_unmapped") if encoder_dropped: failures.append("encoder_dropped_features") if name_conflicts and nav_name_conflict_audit.get("overall") != "PASS": failures.append("nav_name_conflict_audit") if planned != output: failures.append("planned_vs_report_output_mismatch") if output != decoded_features: failures.append("report_output_vs_decoded_pbf_mismatch") if duplicate_feature_ids: failures.append("duplicate_feature_ids_within_tile_layer") if style_contract is not None and style_contract.get("overall") != "PASS": failures.append("style_or_sprite_contract") if s52_runtime_enabled and s52_runtime_contract.get("status") != "PASS": failures.append("s52_runtime_contract") if light_presentation_required and s52_light_presentation_contract.get("status") != "PASS": failures.append("s52_light_presentation_contract") if presentation_conservation_required and presentation_conservation_contract.get("status") != "PASS": failures.append("s52_presentation_conservation_contract") if runtime_capability_audit_path is not None and runtime_capability_contract.get("status") != "PASS": failures.append("s52_runtime_capability_contract") result = { "schema": RELEASE_SCHEMA, "created_at": _utc_now(), "status": "PASS" if not failures else "FAIL", "release_id": plan["release_id"], "plan_id": plan["plan_id"], "dataset_id": plan["dataset_id"], "dataset_fingerprint": plan["dataset_fingerprint"], "profile": plan["profile"], "zooms": plan["zooms"], "mapping_sha256": plan["mapping_sha256"], "policy_sha256": plan["policy_sha256"], "builder_contract_version": plan.get("builder_contract_version"), "production_pipeline_version": plan.get("production_pipeline_version"), "tile_enumeration_mode": tile_enumeration_mode, "tile_footprint_contract": tile_footprint_contract, "s52": plan.get("s52"), "aoi": plan.get("aoi"), "s52_resource_contract": s52_resource_contract, "s52_runtime_contract": s52_runtime_contract, "s52_light_presentation_contract": s52_light_presentation_contract, "s52_presentation_conservation_contract": presentation_conservation_contract, "s52_runtime_capability_contract": runtime_capability_contract, "style_sha256": style_sha, "sprite_json_sha256": sprite_sha, "gap_disposition_sha256": gap_disposition_sha, "semantic_completeness": semantic_completeness, "semantic_backlog": semantic_backlog, "style_sprite_contract": style_contract, "planned_tile_jobs": int(plan["job_count"]), "pbf_file_count": len(pbf_files), "report_file_count": len(report_files), "planned_output_features": planned, "report_output_features": output, "decoded_pbf_features": decoded_features, "non_renderable_omissions": non_renderable_omissions, "non_renderable_omission_layers": dict(sorted(non_renderable_omission_layers.items())), "quantization_fallbacks": quantization_fallbacks, "quantization_omissions": quantization_omissions, "quantization_omission_layers": dict(sorted(quantization_omission_layers.items())), "encoder_reconciled_tiles": encoder_reconciled_tiles, "encoder_reconciled_drop_candidates": encoder_reconciled_candidates, "hard_gate_unmapped": hard_gate, "encoder_dropped_features": encoder_dropped, "nav_name_conflicts": name_conflicts, "nav_name_conflict_omissions": nav_name_conflict_audit.get("safe_omission_occurrences", 0), "nav_name_conflict_unique_outputs": nav_name_conflict_audit.get("unique_conflict_outputs", 0), "nav_name_conflict_audit": nav_name_conflict_audit, "duplicate_feature_ids": duplicate_feature_ids, "output_layer_counts": dict(sorted(decoded_layers.items())), "failures": failures, "inventory": inventory, } out = release_dir / "release_manifest_v1.json" out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") return result