fix(perception): separate preroll history from current sensor binding

Preserve rolling points and original clocks; expose per-modality ages, held pose and rejection reasons. Saved-lineage audit changes only the first admission (75 to 76 of 128). No new model/performance claim. 384 focused Python and 62 frontend tests passed.

experiments/perception/worker/streaming_profile_stage1/pilot_sensor_binding.py

@@ -0,0 +1,129 @@
+"""Causal input binding with explicit preroll history and per-modality age.
+
+No new inference or interpolation. The caller retains all preroll points in its
+bounded rolling window; only the initial current-pair selection is narrowed.
+After the first camera, the original increment admission rules are unchanged.
+"""
+
+from dataclasses import dataclass
+
+POSE_AGE_NS = 100_000_000
+NEWEST_POINT_AGE_NS = 100_000_000
+OLDEST_POINT_AGE_NS = 250_000_000
+POINT_POSE_SKEW_NS = 100_000_000
+
+
+def increment_identity(event):
+    return {
+        "sequence": event.sequence,
+        "host_monotonic_ns": event.time_ns,
+        "points": len(event.value[0]),
+    }
+
+
+def milliseconds(value):
+    return None if value is None else value / 1e6
+
+
+@dataclass(frozen=True)
+class SensorBinding:
+    increments: tuple
+    history_only: tuple
+    pose_age_ns: int | None
+    newest_point_age_ns: int | None
+    oldest_point_age_ns: int | None
+    binding_age_ns: int | None
+    pose_state: str
+    points_state: str
+    reasons: tuple[str, ...]
+
+    @property
+    def available(self):
+        return not self.reasons
+
+    def document(self):
+        return {
+            "schema_version": "missioncore.pilot-sensor-binding/v1",
+            "pose_state": self.pose_state,
+            "points_state": self.points_state,
+            "current_pair_available": self.available,
+            "reason_codes": list(self.reasons),
+            "pose_age_ms": milliseconds(self.pose_age_ns),
+            "newest_point_age_ms": milliseconds(self.newest_point_age_ns),
+            "oldest_point_age_ms": milliseconds(self.oldest_point_age_ns),
+            "point_pose_skew_ms": milliseconds(self.binding_age_ns),
+            "preroll_history_only": [increment_identity(e) for e in self.history_only],
+            "preroll_history_disposition": "retained-in-bounded-rolling-window",
+        }
+
+
+def bind_sensors(camera_time_ns, pose, increments, *, previous_camera_time_ns=None):
+    increments = tuple(increments)
+    if previous_camera_time_ns is not None and previous_camera_time_ns >= camera_time_ns:
+        raise ValueError("camera binding clock must increase")
+    if pose is not None and pose.time_ns > camera_time_ns:
+        raise ValueError("future pose cannot bind to camera")
+    if any(e.time_ns > camera_time_ns for e in increments):
+        raise ValueError("future points cannot bind to camera")
+    if any(a.time_ns > b.time_ns for a, b in zip(increments, increments[1:], strict=False)):
+        raise ValueError("point binding clock moved backwards")
+
+    history_only = ()
+    if previous_camera_time_ns is None:
+        # History used to warm rolling geometry is not one current increment.
+        # Preserve its identity separately; do not retimestamp or silently drop it.
+        selected, history = [], []
+        for event in increments:
+            is_current = (
+                pose is not None
+                and camera_time_ns - event.time_ns <= OLDEST_POINT_AGE_NS
+                and abs(event.time_ns - pose.time_ns) <= POINT_POSE_SKEW_NS
+            )
+            (selected if is_current else history).append(event)
+        increments, history_only = tuple(selected), tuple(history)
+
+    pose_age = None if pose is None else camera_time_ns - pose.time_ns
+    newest_age = None if not increments else camera_time_ns - increments[-1].time_ns
+    oldest_age = None if not increments else camera_time_ns - increments[0].time_ns
+    skew = (
+        max(abs(e.time_ns - pose.time_ns) for e in increments)
+        if increments and pose is not None
+        else None
+    )
+    reasons = []
+    if pose is None:
+        pose_state = "unavailable"
+        reasons.append("pose-unavailable")
+    elif pose_age > POSE_AGE_NS:
+        pose_state = "stale"
+        reasons.append("pose-too-old")
+    else:
+        pose_state = (
+            "held"
+            if previous_camera_time_ns is not None and pose.time_ns <= previous_camera_time_ns
+            else "current"
+        )
+    if not increments or not any(len(e.value[0]) for e in increments):
+        points_state = "unavailable"
+        reasons.append("point-increment-unavailable")
+    else:
+        points_state = "current"
+        if newest_age > NEWEST_POINT_AGE_NS:
+            points_state = "stale"
+            reasons.append("newest-points-too-old")
+        if oldest_age > OLDEST_POINT_AGE_NS:
+            points_state = "stale"
+            reasons.append("oldest-points-too-old")
+    if skew is not None and skew > POINT_POSE_SKEW_NS:
+        reasons.append("point-pose-skew")
+    return SensorBinding(
+        increments,
+        history_only,
+        pose_age,
+        newest_age,
+        oldest_age,
+        skew,
+        pose_state,
+        points_state,
+        tuple(reasons),
+    )

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -27,6 +27,7 @@ import numpy as np
 from pilot_ipc import receive, send
 from pilot_queue import Mailbox
 from pilot_scheduler import GpuStage
+from pilot_sensor_binding import bind_sensors, increment_identity, milliseconds
 from pilot_source import SensorArchive, camera_events, merged_events
 
 
@@ -66,11 +67,14 @@ def produce(args, decoder, mailbox, stop, report):
         full_source_prepass=False,
     )
     pose = None
+    previous_camera_time = None
     rolling = deque()
     fresh = []
     arrivals = Counter()
     release_lags = []
     skipped_prefix = Counter()
+    binding_reasons = Counter()
+    preroll_history_points = 0
     try:
         for event in merged_events(archive, args.camera_index, args.frames):
             if stop.is_set():
@@ -100,6 +104,12 @@ def produce(args, decoder, mailbox, stop, report):
             send(decoder.stdin, {"op": "next"})
             decoded, raw = receive(decoder.stdout)
             image = np.frombuffer(raw, np.uint8).reshape(600, 800, 3)
+            binding = bind_sensors(
+                event.time_ns, pose, fresh, previous_camera_time_ns=previous_camera_time
+            )
+            fresh = binding.increments
+            binding_reasons.update(binding.reasons)
+            preroll_history_points += sum(len(e.value[0]) for e in binding.history_only)
             points = np.concatenate([e.value[0] for e in fresh]) if fresh else np.empty((0, 3))
             rolling_points = (
                 np.concatenate([e.value[0] for e in rolling]) if rolling else np.empty((0, 3))
@@ -109,20 +119,6 @@ def produce(args, decoder, mailbox, stop, report):
                 if rolling
                 else np.empty(0, np.int64)
             )
-            pose_age = event.time_ns - pose.time_ns if pose else None
-            point_age = event.time_ns - fresh[-1].time_ns if fresh else None
-            oldest_age = event.time_ns - fresh[0].time_ns if fresh else None
-            binding_age = (
-                max(abs(e.time_ns - pose.time_ns) for e in fresh) if fresh and pose else None
-            )
-            available = bool(
-                len(points)
-                and pose
-                and 0 <= pose_age <= 100_000_000
-                and 0 <= point_age <= 100_000_000
-                and oldest_age <= 250_000_000
-                and binding_age <= 100_000_000
-            )
             bundle = {
                 "sequence": event.sequence,
                 "time_ns": event.time_ns,
@@ -134,24 +130,18 @@ def produce(args, decoder, mailbox, stop, report):
                 "rolling_points": rolling_points,
                 "rolling_times": rolling_times,
                 "pose": pose.value if pose else None,
-                "available": available,
-                "binding_age_ms": binding_age / 1e6 if binding_age is not None else None,
+                "available": binding.available,
+                "binding_age_ms": milliseconds(binding.binding_age_ns),
+                "sensor_binding": binding.document(),
                 "lineage": {
                     "camera_index_sequence": event.value["sequence"],
                     "camera_host_monotonic_ns": event.time_ns,
                     "pose_sequence": pose.sequence if pose else None,
                     "pose_host_monotonic_ns": pose.time_ns if pose else None,
-                    "point_increments": [
-                        {
-                            "sequence": e.sequence,
-                            "host_monotonic_ns": e.time_ns,
-                            "points": len(e.value[0]),
-                        }
-                        for e in fresh
-                    ],
-                    "pose_age_ms": pose_age / 1e6 if pose_age is not None else None,
-                    "newest_point_age_ms": point_age / 1e6 if point_age is not None else None,
-                    "oldest_point_age_ms": oldest_age / 1e6 if oldest_age is not None else None,
+                    "point_increments": [increment_identity(e) for e in fresh],
+                    "pose_age_ms": milliseconds(binding.pose_age_ns),
+                    "newest_point_age_ms": milliseconds(binding.newest_point_age_ns),
+                    "oldest_point_age_ms": milliseconds(binding.oldest_point_age_ns),
                 },
                 "source_release_lag_ms": max(0, arrived - due) / 1e6,
                 "decode_ms": decoded["decode_ms"],
@@ -161,6 +151,7 @@ def produce(args, decoder, mailbox, stop, report):
             bundle["payload_bytes"] = payload_size(bundle)
             mailbox.put(bundle)
             fresh = []
+            previous_camera_time = event.time_ns
             report["last_camera_due_ns"] = due
             if arrivals["camera"] >= args.frames:
                 break
@@ -170,6 +161,8 @@ def produce(args, decoder, mailbox, stop, report):
         report.update(
             arrivals=dict(arrivals),
             skipped_prefix=dict(skipped_prefix),
+            sensor_binding_reasons=dict(binding_reasons),
+            preroll_history_only_points=preroll_history_points,
             release_lag_ms=distribution(release_lags),
             incremental_reads=archive.counters(),
             window_end_monotonic_ns=time.monotonic_ns(),
@@ -387,6 +380,7 @@ def run(args):
                 scene.update(
                     sequence=bundle["sequence"],
                     lineage=bundle["lineage"],
+                    sensor_binding=bundle["sensor_binding"],
                     available=bundle["available"],
                     original_source_ns=bundle["time_ns"],
                 )