feat(perception): add gravity-aligned TGS evidence gate

tests/test_m49_tgs_fail_closed_evidence.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import tarfile
+from pathlib import Path
+
+import numpy as np
+
+REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = (
+    REPOSITORY_ROOT
+    / "experiments/perception/worker/m49_t3_travel/build_tgs_fail_closed_evidence.py"
+)
+SPEC = importlib.util.spec_from_file_location("m49_tgs_evidence", SCRIPT)
+assert SPEC is not None and SPEC.loader is not None
+MODULE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(MODULE)
+PREPARE_PATH = (
+    REPOSITORY_ROOT
+    / "experiments/perception/worker/m49_t3_travel/prepare_tgs_fail_closed_inputs.py"
+)
+PREPARE_SPEC = importlib.util.spec_from_file_location("m49_tgs_prepare", PREPARE_PATH)
+assert PREPARE_SPEC is not None and PREPARE_SPEC.loader is not None
+PREPARE = importlib.util.module_from_spec(PREPARE_SPEC)
+PREPARE_SPEC.loader.exec_module(PREPARE)
+BUILDER_PATH = REPOSITORY_ROOT / "scripts/build_m49_tgs_fail_closed_worker_artifact.py"
+BUILDER_SPEC = importlib.util.spec_from_file_location("m49_tgs_builder", BUILDER_PATH)
+assert BUILDER_SPEC is not None and BUILDER_SPEC.loader is not None
+BUILDER = importlib.util.module_from_spec(BUILDER_SPEC)
+BUILDER_SPEC.loader.exec_module(BUILDER)
+
+
+def test_gravity_local_input_translates_without_rotating_map_axes() -> None:
+    result = PREPARE.gravity_local_xyzi(
+        np.asarray([[2.0, 4.0, 6.0], [3.0, 6.0, 9.0]], dtype=np.float32),
+        np.asarray([1.0, 2.0, 3.0], dtype=np.float64),
+    )
+    assert np.array_equal(
+        result,
+        np.asarray([[1.0, 2.0, 3.0, 0.0], [2.0, 4.0, 6.0, 0.0]], dtype=np.float32),
+    )
+
+
+def test_exact_input_complement_is_retained_as_unknown_rejected() -> None:
+    native = np.asarray(
+        [[2, 0, 0, 0], [3, 0, 0, 0], [4, 0, 1, 0], [0.5, 0, 0, 0]],
+        dtype=np.float32,
+    )
+    ground = native[[0]]
+    nonground = native[[2]]
+    points, states = MODULE.classify_exact_input(native, ground, nonground)
+    assert np.array_equal(points, native[:3, :3])
+    assert np.array_equal(states, np.asarray([1, 3, 2], dtype=np.uint8))
+
+
+def test_costmap_priority_is_nonground_then_rejected_then_ground() -> None:
+    grid = MODULE.costmap_grid(2.0, 1.0)
+    points = np.asarray(
+        [[0.1, 0.1, 0.0], [0.2, 0.2, 0.1], [0.3, 0.3, 0.2]],
+        dtype=np.float32,
+    )
+    states, ground, nonground, rejected, _ = MODULE.rasterize_costmap(
+        points,
+        np.asarray([1, 3, 2], dtype=np.uint8),
+        grid,
+        cell_size_m=1.0,
+    )
+    target = np.flatnonzero((grid[:, 0] == 0) & (grid[:, 1] == 0))
+    assert target.size == 1
+    index = int(target[0])
+    assert states[index] == 2
+    assert (ground[index], nonground[index], rejected[index]) == (1, 1, 1)
+
+
+def test_deterministic_npz_has_identical_bytes(tmp_path: Path) -> None:
+    arrays = {
+        "b": np.asarray([3, 2, 1], dtype=np.int32),
+        "a": np.asarray([[1.0, 2.0]], dtype=np.float32),
+    }
+    first = tmp_path / "first.npz"
+    second = tmp_path / "second.npz"
+    MODULE.write_deterministic_npz(first, arrays)
+    MODULE.write_deterministic_npz(second, arrays)
+    assert first.read_bytes() == second.read_bytes()
+    with np.load(first, allow_pickle=False) as archive:
+        assert np.array_equal(archive["a"], arrays["a"])
+        assert np.array_equal(archive["b"], arrays["b"])
+
+
+def test_tgs_worker_artifact_is_deterministic_and_cpu_only(tmp_path: Path) -> None:
+    revision = "a" * 40
+    patch_id = "mission-core-m49-tgs-unit-001"
+    first = BUILDER.build(patch_id, tmp_path / "first", revision=revision)
+    second = BUILDER.build(patch_id, tmp_path / "second", revision=revision)
+    first_bytes = Path(first["artifact"]).read_bytes()
+    assert first_bytes == Path(second["artifact"]).read_bytes()
+    assert first["sha256"] == hashlib.sha256(first_bytes).hexdigest()
+    with tarfile.open(first["artifact"], "r:gz") as archive:
+        regular = {
+            member.name: archive.extractfile(member).read()  # type: ignore[union-attr]
+            for member in archive.getmembers()
+            if member.isfile()
+        }
+    release = json.loads(regular["payload/release.json"])
+    assert release["candidate_id"] == "travel-tgs-only"
+    assert release["code_revision"] == revision
+    assert release["authority"]["navigation_or_actuation_allowed"] is False
+    assert "--gpus" not in regular["payload/Invoke-M49TgsFailClosedEvidence.ps1"].decode()