test(perception): add bounded clock-only route and scheduler diagnostics

2026-09-02 21:57:02 +03:00
parent 35d28418f3
commit 59a74a3677
2 changed files with 307 additions and 0 deletions
@@ -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)))
@@ -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())