Bind R4 bounded ZIP staging, question recovery and exact qualified image

EVALUATION.md

@@ -1,12 +1,12 @@
 # SEGMENT evaluation evidence
 
-**R3 local checks completed; official platform qualification remains unproven.** The exact archive is `3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d` (35,613,234,136 bytes), immutable generation `1789069382415008`. The build manifest is `5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095` and configuration `fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79`. The same-generation archive passed full compressed/raw25-layer verification and an actual classic Docker import of config `fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79` with all25 ordered DiffIDs; no imported manifest digest was exposed. Five final H100 invocations produced560 answer executions: two canonical10 interface calls with explicitly recorded degraded deferrals, a strict ordinary20, a strict loose500, and a reversed20. The short-case calls matched the frozen saved answers, and reversal matched the same-image ordinary20 by qID. Canonical deferrals are not ten healthy neural-video results. These are runtime/interface/content checks, not new semantic grading, general order invariance, full500 ZIP capacity or an official score. The prior43d4 canonical failure and all failed attempts remain retained.
+**R4 local runtime, archive and import checks completed; official platform qualification remains unproven.** The exact archive is `e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00` (35,613,293,722 bytes), immutable generation `1789092734891901`. The build manifest is `ba9ddb1e78a45040803ae6f0e68c66de7a538f5e3e86397777e5fc2c488106b0` and configuration `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4`. Full compressed/raw archive verification covers all 26 ordered DiffIDs; actual same-generation classic Docker import reports config `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4` and no imported manifest digest. The agreed eight cases completed 9 application invocations and 1593 answer executions across 533 unique qIDs. Healthy20 and loose/ZIP500 matched saved content; reverse20 matched the same-image ordinary20, and both ZIP quotas matched the same-image loose500. Canonical repeated calls retained the exact disclosed deferrals. The mixed recovery case blanked only its declared failed questions and preserved later healthy outputs. These are bounded runtime/interface/content checks; no new semantic grade or official final result is claimed.
 
 The completed development comparison is between two registered serving implementations under the same fixed Qwen3.5-27B judge and 20-case batching protocol: candidate 2373/4000 (59.325%) versus named reference 1940/4000 (48.500%), +10.825 percentage points. Strict sensitivity is 2334/3940 versus 1906/3940. All ten original-source-video differences are positive; 5b causal consequence reasoning regressed 17.742 micro points on 62 cases from six sources. These are adaptively used development results with inherited training/data-exposure limits, not an independent holdout, an official finals score or an all-candidate ranking. The registered local scorer passed14 convenience controls (6positive,8negative). These check response/parser boundaries; they do not estimate organizer-judge agreement, real-question false-positive rates, official calibration or a finals score. All 4,000 were not run through a Docker archive. Source-derived routing equality on the 4,000 intervals 1–299s does not by itself transfer these scores to new image bytes.
 
-The image owns the literal SEGMENT task; duration no longer chooses the task or rejects otherwise valid request metadata. Both start_time and end_time must remain finite, nonnegative and ordered, and their original values are preserved. Only infer_track changes in the image-owned cap_fmt source; infer_all, format/capability inference and the other 51 serving files are unchanged. The bootstrap retains path, source, model, scratch, output and required initialization guards. It records only exactly attributed pre-generation sampler-duration, frame-survival or observed zero/fewer-than-two-frame deferrals while retaining the original rules answer. Unknown, initialization and generation failures remain fatal. Deferrals mean reduced scientific coverage, not healthy neural inference; an answer count or empty observed-deferral list alone does not prove neural generation for every row.
+R4 retains the image-owned literal SEGMENT task and the original request clocks. It stages only the selected ZIP member for each startup probe or answer, verifies that member through CRC/EOF and SHA-256, closes decoder handles, and removes the lease before continuing. Loose input remains a direct read. Model initialization is still eager. An attributed ordinary per-question exception produces empty content for that qID with measured latency and failure telemetry; successful answers are preserved and later questions continue. Unsafe or ambiguous archives, source mutation, cleanup integrity, required model initialization, environment and fixed-track failures remain fatal. The inherited precise pre-generation rules deferrals remain separately disclosed as degraded coverage. The four changed image files are the bootstrap, contract, probe entrypoint and new ZIP helper; all 51 unchanged inherited serving source files, including cap_fmt, all 50 model/prior assets, prompts, model arithmetic and generation settings remain unchanged.
 
-No new semantic score, independent holdout, official calibration, platform pass, full 500 ZIP capacity, general order invariance or causal speedup follows from these changes. Exact image, runtime, archive and load evidence must be joined separately. Public shim source does not attest the deployed version; forced process death can bypass scratch cleanup. The earlier canonical and controller failures remain retained.
+The full retained ZIP500 was qualified locally at the stated 4 GiB and 16 GiB tmpfs profiles, alongside loose500, healthy20, reversed20, default and persistent canonical calls, and controlled recovery. These fixtures do not establish the platform scratch quota, all possible input sizes, continuous resource peaks, cold-cache requirements, general order invariance, new semantic scores, independent-holdout performance or official platform success. Canonical deferrals and deliberately failed recovery rows are reduced scientific coverage. The prior R3 4 GiB eager-ZIP failure is a conditional local capacity reproduction, not proof of the organizer failure cause. Earlier images and failed attempts remain historical.
 
 Measured grader completions: September 10, 2026, 10:53:44 UTC (candidate) and 10:55:13 UTC (reference). Complete readback and actual GPU-return checks passed for both. Paired reporting completed at 11:07:14 UTC.
 

IMAGE_IDENTITY.json

@@ -36,35 +36,38 @@
     "sha256": "2f4dd5055dc7c33c3e9a43c5a443f2946a6cf999f60aa3e453e8d4ce610ac13b"
   },
   "current_image": {
-    "CPU_acceptance_sha256": "c20a1680bf867569101d04c79910ca1f9c04d67caf08d65729e87e8b9800e030",
+    "CPU_acceptance_sha256": "64a3e419b9b0ea9d320c0de0d0e3d2359a010dd77119927bdf4060d32094f5c3",
     "actual_platform_result_uuid": null,
-    "all24_ancestor_layers_preserved": true,
-    "archive_bytes": 35613234136,
-    "archive_generation": "1789069382415008",
-    "archive_receipt_sha256": "2eeeb4daa5f57840ea15c8ab094eaed4b59d56cd666c77da1a562381833337b1",
-    "archive_sha256": "3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d",
-    "archive_uri": "gs://heydonto-surgfield-research/submissions/segment_forward_fixed_track_cloudbuild_20260910_r5/output/segment-final.tar.gz#1789069382415008",
-    "base_archive_sha256": "43d4fc4fd7a39e4884a5179176a6b6a436956a146494876783d6287855a82825",
-    "base_config_sha256": "f7e22f1822d7a152ec91999dbb639ec63378346ef3f5144723c5b453748f97b9",
-    "base_ordered_layer_count": 24,
-    "bootstrap_sha256": "4078387df9a46989ae0ab8b25275dc6d43e82afd5459018d3eaa12df1cf0cd76",
-    "build_receipt_sha256": "0d360628bdf70daf245d6c0ef1c1e7fe7f7db5f6189a4699672daeb6a94c29fa",
-    "build_store_manifest_sha256": "5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095",
-    "candidate_source_sha256": "88ae763faa28c7333031876d4c6e6f2f005b19fb9b1362b6d6ffdaf3cd992e37",
+    "all25_ancestor_layers_preserved": true,
+    "archive_bytes": 35613293722,
+    "archive_generation": "1789092734891901",
+    "archive_receipt_sha256": "a1e4f11381d1be8cc62a8d3d9f8317324c99426c87f5b6be561b9b9224c627a8",
+    "archive_sha256": "e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00",
+    "archive_uri": "gs://heydonto-surgfield-research/submissions/segment_final_recovery_20260911_r1/output/segment-final.tar.gz#1789092734891901",
+    "base_archive_sha256": "3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d",
+    "base_config_sha256": "fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "base_manifest_sha256": "5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095",
+    "base_ordered_layer_count": 25,
+    "bootstrap_sha256": "c6846ff32df7e7027a3de520536ee917b0927f1d55915dfcf530fd035d0a35c8",
+    "build_receipt_sha256": "9ab1411d614c7142568102381fa3c3db5355e465eec7da708db31d61a31764f8",
+    "build_store_manifest_sha256": "ba9ddb1e78a45040803ae6f0e68c66de7a538f5e3e86397777e5fc2c488106b0",
+    "candidate_source_sha256": "6bb01e766b15a196ba78779fd8169dff20a1af2ada0a47db1aa85cefe26fc2c1",
     "cap_fmt_inferrer_sha256": "2861921a562c19937e38346aec219114e64e419bc7dbc7f1eaee4fcaf7f20713",
-    "contract_sha256": "6af2fb47ced6f8bf0bb9b547c5b6580c1a4663869a6eeceb3e2ab90fe7dcc8b7",
-    "docker_config_bytes": 28206,
-    "docker_config_sha256": "fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
-    "docker_loaded_image_id": "sha256:fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "contract_sha256": "2426fca2beaec028bf506085083f9c0eb681620dd667fb0386ffc41236dcc1de",
+    "docker_config_bytes": 28323,
+    "docker_config_sha256": "3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4",
+    "docker_loaded_image_id": "sha256:3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4",
     "docker_loaded_image_id_kind": "classic_config_digest",
-    "layer_count": 25,
+    "layer_count": 26,
     "normalized_archive_import_manifest_sha256": null,
-    "overlay_blob_sha256": "64309f5fa0609fe53792678b90973e41f9ca34490c547dbcae741d7ddd785220",
-    "overlay_diff_id": "91029e508c7a21f763ab8bf3eeca620f1b5ba76cd9a6280331f34c13af596ee8",
+    "overlay_blob_sha256": "e0f8eb14b59dd28adacb89945febed19517e4e5268db36c96a6cc17fe4a3681c",
+    "overlay_diff_id": "1e3951eb1a4e2a1460649a3b4721adada757d5339f84509dc6c33345d9b21392",
     "platform_qualified": false,
-    "runtime_rehearsal_receipt_sha256": "0a8bc8e0cf2d71bebe9085fefd4b2a968595ad69701ea10f592afccc8c35fcfa",
-    "runtime_rehearsal_status": "passed_final_runtime_interface_and_strict_parity_not_platform",
-    "status": "passed_local_interface_strict_parity_archive_load_not_platform"
+    "probe_entrypoint_sha256": "40e91b3780417c3503fe14164385aa1d91ab7571265de05962026228f8cf4e33",
+    "runtime_rehearsal_receipt_sha256": "8b5daa59751678265917a64d96726115edf01c1b10bc8703bb45ec62df4488f0",
+    "runtime_rehearsal_status": "passed_complete_agreed_local_matrix_not_platform",
+    "segment_zip_input_sha256": "aef9806482af9b03ce08946c01240183393fbf15c6a2046bad9e00c27a3f58a7",
+    "status": "passed_local_complete_runtime_matrix_archive_load_not_platform"
   },
   "manifest_identity_note": "Build-store manifest, classic loaded configuration ID and any physically observed imported manifest are separate. Pending fields remain null.",
   "preserved_asset_manifest_scope": {
@@ -133,6 +136,37 @@
     "runtime_rehearsal_status": "passed_targeted_derivative_runtime_not_platform",
     "status": "passed_local_targeted_runtime_and_full_archive_load_not_platform"
   },
+  "prior_fixed_track_R3_image": {
+    "CPU_acceptance_sha256": "c20a1680bf867569101d04c79910ca1f9c04d67caf08d65729e87e8b9800e030",
+    "actual_platform_result_uuid": null,
+    "all24_ancestor_layers_preserved": true,
+    "archive_bytes": 35613234136,
+    "archive_generation": "1789069382415008",
+    "archive_receipt_sha256": "2eeeb4daa5f57840ea15c8ab094eaed4b59d56cd666c77da1a562381833337b1",
+    "archive_sha256": "3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d",
+    "archive_uri": "gs://heydonto-surgfield-research/submissions/segment_forward_fixed_track_cloudbuild_20260910_r5/output/segment-final.tar.gz#1789069382415008",
+    "base_archive_sha256": "43d4fc4fd7a39e4884a5179176a6b6a436956a146494876783d6287855a82825",
+    "base_config_sha256": "f7e22f1822d7a152ec91999dbb639ec63378346ef3f5144723c5b453748f97b9",
+    "base_ordered_layer_count": 24,
+    "bootstrap_sha256": "4078387df9a46989ae0ab8b25275dc6d43e82afd5459018d3eaa12df1cf0cd76",
+    "build_receipt_sha256": "0d360628bdf70daf245d6c0ef1c1e7fe7f7db5f6189a4699672daeb6a94c29fa",
+    "build_store_manifest_sha256": "5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095",
+    "candidate_source_sha256": "88ae763faa28c7333031876d4c6e6f2f005b19fb9b1362b6d6ffdaf3cd992e37",
+    "cap_fmt_inferrer_sha256": "2861921a562c19937e38346aec219114e64e419bc7dbc7f1eaee4fcaf7f20713",
+    "contract_sha256": "6af2fb47ced6f8bf0bb9b547c5b6580c1a4663869a6eeceb3e2ab90fe7dcc8b7",
+    "docker_config_bytes": 28206,
+    "docker_config_sha256": "fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "docker_loaded_image_id": "sha256:fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "docker_loaded_image_id_kind": "classic_config_digest",
+    "layer_count": 25,
+    "normalized_archive_import_manifest_sha256": null,
+    "overlay_blob_sha256": "64309f5fa0609fe53792678b90973e41f9ca34490c547dbcae741d7ddd785220",
+    "overlay_diff_id": "91029e508c7a21f763ab8bf3eeca620f1b5ba76cd9a6280331f34c13af596ee8",
+    "platform_qualified": false,
+    "runtime_rehearsal_receipt_sha256": "0a8bc8e0cf2d71bebe9085fefd4b2a968595ad69701ea10f592afccc8c35fcfa",
+    "runtime_rehearsal_status": "passed_final_runtime_interface_and_strict_parity_not_platform",
+    "status": "passed_local_interface_strict_parity_archive_load_not_platform"
+  },
   "schema": "segment-associated-container-identity/2",
   "shared_serving_implementation": {
     "asset_revision": "35fc28af1d55a5a9ae8096ff2e2cd2c961aa7119",

LINEAGE.json

@@ -49,42 +49,45 @@
   },
   "attribution": "HeyDonto Labs",
   "current_serving_image": {
-    "CPU_acceptance_sha256": "c20a1680bf867569101d04c79910ca1f9c04d67caf08d65729e87e8b9800e030",
+    "CPU_acceptance_sha256": "64a3e419b9b0ea9d320c0de0d0e3d2359a010dd77119927bdf4060d32094f5c3",
     "actual_platform_result_uuid": null,
-    "all24_ancestor_layers_preserved": true,
-    "archive_bytes": 35613234136,
-    "archive_generation": "1789069382415008",
-    "archive_receipt_sha256": "2eeeb4daa5f57840ea15c8ab094eaed4b59d56cd666c77da1a562381833337b1",
-    "archive_sha256": "3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d",
-    "archive_uri": "gs://heydonto-surgfield-research/submissions/segment_forward_fixed_track_cloudbuild_20260910_r5/output/segment-final.tar.gz#1789069382415008",
-    "base_archive_sha256": "43d4fc4fd7a39e4884a5179176a6b6a436956a146494876783d6287855a82825",
-    "base_config_sha256": "f7e22f1822d7a152ec91999dbb639ec63378346ef3f5144723c5b453748f97b9",
-    "base_ordered_layer_count": 24,
-    "bootstrap_sha256": "4078387df9a46989ae0ab8b25275dc6d43e82afd5459018d3eaa12df1cf0cd76",
-    "build_receipt_sha256": "0d360628bdf70daf245d6c0ef1c1e7fe7f7db5f6189a4699672daeb6a94c29fa",
-    "build_store_manifest_sha256": "5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095",
-    "candidate_source_sha256": "88ae763faa28c7333031876d4c6e6f2f005b19fb9b1362b6d6ffdaf3cd992e37",
+    "all25_ancestor_layers_preserved": true,
+    "archive_bytes": 35613293722,
+    "archive_generation": "1789092734891901",
+    "archive_receipt_sha256": "a1e4f11381d1be8cc62a8d3d9f8317324c99426c87f5b6be561b9b9224c627a8",
+    "archive_sha256": "e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00",
+    "archive_uri": "gs://heydonto-surgfield-research/submissions/segment_final_recovery_20260911_r1/output/segment-final.tar.gz#1789092734891901",
+    "base_archive_sha256": "3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d",
+    "base_config_sha256": "fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "base_manifest_sha256": "5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095",
+    "base_ordered_layer_count": 25,
+    "bootstrap_sha256": "c6846ff32df7e7027a3de520536ee917b0927f1d55915dfcf530fd035d0a35c8",
+    "build_receipt_sha256": "9ab1411d614c7142568102381fa3c3db5355e465eec7da708db31d61a31764f8",
+    "build_store_manifest_sha256": "ba9ddb1e78a45040803ae6f0e68c66de7a538f5e3e86397777e5fc2c488106b0",
+    "candidate_source_sha256": "6bb01e766b15a196ba78779fd8169dff20a1af2ada0a47db1aa85cefe26fc2c1",
     "cap_fmt_inferrer_sha256": "2861921a562c19937e38346aec219114e64e419bc7dbc7f1eaee4fcaf7f20713",
-    "contract_sha256": "6af2fb47ced6f8bf0bb9b547c5b6580c1a4663869a6eeceb3e2ab90fe7dcc8b7",
-    "docker_config_bytes": 28206,
-    "docker_config_sha256": "fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
-    "docker_loaded_image_id": "sha256:fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "contract_sha256": "2426fca2beaec028bf506085083f9c0eb681620dd667fb0386ffc41236dcc1de",
+    "docker_config_bytes": 28323,
+    "docker_config_sha256": "3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4",
+    "docker_loaded_image_id": "sha256:3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4",
     "docker_loaded_image_id_kind": "classic_config_digest",
-    "layer_count": 25,
+    "layer_count": 26,
     "normalized_archive_import_manifest_sha256": null,
-    "overlay_blob_sha256": "64309f5fa0609fe53792678b90973e41f9ca34490c547dbcae741d7ddd785220",
-    "overlay_diff_id": "91029e508c7a21f763ab8bf3eeca620f1b5ba76cd9a6280331f34c13af596ee8",
+    "overlay_blob_sha256": "e0f8eb14b59dd28adacb89945febed19517e4e5268db36c96a6cc17fe4a3681c",
+    "overlay_diff_id": "1e3951eb1a4e2a1460649a3b4721adada757d5339f84509dc6c33345d9b21392",
     "platform_qualified": false,
-    "runtime_rehearsal_receipt_sha256": "0a8bc8e0cf2d71bebe9085fefd4b2a968595ad69701ea10f592afccc8c35fcfa",
-    "runtime_rehearsal_status": "passed_final_runtime_interface_and_strict_parity_not_platform",
-    "status": "passed_local_interface_strict_parity_archive_load_not_platform"
+    "probe_entrypoint_sha256": "40e91b3780417c3503fe14164385aa1d91ab7571265de05962026228f8cf4e33",
+    "runtime_rehearsal_receipt_sha256": "8b5daa59751678265917a64d96726115edf01c1b10bc8703bb45ec62df4488f0",
+    "runtime_rehearsal_status": "passed_complete_agreed_local_matrix_not_platform",
+    "segment_zip_input_sha256": "aef9806482af9b03ce08946c01240183393fbf15c6a2046bad9e00c27a3f58a7",
+    "status": "passed_local_complete_runtime_matrix_archive_load_not_platform"
   },
-  "current_serving_image_reference": "IMAGE_IDENTITY.json#/current_image",
+  "current_serving_image_reference": "IMAGE_IDENTITY.json current_image (R4); all prior image identities are explicitly historical.",
   "derived_annotation_current_scope_clarification": "The known derived packet contains all 13,177 listed extensions: 4,621 paraphrases, 5,217 count remints, 2,317 convention remints and 1,022 adoption rows. Exclusion of the 33,581 organizer rows is intentional, not evidence that those rows must be republished. The unresolved items are the separate private annotation URL/access/delivery receipts and complete signature provenance.",
   "forward_change": "Only tested frame_sampler; same selected models/priors/precision/rendering/generation.",
   "forward_gradient_updates": 0,
-  "full_algorithm_and_image_qualification": "Exact R3 local interface, strict short-case parity and full archive/load accepted. Canonical deferrals remain degraded; no official platform pass or fresh semantic score.",
-  "image_compatibility_change": "The image owns the literal SEGMENT task; duration no longer chooses the task or rejects otherwise valid request metadata. Both start_time and end_time must remain finite, nonnegative and ordered, and their original values are preserved. Only infer_track changes in the image-owned cap_fmt source; infer_all, format/capability inference and the other 51 serving files are unchanged. The bootstrap retains path, source, model, scratch, output and required initialization guards. It records only exactly attributed pre-generation sampler-duration, frame-survival or observed zero/fewer-than-two-frame deferrals while retaining the original rules answer. Unknown, initialization and generation failures remain fatal. Deferrals mean reduced scientific coverage, not healthy neural inference; an answer count or empty observed-deferral list alone does not prove neural generation for every row.",
+  "full_algorithm_and_image_qualification": "R4 local runtime, archive and import checks completed; official platform qualification remains unproven. The exact archive is `e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00` (35,613,293,722 bytes), immutable generation `1789092734891901`. The build manifest is `ba9ddb1e78a45040803ae6f0e68c66de7a538f5e3e86397777e5fc2c488106b0` and configuration `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4`. Full compressed/raw archive verification covers all 26 ordered DiffIDs; actual same-generation classic Docker import reports config `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4` and no imported manifest digest. The agreed eight cases completed 9 application invocations and 1593 answer executions across 533 unique qIDs. Healthy20 and loose/ZIP500 matched saved content; reverse20 matched the same-image ordinary20, and both ZIP quotas matched the same-image loose500. Canonical repeated calls retained the exact disclosed deferrals. The mixed recovery case blanked only its declared failed questions and preserved later healthy outputs. These are bounded runtime/interface/content checks; no new semantic grade or official final result is claimed. The full retained ZIP500 was qualified locally at the stated 4 GiB and 16 GiB tmpfs profiles, alongside loose500, healthy20, reversed20, default and persistent canonical calls, and controlled recovery. These fixtures do not establish the platform scratch quota, all possible input sizes, continuous resource peaks, cold-cache requirements, general order invariance, new semantic scores, independent-holdout performance or official platform success. Canonical deferrals and deliberately failed recovery rows are reduced scientific coverage. The prior R3 4 GiB eager-ZIP failure is a conditional local capacity reproduction, not proof of the organizer failure cause. Earlier images and failed attempts remain historical.",
+  "image_compatibility_change": "R4 retains the image-owned literal SEGMENT task and the original request clocks. It stages only the selected ZIP member for each startup probe or answer, verifies that member through CRC/EOF and SHA-256, closes decoder handles, and removes the lease before continuing. Loose input remains a direct read. Model initialization is still eager. An attributed ordinary per-question exception produces empty content for that qID with measured latency and failure telemetry; successful answers are preserved and later questions continue. Unsafe or ambiguous archives, source mutation, cleanup integrity, required model initialization, environment and fixed-track failures remain fatal. The inherited precise pre-generation rules deferrals remain separately disclosed as degraded coverage. The four changed image files are the bootstrap, contract, probe entrypoint and new ZIP helper; all 51 unchanged inherited serving source files, including cap_fmt, all 50 model/prior assets, prompts, model arithmetic and generation settings remain unchanged.",
   "image_identity": {
     "added_diff_id": "ee65a0fd5c5057739b208abb4bab056ff0b874e40fd9dfa43c18acf1ddbf5e1e",
     "adoption_sha256": "7e46d6b9ff4472e894fbcb2279e79c8ecc25daf251e0be11b4c47f96c6dff4b2",
@@ -279,6 +282,37 @@
     "runtime_rehearsal_status": "passed_targeted_derivative_runtime_not_platform",
     "status": "passed_local_targeted_runtime_and_full_archive_load_not_platform"
   },
+  "prior_fixed_track_R3_serving_image": {
+    "CPU_acceptance_sha256": "c20a1680bf867569101d04c79910ca1f9c04d67caf08d65729e87e8b9800e030",
+    "actual_platform_result_uuid": null,
+    "all24_ancestor_layers_preserved": true,
+    "archive_bytes": 35613234136,
+    "archive_generation": "1789069382415008",
+    "archive_receipt_sha256": "2eeeb4daa5f57840ea15c8ab094eaed4b59d56cd666c77da1a562381833337b1",
+    "archive_sha256": "3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d",
+    "archive_uri": "gs://heydonto-surgfield-research/submissions/segment_forward_fixed_track_cloudbuild_20260910_r5/output/segment-final.tar.gz#1789069382415008",
+    "base_archive_sha256": "43d4fc4fd7a39e4884a5179176a6b6a436956a146494876783d6287855a82825",
+    "base_config_sha256": "f7e22f1822d7a152ec91999dbb639ec63378346ef3f5144723c5b453748f97b9",
+    "base_ordered_layer_count": 24,
+    "bootstrap_sha256": "4078387df9a46989ae0ab8b25275dc6d43e82afd5459018d3eaa12df1cf0cd76",
+    "build_receipt_sha256": "0d360628bdf70daf245d6c0ef1c1e7fe7f7db5f6189a4699672daeb6a94c29fa",
+    "build_store_manifest_sha256": "5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095",
+    "candidate_source_sha256": "88ae763faa28c7333031876d4c6e6f2f005b19fb9b1362b6d6ffdaf3cd992e37",
+    "cap_fmt_inferrer_sha256": "2861921a562c19937e38346aec219114e64e419bc7dbc7f1eaee4fcaf7f20713",
+    "contract_sha256": "6af2fb47ced6f8bf0bb9b547c5b6580c1a4663869a6eeceb3e2ab90fe7dcc8b7",
+    "docker_config_bytes": 28206,
+    "docker_config_sha256": "fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "docker_loaded_image_id": "sha256:fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79",
+    "docker_loaded_image_id_kind": "classic_config_digest",
+    "layer_count": 25,
+    "normalized_archive_import_manifest_sha256": null,
+    "overlay_blob_sha256": "64309f5fa0609fe53792678b90973e41f9ca34490c547dbcae741d7ddd785220",
+    "overlay_diff_id": "91029e508c7a21f763ab8bf3eeca620f1b5ba76cd9a6280331f34c13af596ee8",
+    "platform_qualified": false,
+    "runtime_rehearsal_receipt_sha256": "0a8bc8e0cf2d71bebe9085fefd4b2a968595ad69701ea10f592afccc8c35fcfa",
+    "runtime_rehearsal_status": "passed_final_runtime_interface_and_strict_parity_not_platform",
+    "status": "passed_local_interface_strict_parity_archive_load_not_platform"
+  },
   "responsible_contact": {
     "name": "Reza Nehzati, Ph.D.",
     "profile": "https://huggingface.co/rezanehzati"

LINEAGE.md

@@ -17,3 +17,12 @@ The six inherited prior files remain byte-identical. `cholec_priors_v38.json` ar
 The original forward candidate changed the frame sampler (`cd146570…`). The later R3 compatibility derivative additionally fixes image-owned SEGMENT routing in cap_fmt_inferrer.py and records exactly attributed pre-generation deferrals in its bootstrap; the original timestamps, sampler/model behavior and all other 51 serving files remain unchanged. Public Qwen, private N3, both selected adapters, all six priors, prompts, precision, frame limits and generation settings are unchanged. `TRAINING_RESOURCES.json` preserves selected-save timing/peak scopes and producer identities; `MODEL_IDENTITY.json` preserves every selected file identity.
 
 The known derived packet contains all 13,177 listed extensions: 4,621 paraphrases, 5,217 count remints, 2,317 convention remints and 1,022 adoption rows. Exclusion of the 33,581 organizer rows is intentional, not evidence that those rows must be republished. The unresolved items are the separate private annotation URL/access/delivery receipts and complete signature provenance.
+
+
+## R4 serving compatibility update
+
+R4 local runtime, archive and import checks completed; official platform qualification remains unproven. The exact archive is `e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00` (35,613,293,722 bytes), immutable generation `1789092734891901`. The build manifest is `ba9ddb1e78a45040803ae6f0e68c66de7a538f5e3e86397777e5fc2c488106b0` and configuration `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4`. Full compressed/raw archive verification covers all 26 ordered DiffIDs; actual same-generation classic Docker import reports config `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4` and no imported manifest digest. The agreed eight cases completed 9 application invocations and 1593 answer executions across 533 unique qIDs. Healthy20 and loose/ZIP500 matched saved content; reverse20 matched the same-image ordinary20, and both ZIP quotas matched the same-image loose500. Canonical repeated calls retained the exact disclosed deferrals. The mixed recovery case blanked only its declared failed questions and preserved later healthy outputs. These are bounded runtime/interface/content checks; no new semantic grade or official final result is claimed.
+
+R4 retains the image-owned literal SEGMENT task and the original request clocks. It stages only the selected ZIP member for each startup probe or answer, verifies that member through CRC/EOF and SHA-256, closes decoder handles, and removes the lease before continuing. Loose input remains a direct read. Model initialization is still eager. An attributed ordinary per-question exception produces empty content for that qID with measured latency and failure telemetry; successful answers are preserved and later questions continue. Unsafe or ambiguous archives, source mutation, cleanup integrity, required model initialization, environment and fixed-track failures remain fatal. The inherited precise pre-generation rules deferrals remain separately disclosed as degraded coverage. The four changed image files are the bootstrap, contract, probe entrypoint and new ZIP helper; all 51 unchanged inherited serving source files, including cap_fmt, all 50 model/prior assets, prompts, model arithmetic and generation settings remain unchanged.
+
+The full retained ZIP500 was qualified locally at the stated 4 GiB and 16 GiB tmpfs profiles, alongside loose500, healthy20, reversed20, default and persistent canonical calls, and controlled recovery. These fixtures do not establish the platform scratch quota, all possible input sizes, continuous resource peaks, cold-cache requirements, general order invariance, new semantic scores, independent-holdout performance or official platform success. Canonical deferrals and deliberately failed recovery rows are reduced scientific coverage. The prior R3 4 GiB eager-ZIP failure is a conditional local capacity reproduction, not proof of the organizer failure cause. Earlier images and failed attempts remain historical.

LOAD.md

@@ -1,6 +1,6 @@
 # Local loading from two immutable revisions
 
-These revisions reproduce the unchanged **model components and loading tools**. For full serving, the locally qualified R3 archive has SHA-256 `3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d`. The earlier 43d4 archive failed its canonical platform tryout and remains historical. Its exact GCS generation, configuration, compatibility-wrapper identities and completed or pending validation are recorded in [IMAGE_IDENTITY.json](IMAGE_IDENTITY.json) and [PLATFORM_COMPATIBILITY.md](PLATFORM_COMPATIBILITY.md) at the repository revision containing this page. Use those current records to select the serving archive.
+These revisions reproduce the unchanged **model components and loading tools**. For full serving, the locally qualified R4 archive has SHA-256 `e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00`. Its exact GCS generation, configuration, four compatibility-source identities and bounded local runtime evidence are recorded in [IMAGE_IDENTITY.json](IMAGE_IDENTITY.json) and [PLATFORM_COMPATIBILITY.md](PLATFORM_COMPATIBILITY.md) at the repository revision containing this page. Prior archives remain historical. Use those current records to select the serving archive.
 
 The metadata download below remains pinned to revision M and excludes its archived `README.md`, `LOAD.md` and `IMAGE_IDENTITY.json`. The retained `MODEL_FILES.json` and lineage/binding documents describe component ancestry and may contain the ed85 ancestor's image identity; those bindings do not select the current platform artifact. The 50 model/prior files, immutable A/M revisions, manifests and component loaders remain unchanged. The fixed wrapper is already inside the current Docker archive and is identified by the current records linked above.
 

PLATFORM_COMPATIBILITY.md

@@ -1,11 +1,11 @@
 # Fixed SEGMENT task and explicit degraded coverage
 
-**R3 local checks completed; official platform qualification remains unproven.** The exact archive is `3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d` (35,613,234,136 bytes), immutable generation `1789069382415008`. The build manifest is `5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095` and configuration `fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79`. The same-generation archive passed full compressed/raw25-layer verification and an actual classic Docker import of config `fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79` with all25 ordered DiffIDs; no imported manifest digest was exposed. Five final H100 invocations produced560 answer executions: two canonical10 interface calls with explicitly recorded degraded deferrals, a strict ordinary20, a strict loose500, and a reversed20. The short-case calls matched the frozen saved answers, and reversal matched the same-image ordinary20 by qID. Canonical deferrals are not ten healthy neural-video results. These are runtime/interface/content checks, not new semantic grading, general order invariance, full500 ZIP capacity or an official score. The prior43d4 canonical failure and all failed attempts remain retained.
+**R4 local runtime, archive and import checks completed; official platform qualification remains unproven.** The exact archive is `e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00` (35,613,293,722 bytes), immutable generation `1789092734891901`. The build manifest is `ba9ddb1e78a45040803ae6f0e68c66de7a538f5e3e86397777e5fc2c488106b0` and configuration `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4`. Full compressed/raw archive verification covers all 26 ordered DiffIDs; actual same-generation classic Docker import reports config `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4` and no imported manifest digest. The agreed eight cases completed 9 application invocations and 1593 answer executions across 533 unique qIDs. Healthy20 and loose/ZIP500 matched saved content; reverse20 matched the same-image ordinary20, and both ZIP quotas matched the same-image loose500. Canonical repeated calls retained the exact disclosed deferrals. The mixed recovery case blanked only its declared failed questions and preserved later healthy outputs. These are bounded runtime/interface/content checks; no new semantic grade or official final result is claimed.
 
-The image owns the literal SEGMENT task; duration no longer chooses the task or rejects otherwise valid request metadata. Both start_time and end_time must remain finite, nonnegative and ordered, and their original values are preserved. Only infer_track changes in the image-owned cap_fmt source; infer_all, format/capability inference and the other 51 serving files are unchanged. The bootstrap retains path, source, model, scratch, output and required initialization guards. It records only exactly attributed pre-generation sampler-duration, frame-survival or observed zero/fewer-than-two-frame deferrals while retaining the original rules answer. Unknown, initialization and generation failures remain fatal. Deferrals mean reduced scientific coverage, not healthy neural inference; an answer count or empty observed-deferral list alone does not prove neural generation for every row.
+R4 retains the image-owned literal SEGMENT task and the original request clocks. It stages only the selected ZIP member for each startup probe or answer, verifies that member through CRC/EOF and SHA-256, closes decoder handles, and removes the lease before continuing. Loose input remains a direct read. Model initialization is still eager. An attributed ordinary per-question exception produces empty content for that qID with measured latency and failure telemetry; successful answers are preserved and later questions continue. Unsafe or ambiguous archives, source mutation, cleanup integrity, required model initialization, environment and fixed-track failures remain fatal. The inherited precise pre-generation rules deferrals remain separately disclosed as degraded coverage. The four changed image files are the bootstrap, contract, probe entrypoint and new ZIP helper; all 51 unchanged inherited serving source files, including cap_fmt, all 50 model/prior assets, prompts, model arithmetic and generation settings remain unchanged.
 
-No new semantic score, independent holdout, official calibration, platform pass, full 500 ZIP capacity, general order invariance or causal speedup follows from these changes. Exact image, runtime, archive and load evidence must be joined separately. Public shim source does not attest the deployed version; forced process death can bypass scratch cleanup. The earlier canonical and controller failures remain retained.
+The full retained ZIP500 was qualified locally at the stated 4 GiB and 16 GiB tmpfs profiles, alongside loose500, healthy20, reversed20, default and persistent canonical calls, and controlled recovery. These fixtures do not establish the platform scratch quota, all possible input sizes, continuous resource peaks, cold-cache requirements, general order invariance, new semantic scores, independent-holdout performance or official platform success. Canonical deferrals and deliberately failed recovery rows are reduced scientific coverage. The prior R3 4 GiB eager-ZIP failure is a conditional local capacity reproduction, not proof of the organizer failure cause. Earlier images and failed attempts remain historical.
 
-The earlier 43d4 qualification covered six H100 invocations and 600 answer executions across 520 unique questions, including repeated20 cases and a loose500 case. It omitted the owner canonical long intervals and invalid-media control. Its151 positive 500 startup probes left 349 unmeasured; its ZIP coverage was 20, not 500. Those measurements remain historical and do not qualify R3. The completed R2 diagnostic prototype returned 10 answers twice with identical contents but failed its guard on two adaptive_n duration-domain deferrals and one no-frame control. That interface diagnosis is not an all-neural-health pass.
+The earlier 43d4 qualification covered six H100 invocations and 600 answer executions across 520 unique questions, including repeated20 cases and a loose500 case. It omitted the owner canonical long intervals and invalid-media control. Its151 positive 500 startup probes left 349 unmeasured; its ZIP coverage was 20, not 500. Those measurements remain historical and do not qualify R4. The completed R2 diagnostic prototype returned 10 answers twice with identical contents but failed its guard on two adaptive_n duration-domain deferrals and one no-frame control. That interface diagnosis is not an all-neural-health pass.
 
-The exact R3 source files are published for inspection under platform_wrapper/ and component_source/cap_fmt_inferrer.py. Models, all 50 model/prior assets, immutable assetA35fc28af and loading metadataMbc9e0924, component loaders, arithmetic, prompts, generation settings and original sampler behavior remain unchanged. Only the top-layer bootstrap, contract and one serving helper are replaced.
+The exact four R4 image files are available for inspection under platform_wrapper/. The inherited component_source/cap_fmt_inferrer.py and all component loaders remain unchanged. R4 retains the image-owned literal SEGMENT task and the original request clocks. It stages only the selected ZIP member for each startup probe or answer, verifies that member through CRC/EOF and SHA-256, closes decoder handles, and removes the lease before continuing. Loose input remains a direct read. Model initialization is still eager. An attributed ordinary per-question exception produces empty content for that qID with measured latency and failure telemetry; successful answers are preserved and later questions continue. Unsafe or ambiguous archives, source mutation, cleanup integrity, required model initialization, environment and fixed-track failures remain fatal. The inherited precise pre-generation rules deferrals remain separately disclosed as degraded coverage. The four changed image files are the bootstrap, contract, probe entrypoint and new ZIP helper; all 51 unchanged inherited serving source files, including cap_fmt, all 50 model/prior assets, prompts, model arithmetic and generation settings remain unchanged.

README.md

@@ -16,7 +16,7 @@ tags:
 ---
 # SURGFIELD ORena2026 SEGMENT FINAL
 
-**R3 local checks completed; official platform qualification remains unproven.** The exact archive is `3f86686527e2695697d7e2958c6b6026abba4dd0ad07918c6ddc3dfc4d39c51d` (35,613,234,136 bytes), immutable generation `1789069382415008`. The build manifest is `5752df0b2ac35a0660fd4c5627c59c383ad33ac0b5823dd56647a999ba4af095` and configuration `fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79`. The same-generation archive passed full compressed/raw25-layer verification and an actual classic Docker import of config `fa8e31b08886a361c45c80425485b08de0fa3f837b4951a0fab636a18952fe79` with all25 ordered DiffIDs; no imported manifest digest was exposed. Five final H100 invocations produced560 answer executions: two canonical10 interface calls with explicitly recorded degraded deferrals, a strict ordinary20, a strict loose500, and a reversed20. The short-case calls matched the frozen saved answers, and reversal matched the same-image ordinary20 by qID. Canonical deferrals are not ten healthy neural-video results. These are runtime/interface/content checks, not new semantic grading, general order invariance, full500 ZIP capacity or an official score. The prior43d4 canonical failure and all failed attempts remain retained.
+**R4 local runtime, archive and import checks completed; official platform qualification remains unproven.** The exact archive is `e477d307ff2e09f469d83569ec1ce8703daeef34edbba633a3f240eab955de00` (35,613,293,722 bytes), immutable generation `1789092734891901`. The build manifest is `ba9ddb1e78a45040803ae6f0e68c66de7a538f5e3e86397777e5fc2c488106b0` and configuration `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4`. Full compressed/raw archive verification covers all 26 ordered DiffIDs; actual same-generation classic Docker import reports config `3f43fce1081dc5db454447dbe4678947e5cefec014635e611d8d1032a0ac71c4` and no imported manifest digest. The agreed eight cases completed 9 application invocations and 1593 answer executions across 533 unique qIDs. Healthy20 and loose/ZIP500 matched saved content; reverse20 matched the same-image ordinary20, and both ZIP quotas matched the same-image loose500. Canonical repeated calls retained the exact disclosed deferrals. The mixed recovery case blanked only its declared failed questions and preserved later healthy outputs. These are bounded runtime/interface/content checks; no new semantic grade or official final result is claimed.
 
 Private model and resource handoff by **HeyDonto Labs**. Responsible contact: **Reza Nehzati, Ph.D.**, via [rezanehzati](https://huggingface.co/rezanehzati).
 
@@ -24,11 +24,11 @@ This candidate answers segment-level surgical-video questions through fixed rule
 
 `SERVING_ASSETS.json` lists all **50 required model/prior files**: 22 private assets, 14 pinned public Qwen files, 8 pinned OWLv2 files and 6 priors. `LOAD.md` downloads immutable asset and metadata revisions into separate directories and explains the frozen loading contract. LOAD pins asset revision `35fc28af1d55a5a9ae8096ff2e2cd2c961aa7119` and metadata/tools revision `bc9e092422beac9ae51005694824cdc70205e0d5` separately. The private N3 backbone and public Qwen backbone are distinct; both are required. The inactive OWLv2 `.bin` duplicate and obsolete adapters are excluded.
 
-The image owns the literal SEGMENT task; duration no longer chooses the task or rejects otherwise valid request metadata. Both start_time and end_time must remain finite, nonnegative and ordered, and their original values are preserved. Only infer_track changes in the image-owned cap_fmt source; infer_all, format/capability inference and the other 51 serving files are unchanged. The bootstrap retains path, source, model, scratch, output and required initialization guards. It records only exactly attributed pre-generation sampler-duration, frame-survival or observed zero/fewer-than-two-frame deferrals while retaining the original rules answer. Unknown, initialization and generation failures remain fatal. Deferrals mean reduced scientific coverage, not healthy neural inference; an answer count or empty observed-deferral list alone does not prove neural generation for every row.
+R4 retains the image-owned literal SEGMENT task and the original request clocks. It stages only the selected ZIP member for each startup probe or answer, verifies that member through CRC/EOF and SHA-256, closes decoder handles, and removes the lease before continuing. Loose input remains a direct read. Model initialization is still eager. An attributed ordinary per-question exception produces empty content for that qID with measured latency and failure telemetry; successful answers are preserved and later questions continue. Unsafe or ambiguous archives, source mutation, cleanup integrity, required model initialization, environment and fixed-track failures remain fatal. The inherited precise pre-generation rules deferrals remain separately disclosed as degraded coverage. The four changed image files are the bootstrap, contract, probe entrypoint and new ZIP helper; all 51 unchanged inherited serving source files, including cap_fmt, all 50 model/prior assets, prompts, model arithmetic and generation settings remain unchanged.
 
 The completed development comparison is between two registered serving implementations under the same fixed Qwen3.5-27B judge and 20-case batching protocol: candidate 2373/4000 (59.325%) versus named reference 1940/4000 (48.500%), +10.825 percentage points. Strict sensitivity is 2334/3940 versus 1906/3940. All ten original-source-video differences are positive; 5b causal consequence reasoning regressed 17.742 micro points on 62 cases from six sources. These are adaptively used development results with inherited training/data-exposure limits, not an independent holdout, an official finals score or an all-candidate ranking. The registered local scorer passed14 convenience controls (6positive,8negative). These check response/parser boundaries; they do not estimate organizer-judge agreement, real-question false-positive rates, official calibration or a finals score. All 4,000 were not run through a Docker archive. Source-derived routing equality on the 4,000 intervals 1–299s does not by itself transfer these scores to new image bytes.
 
-No new semantic score, independent holdout, official calibration, platform pass, full 500 ZIP capacity, general order invariance or causal speedup follows from these changes. Exact image, runtime, archive and load evidence must be joined separately. Public shim source does not attest the deployed version; forced process death can bypass scratch cleanup. The earlier canonical and controller failures remain retained.
+The full retained ZIP500 was qualified locally at the stated 4 GiB and 16 GiB tmpfs profiles, alongside loose500, healthy20, reversed20, default and persistent canonical calls, and controlled recovery. These fixtures do not establish the platform scratch quota, all possible input sizes, continuous resource peaks, cold-cache requirements, general order invariance, new semantic scores, independent-holdout performance or official platform success. Canonical deferrals and deliberately failed recovery rows are reduced scientific coverage. The prior R3 4 GiB eager-ZIP failure is a conditional local capacity reproduction, not proof of the organizer failure cause. Earlier images and failed attempts remain historical.
 
 Training lineage and selected-prefix exposure are disclosed in `LINEAGE.md` and the accompanying JSONs. The separately retained annotation packet has 13,177 derived rows, not the complete 46,758-row ancestor corpus; it contains no underlying video pixels or complete signature ledger.
 

REQUIRED_LINKS.json

@@ -1,6 +1,7 @@
 {
-  "current_image_publication_status": "reviewed_local_candidate_pending_private_CAS",
+  "current_image_publication_status": "R4 payload locally qualified; exact replacement private commit/readback must be bound separately.",
   "current_records": {
+    "R4_source": "platform_wrapper/ (four exact image source files)",
     "asset_inventory": "MODEL_FILES.json",
     "asset_revision": "35fc28af1d55a5a9ae8096ff2e2cd2c961aa7119",
     "current_loading_instructions": "LOAD.md",
@@ -17,7 +18,9 @@
     "licenses": "NOTICES.md",
     "lineage": "LINEAGE.md",
     "loading": "LOAD.md",
-    "model_card": "README.md"
+    "model_card": "README.md",
+    "r4_probe_source": "platform_wrapper/probe_entrypoint.py",
+    "r4_zip_source": "platform_wrapper/segment_zip_input.py"
   },
   "historical_availability_verification": {
     "asset_rows_matching_immutable_A_custody": 50,

platform_wrapper/CONTRACT.json

@@ -133,6 +133,14 @@
   },
   "input_root": "/input",
   "output_root": "/output",
+  "per_question_failure_policy": {
+    "fatal": "initialization, source/environment/fixed-track, unsafe archive/path/identity/cleanup integrity, non-Exception BaseException (e.g. SystemExit/KeyboardInterrupt)",
+    "intentional_deferrals": "existing rules behavior retained and disclosed",
+    "ordinary_attributed_exceptions": "blank only affected qID and continue; degraded scientific qualification",
+    "output": "exact qID census and finite nonnegative latency; empty content only with attributed failure",
+    "revision": "R4",
+    "zip_staging": "one selected member lease at a time; full member EOF/CRC before model access; no compressed archive copy"
+  },
   "platform_root_compatibility": {
     "linked_input_parent": "/opt/ml/input/data",
     "linked_input_suffix": "-input",
@@ -196,7 +204,7 @@
     "TRITON_CACHE_DIR": "/tmp/orena-segment/triton",
     "XDG_CACHE_HOME": "/tmp/orena-segment/cache"
   },
-  "runtime_validation": "fixed_segment_and_attributed_pre_generation_deferral_derivative_pending_actual_probe_and_runtime_qualification",
+  "runtime_validation": "Successful fixed-source and model initialization, exact per-question attempt/skip and output census; attributed question failures produce blank answers and degraded scientific qualification; source/environment/task/archive-integrity failures remain fatal.",
   "schema_version": "orena-segment-runtime/1",
   "source_files": [
     {
@@ -260,10 +268,10 @@
       "sha256": "a55143dfdbeca52503e225a8a1d2529ef359d3f905ee00ae4d919c6c885c065f"
     },
     {
-      "bytes": 52738,
-      "package_path": "reference/333cd7fe/solution/entrypoint/probe_entrypoint.py",
+      "bytes": 57832,
+      "package_path": "runtime/platform_input_r4/probe_entrypoint.py",
       "path": "/app/solution/entrypoint/probe_entrypoint.py",
-      "sha256": "82ef94f70fb3ca6ddddd74ff372e7c2eb4a9272425c4031725033f682e7bed13"
+      "sha256": "40e91b3780417c3503fe14164385aa1d91ab7571265de05962026228f8cf4e33"
     },
     {
       "bytes": 4419,
@@ -289,6 +297,12 @@
       "path": "/app/solution/entrypoint/segment_unmerged_main.py",
       "sha256": "7a4df6453eba63c01e7638a5ea402e78653a47f83fb7e25d4cf53f030ef06cf6"
     },
+    {
+      "bytes": 16014,
+      "package_path": "runtime/platform_input_r4/segment_zip_input.py",
+      "path": "/app/solution/entrypoint/segment_zip_input.py",
+      "sha256": "aef9806482af9b03ce08946c01240183393fbf15c6a2046bad9e00c27a3f58a7"
+    },
     {
       "bytes": 21221,
       "package_path": "reference/333cd7fe/solution/entrypoint/test_procedure_byte_identity.py",

platform_wrapper/bootstrap.py

@@ -25,7 +25,7 @@ import zipfile
 
 TRACK = "segment"
 PLATFORM_ROOT_POLICY = {'schema': 'segment-declared-platform-input-root/1', 'logical_input_root': '/input', 'linked_input_parent': '/opt/ml/input/data', 'linked_input_suffix': '-input', 'production_scratch_policy': 'exclusive_per_invocation_children', 'output_and_tmp_aliases': 'refused'}
-CONTRACT_SHA256 = '6af2fb47ced6f8bf0bb9b547c5b6580c1a4663869a6eeceb3e2ab90fe7dcc8b7'
+CONTRACT_SHA256 = '2426fca2beaec028bf506085083f9c0eb681620dd667fb0386ffc41236dcc1de'
 IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".bmp", ".webp"}
 VIDEO_EXTENSIONS = {".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v", ".mpg", ".mpeg"}
 SAFE_COMPONENT = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]{0,254}\Z")
@@ -39,7 +39,7 @@ PROTECTED_NAMES = {"PATH", "LD_LIBRARY_PATH", "LD_PRELOAD", "LD_AUDIT", "HOME",
                    "OWLV2_REVISION", "CUDA_VISIBLE_DEVICES", "NVIDIA_VISIBLE_DEVICES",
                    "NVIDIA_DRIVER_CAPABILITIES"}
 DEGRADED_MARKERS = ("PROBESHELL DEGRADED:", "PROBESHELL: launcher floor stands",
-                    "SHIP-CONFIG ASSERT FAILED", "PROBESHELL: RESOLVE-FAIL",
+                    "SHIP-CONFIG ASSERT FAILED",
                     "TRACK: FRAME RESOLVE-FAIL", "WARN: Owlv2 model load failed:",
                     "keeping the pre-seeded fallback", "L17 eager-load: failed",
                     "L17 eager-load raised:")
@@ -308,6 +308,15 @@ def safe_component(value, label):
     return value
 
 
+def literal_qid(value, label):
+    import unicodedata
+    if (not isinstance(value, str) or not value or value in (".", "..")
+            or "/" in value or "\\" in value
+            or any(unicodedata.category(c) in ("Cc", "Cs") for c in value)):
+        raise ValueError(f"unsafe {label}")
+    return value
+
+
 def validate_inputs(root: Path):
     data, request_sha = read_json(root / "request.json")
     rows = [data] if isinstance(data, dict) else data
@@ -317,7 +326,7 @@ def validate_inputs(root: Path):
     for row in rows:
         if not isinstance(row, dict):
             raise ValueError("request row must be an object")
-        qid = safe_component(row.get("qID"), "qID")
+        qid = literal_qid(row.get("qID"), "qID")
         if qid in seen:
             raise ValueError("duplicate qID")
         seen.add(qid)
@@ -325,8 +334,8 @@ def validate_inputs(root: Path):
             raise ValueError("request track mismatch")
         if not isinstance(row.get("question"), str) or not row["question"].strip():
             raise ValueError("question must be nonempty text")
-        if row.get("videoID") is not None:
-            safe_component(row["videoID"], "videoID")
+        if row.get("videoID") is not None and not isinstance(row["videoID"], str):
+            raise ValueError("videoID must be metadata text")
         for key in ("start_time", "end_time"):
             if key not in row or type(row[key]) not in (int, float) or not math.isfinite(row[key]) or row[key] < 0:
                 raise ValueError(f"invalid or missing {key}")
@@ -380,6 +389,7 @@ def validate_inputs(root: Path):
     if any(name.suffix.lower() in ((VIDEO_EXTENSIONS if TRACK == "frame" else IMAGE_EXTENSIONS)) for name, _ in staged.values()):
         raise ValueError("other-track media present")
     selected = []
+    unresolved = []
     for row in rows:
         qid = row["qID"]
         hits = [name for name in media if name.stem == qid]
@@ -387,11 +397,14 @@ def validate_inputs(root: Path):
         groups.append(hits)
         groups.append([p for p in media if p.stem == row.get("videoID")])
         candidates = next((group for group in groups if group), [])
-        if len(candidates) != 1:
-            raise ValueError(f"missing or ambiguous media for qID: {qid}")
-        selected.append(str(candidates[0]))
+        if len(candidates) > 1:
+            raise ValueError(f"ambiguous media for qID: {qid}")
+        if not candidates:
+            unresolved.append(qid)
+        else:
+            selected.append(str(candidates[0]))
     return {"requests": len(rows), "input_files": len(files), "archives": archive_count,
-            "media_files": len(media), "resolved_requests": len(selected), "request_sha256": request_sha,
+            "media_files": len(media), "resolved_requests": len(selected), "unresolved_qIDs": unresolved, "request_sha256": request_sha,
             "validation": "names, layout and supplied interval metadata only; no media decoding"}
 
 
@@ -399,7 +412,7 @@ def protected(name):
     return name.startswith(PROTECTED_PREFIXES) or name in PROTECTED_NAMES
 
 
-def validate_outputs(output_root: Path, input_root: Path):
+def validate_outputs(output_root: Path, input_root: Path, *, empty_qids=frozenset()):
     regular_path(output_root, directory=True)
     if sorted(p.name for p in output_root.iterdir()) != ["answer.json"]:
         raise ValueError("official output must contain only answer.json")
@@ -413,12 +426,14 @@ def validate_outputs(output_root: Path, input_root: Path):
     for answer in answers:
         if not isinstance(answer, dict) or set(answer) != {"qID", "content", "latency"}:
             raise ValueError("answer schema requires exactly qID/content/latency")
-        qid = safe_component(answer["qID"], "answer qID")
+        qid = literal_qid(answer["qID"], "answer qID")
         if qid in seen:
             raise ValueError("duplicate answer qID")
         seen.add(qid)
-        if not isinstance(answer["content"], str) or not answer["content"].strip():
-            raise ValueError("answer content must be nonempty text")
+        if not isinstance(answer["content"], str):
+            raise ValueError("answer content must be text")
+        if not answer["content"].strip() and qid not in empty_qids:
+            raise ValueError("empty answer lacks an attributed per-question failure")
         latency = answer["latency"]
         if type(latency) not in (int, float) or not math.isfinite(latency) or latency < 0:
             raise ValueError("answer latency must be a finite nonnegative number")
@@ -602,6 +617,12 @@ class EngineObservations:
         self.temporal = temporal
         self.temporal_handler = None
         self.input_deferrals = []
+        self.question_failures = []
+        self.answer_qids = []
+        self.answer_calls_failed = 0
+        self.current_answer_qid = None
+        self.allowed_layer_exceptions = {}
+        self.skipped_qids = set()
         self.allowed_deferrals_by_layer = {}
         owner = self
         active = {"layer": None, "qid": None, "sample": None, "sample_return": None}
@@ -610,6 +631,7 @@ class EngineObservations:
             original = getattr(cls, name)
             self.originals.append((cls, name, original))
             def observed(instance, *args, **kwargs):
+                admitted_answer = False
                 try:
                     if name == "answer":
                         # The original caller supplies infer_all's scratch dict.
@@ -620,6 +642,13 @@ class EngineObservations:
                         if ("scratch_init" in kwargs and len(args) > 1) or not isinstance(scratch, dict) or scratch.get("track") != TRACK:
                             owner.events.append("answer scratch_init track missing, ambiguous or foreign")
                             raise ValueError("answer scratch_init must carry the fixed track")
+                        request = args[0] if args else kwargs.get("request")
+                        qid = request.get("qID") if isinstance(request, dict) else getattr(request, "qID", None)
+                        if not isinstance(qid, str) or qid in owner.answer_qids:
+                            raise ValueError("missing or repeated original answer qID")
+                        owner.answer_qids.append(qid)
+                        owner.current_answer_qid = qid
+                        admitted_answer = True
                         owner.counts["answers_track_checked"] += 1
                     result = original(instance, *args, **kwargs)
                     if result_check is not None and not result_check(instance, result):
@@ -628,10 +657,18 @@ class EngineObservations:
                         owner.counts[counter] += 1
                     if name == "__init__":
                         owner.containers.append(instance)
+                        owner.observe_layer_exceptions(instance)
                     return result
                 except BaseException as error:
-                    owner.events.append(f"{name} raised {type(error).__name__}")
+                    if name == "answer" and admitted_answer and isinstance(error, Exception) and not integrity_exception(error):
+                        owner.answer_calls_failed += 1
+                        owner.record_question_failure(exception_event(error, owner.current_answer_qid, "container_answer_exception"))
+                    else:
+                        owner.events.append(f"{name} raised {type(error).__name__}")
                     raise
+                finally:
+                    if admitted_answer:
+                        owner.current_answer_qid = None
             setattr(cls, name, observed)
 
         loaded = lambda instance, result: getattr(instance, "_model", None) is not None and getattr(instance, "_processor", None) is not None
@@ -704,6 +741,18 @@ class EngineObservations:
                     key = id(instance)
                     owner.allowed_deferrals_by_layer[key] = owner.allowed_deferrals_by_layer.get(key, 0) + 1
                     active["sample"] = None
+                elif instance is active["layer"] and active["qid"] is not None:
+                    event = None
+                    if message == expected_message and error is None:
+                        event = {"qID": active["qid"], "stage": "main_pre_generation_media_resolution",
+                                 "reason": "Original main path could not resolve a usable clip before sampling",
+                                 "producer": {"path": "/app/solution/layers/vlm_cholec.py", "function": "predict", "exception": None}}
+                    elif error is not None:
+                        event = classified_question_failure(error, active["qid"], "main")
+                    if event is not None:
+                        owner.record_question_failure(event)
+                        key = id(instance)
+                        owner.allowed_deferrals_by_layer[key] = owner.allowed_deferrals_by_layer.get(key, 0) + 1
                 return original_failure(instance, message, error)
 
             layer_class.predict = observed_predict
@@ -796,27 +845,86 @@ class EngineObservations:
                                  "actual_grid_t": local.get("grid_t"),
                                  "expected_grid_t": local["expected_t"]})
                             return
+                    if exact_producer and ((record.lineno == 175 and record.msg == "segment_temporal_seam: %s R1/build defer (%s)") or (record.lineno == 179 and record.msg == "segment_temporal_seam: %s inference FAILED (%r); defer to rules")) and isinstance(args[1], Exception):
+                        event = classified_question_failure(args[1], active["qid"], "temporal")
+                        if event is not None:
+                            owner.record_question_failure(event)
+                            return
                     if any(marker in message for marker in ("inference FAILED", "R1/build defer", "frames (<2)", "model load FAILED")):
                         owner.events.append("temporal seam reported failure: " + message[:500])
             self.temporal_handler = TemporalHandler()
             temporal.log.addHandler(self.temporal_handler)
 
+    def observe_layer_exceptions(self, container):
+        for layer in getattr(container, "layers", []):
+            for method_name in ("applies", "predict"):
+                original = getattr(layer, method_name, None)
+                if original is None or not hasattr(layer, "name"):
+                    continue
+                self.originals.append((layer, method_name, original))
+                def observed(*args, _original=original, _layer=layer, _method=method_name, **kwargs):
+                    try:
+                        return _original(*args, **kwargs)
+                    except Exception as error:
+                        if self.current_answer_qid is not None and not integrity_exception(error):
+                            key = (id(container), _layer.name)
+                            self.allowed_layer_exceptions[key] = self.allowed_layer_exceptions.get(key, 0) + 1
+                            self.record_question_failure(exception_event(error, self.current_answer_qid, "layer_" + _method + "_exception", layer=_layer.name))
+                        else:
+                            self.events.append("unattributed or integrity layer exception")
+                        raise
+                setattr(layer, method_name, observed)
+
+    def record_question_failure(self, event):
+        self.question_failures.append({**event, "output_policy": "empty_content", "scientific_coverage": "failed question; no correctness or healthy-neural-coverage claim"})
+
+    def reconcile_skipped(self, expected_qids, events, question_events=(), fence_events=()):
+        expected = set(expected_qids)
+        seen = set()
+        for event in events:
+            validate_staging_event(event, expected)
+            seen.add(event["qID"])
+            self.record_question_failure({"qID": event["qID"], "stage": event["stage"], "reason": event["message"], "producer": dict(event)})
+        question_ids = set()
+        for event in question_events:
+            validate_question_event(event, expected)
+            if event["qID"] in question_ids:
+                raise ValueError("repeated answer-loop failure event")
+            question_ids.add(event["qID"])
+            self.record_question_failure({"qID": event["qID"], "stage": event["stage"], "reason": event["message"], "producer": dict(event)})
+        for event in fence_events:
+            if not isinstance(event, dict) or set(event) != {"qID", "reason", "stage"} or event["qID"] not in expected or event["reason"] not in ("WEDGE", "CRASH") or event["stage"] != "startup_decoder_fence":
+                raise ValueError("unrecognized original decoder fence")
+            self.record_question_failure({"qID": event["qID"], "stage": event["stage"], "reason": event["reason"], "producer": dict(event)})
+        if set(self.answer_qids) & seen:
+            raise ValueError("staging-skipped question unexpectedly reached original answer")
+        seen |= question_ids - set(self.answer_qids)
+        if set(self.answer_qids) | seen != expected:
+            raise ValueError("original answer plus attributed staging-skip qID census mismatch")
+        self.skipped_qids = seen
+
     def finish(self, requests):
-        if self.counts["container_initialized"] != 1 or self.counts["answers_completed"] != requests or self.counts["answers_track_checked"] != requests:
+        completed_expected = requests - len(self.skipped_qids)
+        if self.counts["container_initialized"] != 1 or self.counts["answers_completed"] + self.answer_calls_failed != completed_expected or self.counts["answers_track_checked"] != completed_expected:
             self.events.append("engine initialization/answer census mismatch")
         if self.counts["detector_load_completed"] < 1 or self.counts["vlm_eager_load_completed"] < 1:
             self.events.append("required model initialization was not completed")
         if self.temporal is not None and (self.counts["temporal_load_completed"] < 1 or self.temporal._LOAD_FAILED):
             self.events.append("required temporal initialization was not completed")
         for container in self.containers:
-            if any(getattr(container, "exceptions", {}).values()):
-                self.events.append("container recorded a layer exception")
+            for name, count in getattr(container, "exceptions", {}).items():
+                if count != self.allowed_layer_exceptions.get((id(container), name), 0):
+                    self.events.append("container recorded an unattributed layer exception")
             for layer in getattr(container, "layers", []):
                 allowed = self.allowed_deferrals_by_layer.get(id(layer), 0)
                 if getattr(layer, "_load_failed", False) or getattr(layer, "_n_infer_fail", 0) != allowed:
                     self.events.append("layer recorded model load/inference failure")
         return {"counts": dict(self.counts), "failures": list(dict.fromkeys(self.events)),
                 "observed_input_deferrals": list(self.input_deferrals),
+                "observed_question_failures": list(self.question_failures),
+                "answer_qIDs": list(self.answer_qids), "staging_skipped_qIDs": sorted(self.skipped_qids),
+                "answer_calls_failed": self.answer_calls_failed,
+                "strict_model_and_guard_health_passed": not self.events and not self.input_deferrals and not self.question_failures,
                 "input_deferral_summary": {
                     "events": len(self.input_deferrals),
                     "affected_qIDs": sorted({item["qID"] for item in self.input_deferrals}),
@@ -861,6 +969,131 @@ def verify_shell_counters(counters, requests):
             "scope": "original structured shell counters; no correctness or timing conclusion"}
 
 
+def integrity_exception(error):
+    module = sys.modules.get("solution.entrypoint.segment_zip_input")
+    cls = getattr(module, "ZipIntegrityError", None)
+    return isinstance(cls, type) and isinstance(error, cls)
+
+
+def exception_event(error, qid, stage, **metadata):
+    trace, frames = error.__traceback__, []
+    while trace is not None:
+        frames.append({"path": trace.tb_frame.f_code.co_filename, "function": trace.tb_frame.f_code.co_name, "line": trace.tb_lineno})
+        trace = trace.tb_next
+    return {"qID": qid, "stage": stage, "reason": str(error), "producer": {"exception_type": type(error).__name__, "exception_module": type(error).__module__, "traceback": frames, **metadata}}
+
+
+def classified_question_failure(error, qid, route):
+    """Attribute ordinary per-question exceptions; preserve exact known stages and label unknown stages honestly."""
+    trace, frames = error.__traceback__, []
+    while trace is not None:
+        frames.append({"path": trace.tb_frame.f_code.co_filename,
+                       "function": trace.tb_frame.f_code.co_name, "line": trace.tb_lineno})
+        trace = trace.tb_next
+    if not frames or not isinstance(error, Exception) or integrity_exception(error):
+        return None
+    leaf = frames[-1]
+    rendering = "/app/solution/entrypoint/segment_final_rendering.py"
+    temporal = "/app/solution/entrypoint/segment_temporal_seam.py"
+    clock = "/app/solution/entrypoint/segment_main_source_clock.py"
+    stage = None
+    if route == "temporal" and type(error) is ValueError:
+        if leaf["path"] == temporal and leaf["function"] == "segment_temporal_predict" and leaf["line"] in (151, 155):
+            stage = "temporal_pre_generation_frame_census" if leaf["line"] == 151 else "temporal_pre_generation_source_clock"
+        elif leaf["path"] == rendering and leaf["function"] in ("emit", "seconds", "hms") and any(x["path"] == rendering and x["function"] == "emit" for x in frames):
+            stage = "temporal_post_generation_invalid_format"
+    if route == "main":
+        if type(error) is ValueError and leaf["path"] == clock and leaf["function"] in ("sample", "validate_metadata", "content"):
+            stage = "main_pre_generation_source_clock"
+        cv = sys.modules.get("cv2")
+        cv_error = getattr(cv, "error", None)
+        decode_error = isinstance(error, OSError) or (isinstance(cv_error, type) and isinstance(error, cv_error))
+        if decode_error and any(x["path"] == "/app/solution/frame_sampler.py" for x in frames):
+            stage = "main_pre_generation_decoder_exception"
+    if stage is None:
+        stage = route + "_inference_exception_unclassified_stage"
+    return {"qID": qid, "stage": stage, "reason": str(error),
+            "producer": {"exception_type": type(error).__name__, "traceback": frames,
+                         "model_generation_reached": True if stage == "temporal_post_generation_invalid_format" else None}}
+
+
+def validate_staging_event(event, expected_qids):
+    fields = {"qID", "phase", "stage", "archive", "member", "declared_bytes", "written_bytes",
+              "exception_type", "errno", "message", "output_policy"}
+    if not isinstance(event, dict) or set(event) != fields:
+        raise ValueError("unexpected trusted staging event schema")
+    if event["qID"] not in expected_qids or event["phase"] not in ("startup_probe", "answer") or event["output_policy"] != "empty_content":
+        raise ValueError("staging event qID/phase/output policy differs")
+    if event["stage"] not in ("zip_member_staging", "input_media_resolution"):
+        raise ValueError("unknown staging failure class")
+    if not isinstance(event["message"], str) or not isinstance(event["exception_type"], str):
+        raise ValueError("invalid staging exception evidence")
+    if type(event["written_bytes"]) is not int or event["written_bytes"] < 0 or (event["errno"] is not None and type(event["errno"]) is not int):
+        raise ValueError("invalid staging byte/errno evidence")
+    if event["stage"] == "input_media_resolution":
+        if (event["archive"], event["member"], event["declared_bytes"], event["written_bytes"], event["exception_type"]) != (None, None, None, 0, "MissingMedia"):
+            raise ValueError("missing-media event identity differs")
+    elif (not isinstance(event["archive"], str) or not isinstance(event["member"], str)
+          or type(event["declared_bytes"]) is not int or not 0 <= event["written_bytes"] <= event["declared_bytes"]):
+        raise ValueError("ZIP failure member/byte evidence differs")
+
+
+def blank_failed_outputs(output_root, input_root, events):
+    """Only after the original loop: preserve healthy strings and all latencies."""
+    failed = {event["qID"] for event in events}
+    before = validate_outputs(output_root, input_root, empty_qids=failed)
+    answers, original_sha = read_json(output_root / "answer.json")
+    qids = {answer["qID"] for answer in answers}
+    if not failed <= qids:
+        raise ValueError("question-failure ledger contains an unknown qID")
+    if not failed:
+        return {"blanked_qIDs": [], "answer_sha256_before": original_sha, "answer_sha256_after": original_sha,
+                "successful_answer_contents_unchanged": True, "latencies_unchanged": True}
+    kept = {answer["qID"]: dict(answer) for answer in answers}
+    for answer in answers:
+        if answer["qID"] in failed:
+            answer["content"] = ""
+    target = output_root / "answer.json"
+    temp = output_root / ".r4-answer.json.tmp"
+    try:
+        fd = os.open(temp, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600)
+        with os.fdopen(fd, "w", encoding="utf-8") as stream:
+            json.dump(answers, stream, indent=1, allow_nan=False)
+            stream.write("\n")
+        _, current_sha = read_json(target)
+        if current_sha != original_sha:
+            raise ValueError("answer changed after original loop")
+        os.replace(temp, target)
+    finally:
+        if temp.exists():
+            temp.unlink()
+    after = validate_outputs(output_root, input_root, empty_qids=failed)
+    actual, _ = read_json(target)
+    for answer in actual:
+        old = kept[answer["qID"]]
+        if answer["latency"] != old["latency"] or (answer["qID"] not in failed and answer != old):
+            raise ValueError("healthy answer or measured latency changed")
+    return {"blanked_qIDs": sorted(failed), "answer_sha256_before": original_sha,
+            "answer_sha256_after": after["answer_sha256"],
+            "successful_answer_contents_unchanged": True, "latencies_unchanged": True}
+
+
+def validate_question_event(event, expected_qids):
+    if not isinstance(event, dict) or set(event) != {"qID", "stage", "exception_type", "exception_module", "message", "output_policy", "traceback"}:
+        raise ValueError("unexpected trusted question event schema")
+    if event["qID"] not in expected_qids or event["stage"] != "answer_loop_exception" or event["output_policy"] != "empty_content":
+        raise ValueError("question event qID/stage/output policy differs")
+    if any(not isinstance(event[k], str) for k in ("exception_type", "exception_module", "message")):
+        raise ValueError("invalid per-question exception text")
+    frames = event["traceback"]
+    if not isinstance(frames, list) or not frames or any(not isinstance(t, dict) or set(t) != {"path", "function", "line"} or not isinstance(t["path"], str) or not isinstance(t["function"], str) or type(t["line"]) is not int for t in frames):
+        raise ValueError("invalid per-question exception traceback")
+    if frames[0]["path"] != "/app/solution/entrypoint/probe_entrypoint.py" or frames[0]["function"] != "answer_all":
+        raise ValueError("question exception did not originate at the trusted answer-loop boundary")
+    if event["exception_type"] == "ZipIntegrityError":
+        raise ValueError("archive integrity error cannot be contained as a question failure")
+
+
 def execute(contract, *, input_root: Path, output_root: Path, evidence_root: Path | None = None, backend=None, setup_imports=False, production_invocation=False):
     """Synthetic-fixture seam; production supplies only baked paths and launcher."""
     receipt = {"track": TRACK, "status": "failed", "source_validation": None,
@@ -962,14 +1195,32 @@ def execute(contract, *, input_root: Path, output_root: Path, evidence_root: Pat
             guard.violations.append({"key": "environment API", "operation": "guard replaced"})
         guard.final_check()
         receipt["configuration_violations"] = list(guard.violations)
+        question_events = []
         if observations is not None:
+            rows, _ = read_json(input_root / "request.json")
+            rows = [rows] if isinstance(rows, dict) else rows
+            expected_qids = {row["qID"] for row in rows}
+            shell = sys.modules.get("solution.entrypoint.probe_entrypoint")
+            staging = getattr(shell, "_ZIP_STAGING_EVENTS", None)
+            if not isinstance(staging, list):
+                raise ValueError("trusted shell staging ledger missing")
+            ordinary = getattr(shell, "_QUESTION_FAILURE_EVENTS", None)
+            fences = getattr(shell, "_MEDIA_FENCE_EVENTS", None)
+            if not isinstance(ordinary, list) or not isinstance(fences, list):
+                raise ValueError("trusted question/fence ledger missing")
+            observations.reconcile_skipped(expected_qids, staging, ordinary, fences)
+            for qid in receipt["input_validation"].get("unresolved_qIDs", []):
+                if not any(event["qID"] == qid for event in observations.question_failures):
+                    observations.record_question_failure({"qID": qid, "stage": "input_media_resolution", "reason": "No unambiguous media candidate for request qID", "producer": {"function": "validate_inputs", "request_sha256": receipt["input_validation"]["request_sha256"]}})
+            question_events = list(observations.question_failures)
             receipt["engine_observations"] = observations.finish(receipt["input_validation"]["requests"])
             if receipt["engine_observations"]["failures"]:
                 raise ValueError("original engine initialization or answer failure")
         verify_sources(contract)
         if validate_inputs(input_root) != receipt["input_validation"]:
             raise ValueError("input metadata or layout changed during backend execution")
-        receipt["output_validation"] = validate_outputs(output_root, input_root)
+        receipt["question_failure_output"] = blank_failed_outputs(output_root, input_root, question_events)
+        receipt["output_validation"] = validate_outputs(output_root, input_root, empty_qids={event["qID"] for event in question_events})
         check_input_binding(binding, policy)
         if guard.violations or receipt["degraded_markers"]:
             raise ValueError("backend degraded or attempted fixed configuration mutation")

platform_wrapper/probe_entrypoint.py

@@ -0,0 +1,1118 @@
+"""Submission I/O shell — the live Grand-Challenge PROCEDURE contract.
+
+The contract (live GC socket table, evidence-packet-sourced, never the template):
+
+    read   /input/request.json          Request rows (cardinality: see U2)
+    read   /input/FO_definitions.json   organizer FO taxonomy text (encoding: see ASM-4)
+    read   /input/batch-videos.zip      per-question clips, ZIP kind (layout: U1; delivery: U3)
+    write  /output/answer.json          list of {qID, content, latency} — the ONLY file
+                                        this shell ever leaves in /output
+
+ARMED MODES, REFUSAL OTHERWISE. This shell runs only under a SANCTIONED arming
+form and refuses everything else with exit code 2 and zero writes:
+
+    submission (production):  `--submission`  or  SURGFIELD_SUBMISSION_MODE=1
+                              — the container CMD's form; THIS is the upload path.
+    probe (rehearsal):        `--probe`       or  SURGFIELD_PROBE_MODE=1
+                              — interface rehearsals and gate fixtures; never
+                              uploaded as a production CMD.
+
+The two modes execute the identical contract; they differ only in the stderr
+arming stamp. Any invocation carrying NO sanctioned form — bare, wrong flags,
+misconfigured CMD — exits 2 before any I/O.
+
+Behavioral guarantees:
+  1. `/output/answer.json` exists with FULL qID coverage from the moment the
+     request set is readable: a fallback answer per qID is PRE-SEEDED and each
+     computed answer atomically replaces its entry (progressive flush). A job
+     killed mid-run still leaves a complete, monotonically-improved answer file.
+  2. `/output` contains exactly `answer.json` after a clean exit; scratch lives
+     in a private work directory (the atomic-replace temp is consumed by every
+     flush).
+  3. One `G1DECODE qID=<q> frames=<n> first_frame_sha256=<h> res=<WxH>` line per
+     qID on stderr — the decode-proof channel (external parse contract,
+     byte-stable). Hash = sha256 of the first decoded frame's raw bytes.
+     Decoding runs in a KILLABLE SUBPROCESS with a hard timeout so a wedged
+     video can never hang the job.
+  4. Time budget: the platform declares a 30 s/question unit rate with per-JOB
+     enforcement only (evidence pack, snapshot 2026-07-23T164642Z) —
+     operationally POOLED within a job. The shell budgets TOTAL job time and
+     logs pacing; it NEVER self-kills (the platform owns enforcement; a
+     self-kill could only forfeit answers the platform would have accepted).
+  5. No network. This shell downloads nothing; model weights resolve from the
+     image's local cache (revision pinning is a build-recipe property).
+
+Interface unknowns, handled defensively where marked in-code:
+  U1    zip internal tree (flat / plain/ / nested / .MP4 / videoID-keyed)
+  U2    request.json cardinality (list vs single object)
+  U3    whether GC auto-extracts ZIP-kind sockets
+  ASM-4 FO_definitions.json encoding (JSON string vs structured JSON)
+
+Supported invocation forms:
+
+    surgfield-entrypoint --submission ...                      (dev/rehearsal)
+    surgfield-probe --probe ...                                (rehearsal)
+    python -m solution.entrypoint.probe_entrypoint --submission|--probe ...
+
+P9 (2026-07-31): the PRODUCTION container CMD is NOT this console script — it is the
+stdlib-only import-death-proof launcher `python -m surgfield_shell.launch --submission`
+(docker/Dockerfile.p4 ENTRYPOINT), which writes the never-empty floor BEFORE importing
+this engine. The forms above eager-import the engine (F1 death risk) and are for
+dev/rehearsal only.
+
+Direct file-path invocation (`python .../probe_entrypoint.py`) is NOT a
+supported form — the engine package is not importable from that sys.path.
+Once armed, the outermost coverage net in `main()` guarantees a full-coverage
+`answer.json` (loudly-degraded fallbacks, never a bare /output) under that
+misuse or any other post-arming failure.
+"""
+from __future__ import annotations
+
+import argparse
+from contextlib import ExitStack
+import json
+import traceback
+import math
+import os
+import re
+import subprocess
+import sys
+import tempfile
+import time
+import zipfile
+from pathlib import Path
+
+# The DECLARED unit rate (track page: "30 seconds per question"), applied as a
+# POOLED per-job budget per the evidence pack's operational verdict. Monitoring
+# only — see guarantee 4.
+POOLED_SEC_PER_QUESTION = 30.0
+
+# Hard ceiling for one first-frame decode subprocess. Generous against real
+# clips (a first-frame read is sub-second); tight against a wedged demuxer.
+DECODE_TIMEOUT_SEC = 20.0
+
+# P1-04: total wall budget for the pre-answer decode-proof phase (cold cv2
+# subprocesses). Bounds the O(N) probe cost inside the platform's setup window;
+# past it, remaining clips skip the probe (the in-process decode is still
+# SIGALRM-bounded per P1-03, so coverage is not a safety dependency).
+PROBE_PHASE_BUDGET_SEC = 60.0
+
+# Trusted structured events; only raw-ZIP materialization failures populate it.
+_ZIP_STAGING_EVENTS: list[dict] = []
+_QUESTION_FAILURE_EVENTS: list[dict] = []
+_MEDIA_FENCE_EVENTS: list[dict] = []
+
+
+def _log(msg: str) -> None:
+    """All shell logging goes to stderr: contract-neutral (GC ignores it,
+    harnesses parse it), and it keeps stdout free of accidental output."""
+    print(msg, file=sys.stderr, flush=True)
+
+
+def _safe_float(value, default: float = 0.0) -> float:
+    """Never-raise numeric coercion for request-row fields. A malformed field
+    costs AT MOST its own row's fidelity — never the run (the B2-19-1 class:
+    field access outside a containment must not be able to raise). inf/nan
+    clamp to the default: a non-finite window boundary is malformed, not big."""
+    try:
+        f = float(value)
+        return f if math.isfinite(f) else default
+    except (TypeError, ValueError):
+        return default
+
+
+# Whole-video time-question marker (the Leg-B observability population): the
+# boundary SEMANTICS stay held to the platform observer — the shell only
+# refuses to hide the rows.
+_TIME_Q = re.compile(
+    r"how long|how much time|at what time|when (was|is|did|does)|hh:mm:ss", re.I)
+
+
+# ── request.json ─────────────────────────────────────────────────────────
+
+def read_requests(path: Path) -> list[dict]:
+    """Parse request.json into a list of row dicts.
+
+    U2 (cardinality UNKNOWN): the helptext suggests a single object, the
+    template a list. Accept BOTH: a dict is wrapped into a one-row list, a
+    list is taken as-is. The answer file is always emitted as a LIST — the
+    single-object output mirror is unconfirmed, and a list is the shape the
+    grader's `load_items` reads (observed tolerant, not contract-asserted).
+    """
+    raw = json.loads(path.read_text(encoding="utf-8"))
+    if isinstance(raw, dict):
+        return [raw]
+    if isinstance(raw, list):
+        rows = [r for r in raw if isinstance(r, dict)]
+        if len(rows) != len(raw):
+            _log(f"PROBESHELL: skipping {len(raw) - len(rows)} non-object "
+                 f"request row(s)")
+        return rows
+    raise ValueError(f"request.json: expected object or list, got {type(raw).__name__}")
+
+
+# ── FO_definitions.json ──────────────────────────────────────────────────
+
+def stage_fo_definitions(path: Path, workdir: Path) -> Path | None:
+    """Decode the organizer FO-definitions socket to a plain-text file and
+    return its path (for SURGFIELD_FO_DEFINITIONS_FILE), or None if absent.
+
+    ASM-4 (string-primary, template-observed): the value is expected to be a
+    JSON-encoded plain-text STRING. Tolerate structured JSON too — serialized
+    back to text so the taxonomy's heading parser sees SOMETHING rather than
+    the shell failing. Never fatal: on any error the engine falls back to its
+    built-in class floor.
+    """
+    if not path.is_file():
+        return None
+    try:
+        raw = json.loads(path.read_text(encoding="utf-8"))
+        text = raw if isinstance(raw, str) else json.dumps(raw, indent=1)
+        out = workdir / "FO_definitions.txt"
+        out.write_text(text, encoding="utf-8")
+        return out
+    except (ValueError, OSError) as e:
+        _log(f"PROBESHELL: FO_definitions unusable ({type(e).__name__}: {e}); "
+             f"engine uses its built-in taxonomy floor")
+        return None
+
+
+# ── batch videos ─────────────────────────────────────────────────────────
+
+# Video containers we recognize as clips. Lower-cased suffix match; the
+# resolvers below also honor .MP4 (upper) via the engine's resolve_clip.
+_VIDEO_EXTS = (".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v", ".mpg", ".mpeg")
+
+
+def _log_input_tree(input_dir: Path) -> None:
+    """DIAGNOSTIC (log-only, no behavior change): dump the full /input tree so
+    the NEXT live try-out definitively reveals GC's real batch-videos layout.
+    One `PROBESHELL: /input tree: ...` line per file (path + size). Total by
+    construction — any failure here degrades to a single note, never a raise."""
+    try:
+        files = sorted((p for p in input_dir.rglob("*") if p.is_file()),
+                       key=lambda p: str(p))
+        _log(f"PROBESHELL: /input tree: {len(files)} file(s) under {input_dir}")
+        for p in files:
+            try:
+                size = p.stat().st_size
+            except OSError:
+                size = -1
+            _log(f"PROBESHELL: /input tree:   {p} ({size}B)")
+    except Exception as e:
+        _log(f"PROBESHELL: /input tree unavailable ({type(e).__name__}: {e})")
+
+
+def _extract_zip(zpath: Path, dest: Path) -> None:
+    """Extract a ZIP into `dest` under the zip-slip guard.
+
+    SYMLINK-SAFETY IS LOAD-BEARING HERE: stdlib zipfile.extract writes every
+    member as a REGULAR FILE — it never honors a symlink attribute — so a
+    malicious link member cannot escape `dest`. If this is ever swapped for
+    another extractor (shutil.unpack_archive, an external tool), that guarantee
+    disappears and the escape reopens; re-prove before swapping."""
+    dest.mkdir(parents=True, exist_ok=True)
+    skipped = 0
+    with zipfile.ZipFile(zpath) as zf:
+        for info in zf.infolist():
+            name = info.filename
+            # zip-slip guard: refuse absolute paths and parent escapes.
+            if name.startswith(("/", "\\")) or ".." in Path(name).parts:
+                _log(f"PROBESHELL: skipping unsafe zip member {name!r}")
+                continue
+            # P1-19: PER-MEMBER isolation — a single bad member (BadZipFile / CRC
+            # / ENOSPC) must NOT propagate to the batch-wide except and zero the
+            # video signal for EVERY question. Log+skip the failing member; keep
+            # the rest of the archive.
+            try:
+                zf.extract(info, dest)
+            except Exception as e:
+                skipped += 1
+                _log(f"PROBESHELL: skipping unextractable zip member {name!r} "
+                     f"({type(e).__name__}: {e}); other members still staged")
+    if skipped:
+        _log(f"PROBESHELL: WARN {skipped} zip member(s) failed to extract "
+             f"from {zpath.name}; the rest were staged (per-member isolation)")
+
+
+def _is_zip(path: Path) -> bool:
+    """Content-sniff a regular file for the ZIP signature. Extension-agnostic —
+    GC has delivered the archive as `batch-videos` (no `.zip`), so we must NOT
+    rely on the name. `zipfile.is_zipfile` reads only the end-of-central-dir
+    record, so it is cheap even on a large non-zip (a real .mp4 => False)."""
+    try:
+        return zipfile.is_zipfile(path)
+    except OSError:
+        return False
+
+
+def stage_videos_root(input_dir: Path, workdir: Path) -> Path | None:
+    """Locate the clips ROBUSTLY across GC's actual placement: return a single
+    directory to search, or None if genuinely no videos are present.
+
+    Round-1/try-out RCA (2026-07-25): GC put the `batch-videos` input at a path
+    OTHER than `/input/batch-videos.zip` and OTHER than `/input/batch-videos/`,
+    so the old exact-path resolver deferred every question (frames=0). This
+    resolver instead scans `/input` RECURSIVELY and accepts, order-deterministic:
+
+      (a) ANY ZIP found anywhere under /input — matched by CONTENT, not name
+          (handles `batch-videos` with no extension, or a nested
+          `.../<uuid>/batch-videos.zip`) — extracted into the work dir; AND
+      (b) ANY loose video file (`*.mp4`/`*.avi`/…) anywhere under /input —
+          symlinked into the same work dir PRESERVING its path relative to
+          /input, so a `plain/` sub-tree is kept intact for the qID resolver.
+
+    Everything lands under one `<work>/videos` root (never /output); the
+    existing qID→clip resolver (`normalize_clips`) then runs over it unchanged.
+    The zip-slip guard is preserved (`_extract_zip`). If NOTHING resolves we
+    keep the prior defer-to-rules behavior — never crash, never a bare output.
+
+    Regression: the known-good contract (`/input/batch-videos.zip` with
+    `plain/<qID>.mp4`, the P6 harness staging) is a strict subset of case (a)
+    and stages to `<work>/videos/plain/<qID>.mp4` exactly as before.
+    """
+    dest = workdir / "videos"
+    files = sorted((p for p in input_dir.rglob("*") if p.is_file()),
+                   key=lambda p: str(p))
+    zips = []
+    loose_videos = []
+    for f in files:
+        if f.suffix.lower() in _VIDEO_EXTS:
+            loose_videos.append(f)
+        elif _is_zip(f):
+            zips.append(f)
+
+    staged_any = False
+    for z in zips:  # deterministic (files is sorted)
+        _log(f"PROBESHELL: extracting archive {z} ({z.stat().st_size}B)")
+        _extract_zip(z, dest)
+        staged_any = True
+    for v in loose_videos:  # deterministic (files is sorted)
+        rel = v.relative_to(input_dir)
+        link = dest / rel
+        link.parent.mkdir(parents=True, exist_ok=True)
+        if not link.exists():
+            link.symlink_to(v.resolve())
+            staged_any = True
+
+    if staged_any:
+        _log(f"PROBESHELL: staged videos root {dest} "
+             f"({len(zips)} archive(s), {len(loose_videos)} loose clip(s))")
+        return dest
+    _log("PROBESHELL: no batch-videos archive or video files found anywhere "
+         "under /input; video-dependent layers will defer on every question")
+    return None
+
+
+def _resolve_deterministic(root: Path, key: str):
+    """Recursive, ORDER-DETERMINISTIC clip lookup by stem == `key`, preferring a
+    `plain/` path over `overlayed/` (we tuned on plain), then the shortest path,
+    then lexicographic. Complements the engine's tolerant resolvers for the
+    arbitrarily-nested trees GC may hand us. Returns a Path or None."""
+    hits = [p for p in root.rglob("*")
+            if p.is_file() and p.stem == key and p.suffix.lower() in _VIDEO_EXTS]
+
+    def sort_key(p: Path):
+        parts = [seg.lower() for seg in p.parts]
+        return (0 if "plain" in parts else 1,
+                1 if "overlayed" in parts else 0,
+                len(p.parts), str(p))
+
+    hits.sort(key=sort_key)
+    return hits[0] if hits else None
+
+
+# qID→clip resolution ladder — DOCUMENTED layout first, then bounded fallbacks.
+# Each rung is labeled so the run log shows exactly which layout GC delivered
+# (observe-don't-assume: the try-out enumeration + these lines finally pin the
+# real tree). The template's reference `inference.py` reads `/input/plain/<qID>.mp4`
+# (+ `overlayed/` with a burned-in HH:MM:SS clock).
+# P9 (2026-07-31) — RCA drift corrected: `plain/` is the PRIMARY rung because WE
+# unzip the `batch-videos` ZIP into a plain layout (stage_videos_root), NOT because
+# the platform auto-extracts it. The team's OBSERVED_ON_PLATFORM note recorded the
+# OPPOSITE (the container sees the ZIP as a FILE; no auto-extraction), so the earlier
+# "platform auto-extracts to plain/" premise here was FALSE and is struck. The real
+# /input tree is still un-captured; `_log_input_tree` (see main) emits it verbatim on
+# the next live run so the ladder can finally be verified against ground truth. The
+# rest are known-plausible fallbacks in priority order.
+_RUNG_PRIMARY = "plain"          # documented primary:  <root>/plain/<qID>.mp4
+_RUNG_OVERLAYED = "overlayed"    # documented alt:       <root>/overlayed/<qID>.mp4
+_RUNG_FLAT = "flat"              # bare root:            <root>/<qID>.mp4
+_RUNG_NESTED = "nested-stem"     # recursive stem==qID (plain-preferring)
+_RUNG_VIDEOID = "videoID"        # keyed by row.videoID (qIDs may share a clip)
+_RUNG_UNRESOLVED = "UNRESOLVED"
+
+
+def _resolve_qid(root: Path, qid: str, video_id, resolve_video):
+    """Resolve ONE qID to a clip across the ordered ladder. Returns
+    `(Path|None, rung_label)`. Rungs are tried in strict priority order so the
+    documented layout always wins and any fallback is visible in the label.
+
+    The primary/overlayed/flat rungs reimplement the engine's `resolve_clip`
+    path checks SPLIT BY RUNG (so each is separately labeled); the P6 known-good
+    tree (`plain/<qID>.mp4`) resolves on rung 1 exactly as before — no mapping
+    change. Rungs 4–5 reuse the deterministic recursive + engine resolvers.
+    """
+    q = str(qid)
+    for ext in (".mp4", ".MP4"):
+        c = root / "plain" / f"{q}{ext}"
+        if c.is_file():
+            return c, _RUNG_PRIMARY
+    for ext in (".mp4", ".MP4"):
+        c = root / "overlayed" / f"{q}{ext}"
+        if c.is_file():
+            return c, _RUNG_OVERLAYED
+    for ext in (".mp4", ".MP4"):
+        c = root / f"{q}{ext}"
+        if c.is_file():
+            return c, _RUNG_FLAT
+    c = _resolve_deterministic(root, q)
+    if c is not None:
+        return c, _RUNG_NESTED
+    if video_id:
+        vid = str(video_id)
+        c = _resolve_deterministic(root, vid)
+        if c is not None and Path(c).is_file():
+            return c, _RUNG_VIDEOID
+    return None, _RUNG_UNRESOLVED
+
+
+def normalize_clips(rows: list[dict], root: Path | None, workdir: Path) -> Path:
+    """Build `<work>/norm/plain/<qID>.mp4` links so the engine's clip-mode
+    lookup is deterministic regardless of the delivered tree, resolving each
+    qID through the labeled ladder in `_resolve_qid`.
+
+    OBSERVABILITY (robust-with-visibility, non-negotiable): one `RESOLVE` line
+    per qID naming the rung that fired; a loud `RESOLVE-FALLBACK` when a qID
+    resolves past the documented primary; a `RESOLVE-SUMMARY` rung histogram;
+    and a loud `RESOLVE-FAIL` naming every qID with NO clip. An unresolved qID
+    still gets no link — the engine defers and the rule layers answer it — but
+    it is never silent (Round-1 died on silent tolerant-guessing).
+    """
+    from solution.frame_sampler import resolve_video
+
+    norm = workdir / "norm" / "plain"
+    norm.mkdir(parents=True, exist_ok=True)
+    if root is None:
+        _log("PROBESHELL: RESOLVE-SUMMARY no videos root; every qID defers to rules")
+        return norm.parent
+
+    rung_counts: dict[str, int] = {}
+    unresolved: list[str] = []
+    # P1-18: qIDs resolved via the videoID rung map to a SHARED, un-trimmed
+    # video (not a per-qID clip). Clip-mode whole_clips from frame 0 and would
+    # give every qID sharing that video identical, wrong-window frames. Record
+    # them so the engine SEEKS the row's [start,end] instead of whole-clipping.
+    videoid_qids: list[str] = []
+    for row in rows:
+        qid = str(row.get("qID", ""))
+        if not qid:
+            continue
+        src, rung = _resolve_qid(root, qid, row.get("videoID"), resolve_video)
+        rung_counts[rung] = rung_counts.get(rung, 0) + 1
+        if src is None:
+            unresolved.append(qid)
+            _log(f"PROBESHELL: RESOLVE qID={qid} rung={_RUNG_UNRESOLVED} "
+                 f"(no clip anywhere under videos root; defers to rules)")
+            continue
+        if rung == _RUNG_VIDEOID:
+            videoid_qids.append(qid)
+        if rung != _RUNG_PRIMARY:
+            _log(f"PROBESHELL: RESOLVE-FALLBACK qID={qid} rung={rung} src={src} "
+                 f"(documented plain/{qid}.mp4 absent — resolved via fallback)")
+        else:
+            _log(f"PROBESHELL: RESOLVE qID={qid} rung={rung} src={src}")
+        link = norm / f"{qid}.mp4"
+        if not link.exists():
+            link.symlink_to(Path(src).resolve())
+
+    # P1-18: persist the videoID-rung set next to the normalized tree so the
+    # clip-mode engine can pick SEEK-with-window for exactly these qIDs.
+    marker = norm.parent / "videoid_rung.json"
+    try:
+        marker.write_text(json.dumps(sorted(videoid_qids)))
+    except OSError as e:
+        _log(f"PROBESHELL: could not write videoid_rung marker ({e})")
+    if videoid_qids:
+        _log(f"PROBESHELL: RESOLVE-VIDEOID {len(videoid_qids)} qID(s) share an "
+             f"un-trimmed video — engine will SEEK [start,end], not whole-clip")
+
+    _log(f"PROBESHELL: RESOLVE-SUMMARY rungs={rung_counts} "
+         f"resolved={sum(rung_counts.values()) - len(unresolved)} "
+         f"unresolved={len(unresolved)}")
+    if unresolved:
+        shown = unresolved[:20]
+        _log(f"PROBESHELL: RESOLVE-FAIL {len(unresolved)} qID(s) had NO clip: "
+             f"{shown}{' …' if len(unresolved) > 20 else ''} — video-blind on these")
+    return norm.parent
+
+
+# ── decode-proof channel ─────────────────────────────────────────────────
+
+# Self-contained first-frame prober run in a KILLABLE subprocess (production
+# hardening): cv2 can wedge inside C on a malformed container, beyond any
+# in-process timeout. The snippet prints one JSON line; anything else (crash,
+# hang->kill, no frame) degrades to the frames=0 record. The emitted G1DECODE
+# line stays byte-stable.
+#
+# P1-07: the child prints the sentinel ``CV2_INIT_OK`` on stderr AFTER cv2
+# imports+constructs successfully. That lets the parent tell a clip-specific
+# decoder crash (sentinel present, then rc!=0) from a SYSTEMIC child-env fault
+# (no sentinel: missing libGL, cv2 import failure, OOM-kill of the child) — the
+# latter fires for EVERY clip and must NOT unlink every clip (the Round-1
+# frames=0 collapse).
+_DECODE_SNIPPET = (
+    "import hashlib, json, sys\n"
+    "try:\n"
+    "    import cv2\n"
+    "    cap = cv2.VideoCapture(sys.argv[1])\n"
+    "except Exception:\n"
+    "    sys.exit(3)\n"
+    "sys.stderr.write('CV2_INIT_OK')\n"
+    "ok, frame = cap.read()\n"
+    "cap.release()\n"
+    "if ok and frame is not None:\n"
+    "    print(json.dumps({'sha': hashlib.sha256(frame.tobytes()).hexdigest(),\n"
+    "                      'res': str(frame.shape[1]) + 'x' + str(frame.shape[0])}))\n"
+)
+
+
+def _probe_first_frame(clip: Path) -> tuple[str, int, str, str | None]:
+    """(sha, frames, res, danger) for one clip via the killable subprocess prober.
+
+    `danger` classifies the prober outcome for the H6 in-process fence:
+      - "WEDGE"    → subprocess TimeoutExpired (the child hung on decode); the
+                     in-process engine would hang on the same clip.
+      - "CRASH"    → cv2 INITIALIZED (sentinel seen) then the child exited
+                     non-zero on THIS clip (segfault/OOM/abort in the decoder).
+      - "ENVFAULT" → rc!=0 with NO sentinel: the child could not even init cv2
+                     (P1-07 — a SYSTEMIC env fault, not a clip crash). Reported
+                     so the caller can ABORT fencing instead of unlinking every
+                     clip and going whole-pipeline video-blind.
+      - None       → SAFE: clean rc==0 (frame-0 decoded OR cleanly undecodable),
+                     an unparseable-stdout rc==0, or a subprocess spawn error.
+                     A clean frame-0 miss is NOT dangerous — frame_sampler skips
+                     an unreadable frame and answers from the rest.
+    """
+    try:
+        r = subprocess.run([sys.executable, "-c", _DECODE_SNIPPET, str(clip)],
+                           capture_output=True, text=True,
+                           timeout=DECODE_TIMEOUT_SEC)
+        if r.returncode != 0:
+            if "CV2_INIT_OK" not in (r.stderr or ""):
+                # P1-07: child never initialized cv2 → systemic env fault, not a
+                # clip-specific crash. Do NOT fence (would blind-unlink all clips).
+                _log(f"PROBESHELL: decode probe ENV-FAULT for {clip.name} "
+                     f"(rc={r.returncode}, cv2 never initialized); NOT fencing this clip")
+                return "none", 0, "0x0", "ENVFAULT"
+            _log(f"PROBESHELL: decode probe crashed for {clip.name} "
+                 f"(returncode={r.returncode}); recording frames=0 and fencing")
+            return "none", 0, "0x0", "CRASH"
+        if r.stdout.strip():
+            d = json.loads(r.stdout.strip().splitlines()[-1])
+            return d["sha"], 1, d["res"], None
+    except subprocess.TimeoutExpired:
+        _log(f"PROBESHELL: decode timeout ({DECODE_TIMEOUT_SEC:.0f}s) for "
+             f"{clip.name}; recording frames=0 and fencing")
+        return "none", 0, "0x0", "WEDGE"
+    except Exception as e:
+        _log(f"PROBESHELL: decode probe failed for {clip.name} "
+             f"({type(e).__name__}: {e})")
+    return "none", 0, "0x0", None
+
+
+def _fence_decision(to_fence: dict[str, str], n_probed: int):
+    """P1-07: decide which qIDs to fence, and whether the pattern is SYSTEMIC.
+
+    Returns ``(systemic, env_faults, dangers)``:
+      - ``env_faults`` — qIDs whose child never initialized cv2 (env fault).
+      - ``dangers``    — qIDs with a real per-clip WEDGE/CRASH.
+      - ``systemic``   — True if ANY env fault, OR every probed clip came back
+                         dangerous (>1). When systemic, the caller must NOT
+                         unlink clips (would blind the whole pipeline to the
+                         rules floor at exit 0); the in-process SIGALRM decode
+                         deadline provides containment instead.
+    """
+    env_faults = {q for q, r in to_fence.items() if r == "ENVFAULT"}
+    dangers = {q: r for q, r in to_fence.items() if r in ("WEDGE", "CRASH")}
+    systemic = bool(env_faults) or (n_probed > 1 and len(dangers) == n_probed)
+    return systemic, env_faults, dangers
+
+
+def emit_decode_records(rows: list[dict], norm_root: Path,
+                        probe_budget_sec: float | None = None, zip_batch=None
+                        ) -> tuple[dict[str, str], int]:
+    """One G1DECODE stderr line per UNIQUE qID — the decode-proof channel.
+
+    Hash convention (a parse contract with the external rehearsal harness —
+    never change it unilaterally): sha256 over the raw bytes of the FIRST
+    decoded frame (`frame.tobytes()`). `frames` counts what THIS probe decoded
+    (telemetry — the assertion compares the hash). Absent/undecodable/timed-out
+    clips emit `frames=0 first_frame_sha256=none res=0x0` rather than nothing,
+    so coverage of the channel itself is checkable.
+
+    P1-04: each cold cv2 subprocess is charged nowhere in the pooled budget, so
+    this phase (a) probes each unique qID ONCE (no per-duplicate-row re-decode),
+    (b) caches by resolved clip target so qIDs SHARING a clip decode once, and
+    (c) stops probing once `probe_budget_sec` is exhausted (remaining clips are
+    still safe — the in-process decode is SIGALRM-bounded per P1-03).
+
+    Returns `(to_fence, n_probed)`:
+      - `to_fence` = {qID: "WEDGE"|"CRASH"|"ENVFAULT"} — qIDs whose prober hit a
+        signal. The caller fences WEDGE/CRASH; ENVFAULT (systemic, P1-07) is
+        surfaced but NOT unlinked.
+      - `n_probed` = number of clips actually decoded this phase (for the caller's
+        systemic-pattern guard: all-probed-clips-dangerous ⇒ do not fence).
+    """
+    if zip_batch is not None:
+        return _emit_zip_decode_records(rows, norm_root, zip_batch, probe_budget_sec)
+    to_fence: dict[str, str] = {}
+    seen_q: set[str] = set()
+    by_target: dict[str, tuple] = {}
+    n_probed = 0
+    t0 = time.perf_counter()
+    budget_logged = False
+    for row in rows:
+        qid = str(row.get("qID", ""))
+        if not qid or qid in seen_q:
+            continue
+        seen_q.add(qid)
+        clip = norm_root / "plain" / f"{qid}.mp4"
+        sha, frames, res, danger = ("none", 0, "0x0", None)
+        if clip.is_file():
+            key = os.path.realpath(clip)
+            if key in by_target:                       # shared clip: decode once
+                sha, frames, res, danger = by_target[key]
+            elif (probe_budget_sec is not None
+                  and time.perf_counter() - t0 > probe_budget_sec):
+                if not budget_logged:
+                    budget_logged = True
+                    _log(f"PROBESHELL: decode-proof budget ({probe_budget_sec:.0f}s) "
+                         f"exhausted; skipping probe for remaining clips (the "
+                         f"in-process decode is SIGALRM-bounded, so still safe)")
+            else:
+                sha, frames, res, danger = _probe_first_frame(clip)
+                by_target[key] = (sha, frames, res, danger)
+                n_probed += 1
+        if danger:
+            to_fence[qid] = danger
+        _log(f"G1DECODE qID={qid} frames={frames} first_frame_sha256={sha} res={res}")
+    return to_fence, n_probed
+
+
+# ── answers: pre-seed, compute, flush ────────────────────────────────────
+
+def _record_zip_failure(zip_batch, qid, phase, error):
+    event = zip_batch.record_failure(qid, phase, error)
+    _log("ZIP_STAGING_FAILURE " + json.dumps(event, sort_keys=True))
+
+
+def _emit_zip_decode_records(rows, norm_root, zip_batch, probe_budget_sec):
+    from solution.entrypoint.segment_zip_input import MemberStagingError
+    to_fence, seen_q, by_target = {}, set(), {}
+    n_probed, t0, budget_logged = 0, time.perf_counter(), False
+    for row in rows:
+        qid = str(row.get("qID", ""))
+        if not qid or qid in seen_q:
+            continue
+        seen_q.add(qid)
+        sha, frames, res, danger = "none", 0, "0x0", None
+        key = zip_batch.cache_key(qid)
+        if key in by_target:
+            sha, frames, res, danger = by_target[key]
+        elif probe_budget_sec is not None and time.perf_counter() - t0 > probe_budget_sec:
+            if not budget_logged:
+                _log(f"PROBESHELL: decode-proof budget ({probe_budget_sec:.0f}s) exhausted; remaining ZIP clips unprobed")
+                budget_logged = True
+        else:
+            with ExitStack() as clips:
+                try:
+                    clip = clips.enter_context(zip_batch.member(qid, phase="startup_probe"))
+                except MemberStagingError as error:
+                    _record_zip_failure(zip_batch, qid, "startup_probe", error)
+                else:
+                    # Exception classification ended before invoking the decoder.
+                    if clip.is_file():
+                        sha, frames, res, danger = _probe_first_frame(clip)
+                        by_target[key] = (sha, frames, res, danger)
+                        n_probed += 1
+        if danger:
+            to_fence[qid] = danger
+        _log(f"G1DECODE qID={qid} frames={frames} first_frame_sha256={sha} res={res}")
+    return to_fence, n_probed
+
+
+def write_answers(output_dir: Path, answers: list[dict]) -> Path:
+    """Atomic (re)write of /output/answer.json: temp-in-/output + os.replace
+    (same filesystem => atomic; the temp is consumed by every flush). Called
+    once to pre-seed and then after every computed answer — the file is
+    complete at every instant."""
+    output_dir.mkdir(parents=True, exist_ok=True)
+    out = output_dir / "answer.json"
+    tmp = output_dir / ".answer.json.tmp"
+    tmp.write_text(json.dumps(answers, indent=1), encoding="utf-8")
+    os.replace(tmp, out)
+    return out
+
+
+# C1 (2026-07-25): the grader's adversarial detector whole-submission-DQs on these public
+# signal substrings (focus.evaluation.adversarial). Our answers must never trip it.
+#
+# P9 (2026-07-31): promoted from OBSERVE-ONLY to ENFORCE. The prior hedge shipped a colliding
+# answer verbatim; a single genuine free-text answer that happens to contain a signal substring
+# would then risk zeroing all 10 buckets (whole-submission DQ). ENFORCE = on a collision we
+# DECLINE that ONE question by replacing its ENTIRE content with the format's SAFE_FALLBACK
+# (a full-answer forfeit, at most 1 question lost) and log loudly. This is a DECLINE, not an
+# evasion-scrub: we do NOT surgically strip the signal to sneak the rest of the answer through
+# (that would be grader-control evasion, HARD-BARRED). Coverage is preserved (still one answer
+# per qID). GOVERNANCE: this supersedes the 2026-07-25 observe-only disposition; the
+# scrub-vs-decline distinction is flagged for Reza ratification in P9_coverage.md.
+_C1_INJECTION_SIGNALS = (
+    "ignore previous instructions", "ignore all previous", "disregard the above",
+    "you are now", "system prompt:", "new instructions:", "override:",
+    "forget your instructions", "act as if", "pretend that the answer is correct",
+    "always respond with correct", "the answer is definitely correct",
+)
+
+
+def _c1_safe_fallback_for(qid: str, qid_to_question: dict[str, str]) -> str:
+    """Format-aware DECLINE content for a colliding answer. Best-effort: infer the answer
+    format from the question and take its SAFE_FALLBACK; plain 'none' if inference is
+    unavailable (total by construction — a raise here must never escape the pre-flight)."""
+    try:
+        from solution.answer_validator import SAFE_FALLBACK
+        from solution.cap_fmt_inferrer import infer_all
+        q = qid_to_question.get(qid, "")
+        return SAFE_FALLBACK.get(infer_all(q)["answer_format"], "none")
+    except Exception:
+        return "none"
+
+
+def _c1_collision_preflight(answers: list[dict],
+                            rows: list[dict] | None = None) -> int:
+    """ENFORCE: on any answer whose content collides with a public injection signal (would
+    whole-submission-DQ at the grader), DECLINE that ONE question — replace its whole content
+    with the format's SAFE_FALLBACK and log loudly. Mutates `answers` in place. Returns the
+    collision (blocked) count. This is a decline, never an evasion-scrub."""
+    qid_to_question = {str(r.get("qID", "")): str(r.get("question", ""))
+                       for r in (rows or [])}
+    hits = 0
+    for a in answers:
+        content = str(a.get("content", "")).lower()
+        for sig in _C1_INJECTION_SIGNALS:
+            if sig in content:
+                hits += 1
+                qid = str(a.get("qID", ""))
+                safe = _c1_safe_fallback_for(qid, qid_to_question)
+                _log(f"C1-COLLISION-BLOCKED qID={qid} signal={sig!r} "
+                     f"(answer would risk whole-submission DQ; DECLINED → content={safe!r})")
+                a["content"] = safe
+                break
+    if hits:
+        _log(f"C1-COLLISION-BLOCKED: {hits} answer(s) collided with the public injection-signal "
+             f"list — DECLINED (content replaced with SAFE_FALLBACK; coverage preserved)")
+    return hits
+
+
+def fallback_answers(rows: list[dict]) -> list[dict]:
+    """Full-coverage fallback list that does NOT require the engine package.
+
+    The PRE-SEED payload and the outermost coverage net's payload: it must work
+    even when `solution.*` itself is unimportable (e.g. the unsupported
+    file-path invocation), so the format-aware fallback is best-effort and the
+    floor is a plain "none" per unique qID. Degradation is loud on stderr and
+    fully countable downstream — nothing is masked.
+    """
+    try:
+        from solution.answer_validator import SAFE_FALLBACK
+        from solution.cap_fmt_inferrer import infer_all
+
+        def content_for(question: str) -> str:
+            # Total by construction: this executes INSIDE the outermost except
+            # handler, where a raise would escape the net entirely. The
+            # format-aware attempt is best-effort; "none" is unconditional.
+            try:
+                return SAFE_FALLBACK.get(infer_all(question)["answer_format"], "none")
+            except Exception:
+                return "none"
+    except Exception:
+        def content_for(question: str) -> str:
+            return "none"
+    answers, seen = [], set()
+    for row in rows:
+        qid = str(row.get("qID", ""))
+        if not qid or qid in seen:
+            continue
+        seen.add(qid)
+        answers.append({"qID": qid,
+                        "content": content_for(str(row.get("question", ""))),
+                        "latency": 0.0})
+    return answers
+
+
+def answer_all(rows: list[dict], videos_root: Path, device: str,
+               answers: list[dict], output_dir: Path,
+               proc_t0: float | None = None, zip_batch=None) -> list[dict]:
+    """Upgrade the pre-seeded `answers` in place — one engine answer per unique
+    qID — flushing atomically after every question.
+
+    Failure containment is per-question: an ordinary exception inside one
+    answer records its origin and writes empty content for only that question. Latency
+    is the full per-question wall (infer -> answer -> validate).
+
+    Pooled-budget monitor (guarantee 4): total = 30 s x N; an overrun is logged
+    LOUDLY once and the run CONTINUES — enforcement is the platform's, and the
+    progressive flush means a platform kill costs only the not-yet-upgraded
+    answers, never the file.
+    """
+    from solution.answer_validator import validate
+    from solution.cap_fmt_inferrer import infer_all
+    from solution.container import SubmissionContainer
+    from solution.question_context import Request
+
+    # P1-20: time the eager model/detector load (the largest setup sink) so a
+    # cold-start init overrun is visible and attributable, not silent.
+    _init_t0 = time.perf_counter()
+    container = SubmissionContainer(videos_dir=videos_root, device=device)
+    _init_wall = time.perf_counter() - _init_t0
+    _log(f"PROBESHELL: CONTAINER-INIT {_init_wall:.1f}s (eager model/detector load)"
+         + (f"; total setup {time.perf_counter() - proc_t0:.1f}s from process start"
+            if proc_t0 is not None else ""))
+    slot = {a["qID"]: i for i, a in enumerate(answers)}
+
+    budget = POOLED_SEC_PER_QUESTION * len(answers)
+    job_t0 = time.perf_counter()
+    budget_breached = False
+    done: set[str] = set()
+    for row in rows:
+        qid = str(row.get("qID", ""))
+        if not qid or qid in done or qid not in slot:
+            if qid and qid in done:
+                _log(f"PROBESHELL: skipping duplicate qID row ({qid!r})")
+            continue
+        done.add(qid)
+        elapsed = time.perf_counter() - job_t0
+        if elapsed > budget and not budget_breached:
+            budget_breached = True
+            _log(f"PROBESHELL: POOLED BUDGET EXCEEDED ({elapsed:.0f}s > "
+                 f"{budget:.0f}s for {len(answers)} questions) — continuing; "
+                 f"the platform owns enforcement and the answer file is "
+                 f"already complete at every instant")
+        t0 = time.perf_counter()
+        question = str(row.get("question", ""))
+        # Same B2-19-1 class, sibling site: these run OUTSIDE the per-question
+        # try — a raising coercion here would degrade the WHOLE run via the
+        # outer net instead of costing one row. Never-raise by construction.
+        start = _safe_float(row.get("start_time", 0.0) or 0.0)
+        end = _safe_float(row.get("end_time", 0.0) or 0.0)
+        try:
+            scratch = infer_all(question, start_time=start, end_time=end)
+            # H5: prefer an answer_format EXPOSED on the request over the inferred
+            # one. The observed live contract does NOT carry this field (the
+            # Request schema is qID/videoID/start/end/procedure_type/question),
+            # so on real inputs this is byte-identical to inference — a defensive
+            # guard that honors a true format if a future contract exposes it,
+            # avoiding the misroute the H5 validator policy otherwise absorbs.
+            exposed_fmt = row.get("answer_format")
+            fmt = exposed_fmt if isinstance(exposed_fmt, str) and exposed_fmt \
+                else scratch.get("answer_format", "open_ended")
+            req = Request(
+                qID=qid,
+                videoID=str(row.get("videoID", "")),
+                start_time=start,
+                end_time=end,
+                procedure_type=str(row.get("procedure_type", "")),
+                question=question,
+            )
+            if zip_batch is None:
+                resp = container.answer(req, scratch_init=scratch)
+            else:
+                from solution.entrypoint.segment_zip_input import MemberStagingError
+                if zip_batch.failed(qid):
+                    answers[slot[qid]] = {"qID": qid, "content": "", "latency": time.perf_counter() - t0}
+                    write_answers(output_dir, answers)
+                    continue
+                with ExitStack() as clips:
+                    try:
+                        clips.enter_context(zip_batch.member(qid))
+                    except MemberStagingError as error:
+                        _record_zip_failure(zip_batch, qid, "answer", error)
+                        answers[slot[qid]] = {"qID": qid, "content": "", "latency": time.perf_counter() - t0}
+                        write_answers(output_dir, answers)
+                        continue
+                    # Close all loader FDs BEFORE the lease unlinks its clip.
+                    # This callback is ZIP-only; the original loose path is unchanged.
+                    if container.video_loader is not None:
+                        clips.callback(container.video_loader.close_all)
+                    # Model exceptions occur outside the materialization catch.
+                    resp = container.answer(req, scratch_init=scratch)
+            content = validate(resp.content, fmt)
+            answers[slot[qid]] = {"qID": qid, "content": content,
+                                  "latency": time.perf_counter() - t0}
+        except Exception as e:
+            from solution.entrypoint.segment_zip_input import ZipIntegrityError
+            if isinstance(e, ZipIntegrityError):
+                raise
+            event = {"qID": qid, "stage": "answer_loop_exception", "exception_type": type(e).__name__,
+                     "exception_module": type(e).__module__, "message": str(e), "output_policy": "empty_content",
+                     "traceback": [{"path": t.filename, "function": t.name, "line": t.lineno} for t in traceback.extract_tb(e.__traceback__)]}
+            _QUESTION_FAILURE_EVENTS.append(event)
+            _log("QUESTION_FAILURE " + json.dumps(event, sort_keys=True))
+            answers[slot[qid]] = {"qID": qid, "content": "", "latency": time.perf_counter() - t0}
+        write_answers(output_dir, answers)
+    _log(f"PROBESHELL: job wall={time.perf_counter() - job_t0:.1f}s "
+         f"pooled-budget={budget:.0f}s n={len(answers)}")
+    # P2-06: emit the per-layer latency/attribution summary at end of run (it was
+    # accumulated every question but never reported — the only instrument for
+    # diagnosing which layer won and where budget was burned). Never fatal.
+    try:
+        container.metrics.report()
+    except Exception as e:
+        _log(f"PROBESHELL: metrics report skipped (non-fatal {type(e).__name__}: {e})")
+    return answers
+
+
+# ── entry ────────────────────────────────────────────────────────────────
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser(
+        description="solution submission shell (live GC PROCEDURE contract)")
+    ap.add_argument("--submission", action="store_true",
+                    help="arm SUBMISSION mode — the sanctioned production CMD "
+                         "form (equivalent env: SURGFIELD_SUBMISSION_MODE=1)")
+    ap.add_argument("--probe", action="store_true",
+                    help="arm PROBE mode for interface rehearsal (equivalent "
+                         "env: SURGFIELD_PROBE_MODE=1)")
+    ap.add_argument("--input-dir", default="/input")
+    ap.add_argument("--output-dir", default="/output")
+    ap.add_argument("--device", default="cuda",
+                    help="engine device (default: cuda); in a CPU-only context "
+                         "the detector load fails with one benign WARN and the "
+                         "detection layers defer — never a crash")
+    args = ap.parse_args(argv)
+
+    # P1-20: capture a process-start monotonic BEFORE any staging/model load so
+    # the platform-scored ~120s setup window is measurable end-to-end (the shell
+    # previously only timed from inside answer_all, AFTER eager model loading —
+    # a cold-start overrun was invisible).
+    _proc_t0 = time.perf_counter()
+
+    # Refusal gate (design unchanged by the promotion): no sanctioned arming
+    # form, no run — exit 2 before any I/O. A misbuilt CMD cannot produce
+    # output it was never armed to produce.
+    submission = args.submission or os.environ.get("SURGFIELD_SUBMISSION_MODE") == "1"
+    probe = args.probe or os.environ.get("SURGFIELD_PROBE_MODE") == "1"
+    if not (submission or probe):
+        _log("PROBE-REFUSED: no sanctioned arming form present "
+             "(--submission|--probe, or SURGFIELD_SUBMISSION_MODE=1|"
+             "SURGFIELD_PROBE_MODE=1); nothing written. This shell is the "
+             "submission entrypoint ONLY when armed via a sanctioned form")
+        return 2
+    if submission:
+        _log("PROBESHELL: submission-mode armed (sanctioned production path)")
+    else:
+        _log("PROBESHELL: probe-mode armed (solution tree; interface rehearsal only)")
+
+    input_dir, output_dir = Path(args.input_dir), Path(args.output_dir)
+    _ZIP_STAGING_EVENTS.clear()
+    _QUESTION_FAILURE_EVENTS.clear()
+    _MEDIA_FENCE_EVENTS.clear()
+    zip_batch = None
+    # P9 (2026-07-31): tempfile.mkdtemp() is deferred to INSIDE the coverage net
+    # below (it is only needed for video staging). On a read-only/absent /tmp — the
+    # exact failsafe scenario, GC /tmp semantics UNCONFIRMED — the old top-of-main
+    # mkdtemp raised BEFORE the format-aware pre-seed, shipping the launcher's floor
+    # instead of the strictly-better format-aware fallback that needs no /tmp.
+
+    try:
+        rows = read_requests(input_dir / "request.json")
+    except (OSError, ValueError) as e:
+        # P9: the launcher (surgfield_shell.launch) already wrote a full-coverage floor
+        # to /output BEFORE this shell ran. Without a readable request set there is no
+        # qID universe to re-cover, so do NOT clobber that good floor with an empty '[]'
+        # (a latent re-introduction of the empty-answer regression the floor exists to
+        # prevent, if the launcher's read succeeded where this re-read raised). Only write
+        # '[]' when no answer.json exists at all — the socket is still never left bare.
+        _log(f"PROBESHELL: request.json unreadable ({type(e).__name__}: {e})")
+        ans = output_dir / "answer.json"
+        if ans.exists():
+            _log("PROBESHELL: preserving the existing launcher floor on disk "
+                 "(NOT clobbering with [])")
+        else:
+            write_answers(output_dir, [])
+            _log("PROBESHELL: no prior floor on disk — wrote empty [] "
+                 "(output socket non-bare)")
+        return 1
+
+    # PRE-SEED (guarantee 1): full-coverage fallback answers land in
+    # /output/answer.json before any engine work — a kill at any later instant
+    # still leaves a complete file. This lands BEFORE mkdtemp so a /tmp failure
+    # cannot foreclose the format-aware floor.
+    answers = fallback_answers(rows)
+    write_answers(output_dir, answers)
+    _log(f"PROBESHELL: pre-seeded {len(answers)} fallback answers")
+
+    # OUTERMOST COVERAGE NET (post-arming): once the request set is readable,
+    # NO failure below — engine imports included, i.e. even the unsupported
+    # file-path invocation — may leave /output without a full-coverage
+    # answer.json. Everything between here and the final write is contained;
+    # degradation is loud, never silent.
+    try:
+        # P9: video-staging workdir, created HERE (inside the coverage net, AFTER the
+        # format-aware pre-seed). A read-only/absent /tmp now degrades to the pre-seeded
+        # floor via this net, never forecloses it.
+        workdir = Path(tempfile.mkdtemp(prefix="surgfield-shell-"))
+
+        # RUN-SCOPED ENV RESET (in-process isolation): a prior main() call in
+        # this interpreter may have left these pointing at a dead workdir —
+        # every run starts from a clean slate and sets only what ITS inputs
+        # justify.
+        os.environ.pop("SURGFIELD_FO_DEFINITIONS_FILE", None)
+        os.environ.pop("SURGFIELD_CLIP_INPUT", None)
+
+        # Zeroed-boundary telemetry (B2-19-1 cure: lives INSIDE the net AND the
+        # coercion is never-raise): a row whose end_time is 0.0 — or malformed
+        # and therefore coerced to 0.0 — gives every duration-gated layer
+        # duration 0; count it loudly, never silently, never fatally.
+        zeroed = sum(1 for r in rows
+                     if _safe_float(r.get("end_time", 0.0) or 0.0) == 0.0)
+        if zeroed:
+            _log(f"PROBESHELL: WARN {zeroed} request row(s) with end_time==0.0 "
+                 f"(zeroed boundary)")
+        # Whole-video time-questions with start==0 — the Leg-B population:
+        # covered by the engine, but never invisibly (semantics HELD to the
+        # platform observer; the shell records, it does not resolve).
+        whole_video = sum(
+            1 for r in rows
+            if _safe_float(r.get("start_time", 0.0) or 0.0) == 0.0
+            and _safe_float(r.get("end_time", 0.0) or 0.0) > 0.0
+            and _TIME_Q.search(str(r.get("question", ""))))
+        if whole_video:
+            _log(f"PROBESHELL: WARN {whole_video} whole-video time-question "
+                 f"row(s) with start_time==0.0 (boundary semantics recorded, "
+                 f"not resolved)")
+
+        # DIAGNOSTIC (log-only): dump the full /input tree so the next live
+        # try-out definitively reveals GC's real batch-videos placement.
+        _log_input_tree(input_dir)
+
+        fo_path = stage_fo_definitions(input_dir / "FO_definitions.json", workdir)
+        if fo_path is not None:
+            os.environ["SURGFIELD_FO_DEFINITIONS_FILE"] = str(fo_path)
+            _log(f"PROBESHELL: FO definitions staged ({fo_path.stat().st_size} bytes)")
+
+        # ADDITIVE track dispatch (STEP-2, 2026-08-04, FRAME/SEGMENT foundation):
+        # SURGFIELD_TRACK selects the input resolver. Default/'procedure' runs the
+        # UNCHANGED video ladder (stage_videos_root + normalize_clips) VERBATIM in
+        # the else — the FRAME/SEGMENT branches are entered ONLY when explicitly
+        # armed, so the PROCEDURE resolved-clips + answer.json are byte-identical
+        # (proven by solution.entrypoint.test_procedure_byte_identity). The env read
+        # is pure (no output side effect); track_inputs is imported LAZILY so a
+        # PROCEDURE run never even imports the FRAME/SEGMENT code.
+        _track = os.environ.get("SURGFIELD_TRACK", "procedure").strip().lower()
+        if _track == "segment":
+            from solution.entrypoint.segment_zip_input import maybe_zip_batch
+            zip_batch = maybe_zip_batch(rows, input_dir, workdir, _ZIP_STAGING_EVENTS,
+                receipt_sink=lambda record: _log("ZIP_MEMBER_RECEIPT " + json.dumps(record, sort_keys=True)))
+        if zip_batch is not None:
+            root, norm_root = None, zip_batch.norm_root
+            _log(f"PROBESHELL: bounded ZIP staging for {len(rows)} questions; one selected clip at a time")
+        elif _track in ("frame", "segment"):
+            from solution.entrypoint.track_inputs import stage_and_normalize_track
+            root, norm_root = stage_and_normalize_track(_track, rows, input_dir, workdir)
+        else:
+            root = stage_videos_root(input_dir, workdir)
+            norm_root = normalize_clips(rows, root, workdir)
+        # Both env writes (SURGFIELD_FO_DEFINITIONS_FILE above, CLIP_INPUT here)
+        # are RUN-SCOPED: reset at the top of this block, set for this run's
+        # duration, read per-call by the engine. In the container nothing
+        # survives the exit; in-process reuse is isolated by the reset.
+        os.environ["SURGFIELD_CLIP_INPUT"] = "1"  # per-qID clip semantics engine-side
+        to_fence, n_probed = emit_decode_records(
+            rows, norm_root, probe_budget_sec=PROBE_PHASE_BUDGET_SEC, zip_batch=zip_batch)
+        # H6 fence: a clip whose killable prober hit a WEDGE/CRASH is removed from
+        # the normalized tree so the in-process sampler never touches it; that qID
+        # defers to text/rule like an absent clip. (The in-process decode is ALSO
+        # SIGALRM-bounded now — P1-03 — so a wedge that slips past the probe still
+        # can't hang the job.) NEVER fenced on a clean frame-0 miss.
+        systemic, env_faults, dangers = _fence_decision(to_fence, n_probed)
+        if systemic:
+            _log("PROBESHELL: H6-FENCE ABORT — systemic decode fault "
+                 f"(env_faults={sorted(env_faults)[:10]}, "
+                 f"dangerous={len(dangers)}/{n_probed} probed); NOT unlinking clips "
+                 "(would blind the whole pipeline); relying on the in-process "
+                 "SIGALRM decode deadline for containment")
+        elif dangers:
+            _log("PROBESHELL: H6-FENCE unlinking "
+                 + ", ".join(f"{q}({r})" for q, r in sorted(dangers.items()))
+                 + " — in-process decode fenced; affected qIDs recorded for empty responses")
+            for qid in dangers:
+                _MEDIA_FENCE_EVENTS.append({"qID": qid, "stage": "startup_decoder_fence", "reason": dangers[qid]})
+                link = norm_root / "plain" / f"{qid}.mp4"
+                try:
+                    if link.is_symlink() or link.exists():
+                        link.unlink()
+                except OSError as error:
+                    from solution.entrypoint.segment_zip_input import ZipIntegrityError
+                    raise ZipIntegrityError("failed to fence dangerous decoder input") from error
+        _setup_wall = time.perf_counter() - _proc_t0
+        _log(f"PROBESHELL: SETUP-WALL {_setup_wall:.1f}s (arm→staging→decode-proof, "
+             f"before model load); platform setup window ~120s")
+        if _setup_wall > 90.0:
+            _log(f"PROBESHELL: WARN setup wall {_setup_wall:.0f}s is within 30s of "
+                 f"the ~120s platform setup window — cold-start overrun risk")
+        answers = answer_all(rows, norm_root, args.device, answers, output_dir,
+                             proc_t0=_proc_t0, zip_batch=zip_batch)
+    except Exception as e:
+        _log(f"PROBESHELL DEGRADED: shell/engine failure after arming "
+             f"({type(e).__name__}: {e}); the pre-seeded full-coverage "
+             f"answers stand for {len(rows)} rows")
+    finally:
+        if zip_batch is not None:
+            zip_batch.close()
+
+    try:
+        _c1_collision_preflight(answers, rows)  # C1 ENFORCE: DECLINE colliding answers (P9)
+    except Exception as e:                       # belt: the pre-flight must never crash the run
+        _log(f"C1 pre-flight skipped (non-fatal {type(e).__name__}: {e})")
+
+    # P2-14/P2-18: OUTPUT-HYGIENE self-DQ preflight (F0/DUP/COV/FMT/P1/P2/P3).
+    # OBSERVE-ONLY in-container (like C1): it LOGS a self-DQ risk so it is visible
+    # in the container log; it NEVER mutates or blocks (the never-empty coverage
+    # guarantee stands). The blocking pre-upload gate is the CLI wrapper
+    # `run_preflight_on_file` (exit-nonzero on findings) that Reza runs before an
+    # upload. Coverage is checked against the request qID universe; the FMT check
+    # uses the answer_format inferred/exposed per row. Never fatal here.
+    try:
+        from solution.cap_fmt_inferrer import infer_all
+        from solution.answer_validator import normalize_answer_format
+        from solution.entrypoint.output_hygiene_preflight import run_preflight
+        expected_qids = [str(r.get("qID", "")) for r in rows if r.get("qID", "")]
+        formats_by_qid: dict[str, str] = {}
+        for r in rows:
+            qid = str(r.get("qID", ""))
+            if not qid:
+                continue
+            exposed = r.get("answer_format")
+            try:
+                fmt = (normalize_answer_format(str(exposed)) if exposed
+                       else infer_all(str(r.get("question", "")))["answer_format"])
+            except Exception:
+                fmt = None
+            if fmt:
+                formats_by_qid[qid] = fmt
+        rep = run_preflight(answers, expected_qids=expected_qids,
+                            formats_by_qid=formats_by_qid)
+        _log(f"PROBESHELL: {rep.summary()}")
+        for f in rep.findings:
+            _log(f"  HYGIENE [{f.check}] qID={f.qID}: {f.reason} :: {f.detail}")
+    except Exception as e:                     # observe-only: never affect output
+        _log(f"output-hygiene pre-flight skipped (non-fatal {type(e).__name__}: {e})")
+
+    out = write_answers(output_dir, answers)
+    _log(f"PROBESHELL: wrote {out} ({len(answers)} answers)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

platform_wrapper/segment_zip_input.py

@@ -0,0 +1,345 @@
+"""Bounded SEGMENT ZIP staging. The bootstrap validates request/root custody first.
+Only the selected clip is materialized; the source ZIP remains on read-only input.
+"""
+from contextlib import contextmanager
+import errno
+import hashlib
+import json
+import os
+from pathlib import Path, PurePosixPath
+import stat
+import zipfile
+
+class ZipIntegrityError(RuntimeError):
+    """Archive/path/owned-state corruption must remain fatal for the invocation."""
+
+
+BLOCK = 1024 * 1024
+MEDIA = {".mp4", ".avi", ".mov", ".mkv", ".webm", ".m4v", ".mpg", ".mpeg"}
+
+
+def stamp(s):
+    return tuple(getattr(s, k) for k in ("st_dev", "st_ino", "st_mode", "st_uid", "st_gid", "st_size", "st_mtime_ns", "st_ctime_ns", "st_nlink"))
+
+
+def validate_members(infos):
+    files, directories = {}, set()
+    if len(infos) > 100000:
+        raise ZipIntegrityError("ZIP member-count bound")
+    for info in infos:
+        name = info.filename
+        trimmed = name[:-1] if info.is_dir() else name
+        parts = trimmed.split("/")
+        if not trimmed or "\\" in name or any(x in ("", ".", "..") or ":" in x or any(ord(c) < 32 for c in x) for x in parts):
+            raise ZipIntegrityError("unsafe ZIP name")
+        kind = stat.S_IFMT(info.external_attr >> 16)
+        if kind not in (0, stat.S_IFREG, stat.S_IFDIR) or info.flag_bits & 1:
+            raise ZipIntegrityError("special or encrypted ZIP member")
+        key = trimmed.casefold()
+        if info.is_dir():
+            if key in files:
+                raise ZipIntegrityError("file-directory collision")
+            directories.add(key)
+        else:
+            if key in files or key in directories:
+                raise ZipIntegrityError("duplicate or case-colliding ZIP member")
+            files[key] = info
+    for info in files.values():
+        if any(str(a).casefold() in files for a in PurePosixPath(info.filename).parents if str(a) != "."):
+            raise ZipIntegrityError("ancestor file collision")
+    return {info.filename: info for info in files.values()}
+
+
+def plan_rows(rows, members):
+    """Same strict selection groups as current R3 input validator.
+    It never rewrites request clocks, qIDs, question text or videoID.
+    """
+    media = [PurePosixPath(n) for n in members if PurePosixPath(n).suffix.lower() in MEDIA]
+    result, seen = [], set()
+    for row in rows:
+        qid = row["qID"]
+        if (not isinstance(qid, str) or not qid or qid in seen or qid in (".", "..")
+                or "/" in qid or "\\" in qid or any(ord(c) < 32 for c in qid)):
+            raise ZipIntegrityError("invalid or duplicate request qID")
+        seen.add(qid)
+        hits = [p for p in media if p.stem == qid]
+        groups = [[p for p in hits if p.parent == PurePosixPath(sub)] for sub in ("plain", "overlayed", ".")]
+        groups.append(hits)
+        groups.append([p for p in media if p.stem == row.get("videoID")])
+        index, selected = next(((i, g) for i, g in enumerate(groups) if g), (None, []))
+        if len(selected) > 1:
+            raise ZipIntegrityError("ambiguous selected media")
+        result.append({"qID": qid, "member": str(selected[0]) if selected else None, "videoid_rung": index == 4})
+    return result
+
+
+def _open_exclusive(path):
+    return os.fdopen(os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600), "wb")
+
+
+class MemberStagingError(Exception):
+    """Raised only by lease materialization, never by the model consumer."""
+    def __init__(self, details):
+        self.details = details
+        super().__init__(details["message"])
+
+
+def member_error(path, info, count, cause):
+    return MemberStagingError({"stage": "zip_member_staging", "archive": str(path),
+        "member": info.filename, "declared_bytes": info.file_size,
+        "written_bytes": count, "exception_type": type(cause).__name__,
+        "errno": getattr(cause, "errno", None), "message": str(cause)})
+
+
+class ZipLease:
+    def __init__(self, archive_path, root, *, max_working_bytes, reserve_bytes=64 * 1024 * 1024, receipt_sink=None):
+        self.path, self.root = Path(archive_path), Path(root)
+        self.limit, self.reserve = max_working_bytes, reserve_bytes
+        self.receipt_sink = receipt_sink
+        if type(self.limit) is not int or type(self.reserve) is not int or self.limit <= self.reserve or self.reserve < 0:
+            raise ZipIntegrityError("invalid working budget")
+        self.root.mkdir(mode=0o700, parents=False, exist_ok=False)
+        self.root_id = (self.root.stat().st_dev, self.root.stat().st_ino)
+        self.raw = None
+        self.zf = None
+        self.active = False
+        self.receipts = []
+        try:
+            fd = os.open(self.path, os.O_RDONLY | os.O_NOFOLLOW)
+            self.raw = os.fdopen(fd, "rb")
+            s = os.fstat(fd)
+            if not stat.S_ISREG(s.st_mode):
+                raise ZipIntegrityError("archive is not a regular file")
+            self.source = stamp(s)
+            self.check_source()
+            self.zf = zipfile.ZipFile(self.raw)
+            self.members = validate_members(self.zf.infolist())
+        except BaseException:
+            self.close()
+            raise
+
+    def check_source(self):
+        try:
+            return self._check_source()
+        except OSError as error:
+            raise ZipIntegrityError("archive/lease identity cannot be checked") from error
+
+    def _check_source(self):
+        if stamp(os.fstat(self.raw.fileno())) != self.source or stamp(self.path.lstat()) != self.source:
+            raise ZipIntegrityError("archive identity changed")
+        r = self.root.lstat()
+        if not stat.S_ISDIR(r.st_mode) or (r.st_dev, r.st_ino) != self.root_id:
+            raise ZipIntegrityError("owned lease directory replaced")
+
+    @contextmanager
+    def member(self, name, *, qid=None, phase=None):
+        if self.active:
+            raise ZipIntegrityError("only one media lease may be live")
+        self.check_source()
+        info = self.members[name]
+        available = os.statvfs(self.root)
+        free_bytes = available.f_bavail * available.f_frsize
+        if info.file_size > min(self.limit, free_bytes) - self.reserve:
+            cause = OSError(errno.ENOSPC, "member does not fit bounded scratch reserve")
+            raise member_error(self.path, info, 0, cause)
+        self.active = True
+        token = hashlib.sha256(name.encode("utf-8")).hexdigest()
+        part, final = self.root / (token + ".part"), self.root / (token + ".mp4")
+        owned = {}
+        count, digest, record = 0, hashlib.sha256(), None
+        try:
+            try:
+                with _open_exclusive(part) as out:
+                    s = part.lstat(); owned[part] = (s.st_dev, s.st_ino)
+                    with self.zf.open(info, "r") as src:
+                        while True:
+                            block = src.read(BLOCK)
+                            if not block:
+                                break
+                            if count + len(block) > info.file_size or count + len(block) + self.reserve > self.limit:
+                                raise OSError(errno.ENOSPC, "stream exceeds member or working budget")
+                            if out.write(block) != len(block):
+                                raise OSError(errno.EIO, "short write while staging media")
+                            digest.update(block)
+                            count += len(block)
+                if count != info.file_size:
+                    raise zipfile.BadZipFile("member size differs at full EOF")
+                self.check_source()
+                # Full ZipExtFile EOF validates CRC before the clip becomes visible.
+                part.chmod(0o400)
+                part.rename(final)
+                owned[final] = owned.pop(part)
+                name_bytes = name.encode("utf-8")
+                record = {"qID": qid, "phase": phase, "stage": "zip_member_staging",
+                    "member": name if len(name_bytes) <= 1024 else None,
+                    "member_name_bytes": len(name_bytes), "member_name_sha256": hashlib.sha256(name_bytes).hexdigest(),
+                    "archive_identity": list(self.source), "bytes": count,
+                    "sha256": digest.hexdigest(), "crc32": info.CRC, "full_member_eof": True,
+                    "file_allocated_bytes": final.stat().st_blocks * 512,
+                    "working_budget_bytes": min(self.limit, free_bytes), "free_bytes_before": free_bytes,
+                    "reserve_bytes": self.reserve, "cleanup_completed": False}
+                self.receipts.append(record)
+            except (zipfile.BadZipFile, EOFError, OSError) as cause:
+                if isinstance(cause, OSError) and cause.errno not in (errno.ENOSPC, errno.EIO):
+                    raise ZipIntegrityError("unclassified staging filesystem error") from cause
+                raise member_error(self.path, info, count, cause) from cause
+            yield final
+        finally:
+            self.active = False
+            for path, identity in owned.items():
+                try:
+                    s = path.lstat()
+                except FileNotFoundError:
+                    continue
+                except OSError as error:
+                    raise ZipIntegrityError("owned media cleanup identity unavailable") from error
+                if not stat.S_ISREG(s.st_mode) or (s.st_dev, s.st_ino) != identity:
+                    raise ZipIntegrityError("refusing cleanup of replaced media path")
+                try:
+                    path.unlink()
+                except OSError as error:
+                    raise ZipIntegrityError("owned media cleanup failed") from error
+            self.check_source()
+            if record is not None:
+                try:
+                    after = os.statvfs(self.root)
+                except OSError as error:
+                    raise ZipIntegrityError("post-cleanup filesystem observation failed") from error
+                record["free_bytes_after_unlink"] = after.f_bavail * after.f_frsize
+                record["cleanup_completed"] = True
+                record["source_identity_unchanged_after_use"] = True
+                if self.receipt_sink is not None:
+                    self.receipt_sink(dict(record))
+
+    def close(self):
+        if self.active:
+            raise ZipIntegrityError("cannot close with active lease")
+        try:
+            if self.raw is not None and not self.raw.closed and hasattr(self, "source"):
+                self.check_source()
+        finally:
+            if self.zf is not None:
+                self.zf.close()
+            if self.raw is not None:
+                self.raw.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *unused):
+        self.close()
+
+
+class ZipBatch:
+    """Invocation-owned union of admitted ZIP and loose paths.
+    Only one sequential question is leased at a time. No answer/model code here.
+    """
+    def __init__(self, rows, input_root, workdir, archives, loose, events, receipt_sink=None):
+        self.events, self.failures, self.stores = events, {}, []
+        self.active = False
+        self.norm_root = Path(workdir) / "norm"
+        self.plain = self.norm_root / "plain"
+        self.plain.mkdir(parents=True, exist_ok=False)
+        self.entries, union, case_names = {}, {}, set()
+        try:
+            for i, archive in enumerate(archives):
+                store = ZipLease(archive, Path(workdir) / ("zip-lease-" + str(i)),
+                    max_working_bytes=(1 << 63) - 1, receipt_sink=receipt_sink)
+                self.stores.append(store)
+                for name in store.members:
+                    if name.casefold() in case_names:
+                        raise ZipIntegrityError("ambiguous ZIP/loose staging path")
+                    case_names.add(name.casefold());union[name] = (store, name)
+            for file in loose:
+                name = file.relative_to(input_root).as_posix()
+                if name.casefold() in case_names:
+                    raise ZipIntegrityError("ambiguous ZIP/loose staging path")
+                case_names.add(name.casefold());union[name] = (None, file)
+            for name in union:
+                if any(str(a).casefold() in case_names for a in PurePosixPath(name).parents if str(a) != "."):
+                    raise ZipIntegrityError("cross-input ancestor file collision")
+            plan = plan_rows(rows, union)
+            for item in plan:
+                if item["member"] is None:
+                    link = self.plain / (item["qID"] + ".mp4")
+                    self.entries[item["qID"]] = (None, None, link, None)
+                    continue
+                store, member = union[item["member"]]
+                target = (store.root / (hashlib.sha256(member.encode("utf-8")).hexdigest() + ".mp4")
+                          if store is not None else Path(member).resolve())
+                link = self.plain / (item["qID"] + ".mp4")
+                link.symlink_to(target)
+                self.entries[item["qID"]] = (store, member, link, target)
+            (self.norm_root / "videoid_rung.json").write_text(json.dumps(sorted(x["qID"] for x in plan if x["videoid_rung"])))
+            self.plan = plan
+        except BaseException:
+            self.close()
+            raise
+
+    def cache_key(self, qid):
+        store, member, _, _ = self.entries[qid]
+        if member is None:
+            return ("missing", qid)
+        return (str(store.path), member) if store is not None else ("loose", str(member))
+
+    def failed(self, qid):
+        return qid in self.failures
+
+    def record_failure(self, qid, phase, error):
+        if type(error) is not MemberStagingError:
+            raise TypeError("only materialization failures may be recorded here")
+        event = {"qID": qid, "phase": phase, **error.details, "output_policy": "empty_content"}
+        self.events.append(event);self.failures[qid] = event
+        return event
+
+    @contextmanager
+    def member(self, qid, *, phase="answer"):
+        if self.active:
+            raise ZipIntegrityError("only one question media lease may be live")
+        self.active = True
+        try:
+            store, member, link, target = self.entries[qid]
+            if target is None:
+                raise MemberStagingError({"stage": "input_media_resolution", "archive": None,
+                    "member": None, "declared_bytes": None, "written_bytes": 0,
+                    "exception_type": "MissingMedia", "errno": None,
+                    "message": "no selected media for this request"})
+            # A startup H6 fence removes the normalized link. Never resurrect it.
+            if not link.is_symlink():
+                if link.exists():
+                    raise ZipIntegrityError("normalized lease link replaced by non-symlink")
+                yield link
+                return
+            if Path(os.readlink(link)) != target:
+                raise ZipIntegrityError("normalized lease target changed")
+            if store is None:
+                yield link
+            else:
+                with store.member(member, qid=qid, phase=phase):
+                    yield link
+        finally:
+            self.active = False
+
+    def close(self):
+        errors = []
+        for store in self.stores:
+            try:
+                store.close()
+            except Exception as error:
+                errors.append(error)
+        if errors:
+            raise errors[0]
+
+
+def maybe_zip_batch(rows, input_root, workdir, events, receipt_sink=None):
+    """Return None for loose-only input so its original staging path is untouched."""
+    files = sorted((x for x in Path(input_root).rglob("*") if x.is_file()), key=str)
+    archives, loose = [], []
+    for file in files:
+        if file.suffix.lower() in MEDIA:
+            loose.append(file)
+        elif zipfile.is_zipfile(file):
+            archives.append(file)
+    if not archives:
+        return None
+    return ZipBatch(rows, Path(input_root), Path(workdir), archives, loose, events, receipt_sink=receipt_sink)