perf(m49): verify source payloads during assembly

src/k1link/observatory/m49_portable_executor.py

@@ -34,7 +34,7 @@ from k1link.observatory.m49_portable_source import (
     M49_PORTABLE_TGS_SEQUENCE,
     M49PortableSourceStage,
     materialize_m49_portable_source_from_worker_stage,
-    validate_m49_portable_source_stage,
+    validate_m49_portable_source_stage_binding,
 )
 from k1link.observatory.portable_result_contract import (
     OBSERVATION_ONLY_AUTHORITY,
@@ -249,7 +249,7 @@ class M49PortableProfileRunnerAdapter:
             )
         job = source.job
         _verify_plan(plan, job=job, definition=self.definition)
-        stage = validate_m49_portable_source_stage(source.root)
+        stage = validate_m49_portable_source_stage_binding(source.m49_stage)
         _verify_installation_unchanged(self.installation)
         workspace = Path(
             tempfile.mkdtemp(prefix=".m49-portable-run-", dir=self.installation.output_parent)
@@ -268,7 +268,7 @@ class M49PortableProfileRunnerAdapter:
                 timeout_seconds=self.installation.timeout_seconds,
             )
             package = assemble_m49_portable_result(
-                source_stage_root=stage.root,
+                source_stage=stage,
                 runner_output_root=output,
                 runner_timing_path=timing,
                 profile_path=self.installation.profile_path,

src/k1link/observatory/m49_portable_result.py

@@ -33,8 +33,9 @@ from k1link.observatory.m49_portable_source import (
     M49_PORTABLE_PROFILE_SCHEMA,
     M49_PORTABLE_STAGE_INDEX,
     M49_PORTABLE_STAGE_MANIFEST,
+    M49PortableSourceStage,
     read_m49_source_index,
-    validate_m49_portable_source_stage,
+    validate_m49_portable_source_stage_binding,
 )
 from k1link.observatory.portable_result_contract import (
     OBSERVATION_ONLY_AUTHORITY,
@@ -146,7 +147,7 @@ class _TimingRow:
 
 def assemble_m49_portable_result(
     *,
-    source_stage_root: Path,
+    source_stage: M49PortableSourceStage,
     runner_output_root: Path,
     runner_timing_path: Path,
     profile_path: Path,
@@ -157,14 +158,17 @@ def assemble_m49_portable_result(
 ) -> M49PortableResultPackage:
     """Assemble one exact content-addressed portable result package.
 
-    The input ``job`` is either the running durable queue record or its exact
+    ``source_stage`` is a previously admitted, content-addressed stage.  Its
-    path-free Worker projection.  The returned manifest digest and result ID
+    small manifest, schedule, and index are rebound here after the external
-    are the two values the Worker later supplies to the queue's success
+    runner returns; each large point file is digest-checked during the required
-    transition.
+    assembly read.  The input ``job`` is either the running durable queue record
+    or its exact path-free Worker projection.  The returned manifest digest and
+    result ID are the two values the Worker later supplies to the queue's
+    success transition.
     """
 
     _verify_running_job_definition(job, definition)
-    source_stage = validate_m49_portable_source_stage(source_stage_root)
+    source_stage = validate_m49_portable_source_stage_binding(source_stage)
     _verify_source_stage_job_binding(source_stage.root, job)
     profile = _read_profile(profile_path, definition)
     source_index_path = source_stage.root / M49_PORTABLE_STAGE_INDEX
@@ -228,6 +232,8 @@ def assemble_m49_portable_result(
                     *PurePosixPath(cast(str, source["relative_path"])).parts
                 ),
                 label="materialized TGS input",
+                expected_byte_length=cast(int, source["byte_length"]),
+                expected_sha256=cast(str, source["sha256"]),
             )
             ground = _load_xyzi(
                 output_root / f"{frame_index}_ground.bin",
@@ -1210,11 +1216,30 @@ def _read_existing_package(root: Path) -> PortableResultPackageManifest:
     return package
 
 
-def _load_xyzi(path: Path, *, label: str) -> npt.NDArray[np.float32]:
+def _load_xyzi(
+    path: Path,
+    *,
+    label: str,
+    expected_byte_length: int | None = None,
+    expected_sha256: str | None = None,
+) -> npt.NDArray[np.float32]:
     candidate = _safe_file(path, label)
-    if candidate.stat().st_size % 16:
+    byte_length = candidate.stat().st_size
+    if (expected_byte_length is None) != (expected_sha256 is None):
+        raise M49PortableResultError(f"{label} expected identity is incomplete")
+    if expected_byte_length is not None and (
+        byte_length != expected_byte_length or expected_byte_length < 1
+    ):
+        raise M49PortableResultError(f"{label} byte length changed")
+    if byte_length % 16:
         raise M49PortableResultError(f"{label} byte shape changed")
-    values = np.fromfile(candidate, dtype="<f4")
+    if expected_sha256 is None:
+        values = np.fromfile(candidate, dtype="<f4")
+    else:
+        payload = candidate.read_bytes()
+        if hashlib.sha256(payload).hexdigest() != expected_sha256:
+            raise M49PortableResultError(f"{label} digest changed")
+        values = np.frombuffer(payload, dtype="<f4")
     result = values.reshape(-1, 4)
     if not np.isfinite(result).all():
         raise M49PortableResultError(f"{label} contains non-finite points")

src/k1link/observatory/m49_portable_source.py

@@ -346,6 +346,68 @@ def validate_m49_portable_source_stage(root: Path) -> M49PortableSourceStage:
     )
 
 
+def validate_m49_portable_source_stage_binding(
+    stage: M49PortableSourceStage,
+) -> M49PortableSourceStage:
+    """Rebind an admitted stage without re-hashing its large point files.
+
+    A fresh stage was produced and hashed by this materializer; a reused stage
+    has already passed ``validate_m49_portable_source_stage``.  The runner only
+    needs to prove that the bound root and its small canonical manifest still
+    name that stage.  Result assembly checks each point-file digest while doing
+    its required read, avoiding a separate multi-gigabyte pass.
+    """
+
+    resolved = _safe_directory(stage.root, "bound portable source stage")
+    payload, manifest = _read_canonical_document(
+        resolved / M49_PORTABLE_STAGE_MANIFEST,
+        expected_sha256=stage.manifest_sha256,
+        label="bound M4.9 source stage manifest",
+    )
+    identity = _object(manifest.get("identity"), "bound M4.9 source stage identity")
+    identity_sha256 = _string(
+        manifest.get("identity_sha256"),
+        "bound M4.9 source stage identity sha256",
+    )
+    timeline = _object(identity.get("timeline"), "bound M4.9 source stage timeline")
+    frame_count = _positive_int(timeline.get("frame_count"), "timeline frame count")
+    available_count = _positive_int(
+        timeline.get("available_lidar_frame_count"),
+        "available LiDAR frame count",
+    )
+    artifacts = _object(manifest.get("artifacts"), "bound M4.9 source stage artifacts")
+    if set(artifacts) != {"schedule", "sequence-index"}:
+        raise M49PortableSourceError("bound M4.9 source stage artifacts changed")
+    schedule = _verify_stage_artifact(resolved, artifacts["schedule"], "schedule")
+    index = _verify_stage_artifact(
+        resolved,
+        artifacts["sequence-index"],
+        "sequence-index",
+    )
+    records = read_m49_source_index(index, expected_frame_count=frame_count)
+    _verify_schedule(schedule, records)
+    if (
+        manifest.get("schema_version") != M49_PORTABLE_SOURCE_STAGE_SCHEMA
+        or manifest.get("authority") != _AUTHORITY
+        or identity.get("authority") != _AUTHORITY
+        or hashlib.sha256(canonical_json(identity)).hexdigest() != identity_sha256
+        or identity_sha256 != stage.identity_sha256
+        or hashlib.sha256(payload).hexdigest() != stage.manifest_sha256
+        or resolved.name != f"{M49_PORTABLE_STAGE_PREFIX}{identity_sha256}"
+        or frame_count != stage.timeline_frame_count
+        or available_count != stage.available_lidar_frame_count
+        or sum(bool(row["sample_available"]) for row in records) != available_count
+    ):
+        raise M49PortableSourceError("bound portable source stage changed")
+    return M49PortableSourceStage(
+        root=resolved,
+        identity_sha256=identity_sha256,
+        manifest_sha256=stage.manifest_sha256,
+        timeline_frame_count=frame_count,
+        available_lidar_frame_count=available_count,
+    )
+
+
 def read_m49_source_index(
     path: Path,
     *,

tests/test_m49_portable_executor_release.py

@@ -31,6 +31,7 @@ from k1link.observatory.m49_portable_source import (
     materialize_m49_portable_source,
     read_m49_source_index,
     validate_m49_portable_source_stage,
+    validate_m49_portable_source_stage_binding,
 )
 from k1link.observatory.portable_result_contract import (
     PortableResultPackageManifest,
@@ -566,6 +567,36 @@ def test_fresh_source_stage_defers_full_validation_until_execution_boundary(
     assert validated_roots == [first_root]
 
 
+def test_bound_source_stage_rechecks_small_documents_without_rehashing_sequence(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    stage_root, _expected = _materialized_source(tmp_path, monkeypatch)
+    stage = validate_m49_portable_source_stage(stage_root)
+
+    def unexpected_sequence_rehash(*_args: object, **_kwargs: object) -> None:
+        raise AssertionError("bound-stage validation rehashed the point sequence")
+
+    monkeypatch.setattr(source_module, "_verify_sequence_files", unexpected_sequence_rehash)
+    rebound = validate_m49_portable_source_stage_binding(stage)
+
+    assert rebound == stage
+
+
+@pytest.mark.parametrize("relative_path", ["schedule.tsv", "sequence-index.ndjson"])
+def test_bound_source_stage_rejects_changed_small_documents(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    relative_path: str,
+) -> None:
+    stage_root, _expected = _materialized_source(tmp_path, monkeypatch)
+    stage = validate_m49_portable_source_stage(stage_root)
+    artifact = stage.root / relative_path
+    artifact.write_bytes(artifact.read_bytes() + b"changed")
+
+    with pytest.raises(M49PortableSourceError):
+        validate_m49_portable_source_stage_binding(stage)
+
+
 def test_source_materializer_rejects_unadmitted_adjacent_metadata(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:
@@ -756,20 +787,18 @@ def test_result_v2_assembler_and_exact_validator_round_trip(
         created_at_utc=lambda: NOW,
         invoker=fake_invoke,
     )
-    draft = runner.run(
+    plan = PortableWorkerRuntimePlan(
-        PortableWorkerRuntimePlan(
+        job_id=sealed.job_id,
-            job_id=sealed.job_id,
+        adapter_id="m49-tgs-worker006-portable-v2",
-            adapter_id="m49-tgs-worker006-portable-v2",
+        candidate_sha256="f" * 64,
-            candidate_sha256="f" * 64,
+        setup_id=sealed.setup_id,
-            setup_id=sealed.setup_id,
+        definition_sha256=sealed.definition_sha256,
-            definition_sha256=sealed.definition_sha256,
+        source_bundle_sha256=sealed.source_bundle_sha256,
-            source_bundle_sha256=sealed.source_bundle_sha256,
+        source_capability_manifest_sha256=sealed.source_capability_manifest_sha256,
-            source_capability_manifest_sha256=sealed.source_capability_manifest_sha256,
+        result_contract_sha256=definition.result_contract.contract_sha256,
-            result_contract_sha256=definition.result_contract.contract_sha256,
+        phases=M49_PORTABLE_RUNTIME_PHASES,
-            phases=M49_PORTABLE_RUNTIME_PHASES,
-        ),
-        bound_source,
     )
+    draft = runner.run(plan, bound_source)
     package = PortableResultPackageManifest.from_bytes((draft.root / "manifest.json").read_bytes())
     assert draft.root.name == package.manifest_sha256
     assert draft.result_id.startswith("m49-tgs-portable-review-")
@@ -816,6 +845,18 @@ def test_result_v2_assembler_and_exact_validator_round_trip(
             )
         )
 
+    source_rows = read_m49_source_index(
+        source_stage.root / "sequence-index.ndjson",
+        expected_frame_count=source_stage.timeline_frame_count,
+    )
+    available_row = next(row for row in source_rows if row["sample_available"] is True)
+    native_path = source_stage.root / cast(str, available_row["relative_path"])
+    native_payload = bytearray(native_path.read_bytes())
+    native_payload[-1] ^= 0x01
+    native_path.write_bytes(native_payload)
+    with pytest.raises(M49PortableResultError, match="digest changed"):
+        runner.run(plan, bound_source)
+
 
 def test_executor_release_candidate_is_deterministic_blocked_and_tamper_evident(
     tmp_path: Path,