fix(perception): require joint clock readiness before source activation

experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py

@@ -84,6 +84,10 @@ async def run(args):
         client = reader = None
 
     try:
+        # Reject a directory/missing input before establishing an activation clock.
+        first = next(camera_events(args.camera_index, 1)).time_ns
+        if not Path(args.sensor_archive).is_file() or not (root / "init.mp4").is_file():
+            raise ValueError("raw recording inputs are missing")
         deadline = time.monotonic() + 120
         startup_path = control / "bootstrap.json" if args.cross_host else grant_path
         while not startup_path.exists():
@@ -104,7 +108,6 @@ async def run(args):
             }
         else:
             clock, _ = read_grant(grant_path)
-        first = next(camera_events(args.camera_index, 1)).time_ns
         if clock["source_zero_ns"] != first - 500_000_000:
             raise ValueError("recording prefix does not match admitted source clock")
         report.update(

experiments/perception/worker/streaming_profile_stage1/pilot_source_control.py

@@ -22,6 +22,8 @@ class SourceControl:
         )
         self.source_zero_ns = int(bootstrap["source_zero_ns"])
         self.anchor = None  # Latch once after the initial clock handshake, never on resume.
+        self.proposal = None  # Retain an uncertain proposal across a lost ACK.
+        self.peer = None
         self.access = None
         self.confirmed = False
         self.failure = None
@@ -31,43 +33,79 @@ class SourceControl:
 
     async def update(self):
         probe, access = await self.client.poll()
-        bounds = self.window.add(probe)
+        self.window.add(probe)
         now = time.monotonic_ns()
-        if self.anchor is None:
-            try:
-                self.window.current(now).require(now)
-            except ClockMappingError:
-                pass  # Pre-start clock warmup, not a running source pause or retimestamp.
-            else:
-                self.anchor = SourceAnchor(self.source_zero_ns, now + 2_000_000_000)
-        if self.anchor is not None:
-            await self.client.acknowledge(probe, self.anchor, ended=self.ending)
-            self.confirmed = True
+        if (
+            self.anchor is None
+            and self.proposal is None
+            and not self.ending
+            and self.clocks_ready(now)
+        ):
+            # Lead only allows bounded grant delivery/source setup AFTER two-sided readiness.
+            # It is not a substitute for either clock gate or a retimestamp on recovery.
+            self.proposal = SourceAnchor(self.source_zero_ns, now + 1_000_000_000)
+        row = {
+            "t1_source_send_ns": str(probe.local_send_ns),
+            "t2_worker_receive_ns": str(probe.remote_receive_ns),
+            "t3_worker_send_ns": str(probe.remote_send_ns),
+            "t4_source_receive_ns": str(probe.local_receive_ns),
+            "source_report_send_ns": str(now),
+            "nonce": probe.nonce,
+            "acknowledged": False,
+        }
+        if len(self.samples) >= 2048:
+            raise ValueError("bounded clock diagnostic count exceeded")
+        self.samples.append(row)
+        # Every probe is echoed, even before an anchor exists. Worker warmup must
+        # see the same evidence; starting the source requires its explicit ACK.
+        self.peer = await self.client.acknowledge(
+            probe, self.anchor or self.proposal, ended=self.ending
+        )
+        now = time.monotonic_ns()
+        if self.anchor is not None and self.peer.anchor != self.anchor:
+            raise ValueError("accepted source anchor disappeared")
+        if self.peer.anchor is not None:
+            if self.peer.anchor != (self.anchor or self.proposal):
+                raise ValueError("foreign accepted source anchor")
+            self.anchor = self.peer.anchor
+        else:
+            self.proposal = None  # Explicit nonacceptance, not a lost response.
+        self.confirmed = True
         self.access = access
-        now = time.monotonic_ns()
-        self.samples.append(
+        bounds = self.window.current(now)
+        row.update(
             {
                 "at_ns": now,
+                "t5_worker_report_receive_ns": str(self.peer.bounds.measured_at_ns),
+                "t6_source_ack_receive_ns": str(now),
                 "bounds": bounds.to_dict(),
+                "peer": self.peer.to_dict(),
                 "uncertainty_ns": bounds.uncertainty_ns(now),
-                "ready": self.ready(),
-                "acknowledged": self.anchor is not None,
+                "ready": self.anchor is not None and self.confirmed and self.clocks_ready(now),
+                "acknowledged": True,
             }
         )
-        if len(self.samples) > 2048:
-            raise ValueError("bounded clock diagnostic count exceeded")
+
+    def clocks_ready(self, now):
+        if self.peer is None:
+            return False
+        try:
+            own = self.window.current(now)
+            own.require(now)
+            # Conservative latest possible Worker time includes ACK transit and
+            # elapsed time since receipt. A cached peer-ready boolean is unsafe.
+            _, upper = own.offset_at(now)
+            self.peer.bounds.require(now + upper)
+        except ClockMappingError:
+            return False
+        return True
 
     def ready(self):
         if self.failure:
             raise RuntimeError("source clock task failed") from self.failure
-        if not self.confirmed:
+        if not self.confirmed or self.anchor is None:
             return False
-        try:
-            now = time.monotonic_ns()
-            self.window.current(now).require(now)
-        except ClockMappingError:
-            return False
-        return True
+        return self.clocks_ready(time.monotonic_ns())
 
     async def loop(self):
         while not self.stopping:
@@ -91,13 +129,17 @@ class SourceControl:
 
     async def wait_anchor(self):
         deadline = time.monotonic() + 10
-        while self.anchor is None or not self.confirmed:
+        while True:
             if self.failure:
                 raise RuntimeError("initial source clock handshake failed") from self.failure
             if time.monotonic() > deadline:
                 raise TimeoutError("bounded initial source clock handshake")
+            if self.anchor is not None:
+                if time.monotonic_ns() >= self.anchor.local_zero_ns - 100_000_000:
+                    raise TimeoutError("joint source admission missed immutable start")
+                if self.ready() and self.access is not None:
+                    return self.anchor
             await asyncio.sleep(0.01)
-        return self.anchor
 
     async def finish(self):
         self.ending = True

experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py

@@ -276,7 +276,7 @@ def run(args):
         "costmap_freshness_mode": args.costmap_freshness,
         "ddrnet_min_source_interval_ms": args.ddrnet_min_source_interval_ms,
         "input_transport": args.input_transport,
-        "network_clock_mode": "acknowledged-monotonic/v1"
+        "network_clock_mode": "joint-start-monotonic/v2"
         if getattr(args, "network_cross_host", False)
         else "local-only",
         "network_clock_envelope": {