perf(m49): defer fresh stage revalidation

src/k1link/observatory/m49_portable_source.py

@@ -599,16 +599,28 @@ def _materialize_stage(
             "authority": dict(_AUTHORITY),
         }
         manifest_path = staging / M49_PORTABLE_STAGE_MANIFEST
-        manifest_path.write_bytes(canonical_json(manifest))
+        manifest_payload = canonical_json(manifest)
+        manifest_path.write_bytes(manifest_payload)
+        manifest_sha256 = hashlib.sha256(manifest_payload).hexdigest()
         final = parent / f"{M49_PORTABLE_STAGE_PREFIX}{identity_sha256}"
         if final.exists():
             existing = validate_m49_portable_source_stage(final)
-            if existing.manifest_sha256 != hashlib.sha256(manifest_path.read_bytes()).hexdigest():
+            if existing.manifest_sha256 != manifest_sha256:
                 raise M49PortableSourceError("existing portable source stage has another manifest")
             return existing
         os.replace(staging, final)
         published = True
-        return validate_m49_portable_source_stage(final)
+        # The fresh stage consists only of files produced and hashed above.
+        # Its runner still performs the complete validation immediately before
+        # execution, so validating here would duplicate all multi-gigabyte I/O
+        # without adding a trust boundary.
+        return M49PortableSourceStage(
+            root=final,
+            identity_sha256=identity_sha256,
+            manifest_sha256=manifest_sha256,
+            timeline_frame_count=len(anchors),
+            available_lidar_frame_count=available_slot,
+        )
     finally:
         if not published:
             shutil.rmtree(staging, ignore_errors=True)

tests/test_m49_portable_executor_release.py

@@ -540,6 +540,32 @@ def test_dynamic_source_materializer_is_source_derived_and_tamper_evident(
         validate_m49_portable_source_stage(stage.root)
 
 
+def test_fresh_source_stage_defers_full_validation_until_execution_boundary(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    original = source_module.validate_m49_portable_source_stage
+    validated_roots: list[Path] = []
+
+    def counting_validator(root: Path) -> source_module.M49PortableSourceStage:
+        validated_roots.append(root)
+        return original(root)
+
+    monkeypatch.setattr(
+        source_module,
+        "validate_m49_portable_source_stage",
+        counting_validator,
+    )
+
+    first_root, _expected = _materialized_source(tmp_path, monkeypatch)
+    assert validated_roots == []
+    validated = original(first_root)
+    assert validated.timeline_frame_count == 3
+
+    second_root, _expected = _materialized_source(tmp_path, monkeypatch)
+    assert second_root == first_root
+    assert validated_roots == [first_root]
+
+
 def test_source_materializer_rejects_unadmitted_adjacent_metadata(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None: