diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py new file mode 100644 index 0000000..885ce5b --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_clock_route_probe.py @@ -0,0 +1,225 @@ +"""Small clock-only route diagnostic; no source, data grant, GPU or model workload.""" + +import argparse +import asyncio +import json +import os +import secrets +import time +from datetime import UTC, datetime +from pathlib import Path + +import grpc +from pilot_network_control import read_control, write_control + +from k1link.perception.realtime_contract import StreamStart +from k1link.perception.streaming_clock import ClockMappingError, ClockWindow +from k1link.perception.streaming_control_grpc import ( + MAX_CONTROL, + ControlTicket, + StreamControlClient, + StreamControlEndpoint, +) +from k1link.perception.streaming_source_clock import SourceClockMonitor + + +class LoopWitness: + """Observe scheduler delay; never subtract it from an admission timestamp.""" + + def __init__(self): + self.rows = [] + self.stopping = False + + async def run(self): + while not self.stopping: + due = time.monotonic_ns() + 10_000_000 + await asyncio.sleep(0.01) + now = time.monotonic_ns() + if len(self.rows) >= 8192: + raise ValueError("bounded scheduler witness exceeded") + self.rows.append({"due_ns": str(due), "observed_ns": str(now)}) + + +def assessment(own, peer, now): + own_uncertainty = own.uncertainty_ns(now) + _, upper = own.offset_at(now) + worker_latest = now + upper + peer_uncertainty = None + try: + peer_uncertainty = peer.uncertainty_ns(worker_latest) + own.require(now) + peer.require(worker_latest) + ready = True + except ClockMappingError: + ready = False + return { + "at_ns": str(now), + "own_bounds": own.to_dict(), + "peer_bounds": peer.to_dict(), + "own_uncertainty_ns": str(own_uncertainty), + "peer_at_use_uncertainty_ns": None if peer_uncertainty is None else str(peer_uncertainty), + "ready": ready, + } + + +async def client(args, report): + bootstrap = read_control(Path(args.bootstrap)) + ticket = ControlTicket(StreamStart.from_dict(bootstrap["activation"]), bootstrap["token"]) + client = StreamControlClient( + args.target, + Path(args.certificate).read_bytes(), + ticket, + clock_id="route-source-" + secrets.token_hex(12), + ) + window = ClockWindow( + client.clock_id, bootstrap["remote_clock_id"], rate_ppm=500, timestamp_error_ns=50_000 + ) + try: + for index in range(args.samples): + probe, grant = await client.poll() + if grant is not None: + raise ValueError("clock-only diagnostic must never receive a data grant") + window.add(probe) + report_sent = time.monotonic_ns() + peer = await client.acknowledge(probe, None, ended=index == args.samples - 1) + arrived = time.monotonic_ns() + if peer.anchor is not None: + raise ValueError("clock-only diagnostic cannot start a recording") + report["samples"].append( + { + "nonce": probe.nonce, + "t1_ns": str(probe.local_send_ns), + "t2_ns": str(probe.remote_receive_ns), + "t3_ns": str(probe.remote_send_ns), + "t4_ns": str(probe.local_receive_ns), + "report_sent_ns": str(report_sent), + "t5_ns": str(peer.bounds.measured_at_ns), + "t6_ns": str(arrived), + **assessment(window.current(arrived), peer.bounds, arrived), + } + ) + if index != args.samples - 1: + await asyncio.sleep(0.05) + finally: + await client.close() + + +async def server(args, report): + activation = StreamStart( + "clock-route-probe", + "no-recording", + "worker-006", + "clock-only", + 1, + *(["a" * 64] * 4), + "clock-diagnostic", + "live", + ) + ticket = ControlTicket(activation, secrets.token_hex(32)) + clock_id = "route-worker-" + secrets.token_hex(12) + monitor = SourceClockMonitor(clock_id, 10**12) + + def observe(probe, received, anchor, ended): + if anchor is not None: + raise ValueError("this diagnostic cannot activate a recording") + entered = time.monotonic_ns() + result = monitor.observe(probe, received, anchor, ended) + if len(report["samples"]) >= 128: + raise ValueError("bounded responder diagnostic exceeded") + report["samples"].append( + { + "nonce": probe.nonce, + "t5_ns": str(received), + "observer_entered_ns": str(entered), + "observer_finished_ns": str(time.monotonic_ns()), + "bounds": result.bounds.to_dict(), + } + ) + return result + + endpoint = StreamControlEndpoint(ticket, lambda: None, clock_id=clock_id, observe=observe) + rpc = grpc.aio.server( + options=( + ("grpc.max_send_message_length", MAX_CONTROL), + ("grpc.max_receive_message_length", MAX_CONTROL), + ), + maximum_concurrent_rpcs=3, + ) + rpc.add_generic_rpc_handlers((endpoint.handler(),)) + credentials = grpc.ssl_server_credentials( + ((Path(args.private_key).read_bytes(), Path(args.certificate).read_bytes()),) + ) + if not rpc.add_secure_port(args.target, credentials): + raise ValueError("clock diagnostic bind failed") + await rpc.start() + try: + write_control( + Path(args.bootstrap), + { + "activation": activation.to_dict(), + "token": ticket.token, + "remote_clock_id": clock_id, + }, + ) + print("CLOCK_ENDPOINT_READY", flush=True) + while not monitor.ended: + await asyncio.sleep(0.02) + await asyncio.sleep(0.2) # Allow final receipt delivery before controlled retirement. + report.update(accepted=endpoint.accepted, rejected=endpoint.rejected) + finally: + Path(args.bootstrap).unlink(missing_ok=True) + await rpc.stop(1) + + +async def run(args): + output = Path(args.output) + if output.exists(): + raise ValueError("immutable diagnostic report already exists") + report = { + "schema_version": "missioncore.clock-route-diagnostic/v1", + "role": args.role, + "started_utc": datetime.now(UTC).isoformat(), + "started_ns": str(time.monotonic_ns()), + "pid": os.getpid(), + "models": 0, + "gpu_lease": False, + "source_started": False, + "data_payloads": 0, + "actuation_allowed": False, + "samples": [], + "error": None, + } + witness = LoopWitness() + task = asyncio.create_task(witness.run()) + try: + async with asyncio.timeout(40 if args.role == "server" else 25): + await (server(args, report) if args.role == "server" else client(args, report)) + except Exception as exc: + report["error"] = type(exc).__name__ + ": " + str(exc) + finally: + witness.stopping = True + await asyncio.wait_for(task, timeout=1) + report.update(ended_ns=str(time.monotonic_ns()), scheduler_witness=witness.rows) + output.write_text(json.dumps(report, indent=2, allow_nan=False) + "\n") + print( + json.dumps( + {"role": args.role, "samples": len(report["samples"]), "error": report["error"]} + ), + flush=True, + ) + return int(report["error"] is not None) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("role", choices=("client", "server")) + parser.add_argument("--target", required=True) + parser.add_argument("--bootstrap", required=True) + parser.add_argument("--certificate", required=True) + parser.add_argument("--private-key") + parser.add_argument("--samples", type=int, default=96) + parser.add_argument("--output", required=True) + args = parser.parse_args() + if not 8 <= args.samples <= 128 or (args.role == "server" and not args.private_key): + parser.error("bounded 8..128 samples and server private key required") + raise SystemExit(asyncio.run(run(args))) diff --git a/tests/test_perception_clock_route_probe.py b/tests/test_perception_clock_route_probe.py new file mode 100644 index 0000000..5c39b97 --- /dev/null +++ b/tests/test_perception_clock_route_probe.py @@ -0,0 +1,82 @@ +"""Small diagnostic contract tests; no synthetic load on the operator Mac.""" + +# ruff: noqa: F811 -- shared fixtures +import asyncio +import importlib +import json +from dataclasses import replace +from types import SimpleNamespace + +import pytest +from test_perception_network_graph_adapter import adapter # noqa: F401 +from test_perception_source_clock import sample +from test_perception_streaming_control_grpc import controlled +from test_perception_streaming_grpc import tls # noqa: F401 + +from k1link.perception.streaming_clock import ClockReceipt, ClockWindow +from k1link.perception.streaming_source_clock import SourceClockMonitor + + +@pytest.mark.parametrize("fault", [None, "wide", "expired"]) +def test_diagnostic_uses_both_original_bounds_without_retiming(adapter, fault): + module = importlib.import_module("pilot_clock_route_probe") + probe = sample() + own = ClockWindow("source", "worker", rate_ppm=500, timestamp_error_ns=50_000).add(probe) + t5 = probe.local_receive_ns + 10**12 + 1_000_000 + peer = ClockWindow("worker", "source", rate_ppm=500, timestamp_error_ns=50_000).add( + ClockReceipt(probe, t5) + ) + now = probe.local_receive_ns + 2_000_000 + if fault == "wide": + peer = replace( + peer, offset_lower_ns=-(10**12) - 6_000_000, offset_upper_ns=-(10**12) + 6_000_000 + ) + elif fault == "expired": + peer = replace(peer, expires_at_ns=t5 + 1) + result = module.assessment(own, peer, now) + assert result["ready"] == (fault is None) + assert result["own_bounds"] == own.to_dict() and result["peer_bounds"] == peer.to_dict() + if fault == "wide": + assert int(result["peer_at_use_uncertainty_ns"]) > 5_000_000 + if fault == "expired": + assert result["peer_at_use_uncertainty_ns"] is None + + +def test_clock_only_client_has_no_source_anchor_data_grant_or_hidden_load(tmp_path, tls, adapter): + module = importlib.import_module("pilot_clock_route_probe") + + async def check(): + async with controlled(tmp_path, tls) as (_, _, server, client, pending, port): + pending[0] = None + monitor = SourceClockMonitor("worker", 10**12) + server.observe = monitor.observe + certificate, bootstrap = tmp_path / "cert.pem", tmp_path / "bootstrap.json" + certificate.write_bytes(tls[0]) + module.write_control( + bootstrap, + { + "activation": client.ticket.activation.to_dict(), + "token": client.ticket.token, + "remote_clock_id": "worker", + }, + ) + args = SimpleNamespace( + role="client", + target=f"localhost:{port}", + samples=8, + bootstrap=bootstrap, + certificate=certificate, + output=tmp_path / "client.json", + ) + assert await module.run(args) == 0 + report = json.loads(args.output.read_text()) + assert len(report["samples"]) == 8 and report["error"] is None + assert monitor.ended and monitor.anchor is None + assert report["models"] == report["data_payloads"] == 0 + assert not report["gpu_lease"] and not report["source_started"] + assert report["scheduler_witness"] + assert all(int(r["t4_ns"]) <= int(r["t6_ns"]) for r in report["samples"]) + with pytest.raises(ValueError, match="immutable"): + await module.run(args) + + asyncio.run(check())