fix(perception): preserve quaternion alignment across binary ingress

experiments/perception/worker/streaming_profile_stage1/pilot_projection_parity_probe.py

@@ -0,0 +1,156 @@
+"""CPU-only numerical diagnosis on a bounded source prefix, no model execution.
+
+Frozen observations select ranges to compare; they never feed a new graph.
+Probe changes only storage alignment of identical pose values, not geometry math.
+"""
+
+import argparse
+import hashlib
+import json
+import time
+from datetime import UTC, datetime
+from pathlib import Path
+from types import SimpleNamespace
+
+import numpy as np
+import run_joint_pilot as legacy
+from pilot_graph import calibration
+
+from k1link.perception.geometry_math import (
+    project_map_points_kb4,
+    quaternion_xyzw_to_rotation_matrix,
+)
+from k1link.perception.streaming_sensors import normalized_sensor
+
+
+def aligned_pose(pose, offset):
+    # Force a known address modulo 64, preserving bytes and C order.
+    position, quaternion = pose
+    storage = np.empty(128, np.uint8)
+    start = (offset - storage.ctypes.data) % 64
+    q = np.ndarray((4,), dtype=np.float64, buffer=storage, offset=start)
+    q[:] = quaternion
+    return position, q
+
+
+def run(args):
+    root = Path(args.output)
+    root.mkdir()
+    report = {
+        "created_utc": datetime.now(UTC).isoformat(),
+        "started_monotonic_ns": time.monotonic_ns(),
+        "scope": "CPU-only offline bounded numerical diagnosis; not latency qualification",
+        "numpy_version": np.__version__,
+        "model_runs": 0,
+        "frames": [],
+    }
+    with Path(args.reference).open() as stream:
+        reference = [json.loads(next(stream)) for _ in range(args.frames)]
+    projection = calibration(args.calibration)
+
+    class Sink:
+        error = None
+
+        def put(self, bundle):
+            row = {"sequence": bundle["sequence"], "available": bundle["available"]}
+            report["frames"].append(row)
+            if not bundle["available"]:
+                return True
+            position, quaternion = bundle["pose"]
+            raw = position.tobytes() + quaternion.tobytes()
+            wire_pose = normalized_sensor("pose", 1, 1, raw).value
+            variants = {
+                "archive": bundle["pose"],
+                "wire": wire_pose,
+                "owned_copy": tuple(np.array(a, copy=True) for a in wire_pose),
+                "separate_bytes": (np.frombuffer(raw[:24], "<f8"), np.frombuffer(raw[24:], "<f8")),
+                **{
+                    f"offset_{offset}": aligned_pose(bundle["pose"], offset)
+                    for offset in (0, 8, 16, 24, 32, 40, 48, 56)
+                },
+            }
+            row["variants"] = {}
+            observations = reference[bundle["sequence"]]["observations"]
+            for name, pose in variants.items():
+                assert all(np.array_equal(a, b) for a, b in zip(pose, bundle["pose"], strict=True))
+                projected = project_map_points_kb4(
+                    bundle["points"],
+                    position_map_xyz=pose[0],
+                    orientation_map_from_lidar_xyzw=pose[1],
+                    profile=projection,
+                )
+                depths = dict(zip(projected.source_indices, projected.depths_m, strict=True))
+                ranges = []
+                for index, observation in enumerate(observations):
+                    if observation["basis"] != "fused" or observation["metric_geometry"] is None:
+                        continue
+                    distance = float(
+                        np.median([depths[k] for k in observation["source_point_ids"]])
+                    )
+                    target = observation["metric_geometry"]["range_m"]
+                    ranges.append(
+                        {
+                            "observation": index,
+                            "actual": distance,
+                            "reference": target,
+                            "exact": distance == target,
+                        }
+                    )
+                row["variants"][name] = {
+                    "quaternion_address_mod64": pose[1].ctypes.data % 64,
+                    "norm_hex": float(np.linalg.norm(pose[1])).hex(),
+                    "rotation_sha256": hashlib.sha256(
+                        quaternion_xyzw_to_rotation_matrix(pose[1])
+                    ).hexdigest(),
+                    "depth_sha256": hashlib.sha256(projected.depths_m).hexdigest(),
+                    "pixels_sha256": hashlib.sha256(projected.pixels_xy).hexdigest(),
+                    "source_indices_sha256": hashlib.sha256(projected.source_indices).hexdigest(),
+                    "ranges": ranges,
+                }
+            return True
+
+        def finish(self, error=None):
+            self.error = error
+
+    sink = Sink()
+    old_send, old_receive = legacy.send, legacy.receive
+    image = bytes(1440000)
+    legacy.send = lambda *args: None
+    legacy.receive = lambda *args: ({"decode_ms": 0}, image)
+    try:
+        legacy.produce(
+            args,
+            SimpleNamespace(stdin=None, stdout=None),
+            sink,
+            SimpleNamespace(is_set=lambda: False, wait=lambda _: False),
+            {},
+        )
+    finally:
+        legacy.send, legacy.receive = old_send, old_receive
+    if sink.error:
+        raise ValueError(sink.error)
+    report["finished_monotonic_ns"] = time.monotonic_ns()
+    report["mismatches"] = {}
+    for row in report["frames"]:
+        for name, variant in row.get("variants", {}).items():
+            report["mismatches"].setdefault(name, []).extend(
+                {"sequence": row["sequence"], **value}
+                for value in variant["ranges"]
+                if not value["exact"]
+            )
+    (root / "report.json").write_text(json.dumps(report, indent=2) + "\n")
+    print(json.dumps({"frames": len(report["frames"]), "mismatches": report["mismatches"]}))
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--output", required=True)
+    parser.add_argument("--frames", type=int, default=32)
+    parser.add_argument("--reference", required=True)
+    parser.add_argument("--calibration", default="/calibration.npz")
+    parser.add_argument("--camera-index", default="/camera-index.jsonl")
+    parser.add_argument("--sensor-archive", default="/sensor-source.npz")
+    args = parser.parse_args()
+    if not 1 <= args.frames <= 128:
+        parser.error("bounded prefix only")
+    run(args)

src/k1link/perception/streaming_sensors.py

@@ -170,9 +170,18 @@ def normalized_sensor(modality: str, sequence: int, time_ns: int, raw: bytes) ->
     if modality != "pose" or len(raw) != 56:
         raise ValueError("normalized pose layout outside bound")
     values = np.frombuffer(raw, "<f8")
-    if not np.isfinite(values).all() or abs(float(np.linalg.norm(values[3:])) - 1) > 0.01:
+    # The quaternion begins at byte 24 of the wire body. On the qualified x86
+    # NumPy runtime, its 8-mod-16 address selects a different dot/norm reduction
+    # and changes two downstream ranges by one ULP. Own a tiny aligned copy,
+    # matching the archive reader's numeric layout without changing any values
+    # or the projection/norm algorithm. Covered by the cache scratch allowance.
+    quaternion = np.array(values[3:], copy=True)
+    if quaternion.ctypes.data % 16:
+        raise ValueError("unqualified quaternion numeric buffer alignment")
+    quaternion.setflags(write=False)
+    if not np.isfinite(values).all() or abs(float(np.linalg.norm(quaternion)) - 1) > 0.01:
         raise ValueError("invalid source pose")
-    return SensorEvent(time_ns, "pose", sequence, (values[:3], values[3:]))
+    return SensorEvent(time_ns, "pose", sequence, (values[:3], quaternion))
 
 
 class CausalSensorWindow:

tests/test_perception_binary_bridge.py

@@ -187,6 +187,21 @@ def test_cloud_validation_checks_later_bounded_chunks():
         normalized_sensor("lidar", 1, 100, struct.pack("<I", 2000) + xyz.tobytes() + bytes(2000))
 
 
+def test_wire_pose_owns_aligned_immutable_quaternion_with_exact_values():
+    from k1link.perception.geometry_math import quaternion_xyzw_to_rotation_matrix
+
+    quaternion = np.array([0.1, -0.2, 0.3, 0.9273618495495703], "<f8")
+    raw = np.array([1, 2, 3], "<f8").tobytes() + quaternion.tobytes()
+    position, decoded = normalized_sensor("pose", 1, 100, raw).value
+    assert decoded.flags.owndata and decoded.ctypes.data % 16 == 0
+    assert not decoded.flags.writeable and not position.flags.writeable
+    assert decoded.tobytes() == quaternion.tobytes()
+    assert np.array_equal(
+        quaternion_xyzw_to_rotation_matrix(decoded),
+        quaternion_xyzw_to_rotation_matrix(quaternion),
+    )
+
+
 @pytest.mark.parametrize("failure", ["socket", "receiver-registration", "source-registration"])
 def test_bridge_initialization_failure_releases_unstarted_resources(monkeypatch, failure):
     monkeypatch.syspath_prepend(