From 35d99e40d53e78a05803d94935e5079f84ef9cdc Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Mon, 31 Aug 2026 19:06:59 +0300 Subject: [PATCH] perf(m49): verify source payloads during assembly --- .../observatory/m49_portable_executor.py | 6 +- src/k1link/observatory/m49_portable_result.py | 45 ++++++++++--- src/k1link/observatory/m49_portable_source.py | 62 +++++++++++++++++ tests/test_m49_portable_executor_release.py | 67 +++++++++++++++---- 4 files changed, 154 insertions(+), 26 deletions(-) diff --git a/src/k1link/observatory/m49_portable_executor.py b/src/k1link/observatory/m49_portable_executor.py index 3e848ea..cc631cb 100644 --- a/src/k1link/observatory/m49_portable_executor.py +++ b/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, diff --git a/src/k1link/observatory/m49_portable_result.py b/src/k1link/observatory/m49_portable_result.py index 8a3efd7..718b0c7 100644 --- a/src/k1link/observatory/m49_portable_result.py +++ b/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 - 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. + ``source_stage`` is a previously admitted, content-addressed stage. Its + small manifest, schedule, and index are rebound here after the external + runner returns; each large point file is digest-checked during the required + 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=" 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, *, diff --git a/tests/test_m49_portable_executor_release.py b/tests/test_m49_portable_executor_release.py index c8de07d..e42bf8f 100644 --- a/tests/test_m49_portable_executor_release.py +++ b/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( - PortableWorkerRuntimePlan( - job_id=sealed.job_id, - adapter_id="m49-tgs-worker006-portable-v2", - candidate_sha256="f" * 64, - setup_id=sealed.setup_id, - definition_sha256=sealed.definition_sha256, - source_bundle_sha256=sealed.source_bundle_sha256, - source_capability_manifest_sha256=sealed.source_capability_manifest_sha256, - result_contract_sha256=definition.result_contract.contract_sha256, - phases=M49_PORTABLE_RUNTIME_PHASES, - ), - bound_source, + plan = PortableWorkerRuntimePlan( + job_id=sealed.job_id, + adapter_id="m49-tgs-worker006-portable-v2", + candidate_sha256="f" * 64, + setup_id=sealed.setup_id, + definition_sha256=sealed.definition_sha256, + source_bundle_sha256=sealed.source_bundle_sha256, + source_capability_manifest_sha256=sealed.source_capability_manifest_sha256, + result_contract_sha256=definition.result_contract.contract_sha256, + phases=M49_PORTABLE_RUNTIME_PHASES, ) + 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,