"""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())