feat: add S-52 production pipeline v2

docs/CM95_S52_ProductionPipeline_v2.md

@@ -0,0 +1,110 @@
+# CM95 S-52 Production Pipeline v2
+
+## 1. 目标
+
+Production Pipeline v2 将已经通过单 Tile 验证的 S-52 runtime PBF 正式接入现有 CM95 生产编排层，并增加可审计 AOI 计划能力。
+
+核心要求是：同一个 release identity 不允许混入不同 S-52 Lookup、不同 mariner mode 或不同 AOI 的 Tile。
+
+## 2. Production Plan v2
+
+新计划 schema：
+
+```text
+navsea.cm95.production-plan/v2
+```
+
+Pipeline 版本：
+
+```text
+cm95pbf-production/2.0-s52-aoi
+```
+
+Plan immutable hash 现在除 Dataset、mapping、policy、FID manifest、Builder Contract、jobs 外，还冻结：
+
+- S-52 是否启用；
+- S-52 Lookup SHA256；
+- Point Table；
+- Boundary Table；
+- `runtime/reference` mariner mode；
+- AOI name；
+- AOI WGS84 bbox。
+
+S-52 Lookup 路径本身不参与 identity，内容 SHA256 才参与 identity；同内容换路径不会制造新 release。
+
+## 3. AOI 枚举
+
+`plan-production` 新增：
+
+```text
+--aoi-name
+--aoi-bbox west,south,east,north
+```
+
+AOI 在 Web Mercator 中与 Dataset coverage 几何相交并裁剪后再枚举 XYZ job。这样不是简单按 Tile 中心筛选，边界 coverage 也不会被漏掉。
+
+Plan `source_stats` 同时记录 AOI 裁剪前后的 coverage piece 数量。
+
+## 4. S-52 生产参数
+
+`plan-production` 新增：
+
+```text
+--s52-lookups
+--s52-point-table Simplified|Paper
+--s52-boundary-table Plain|Symbolized
+--s52-mariner-mode runtime|reference
+```
+
+`build-production` 不再从命令行临时接收这些值，而是只能读取冻结 plan，随后传给 `Cm95TileBuilder`。
+
+Build 启动前重新校验 Lookup 文件存在且 SHA256 与 plan 一致；文件发生变化时 fail-closed。
+
+## 5. Runtime release gate
+
+`finalize-production` 在 S-52 runtime plan 下增加 release 级 runtime contract gate。
+
+它会解码所有输出 PBF，并检查：
+
+- `s52_mariner_dependent` 与 `s52_runtime_csp` 一致；
+- DEPARE 存在 `depth_min_m`，且没有静态 `s52_area_color` 泄漏；
+- DEPCNT 存在 `contour_depth_m`；
+- SOUNDG 存在 `s52_sounding_depth_m`；
+- OBSTRN/WRECKS 存在 `s52_hazard_depth_m`；
+- 不允许 `s52_is_safety_contour`、`s52_isolated_danger`、`s52_display_base_promoted` 等动态最终状态泄漏。
+
+失败会把整个 release 标记为 FAIL。
+
+## 6. Style contract 过渡
+
+当前旧 NavSea semantic style/sprite contract 不是 S-52 runtime Adapter 的最终 contract。
+
+因此 CLI 新增：
+
+```text
+--skip-style-contract
+```
+
+它只跳过旧 style/sprite 审计，不跳过 PBF 计数、FID、hard gate、encoder、名称冲突和 S-52 runtime contract。待 NavSea `S52RuntimePresentationAdapter` 完成后，再接入正式 S-52 style/runtime contract。
+
+## 7. 兼容性
+
+Loader 继续接受历史 `production-plan/v1`，但新建 plan 一律生成 v2。
+
+旧 release 若没有与新 plan 完全一致的 S-52 production contract，不允许直接通过 `migrate-production-release` 复用 Tile。
+
+## 8. 九州首次运行
+
+本次 AOI：
+
+```text
+name = kyushu-main-nearby-v1
+bbox = 128.0,29.5,132.5,34.8
+zooms = 0-12
+profile = delivery
+point_table = Paper
+boundary_table = Plain
+mariner_mode = runtime
+```
+
+该范围用于本轮工程验证，覆盖九州本岛、对马、五岛、种子岛/屋久岛及近海，不作为日本行政区边界定义。

docs/报告/CM95_S52_ProductionPipelineV2与九州运行报告_v1.md

@@ -0,0 +1,57 @@
+# CM95 S-52 Production Pipeline v2 与九州运行报告 v1
+
+日期：2026-09-18
+
+## Pipeline 升级
+
+已完成 Production Plan v2、S-52 immutable contract、AOI coverage 裁剪、build 参数接线、Lookup SHA fail-closed、release runtime contract gate，以及旧 style contract 的显式过渡开关。
+
+## 福冈湾端到端 Smoke
+
+AOI：`130.3418,33.5780,130.4297,33.6512`，z12。
+
+Plan 枚举 4 个 Tile，`build-production --workers 4` 结果：
+
+- PASS：4/4
+- PBF：4
+- output features：3235
+- 主 Tile `12/3531/1641`：3164 features / 22 layers
+
+`finalize-production --skip-style-contract`：PASS。
+
+S-52 runtime gate：
+
+- `mariner_dependent_count = 2261`
+- `DEPARE01 = 173`
+- `DEPCNT02 = 188`
+- `OBSTRN04 = 19`
+- `SOUNDG02 = 1881`
+- `dynamic_final_leaks = {}`
+- `failure_count = 0`
+
+planned/report/decoded feature count 均为 3235。
+
+## 九州 Production Plan
+
+工程 AOI：`128.0,29.5,132.5,34.8`，名称 `kyushu-main-nearby-v1`。
+
+Plan：
+
+- schema：`navsea.cm95.production-plan/v2`
+- pipeline：`cm95pbf-production/2.0-s52-aoi`
+- release：`cm95-d1-62a85bd40d3a-bbaaeafee1b4`
+- jobs：1109
+- S-52 mariner mode：runtime
+- Lookup SHA256：`f63cbc0d3c9317b1873c0b0c470a5e7371d1b6d8d58393da660f70b837a52eb7`
+
+各级 job 数：z5=2、z6=4、z7=12、z8=24、z9=70、z10=197、z11=503、z12=297。z0-z4 在该 AOI 内没有对应 1:20M coverage，因此不制造空 job。
+
+## 当前运行
+
+九州 `build-production` 已使用 4 workers、最多 3 attempts 启动。
+
+日志：`build/logs/cm95_s52_kyushu_pipeline_v2.log`
+
+Release root：`build/releases-s52-v2/cm95-d1-62a85bd40d3a-bbaaeafee1b4`
+
+初始 ledger：1109 jobs，4 RUNNING，1105 PENDING。后续以 `status-production` 和最终 `finalize-production` 为权威。

src/cm95pbf/cli.py

@@ -28,6 +28,18 @@ def _add_common_db(p):
     p.add_argument("--dataset-id", type=int, default=1)
 
 
+def _parse_aoi_bbox(value: str | None):
+    if value is None or not str(value).strip():
+        return None
+    parts = [part.strip() for part in str(value).split(",")]
+    if len(parts) != 4:
+        raise argparse.ArgumentTypeError("AOI bbox 必须是 west,south,east,north")
+    try:
+        return tuple(float(part) for part in parts)
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("AOI bbox 必须是四个数字") from exc
+
+
 def build_parser() -> argparse.ArgumentParser:
     p = argparse.ArgumentParser(
         prog="cm95pbf",
@@ -121,7 +133,13 @@ def build_parser() -> argparse.ArgumentParser:
     prod_plan.add_argument("--mapping", default=str(DEFAULT_MAPPING))
     prod_plan.add_argument("--policy", default=str(DEFAULT_POLICY))
     prod_plan.add_argument("--manifest", default=str(DEFAULT_MANIFEST))
-    prod_plan.add_argument("--output", default="build/production/cm95_production_plan_v1.json")
+    prod_plan.add_argument("--output", default="build/production/cm95_production_plan_v2.json")
+    prod_plan.add_argument("--s52-lookups", default=None, help="S-52 lookup_rule_ir.json；提供后冻结 SHA256 并进入 production builder")
+    prod_plan.add_argument("--s52-point-table", choices=["Simplified", "Paper"], default="Paper")
+    prod_plan.add_argument("--s52-boundary-table", choices=["Plain", "Symbolized"], default="Plain")
+    prod_plan.add_argument("--s52-mariner-mode", choices=["runtime", "reference"], default="runtime")
+    prod_plan.add_argument("--aoi-name", default=None, help="可审计 AOI 名称")
+    prod_plan.add_argument("--aoi-bbox", type=_parse_aoi_bbox, default=None, metavar="W,S,E,N", help="只枚举与 WGS84 bbox 相交的 coverage/tile")
 
     prod_build = sub.add_parser(
         "build-production",
@@ -162,6 +180,7 @@ def build_parser() -> argparse.ArgumentParser:
     prod_final.add_argument("--style", default="config/style/style.navsea-delivery-full-semantic-unified-20260808-cm95-v1.json")
     prod_final.add_argument("--sprite-json", default="config/sprite/semantic-sprite-20260808.json")
     prod_final.add_argument("--gap-disposition", default="config/mappings/cm95_gap_disposition_v1.yaml")
+    prod_final.add_argument("--skip-style-contract", action="store_true", help="跳过旧 NavSea style/sprite contract；S-52 runtime release 仍执行 runtime contract gate")
 
     return p
 
@@ -206,11 +225,20 @@ def main() -> None:
             output_path=Path(args.output),
             zooms=parse_zooms(args.zooms),
             profile=args.profile,
+            s52_lookups_path=(Path(args.s52_lookups) if args.s52_lookups else None),
+            s52_point_table=args.s52_point_table,
+            s52_boundary_table=args.s52_boundary_table,
+            s52_mariner_mode=args.s52_mariner_mode,
+            aoi_name=args.aoi_name,
+            aoi_bbox_wgs84=args.aoi_bbox,
         )
         print(json.dumps({
             "plan_id": result["plan_id"],
             "release_id": result["release_id"],
             "job_count": result["job_count"],
+            "production_pipeline_version": result.get("production_pipeline_version"),
+            "s52": result.get("s52"),
+            "aoi": result.get("aoi"),
             "source_stats": result["source_stats"],
             "output": str(Path(args.output)),
         }, ensure_ascii=False, indent=2))
@@ -249,8 +277,8 @@ def main() -> None:
         result = finalize_release(
             Path(args.plan),
             Path(args.release_root),
-            Path(args.style) if args.style else None,
-            Path(args.sprite_json) if args.sprite_json else None,
+            None if args.skip_style_contract else (Path(args.style) if args.style else None),
+            None if args.skip_style_contract else (Path(args.sprite_json) if args.sprite_json else None),
             Path(args.gap_disposition) if args.gap_disposition else None,
         )
         print(json.dumps({
@@ -269,6 +297,7 @@ def main() -> None:
                 "contract_unmatched_count": (result.get("style_sprite_contract") or {}).get("contract_unmatched_count"),
                 "sprite_missing_count": (result.get("style_sprite_contract") or {}).get("sprite_missing_count"),
             },
+            "s52_runtime_contract": result.get("s52_runtime_contract"),
             "quantization_fallbacks": result.get("quantization_fallbacks"),
             "quantization_omissions": result.get("quantization_omissions"),
             "nav_name_conflict_audit": {

src/cm95pbf/production.py

@@ -32,7 +32,10 @@ from .policy import (
 )
 
 
-PRODUCTION_PLAN_SCHEMA = "navsea.cm95.production-plan/v1"
+PRODUCTION_PLAN_SCHEMA_V1 = "navsea.cm95.production-plan/v1"
+PRODUCTION_PLAN_SCHEMA = "navsea.cm95.production-plan/v2"
+SUPPORTED_PRODUCTION_PLAN_SCHEMAS = {PRODUCTION_PLAN_SCHEMA_V1, PRODUCTION_PLAN_SCHEMA}
+PRODUCTION_PIPELINE_VERSION = "cm95pbf-production/2.0-s52-aoi"
 PRODUCTION_RUN_SCHEMA = "navsea.cm95.production-run/v1"
 
 
@@ -231,6 +234,13 @@ def create_production_plan(
     output_path: Path,
     zooms: list[int],
     profile: str = "delivery",
+    *,
+    s52_lookups_path: Path | None = None,
+    s52_point_table: str = "Paper",
+    s52_boundary_table: str = "Plain",
+    s52_mariner_mode: str = "runtime",
+    aoi_name: str | None = None,
+    aoi_bbox_wgs84: tuple[float, float, float, float] | None = None,
 ) -> dict[str, Any]:
     if profile not in {"delivery", "engineering"}:
         raise ValueError("profile must be delivery or engineering")
@@ -240,6 +250,29 @@ def create_production_plan(
     output_path = Path(output_path)
     policy = _load_policy(policy_path)
     zooms = parse_zooms(zooms)
+    if s52_point_table not in {"Simplified", "Paper"}:
+        raise ValueError("s52_point_table must be Simplified or Paper")
+    if s52_boundary_table not in {"Plain", "Symbolized"}:
+        raise ValueError("s52_boundary_table must be Plain or Symbolized")
+    if s52_mariner_mode not in {"runtime", "reference"}:
+        raise ValueError("s52_mariner_mode must be runtime or reference")
+
+    s52_path = Path(s52_lookups_path) if s52_lookups_path is not None else None
+    if s52_path is not None and not s52_path.exists():
+        raise RuntimeError(f"S-52 lookup file not found: {s52_path}")
+
+    aoi_bbox = None
+    aoi_box_m = None
+    if aoi_bbox_wgs84 is not None:
+        west, south, east, north = (float(v) for v in aoi_bbox_wgs84)
+        if not (-180.0 <= west < east <= 180.0):
+            raise ValueError(f"invalid AOI longitude range: {aoi_bbox_wgs84}")
+        if not (-85.0 <= south < north <= 85.0):
+            raise ValueError(f"invalid AOI latitude range: {aoi_bbox_wgs84}")
+        aoi_bbox = [west, south, east, north]
+        minx, miny = lonlat_to_mercator(west, south)
+        maxx, maxy = lonlat_to_mercator(east, north)
+        aoi_box_m = box(minx, miny, maxx, maxy)
 
     all_scales = sorted({s for z in zooms for s in _enumeration_scales(policy, z)})
 
@@ -259,7 +292,16 @@ def create_production_plan(
         ladder = _scale_ladder(policy, z)
         enumeration = _enumeration_scales(policy, z)
         enumeration_set = set(enumeration)
-        z_pieces = [(cid, scale, geom, src) for cid, scale, geom, src in pieces if scale in enumeration_set]
+        raw_z_pieces = [(cid, scale, geom, src) for cid, scale, geom, src in pieces if scale in enumeration_set]
+        z_pieces = []
+        for cid, scale, geom, src in raw_z_pieces:
+            if aoi_box_m is not None:
+                if not geom.intersects(aoi_box_m):
+                    continue
+                geom = geom.intersection(aoi_box_m)
+                if geom.is_empty:
+                    continue
+            z_pieces.append((cid, scale, geom, src))
         z_jobs: set[tuple[int, int, int]] = set()
         fallback_pieces = 0
         for _cid, _scale, geom, source in z_pieces:
@@ -270,6 +312,7 @@ def create_production_plan(
         source_stats[str(z)] = {
             "enumeration_scales": enumeration,
             "scale_ladder": ladder,
+            "coverage_pieces_before_aoi": len(raw_z_pieces),
             "coverage_pieces": len(z_pieces),
             "cell_bbox_fallback_pieces": fallback_pieces,
             "tile_count": len(z_jobs),
@@ -280,8 +323,20 @@ def create_production_plan(
         for z, x, y in sorted(jobs)
     ]
 
+    s52_contract = {
+        "enabled": s52_path is not None,
+        "lookups_sha256": sha256_file(s52_path) if s52_path is not None else None,
+        "point_table": s52_point_table if s52_path is not None else None,
+        "boundary_table": s52_boundary_table if s52_path is not None else None,
+        "mariner_mode": s52_mariner_mode if s52_path is not None else None,
+    }
+    aoi_contract = {
+        "name": str(aoi_name or "").strip() or None,
+        "bbox_wgs84": aoi_bbox,
+    }
     immutable = {
         "schema": PRODUCTION_PLAN_SCHEMA,
+        "production_pipeline_version": PRODUCTION_PIPELINE_VERSION,
         "dataset_id": int(dataset_id),
         "dataset_fingerprint": fingerprint,
         "profile": profile,
@@ -291,6 +346,8 @@ def create_production_plan(
         "builder_contract_version": BUILDER_CONTRACT_VERSION,
         "manifest_dataset_fingerprint": manifest_info["dataset_fingerprint"],
         "manifest_identity_count": int(manifest_info["identity_count"]),
+        "s52": s52_contract,
+        "aoi": aoi_contract,
         "jobs": job_rows,
     }
     plan_id = canonical_sha256(immutable)
@@ -302,6 +359,7 @@ def create_production_plan(
         "mapping_path": str(mapping_path),
         "policy_path": str(policy_path),
         "manifest_path": str(manifest_path),
+        "s52_lookups_path": str(s52_path) if s52_path is not None else None,
         "source_stats": source_stats,
         "job_count": len(job_rows),
     }
@@ -312,7 +370,7 @@ def create_production_plan(
 
 def load_plan(path: Path) -> dict[str, Any]:
     obj = json.loads(Path(path).read_text(encoding="utf-8"))
-    if obj.get("schema") != PRODUCTION_PLAN_SCHEMA:
+    if obj.get("schema") not in SUPPORTED_PRODUCTION_PLAN_SCHEMAS:
         raise RuntimeError(f"unsupported plan schema: {obj.get('schema')}")
     return obj
 
@@ -476,6 +534,16 @@ def _validate_plan_inputs(db: DbConfig, plan: dict[str, Any]):
         raise RuntimeError("mapping file changed after plan creation")
     if sha256_file(policy_path) != plan["policy_sha256"]:
         raise RuntimeError("policy file changed after plan creation")
+    s52 = dict(plan.get("s52") or {})
+    if s52.get("enabled"):
+        s52_path_raw = plan.get("s52_lookups_path")
+        if not s52_path_raw:
+            raise RuntimeError("S-52 enabled plan missing s52_lookups_path")
+        s52_path = Path(s52_path_raw)
+        if not s52_path.exists():
+            raise RuntimeError(f"S-52 lookup file missing: {s52_path}")
+        if sha256_file(s52_path) != str(s52.get("lookups_sha256") or ""):
+            raise RuntimeError("S-52 lookup file changed after plan creation")
     with _connect(db) as conn:
         fp = dataset_fingerprint(conn, int(plan["dataset_id"]))
         if fp != plan["dataset_fingerprint"]:
@@ -525,6 +593,10 @@ def run_production_plan(
                 output_root=output_root,
                 report_root=report_root,
                 profile=profile,
+                s52_lookups_path=(Path(plan["s52_lookups_path"]) if (plan.get("s52") or {}).get("enabled") else None),
+                s52_point_table=str((plan.get("s52") or {}).get("point_table") or "Paper"),
+                s52_boundary_table=str((plan.get("s52") or {}).get("boundary_table") or "Plain"),
+                s52_mariner_mode=str((plan.get("s52") or {}).get("mariner_mode") or "runtime"),
             )
             result = builder.build_tile(TileJob(z, x, y))
             if result.get("hard_gate_unmapped"):
@@ -586,6 +658,9 @@ def _write_run_summary(plan: dict[str, Any], release_dir: Path, summary: dict[st
         "dataset_fingerprint": plan["dataset_fingerprint"],
         "profile": plan["profile"],
         "builder_contract_version": plan.get("builder_contract_version"),
+        "production_pipeline_version": plan.get("production_pipeline_version"),
+        "s52": plan.get("s52"),
+        "aoi": plan.get("aoi"),
         "updated_at": utc_now(),
         **summary,
     }
@@ -602,6 +677,9 @@ def production_status(plan_path: Path, release_root: Path) -> dict[str, Any]:
     return {
         "release_id": plan["release_id"],
         "release_dir": str(release_dir),
+        "production_pipeline_version": plan.get("production_pipeline_version"),
+        "s52": plan.get("s52"),
+        "aoi": plan.get("aoi"),
         **summary,
     }
 
@@ -689,6 +767,10 @@ def migrate_production_release(
             raise RuntimeError("old/new dataset fingerprint mismatch")
         if str(old_summary.get("profile") or "") != str(new_plan["profile"]):
             raise RuntimeError("old/new profile mismatch")
+        new_s52 = dict(new_plan.get("s52") or {})
+        old_s52 = dict(old_summary.get("s52") or {})
+        if new_s52.get("enabled") and old_s52 != new_s52:
+            raise RuntimeError("old/new S-52 production contract mismatch; direct tile reuse is unsafe")
 
     profile = str(new_plan["profile"])
     new_release_dir = Path(release_root) / str(new_plan["release_id"])
@@ -805,6 +887,9 @@ def migrate_production_release(
         "new_release_dir": str(new_release_dir),
         "new_release_id": str(new_plan["release_id"]),
         "builder_contract_version": new_plan.get("builder_contract_version"),
+        "production_pipeline_version": new_plan.get("production_pipeline_version"),
+        "s52": new_plan.get("s52"),
+        "aoi": new_plan.get("aoi"),
         "changed_delivery_classes": sorted(changed_classes),
         "reused_pass_jobs": reused,
         "reused_pbf_files": reused_pbf,

src/cm95pbf/release.py

@@ -336,6 +336,22 @@ def finalize_release(
     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_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)
@@ -348,12 +364,45 @@ def finalize_release(
             pbf_rel = str(path.relative_to(pbf_root)).replace("\\", "/")
             for feat in feats:
                 fid = feat.get("id")
+                props = feat.get("properties") or {}
+                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:
+                                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:
+                            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,
-                    feat.get("properties") or {},
+                    props,
                 )
                 if fid is None:
                     continue
@@ -402,6 +451,20 @@ def finalize_release(
         nav_name_conflict_state
     )
 
+    if s52_runtime_leaks:
+        s52_runtime_failures.append(f"dynamic final state leaked into runtime PBF: {dict(s52_runtime_leaks)}")
+    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())),
+        "dynamic_final_leaks": dict(sorted(s52_runtime_leaks.items())),
+        "failure_count": len(s52_runtime_failures),
+        "failure_examples": s52_runtime_failures[:100],
+    }
+
     failures = []
     if hard_gate:
         failures.append("hard_gate_unmapped")
@@ -417,6 +480,8 @@ def finalize_release(
         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")
 
     result = {
         "schema": RELEASE_SCHEMA,
@@ -431,6 +496,10 @@ def finalize_release(
         "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"),
+        "s52": plan.get("s52"),
+        "aoi": plan.get("aoi"),
+        "s52_runtime_contract": s52_runtime_contract,
         "style_sha256": style_sha,
         "sprite_json_sha256": sprite_sha,
         "gap_disposition_sha256": gap_disposition_sha,