diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py b/experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py index 60c9b96..8e65999 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_grpc_source.py +++ b/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( diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_source_control.py b/experiments/perception/worker/streaming_profile_stage1/pilot_source_control.py index b989fbc..94edb7b 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_source_control.py +++ b/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 diff --git a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py b/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py index 572b60b..491c993 100644 --- a/experiments/perception/worker/streaming_profile_stage1/run_joint_pilot.py +++ b/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": { diff --git a/src/k1link/perception/streaming_clock.py b/src/k1link/perception/streaming_clock.py index 4c937da..90258eb 100644 --- a/src/k1link/perception/streaming_clock.py +++ b/src/k1link/perception/streaming_clock.py @@ -145,6 +145,7 @@ class ClockBounds: or type(self.offset_lower_ns) is not int or type(self.offset_upper_ns) is not int or self.offset_lower_ns > self.offset_upper_ns + or not -(2**63) < self.offset_lower_ns <= self.offset_upper_ns < 2**63 or type(self.rate_ppm) is not int or not 1 <= self.rate_ppm <= 1000 or type(self.timestamp_error_ns) is not int @@ -187,6 +188,46 @@ class ClockBounds: "symmetric_network_assumed": False, } + @classmethod + def from_dict(cls, value: object) -> ClockBounds: + """Closed, exact-integer wire contract, including signed clock offsets.""" + if not isinstance(value, dict): + raise ClockMappingError("invalid clock bounds document") + + def integer(key: str) -> int: + raw = value.get(key) + if not isinstance(raw, str) or not 1 <= len(raw) <= 20: + raise ClockMappingError("clock nanoseconds must be canonical decimal strings") + try: + parsed = int(raw) + except ValueError as exc: + raise ClockMappingError("invalid clock nanoseconds") from exc + if str(parsed) != raw: + raise ClockMappingError("noncanonical clock nanoseconds") + return parsed + + try: + result = cls( + value["local_clock_id"], + value["remote_clock_id"], + integer("measured_at_ns"), + integer("expires_at_ns"), + integer("offset_lower_ns"), + integer("offset_upper_ns"), + value["relative_rate_budget_ppm"], + integer("timestamp_error_budget_ns"), + value["samples"], + ) + except KeyError as exc: + raise ClockMappingError("incomplete clock bounds") from exc + if ( + value != result.to_dict() + or value.get("conditional_rate_error_envelope") is not True + or value.get("symmetric_network_assumed") is not False + ): + raise ClockMappingError("unsupported clock bounds document") + return result + class ClockWindow: def __init__( diff --git a/src/k1link/perception/streaming_control_grpc.py b/src/k1link/perception/streaming_control_grpc.py index ab6904c..02d5976 100644 --- a/src/k1link/perception/streaming_control_grpc.py +++ b/src/k1link/perception/streaming_control_grpc.py @@ -21,9 +21,9 @@ import grpc # type: ignore[import-untyped] from . import streaming_wire as wire from .realtime_contract import StreamStart -from .streaming_clock import ClockProbe +from .streaming_clock import ClockProbe, ClockReceipt from .streaming_grpc import OPTIONS, StreamAccess -from .streaming_source_clock import SourceAnchor +from .streaming_source_clock import SourceAnchor, SourceClockState SERVICE = "missioncore.perception.v1.StreamControl" METHOD = f"/{SERVICE}/Poll" @@ -79,7 +79,8 @@ class StreamControlEndpoint: *, clock_id: str, clock_ns: Callable[[], int] = time.monotonic_ns, - observe: Callable[[ClockProbe, int, SourceAnchor, bool], None] | None = None, + observe: Callable[[ClockProbe, int, SourceAnchor | None, bool], SourceClockState] + | None = None, ) -> None: wire.identifier(clock_id) self.activation, self.pending, self.clock_id, self.clock_ns = ( @@ -198,7 +199,11 @@ class StreamControlEndpoint: try: self._authenticate(context) value = _document(raw) - if set(value) != {"probe", "anchor", "ended"} or self.observe is None: + if ( + set(value) != {"schema_version", "probe", "anchor", "ended"} + or value["schema_version"] != "missioncore.stream-clock-report/v2" + or self.observe is None + ): raise ValueError("clock receipt not admitted") evidence = value["probe"] with self._lock: @@ -225,17 +230,18 @@ class StreamControlEndpoint: ) ), ) - anchor = SourceAnchor.from_dict(value["anchor"]) - self.observe(probe, received, anchor, value["ended"]) + anchor = None if value["anchor"] is None else SourceAnchor.from_dict(value["anchor"]) + state = self.observe(probe, received, anchor, value["ended"]) except (ValueError, TypeError): self.rejected += 1 await context.abort(grpc.StatusCode.INVALID_ARGUMENT, "clock receipt rejected") return b"" return wire.canonical( { - "schema_version": "missioncore.stream-clock-receipt/v1", + "schema_version": "missioncore.stream-clock-receipt/v2", "activation_binding": wire.binding(self.activation), "nonce": probe.nonce, + "state": state.to_dict(), } ) @@ -338,8 +344,8 @@ class StreamControlClient: await self.channel.close() async def acknowledge( - self, probe: ClockProbe, anchor: SourceAnchor, *, ended: bool = False - ) -> None: + self, probe: ClockProbe, anchor: SourceAnchor | None, *, ended: bool = False + ) -> SourceClockState: if self.polling or probe.local_clock_id != self.clock_id or type(ended) is not bool: raise ValueError("one scoped control request at a time") self.polling = True @@ -365,17 +371,37 @@ class StreamControlClient: ) response = _document( await self.report_call( - wire.canonical({"probe": evidence, "anchor": anchor.to_dict(), "ended": ended}), + wire.canonical( + { + "schema_version": "missioncore.stream-clock-report/v2", + "probe": evidence, + "anchor": None if anchor is None else anchor.to_dict(), + "ended": ended, + } + ), metadata=self.ticket.metadata(), timeout=0.5, wait_for_ready=False, ) ) + state = SourceClockState.from_dict(response.get("state")) if response != { - "schema_version": "missioncore.stream-clock-receipt/v1", + "schema_version": "missioncore.stream-clock-receipt/v2", "activation_binding": wire.binding(self.ticket.activation), "nonce": probe.nonce, + "state": state.to_dict(), }: raise ValueError("clock receipt response binding mismatch") + if ( + (state.bounds.local_clock_id, state.bounds.remote_clock_id) + != (probe.remote_clock_id, probe.local_clock_id) + or state.bounds.rate_ppm != 500 + or state.bounds.timestamp_error_ns != 50_000 + or state.ended != ended + or (state.anchor is not None and state.anchor != anchor) + ): + raise ValueError("clock receipt state binding mismatch") + ClockReceipt(probe, state.bounds.measured_at_ns) + return state finally: self.polling = False diff --git a/src/k1link/perception/streaming_source_clock.py b/src/k1link/perception/streaming_source_clock.py index 1352d50..e8a680c 100644 --- a/src/k1link/perception/streaming_source_clock.py +++ b/src/k1link/perception/streaming_source_clock.py @@ -6,7 +6,14 @@ import threading from dataclasses import dataclass from . import streaming_wire as wire -from .streaming_clock import ClockMappingError, ClockProbe, ClockReceipt, ClockWindow, _stamp +from .streaming_clock import ( + ClockBounds, + ClockMappingError, + ClockProbe, + ClockReceipt, + ClockWindow, + _stamp, +) from .streaming_continuity import StreamSuspended @@ -32,6 +39,36 @@ class SourceAnchor: return cls(wire.uint64(value["source_zero_ns"]), wire.uint64(value["local_zero_ns"])) +@dataclass(frozen=True) +class SourceClockState: + bounds: ClockBounds + anchor: SourceAnchor | None + ended: bool + + def __post_init__(self) -> None: + if not isinstance(self.bounds, ClockBounds) or type(self.ended) is not bool: + raise ValueError("invalid source clock state") + if self.anchor is not None and not isinstance(self.anchor, SourceAnchor): + raise ValueError("invalid accepted source anchor") + + def to_dict(self) -> dict[str, object]: + return { + "bounds": self.bounds.to_dict(), + "anchor": None if self.anchor is None else self.anchor.to_dict(), + "ended": self.ended, + } + + @classmethod + def from_dict(cls, value: object) -> SourceClockState: + if not isinstance(value, dict) or set(value) != {"bounds", "anchor", "ended"}: + raise ValueError("invalid source clock state document") + return cls( + ClockBounds.from_dict(value["bounds"]), + None if value["anchor"] is None else SourceAnchor.from_dict(value["anchor"]), + value["ended"], + ) + + class SourceClockMonitor: def __init__(self, worker_clock_id: str, source_zero_ns: int) -> None: wire.identifier(worker_clock_id) @@ -44,30 +81,40 @@ class SourceClockMonitor: self._lock = threading.RLock() def observe( - self, probe: ClockProbe, received_ns: int, anchor: SourceAnchor, ended: bool - ) -> None: + self, probe: ClockProbe, received_ns: int, anchor: SourceAnchor | None, ended: bool + ) -> SourceClockState: """Called only after the transport verifies the one-use issued challenge.""" receipt = ClockReceipt(probe, received_ns) if probe.remote_clock_id != self.worker_clock_id or type(ended) is not bool: raise ValueError("foreign clock/status") with self._lock: - if self.ended or anchor.source_zero_ns != self.source_zero: + if self.ended or (anchor is not None and anchor.source_zero_ns != self.source_zero): raise ValueError("source activation ended or source anchor changed") if self.anchor is None: - if not 0 <= anchor.local_zero_ns - probe.local_receive_ns <= 10_000_000_000: + if ( + anchor is not None + and not 0 <= anchor.local_zero_ns - probe.local_receive_ns <= 10_000_000_000 + ): raise ValueError("initial source start must be bounded and not in the past") - self.anchor = anchor + elif anchor != self.anchor: + raise ValueError("source timeline is immutable for the activation") + if self.window is None: self.window = ClockWindow( self.worker_clock_id, probe.local_clock_id, rate_ppm=500, timestamp_error_ns=50_000, ) - elif anchor != self.anchor: - raise ValueError("source timeline is immutable for the activation") - assert self.window is not None - self.window.add(receipt) + bounds = self.window.add(receipt) + if self.anchor is None and anchor is not None: + try: + bounds.require(received_ns) + except ClockMappingError: + pass # Explicitly unaccepted proposal; keep warming BOTH clock windows. + else: + self.anchor = anchor self.ended = ended + return SourceClockState(bounds, self.anchor, ended) def observed(self, now_ns: int) -> tuple[int, int]: """Source time midpoint + integer half-width, freshly checked at use.""" diff --git a/tests/test_perception_network_graph_adapter.py b/tests/test_perception_network_graph_adapter.py index ef55f1a..2ad9a0b 100644 --- a/tests/test_perception_network_graph_adapter.py +++ b/tests/test_perception_network_graph_adapter.py @@ -161,68 +161,6 @@ def test_source_eof_ack_precedes_data_channel_retirement(adapter, has_stream): assert calls == (["ack", "end", "drain"] if has_stream else ["ack"]) -def test_source_anchor_latches_only_after_initial_clock_gate_and_never_on_recovery( - adapter, monkeypatch -): - module = importlib.import_module("pilot_source_control") - healthy, acknowledgements = [False], [] - - class Bounds: - def require(self, now): - if not healthy[0]: - raise module.ClockMappingError("synthetic excessive uncertainty") - - def to_dict(self): - return {} - - def uncertainty_ns(self, now): - return 1_000_000 if healthy[0] else 6_000_000 - - class Client: - def __init__(self, *args, **kwargs): - pass - - async def poll(self): - return object(), None - - async def acknowledge(self, probe, anchor, *, ended): - acknowledgements.append(anchor) - - monkeypatch.setattr(module, "StreamControlClient", Client) - monkeypatch.setattr( - module, - "ClockWindow", - lambda *a, **kw: SimpleNamespace(add=lambda probe: Bounds(), current=lambda now: Bounds()), - ) - control = module.SourceControl( - "unused", - b"", - { - "activation": identity().to_dict(), - "token": "b" * 64, - "remote_clock_id": "worker", - "source_zero_ns": str(10**12), - }, - ) - - async def check(): - await control.update() - assert control.anchor is None and not control.ready() and not acknowledgements - healthy[0] = True - await control.update() - anchor = await control.wait_anchor() - assert anchor == control.anchor and control.ready() - healthy[0] = False - await control.update() - assert not control.ready() and control.anchor == anchor - healthy[0] = True - await control.update() - assert control.ready() and acknowledgements == [anchor] * 3 - assert [row["acknowledged"] for row in control.samples] == [False, True, True, True] - - asyncio.run(check()) - - def test_cross_host_bridge_clock_expiry_retains_owner_and_rotates_epoch(tmp_path, tls, adapter): # noqa: F811 module, _ = adapter cert, key = tmp_path / "cert", tmp_path / "key" diff --git a/tests/test_perception_source_clock.py b/tests/test_perception_source_clock.py index a81817a..fe8a055 100644 --- a/tests/test_perception_source_clock.py +++ b/tests/test_perception_source_clock.py @@ -18,6 +18,37 @@ from k1link.perception.streaming_queue import StreamMailbox from k1link.perception.streaming_source_clock import SourceAnchor, SourceClockMonitor +def test_responder_warms_without_source_admission_and_rejects_wide_start(): + monitor = SourceClockMonitor("worker", 10**12) + wide = sample(outbound=20_000_000, inbound=20_000_000) + now = wide.local_receive_ns + 10**12 + 1_000_000 + anchor = SourceAnchor(10**12, wide.local_receive_ns + 1_000_000_000) + response = monitor.observe(wide, now, anchor, False) + assert response.anchor is monitor.anchor is None and response.bounds.samples == 1 + with pytest.raises(StreamSuspended): + monitor.observed(now) + narrow = sample(2) + now = narrow.local_receive_ns + 10**12 + 1_000_000 + response = monitor.observe(narrow, now, None, False) + assert response.anchor is None and response.bounds.samples == 2 + response.bounds.require(now) + with pytest.raises(StreamSuspended): + monitor.observed(now) # Valid clock is not permission to start without an anchor. + third = sample(3) + now = third.local_receive_ns + 10**12 + 1_000_000 + assert monitor.observe(third, now, anchor, False).anchor == anchor + assert monitor.observed(now)[1] < 5_000_000 + + +def test_end_before_start_is_terminal_without_inventing_a_source_timeline(): + monitor = SourceClockMonitor("worker", 10**12) + first = sample() + state = monitor.observe(first, first.local_receive_ns + 10**12 + 1_000_000, None, True) + assert state.ended and state.anchor is None + with pytest.raises(StreamSuspended): + monitor.observed(state.bounds.measured_at_ns) + + def sample(number=1, outbound=1_000_000, inbound=2_000_000, offset=10**12): t1 = 10**16 + number * 100_000_000 return ClockProbe( diff --git a/tests/test_perception_source_start.py b/tests/test_perception_source_start.py new file mode 100644 index 0000000..18ae62b --- /dev/null +++ b/tests/test_perception_source_start.py @@ -0,0 +1,231 @@ +"""Joint pre-start admission, using real interval windows with synthetic time.""" + +# ruff: noqa: F811 -- shared probe path fixture +import asyncio +import importlib +from dataclasses import replace +from types import SimpleNamespace + +import pytest +from test_perception_network_graph_adapter import adapter # noqa: F401 +from test_perception_streaming_control_grpc import controlled +from test_perception_streaming_grpc import identity, tls # noqa: F401 + +from k1link.perception.streaming_clock import ClockProbe +from k1link.perception.streaming_source_clock import SourceClockMonitor + + +@pytest.fixture +def joint(adapter, monkeypatch): + module = importlib.import_module("pilot_source_control") + clock = [10**16] + monitor = SourceClockMonitor("worker", 10**12) + state = SimpleNamespace(lost_ack=False, wide_peer=False, grant=True, proposals=[]) + + class Client: + def __init__(self, *args, clock_id): + self.clock_id = clock_id + + async def poll(self): + clock[0] += 100_000_000 + t1 = clock[0] + probe = ClockProbe( + self.clock_id, + "worker", + str(t1), + t1, + t1 + 10**12 + 1_000_000, + t1 + 10**12 + 1_010_000, + t1 + 3_010_000, + ) + clock[0] = probe.local_receive_ns + return probe, object() if monitor.anchor and state.grant else None + + async def acknowledge(self, probe, anchor, *, ended): + state.proposals.append(anchor) + peer = monitor.observe(probe, clock[0] + 10**12 + 1_000_000, anchor, ended) + clock[0] += 2_000_000 + if state.lost_ack: + state.lost_ack = False + raise TimeoutError("lost ACK after Worker accepted anchor") + if state.wide_peer: + peer = replace( + peer, + bounds=replace( + peer.bounds, + offset_lower_ns=-(10**12) - 6_000_000, + offset_upper_ns=-(10**12) + 6_000_000, + ), + ) + return peer + + monkeypatch.setattr( + module, + "time", + SimpleNamespace(monotonic_ns=lambda: clock[0], monotonic=lambda: clock[0] / 1e9), + ) + monkeypatch.setattr(module, "StreamControlClient", Client) + control = module.SourceControl( + "unused", + b"", + { + "activation": identity().to_dict(), + "token": "b" * 64, + "remote_clock_id": "worker", + "source_zero_ns": str(10**12), + }, + ) + return control, monitor, state, clock + + +def test_both_windows_warm_before_anchor_and_grant_and_never_reanchor(joint): + control, monitor, state, clock = joint + + async def check(): + await control.update() + assert control.anchor is monitor.anchor is None + assert len(monitor.window.samples) == 1 and state.proposals == [None] + assert not control.ready() + await control.update() + anchor = control.anchor + assert anchor is not None and anchor == monitor.anchor and control.access is None + await control.update() + assert await control.wait_anchor() == anchor and control.ready() + state.wide_peer = True + await control.update() + assert not control.ready() and control.anchor == anchor + state.wide_peer = False + await control.update() + assert control.ready() and control.anchor == anchor + assert all(p == anchor for p in state.proposals[1:]) + assert all(row["acknowledged"] for row in control.samples) + row = control.samples[-1] + assert ( + int(row["t1_source_send_ns"]) + < int(row["t4_source_receive_ns"]) + < int(row["t6_source_ack_receive_ns"]) + ) + assert int(row["t3_worker_send_ns"]) < int(row["t5_worker_report_receive_ns"]) + clock[0] += 2_000_000_000 + assert not control.ready() and control.anchor == anchor + + asyncio.run(check()) + + +def test_one_sided_readiness_never_proposes_start(joint): + control, monitor, state, _ = joint + + async def check(): + state.wide_peer = True + for _ in range(4): + await control.update() + assert control.anchor is monitor.anchor is None + assert state.proposals == [None] * 4 + state.wide_peer = False + await control.update() + assert control.anchor is None # Fresh peer evidence arrives in this ACK. + await control.update() + assert control.anchor == monitor.anchor and control.ready() + + asyncio.run(check()) + + +def test_lost_acceptance_echo_keeps_exact_same_proposal(joint): + control, monitor, state, _ = joint + + async def check(): + await control.update() + state.lost_ack = True + with pytest.raises(TimeoutError): + await control.update() + proposal = control.proposal + assert proposal is not None and control.anchor is None and monitor.anchor == proposal + await control.update() + assert await control.wait_anchor() == proposal + assert state.proposals[-2:] == [proposal, proposal] + assert not control.samples[-2]["acknowledged"] + + asyncio.run(check()) + + +def test_missing_grant_cannot_silently_start_late_or_reanchor(joint): + control, _, state, clock = joint + + async def check(): + state.grant = False + await control.update() + await control.update() + anchor = control.anchor + waiting = asyncio.create_task(control.wait_anchor()) + await asyncio.sleep(0) + assert not waiting.done() + clock[0] = anchor.local_zero_ns + with pytest.raises(TimeoutError, match="immutable start"): + await waiting + assert control.anchor == anchor and control.access is None + + asyncio.run(check()) + + +def test_peer_ready_does_not_replace_local_gate_or_refresh_peer_expiry(joint, monkeypatch): + control, _, _, clock = joint + + async def check(): + await control.update() + await control.update() + assert control.ready() + original = control.window.current + + def wide(now): + return replace( + original(now), + offset_lower_ns=10**12 - 6_000_000, + offset_upper_ns=10**12 + 6_000_000, + ) + + monkeypatch.setattr(control.window, "current", wide) + assert not control.ready() + monkeypatch.setattr(control.window, "current", original) + assert control.ready() + old_peer = control.peer + clock[0] += 2_000_000_000 + await control.update() + assert control.ready() + control.peer = old_peer + assert not control.ready() # Fresh Mac probe cannot revive stale Worker evidence. + + asyncio.run(check()) + + +def test_real_tls_joint_handshake_waits_for_worker_anchor_and_data_grant(tmp_path, tls, adapter): + module = importlib.import_module("pilot_source_control") + + async def check(): + async with controlled(tmp_path, tls) as (run, endpoint, server, client, pending, port): + monitor = SourceClockMonitor("worker", 10**12) + server.observe = monitor.observe + server.pending = lambda: pending[0] if monitor.anchor is not None else None + control = module.SourceControl( + f"localhost:{port}", + tls[0], + { + "activation": identity().to_dict(), + "token": client.ticket.token, + "remote_clock_id": "worker", + "source_zero_ns": str(10**12), + }, + ) + control.start() + try: + anchor = await control.wait_anchor() + assert anchor == monitor.anchor + assert control.access == pending[0] and control.ready() + assert control.samples[0]["peer"]["anchor"] is None + assert len(control.samples) >= 3 + assert endpoint.active is None and run.lease.start == identity() + await control.finish() + assert monitor.ended + finally: + await control.close() + + asyncio.run(check()) diff --git a/tests/test_perception_streaming_clock.py b/tests/test_perception_streaming_clock.py index ee4c00c..31da44b 100644 --- a/tests/test_perception_streaming_clock.py +++ b/tests/test_perception_streaming_clock.py @@ -4,11 +4,62 @@ from dataclasses import replace import pytest -from k1link.perception.streaming_clock import ClockMappingError, ClockProbe, ClockWindow +from k1link.perception.streaming_clock import ( + ClockBounds, + ClockMappingError, + ClockProbe, + ClockWindow, +) BASE = 10**16 # Deliberately above IEEE754 exact-integer range. +@pytest.mark.parametrize( + "fault", + [ + None, + "float", + "integer", + "plus", + "leading", + "negative_zero", + "extra", + "missing", + "flags", + "wide", + "schema", + ], +) +def test_bounds_wire_exact_closed_and_signed(fault): + bounds = window().add(probe()) + document = bounds.to_dict() + if fault is None: + assert ClockBounds.from_dict(document) == bounds + return + if fault == "float": + document["measured_at_ns"] = float(bounds.measured_at_ns) + elif fault == "integer": + document["measured_at_ns"] = bounds.measured_at_ns + elif fault == "plus": + document["offset_upper_ns"] = "+1" + elif fault == "leading": + document["measured_at_ns"] = "0" + document["measured_at_ns"] + elif fault == "negative_zero": + document["offset_upper_ns"] = "-0" + elif fault == "extra": + document["ready"] = True + elif fault == "missing": + del document["samples"] + elif fault == "flags": + document["conditional_rate_error_envelope"] = 1 + elif fault == "wide": + document["offset_upper_ns"] = str(2**63) + else: + document["schema_version"] = "old" + with pytest.raises(ClockMappingError): + ClockBounds.from_dict(document) + + def probe(number=1, *, outbound=1_000_000, inbound=2_000_000, offset=-(10**12)): sent = BASE + number * 100_000_000 return ClockProbe( diff --git a/tests/test_perception_streaming_control_grpc.py b/tests/test_perception_streaming_control_grpc.py index 76c9af7..f0bb5de 100644 --- a/tests/test_perception_streaming_control_grpc.py +++ b/tests/test_perception_streaming_control_grpc.py @@ -254,3 +254,39 @@ def test_wrong_clock_receipt_cannot_update_mapping(tmp_path, tls, fault): assert monitor.anchor is None and monitor.window is None asyncio.run(check()) + + +@pytest.mark.parametrize("fault", ["v1", "nonce", "clock", "anchor", "time", "envelope", "end"]) +def test_receipt_v2_client_rejects_unbound_or_legacy_state(tmp_path, tls, fault): + async def check(): + async with controlled(tmp_path, tls) as (_, _, control, client, _, _): + control.observe = SourceClockMonitor("worker", 10**12).observe + probe, _ = await client.poll() + original = client.report_call + + async def altered(raw, **kwargs): + value = wire.parse_header(bytearray(await original(raw, **kwargs))) + if fault == "v1": + value["schema_version"] = "missioncore.stream-clock-receipt/v1" + elif fault == "nonce": + value["nonce"] = "foreign" + elif fault == "clock": + value["state"]["bounds"]["local_clock_id"] = "foreign" + elif fault == "anchor": + value["state"]["anchor"] = SourceAnchor( + 10**12, probe.local_receive_ns + ).to_dict() + elif fault == "time": + value["state"]["bounds"]["measured_at_ns"] = str(probe.remote_send_ns - 1) + elif fault == "envelope": + value["state"]["bounds"]["relative_rate_budget_ppm"] = 1 + else: + value["state"]["ended"] = True + return wire.canonical(value) + + client.report_call = altered + with pytest.raises(ValueError): + await client.acknowledge(probe, None) + assert not client.polling + + asyncio.run(check())