fix(lab): preserve replay view and restore SLAM semantics

This commit is contained in:
DCCONSTRUCTIONS
2026-08-26 23:37:05 +03:00
parent 46cef8df17
commit 296cf610cd
8 changed files with 230 additions and 45 deletions
+72
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import hashlib
import importlib.util
import json
import sys
@@ -7,6 +8,7 @@ from pathlib import Path
import numpy as np
from k1link.laboratory.m49_tgs_full_shadow import seal_m49_tgs_full_shadow
from k1link.web import m49_tgs_full_shadow_api as full_shadow_api
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
@@ -59,6 +61,76 @@ def test_full_shadow_worker_artifact_is_deterministic_and_cpu_only(tmp_path: Pat
assert "gpu_requested = $false" in runner_text
def test_full_shadow_seal_binds_visual_and_semantic_timelines(tmp_path: Path) -> None:
source = tmp_path / "worker"
source.mkdir()
files: dict[str, dict[str, object]] = {}
for name in (
"costmap-cell-centers-xy-m.npy",
"costmap-cell-indices-xy.npy",
"costmap-states.npy",
"costmap-z-bounds-m.npy",
"frames.ndjson",
):
payload = f"sealed:{name}\n".encode()
(source / name).write_bytes(payload)
files[name] = {
"bytes": len(payload),
"sha256": hashlib.sha256(payload).hexdigest(),
}
worker = {
"schema_version": "missioncore.m49-tgs-full-shadow-result/v1",
"status": "passed",
"source_pack_sha256": "a" * 64,
"input_manifest_sha256": "b" * 64,
"config_sha256": "c" * 64,
"timeline": {
"frame_count": 4489,
"available_lidar_frame_count": 3928,
"missing_lidar_frame_count": 561,
},
"point_accounting": {"unaccounted": 0},
"costmap": {"cell_size_m": 0.45, "radius_m": 12.0},
"performance": {},
"acceptance": {},
"files": files,
}
(source / "result.json").write_text(json.dumps(worker), encoding="utf-8")
(source / "worker-summary.json").write_text(json.dumps({
"schema_version": "missioncore.m49-tgs-full-shadow-worker-summary/v1",
"gpu_requested": False,
"aos_used": False,
"all_timeline_frames_accounted": True,
"all_eligible_points_accounted": True,
"canonical_triton_health": "healthy",
"canonical_triton_id": "triton",
"wall_seconds": 1.0,
}), encoding="utf-8")
profile = tmp_path / "profile.json"
profile.write_text(json.dumps({
"schema_version": "missioncore.m49-tgs-full-shadow-profile/v1",
"profile_id": "test",
"profile": {"history_seconds": 1.0},
"costmap": {"state_priority": ["NONGROUND_OCCUPIED"]},
"state_codes": {"UNOBSERVED": 0},
}), encoding="utf-8")
visual = "m4-threat-replay-" + "d" * 64
semantic = "e47-semantic-slam-" + "e" * 64
sealed = seal_m49_tgs_full_shadow(
source_root=source,
destination_root=tmp_path / "results",
profile_path=profile,
linked_visual_result_id=visual,
linked_semantic_result_id=semantic,
created_at_utc="2026-08-26T20:27:19Z",
)
assert sealed.report["source"]["linked_visual_result_id"] == visual
assert sealed.report["source"]["linked_semantic_result_id"] == semantic
assert sealed.manifest["identity"]["linked_semantic_result_id"] == semantic
def test_full_shadow_chunk_contract_keeps_missing_lidar_unobserved(
monkeypatch,
tmp_path: Path,